LangGraph 实战：从硬编码调用链到可控的多智能体协作

接手一个内部知识管理平台时，我最初没有碰 LangGraph，而是用一套看起来更轻的方案：把几个 agent 封装成 Python 函数，按业务顺序依次调用。搜索结果传给写作函数，写作结果传给审校函数，审校不通过就再调一次写作函数。前两周跑得挺顺，第三周需求变了：要接入第三个数据库、增加一个“是否联系人工确认”的分支，还要在某个任务失败后能接着上一次进度继续。我盯着那几百行硬编码的调用链，意识到一个再简单不过的问题——多智能体应用真正难的不是让每个智能体能干活，而是让它们之间的分工、流转、恢复和终止变得可控。后来换到 LangGraph，整个思路才顺过来。

LangGraph 解决的不是“调用多个大模型”这件事，而是把多智能体协作从一段脆弱的过程式脚本，变成一张有状态、可恢复、可观测的图。这篇内容就围绕这个判断展开，从架构、核心组件到代码实战，再聊到落地时最容易踩的坑。

1. 为什么“多智能体”最容易被做成高维护成本的工程

1.1 多智能体不是多个函数互相调用

很多团队开始聊多智能体，实际写出来的东西却是“顺序调用多个模型”。这不能说错，但如果只是按固定顺序调一次，那确实不需要多智能体框架，一个 for 循环就够了。真正的多智能体场景，通常有两个特征。

第一，每个智能体拥有自己的上下文、目标和能力边界。比如搜索智能体负责收集信息，写作智能体负责组织内容，审阅智能体负责挑毛病。它们并不能简单用“一个 prompt 链”完全替代，因为各自的输入输出结构不同，失败模式也不同。第二，彼此之间存在复杂的流转关系：可能需要循环、并行、分支，甚至需要人工介入。这时再用硬编码的 if-else 串起来，代码会迅速失控。

从工程经验看，判断你是否真的需要多智能体框架，可以看一个问题：如果某个中间节点大幅修改了任务目标，你的流程能否在不重写调用链的前提下响应变化？如果不能，说明调用链已经把逻辑焊死了，这不是多智能体，只是几个函数的捆绑销售。

1.2 硬编码流程图的三类典型症状

我见过不少项目在早期都选择“自己写流程控制”，最后几乎都会出现三个症状。

第一个症状是状态散落。每个函数都维护自己的局部变量，搜索结果的格式、写作结果的长度、审阅意见所在的字段，全靠调用方心领神会。只要有人换了一个模型或改了一个返回结构，所有下游代码都要跟着改。

第二个症状是失败后无法恢复。一次任务要跑几十秒甚至几分钟，中途某个 API 超时导致异常退出。重启之后从头开始，前面的结果全部作废。如果任务消耗的是外部接口配额，那种浪费会非常明显。

第三个症状是流程不可观测。出了问题时只能靠每个函数自己打日志，但日志之间缺乏关联 ID。你很难回答“这一步的输入到底是哪一次搜索产生的”“审阅为什么否定这个版本”“循环是从第几轮开始偏离的”。这些问题的本质，是缺少一个显式的状态对象和一组显式的转移规则。

1.3 LangGraph 改变的是“控制流”的建模层级

LangGraph 的核心贡献，是让开发者从“写循环和调用关系”升级到“声明图结构”。它不再要求你手动维护调用状态，而是把状态定义成一个独立的 State 对象，把每一步动作定义成节点（Node），把节点之间的流转规则定义成边（Edge）。

这意味着你可以像画流程图一样设计多智能体协作，然后直接编译运行。更重要的是，LangGraph 本身内置了持久化和恢复能力。它会在每次节点执行后保存一份检查点，流程中断时，可以从最近的检查点继续。这个能力对长耗时、多分支、需要人工审核的任务价值极大。

理解了这个层级差异，后面所有组件才有意义。否则你只是在用 LangGraph 写另一种风格的脚本。

2. LangGraph 多智能体的核心组件，先理解它们再写代码

2.1 State：所有节点共享的状态对象

