LangGraph多智能体实战：核心组件与工程化落地

在之前的多个项目里，我一直想用 LangGraph 把“多个大模型角色协作”这件事做成可控、可观测、能上线的工程方案，但早期资料零散，网上教程大多停在“用 LangChain 调一次模型”的程度，真正把多智能体架构、核心组件、分支路由、状态持久化讲透的内容很少。这篇文章我会从多智能体的基本概念讲起，逐个拆解 LangGraph 的 State、Node、Edge、Conditional Edge、Send、Checkpoint 等核心组件，然后给出一套可以本地运行的多智能体协作实战项目，最后整理开发中常见的问题排查思路和工程化建议，适合正在学习 AI 大模型应用开发、准备上手 LangGraph 多智能体的开发者。

1. 多智能体与 LangGraph 背景与核心概念

1.1 什么是多智能体系统

先来理解“多智能体”这个概念。在 AI 大模型领域，单个智能体通常是指“一个大模型 + 若干工具 + 一套提示词”组成的最小执行单元。你可以把它理解成一个能独立完成某类任务的数字员工。

多智能体系统，就是多个这样的数字员工组成一个团队，每个员工负责一个独立角色，比如数据分析、文案撰写、代码审查、质量检查等，然后通过消息传递、状态共享和路由调度，协同完成一个更复杂的任务。

一个最简单的多智能体协作流程是：

1. 用户提出一个复杂需求。
2. 调度者分析需求，决定先让哪个角色处理。
3. 角色 A 处理完后，把结果写入共享状态。
4. 调度者根据结果决定是否交给角色 B。
5. 所有角色完成后，汇总输出最终答案。

这种模式的好处是任务分工明确、每一步都能观察和干预、出现问题容易定位。比如“先生成报告，再检查报告质量，质量不合格就退回重写”，这种流程用单个大模型调用做出来比较笨拙，但用多智能体架构却很自然。

1.2 LangGraph 是什么

LangGraph 是 LangChain 社区推出的一个面向 LLM 应用的状态化编排框架，它的核心思路是把智能体的执行过程建模成一张“图”。

在这张图里：

• 节点表示一次计算，比如调用大模型、执行一个函数、调用一个工具。
• 边表示节点之间的流转关系，比如“A 执行完必须执行 B”。
• 状态则是整个图的共享数据，每个节点都能读取和修改状态，下一个节点能看到上一个节点的修改结果。

相比直接用 Python 代码写 if-else 来回调用大模型，LangGraph 有几点很关键：

• 天然的循环支持：智能体经常会遇到“工具调用 -> 返回结果 -> 再调用 -> 再返回”，这本质上是一个循环，LangGraph 对循环的支持非常自然。
• 可观测性：图执行过程中的每一步状态变化都能被追踪。
• 可持久化：通过 Checkpoint 机制把图的状态保存下来，支持记忆和断点恢复。
• 有向图带来的可控性：业务怎么流转，图就怎么画，不依赖模型自由发挥。

1.3 LangGraph 与 LangChain 的区别

很多刚开始接触大模型开发的同学会混淆 LangChain 和 LangGraph。从定位上讲：

• LangChain 的核心是“链”，也就是把一次提示词调用、一个工具调用、一次输出解析串成一条顺序执行的管道，适合做相对固定的流程。
• LangGraph 的核心是“图”，支持分支、循环、并行、条件路由，适合做动态、多角色、有反馈闭环的复杂编排。

可以这样简单理解：如果业务是固定的三步走，用 LangChain 就够；如果业务是“根据中间结果决定下一步去哪里”，或者需要多个智能体来回协作，LangGraph 更合适。

需要注意的是，LangGraph 并不是 LangChain 的替代品，两者可以共存。LangGraph 节点内部依然可以使用 LangChain 的模型封装、Prompt 模板、输出解析器等能力。

1.4 为什么需要 LangGraph 来构建多智能体应用

