多智能体开发实战：用 LangGraph 编排搜索+代码生成+数据分析 3 类 Agent 协同工作的架构与踩坑

拆分多智能体系统的判据只有一条：任务同时满足三个条件——需要 2 种以上差异工具、子结果要被后续步骤消费、单点失败要可隔离重试——才值得拆。本文用图编排框架 LangGraph 搭一条「搜索 → 代码生成 → 数据分析」三步流水线，把状态设计、通信模式、词元成本与幻觉拦截讲透。

TL;DR

• 边界判据：任务要 2 种以上差异工具、子结果会被后续步骤消费、失败需要局部重试 → 拆；否则不拆。
• 状态传递：用共享 TypedDict 状态代替提示词拼接，可 checkpoint、可断点恢复。
• 通信选型：同进程优先共享状态；跨进程、异构服务才上消息队列。
• 幻觉拦截：幻觉会在多节点链上逐级放大，必须在关键产出后加 verifier「事实门禁」。
• 成本口径：三节点单次任务词元总量约 15–25k（估算），比单一大模型调用高 20–40%，但重试率下降后总成本更可控。

单节点还是多节点：先过决策树，再动手拆

拆分的收益是隔离复杂度，代价是同步、通信开销与调试难度。写代码前先回答 4 个问题：

1. 任务能拆成有清晰输入输出的子任务吗？拆不开的任务强行拆，只会让提示词上下文在节点间重复搬运。
2. 子任务需要差异化的工具或上下文窗口吗？搜索要联网工具，代码生成要长输出，数据分析要执行环境——三者工具集完全不同，这才是拆的硬理由。
3. 中间结果会被后续步骤消费吗？搜索结果要喂给代码生成，代码要喂给分析，形成链条才有编排价值。
4. 失败需要局部重试吗？搜索失败重搜 3 次，不该把整条流水线重跑一遍。

反例同样明确：写邮件、翻译、单轮问答这类原子任务，拆成多节点只会增加延迟和成本，没有任何收益。

市场数据也印证这个判断。Presenc AI 对 2026 年生产系统的调研显示，约 38% 的生产级多智能体系统选择该框架作为编排层，其次是自研编排（约 28%）与 CrewAI（约 12%）。报告同时提醒：框架选型远不如模型选择、评估设施和人工检查点设计重要——多数生产事故出在缺少评估与人工兜底，而不是选错了图框架。

StateGraph 编排：搜索 → 代码生成 → 数据分析的代码骨架

该框架的核心是状态图：所有节点共享一份 TypedDict 上下文，每个节点从上下文中读输入、把输出写回，边的条件决定下一个节点。三个角色职责不同：搜索节点只负责把用户问题转成检索词并拉取结构化结果，不碰代码；代码生成节点消费搜索结果、只产出可执行代码；数据分析节点拿到代码与测试输出，输出带数据依据的结论。职责越单一，提示词越短，越不容易互相污染。

先定义上下文与搜索节点：

from langgraph import graph as lg
from langgraph.checkpoint.memory import InMemorySaver

class TaskCtx(TypedDict):
    question: str
    search_results: list[dict]
    code: str
    test_output: str
    analysis: str
    error: str | None

def search_node(s: TaskCtx) -> dict:
    raw = TavilyClient(api_key=TAVILY_API_KEY).search(query=s.question, max_results=5)
    return dict(search_results=[dict(title=r.title, url=r.url) for r in raw.results])

再把三个节点接成图，用条件边实现失败局部重试：

g = lg.StateGraph(TaskCtx)
g.add_node(N1, search_node)
g.add_node(N2, codegen_node)
g.add_node(N3, analysis_node)
g.add_edge(N1, N2)
g.add_conditional_edges(N2, retry_if_error, {N2: N2, N3: N3})
g.add_edge(N3, END)

中断恢复是生产环境的关键能力。用 checkpointer 加 thread_id，可以在任意节点前暂停、由人工确认后继续：

app = g.compile(checkpointer=InMemorySaver(), interrupt_before=[N3])
config = dict(configurable=dict(thread_id=TASK_ID))
result = app.invoke(dict(question=QUESTION), config)
app.invoke(Command(resume=RESUME_MSG), config)

三个要点：上下文是唯一的共享事实源，不要在节点里偷偷拼全局变量；checkpointer 让整条链路可回放、可续跑；interrupt 机制把「人工确认」变成图的一等公民，而不是旁路脚本。

节点间通信 3 种模式：共享状态、消息队列、直接调用

选型建议：80% 的场景用共享状态就够了，它自带 checkpoint 与重放能力；只有某个节点要部署成独立服务（比如分析节点用 GPU 推理集群）或需要独立扩容时，才在边界上引入消息队列。直接函数调用只配出现在原型里。判断要不要上消息队列，看两点：节点是否要独立水平扩容，以及上下游是否允许异步。两个答案都是否，就继续用共享状态。

幻觉传递链：错误从搜索一路放大到结论

