FastAPI + LangGraph：把 Agent 编排拆成可观测的节点

早期写 Agent 的时候，一个请求进来就是一串函数从头跑到尾，出问题只能靠打日志，从一堆输出里翻哪一步产生了错误的输入。后来换成状态图之后，最大的收获不是流程变好看，而是每一步都有了明确的边界，可以被独立观察、独立重试。

节点粒度：一个节点只做一件事

判断粒度是否合适，我用的标准很简单：这个节点出问题时，能不能用一句话描述它做错了什么。如果描述里出现「并且」，就该拆开。典型拆分：

• 解析输入、校验参数 —— 一个节点
• 决定调用哪个工具 —— 一个节点
• 执行工具、归一化结果 —— 一个节点
• 生成最终回复 —— 一个节点

拆开的好处是重试的范围可以精确控制：工具超时只需要重跑执行节点，不用重跑整个规划。

状态里放什么

状态对象是节点之间唯一的契约，我倾向只放两类东西：业务数据本身，以及可复现所需的元信息。不放临时变量，也不放已经格式化好的展示文本。

class AgentState(TypedDict, total=False):
    session_id: str
    round: int
    messages: list[dict]
    tool_calls: list[dict]
    tool_results: list[dict]
    error: str | None
    trace: list[dict]

trace 字段的设计

每个节点执行结束后追加一条记录，字段固定下来，不要随意增减：

• node：节点名，和代码里的注册名保持一致
• started_at / elapsed_ms：耗时，判断性能瓶颈的唯一依据
• input_keys / output_keys：只记键名不记全文，避免状态膨胀
• tokens：模型调用的 token 用量，成本统计靠它
• error：出错时的类型和简短信息

只记键名的做法一开始觉得信息不够，实际排查时配合结构化的日志系统，比把整个状态 dump 进去更好用——后者通常大到没法读，还容易带出敏感数据。

重试该放在哪

放在节点内部的重试适合处理网络抖动、限流这类瞬时错误，因为它不需要改变图的结构。但要跨节点重试（比如工具执行失败后让规划节点重新决策），就必须把中间状态完整持久化，否则回放会从错误的起点开始。

实践中我是这样分的：节点内重试 2 次处理瞬时错误，仍然失败则写 error 并走条件边回到规划节点，规划节点看到 error 才会换工具。这样图本身永远只有一条主线，重试逻辑不会散落在各处。

接 FastAPI 时注意什么

状态图是有状态的，而 HTTP 请求是无状态的，中间需要一层会话存储。我把状态按 session_id 存进 Redis，请求进来先取、执行完写回，超时时间设得比对话超时略长。要留意并发：同一个 session 的两个请求同时进来会互相覆盖状态，需要在会话上加锁，或者干脆串行化同一个会话的请求。