Flows¶

A flow is a graph of nodes over one pydantic state. You write it once with Flow. The local FlowRunner runs it on the executor; the LangGraph and ADK compilers consume the same validated graph.

flowchart LR
  classify --> retrieve["retrieve<br/>Map: one call per plan step"]
  retrieve --> merge["merge<br/>Join"]
  merge --> reason
  reason -- "needs_more (max 3 loops)" --> retrieve
  reason -- done --> approve["approve<br/>pause"]

State and merge rules¶

class Hit(BaseModel):
    source_id: str
    text: str


class ResearchState(BaseModel):
    question: str
    plan_steps: list[dict[str, Any]] = Field(default_factory=list)
    evidence: Annotated[list[Hit], Merge.keyed(by="source_id")] = Field(default_factory=list)
    rounds: Annotated[int, Merge.sum] = 0
    needs_more: bool = False
    answer: str = ""

Each field declares how several writes in one superstep combine (unannotated: replace).

Rule	Several updates in one superstep
`Merge.replace`	equal values → that value; different values → `FlowError("conflicting writes to <field>")`
`Merge.append`	list concatenation in update order
`Merge.sum`	added onto the current value
`Merge.keyed(by)`	a later item with the same key replaces the earlier one; first-seen order kept

append and keyed extend with the elements when a value is a list, else add the value itself. Update order is node name order (LangGraph's task order), then Map item index, so with keyed the node whose name sorts last wins a shared key. Unknown keys raise FlowError; the merged state is validated again with the model. Nodes of one superstep all read the same input state: treat it as read-only and return updates.

Nodes¶

async def classify(state: ResearchState) -> dict[str, Any]:
    steps = [
        {"tool": name, "args": {"query": "returns"}}
        for name in ("search_articles", "search_tickets")
    ]
    return {"plan_steps": steps}


async def reason(state: ResearchState, ctx: ToolContext) -> dict[str, Any]:
    model = node_model(ctx)  # the runner's model, bound to this node
    if model is None:
        raise ConfigError("the reason node needs FlowRunner(model=...)")
    request = LLMRequest(
        messages=(
            Message(role="system", content=node_prompt(ctx) or ""),
            Message(role="user", content=f"{len(state.evidence)} sources"),
        ),
        stage="reason",
    )
    reply = await model.generate(request)
    first_round = state.rounds == 0
    update: dict[str, Any] = {"rounds": 1, "needs_more": first_round, "answer": reply.text}
    if first_round:  # ask for one more source, then loop back to retrieval
        update["plan_steps"] = [
            *state.plan_steps,
            {"tool": "search_articles", "args": {"query": "shipping"}},
        ]
    return update


def approve_step(state: ResearchState) -> dict[str, str]:
    return {"answer": f"{state.answer} (approved)"}

Action	Runs	Writes
`fn(state)` / `fn(state, ctx)`	sync (in a worker thread) or async	the mapping it returns (or nothing for `None`)
`Call(tool, args=…, into=…)`	`executor.run_spec`; `args` may be `fn(state)`	`{into: value}`
`Map(over, tool= / tool_from= / fn=, into=…)`	one call per item of the list field `over`	one `{into: value}` per item
`Join(sources, fn=None)`	once nothing else is pending (fan-in)	what `fn` returns
`SubFlow(flow, input=…, into=…)`	another flow, as one node (sub-flows)	the child's result, mapped by `into`

Map(tool=…) passes the item (or args_fn(item)); tool_from="tool" reads the tool name from each item, arguments from item[args_from] (else the rest of the item); fn(item, state, ctx) maps without a tool. on_error="fail" fails on the first failed item in item order, "skip" drops it with a WARNING. Node options: concurrency (Map), timeout, retries (executor backoff), pause, prompt, cache (node cache).

Building and validating¶

flow = (
    Flow("research", state=ResearchState)
    .node("classify", classify)
    .node("retrieve", Map("plan_steps", tool_from="tool", into="evidence"), concurrency=8)
    .node("merge", Join(["retrieve"]))
    .node("reason", reason, prompt="Answer from the evidence only.")
    .node("approve", approve_step, pause=lambda s: {"answer": s.answer})
    .edge("classify", "retrieve")
    .edge("retrieve", "merge")
    .edge("merge", "reason")
    .route(
        "reason",
        Route(lambda s: "retrieve" if s.needs_more else "approve", targets=["retrieve", "approve"]),
        max_loops=3,
        on_exhausted="approve",
    )
    .end("approve")
)

edge(src, [a, b]) runs a and b in one superstep; Route(fn, targets) picks nodes by label (names, or label → node). validate() returns the FlowGraph, or one FlowDefinitionError listing every problem:

Checked
names	the flow and every node match `^[A-Za-z_][A-Za-z0-9_]{0,63}$`
endpoints	every edge, route target, `start`, `end` and `on_exhausted` exists
exits	static edges or one route, not both; end nodes have no exit, others do
reachability	every node is reachable from `start` (default: the first node)
joins	each source is a predecessor
cycles	every cycle passes through a route with `max_loops`
state	`Map.over`, `Map` / `Call` targets are fields; `aa_loop_counts`, `aa_visits` are reserved

Running¶

store = MemoryApprovalStore()
runner = FlowRunner(
    Executor(registry), approval=DeferredApproval(store), model=CallableModel(answer)
)
paused = await runner.run(flow, {"question": "What does the returns policy allow?"})
if paused.pending is None or paused.checkpoint is None:
    raise FlowError(f"expected the approval pause, got {paused.status}")
print(paused.status, paused.pending.tool, paused.pending.arguments)

done = await runner.resume(flow, paused.checkpoint, ApprovalDecision(approved=True))
print(done.status, done.steps, dict(done.visits))
print([hit.source_id for hit in done.state.evidence])
print(done.state.answer)

flowchart TB
  F["frontier = {start}"] --> R["ready = non-join nodes<br/>(joins only when nothing else is left)"]
  R --> G{"pause gates"}
  G -- pending --> P["status paused + checkpoint"]
  G -- denied --> D["status denied"]
  G -- approved --> E["apply edited_args<br/>run ready nodes concurrently"]
  E --> M["merge updates by rule"]
  M --> N["next = rest + edges + route targets"]
  N --> F

A route sees the superstep's input plus its own node's updates (not a sibling's). Picking its own node, or a node visited before this superstep, counts a loop; past max_loops it goes to on_exhausted or raises FlowLoopExceeded, as does max_steps.

`FlowRunner(executor, *, …)`	Default	Meaning
`approval`	the executor's	port asked by `pause` gates
`hooks`, `sink`	`()`, the executor's	node hooks (after the executor's), event sink
`max_steps`	`100`	supersteps per run
`concurrency`	none	cap on nodes per superstep
`model`	`None`	given to nodes through `node_model(ctx)`
`checkpoint_key`	per process	key that seals checkpoints (bytes, 16+)
`seal_encoder`, `seal_decoder`	`None`	seal (and load) state values of custom types: see Flow pause and resume
`trace_state`	`False`	`node.finished` also carries `updates` (each changed field's value, JSON, 2 KB cap); the run viewer keeps them for that run

Pause and resume¶

A pause node asks the port with ApprovalRequest(kind="node", tool="node:<flow>.<node>", arguments=pause(state) or {}) before its superstep; nothing in that superstep runs first.

Port answer	Result
approved	the node runs with `ctx.approval` set; `edited_args` is a state update applied first
denied (or no port)	`status="denied"`, `denied_at=<node>`
`ApprovalPending`	`status="paused"`, `pending`, `checkpoint`; nothing in that superstep ran

ApprovalPending from inside a node also pauses the flow. Resuming re-runs only the paused node, replays what it finished, and is sealed and single-use (details).

Inside a node¶

Read	Gives
`ctx.call_id`	`"<flow>:<node>:<visit>"`; its tool calls are `"…:<i>"`
`node_prompt(ctx)`	the node's `prompt` (optimisers edit them with `flow.with_prompts`)
`node_model(ctx)`	the runner's model, instrumented by the executor (`executor.model(model).bind(ctx)`; an `InstrumentedModel` is only bound, so each call is billed once)
`current_context()`	the node context while the node runs

Events: flow.started, flow.finished (status: done, paused, denied, failed or cancelled; steps), node.started, node.finished (changed: the field names it set), node.failed (error_kind), node.paused (approval_id), run.started / run.finished when the runner made the run. A node error becomes FlowError(node=…) raised from the original; control flow passes through. With trace_state the run recorder keeps state values (secret-named keys masked, 2 KB each) without capture_content; compiled flows keep names.

Files¶

File	Holds
`flow/spec.py`, `actions.py`	`Flow` builder; `Call`, `Map`, `Join`, `Route`, `node_prompt`, `node_model`
`flow/state.py`	merge rules, `merge_rules`, `apply_updates`
`flow/graph.py`	`FlowGraph`, `NodeSpec`, `RouteSpec`, validation
`flow/runner.py`, `_loop.py`	`FlowRunner`, `FlowResult`, `FlowCheckpoint`; the superstep loop
`flow/_nodes.py`, `_gates.py`, `_step.py`, `_invoke.py`	node visit, gates and resume answer, merge and routes, user code