多智能体协作最重要的不是代码复用，而是状态一致。LangGraph 里，所有节点接收一个 State，处理后返回一个状态更新，系统会把这个更新合并到全局状态中。State 通常是一个 TypedDict，也可以使用 Pydantic 模型，关键点是字段的类型定义要明确。

实际编码时，最常见的 State 会包含这样几类内容：

• 用户原始输入，比如 query、topic。
• 中间结果，比如 search_results、draft_content。
• 控制字段，比如 iteration_count、review_passed。
• 消息列表，比如 messages，用来记录每个智能体产生的对话记录。

一个容易忽视的点是字段合并规则。默认情况下，如果节点返回一个新值，它会覆盖旧值。但 messages 这类需要不断追加的字段，必须使用 reducer，比如 operator.add。如果你忘了给 messages 设置追加逻辑，每次节点返回时旧消息就会被覆盖，智能体会失去上下文。

python复制from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    topic: str
    search_results: list
    draft: str
    review_feedback: str
    review_passed: bool
    iterations: int
    messages: Annotated[list, add_messages]

这个设计的意义在于：所有节点都从同一个 State 读取信息，也向同一个 State 写入更新。它避免了“通过函数参数逐层传递”造成的上下文散落。也正因为 State 是可序列化的数据，LangGraph 才能把它持久化到检查点里，实现恢复和回溯。

2.2 Node 和 Edge：把动作和转移分开表达

Node 是一个普通函数，输入 State，输出一个字典，字典里的键会被合并回 State。比如搜索节点可以这样写：

python复制def search_node(state):
    results = do_search(state["topic"])
    return {"search_results": results, "messages": [("assistant", f"已完成搜索，找到 {len(results)} 条结果")]}

Edge 则定义节点之间的连接。LangGraph 中有普通边和条件边。普通边表示“无条件转移”，比如写完必有审阅。条件边表示一个节点执行完后，根据当前 State 的值决定去哪个节点。

条件边是 LangGraph 最灵活的部分，也是很多“多智能体循环”的基础。它可以实现审阅不通过就回到写作节点，连续超过两轮就强制结束等策略。

python复制def should_review_pass(state):
    if state.get("review_passed"):
        return "finish"
    return "revise"

graph.add_conditional_edges("review", should_review_pass, {"finish": "__end__", "revise": "write"})

理解 Node 和 Edge 分离的价值，你就会明白为什么 LangGraph 比普通循环脚本更适合多智能体：动作只负责“做什么”，转移规则只负责“下一步去哪”。两者解耦后，后续增加一个节点或调整一个分支，只需要改图和条件函数，不需要翻整个调用链。

2.3 Checkpointer：让流程支持断点、恢复和人工审核

多智能体任务往往不是一次函数调用，而是一个持续过程。LangGraph 的 Checkpointer 负责在每个步骤完成后保存状态快照。这个机制带来的能力很实在。

第一，流程中断后可以从最近保存点恢复，不需要从头再来。第二，可以在节点之间穿插人工审核，比如“生成方案”完成后暂停，等人工确认后再继续下一步。第三，可以回看历史状态，甚至把流程从一个检查点重新派生。

在 LangGraph 里，使用 Checkpointer 需要安装对应的存储后端，比如内存版或 SQLite/Postgres 版本。常用写法是编译时传入 checkpointer。

python复制from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

需要提醒的是，内存版只适合开发测试，生产环境要使用持久化存储，否则进程一重启检查点就丢了。如果你要支持长时间运行、多实例并发，更需要数据库级别的检查点后端。

2.4 多智能体协作的三种常见模式

看官方文档和社区实践，LangGraph 里多智能体协作大致可以归为三种模式。

第一种是 Supervisor 模式。有一个主智能体作为调度者，负责分析任务、分配给子智能体、汇总结果。这种方式适合目标不明确、需要动态拆解的任务，但调度者本身会有额外 token 开销，也可能成为瓶颈。

第二种是图内边连接模式。所有智能体都是图上的节点，通过普通边和条件边直接形成工作流。适合流程相对固定的场景，比如“搜索→写作→审阅→是否返回修改”。它最可控，也最容易调试。

