05 · Agent 编排：LangGraph 状态图与降级内核

对应代码：app/agent/graph.py、state.py

一、为什么不是“一次 LLM + Function Calling”

单轮 Function Calling 有三个硬伤：

1. 只能调用一次工具：拿到结果后不能再触发新的查询。
2. 工具报错就整轮失败：没有自愈机会。
3. 复杂问题做不完：例如“先查制度阈值 → 再查实际数据 → 对比 → 给建议”。

所以引入 Plan → Execute → Reflect → Answer 流程。

二、四个节点在做什么

节点	输入	输出	关键点
Plan	问题 + 工具清单 + 历史摘要	JSON 步骤（1～3 步）	温度 0、强制 JSON、容错解析
Execute	步骤列表	observations	逐个执行；报错不抛异常，而是写入 observation
Reflect	observations	enough / next_steps	判断信息是否足够；不足时最多补充 2 步
Answer	context + observations	流式答案	引用编号 + 口径说明

Plan 节点

python prompt = PLANNER_TMPL.format( system_role=SYSTEM_ROLE, tools=self.registry.describe_for_prompt(),
summary=state["summary"], question=state["query"]) raw = llm.chat(messages, json_mode=True) data = parse_json_safe(raw) plan = [s for s in data["steps"] if self.registry.get(s["tool"])][:3]

两个细节：

• 工具清单自动生成：新增工具时无需修改 Prompt。
• 过滤未知工具：模型偶尔会“幻觉”出一个不存在的工具名。

Execute 节点

python for step in plan: result = self.registry.execute(name, args, trace_id) if result.error == "HITL_REQUIRED": return {"status": "waiting_human", "pending_hitl": {...}}
observations.append({"tool": name, "ok": result.ok, "output": ...})

关键决策：工具报错不抛异常，而是变成一条 observation 交给模型。 这样模型能够看到“SQL 字段不存在”等错误，并自行尝试另一种查询写法。

Reflect 节点

{"enough": true/false, "missing": "...", "next_steps": [...]}

“让模型判断信息是否足够”这件事看起来简单，实际是提升准确率最明显的一环。

三、用 LangGraph 建图

python g = StateGraph(AgentState) g.add_node("plan", self._node_plan) g.add_node("execute", self._node_execute) g.add_node("reflect", self._node_reflect) g.add_node("answer", self._node_answer)

g.set_entry_point("plan") g.add_conditional_edges("plan", self._route_after_plan, {"execute": "execute", "answer": "answer"}) g.add_edge("execute", "reflect") g.add_conditional_edges("reflect", self._route_after_reflect, {"execute": "execute", "answer": "answer"}) g.add_edge("answer", END)

路由函数：

python @staticmethod def _route_after_plan(state): return "execute" if state.get("plan") else "answer"

def _route_after_reflect(self, state): if state.get("status") == "answer_ready": return "answer" if state.get("iteration", 0) >= settings.max_iterations: return "answer" return "execute" if state.get("plan") else "answer"

为什么 State 使用 TypedDict 而不是 Pydantic

LangGraph 会对 dict 做增量合并，序列化成本较低，整份 state 可以直接落盘以支持回放。

python class AgentState(TypedDict, total=False): query, trace_id, thread_id summary, history context, citations, docs plan, observations, iteration pending_hitl, hitl_approved final_answer, error, status

四、防死循环：三道闸

这是血泪经验。 没有上限的 Agent 在“工具持续返回空结果”时可能无限循环，一晚上消耗数百元 token。

python MAX_ITERATIONS = 6
Reflect 最多补 2 步 Plan 最多 3 步

五、降级内核：不依赖框架也能运行

python def _build_langgraph(self): try: from langgraph.graph import END, StateGraph except Exception: return None
...

def _execute_graph(self, state): if self.graph is not None: try: return self.graph.invoke(dict(state)) except Exception as e: self._emit({"type": "warn", "message": f"langgraph 异常，降级：{e}"})

cur.update(self._node_plan(cur))
while self._route_after_plan(cur) == "execute":
    cur.update(self._node_execute(cur))
    if cur.get("status") == "waiting_human": return cur
    cur.update(self._node_reflect(cur))
    if self._route_after_reflect(cur) != "execute": break
cur.update(self._node_answer(cur))

为什么值得写这 20 行：