单智能体应用的逻辑通常比较简单，就是“问题 -> 模型 -> 答案”。一旦到了多智能体场景，问题就变复杂了：

• 多个智能体之间如何传递数据？
• 某个智能体失败了，是由另一个智能体重试还是直接退出？
• 是顺序执行，还是并行执行？
• 如何让模型来动态决定下一步交给哪个智能体？
• 用户多轮对话时，智能体协作的中间状态怎么保存？

这些问题本质上都是“流程控制”问题，而流程控制是图最擅长的事。LangGraph 把状态管理、路由、循环、并行、持久化这些能力都内置了，让我们可以专注于业务，而不是自己写一套状态机。

2. 环境准备与版本说明

2.1 安装依赖

在开始写代码之前，先准备 Python 环境。建议使用 Python 3.10 及以上版本，避免旧版本对类型注解和异步语法支持不完整。

创建虚拟环境后，安装以下核心依赖：

pip install langgraph langchain-core langchain-openai

如果你的环境需要访问 OpenAI 接口，还需要配置 API Key：

export OPENAI_API_KEY="你的 API Key"

如果使用的是国内模型服务或者其他兼容 OpenAI 协议的接口，可以通过 base_url 参数指定服务地址，比如：

from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="你的模型名称", api_key="你的 API Key", base_url="你的接口地址" )

为了文章示例的安全性和可复制性，我会先用模拟函数扮演大模型，跑通完整的图结构；在第四节会给出如何替换成真实大模型调用的方式。

2.2 版本差异提醒

LangGraph 目前迭代速度比较快，不同小版本之间接口细节可能不同。本文示例基于常见稳定版本的常用 API，核心组件如StateGraph、TypedDict、add_messages、conditional_edges、Send都属于主接口，在大版本内基本保持一致。

如果你用的版本较旧，部分命名可能不同，比如早期版本中图的入口方法叫set_entry_point，后续版本仍然兼容；而一些新增的快捷方法则可能只在较新版本里可用。建议安装时不要锁定过旧版本，直接安装最新版即可。

pip install --upgrade langgraph

安装完成后可以验证版本：

python -c "import langgraph; print(langgraph.__version__)"

如果执行报错，说明当前版本可能还没有暴露这个包入口，可以参考官方文档确认导入方式。

2.3 项目结构

本文的实战项目是一个多智能体协作的“情报简报生成系统”，项目结构如下：

langgraph-multi-agent-demo/ ├── main.py # 入口文件，构建图并执行 ├── state.py # 定义共享状态 ├── agents.py # 定义各个智能体节点 ├── tools.py # 定义工具函数 └── requirements.txt # 依赖清单

实际开发中，建议把状态定义、节点函数、图构建逻辑拆分成不同模块，避免把所有代码堆在一个文件里。

3. LangGraph 核心组件拆解

3.1 State：共享状态

在 LangGraph 中，State 就是图的全局状态。每个节点执行后，可以返回一个字典，LangGraph 会把字典里的字段更新到全局状态里。

State 通常用 TypedDict 或 Pydantic 模型定义。下面是一个示例：

from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): # 使用 add_messages reducer，使每次写入 messages 都是追加而不是覆盖 messages: Annotated[list, add_messages] # 用户原始问题 query: str # 分析结果 analysis: str # 写作文案 draft: str # 审核意见 feedback: str

Annotated[list, add_messages]是 LangGraph 里一个很有用的写法。默认情况下，多个节点如果返回同一个字段，后执行的节点会直接覆盖前一个节点的值。但加上add_messages后，每次写入messages字段都会自动合并到已有列表里，适合保存多轮对话记录。

理解这一点很重要：State 不只是简单字典，字段的合并策略是通过 reducer 来控制的。

3.2 Node：节点

节点就是普通 Python 函数，接收整个 State 作为参数，返回一个字典，把需要更新的字段返回给图。

