Metadata-Version: 2.4
Name: openworkflows
Version: 0.2.0
Summary: A node-based DAG workflow engine for AI pipelines
Requires-Python: >=3.10
Requires-Dist: httpx>=0.28.1
Requires-Dist: pydantic>=2.0.0
Provides-Extra: audio
Requires-Dist: nemo-toolkit[asr]>=1.23.0; extra == 'audio'
Requires-Dist: torch>=2.5.1; extra == 'audio'
Requires-Dist: transformers>=4.46.3; extra == 'audio'
Provides-Extra: litellm
Requires-Dist: litellm>=1.77.5; extra == 'litellm'
Description-Content-Type: text/markdown

# OpenWorkflows

A node-based DAG workflow engine for AI pipelines. Build workflows in Python or load them from a JSON document, run them as an observable event stream, and let frontends render every node from generated schemas.

- **Async-first** — workflows execute as an async stream of typed events; `run()` is just the drained stream.
- **Node errors never raise** — they surface as events and in the result, so one bad node doesn't crash your service.
- **Dynamic handles** — nodes can grow per-instance inputs/outputs (template variables, LLM tools, switch cases).
- **Control flow without cycles** — if/else and switch via branch pruning; loops via nested workflow documents.
- **Provider-agnostic LLM protocol** — inject any `ChatModel`; credentials stay with your application.
- **One versioned document** — engine definition and editor UI state live in the same JSON file.
- **Lean core** — only `httpx` and `pydantic`; LLM and audio integrations are optional extras.

Requires Python 3.10+.

## Installation

```bash
pip install openworkflows               # core
pip install "openworkflows[litellm]"    # + LiteLLM-backed ChatModel
pip install "openworkflows[audio]"      # + Whisper transcription nodes
```

## Quickstart

```python
import asyncio
from openworkflows import Workflow

async def main():
    workflow = Workflow("Basic Example")
    workflow.add_node("in", "input", {"name": "text"})
    workflow.add_node("upper", "transform", {"transform": "upper"})
    workflow.add_node("out", "output", {"name": "shouted"})

    workflow.connect("in.value", "upper.input")     # "node.handle" notation
    workflow.connect("upper.output", "out.value")

    result = await workflow.run({"text": "hello world"})
    print(result.status)    # "succeeded" | "failed" | "partial"
    print(result.outputs)   # {"shouted": "HELLO WORLD"} — from output nodes, keyed by name
    print(result.nodes["upper"].outputs)  # every node's individual result

asyncio.run(main())
```

`input` nodes define the workflow's inputs (`workflow.required_inputs()` describes them); `output` nodes collect `result.outputs`. Node errors never raise from `run()` — check `result.status` and `result.nodes[id].error`. Graph errors (cycles, unknown types, invalid config) raise early: config is validated at `add_node()`, cycles at run setup, before any event.

## Streaming execution

`stream()` is the primary API — `run()` simply drains it. It yields typed events you can forward straight to a WebSocket via `event.to_dict()`:

```python
from openworkflows import NodeFinished, NodeStreamed, WorkflowFinished

async for event in workflow.stream({"text": "hi"}):
    match event:
        case NodeStreamed():   print(event.node_id, event.channel, event.chunk)  # LLM tokens, progress
        case NodeFinished():   print(event.node_id, event.outputs)
        case WorkflowFinished(): print(event.status, event.outputs)
```

| Event | When |
|---|---|
| `WorkflowStarted` | Run begins; carries `workflow_name` and `node_ids`. |
| `NodeStarted` | A node begins executing. |
| `NodeStreamed` | An intra-node chunk (`channel`, `chunk`) — LLM tokens, loop iteration ticks. |
| `NodeFinished` | A node succeeded; carries `outputs` and `duration_ms`. |
| `NodeFailed` | A node raised; carries `error`. Never propagates as an exception. |
| `NodeSkipped` | A node was not executed (`reason`: `branch_not_taken`, `upstream_failed`, `halted`). |
| `WorkflowFinished` | Terminal; carries `status`, collected `outputs`, `duration_ms`. |

`on_error="halt"` (default) skips everything after the first failure and finishes `"failed"`; `on_error="continue"` keeps running unaffected nodes and finishes `"partial"`.

## LLM generation

The engine defines only a protocol. Nodes get the model via `ctx.service("llm")` — they never read environment variables, so credential policy stays with your application:

```python
class ChatModel(Protocol):
    async def complete(self, messages, *, model=None, tools=None,
                       temperature=None, max_tokens=None, **extra) -> ChatResult: ...
    def stream(self, messages, *, model=None, temperature=None,
               max_tokens=None, **extra) -> AsyncIterator[ChatChunk]: ...
```

