Per-task failure isolation

agent/graph.py
python
async def _run_plain(text: str, tasks: list[str]) -> dict:
    out: dict = {"errors": {}}
    for name in tasks:
        fn = TASK_REGISTRY[name]
        try:
            out[name] = await fn(text)
        except Exception as e:
            out["errors"][name] = str(e)
    return out

Every task runs in its own try/except. An exception in emotion never touches sentiment. The error message goes into out["errors"][name] and the loop continues.

agent/graph.py
python
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

Same pattern inside the LangGraph node. The state merge means errors accumulate across nodes without overwriting each other. The graph keeps running after a node fails.

Checkpoint: Resilient pipeline checkpoint

Loading practice…