def analyze_node(state: AgentState) -> dict: query = state["query"] # 这里是模拟大模型分析 analysis = f"针对问题「{query}」的分析结果：这是一个 LangGraph 多智能体示例。" return {"analysis": analysis}

节点的编写规范：

• 函数参数是 state，类型是整个 AgentState。
• 返回值必须是字典，里面包含要更新的字段。
• 不要直接修改入参 state，而是返回新字典。LangGraph 内部会负责合并状态。

在图中注册节点使用add_node：

from langgraph.graph import StateGraph graph = StateGraph(AgentState) graph.add_node("analyze", analyze_node)

3.3 Edge：普通边和条件边

普通边表示无条件流转。比如分析结束后必须写文案：

graph.add_edge("analyze", "write")

条件边表示根据当前状态动态决定下一步进入哪个节点。条件边由一个路由函数返回目标节点名称：

def route_after_write(state: AgentState) -> str: if "需要修改" in state.get("feedback", ""): return "write" # 回到写作文案节点重写 return "finish" # 质量合格，进入结束节点

注册条件边：

graph.add_conditional_edges( "write", route_after_write, { "write": "write", "finish": END } )

这里要注意：条件边函数返回的字符串必须是第三个参数映射表里的 key，否则运行时会报错。这个映射表的作用是把路由函数返回值映射到真实的节点名。

3.4 编译图与执行

所有节点和边都定义好后，需要调用compile()方法，把图编译成可执行对象：

app = graph.compile() result = app.invoke({ "query": "帮我总结 LangGraph 的核心组件", "messages": [] })

invoke的入参是初始状态字典，返回值是最终状态字典。整个过程可以用stream方法逐步观察：

for chunk in app.stream({"query": "你好"}): print(chunk)

stream每次输出一个节点执行后的状态变化，非常适合调试。

3.5 Checkpoint：持久化与记忆

Checkpoint 是 LangGraph 中实现对话记忆和断点恢复的机制。在 compile 时传入一个 CheckpointSaver，图每次执行后都会保存当前状态，下次执行时可以从某个历史状态继续。

from langgraph.checkpoint.memory import InMemorySaver memory = InMemorySaver() app = graph.compile(checkpointer=memory) config = {"configurable": {"thread_id": "user-session-001"}} result = app.invoke( {"query": "第二轮对话"}, config=config )

通过thread_id区分不同会话，同一个会话的多次执行会共享历史状态。需要注意：InMemorySaver只适合开发和测试使用，生产环境建议使用langgraph-checkpoint-postgres或langgraph-checkpoint-dynamodb等持久化存储。

3.6 Send：动态并行分支

在多智能体场景中，经常需要把一个任务拆分成多个子任务并行执行。比如要分析 5 份报告，每个报告交给同一个分析智能体并行处理。这时可以使用Send。

Send的作用是：在节点里返回多个Send对象，每个 Send 对象指定目标节点和该节点要接收的状态。

from langgraph.types import Send def dispatch_node(state: AgentState): topics = ["LangGraph", "多智能体", "AI Agent", "状态编排", "Checkpoint"] return [ Send("analyze_topic", {"topic": t}) for t in topics ]

这里每个Send("analyze_topic", {"topic": t})表示：把{"topic": t}作为初始状态，交给analyze_topic节点执行。LangGraph 会并行创建多个analyze_topic的执行分支。

Send适合任务数量在运行时才能确定的场景，比如读取了一堆文档、生成了多个子任务，和普通一次性 add_node 完全不同。

3.7 常见理解误区

• 节点返回的状态只会更新它返回的字段，不会清空其他字段。
• 如果不加 reducer，后写入的字段会覆盖前值。想追加就用Annotated搭配 reducer。
• END不是一个节点，它表示图的终点，不能给它 add_node。
• 条件边的映射表 key 要和路由函数返回值完全一致，大小写都不能错。

4. 完整实战案例：多智能体协作系统

4.1 需求与架构设计

接下来我们实现一个“情报简报生成系统”。系统的目标是根据用户提出的主题，生成一份结构完整、质量合格的技术简报。