Use `MockChatModel` for tests, or the LiteLLM implementation (`[litellm]` extra) for real providers — the provider folds into the model string (`"openai/gpt-4o"`, `"ollama/llama3.2"`, `"openrouter/openai/gpt-4o"`):

```python
from openworkflows.contrib.litellm import LiteLLMChatModel

workflow.add_node("llm", "generate_text", {
    "model": "openai/gpt-4o",   # params: model, temperature, max_tokens, stream
    "temperature": 0.7,
    "stream": True,             # emit tokens live as NodeStreamed(channel="text")
})
workflow.connect("prompt.text", "llm.prompt")   # inputs: prompt, system
workflow.add_service("llm", LiteLLMChatModel(api_key="..."))

# or per run, overriding workflow-level services:
await workflow.run(inputs, services={"llm": MockChatModel("canned")})
```

`LLMError` carries a coarse `code` (`auth`, `not_found`, `quota`, `connection`, `unknown`) for upstream mapping.

## Dynamic handle groups

A node class declares a *group*; each instance defines its concrete handles in config. Edges reference them by `name`, exactly like static handles.

**Template variables** — `template` fills `{{variable}}` placeholders (double braces; single braces pass through, so templates can contain JSON). Unconnected variables fall back to workflow inputs:

```python
workflow.add_node("prompt", "template", {
    "template": "Write a haiku about {{topic}}.",
    "variables": [{"name": "topic"}],          # each becomes an input handle
})
workflow.connect("in.value", "prompt.topic")
```

**LLM tools** — each tool in `generate_text`'s `tools` group becomes an *output* handle carrying the tool-call argument. Uncalled tools produce no output, so their downstream branches are pruned — the model routes the workflow:

```python
workflow.add_node("llm", "generate_text", {
    "model": "openai/gpt-4o",
    "tools": [{"name": "search", "description": "Search the web"}],
})
workflow.connect("llm.search", "searcher.query")  # runs only if the model calls 'search'
workflow.connect("llm.text", "answer.value")
```

## Control flow

Branching nodes return only the taken output handle; the runner's **branch pruning** skips everything downstream of untaken handles (`NodeSkipped`, reason `branch_not_taken`). Rejoin branches with `merge` mode `"first"`:

```python
workflow.add_node("check", "if_else", {"operator": "gte", "compare": 50})
# operators: equals, not_equals, contains, not_contains, gt, gte, lt, lte, is_empty, regex
workflow.connect("score.output", "check.value")
workflow.connect("check.true", "pass_msg.score")
workflow.connect("check.false", "fail_msg.score")

workflow.add_node("join", "merge", {"mode": "first"})   # first available value wins
workflow.connect("pass_msg.text", "join.a")
workflow.connect("fail_msg.text", "join.b")
```

`switch` routes by string match against a dynamic `cases` group (each case name is both the handle and the matched value; misses go to `default`):

```python
workflow.add_node("route", "switch", {"cases": [{"name": "billing"}, {"name": "support"}]})
```

Loops and composition embed a **nested workflow document** in config, keeping the parent graph acyclic. The child's `input`/`output` nodes become the parent node's handles, and child events are forwarded with `parent/child` node ids:

```python
child = Workflow("double")
# ... build child with an input named "item" and an output ...

workflow.add_node("sub", "subworkflow", {"workflow": child.to_dict()})   # run once
workflow.add_node("map", "for_each", {"workflow": child.to_dict(), "concurrency": 4})
# for_each feeds each iteration 'item' and 'index'; collects into 'results'
workflow.add_node("loop", "while", {"workflow": child.to_dict(),
                                    "condition": "keep_going",          # child output name
                                    "max_iterations": 100})
# while threads same-named child outputs into the next iteration's inputs
```

## The workflow document

A workflow serializes to one versioned JSON document — the engine reads `nodes`/`edges`, while `ui` and `meta` (top-level and per-node) are opaque namespaces preserved round-trip, so a visual editor's state lives in the same file:

```python
doc = workflow.to_dict()        # or workflow.to_json()
restored = Workflow.from_dict(doc)   # or Workflow.from_json(...)
```

```json
{
  "version": 1,
  "name": "Basic Example",
  "meta": {},
  "ui": {},
  "nodes": [
    {"id": "in", "type": "input", "config": {"name": "text"}, "ui": {"position": {"x": 0, "y": 0}}}
  ],
  "edges": [
    {"id": null, "source": "in", "target": "upper", "source_handle": "value", "target_handle": "input"}
  ]
}
```