第三种是层级团队模式。每个子图本身就是一个多智能体系统，再通过一个更大的图把它们组合起来。适合复杂组织架构，但层级过多会增加调试和理解成本。

三种模式没有绝对的优劣。我更建议从第二种模式入手，因为它性问题最少。当你发现单个图太复杂、节点关系混乱时，再考虑用 Supervisor 或拆子图。

下面用一个具体的三智能体报告生成系统，把这套组件串起来。

3. 代码实战：一个三智能体报告生成系统

3.1 先想清楚场景，再动手编码

这里我们做一个演示场景：用户输入一个主题，系统依次经历“搜索资料→撰写初稿→审阅修改→通过或返回修改”的过程。三个智能体分别为搜索智能体、写作智能体、审阅智能体。搜索智能体不直接调用外部搜索，而是从预先准备的本地知识库中匹配相关内容，这样示例不需要外部 API 也能跑通。写作智能体负责生成初稿。审阅智能体负责检查内容是否符合要求，如果不符合，就带着审阅意见重新回到写作节点修改。

这个场景覆盖了 LangGraph 最核心的用法：多节点协作、状态共享、条件边和循环控制。

3.2 环境准备

需要先安装 langgraph 和相应的模型接口库。生产项目中通常还会用到 langchain-openai 或其他供应商适配包。示例结构如下：

bash复制pip install langgraph langchain-core

如果你要调用 OpenAI 风格接口，再安装：

bash复制pip install langchain-openai

然后设置环境变量：

bash复制export OPENAI_API_KEY="your-api-key"

这里要特别说明：LangGraph 本身不限定模型供应商。你只要提供能被 LangChain 语言模型接口包裹的模型即可。这意味着你可以接入 OpenAI、DeepSeek、通义、本地 Ollama 等不同来源。开发阶段建议先用一个便宜、响应快的模型，先跑通流程，再换更强模型做效果验证。

3.3 定义状态和节点函数

我们先定义状态：

python复制from typing import Annotated, TypedDict, List
from langgraph.graph.message import add_messages

class ReportState(TypedDict):
    topic: str
    source_chunks: List[str]
    draft: str
    review_comment: str
    passed: bool
    iteration: int
    messages: Annotated[List, add_messages]

然后模拟一个本地资料库搜索函数。不用真实网络请求，只是为了演示节点如何返回值：

python复制mock_knowledge = {
    "langgraph": ["LangGraph 是基于图的 agent 编排框架", "核心组件包括 State、Node、Edge、Checkpointer"],
    "agent": ["智能体能调用工具并自主决策", "多智能体需要共享状态和控制流程"],
}

def search_node(state: ReportState):
    topic = state["topic"]
    chunks = mock_knowledge.get(topic, ["暂无资料，请用通用知识生成内容"])
    return {
        "source_chunks": chunks,
        "messages": [("assistant", f"搜索完成，找到 {len(chunks)} 个片段。")],
    }

写作节点调用一个语言模型，把搜索资料和要求写入提示词：

python复制def write_node(state: ReportState):
    topic = state["topic"]
    chunks = "\n".join(state["source_chunks"])

    prompt = f"""主题：{topic}
参考材料：
{chunks}
请写一份简洁的技术报告，包含概述、现状、关键实践。"""
    response = chat_model.invoke(prompt)
    return {
        "draft": response.content,
        "messages": [("assistant", "初稿已生成。")],
    }

审阅节点同样调用语言模型，输出是否通过和修改意见：

python复制def review_node(state: ReportState):
    draft = state["draft"]
    prompt = f"""请审阅以下报告，指出逻辑、结构、事实方面的问题。
如果可以直接通过，输出：PASS
如果需要修改，输出：NEED_REVISION
然后换行给出具体修改意见。

报告：
{draft}"""
    response = chat_model.invoke(prompt)
    content = response.content
    passed = content.startswith("PASS")
    return {
        "passed": passed,
        "review_comment": content,
        "messages": [("assistant", "审阅完成。")]
    }

这里注意：chat_model 需要在函数外部定义。编写大型项目时，我更建议把模型实例放到依赖注入或全局配置中，不要在节点内部反复创建。

