From a single agent. To a complete system.

def build_pipeline():
    """Build the IDP pipeline DAG with event handler."""
    dag = (
        PipelineBuilder("idp-pipeline")
        .add_node("ingest", CallableStep(ingest_step), timeout_seconds=90)
        .add_node("split", CallableStep(split_step_fn), timeout_seconds=120)
        .add_node("classify", CallableStep(classify_step_fn), timeout_seconds=180)
        .add_node(
            "extract",
            CallableStep(extract_step_fn),
            retry_max=1,
            timeout_seconds=300,
        )
        .add_node("validate", CallableStep(validate_step_fn), timeout_seconds=180)
        .add_node("assemble", CallableStep(assemble_step_fn))
        .add_node("explain", CallableStep(explain_step_fn), timeout_seconds=120)
        .chain(
            "ingest",
            "split",
            "classify",
            "extract",
            "validate",
            "assemble",
            "explain",
        )
        .build_dag()
    )
    return PipelineEngine(dag, event_handler=IDPEventHandler())

class MapReduceState(BaseModel):
    items: list[str] = []
    processed: Annotated[list[str], extend] = []
    summary: str | None = None
    # Per-Send payload field — each worker receives its own item here.
    item: str | None = None


async def plan(state: MapReduceState) -> dict:
    # No state mutation; the dispatch router below decides what runs next.
    return {}


async def process_item(state: MapReduceState) -> dict:
    assert state.item is not None
    return {"processed": [f"processed:{state.item}"]}


async def aggregate(state: MapReduceState) -> dict:
    return {"summary": f"Processed {len(state.processed)} items: {state.processed}"}


def dispatch(state: MapReduceState) -> list[Send]:
    # One Send per item — workers run concurrently. The ``extend`` reducer on
    # ``processed`` merges all worker outputs into one list.
    return [Send("process_item", {"item": x}) for x in state.items]


async def run_map_reduce() -> None:
    print("=== 2. Map-reduce via Send ===\n")

    pipeline = (
        PipelineBuilder("mapreduce", state=MapReduceState)
        .add_node(plan)
        .add_node(process_item)
        .add_node(aggregate)
        .add_edge(process_item, aggregate)
        .branch(plan, dispatch)
        .build()
    )
    result = await pipeline.invoke(MapReduceState(items=["alpha", "beta", "gamma", "delta"]))
    print(f"  summary: {result.state.summary}")

@workflow(name="triage")
async def triage(args, ctx):
    draft = await agent(f"draft a reply to: {args}")
    decision = await human(f"Approve this reply?\n\n{draft}")
    if decision.lower().startswith("y"):
        return draft
    return await agent(f"revise per feedback '{decision}': {draft}")