我们设计三个智能体角色：

1. 分析师（ResearchAgent）：负责整理主题相关的分析要点。
2. 撰稿人（WriterAgent）：负责把分析要点扩展成完整简报。
3. 审核员（CriticAgent）：负责检查简报质量，决定是否通过。

完整流程如下：

1. 用户输入主题。
2. 分析师写出分析要点。
3. 撰稿人基于分析要点生成简报。
4. 审核员检查简报。
5. 如果简报不合格，回到撰稿人节点重写。
6. 如果合格，结束流程并输出最终简报。

这个案例虽然简单，但已经覆盖了 LangGraph 的 State、Node、Edge、Conditional Edge、循环、Compile、Stream 这些核心能力。

4.2 定义共享状态

文件：state.py

from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class BriefState(TypedDict): # 用户输入的主题 topic: str # 分析师的分析要点 analysis: str # 撰稿人生成的简报 brief: str # 审核员的反馈意见 feedback: str # 重写次数，避免无限循环 rewrite_count: int # 消息记录 messages: Annotated[list, add_messages]

rewrite_count字段很关键，多智能体循环中必须要有循环终止条件，否则可能陷入死循环。

4.3 模拟大模型调用的工具函数

文件：tools.py

先定义几个模拟大模型的函数。后面要接真实大模型时，只需要替换这些函数内部实现。

import random def mock_llm_analyze(topic: str) -> str: """模拟分析师对主题进行要点拆解。""" return ( f"针对《{topic}》的分析要点如下：\n" f"1. 大模型应用正在从单模型调用走向多智能体协作。\n" f"2. LangGraph 是当前主流的状态化编排框架。\n" f"3. 工程落地的关键在于状态管理、路由设计和可观测性。\n" f"4. 建议先以图思维拆解业务流程，再实现代码。" ) def mock_llm_write(analysis: str) -> str: """模拟撰稿人根据分析要点生成完整简报。""" return ( f"【技术简报】\n" f"一、背景\n{analysis}\n" f"二、核心挑战\n" f"多智能体场景下，角色分工、数据传递、流程控制都是必须解决的问题。\n" f"三、LangGraph 方案\n" f"使用状态图编排多个智能体，利用条件边实现质量反馈闭环。\n" f"四、结论\n" f"LangGraph 将复杂流程显式建模，显著提高了多智能体应用的稳定性。" ) def mock_llm_critic(brief: str) -> tuple[str, str]: """ 模拟审核员检查简报质量。 返回元组：(是否通过, 审核反馈) """ # 为了让示例更容易看到“重写”效果，这里随机返回一次不合格 if "LangGraph" in brief and random.random() > 0.3: return "pass", "内容完整，结构清晰，审核通过。" return "fail", "缺少实际代码示例，需要补充技术细节后重写。"

这里我用随机数让审核有一定概率失败，从而演示循环重写机制。实际项目中应该用真实大模型判断，或者接入规则引擎。

4.4 实现智能体节点

文件：agents.py

from state import BriefState from tools import mock_llm_analyze, mock_llm_write, mock_llm_critic def analysis_node(state: BriefState) -> dict: """分析师节点：对主题进行要点拆解。""" topic = state["topic"] analysis = mock_llm_analyze(topic) return { "analysis": analysis, "messages": [{"role": "assistant", "content": f"分析师已完成要点拆解：{analysis}"}] } def write_node(state: BriefState) -> dict: """撰稿人节点：基于分析要点生成简报。""" analysis = state.get("analysis", "") brief = mock_llm_write(analysis) return { "brief": brief, "rewrite_count": state.get("rewrite_count", 0) + 1, "messages": [{"role": "assistant", "content": f"撰稿人已完成第 {state.get('rewrite_count', 0) + 1} 版简报。"}] } def critic_node(state: BriefState) -> dict: """审核员节点：检查简报质量。""" brief = state.get("brief", "") verdict, feedback = mock_llm_critic(brief) return { "feedback": feedback, "messages": [{"role": "assistant", "content": f"审核结果：{verdict}，{feedback}"}] } def route_after_critic(state: BriefState) -> str: """ 审核后路由： 如果审核未通过且重写次数未超过上限，回到撰稿人节点重写； 否则进入结束节点。 """ feedback = state.get("feedback", "") rewrite_count = state.get("rewrite_count", 0) if "通过" in feedback: return "finish" if rewrite_count >= 3: return "finish" return "rewrite"