3.4 构建图：添加节点、普通边、条件边

接下来把节点加入图：

python复制from langgraph.graph import StateGraph, START, END

graph = StateGraph(ReportState)
graph.add_node("search", search_node)
graph.add_node("write", write_node)
graph.add_node("review", review_node)

graph.add_edge(START, "search")
graph.add_edge("search", "write")
graph.add_edge("write", "review")

现在添加审阅后的条件边：如果通过，进入 END；如果没通过，回到 write，但需要限制最大循环次数。

python复制MAX_ITERATIONS = 3

def review_router(state: ReportState):
    if state.get("passed"):
        return "pass"
    if state.get("iteration", 0) >= MAX_ITERATIONS - 1:
        return "pass_with_warning"
    return "revise"

def update_iteration(state):
    return {"iteration": state.get("iteration", 0) + 1}

我们可以在 review 节点返回时手动累加 iteration，也可以在 review 前加一个变量。为了清楚，修改 review_node 返回：

python复制    return {
        "passed": passed,
        "review_comment": content,
        "iteration": state.get("iteration", 0) + 1,
        "messages": [("assistant", "审阅完成。")],
    }

然后添加条件边：

python复制graph.add_conditional_edges(
    "review",
    review_router,
    {
        "pass": END,
        "pass_with_warning": END,
        "revise": "write",
    },
)

最后编译：

python复制app = graph.compile()

3.5 运行并检查输出

调用和正常的链式调用类似：

python复制result = app.invoke({"topic": "langgraph"})
print(result["draft"])
print(result["passed"])
print(result["iteration"])

如果要看到每个步骤的状态，可以使用流式执行或带回调的版本。LangGraph 支持 stream，它会产生按步执行的事件。开发阶段建议开启，这样可以快速确认每一步的输入输出是否符合预期。

python复制for event in app.stream({"topic": "langgraph"}, stream_mode="values"):
    if "draft" in event:
        print("当前草稿长度：", len(event["draft"]))
    if "passed" in event:
        print("是否通过：", event["passed"])

这样一个基本的多智能体循环就完成了。你看代码量并不大，但控制流已经比一层层函数调用清晰很多：你可以在 graph 对象上查看它的结构，可以单独调试某个节点，甚至随时插入一个人工审核节点。

4. 真正落地时的五个易错点

4.1 条件边里的“状态未更新”陷阱

条件边函数读取的状态，是当前节点返回并被合并后的状态。如果你在某节点忘记返回某个字段，条件边读取到的就是旧值。比如上面代码里，如果 review_node 忘记返回 passed，review_router 可能会一直走到 revise，造成死循环。

排查这类问题的思路很简单：在条件边函数第一行打印整个 state，或使用事件流观察每一步的状态。看到“当前步骤没有预期字段”时，优先检查节点返回值，而不是检查边配置。

4.2 循环没有终止条件

多智能体系统里最危险的错误就是无限循环。LangGraph 默认会执行到没有可走的边或到达 END，但如果条件边永远把你带回某个节点，它就会一直跑下去。

控制手段通常有三种：

• 在状态里维护 iteration 字段，超过阈值强制 END。
• 在条件边里检查内容质量分数，比如“连续两次修改意见相同”就停止。
• 在编译或调用时设置整体执行步数上限，比如 recursion_limit。

python复制result = app.invoke(input_data, config={"recursion_limit": 30})

一旦触发 recursion_limit，LangGraph 会抛出异常，这也算兜底。但最好的做法还是业务逻辑里有明确终止条件。

4.3 消息列表无限膨胀

多智能体协作中，messages 字段通常使用 add_messages 追加。但一轮轮对话下来，这个列表会越来越大，最终把模型输入上下文撑爆。

常用的处理策略包括：

• 只保留最近 N 轮对话，更早的进行摘要。
• 把中间过程放到 State 其他字段里，messages 只存“决策要点”。
• 定期压缩 prompt：把之前的结论汇总后，替换掉原始长文本。

这不是 LangGraph 的问题，而是所有多轮 Agent 都会遇到的上下文管理问题。设计 State 时就要想清楚 messages 的“生命周期”。

