<a href="https://graflow.ai"><img width="500" height="200" alt="banner" src="https://github.com/user-attachments/assets/932a3403-7d2c-4e9d-a900-8781f97fb14b" /></a>

Graflow

Stop wrestling graph composition. Start writing Pythonic agent pipelines. Compose tasks with >> and |.

Graflow is a Python-based agent orchestration framework with a workflow-first, defined-by-run approach focused on simplicity and developer experience. Compose tasks — from ETL jobs to multi-agent LLM pipelines — using plain Python functions and two intuitive operators.

<a href="https://x.com/GraflowAI"><img src="https://img.shields.io/twitter/follow/GraflowAI?logo=X&color=%20%23f5f5f5" alt="follow on X(Twitter)"></a>

---

Why Graflow?

Most agent frameworks make you think in framework concepts — state machines, conditional edges, graph builders, node factories. Graflow lets you think in Python. Define tasks, connect them with >> and |, and run.

uv add graflow   # or: pip install graflow

from graflow import task, workflow

@task
def fetch():
    return {"users": 1200}

@task
def validate(users: int):        # auto-resolved from upstream channel
    assert users > 0
    return users

@task
def notify(users: int):
    print(f"Pipeline done: {users} users processed")

with workflow("my_pipeline") as ctx:
    fetch >> validate >> notify   # that's it
    ctx.execute("fetch")

3 decorators, 1 operator, zero boilerplate.

Try hands-on in the browser — no install required:

Before / After

Tired of writing add_conditional_edges, builder patterns, and explicit state objects? Here's the same agent loop in LangGraph vs. Graflow.

Define-and-compile vs. define-by-run. LangGraph makes you declare every edge and conditional transition before execution — the graph is frozen at compile() time, and all possible routes must be known upfront. Graflow is define-by-run (the same paradigm PyTorch uses vs. TensorFlow 1.x): the graph shape is whatever the Python code does at runtime. Control flow is just if/return/context.next_task() — no separate routing DSL, no END sentinels, no conditional-edge registry. You debug workflows with pdb, not with a graph inspector.

<table> <tr><th>LangGraph (conditional routing)</th><th>Graflow</th></tr> <tr> <td>

from langgraph.graph import StateGraph, END

builder = StateGraph(State)
builder.add_node("agent", agent_node)
builder.add_node("tool", tool_node)
builder.add_conditional_edges(
    "agent",
    should_continue,
    {"continue": "tool", "end": END},
)
builder.add_edge("tool", "agent")
builder.set_entry_point("agent")
graph = builder.compile()
result = graph.invoke({"input": "hello"})

</td> <td>

from graflow import task, workflow

with workflow("agent_loop") as ctx:
    @task(inject_context=True)
    def agent(context):
        out = call_llm(context.get_channel().get("input"))
        if not should_continue(out):
            context.terminate_workflow("done")
        return out

    @task
    def tool(out: str):
        return run_tool(out)

    agent >> tool >> agent      # loop
    ctx.execute("agent")

</td> </tr> </table>

Key Features

| Feature | What it does | |---|---| | >> and \| operators | Chain tasks sequentially or run them in parallel — no graph builder API | | Auto data flow | Task parameters are automatically resolved from upstream channels | | Fan-out / Fan-in | fetch >> (transform_a \| transform_b) >> store — diamond patterns in one line | | Dynamic tasks | Add tasks at runtime via ctx.next_task() / ctx.next_iteration() — true defined-by-run | | Distributed execution | Swap to Redis-backed queues with one config change; add workers with python -m graflow.worker.main | | Checkpoint & resume | Pause long-running workflows, resume from exact state | | Human-in-the-loop | Approvals, text input, selections mid-workflow — with Slack/webhook notifications | | LLM agents | Plug in LiteLLM, Pydantic AI, or Google ADK agents and inject them into tasks | | Tracing | Langfuse integration for observability across distributed workflows |

Workflow Patterns

Linear pipeline:   a >> b >> c >> d
Fan-out:           source >> (proc_a | proc_b | proc_c)
Fan-in:            (input_a | input_b) >> aggregate
Diamond:           fetch >> (transform_a | transform_b) >> store
Multi-stage:       (load_a | load_b) >> validate >> (proc_a | proc_b) >> store