注意route_after_critic返回的是映射表的 key，不是直接返回节点名。上一节我们提过这个细节，这里再次验证。

4.5 构建图

文件：main.py

from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import InMemorySaver from state import BriefState from agents import analysis_node, write_node, critic_node, route_after_critic def build_graph(): graph = StateGraph(BriefState) # 注册节点 graph.add_node("analysis", analysis_node) graph.add_node("write", write_node) graph.add_node("critic", critic_node) # 设置入口节点 graph.set_entry_point("analysis") # 固定顺序边 graph.add_edge("analysis", "write") graph.add_edge("write", "critic") # 条件边：审核员决定是重写还是结束 graph.add_conditional_edges( "critic", route_after_critic, { "rewrite": "write", "finish": END, } ) return graph.compile(checkpointer=InMemorySaver()) if __name__ == "__main__": app = build_graph() config = {"configurable": {"thread_id": "demo-001"}} result = app.invoke( { "topic": "LangGraph 多智能体实战", "analysis": "", "brief": "", "feedback": "", "rewrite_count": 0, "messages": [] }, config=config ) print("=" * 50) print("最终简报：") print(result["brief"]) print("=" * 50) print("审核反馈：") print(result["feedback"])

4.6 运行与验证

保存所有文件后执行：

python main.py

预期输出会包含分析师要点拆解、撰稿人简报生成、审核员的反馈。如果审核返回“不通过”，图会自动回到 write 节点重新生成，最多重写 3 次后强制结束。

这里有一个值得观察的点：因为用了add_messages，messages 字段会累积所有节点的日志，你可以通过打印result["messages"]查看完整的执行轨迹。

如果运行时报错expected value of type "str" for key "topic"...之类的类型异常，多半是初始 state 里字段缺失或类型不对，对照状态定义补齐字段即可。

4.7 接入真实大模型

上面示例中的mock_llm_*函数返回的是固定文本，实际项目中需要替换成真实模型调用。

以 OpenAI 风格接口为例，可以这样实现分析师节点：

from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="gpt-4o-mini", temperature=0.2 ) def real_analysis_node(state: BriefState) -> dict: topic = state["topic"] prompt = f"你是技术分析师，请针对《{topic}》输出 4 条分析要点。" response = llm.invoke(prompt) analysis = response.content return { "analysis": analysis, "messages": [{"role": "assistant", "content": f"分析师完成：{analysis}"}] }

同理，撰稿人和审核员节点也可以改成提示词模板 + 模型调用的方式。审核员节点可以要求模型返回固定格式，方便代码解析：

prompt = f""" 请作为审核员检查以下简报是否合格。 简报内容： {brief} 如果合格，请只回复：pass 如果不合格，请只回复：fail:需要修改的原因 """

通过设置强约束提示词，将模型输出保持为机器可解析的格式，这是工程化的常用做法。

5. 多智能体架构模式总结

5.1 顺序执行模式

与上面 4.5 的案例类似，多个节点按固定顺序执行，适合流程特别明确的业务。优点是简单、稳定，缺点是模型没有决策空间。

适用场景：固定流水线任务，比如数据清洗 -> 特征提取 -> 写报告。

5.2 并行扇出模式

使用Send将一个任务拆分成多个子任务并行执行。适合“处理多份文档”“批量分析多个维度”等场景。

参考结构：