4.4 恢复与重试不能简单重放

Checkpointer 可以恢复流程状态，但外部副作用不会自动回滚。比如搜索节点调用了一个第三方 API，检查点恢复时可能再次调用；写作文本已经发送到审批系统，恢复后可能会重复发送。

因此，只要节点涉及不可自然幂等的操作，就要设计幂等键，或者把外部调用拆成“准备数据”和“提交执行”两步。恢复执行时，优先复用已经产生的结果，而不是盲目重放。

4.5 并发与资源控制

多智能体不是越快越好。当多个节点并行运行甚至多个用户同时请求时，模型 API 的速率限制、数据库连接、token 消耗都会成倍增加。

LangGraph 本身支持并行分支，但你要在代码层面控制并发度。可以在节点内部使用信号量限制模型请求并发，也可以在外部对请求进行排队。建议上线前做一个小规模压测，看速率限制下哪个节点最容易被触发限流。

5. 排查链路：从“跑不通”到“能稳定跑”

5.1 按输入、状态、图、日志四步排查

遇到 LangGraph 流程异常，不要急着改代码，按这个顺序排查。

先看输入：你的初始状态是否包含所有必要字段？比如 topic 没有传，搜索节点就会报 KeyError。再看状态：在事件流里输出每一步的 State，确认每一步更新是否覆盖了预期字段，有没有字段意外丢失。接着看图：使用可视化或 app.get_graph().draw_mermaid() 检查边是否指错节点、条件边映射表是否有遗漏。最后看日志：如果模型调用报错，优先确认是 API 返回格式异常，还是提示词触发了内容过滤。

5.2 利用图可视化和状态历史

开发阶段最有用的两个调试工具：

• app.get_graph().draw_mermaid() 生成流程可视化，直接看到节点连接是否符合设计。
• 如果用了 Checkpointer，可以通过 app.get_state(config) 查看当前检查点状态，通过 app.get_state_history(config) 查看历史状态，定位流程走到哪一步出错。

这是一个很符合直觉的调试方式：你不再是在一堆函数间跳转，而是看着流程走到某个节点后状态不对，回头改那个节点。

5.3 有这样的判断：什么时候别用 LangGraph

不是所有 Agent 应用都需要 LangGraph。如果你的任务真的是“一次模型调用+一个工具结果”，直接写个函数更省事。如果只是固定顺序链路，没有分支、循环、恢复需求，普通 LangChain 链也够。LangGraph 最大的价值来自复杂的“非确定流转”。你越需要动态决策、循环修订、人工介入，LangGraph 越能体现优势。

什么时候我推荐先别引入？第一，团队刚刚接触 Agent，还在试 prompt 阶段。第二，流程固定且只有三五个节点。第三，没有持久化需求，也没有恢复场景。这些情况下引入 LangGraph 会平白增加学习成本。

5.4 把原型变成工程的检查清单

一个带 LangGraph 的多智能体应用，如果要放进生产环境，至少要补齐下面这些能力。

• 日志：每个节点都能输出结构化日志，包含请求 ID、节点名、耗时、token 消耗。
• 超时：模型调用和图执行都设置超时时间，避免一直挂起。
• 重试：区分“可重试错误”和“不可重试错误”，不要盲目重试。
• 安全校验：对模型返回值做格式校验，防止内容或字段异常导致后续节点崩溃。
• 版本管理：图结构、提示词、模型版本都要纳入版本控制，否则很难复现线上问题。
• 评估集：至少准备 10 到 20 条典型输入，建立多轮回归测试，防止改一个节点后其他节点效果下降。

做到这些，LangGraph 才真正成为工程化工具，而不是又一个 demo 框架。

写到最后，我还是想回到最初的判断：LangGraph 真正值得学的地方，不是它的代码有多华丽，而是它逼着你用“图”的视角思考多智能体协作。流程、状态、决策点一目了然，失败和恢复有迹可循，扩展新节点时心里有底。下一次接到“多个智能体配合完成复杂任务”的需求，你也可以先把流程画成一张图，再让 LangGraph 帮你把这张图变成可运行的程序。