多节点系统最隐蔽的风险是幻觉被逐级放大：搜索节点对来源的误读，会被代码生成节点固化成代码里的硬编码假设，数据分析节点再基于错误假设输出一份「看起来严谨」的结论。单一 LLM 调用的幻觉影响一跳，三步链上会变成三跳，且中间节点越多越难定位。举例：搜索把「2026 年销量」误读成「2025 年」，代码生成把年份硬编码进 SQL，分析结果就整体偏移一年，而每一跳输出看起来都自洽。

拦截方案是加「事实门禁」：在关键产出节点之后插入一个 verifier 节点，用结构化输出逐条校验结论是否有搜索来源支撑，不通过就打回重做：

def fact_gate(s: TaskCtx) -> dict:
    verdict = llm.with_structured_output(VerifyResult).invoke(verify_req(s))
    unsupported = [c for c in verdict.claims if not c.supported]
    return dict(error=GATE_FAIL) if unsupported else {}

同类思路在行业里已有落地：NVIDIA 的 NeMo Guardrails 提供幻觉检测组件，Patronus AI 也验证过「独立 verifier 模型给生成结果打分」在事实性任务上的有效性。通用原则就三条：关键产出必须过校验节点；校验要用结构化输出，不要靠自由文本「你觉得对吗」；校验失败的路径要能自动回退到产生该产出的节点，而不是直接给用户看错误结果。落地时把校验标准写清楚：支持、反对、无证据，三类都要有判定样例，避免 verifier 自己犯含糊。

三节点协同的词元消耗与成本口径

以下是按常见模型输入输出比例估算的口径（估算值，非实测；实际以你的模型与提示词设计为准）：

单次任务总量约 15–25k 词元。以 2026 年 8 月生效的 DeepSeek 峰谷定价为例，高峰时段输出侧约 27 元/百万词元、输入侧低一个数量级，折算单次任务纯模型成本约 0.1–0.3 元（不含搜索 API 与执行环境费用）。

对比单次大模型调用「一次性全做」：单一调用省掉了上下文搬运，词元总量低 20–40%；但一旦生成代码有 bug 需要整体重跑，重试一次的成本就追平了。实测场景里，搜索失败重试 1–2 次 + 代码编译失败重试 1 次，分节点的局部重试反而比整体重跑更省钱。这也是「拆」的隐含收益：重试粒度越细，成本越可控。压成本还有三招：上下文只传摘要不传全文、固定前缀缓存、错峰执行非实时任务。

不适合拆成多节点的 3 种场景

1. 单次原子任务：一次翻译、一封邮件、一个 SQL 查询，单次调用一步完成，拆开只会把延迟和成本乘 3。
2. 毫秒级实时链路：多节点编排本身有 0.3–1s 的图调度开销，加上每个节点的 LLM 调用，端到端通常要 15–45s（估算）。网关鉴权、风控拦截这类实时路径，别放进来。
3. 没有可观测性基建的团队：多节点是黑盒叠加黑盒的灾难。没有 LangSmith 一类的 trace 工具、没有每个节点的输入输出日志，出问题时你连在哪一跳丢的事实都找不到。先建观测，再上复杂度。

常见问题

LangGraph 和 CrewAI、AutoGen 怎么选？

前者胜在状态机式编排、checkpoint 与 human-in-the-loop 机制，适合生产级、流程明确的系统；CrewAI 上手快、角色抽象简单，适合快速原型；AutoGen 的会话式多角色设计适合需要节点间自由对话的研究场景。生产系统优先前者，Presenc AI 的 2026 年调研也显示它是生产部署占比最高的框架。

多节点一定比单次调用更准吗？

不一定。准确率提升来自「每个子任务用更专注的上下文 + 失败隔离重试」，而不是节点数量本身。如果子任务之间上下文大量重叠、又没有校验节点，反而会引入幻觉放大。拆之前先过边界决策树。

共享状态和消息队列，生产环境选哪个？

同进程、强顺序依赖选共享状态，它自带 checkpoint 与重放；节点独立部署、需要各自扩容或跨语言协作时，才在边界上引入消息队列。先用共享状态跑通，再把真正需要解耦的边界抽成队列，不要一开始就上 Kafka。

多节点编排的词元成本怎么压？

三招：控制传递的内容（只传结构化摘要，不传全文）；把「重试」做成局部节点重试而非整图重跑；对高频固定任务缓存搜索结果。峰谷定价的模型（如 DeepSeek 2026 年 8 月生效的分时计价）可以把非实时任务挪到低谷时段。

节点之间怎么排查问题？

给每个节点记录输入输出的完整日志，用 LangSmith 一类的 trace 工具串起来看每跳消耗；在 fact_gate 之类的校验节点保留未通过列表，这通常是幻觉发生的第一现场。先确认「哪一跳产出错了」，再回头改该节点的提示词或数据源。强烈建议上线前先跑 20–50 个真实用例把各节点的典型失败模式记下来，形成排查清单。

参考

• Presenc AI：Multi-Agent Orchestration Frameworks 2026（框架生产部署占比、选型建议）
• 框架官方文档（状态图、checkpoint、human-in-the-loop 特性）
• Tavily 搜索 API（示例代码中的搜索工具）
• NVIDIA NeMo Guardrails（幻觉检测与护栏参考实现）
• 站内关联：多智能体编排实战：2026 年生产级协作的 4 种架构模式与故障排查