第 08 章 — Agent 化编排

作者：袖梨 2026-07-14

第 08 章 — Agent 化编排

上一章的痛点

到第 07 章，功能已经很完整了。但看看我们的代码 —— 流程逻辑散落在各个 if/else 里：

澄清判断在 ConversationalAgent.chat；
重试循环在 SelfCorrectingAgent.ask 的 for 里；
错误分支、软失败、降级……全靠手写控制流拼起来。

这在功能上没问题，但有三个痛点：

看不清全貌：流程藏在代码里，想画出"到底有哪些状态、怎么流转"很费劲。
难扩展：想在"生成 SQL 前"加一个"权限检查"节点，得改好几处控制流。
难观测/中断：想知道卡在哪一步、想在"执行前"暂停等人确认（Human-in-the-Loop），手写循环里很别扭。

这一章我们不加新功能，而是把流程显式化为状态图。这是从"脚本"到"Agent 架构"的关键一跃。

核心原理：把流程建模成状态图

Agent 的运行本质是状态在节点间流转。用图来描述：

State（状态）：贯穿全程的数据（问题、schema、sql、错误、尝试次数、结果…）。
Node（节点）：一个处理步骤（澄清、检索、生成、校验、执行、诊断、解读）。
Edge（边）：节点间的流转，可以是条件边（根据状态决定去哪）。

我们第 01 章画的流程图，现在要变成可执行的代码。

 复制代码              ┌─────────┐
   问题 ─────▶│ clarify │──模糊──▶ [END: 追问]
              └────┬────┘
                   │清晰
                   ▼
              ┌─────────┐   ┌──────────┐   ┌──────────┐
              │ retrieve│──▶│ generate │──▶│ validate │
              └─────────┘   └──────────┘   └────┬─────┘
                                 ▲               │
                          诊断后重试              │通过
                                 │               ▼
                            ┌────┴────┐     ┌──────────┐
                            │ diagnose│◀────│ execute  │
                            └─────────┘ 失败└────┬─────┘
                                                 │成功
                                                 ▼
                                            [END: 结果]

层级设计：图是 L3 的新形态

LangGraph 图就是 L3 Agent 层，只是从"手写循环"换成了"声明式状态机"。节点内部仍调用 L2 的能力（validate/execute），L4 的澄清成为图的一个入口节点。分层没变，编排方式变了。

定义状态

 复制代码pip install langgraph

graph_state.py：

 复制代码"""L3: 问数状态图的状态定义。"""
from typing import TypedDict, Optional
class QueryState(TypedDict, total=False):
    # 输入
    question: str
    resolved_question: str
    context: str
    # 中间产物
    schema_text: str
    examples_text: str
    sql: str
    # 控制
    attempts: int
    max_attempts: int
    error: str
    diagnosis_hint: str
    # 输出
    clarify_question: str    # 需要追问时填充
    columns: tuple
    rows: tuple
    done: bool

State 用 TypedDict，节点每次返回要更新的字段（LangGraph 帮你合并），符合"不改旧值、返回增量"的不可变思路。

定义节点

每个节点是一个纯函数 State -> 部分State。复用前几章的模块。

nodes.py：

 复制代码"""L3: 状态图的各个节点 —— 复用前面章节的 L2/L4 能力。"""
from schema import extract_schema, render_schema
from retriever import SchemaRetriever
from example_retriever import ExampleRetriever, render_examples
from validator import validate
from executor import execute
from diagnosis import diagnose
from clarify import clarify
import anthropicclient = anthropic.Anthropic()
_schema_retriever = SchemaRetriever(extract_schema("drive.db"))
_example_retriever = ExampleRetriever()
def clarify_node(state):
    c = clarify(state["question"], state.get("context", "（无历史）"),
                "路测库：车辆/场景/路测/接管明细")
    if not c.clear:
        return {"clarify_question": c.question_to_user, "done": True}
    return {"resolved_question": c.resolved_question}
def retrieve_node(state):
    q = state["resolved_question"]
    return {
        "schema_text": render_schema(_schema_retriever.retrieve(q, k=5)),
        "examples_text": render_examples(_example_retriever.retrieve(q, k=2)),
        "attempts": 0,
    }
def generate_node(state):
    prompt = f"""# 表结构
{state['schema_text']}
# 相似示例（遵循口径）
{state['examples_text']}
# 问题
{state['resolved_question']}
"""
    # 若上一轮有诊断，带上修正提示
    if state.get("diagnosis_hint"):
        prompt += f"n# 上次错误n{state['diagnosis_hint']}n请修正。"    resp = client.messages.create(
        model="claude-opus-4-8", max_tokens=600,
        system="只输出一条 SQLite SELECT。",
        messages=[{"role": "user", "content": prompt}],
    )
    return {"sql": resp.content[0].text.strip(),
            "attempts": state.get("attempts", 0) + 1}
def validate_node(state):
    v = validate(state["sql"])
    if not v.ok:
        return {"error": v.reason, "diagnosis_hint": diagnose(v.reason).hint}
    return {"sql": v.normalized_sql, "error": ""}
