在实际构建基于大语言模型的智能体应用时单个智能体往往难以处理复杂的、需要多步骤协作或具备不同专业能力的任务。这时多智能体系统Multi-Agent System, MAS就成为了一个关键架构。LangGraph 作为 LangChain 生态中用于构建有状态、多步骤应用特别是智能体的框架其核心设计思想天然契合多智能体协作场景。它通过“图”Graph来定义智能体之间的工作流将状态State作为信息传递的载体使得构建复杂的多智能体协作系统变得前所未有的清晰和可控。本文将从零开始深入讲解如何利用 LangGraph 构建一个实用的多智能体系统。我们将首先理解其核心架构与组件然后通过一个完整的代码实战项目——一个包含“规划者”、“执行者”和“审查者”的协作写作智能体系统——来演示从环境搭建、架构设计、代码实现到运行验证的全过程。最后我们会探讨在生产环境中部署此类系统时需要注意的常见问题、排查路径以及最佳实践。无论你是希望将 LangChain 应用升级为更健壮的流程还是从头开始设计一个多智能体协作任务这篇文章都将提供一条清晰的路径。1. 理解 LangGraph 的核心图、状态与节点在深入代码之前必须清晰理解 LangGraph 的三个核心概念图Graph、状态State和节点Node。这是构建任何 LangGraph 应用尤其是多智能体系统的基石。1.1 图Graph定义工作流的骨架LangGraph 中的“图”是一个有向图它定义了应用的工作流程。图中的节点代表一个独立的处理单元例如一个智能体、一个工具调用或一个条件判断边则代表节点之间的执行顺序和条件流转。通俗理解你可以把图看作一个公司的组织架构图或一个产品的流程图。它不关心每个部门节点内部具体如何运作只关心任务状态应该按照什么路径在不同部门间传递。技术定义在 LangGraph 中StateGraph类用于构建这个图。你向其中添加节点函数并定义这些节点之间的连接关系边。作用在多智能体系统中图清晰地定义了智能体之间的协作协议。例如“规划者”节点完成后任务应该交给“执行者”节点“执行者”完成后可能需要“审查者”节点进行校验。关键特性图支持循环Cycles这是实现多轮对话、迭代优化如“执行-审查-再执行”等复杂逻辑的基础。这是 LangGraph 区别于简单线性链Chain的核心能力。1.2 状态State智能体间通信的共享内存状态是一个字典Dict或 Pydantic 模型它随着图的执行在不同节点间流动和更新。每个节点都可以读取当前状态并返回一个更新后的状态。通俗理解状态就像一份共享的“任务工单”或“项目看板”。所有智能体节点都在这份工单上读取信息、填写自己的进展、提出修改意见。技术定义在 LangGraph 中你需要定义一个State类型。通常我们使用TypedDict或 PydanticBaseModel来明确定义状态中包含哪些字段例如task_description,plan,draft,feedback等。作用它是多智能体协作的“粘合剂”。规划者的输出计划被写入状态执行者从状态中读取计划并生成草稿审查者再读取草稿给出反馈。所有信息都通过状态共享避免了智能体间复杂的直接通信。设计原则状态设计是系统设计的关键。字段应清晰、职责明确。避免一个字段承载过多含义也避免字段间存在隐式依赖。1.3 节点Node执行具体任务的单元节点是图中的一个步骤对应一个可调用的函数。这个函数接收当前State作为输入并返回一个更新后的State。通俗理解节点就是公司里的各个职能部门如市场部、研发部、质检部。每个部门有自己明确的职责函数逻辑。技术定义在代码中节点就是一个普通的 Python 函数或异步函数其第一个参数是当前状态返回值是一个字典包含要更新到状态中的键值对。作用在多智能体系统中一个节点通常封装了一个智能体的完整逻辑调用大模型、使用工具、处理结果。节点函数内部可以使用 LangChain 的 LCEL、Runnable等任何你熟悉的模式。关键点节点函数应该保持“纯净”其逻辑应仅依赖于输入的状态和其内部封装的工具/模型避免副作用和外部状态依赖这有利于测试和调试。理解了这三个概念我们就知道构建一个 LangGraph 多智能体系统的标准流程1) 设计协作流程画图2) 定义共享信息结构设计状态3) 实现每个智能体的功能编写节点函数4) 将节点组装到图中并定义流转逻辑。2. 环境准备与项目初始化在开始编写多智能体系统之前我们需要搭建一个隔离、可复现的 Python 开发环境并安装必要的依赖。2.1 创建虚拟环境与安装依赖强烈建议使用虚拟环境来管理项目依赖以避免与系统或其他项目的包发生冲突。# 1. 创建项目目录并进入 mkdir langgraph-multi-agent-tutorial cd langgraph-multi-agent-tutorial # 2. 创建 Python 虚拟环境 (这里使用 venv你也可以用 conda) python -m venv .venv # 3. 激活虚拟环境 # 在 Linux/macOS 上 source .venv/bin/activate # 在 Windows 上 # .venv\Scripts\activate # 4. 升级 pip 和 setuptools pip install --upgrade pip setuptools wheel # 5. 安装核心依赖 pip install langgraph langchain langchain-openai依赖说明langgraph: 本文的核心框架用于构建图工作流。langchain: 提供了智能体、工具链、提示词模板等基础组件与 LangGraph 无缝集成。langchain-openai: LangChain 的 OpenAI 集成包方便我们调用 GPT 系列模型。你也可以使用langchain-anthropic,langchain-google-genai等对应其他模型的包。2.2 配置 API 密钥为了调用 OpenAI 的模型你需要设置 API 密钥。切勿将密钥硬编码在代码中推荐使用环境变量管理。# 在 Linux/macOS 的终端中设置环境变量 export OPENAI_API_KEY你的-openai-api-key # 在 Windows 的 CMD 中 # set OPENAI_API_KEY你的-openai-api-key # 在 Windows PowerShell 中 # $env:OPENAI_API_KEY你的-openai-api-key在你的 Python 代码中可以通过os.environ读取import os from langchain_openai import ChatOpenAI api_key os.environ.get(OPENAI_API_KEY) if not api_key: raise ValueError(请设置 OPENAI_API_KEY 环境变量) # 初始化 LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0.5, api_keyapi_key)2.3 项目结构规划一个清晰的项目结构有助于代码管理和后期扩展。建议按如下方式组织langgraph-multi-agent-tutorial/ ├── .venv/ # Python 虚拟环境.gitignore 中排除 ├── .env # 本地环境变量文件.gitignore 中排除 ├── requirements.txt # 项目依赖清单 ├── src/ # 源代码目录 │ ├── __init__.py │ ├── state.py # 定义状态State的模块 │ ├── nodes/ # 存放所有节点函数的模块 │ │ ├── __init__.py │ │ ├── planner.py # 规划者节点 │ │ ├── writer.py # 执行者/写作者节点 │ │ └── reviewer.py # 审查者节点 │ ├── graph.py # 构建和编译 LangGraph 的模块 │ └── main.py # 应用主入口 ├── tests/ # 测试目录 └── README.md现在运行pip freeze requirements.txt可以生成依赖文件。接下来我们将从定义状态开始逐步构建系统。3. 实战构建一个协作写作多智能体系统我们将构建一个包含三个智能体的系统规划者Planner分析用户需求制定写作大纲和步骤。执行者/写作者Writer根据规划者的大纲撰写具体的文章内容。审查者Reviewer检查写作者生成的内容从逻辑、风格、语法等方面提供反馈。系统可以根据反馈决定是否让写作者重写。这是一个经典的“规划-执行-审查”循环非常适合用 LangGraph 的图循环来实现。3.1 第一步定义协作状态State在src/state.py中我们使用 PydanticBaseModel来定义状态。Pydantic 提供了类型检查和序列化支持比简单的TypedDict更强大。# src/state.py from typing import List, Optional, Literal from pydantic import BaseModel, Field class WritingState(BaseModel): 写作多智能体系统的共享状态 # 输入 task_description: str Field(description用户原始的任务描述) # 规划者输出 outline: Optional[str] Field(defaultNone, description规划者生成的详细大纲) writing_instructions: Optional[str] Field(defaultNone, description给写作者的具体指令) # 执行者输出 draft: Optional[str] Field(defaultNone, description写作者生成的草稿) # 审查者输出 feedback: Optional[str] Field(defaultNone, description审查者提供的反馈) revision_count: int Field(default0, description修订次数计数器) # 控制流 next: Optional[Literal[write, review, end]] Field(defaultNone, description决定下一个执行的节点)状态字段解释task_description: 用户输入整个流程的起点。outlinewriting_instructions: 由规划者填充是给执行者的“蓝图”。draft: 由执行者填充是工作的核心产出。feedback: 由审查者填充用于指导迭代。revision_count: 一个计数器用于限制循环次数防止无限循环。next:这是实现循环的关键。每个节点在结束时通过更新这个字段来告诉 LangGraph 下一步该去哪个节点。例如审查者可以决定nextwrite需要重写或nextend任务完成。3.2 第二步实现智能体节点Nodes每个节点都是一个独立的函数。我们首先在src/nodes/__init__.py中导出它们然后分别实现。3.2.1 规划者节点Planner规划者的职责是理解任务并拆解为可执行的步骤。它在流程中只执行一次。# src/nodes/planner.py from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from ..state import WritingState import os # 初始化 LLM注意从环境变量读取密钥 llm ChatOpenAI(modelgpt-4o-mini, temperature0.7, api_keyos.getenv(OPENAI_API_KEY)) # 定义规划者的提示词模板 PLANNER_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个专业的写作规划师。你的任务是根据用户需求制定一个清晰、可执行的写作大纲和具体的写作指令。), (human, 用户的需求是{task_description}\n\n请生成以下内容\n1. 一份详细的文章大纲包含主要章节和子标题。\n2. 给写作者的具体指令包括文章风格、目标读者、字数要求、关键点等。) ]) def planner_node(state: WritingState) - dict: 规划者节点分析任务生成大纲和指令。 print( 规划者节点开始工作 ) # 1. 构建调用链提示词 LLM chain PLANNER_PROMPT | llm # 2. 调用 LLM response chain.invoke({task_description: state.task_description}) # 3. 解析响应这里简单处理实际项目可能需要更复杂的解析 # 假设 LLM 的回复中大纲和指令是分开的段落。 content response.content # 一个简单的分割逻辑实际应用中可能需要更鲁棒的解析或用 Output Parser parts content.split(\n\n) outline parts[0] if len(parts) 0 else instructions \n\n.join(parts[1:]) if len(parts) 1 else # 4. 更新状态并决定下一步去“写作者”节点 updates { outline: outline, writing_instructions: instructions, next: write # 规划完成后交给写作者 } print(f规划完成。下一步{updates[next]}) return updates3.2.2 写作者节点Writer写作者根据规划者提供的大纲和指令生成文章草稿。# src/nodes/writer.py from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from ..state import WritingState import os llm ChatOpenAI(modelgpt-4o-mini, temperature0.8, api_keyos.getenv(OPENAI_API_KEY)) WRITER_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一位专业的写作者。请严格按照规划师提供的大纲和指令进行写作。如果存在审查反馈请认真参考并修改你的草稿。), (human, # 写作任务 {task_description} # 规划师提供的大纲 {outline} # 规划师提供的具体指令 {writing_instructions} # 审查反馈如果是修订 {feedback} # 当前草稿如果是修订 {draft} 请生成或修订文章草稿。 ) ]) def writer_node(state: WritingState) - dict: 写作者节点根据规划和反馈撰写或修订草稿。 print( 写作者节点开始工作 ) chain WRITER_PROMPT | llm response chain.invoke({ task_description: state.task_description, outline: state.outline or , writing_instructions: state.writing_instructions or , feedback: state.feedback or 这是初稿尚无反馈。, draft: state.draft or 这是初稿尚无草稿。 }) new_draft response.content updates { draft: new_draft, next: review # 写作完成后交给审查者 } print(f草稿生成/修订完成。下一步{updates[next]}) return updates3.2.3 审查者节点Reviewer审查者评估草稿质量并决定流程是结束还是需要修订。# src/nodes/reviewer.py from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from ..state import WritingState import os llm ChatOpenAI(modelgpt-4o-mini, temperature0.3, api_keyos.getenv(OPENAI_API_KEY)) # 温度低一些评价更稳定 REVIEWER_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一位严格的写作审查员。你需要从逻辑连贯性、语言风格、语法、是否遵循大纲和指令等方面评价草稿。如果草稿质量合格则批准通过否则提供具体、可操作的修改反馈。), (human, # 原始任务 {task_description} # 规划大纲 {outline} # 写作指令 {writing_instructions} # 待审查的草稿 {draft} # 当前修订轮次 第 {revision_count} 轮 请提供你的审查意见。你的回复必须严格遵循以下格式 评价[简要的整体评价如优秀、良好、合格、需要重大修改] 详细反馈[具体的修改建议分点列出] 决定[只能填写“通过”或“重写”] ) ]) def reviewer_node(state: WritingState) - dict: 审查者节点评价草稿决定通过或重写。 print( 审查者节点开始工作 ) chain REVIEWER_PROMPT | llm response chain.invoke({ task_description: state.task_description, outline: state.outline or , writing_instructions: state.writing_instructions or , draft: state.draft or , revision_count: state.revision_count }) content response.content # 简单解析决定生产环境应用更稳健的解析器如 OutputParser decision 重写 # 默认值 feedback content if 决定通过 in content: decision 通过 elif 决定重写 in content: decision 重写 # 更新状态 updates { feedback: feedback, revision_count: state.revision_count 1 } # 基于决定和修订次数判断下一步 if decision 通过 or state.revision_count 3: # 设置最大修订次数为3 updates[next] end print(f审查完成。决定{decision}。流程结束。) else: updates[next] write print(f审查完成。决定{decision}。需要第{updates[revision_count]}轮修订。下一步{updates[next]}) return updates关键点审查者节点包含了业务逻辑判断。它根据 LLM 的输出和修订次数计数器决定下一个节点是“write”继续循环还是“end”终止循环。这是实现条件分支和循环的核心。3.3 第三步组装图并定义工作流在src/graph.py中我们将所有节点组装成一个完整的图工作流。# src/graph.py from langgraph.graph import StateGraph, END from .state import WritingState from .nodes.planner import planner_node from .nodes.writer import writer_node from .nodes.reviewer import reviewer_node def create_writing_agent_graph(): 创建并编译写作智能体图 # 1. 创建图并指定状态模式 workflow StateGraph(WritingState) # 2. 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(writer, writer_node) workflow.add_node(reviewer, reviewer_node) # 3. 设置入口点 workflow.set_entry_point(planner) # 4. 定义节点间的流转条件基于 state 中的 next 字段 # planner 完成后无条件前往 writer workflow.add_edge(planner, writer) # writer 完成后无条件前往 reviewer workflow.add_edge(writer, reviewer) # reviewer 完成后根据 state.next 的值决定去向 def decide_next_step(state: WritingState): 条件判断函数决定 reviewer 之后去哪 return state.next # 返回 “write” 或 “end” workflow.add_conditional_edges( reviewer, decide_next_step, { write: writer, # 如果需要重写回到 writer 节点 end: END # 如果通过或超限结束流程 } ) # 5. 编译图生成可执行对象 graph workflow.compile() return graph图结构解析add_node: 将我们之前定义的函数注册为图中的节点。set_entry_point: 指定流程从 “planner” 节点开始。add_edge: 添加无条件边。planner - writer和writer - reviewer是固定的顺序。add_conditional_edges: 添加条件边。这是实现循环的关键。reviewer节点之后根据state.next的值由reviewer_node函数设置决定下一步是回到writer还是走向END终止。至此一个完整的、具备循环能力的多智能体图就构建完成了。4. 运行、验证与结果分析现在让我们编写主程序来运行这个多智能体系统并观察其协作过程。4.1 编写主程序入口在src/main.py中# src/main.py import asyncio from .graph import create_writing_agent_graph async def main(): 主函数 # 1. 创建图 print(正在初始化协作写作智能体图...) graph create_writing_agent_graph() # 2. 定义初始状态用户输入 initial_state { task_description: 撰写一篇关于 LangGraph 如何简化多智能体系统开发的博客文章面向中级 Python 开发者字数约1500字。, revision_count: 0 } # 3. 执行图 print(\n开始执行多智能体协作流程...) print(- * 50) final_state await graph.ainvoke(initial_state) # 4. 输出最终结果 print(\n *50) print(流程执行完毕) print(*50) print(f\n最终修订轮次: {final_state[revision_count]}) print(f\n最终生成的草稿:\n) print(final_state[draft]) print(f\n最终审查反馈:\n) print(final_state[feedback]) if __name__ __main__: asyncio.run(main())4.2 运行与输出分析在项目根目录下运行python -m src.main你将看到类似下面的控制台输出内容因模型生成随机性而异正在初始化协作写作智能体图... 开始执行多智能体协作流程... -------------------------------------------------- 规划者节点开始工作 规划完成。下一步write 写作者节点开始工作 草稿生成/修订完成。下一步review 审查者节点开始工作 审查完成。决定重写。需要第1轮修订。下一步write 写作者节点开始工作 草稿生成/修订完成。下一步review 审查者节点开始工作 审查完成。决定通过。流程结束。 流程执行完毕 最终修订轮次: 2 最终生成的草稿: 一篇关于 LangGraph 的博客文章草稿... 最终审查反馈: 评价良好 详细反馈1. 结构清晰符合大纲... 2. 语言面向开发者易懂... 3. 代码示例准确... 决定通过流程解读流程从planner开始生成大纲和指令。无条件流转到writer生成第一版草稿。无条件流转到reviewer进行审查。reviewer判断第一版草稿需要“重写”next”write”于是图通过条件边回到writer节点。writer根据反馈生成第二版草稿。再次进入reviewer此次审查“通过”next”end”流程终止。最终状态中包含了经过两轮修订后的草稿和最终的审查反馈。4.3 可视化工作流LangGraph 提供了一个强大的可视化工具可以帮助你理解和调试图的结构。在src/main.py中添加# ... 在 main 函数中创建 graph 后 ... # 将图结构导出为 PNG 图片 from langchain_core.runnables.graph import MermaidDrawer try: drawer MermaidDrawer() png_data drawer.draw(graph).png_data with open(writing_agent_graph.png, wb) as f: f.write(png_data) print(工作流图已保存为 ‘writing_agent_graph.png‘) except Exception as e: print(f生成流程图时出错: {e})运行后你会得到一个writing_agent_graph.png文件清晰地展示了planner - writer - reviewer - (write/end)的循环结构。5. 核心机制详解与高级配置5.1 状态State的更新机制部分更新与合并在节点函数中我们返回一个字典如{“outline”: “大纲内容”, “next”: “write”}。LangGraph 不会简单地替换整个状态而是执行一次浅合并。这意味着返回的字典中的键会更新到状态中。状态中未在返回字典里出现的键保持不变。如果返回字典中的值是None状态中对应的键也会被更新为None使用Pydantic模型时需注意字段的Optional类型。这种机制使得每个节点只需关心自己负责更新的部分非常符合单一职责原则。5.2 条件边Conditional Edges与循环控制add_conditional_edges是实现动态工作流的核心。其第二个参数是一个函数它接收当前State并返回一个字符串或END。这个返回值必须与第三个参数字典中的某个键匹配以决定下一步走向。 在我们的例子中decide_next_step函数返回state.next其值由reviewer_node根据 LLM 的判断和业务规则如修订次数来设置。这是一种非常灵活的模式你可以基于状态的任何属性来做路由决策。5.3 图的编译与执行workflow.compile()返回一个CompiledGraph对象它是可执行的。主要执行方法有.invoke(initial_state): 同步执行。.ainvoke(initial_state): 异步执行推荐尤其当节点涉及网络 IO 时。.stream(initial_state): 流式执行可以观察到每个节点执行前后的状态快照非常适合调试。5.4 为节点添加“中断”或“人工审核”在实际生产系统中有时需要在特定节点如审查者后暂停流程等待人工介入。这可以通过“中断”机制实现。LangGraph 支持在图中设置检查点Checkpoint并在特定条件下暂停等待外部信号如 API 调用来恢复。 基本思路是在reviewer_node中如果检测到需要人工审核例如反馈中存在“不确定”关键词则更新状态next”human_review”并在图中为“human_review”配置一个特殊的节点或边该节点不执行实际逻辑而是将流程挂起并通过外部系统通知人工。6. 常见问题排查与调试指南构建 LangGraph 多智能体系统时你可能会遇到以下典型问题。6.1 状态更新不生效问题现象可能原因检查方式处理建议节点中更新的字段在下一个节点中读取不到。1. 节点函数返回的字典键名与状态字段名不匹配。2. 使用了嵌套字典但更新是浅合并导致内部字典被整个替换。1. 打印节点函数的输入state和返回的updates。2. 使用graph.stream()观察每个步骤的状态变化。1. 确保返回字典的键与State类定义的字段名完全一致。2. 对于嵌套结构考虑在状态中使用 Pydantic 模型或在节点内进行深拷贝和合并。6.2 图陷入无限循环问题现象可能原因检查方式处理建议流程在writer和reviewer间不停循环无法结束。1.reviewer_node的逻辑始终返回next”write”。2. 条件边判断函数decide_next_step逻辑错误。3. 缺少循环终止条件如最大重试次数。1. 在reviewer_node中打印 LLM 的原始输出和解析后的decision。2. 检查state.revision_count是否被正确递增和读取。1. 强化reviewer_node中对 LLM 输出的解析确保能准确识别“通过”信号。2.务必设置安全阀如在状态中维护revision_count并在reviewer_node中判断若超过阈值则强制next”end”。6.3 LLM 调用失败或超时问题现象可能原因检查方式处理建议节点执行卡住或报错OpenAIError。1. API 密钥未设置或错误。2. 网络问题。3. 模型上下文超长。4. 请求速率超限。1. 检查os.getenv(“OPENAI_API_KEY”)是否获取到值。2. 在节点函数内用try…except包裹 LLM 调用并打印错误。1. 使用.env文件配合python-dotenv管理密钥。2. 为ChatOpenAI初始化增加max_retries,timeout参数。3. 在状态设计中避免在每次循环中累积过长的历史信息。6.4 条件边路由错误问题现象可能原因检查方式处理建议流程没有按预期走向某个分支或报错KeyError。1.decide_next_step函数返回的值不在add_conditional_edges指定的映射字典中。2. 前置节点没有正确设置用于判断的字段如state.next。1. 在decide_next_step函数中打印其返回值。2. 检查映射字典的键是否与返回值完全匹配大小写敏感。1. 在decide_next_step函数中增加默认返回值如return state.get(‘next’, ‘end’)。2. 使用枚举类型Literal来定义和约束可能的路由值如我们状态中定义的next字段类型。6.5 调试技巧使用stream方法这是最强大的调试工具。它返回一个迭代器每次 yield 一个(节点名, 节点执行后的状态)对。async for step in graph.astream(initial_state, stream_mode”values”): node, state step print(f”节点 [{node}] 执行完毕。状态片段:”, {k: state[k] for k in [‘next’, ‘revision_count’]})在节点内打印日志如示例代码中的print(“ 节点开始工作 ”)可以清晰跟踪执行流。可视化图如前所述生成 Mermaid 图确保图的结构符合你的设计预期。7. 生产环境最佳实践与扩展方向将 LangGraph 多智能体系统用于生产环境需要考虑更多工程化因素。7.1 状态设计优化使用 Pydantic 模型正如示例所示利用 Pydantic 进行类型验证、字段说明和默认值设置能极大减少运行时错误。避免状态爆炸不要在状态中存储不断增长的完整历史记录如每轮对话。只保留当前轮次必需的信息。对于需要记忆的场景考虑使用外部向量数据库或记忆组件。区分配置与运行时数据将模型温度、API 密钥等配置信息与运行时状态分离通过依赖注入或上下文传递。7.2 节点函数设计原则单一职责一个节点只做一件事。例如planner_node只负责规划不要在里面调用写作工具。幂等性与可重入设计节点函数时尽量使其幂等。即给定相同的输入状态应产生相同的输出更新。这有利于错误恢复和重试。异常处理节点内部应有完善的try…except捕获 LLM 调用、工具调用、网络等异常并选择是更新状态如error_message还是直接抛出。LangGraph 支持在图上设置错误处理边add_edge(…, conditional_after[…])可以将出错节点路由到特定的错误处理节点。7.3 性能与可观测性异步执行所有节点函数都使用async def定义并在调用时使用ainvoke或astream可以更好地利用 IO 等待时间提升吞吐量。设置超时与重试在初始化 LLM 或工具时配置合理的timeout和max_retries。集成日志与监控使用logging模块替代print并集成像 LangSmith 这样的可观测性平台。LangSmith 可以可视化跟踪每次图执行的完整链路、每个节点的输入输出、耗时和 Token 使用量是调试和优化生产系统的利器。持久化检查点对于长时间运行的工作流可以利用 LangGraph 的持久化检查点功能将状态保存到数据库如 PostgreSQL实现故障恢复和暂停/继续。7.4 系统扩展方向动态节点节点的数量和类型可以根据状态动态决定。例如一个“任务分发器”节点分析任务后可以动态地向图中添加需要特定工具的专家节点。子图Subgraph将复杂的节点进一步拆分为一个子图实现模块化和复用。例如writer_node本身可以是一个包含“调研”、“起草”、“润色”三个步骤的子图。与外部系统集成节点内可以调用任何外部 API、数据库或服务。这使得 LangGraph 可以作为智能“编排层”协调传统的软件模块与 LLM 智能体。多模态与工具增强为智能体配备图像识别、代码执行、网络搜索、数据库查询等工具可以极大扩展其能力边界。利用 LangChain 丰富的工具集成可以轻松实现。通过遵循以上实践你可以将本文的示例从一个简单的原型逐步演进为一个健壮、可维护、可扩展的企业级多智能体应用。LangGraph 提供的图抽象正是管理这种复杂性的强大工具。 LangGraph实战：从零构建多智能体协作系统
在实际构建基于大语言模型的智能体应用时单个智能体往往难以处理复杂的、需要多步骤协作或具备不同专业能力的任务。这时多智能体系统Multi-Agent System, MAS就成为了一个关键架构。LangGraph 作为 LangChain 生态中用于构建有状态、多步骤应用特别是智能体的框架其核心设计思想天然契合多智能体协作场景。它通过“图”Graph来定义智能体之间的工作流将状态State作为信息传递的载体使得构建复杂的多智能体协作系统变得前所未有的清晰和可控。本文将从零开始深入讲解如何利用 LangGraph 构建一个实用的多智能体系统。我们将首先理解其核心架构与组件然后通过一个完整的代码实战项目——一个包含“规划者”、“执行者”和“审查者”的协作写作智能体系统——来演示从环境搭建、架构设计、代码实现到运行验证的全过程。最后我们会探讨在生产环境中部署此类系统时需要注意的常见问题、排查路径以及最佳实践。无论你是希望将 LangChain 应用升级为更健壮的流程还是从头开始设计一个多智能体协作任务这篇文章都将提供一条清晰的路径。1. 理解 LangGraph 的核心图、状态与节点在深入代码之前必须清晰理解 LangGraph 的三个核心概念图Graph、状态State和节点Node。这是构建任何 LangGraph 应用尤其是多智能体系统的基石。1.1 图Graph定义工作流的骨架LangGraph 中的“图”是一个有向图它定义了应用的工作流程。图中的节点代表一个独立的处理单元例如一个智能体、一个工具调用或一个条件判断边则代表节点之间的执行顺序和条件流转。通俗理解你可以把图看作一个公司的组织架构图或一个产品的流程图。它不关心每个部门节点内部具体如何运作只关心任务状态应该按照什么路径在不同部门间传递。技术定义在 LangGraph 中StateGraph类用于构建这个图。你向其中添加节点函数并定义这些节点之间的连接关系边。作用在多智能体系统中图清晰地定义了智能体之间的协作协议。例如“规划者”节点完成后任务应该交给“执行者”节点“执行者”完成后可能需要“审查者”节点进行校验。关键特性图支持循环Cycles这是实现多轮对话、迭代优化如“执行-审查-再执行”等复杂逻辑的基础。这是 LangGraph 区别于简单线性链Chain的核心能力。1.2 状态State智能体间通信的共享内存状态是一个字典Dict或 Pydantic 模型它随着图的执行在不同节点间流动和更新。每个节点都可以读取当前状态并返回一个更新后的状态。通俗理解状态就像一份共享的“任务工单”或“项目看板”。所有智能体节点都在这份工单上读取信息、填写自己的进展、提出修改意见。技术定义在 LangGraph 中你需要定义一个State类型。通常我们使用TypedDict或 PydanticBaseModel来明确定义状态中包含哪些字段例如task_description,plan,draft,feedback等。作用它是多智能体协作的“粘合剂”。规划者的输出计划被写入状态执行者从状态中读取计划并生成草稿审查者再读取草稿给出反馈。所有信息都通过状态共享避免了智能体间复杂的直接通信。设计原则状态设计是系统设计的关键。字段应清晰、职责明确。避免一个字段承载过多含义也避免字段间存在隐式依赖。1.3 节点Node执行具体任务的单元节点是图中的一个步骤对应一个可调用的函数。这个函数接收当前State作为输入并返回一个更新后的State。通俗理解节点就是公司里的各个职能部门如市场部、研发部、质检部。每个部门有自己明确的职责函数逻辑。技术定义在代码中节点就是一个普通的 Python 函数或异步函数其第一个参数是当前状态返回值是一个字典包含要更新到状态中的键值对。作用在多智能体系统中一个节点通常封装了一个智能体的完整逻辑调用大模型、使用工具、处理结果。节点函数内部可以使用 LangChain 的 LCEL、Runnable等任何你熟悉的模式。关键点节点函数应该保持“纯净”其逻辑应仅依赖于输入的状态和其内部封装的工具/模型避免副作用和外部状态依赖这有利于测试和调试。理解了这三个概念我们就知道构建一个 LangGraph 多智能体系统的标准流程1) 设计协作流程画图2) 定义共享信息结构设计状态3) 实现每个智能体的功能编写节点函数4) 将节点组装到图中并定义流转逻辑。2. 环境准备与项目初始化在开始编写多智能体系统之前我们需要搭建一个隔离、可复现的 Python 开发环境并安装必要的依赖。2.1 创建虚拟环境与安装依赖强烈建议使用虚拟环境来管理项目依赖以避免与系统或其他项目的包发生冲突。# 1. 创建项目目录并进入 mkdir langgraph-multi-agent-tutorial cd langgraph-multi-agent-tutorial # 2. 创建 Python 虚拟环境 (这里使用 venv你也可以用 conda) python -m venv .venv # 3. 激活虚拟环境 # 在 Linux/macOS 上 source .venv/bin/activate # 在 Windows 上 # .venv\Scripts\activate # 4. 升级 pip 和 setuptools pip install --upgrade pip setuptools wheel # 5. 安装核心依赖 pip install langgraph langchain langchain-openai依赖说明langgraph: 本文的核心框架用于构建图工作流。langchain: 提供了智能体、工具链、提示词模板等基础组件与 LangGraph 无缝集成。langchain-openai: LangChain 的 OpenAI 集成包方便我们调用 GPT 系列模型。你也可以使用langchain-anthropic,langchain-google-genai等对应其他模型的包。2.2 配置 API 密钥为了调用 OpenAI 的模型你需要设置 API 密钥。切勿将密钥硬编码在代码中推荐使用环境变量管理。# 在 Linux/macOS 的终端中设置环境变量 export OPENAI_API_KEY你的-openai-api-key # 在 Windows 的 CMD 中 # set OPENAI_API_KEY你的-openai-api-key # 在 Windows PowerShell 中 # $env:OPENAI_API_KEY你的-openai-api-key在你的 Python 代码中可以通过os.environ读取import os from langchain_openai import ChatOpenAI api_key os.environ.get(OPENAI_API_KEY) if not api_key: raise ValueError(请设置 OPENAI_API_KEY 环境变量) # 初始化 LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0.5, api_keyapi_key)2.3 项目结构规划一个清晰的项目结构有助于代码管理和后期扩展。建议按如下方式组织langgraph-multi-agent-tutorial/ ├── .venv/ # Python 虚拟环境.gitignore 中排除 ├── .env # 本地环境变量文件.gitignore 中排除 ├── requirements.txt # 项目依赖清单 ├── src/ # 源代码目录 │ ├── __init__.py │ ├── state.py # 定义状态State的模块 │ ├── nodes/ # 存放所有节点函数的模块 │ │ ├── __init__.py │ │ ├── planner.py # 规划者节点 │ │ ├── writer.py # 执行者/写作者节点 │ │ └── reviewer.py # 审查者节点 │ ├── graph.py # 构建和编译 LangGraph 的模块 │ └── main.py # 应用主入口 ├── tests/ # 测试目录 └── README.md现在运行pip freeze requirements.txt可以生成依赖文件。接下来我们将从定义状态开始逐步构建系统。3. 实战构建一个协作写作多智能体系统我们将构建一个包含三个智能体的系统规划者Planner分析用户需求制定写作大纲和步骤。执行者/写作者Writer根据规划者的大纲撰写具体的文章内容。审查者Reviewer检查写作者生成的内容从逻辑、风格、语法等方面提供反馈。系统可以根据反馈决定是否让写作者重写。这是一个经典的“规划-执行-审查”循环非常适合用 LangGraph 的图循环来实现。3.1 第一步定义协作状态State在src/state.py中我们使用 PydanticBaseModel来定义状态。Pydantic 提供了类型检查和序列化支持比简单的TypedDict更强大。# src/state.py from typing import List, Optional, Literal from pydantic import BaseModel, Field class WritingState(BaseModel): 写作多智能体系统的共享状态 # 输入 task_description: str Field(description用户原始的任务描述) # 规划者输出 outline: Optional[str] Field(defaultNone, description规划者生成的详细大纲) writing_instructions: Optional[str] Field(defaultNone, description给写作者的具体指令) # 执行者输出 draft: Optional[str] Field(defaultNone, description写作者生成的草稿) # 审查者输出 feedback: Optional[str] Field(defaultNone, description审查者提供的反馈) revision_count: int Field(default0, description修订次数计数器) # 控制流 next: Optional[Literal[write, review, end]] Field(defaultNone, description决定下一个执行的节点)状态字段解释task_description: 用户输入整个流程的起点。outlinewriting_instructions: 由规划者填充是给执行者的“蓝图”。draft: 由执行者填充是工作的核心产出。feedback: 由审查者填充用于指导迭代。revision_count: 一个计数器用于限制循环次数防止无限循环。next:这是实现循环的关键。每个节点在结束时通过更新这个字段来告诉 LangGraph 下一步该去哪个节点。例如审查者可以决定nextwrite需要重写或nextend任务完成。3.2 第二步实现智能体节点Nodes每个节点都是一个独立的函数。我们首先在src/nodes/__init__.py中导出它们然后分别实现。3.2.1 规划者节点Planner规划者的职责是理解任务并拆解为可执行的步骤。它在流程中只执行一次。# src/nodes/planner.py from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from ..state import WritingState import os # 初始化 LLM注意从环境变量读取密钥 llm ChatOpenAI(modelgpt-4o-mini, temperature0.7, api_keyos.getenv(OPENAI_API_KEY)) # 定义规划者的提示词模板 PLANNER_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一个专业的写作规划师。你的任务是根据用户需求制定一个清晰、可执行的写作大纲和具体的写作指令。), (human, 用户的需求是{task_description}\n\n请生成以下内容\n1. 一份详细的文章大纲包含主要章节和子标题。\n2. 给写作者的具体指令包括文章风格、目标读者、字数要求、关键点等。) ]) def planner_node(state: WritingState) - dict: 规划者节点分析任务生成大纲和指令。 print( 规划者节点开始工作 ) # 1. 构建调用链提示词 LLM chain PLANNER_PROMPT | llm # 2. 调用 LLM response chain.invoke({task_description: state.task_description}) # 3. 解析响应这里简单处理实际项目可能需要更复杂的解析 # 假设 LLM 的回复中大纲和指令是分开的段落。 content response.content # 一个简单的分割逻辑实际应用中可能需要更鲁棒的解析或用 Output Parser parts content.split(\n\n) outline parts[0] if len(parts) 0 else instructions \n\n.join(parts[1:]) if len(parts) 1 else # 4. 更新状态并决定下一步去“写作者”节点 updates { outline: outline, writing_instructions: instructions, next: write # 规划完成后交给写作者 } print(f规划完成。下一步{updates[next]}) return updates3.2.2 写作者节点Writer写作者根据规划者提供的大纲和指令生成文章草稿。# src/nodes/writer.py from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from ..state import WritingState import os llm ChatOpenAI(modelgpt-4o-mini, temperature0.8, api_keyos.getenv(OPENAI_API_KEY)) WRITER_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一位专业的写作者。请严格按照规划师提供的大纲和指令进行写作。如果存在审查反馈请认真参考并修改你的草稿。), (human, # 写作任务 {task_description} # 规划师提供的大纲 {outline} # 规划师提供的具体指令 {writing_instructions} # 审查反馈如果是修订 {feedback} # 当前草稿如果是修订 {draft} 请生成或修订文章草稿。 ) ]) def writer_node(state: WritingState) - dict: 写作者节点根据规划和反馈撰写或修订草稿。 print( 写作者节点开始工作 ) chain WRITER_PROMPT | llm response chain.invoke({ task_description: state.task_description, outline: state.outline or , writing_instructions: state.writing_instructions or , feedback: state.feedback or 这是初稿尚无反馈。, draft: state.draft or 这是初稿尚无草稿。 }) new_draft response.content updates { draft: new_draft, next: review # 写作完成后交给审查者 } print(f草稿生成/修订完成。下一步{updates[next]}) return updates3.2.3 审查者节点Reviewer审查者评估草稿质量并决定流程是结束还是需要修订。# src/nodes/reviewer.py from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from ..state import WritingState import os llm ChatOpenAI(modelgpt-4o-mini, temperature0.3, api_keyos.getenv(OPENAI_API_KEY)) # 温度低一些评价更稳定 REVIEWER_PROMPT ChatPromptTemplate.from_messages([ (system, 你是一位严格的写作审查员。你需要从逻辑连贯性、语言风格、语法、是否遵循大纲和指令等方面评价草稿。如果草稿质量合格则批准通过否则提供具体、可操作的修改反馈。), (human, # 原始任务 {task_description} # 规划大纲 {outline} # 写作指令 {writing_instructions} # 待审查的草稿 {draft} # 当前修订轮次 第 {revision_count} 轮 请提供你的审查意见。你的回复必须严格遵循以下格式 评价[简要的整体评价如优秀、良好、合格、需要重大修改] 详细反馈[具体的修改建议分点列出] 决定[只能填写“通过”或“重写”] ) ]) def reviewer_node(state: WritingState) - dict: 审查者节点评价草稿决定通过或重写。 print( 审查者节点开始工作 ) chain REVIEWER_PROMPT | llm response chain.invoke({ task_description: state.task_description, outline: state.outline or , writing_instructions: state.writing_instructions or , draft: state.draft or , revision_count: state.revision_count }) content response.content # 简单解析决定生产环境应用更稳健的解析器如 OutputParser decision 重写 # 默认值 feedback content if 决定通过 in content: decision 通过 elif 决定重写 in content: decision 重写 # 更新状态 updates { feedback: feedback, revision_count: state.revision_count 1 } # 基于决定和修订次数判断下一步 if decision 通过 or state.revision_count 3: # 设置最大修订次数为3 updates[next] end print(f审查完成。决定{decision}。流程结束。) else: updates[next] write print(f审查完成。决定{decision}。需要第{updates[revision_count]}轮修订。下一步{updates[next]}) return updates关键点审查者节点包含了业务逻辑判断。它根据 LLM 的输出和修订次数计数器决定下一个节点是“write”继续循环还是“end”终止循环。这是实现条件分支和循环的核心。3.3 第三步组装图并定义工作流在src/graph.py中我们将所有节点组装成一个完整的图工作流。# src/graph.py from langgraph.graph import StateGraph, END from .state import WritingState from .nodes.planner import planner_node from .nodes.writer import writer_node from .nodes.reviewer import reviewer_node def create_writing_agent_graph(): 创建并编译写作智能体图 # 1. 创建图并指定状态模式 workflow StateGraph(WritingState) # 2. 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(writer, writer_node) workflow.add_node(reviewer, reviewer_node) # 3. 设置入口点 workflow.set_entry_point(planner) # 4. 定义节点间的流转条件基于 state 中的 next 字段 # planner 完成后无条件前往 writer workflow.add_edge(planner, writer) # writer 完成后无条件前往 reviewer workflow.add_edge(writer, reviewer) # reviewer 完成后根据 state.next 的值决定去向 def decide_next_step(state: WritingState): 条件判断函数决定 reviewer 之后去哪 return state.next # 返回 “write” 或 “end” workflow.add_conditional_edges( reviewer, decide_next_step, { write: writer, # 如果需要重写回到 writer 节点 end: END # 如果通过或超限结束流程 } ) # 5. 编译图生成可执行对象 graph workflow.compile() return graph图结构解析add_node: 将我们之前定义的函数注册为图中的节点。set_entry_point: 指定流程从 “planner” 节点开始。add_edge: 添加无条件边。planner - writer和writer - reviewer是固定的顺序。add_conditional_edges: 添加条件边。这是实现循环的关键。reviewer节点之后根据state.next的值由reviewer_node函数设置决定下一步是回到writer还是走向END终止。至此一个完整的、具备循环能力的多智能体图就构建完成了。4. 运行、验证与结果分析现在让我们编写主程序来运行这个多智能体系统并观察其协作过程。4.1 编写主程序入口在src/main.py中# src/main.py import asyncio from .graph import create_writing_agent_graph async def main(): 主函数 # 1. 创建图 print(正在初始化协作写作智能体图...) graph create_writing_agent_graph() # 2. 定义初始状态用户输入 initial_state { task_description: 撰写一篇关于 LangGraph 如何简化多智能体系统开发的博客文章面向中级 Python 开发者字数约1500字。, revision_count: 0 } # 3. 执行图 print(\n开始执行多智能体协作流程...) print(- * 50) final_state await graph.ainvoke(initial_state) # 4. 输出最终结果 print(\n *50) print(流程执行完毕) print(*50) print(f\n最终修订轮次: {final_state[revision_count]}) print(f\n最终生成的草稿:\n) print(final_state[draft]) print(f\n最终审查反馈:\n) print(final_state[feedback]) if __name__ __main__: asyncio.run(main())4.2 运行与输出分析在项目根目录下运行python -m src.main你将看到类似下面的控制台输出内容因模型生成随机性而异正在初始化协作写作智能体图... 开始执行多智能体协作流程... -------------------------------------------------- 规划者节点开始工作 规划完成。下一步write 写作者节点开始工作 草稿生成/修订完成。下一步review 审查者节点开始工作 审查完成。决定重写。需要第1轮修订。下一步write 写作者节点开始工作 草稿生成/修订完成。下一步review 审查者节点开始工作 审查完成。决定通过。流程结束。 流程执行完毕 最终修订轮次: 2 最终生成的草稿: 一篇关于 LangGraph 的博客文章草稿... 最终审查反馈: 评价良好 详细反馈1. 结构清晰符合大纲... 2. 语言面向开发者易懂... 3. 代码示例准确... 决定通过流程解读流程从planner开始生成大纲和指令。无条件流转到writer生成第一版草稿。无条件流转到reviewer进行审查。reviewer判断第一版草稿需要“重写”next”write”于是图通过条件边回到writer节点。writer根据反馈生成第二版草稿。再次进入reviewer此次审查“通过”next”end”流程终止。最终状态中包含了经过两轮修订后的草稿和最终的审查反馈。4.3 可视化工作流LangGraph 提供了一个强大的可视化工具可以帮助你理解和调试图的结构。在src/main.py中添加# ... 在 main 函数中创建 graph 后 ... # 将图结构导出为 PNG 图片 from langchain_core.runnables.graph import MermaidDrawer try: drawer MermaidDrawer() png_data drawer.draw(graph).png_data with open(writing_agent_graph.png, wb) as f: f.write(png_data) print(工作流图已保存为 ‘writing_agent_graph.png‘) except Exception as e: print(f生成流程图时出错: {e})运行后你会得到一个writing_agent_graph.png文件清晰地展示了planner - writer - reviewer - (write/end)的循环结构。5. 核心机制详解与高级配置5.1 状态State的更新机制部分更新与合并在节点函数中我们返回一个字典如{“outline”: “大纲内容”, “next”: “write”}。LangGraph 不会简单地替换整个状态而是执行一次浅合并。这意味着返回的字典中的键会更新到状态中。状态中未在返回字典里出现的键保持不变。如果返回字典中的值是None状态中对应的键也会被更新为None使用Pydantic模型时需注意字段的Optional类型。这种机制使得每个节点只需关心自己负责更新的部分非常符合单一职责原则。5.2 条件边Conditional Edges与循环控制add_conditional_edges是实现动态工作流的核心。其第二个参数是一个函数它接收当前State并返回一个字符串或END。这个返回值必须与第三个参数字典中的某个键匹配以决定下一步走向。 在我们的例子中decide_next_step函数返回state.next其值由reviewer_node根据 LLM 的判断和业务规则如修订次数来设置。这是一种非常灵活的模式你可以基于状态的任何属性来做路由决策。5.3 图的编译与执行workflow.compile()返回一个CompiledGraph对象它是可执行的。主要执行方法有.invoke(initial_state): 同步执行。.ainvoke(initial_state): 异步执行推荐尤其当节点涉及网络 IO 时。.stream(initial_state): 流式执行可以观察到每个节点执行前后的状态快照非常适合调试。5.4 为节点添加“中断”或“人工审核”在实际生产系统中有时需要在特定节点如审查者后暂停流程等待人工介入。这可以通过“中断”机制实现。LangGraph 支持在图中设置检查点Checkpoint并在特定条件下暂停等待外部信号如 API 调用来恢复。 基本思路是在reviewer_node中如果检测到需要人工审核例如反馈中存在“不确定”关键词则更新状态next”human_review”并在图中为“human_review”配置一个特殊的节点或边该节点不执行实际逻辑而是将流程挂起并通过外部系统通知人工。6. 常见问题排查与调试指南构建 LangGraph 多智能体系统时你可能会遇到以下典型问题。6.1 状态更新不生效问题现象可能原因检查方式处理建议节点中更新的字段在下一个节点中读取不到。1. 节点函数返回的字典键名与状态字段名不匹配。2. 使用了嵌套字典但更新是浅合并导致内部字典被整个替换。1. 打印节点函数的输入state和返回的updates。2. 使用graph.stream()观察每个步骤的状态变化。1. 确保返回字典的键与State类定义的字段名完全一致。2. 对于嵌套结构考虑在状态中使用 Pydantic 模型或在节点内进行深拷贝和合并。6.2 图陷入无限循环问题现象可能原因检查方式处理建议流程在writer和reviewer间不停循环无法结束。1.reviewer_node的逻辑始终返回next”write”。2. 条件边判断函数decide_next_step逻辑错误。3. 缺少循环终止条件如最大重试次数。1. 在reviewer_node中打印 LLM 的原始输出和解析后的decision。2. 检查state.revision_count是否被正确递增和读取。1. 强化reviewer_node中对 LLM 输出的解析确保能准确识别“通过”信号。2.务必设置安全阀如在状态中维护revision_count并在reviewer_node中判断若超过阈值则强制next”end”。6.3 LLM 调用失败或超时问题现象可能原因检查方式处理建议节点执行卡住或报错OpenAIError。1. API 密钥未设置或错误。2. 网络问题。3. 模型上下文超长。4. 请求速率超限。1. 检查os.getenv(“OPENAI_API_KEY”)是否获取到值。2. 在节点函数内用try…except包裹 LLM 调用并打印错误。1. 使用.env文件配合python-dotenv管理密钥。2. 为ChatOpenAI初始化增加max_retries,timeout参数。3. 在状态设计中避免在每次循环中累积过长的历史信息。6.4 条件边路由错误问题现象可能原因检查方式处理建议流程没有按预期走向某个分支或报错KeyError。1.decide_next_step函数返回的值不在add_conditional_edges指定的映射字典中。2. 前置节点没有正确设置用于判断的字段如state.next。1. 在decide_next_step函数中打印其返回值。2. 检查映射字典的键是否与返回值完全匹配大小写敏感。1. 在decide_next_step函数中增加默认返回值如return state.get(‘next’, ‘end’)。2. 使用枚举类型Literal来定义和约束可能的路由值如我们状态中定义的next字段类型。6.5 调试技巧使用stream方法这是最强大的调试工具。它返回一个迭代器每次 yield 一个(节点名, 节点执行后的状态)对。async for step in graph.astream(initial_state, stream_mode”values”): node, state step print(f”节点 [{node}] 执行完毕。状态片段:”, {k: state[k] for k in [‘next’, ‘revision_count’]})在节点内打印日志如示例代码中的print(“ 节点开始工作 ”)可以清晰跟踪执行流。可视化图如前所述生成 Mermaid 图确保图的结构符合你的设计预期。7. 生产环境最佳实践与扩展方向将 LangGraph 多智能体系统用于生产环境需要考虑更多工程化因素。7.1 状态设计优化使用 Pydantic 模型正如示例所示利用 Pydantic 进行类型验证、字段说明和默认值设置能极大减少运行时错误。避免状态爆炸不要在状态中存储不断增长的完整历史记录如每轮对话。只保留当前轮次必需的信息。对于需要记忆的场景考虑使用外部向量数据库或记忆组件。区分配置与运行时数据将模型温度、API 密钥等配置信息与运行时状态分离通过依赖注入或上下文传递。7.2 节点函数设计原则单一职责一个节点只做一件事。例如planner_node只负责规划不要在里面调用写作工具。幂等性与可重入设计节点函数时尽量使其幂等。即给定相同的输入状态应产生相同的输出更新。这有利于错误恢复和重试。异常处理节点内部应有完善的try…except捕获 LLM 调用、工具调用、网络等异常并选择是更新状态如error_message还是直接抛出。LangGraph 支持在图上设置错误处理边add_edge(…, conditional_after[…])可以将出错节点路由到特定的错误处理节点。7.3 性能与可观测性异步执行所有节点函数都使用async def定义并在调用时使用ainvoke或astream可以更好地利用 IO 等待时间提升吞吐量。设置超时与重试在初始化 LLM 或工具时配置合理的timeout和max_retries。集成日志与监控使用logging模块替代print并集成像 LangSmith 这样的可观测性平台。LangSmith 可以可视化跟踪每次图执行的完整链路、每个节点的输入输出、耗时和 Token 使用量是调试和优化生产系统的利器。持久化检查点对于长时间运行的工作流可以利用 LangGraph 的持久化检查点功能将状态保存到数据库如 PostgreSQL实现故障恢复和暂停/继续。7.4 系统扩展方向动态节点节点的数量和类型可以根据状态动态决定。例如一个“任务分发器”节点分析任务后可以动态地向图中添加需要特定工具的专家节点。子图Subgraph将复杂的节点进一步拆分为一个子图实现模块化和复用。例如writer_node本身可以是一个包含“调研”、“起草”、“润色”三个步骤的子图。与外部系统集成节点内可以调用任何外部 API、数据库或服务。这使得 LangGraph 可以作为智能“编排层”协调传统的软件模块与 LLM 智能体。多模态与工具增强为智能体配备图像识别、代码执行、网络搜索、数据库查询等工具可以极大扩展其能力边界。利用 LangChain 丰富的工具集成可以轻松实现。通过遵循以上实践你可以将本文的示例从一个简单的原型逐步演进为一个健壮、可维护、可扩展的企业级多智能体应用。LangGraph 提供的图抽象正是管理这种复杂性的强大工具。