One node per task

A registry plus a for-loop is a pipeline. A StateGraph is a pipeline that understands state, edges, and branching. For five sequential tasks the difference looks cosmetic, but once you want conditional routing, retries, or parallel fan-out, the graph pays for itself. You will treat LangGraph as an optional upgrade, so the service runs either way.

LangGraph StateGraph, one node per task

Each task is a node. Edges sequence them. The final node points at END.

agent/graph.py
python
try:
    from langgraph.graph import END, StateGraph
    HAS_LANGGRAPH = True
except Exception:
    HAS_LANGGRAPH = False


async def _run_with_langgraph(text: str, tasks: list[str]) -> dict:
    State = dict

    def make_node(task_name: str):
        async def node(state: dict) -> dict:
            fn = TASK_REGISTRY[task_name]
            try:
                result = await fn(state["text"])
                return {task_name: result}
            except Exception as e:
                errs = dict(state.get("errors", {}))
                errs[task_name] = str(e)
                return {"errors": errs}
        return node

    graph = StateGraph(State)
    for name in tasks:
        graph.add_node(name, make_node(name))
    graph.set_entry_point(tasks[0])
    for i in range(len(tasks) - 1):
        graph.add_edge(tasks[i], tasks[i + 1])
    graph.add_edge(tasks[-1], END)

    compiled = graph.compile()
    final = await compiled.ainvoke({"text": text, "errors": {}})
    return final

The try/except import is the entire optional-dependency pattern. If LangGraph is installed, HAS_LANGGRAPH is True and the graph path runs. If not, the plain-async fallback runs the same tasks in the same order and produces the same shape of result.

Quiz: Quiz

Loading practice…