A diamond workflow visualized:

graph LR
    fetch --> transform_a
    fetch --> transform_b
    transform_a --> store
    transform_b --> store

Real-World Snippets

ETL Pipeline

with workflow("etl") as ctx:
    @task
    def extract():
        return db.query("SELECT * FROM raw_events")

    @task
    def validate(events: list):
        return [e for e in events if e["timestamp"]]

    @task
    def load(events: list):
        warehouse.bulk_insert(events)

    extract >> validate >> load
    ctx.execute("extract")

Parallel Enrichment

with workflow("enrich") as ctx:
    fetch_data >> (enrich_demographics | enrich_geo | enrich_behavioral) >> merge_and_store
    ctx.execute("fetch_data")

Multi-Agent LLM Workflow

Register LLM agents once, then inject them directly into tasks with inject_llm_agent="<name>" — no manual lookup, no boilerplate.

from graflow import task, workflow
from graflow.llm.agents.base import LLMAgent

with workflow("research") as ctx:
    # Register agents once; factories receive the ExecutionContext
    ctx.register_llm_agent("researcher", lambda _: build_researcher_agent())
    ctx.register_llm_agent("analyst",    lambda _: build_analyst_agent())

    @task(inject_llm_agent="researcher")
    def research(agent: LLMAgent):
        return agent.run("Summarize recent ML papers")["output"]

    @task(inject_llm_agent="analyst")
    def analyze(agent: LLMAgent, research: str):   # `research` auto-resolved from channel
        return agent.run(f"Analyze: {research}")["output"]

    @task
    def publish(analyze: str):
        send_report(analyze)

    research >> analyze >> publish
    ctx.execute("research")

Dynamic Workflows: next_task and next_iteration

Graflow is defined-by-run — tasks can spawn new tasks or loop themselves based on runtime data.

next_task — conditionally add a downstream task at runtime:

from graflow import task, workflow
from graflow.core.task import TaskWrapper

with workflow("router") as ctx:
    @task(id="classify", inject_context=True)
    def classify(context, value: int = 150):
        if value > 100:
            context.next_task(TaskWrapper("high", lambda: print(f"high-value: {value * 2}")))
        else:
            context.next_task(TaskWrapper("low",  lambda: print(f"low-value: {value + 10}")))

    ctx.execute("classify")

next_task(..., goto=True) — branch to a task and skip the statically-declared successors:

with workflow("pipeline") as ctx:
    @task(id="validate", inject_context=True)
    def validate(context, record: dict):
        if not record.get("ok"):
            # Jump to error handler and SKIP process/publish below
            context.next_task(error_handler, goto=True)
            return
        # Fall through: process >> publish will run normally

    @task
    def process(): ...

    @task
    def publish(): ...

    @task
    def error_handler():
        print("recovery path")

    validate >> process >> publish
    ctx.execute("validate", value={"ok": False})

Semantics at a glance:

| Call | New task? | Runs static successors? | |------------------------------------|-----------|-------------------------| | next_task(new_task) | creates | ✅ yes (normal fan-out) | | next_task(existing_task) | — | ❌ no (auto-goto) | | next_task(task, goto=True) | either | ❌ no (explicit goto) |

Use goto=True for error recovery, early exits, or replacing the rest of the pipeline with a different branch — it's the Graflow equivalent of a conditional edge pointing at END and then jumping elsewhere, in one line.

next_iteration — loop the current task with new data until a convergence condition:

with workflow("optimize") as ctx:
    @task(id="optimizer", inject_context=True, max_cycles=20)
    def optimize(context, data=None):
        params = data or {"iteration": 0, "accuracy": 0.5}
        params = {"iteration": params["iteration"] + 1,
                  "accuracy": min(params["accuracy"] + 0.15, 0.95)}
        print(f"Iter {params['iteration']}: accuracy={params['accuracy']:.2f}")

        if params["accuracy"] >= 0.9:
            return params            # converged → workflow ends
        context.next_iteration(params)  # loop with new data

    ctx.execute("optimizer")

Use next_task for branching, next_iteration for cycles — both respect checkpointing, retries, and tracing.

Run Tasks in Docker

Nee