def execute_node(state):
    r = execute(state["sql"])
    if not r.ok:
        return {"error": r.error, "diagnosis_hint": diagnose(r.error).hint}
    if len(r.rows) == 0:
        return {"error": "empty",
                "diagnosis_hint": diagnose("", row_count=0).hint}
    return {"columns": r.columns, "rows": r.rows, "error": "", "done": True}

定义条件边与组图

条件边是图的灵魂 —— 它把"重试上限""成功/失败分支"这些控制逻辑从代码里提取成显式的路由函数。

build_graph.py：

 复制代码"""L3: 组装问数状态图。"""
from langgraph.graph import StateGraph, END
from graph_state import QueryState
from nodes import (clarify_node, retrieve_node, generate_node,
                   validate_node, execute_node)
def route_after_clarify(state) -> str:
    return END if state.get("done") else "retrieve"
def route_after_validate(state) -> str:
    return "execute" if not state.get("error") else "check_retry"
def route_after_execute(state) -> str:
    if not state.get("error"):
        return END                       # 成功
    return "check_retry"                 # 失败/空 → 判断能否重试
def check_retry(state):
    """纯路由节点：不改状态，只决定重试还是放弃。"""
    return {}
def route_retry(state) -> str:
    if state.get("attempts", 0) >= state.get("max_attempts", 3):
        return END                       # 到上限，优雅放弃
    return "generate"                    # 带着 diagnosis_hint 重新生成
def build():
    g = StateGraph(QueryState)
    g.add_node("clarify", clarify_node)
    g.add_node("retrieve", retrieve_node)
    g.add_node("generate", generate_node)
    g.add_node("validate", validate_node)
    g.add_node("execute", execute_node)
    g.add_node("check_retry", check_retry)    g.set_entry_point("clarify")
    g.add_conditional_edges("clarify", route_after_clarify)
    g.add_edge("retrieve", "generate")
    g.add_edge("generate", "validate")
    g.add_conditional_edges("validate", route_after_validate)
    g.add_conditional_edges("execute", route_after_execute)
    g.add_conditional_edges("check_retry", route_retry)
    return g.compile()
if __name__ == "__main__":
    app = build()
    result = app.invoke({
        "question": "中止率最高的城市", "max_attempts": 3,
    })
    if result.get("clarify_question"):
        print("需要澄清:", result["clarify_question"])
    elif result.get("rows"):
        print("SQL:", result["sql"])
        print("结果:", result["rows"])
    else:
        print("未能回答，错误:", result.get("error"))

功能和第 06/07 章完全一样，但现在流程是声明式的 —— 一眼看清所有状态和流转，加节点只需 add_node + 连边。

白赚的好处

重构成图后，几个之前很难做的能力几乎白赚：

可观测：LangGraph 原生支持 streaming/追踪每个节点的输入输出 —— 第 10 章监控的现成数据源。
Human-in-the-Loop：想在 execute 前加人工确认？用 interrupt_before=["execute"] 一行搞定，无需改控制流。
可扩展：想加"权限检查"节点？add_node("authz", ...) 插在 generate 前，连两条边即可。
持久化/断点续跑：配 checkpointer 后，会话可暂停、可恢复 —— 天然支撑第 07 章的多轮。

这就是为什么生产级 Agent 几乎都用图/状态机来编排，而不是裸 while 循环。

本章小结

手写控制流能跑，但看不清、难扩展、难观测。
把 Agent 建模成 State + Node + 条件Edge 的状态图，让流程显式化。
图就是 L3 的新形态，节点内部仍复用 L2/L4 能力，分层不变。
图化后，可观测、HITL、可扩展、断点续跑几乎白赚。

【开放性问题 · 生产案例】

某团队把问数系统迁到 LangGraph 后遇到新问题：

图跑起来了，但线上偶发"卡在某节点不返回"，排查半天才定位到是 generate 节点调模型超时无重试。
产品想要"复杂问题拆成多个子查询分别执行再合并"（如"对比两个软件版本各城市的接管率"），当前单链路的图撑不住。
两个用户并发问数，发现全局的 _schema_retriever 没问题，但有人往 State 里塞了可变对象，导致偶发的串数据。

请思考：

问题 1：节点级的超时、重试、熔断该在图层面统一做，还是每个节点自己管？如何避免重复代码？
问题 2 指向子图 / 多 Agent：把"问题分解 → 并行子查询 → 结果合并"建模成图，你会怎么设计？这和多 Agent 编排有什么关系？（这是通往更复杂 Agent 系统的入口）
问题 3 又回到不可变原则：为什么 State 里放可变对象会串数据？LangGraph 的 State 合并机制下，你会立什么规矩来杜绝这类 bug？

下一章：09 — 结果解读与可视化，把裸数据变成用户看得懂、信得过的答案。

 复制代码项目地址：https://github.com/qiuqiuqiu123/text2sql-agent.git 
如果你也在做 AI Agent 编排相关的事情，欢迎交流。 
微信公众号「夜猫子agent工坊」 
感谢小木API的赞助:https://xiaomuai.cn

第 08 章 — Agent 化编排