def fan_out_node(state): items = state["items"] return [Send("process_item", {"item": item}) for item in items] def process_item_node(state): item = state["item"] result = do_something(item) return {"results": [result]} # 需要配合 reducer 合并结果

并行子任务的结果合并需要额外处理。如果多个子节点返回同一个字段，必须定义 reducer，比如：

from typing import Annotated from langgraph.graph.message import add_messages class ParallelState(TypedDict): results: Annotated[list, lambda a, b: a + b] # 自定义 reducer 合并列表

5.3 监督者模式

监督者模式是多智能体架构中最常见的模式之一。核心思路是：一个 Supervisor Agent 负责理解用户意图，动态决定下一步把任务交给哪个专业智能体。

在 LangGraph 中，Supervisor 就是路由节点，用大模型来替换条件边里的规则判断。这样系统的流程控制不再是死板的 if-else，而是模型的动态决策。

5.4 分层模式

当智能体数量比较多时，可以把一部分智能体封装成一个子图，再把子图作为整体嵌入到更大的图里。这种模式和软件工程里的模块化思想一致。

LangGraph 允许直接在一个图里编译另一个子图并作为节点添加，适合大型复杂业务。

5.5 架构选型建议

• 如果智能体角色少、流程固定，顺序模式最简单。
• 如果子任务相互独立，优先并行，缩短耗时。
• 如果流程会随着用户意图变化，监督者模式更合适。
• 如果智能体数量超过 5 个，优先考虑分层分组，降低单张图的复杂度。

6. 常见问题与排查思路

以下是我在开发 LangGraph 多智能体应用时遇到的典型问题，整理成表格方便查阅：

问题现象	常见原因	解决思路
条件边报错 KeyError	路由函数返回的 key 不在映射表中	检查返回值和映射表的 key 是否完全一致
状态字段被覆盖	没有使用 reducer	将需要追加的字段声明为Annotated[list, add_messages]或自定义 reducer
图不执行某节点	入口节点设置错误	确认set_entry_point指向的节点已注册
死循环	缺少循环终止条件	添加重试次数统计节点，达到上限强制结束
并行子任务结果丢失	多节点同时返回同名字段，没有合并	给目标字段配置自定义 reducer
提示词模型输出无法解析	模型没有按格式返回	使用输出解析器或要求模型返回 JSON，并做异常兜底
调用模型报 401	API Key 错误或未设置	检查环境变量和实例化参数
调用模型报 429	请求频率超过限制	降低并发、增加重试退避

除了表格里的问题，还有两个经常踩的坑。

第一个坑是版本差异。LangGraph 迭代很快，网上资料可能是几个月前的，接口命名可能已经变化。遇到接口报错，优先去官方文档核对当前版本的StateGraphAPI，不要盲目照抄老教程。

第二个坑是状态字典里塞了太多数据。有人会把大文本、图片、临时计算结果全部都塞进 State，导致每次状态传递都开销很大，调试也不方便。建议 State 只保存核心流程数据，需要隔离的大对象存入外部存储，用 id 引用。

还有一个关于执行方式的排查建议：如果你使用invoke发现结果和预期不符，可以把程序改成stream逐节点观察：

for event in app.stream(initial_state, config=config): for key, value in event.items(): print(f"节点: {key}") print(value)

这样能清楚看到每个节点到底返回了什么、路由函数走了哪条边，比直接观察最终结果直观得多。

7. 最佳实践与工程建议

7.1 状态设计要克制

State 是图的公共黑板，但不是垃圾桶。在设计状态时，可以问自己三个问题：

• 这个字段会被两个以上节点用到吗？如果只在一个节点内部使用，就不应该放进 State。
• 这个字段会持续增长吗？如果是消息列表，要考虑裁剪旧消息，避免上下文过长。
• 这个字段是临时中间结果吗？如果是，建议在一个节点内部处理完再输出最终结果。

7.2 节点函数保持单一职责

每个节点只做一件事情，节点命名要能直接反映职责。