Documents without a `version` field are treated as the legacy v0 shape and migrated on load.

## Custom nodes

Subclass `Node` and declare the interface as class attributes. Everything a frontend needs — labels (plain strings or `{lang: text}` dicts), handle metadata, parameter editors — lives on the declarations:

```python
from openworkflows import ExecutionContext, Handle, Node, Parameter, register_node

@register_node("repeat")                 # register_node(name, override=True) to replace
class RepeatNode(Node):
    inputs = {"text": Handle(str, label={"en": "Text", "pl": "Tekst"})}
    outputs = {"result": str}
    parameters = {
        "times": Parameter(name="times", type=int, default=2, min=1, max=10,
                           label={"en": "Times"}, description="How many repetitions"),
    }
    label = {"en": "Repeat", "pl": "Powtórz"}
    category = "text"
    icon = "🔁"

    async def execute(self, ctx: ExecutionContext) -> dict:
        return {"result": " ".join([ctx.input("text")] * self.param("times"))}
```

`Parameter` carries the UI hints (`label`, `placeholder`, `choices`, `min`/`max`, and a `component` override such as `"textarea"`, `"model_picker"`, `"workflow_picker"`); validation runs at `add_node()` time. Inside `execute`, use `ctx.input(name, default)`, `ctx.all_inputs()`, `ctx.service(name)`, and `await ctx.emit(channel, chunk)` to stream progress.

For simple cases, the `@node` decorator turns a function into a node type — inputs are inferred from the signature, outputs from the return annotation:

```python
from openworkflows import node, register_node

@register_node("count_words")
@node(outputs={"count": int})
async def count_words(text: str) -> int:
    return len(text.split())
```

A `"*"` key in `inputs`/`outputs` accepts arbitrary handle names (see `merge`); `dynamic_inputs`/`dynamic_outputs` declare `HandleGroup`s resolved from instance config.

## Schemas for frontends

Schemas are generated entirely from node declarations, so a visual editor can render any node type — palette entry, canvas handles, config form — without hardcoded knowledge:

```python
from openworkflows import get_node_schema, get_all_node_schemas, list_nodes

schema = get_node_schema("generate_text")
# {"type", "kind", "label", "description", "category", "icon", "color", "tags",
#  "inputs": [{"name", "type", "required", "label", ...}], "outputs": [...],
#  "wildcard_inputs", "wildcard_outputs",
#  "dynamic_inputs": [{"group", "item_type", "min", "max", ...}], "dynamic_outputs": [...],
#  "parameters": [{"name", "type", "component", "default", "choices", "min", "max", ...}]}
```

All `label`/`description`/`placeholder` fields are language-keyed dicts (plain strings normalize to `{"en": ...}`).

## Built-in node types

| Type | Purpose |
|---|---|
| `input` / `output` | Workflow boundary — define run inputs and collect results. |
| `template` | Fill `{{variable}}` placeholders; dynamic `variables` input group. |
| `transform` | Built-in transforms: `upper`, `lower`, `strip`, `length`, `str`, `int`, `float`, `identity`. |
| `merge` | Combine inputs — modes `dict`, `list`, `first` (branch rejoin). |
| `generate_text` | ChatModel completion with token streaming and a dynamic `tools` output group. |
| `http_request` / `http_get` / `http_post` | HTTP calls with `{variable}`-templated URL and body. |
| `if_else` / `switch` | Conditional routing via branch pruning. |
| `subworkflow` / `for_each` / `while` | Workflow-as-node composition from a nested document. |
| `transcribe_audio` / `transcribe_audio_batch` | Whisper transcription — register on `import openworkflows.contrib.audio` (`[audio]` extra). |

## Examples

Runnable scripts in [`examples/`](examples/): [basic_workflow.py](examples/basic_workflow.py), [llm_workflow.py](examples/llm_workflow.py), [control_flow.py](examples/control_flow.py), [custom_node.py](examples/custom_node.py), [parameters_example.py](examples/parameters_example.py), [json_export_import.py](examples/json_export_import.py), [http_workflow.py](examples/http_workflow.py), [schema_export.py](examples/schema_export.py), [audio_transcription.py](examples/audio_transcription.py). LLM examples use `MockChatModel`, so no API keys are needed:

```bash
uv run python examples/basic_workflow.py
```

## Development

```bash
uv sync --all-extras
uv run pytest tests/
uv run black openworkflows tests
uv run ruff check openworkflows tests
uv run mypy openworkflows
```

## License

MIT