1. 即使环境无法安装 langgraph，也能运行，例如 CI 或内网环境。
2. 面试时最能说明你不是只会调库——你可以清楚解释“状态图”和“手写循环”之间的等价关系。
3. 当框架出现 bug 时可以自动降级，为生产环境增加一层保险。

六、Human-in-the-loop：中断而不是拒绝

高风险动作的处理流程：

text Execute 遇到 risk=high 的工具 │ ├─ 未确认 → 记入 self._pending[thread_id]，状态改 waiting_human，路由到 Answer │ → 前端弹出确认卡片 │ └─ 用户点确认 → 带 approved=true 再次请求 → _resume_after_approval() 取出 pending 并执行 → 继续走到 Answer

python def _resume_after_approval(self, state): pending = self._pending.pop(state["thread_id"]) result = self.registry.execute(pending["tool"], pending["args"], state["trace_id"], auto_approve_high_risk=True) state["observations"].append({...}) state["status"] = "answer_ready"

七、流式输出：如何做到“边想边说”

• 规划、工具调用、反思等过程事件，通过 sink 回调实时推送至 SSE。
• 最终答案使用 llm.stream() 逐 token 推送：

python for delta in get_llm().stream([...], trace_id=trace_id): chunks.append(delta) self._emit({"type": "token", "delta": delta})

前端因此可以同时展示“执行链路时间线”和“打字机式答案”。

八、一次真实执行（DeepSeek 实测）

text 用户：本月各项目成本排名前 3 是多少？

[plan] sql_query(question="本月各项目成本排名前3") [tool] sql_query ✅ → 返回空（月份格式问题） [reflect] enough=false, missing="需要确认月份字段格式" [plan] sql_query(question="最近有数据的月份各项目成本合计排名前3") [tool] sql_query ✅ → 3 行数据 [reflect] enough=true [answer] 结论：2026年9月成本 Top3 为 海外支付网关(80.26)、数据资产平台(77.94)… 说明：直接查询“本月”返回空行 [2][3]，可能因月份字段格式为 '2026-09-01'…

注意最后一段——模型自行解释了数据异常的原因。这正是 Reflect 节点的价值：它不只是“再调用一次工具”，而是让整个回答更可信。

---

下一篇：06 · 工具调用：Text2SQL 与四层护栏

对应代码：app/agent/graph.py、state.py

一、为什么不是"一次 LLM + Function Calling"

单轮 Function Calling 的三个硬伤：

1. 只能调用一次工具 ：拿到结果后不能再触发新的查询
2. 工具报错就整轮失败 ：没有自愈机会
3. 复杂问题做不完 ：比如"先查制度阈值 → 再查实际数据 → 对比 → 给建议"

所以引入 Plan → Execute → Reflect → Answer 。

二、四个节点在干什么

节点	输入	输出	关键点
Plan	问题 + 工具清单 + 历史摘要	JSON 步骤（1~3 步）	温度 0、强制 JSON、容错解析
Execute	步骤列表	observations	逐个执行；报错不抛异常 ，写进 observation
Reflect	observations	enough / next_steps	判断信息是否足够；不够最多补 2 步
Answer	context + observations	流式答案	引用编号 + 口径说明

Plan 节点

prompt = PLANNER_TMPL.format(
    system_role=SYSTEM_ROLE,
    tools=self.registry.describe_for_prompt(),   
    summary=state["summary"],
    question=state["query"])
raw = llm.chat(messages, json_mode=True)
data = parse_json_safe(raw)
plan = [s for s in data["steps"] if self.registry.get(s["tool"])][:3]

两个细节：

• 工具清单自动生成 ：新增工具不用改 Prompt
• 过滤未知工具 ：模型偶尔会"幻觉"出一个不存在的工具名

Execute 节点

for step in plan:
    result = self.registry.execute(name, args, trace_id)
    if result.error == "HITL_REQUIRED":
        return {"status": "waiting_human", "pending_hitl": {...}}   
    observations.append({"tool": name, "ok": result.ok, "output": ...})

关键决策：工具报错不抛异常，而是变成一条 observation 交给模型。 这样模型能看到"SQL 字段不存在"并自己换一种写法。

Reflect 节点

{"enough": true/false, "missing": "...", "next_steps": [...]}

"让模型判断信息够不够"这件事，看起来简单，实际是准确率提升最明显的一环 。

三、用 LangGraph 建图

g = StateGraph(AgentState)
g.add_node("plan", self._node_plan)
g.add_node("execute", self._node_execute)
g.add_node("reflect", self._node_reflect)
g.add_node("answer", self._node_answer)