比如不要写一个process_all_node，里面既调模型，又解析输出，还写日志。应该拆成format_input_node、call_model_node、parse_output_node。

这样拆分的好处是调试方便、复用方便、测试方便。模型调用失败时，你只需要替换call_model_node的实现。

7.3 循环控制必须要有硬上限

多智能体应用的运行时间模型和普通接口完全不同。普通接口通常毫秒级响应，多智能体可能涉及多轮模型调用和工具调用，耗时可能突破几十秒甚至几分钟。所以必须设置：

• 节点重试次数上限。
• 智能体重写轮次上限。
• 总执行步数上限。

在上面的案例中，rewrite_count >= 3就是循环终止条件。实际项目中，可以在路由函数里增加这种判断，也可以在invoke前后用超时机制兜底。

7.4 日志与可观测性

生产环境一定要记录足够详细的日志。每次模型调用要记录：

model=xxx prompt_tokens=xxx completion_tokens=xxx latency_ms=xxx agent_name=xxx thread_id=xxx

这样不仅能排查问题，还能做成本分析。建议在节点执行入口和出口各加一行日志，输出节点名称和关键状态字段的变化。

7.5 工具调用权限与安全边界

多智能体通常伴随工具调用。凡是给智能体开放的工具，都要遵守最小权限原则：

• 只开放业务必需的工具。
• 工具内部做参数校验和越权检查。
• 对可能产生外部副作用的工具，增加人工确认环节。
• 敏感信息不要出现在提示词里。

例如一个“发送邮件”工具，不应该允许智能体直接向任意地址发送邮件，至少要校验收件人是否在白名单内。

7.6 成本控制

多智能体的 token 消耗通常是单次模型调用的数倍。控制成本可以从以下几方面入手：

• 缓存高频问题。
• 多智能体之间传递内容时，只传摘要而不是全文。
• 对同一文本内容，避免多个智能体重复读取完整原文。
• 设置最大执行步数，防止循环导致 token 浪费。

7.7 异步与并发部署

invoke是同步执行，高并发场景下会阻塞线程。生产环境推荐使用异步接口：

result = await app.ainvoke(initial_state, config=config) async for event in app.astream(initial_state, config=config): # 处理事件 pass

如果你通过 FastAPI 暴露接口，在async def端点里调用ainvoke更合适，避免阻塞事件循环。

8. 总结与下一步学习路线

写到这里，我们已经完成了一套从零开始的 LangGraph 多智能体实战：

• 理解了多智能体的概念，以及 LangGraph 在其中的定位。
• 掌握了 State、Node、Edge、Conditional Edge、Send、Checkpoint 等核心组件。
• 实现了一个“分析师 + 撰稿人 + 审核员”的三智能体协作系统，包含循环和条件路由。
• 整理了多智能体架构的几种常见模式。
• 梳理了开发中的高频问题和高阶工程建议。

如果你准备继续深入学习，建议按下面的路线走：

1. 精读 LangGraph 官方文档，重点关注 Concepts 部分对 State、Graph、Checkpoint 的说明。
2. 在本地跑一个基于真实模型的监督者模式案例，体会 Supervisor 动态路由。
3. 研究Send的并行扇出机制，尝试构建一个批量处理的智能体。
4. 阅读 LangGraph 源码中关于 reducer 的合并逻辑，加深对状态管理的理解。
5. 尝试连接 Postgres 或 Redis 作为 Checkpoint 存储，实现会话级记忆持久化。

最后给你一个实操建议：不要在项目一开始就追求复杂的智能体架构。先用模拟数据把图的结构、路由逻辑、状态流转跑通，再逐步把模拟节点替换成真实模型调用。这种“先搭骨架再填肉”的方式，能帮你更快定位问题是出在提示词、状态还是流程控制上。如果这篇文章对你理解 LangGraph 多智能体有帮助，可以收藏备用，也欢迎动手把案例跑起来，遇到问题可以在评论区交流。