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.
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 graflowfrom 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:
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 justif/return/context.next_task()— no separate routing DSL, noENDsentinels, no conditional-edge registry. You debug workflows withpdb, not with a graph inspector.
| LangGraph (conditional routing) | Graflow |
|---|---|
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"}) |
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") |
| 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 |
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
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")with workflow("enrich") as ctx:
fetch_data >> (enrich_demographics | enrich_geo | enrich_behavioral) >> merge_and_store
ctx.execute("fetch_data")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")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.
Need environment isolation, pinned Python versions, or GPU runtimes? Tag a task with handler="<name>" and register a DockerTaskHandler bound to that name — the same workflow code now runs each task inside a container.
Each handler is bound to one image, so pick a container per task by registering multiple handlers under distinct names:
from graflow import task, workflow
from graflow.core.context import ExecutionContext
from graflow.core.engine import WorkflowEngine
from graflow.core.handlers.docker import DockerTaskHandler
with workflow("docker_pipeline") as wf:
@task
def local_step():
return "ran on host"
@task(handler="py311") # -> python:3.11-slim
def etl_step():
import sys
return f"etl on python {sys.version_info[:2]}"
@task(handler="pytorch_gpu") # -> pytorch/pytorch:latest
def train_step():
import torch
return f"cuda available: {torch.cuda.is_available()}"
local_step >> etl_step >> train_step
engine = WorkflowEngine()
# Register one handler per image; task's `handler=` picks which one
engine.register_handler("py311", DockerTaskHandler(image="python:3.11-slim"))
engine.register_handler("pytorch_gpu", DockerTaskHandler(image="pytorch/pytorch:latest"))
engine.execute(ExecutionContext.create(wf.graph))Mix and match freely: CPU vs. GPU, Python 3.10 vs. 3.12, custom-built images with pre-installed dependencies — it's all just "register a handler, tag the task."
💡 Custom handlers.
dockeris just one built-in handler. You can implement your ownTaskHandlerfor Kubernetes Jobs, AWS Lambda, Modal, Ray, SSH-to-remote-host, or any other execution backend — same workflow code, custom runtime. See the Task Handlers guide for the handler interface and full examples.
The same workflow code runs locally or across a Redis-backed worker pool — just change the backend.
from graflow.core.context import create_execution_context
from graflow.queue.factory import QueueBackend
context = create_execution_context(
queue_backend=QueueBackend.REDIS,
channel_backend="redis",
)# Start workers in separate terminals
python -m graflow.worker.main --worker-id worker-1
python -m graflow.worker.main --worker-id worker-2- Product Documentation — Guides, API reference, architecture
- Examples — Production-ready examples organized by complexity
- CHANGELOG
Contributions are welcome — see CONTRIBUTING.md for setup, development workflow, and PR guidelines.
Ways to contribute:
- 🐛 Report bugs or request features via GitHub Issues
- 🔌 Add a new backend (queue / channel / handler) — follow the factory pattern in
graflow/queue/andgraflow/channels/ - 📚 Contribute an example to
examples/— especially real-world agent/LLM workflows - 📖 Improve the docs at graflow.ai
- 💬 Discuss on r/GraflowAI
This project is licensed under the Apache License 2.0 — see the LICENSE file for details.