g.set_entry_point("plan")
g.add_conditional_edges("plan", self._route_after_plan,
                        {"execute": "execute", "answer": "answer"})
g.add_edge("execute", "reflect")
g.add_conditional_edges("reflect", self._route_after_reflect,
                        {"execute": "execute", "answer": "answer"})
g.add_edge("answer", END)

路由函数：

@staticmethod
def _route_after_plan(state):
    return "execute" if state.get("plan") else "answer"

def _route_after_reflect(self, state):
    if state.get("status") == "answer_ready": return "answer"
    if state.get("iteration", 0) >= settings.max_iterations: return "answer"
    return "execute" if state.get("plan") else "answer"

为什么 State 用 TypedDict 而不是 Pydantic

LangGraph 对 dict 做增量合并 ，序列化成本低，整份 state 可以直接落盘做回放。

class AgentState(TypedDict, total=False):
    query, trace_id, thread_id
    summary, history
    context, citations, docs
    plan, observations, iteration
    pending_hitl, hitl_approved
    final_answer, error, status

四、防死循环：三道闸

这是血泪经验。 没有上限的 Agent 在"工具一直返回空"时会无限循环， 一晚上烧掉几百块 token。

MAX_ITERATIONS = 6        
Reflect 最多补 2 步
Plan 最多 3 步

五、降级内核：不依赖框架也能跑

def _build_langgraph(self):
    try:
        from langgraph.graph import END, StateGraph
    except Exception:
        return None          
    ...

def _execute_graph(self, state):
    if self.graph is not None:
        try:
            return self.graph.invoke(dict(state))
        except Exception as e:
            self._emit({"type": "warn", "message": f"langgraph 异常，降级：{e}"})
    
    cur.update(self._node_plan(cur))
    while self._route_after_plan(cur) == "execute":
        cur.update(self._node_execute(cur))
        if cur.get("status") == "waiting_human": return cur
        cur.update(self._node_reflect(cur))
        if self._route_after_reflect(cur) != "execute": break
    cur.update(self._node_answer(cur))

为什么值得写这 20 行 ：

1. 环境装不上 langgraph 也能跑（CI / 内网）
2. 面试时最能说明你不是在调库 ——你能说清"图"和"手写循环"的等价关系
3. 框架出 bug 时自动降级，生产多一层保险

六、Human-in-the-loop：中断而不是拒绝

高危动作的处理流程：

Execute 遇到 risk=high 的工具
   │
   ├─ 未确认 → 记入 self._pending[thread_id]，状态改 waiting_human，路由到 Answer
   │           → 前端弹出确认卡片
   │
   └─ 用户点确认 → 带 approved=true 再次请求
                 → _resume_after_approval() 取出 pending 并执行
                 → 继续走到 Answer



def _resume_after_approval(self, state):
    pending = self._pending.pop(state["thread_id"])
    result = self.registry.execute(pending["tool"], pending["args"],
                                   state["trace_id"], auto_approve_high_risk=True)
    state["observations"].append({...})
    state["status"] = "answer_ready"

七、流式输出：怎么做到"边想边说"

• 规划/工具/反思这些过程事件 通过 sink 回调实时推给 SSE

• 最终答案用 llm.stream() 逐 token 推：

  for delta in get_llm().stream([...], trace_id=trace_id): chunks.append(delta) self._emit({"type": "token", "delta": delta})

前端就能同时看到"执行链路时间线"和"打字机答案"。

八、一次真实执行（DeepSeek 实测）

用户：本月各项目成本排名前 3 是多少？

[plan]  sql_query(question="本月各项目成本排名前3")
[tool]  sql_query ✅  → 返回空（月份格式问题）
[reflect] enough=false, missing="需要确认月份字段格式"
[plan]  sql_query(question="最近有数据的月份各项目成本合计排名前3")
[tool]  sql_query ✅  → 3 行数据
[reflect] enough=true
[answer] 结论：2026年9月成本 Top3 为 海外支付网关(80.26)、数据资产平台(77.94)…
         说明：直接查询"本月"返回空行 [2][3]，可能因月份字段格式为 '2026-09-01'…

注意最后那段——模型自己解释了数据异常的原因 。 这就是 Reflect 节点的价值：它不只是"多调一次工具"，而是让整个回答更可信。

---

下一篇：06 · 工具调用：Text2SQL 与四层护栏