LangGraph实战：构建多智能体协作系统的核心原理与工程指南

在实际 AI 应用开发中，构建一个能处理复杂、多步骤任务的智能体系统，远比实现单一功能调用要困难。开发者常常面临状态管理混乱、流程控制复杂、多智能体协作困难等挑战。LangGraph 作为 LangChain 生态中用于构建有状态、多参与者应用的工作流库，提供了一种基于图（Graph）的清晰范式来编排智能体（Agent）和工具（Tool），尤其擅长处理带有循环、分支和状态共享的复杂任务。本文将深入解析 LangGraph 的核心组件，并通过一个从零开始的多智能体协作项目，手把手展示如何构建一个可运行、可调试的智能体系统。无论你是希望将现有 LangChain 应用升级为更健壮的工作流，还是计划从零设计一个多智能体协作架构，本文提供的概念、代码和排错路径都将帮助你建立清晰的技术认知。

1. 理解 LangGraph 的核心：图、状态与节点

在开始编码之前，必须理解 LangGraph 解决问题的基本模型。它并非要取代 LangChain，而是对其在复杂流程编排能力上的重要补充。

1.1 为什么需要图（Graph）来管理智能体？

传统的链式调用（Chain）适用于线性任务，例如：解析用户问题 -> 检索知识 -> 生成回答。然而，现实中的任务往往是非线性的。以一个“研究助手”任务为例：用户要求“分析一下 LangGraph 的最新特性并写一份报告”。这个任务可能包含：1）联网搜索最新信息；2）总结搜索到的多篇文档；3）根据总结草拟报告大纲；4）检查大纲完整性，若不完整则返回步骤1补充搜索；5）撰写完整报告。这个过程存在明显的循环（检查->补充）和条件分支。

如果只用简单的链，开发者需要手动维护大量的中间变量和if-else逻辑，代码会迅速变得难以维护和调试。LangGraph 将整个工作流抽象为一个有向图，图中的节点代表一个执行单元（如调用一个 LLM、运行一个工具），边代表执行路径。通过明确定义状态（State）的结构和在节点间如何流转，它使得复杂、有状态的工作流变得清晰、可预测。

1.2 核心三要素：State、Node、Edge

State（状态）：这是 LangGraph 工作流的“记忆体”和“共享白板”。它是一个字典（或 Pydantic 模型），定义了工作流执行过程中需要传递和修改的所有数据。例如，一个研究助手的状态可能包含：input（用户原始问题）、search_results（网络搜索结果列表）、summary（内容摘要）、report_outline（报告大纲）、report（最终报告）。每个节点都可以读取和修改状态中的特定字段。

Node（节点）：节点是一个可调用对象（函数），它接收当前整个 State 作为输入，执行特定操作（如调用 LLM、运行工具），并返回一个包含对 State 更新内容的字典。例如，一个search_node函数会读取state[‘input’]，调用搜索引擎工具，然后将结果写入state[‘search_results’]。

Edge（边）：边定义了控制流，即决定执行完一个节点后，下一步该去哪个节点。边分为两种：

1. 条件边（Conditional Edge）：根据 State 中的某个条件（例如，检查大纲是否完整）决定下一个节点。这实现了分支逻辑。
2. 普通边（Normal Edge）：无条件地指向下一个节点，实现线性执行。

一个特殊节点是END，表示工作流正常终止。

1.3 LangGraph 与 LangChain 的关系

这是一个常见的困惑点。你可以这样理解：

• LangChain提供了构建 AI 应用所需的丰富“乐高积木”，如各种 LLM 封装、提示模板、文档加载器、检索器、工具和基础的链。
• LangGraph提供了组装这些“乐高积木”以构建复杂动态流程的“设计图”和“组装说明书”。它尤其擅长处理那些需要循环、多参与者（智能体）协作的场景。

在实践中，你通常会同时使用两者：用 LangChain 的组件（ChatOpenAI，Tool）作为节点内的实现，用 LangGraph 来编排这些组件的执行顺序和状态流转。

2. 环境准备与项目初始化

我们将构建一个“多智能体协作写作助手”。这个系统包含两个智能体：一个“研究员”负责搜索和总结资料，一个“作家”负责根据资料撰写文章。它们通过共享的 State 进行协作。

2.1 环境与依赖配置

首先，确保你的 Python 环境版本在 3.8 以上。建议使用虚拟环境。通过pip安装核心依赖。

# 创建并激活虚拟环境（可选） python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心库 pip install langgraph langchain-openai langchain-community

关键依赖说明：

• langgraph: 核心工作流编排库。
• langchain-openai: 官方维护的 OpenAI 集成，用于调用 GPT 模型。
• langchain-community: 包含大量社区贡献的第三方集成，如网络搜索工具。

你还需要准备一个可用的 OpenAI API 密钥，并将其设置为环境变量。

# Linux/Mac export OPENAI_API_KEY='your-api-key-here' # Windows (PowerShell) $env:OPENAI_API_KEY='your-api-key-here'

2.2 项目结构规划

一个清晰的项目结构有助于管理复杂度。建议如下：

multi_agent_writer/ ├── agents/ │ ├── __init__.py │ ├── researcher.py # 研究员智能体定义 │ └── writer.py # 作家智能体定义 ├── graph/ │ ├── __init__.py │ └── workflow.py # LangGraph 工作流定义 ├── state.py # 共享状态定义 ├── tools.py # 自定义工具定义（如搜索） ├── config.py # 配置管理（API密钥等） └── main.py # 应用入口

我们先从定义共享状态开始。

3. 定义工作流状态与智能体节点

状态是工作流的基石，必须首先明确。

3.1 使用 TypedDict 定义 State

在state.py中，我们使用TypedDict来清晰地定义状态结构。这比普通字典更利于类型检查和代码提示。

# state.py from typing import TypedDict, List, Optional class AgentState(TypedDict): """ 多智能体写作助手的工作流状态。 所有节点都读写此状态的字段。 """ # 输入 topic: str # 用户指定的写作主题 # 研究员智能体的输出 research_materials: List[str] # 搜索到的原始材料列表 research_summary: str # 研究摘要 # 作家智能体的输出 article_outline: str # 文章大纲 final_article: str # 最终文章 # 控制流标志 needs_more_research: bool # 是否需要进一步研究（用于循环控制）

3.2 构建研究员智能体节点

研究员智能体的职责是：根据主题进行搜索，并生成一份摘要。我们在agents/researcher.py中实现。

首先，需要定义一个搜索工具。这里我们使用 LangChain 社区的 DuckDuckGo 搜索工具作为示例（注意：生产环境可能需要更稳定或付费的搜索 API）。

# tools.py from langchain_community.tools import DuckDuckGoSearchRun # 实例化一个搜索工具 search_tool = DuckDuckGoSearchRun()

然后实现研究员节点：

# agents/researcher.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from .tools import search_tool # 导入搜索工具 from typing import Dict # 初始化 LLM llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5) # 使用一个较小、较快的模型进行研究 def research_node(state: Dict) -> Dict: """ 研究员节点：执行搜索并总结。 输入：state (包含 ‘topic‘) 输出：更新后的 state (包含 ‘research_materials‘, ‘research_summary‘) """ topic = state[‘topic‘] # 1. 执行搜索 print(f“[研究员] 正在搜索主题: {topic}“) search_results = search_tool.invoke(topic) # 注意：实际返回可能是大段文本，这里简单处理为列表的一项 materials = [search_results] if search_results else [] # 2. 生成研究摘要 prompt_template = ChatPromptTemplate.from_messages([ (“system“, “你是一个专业的研究员。请根据提供的网络搜索材料，生成一份简洁、关键点突出的摘要。“), (“human“, “主题：{topic}\n\n原始材料：{materials}\n\n请生成研究摘要：“) ]) research_chain = prompt_template | llm | StrOutputParser() summary = research_chain.invoke({“topic“: topic, “materials“: materials}) print(f“[研究员] 研究摘要生成完成，长度：{len(summary)} 字符“) # 3. 返回状态更新 return { “research_materials“: materials, “research_summary“: summary, “needs_more_research“: False # 默认一次研究足够，后续可由其他节点修改 }

3.3 构建作家智能体节点

作家智能体的职责是：基于研究摘要，先撰写大纲，再撰写完整文章。我们在agents/writer.py中实现。

# agents/writer.py from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from typing import Dict # 作家可以使用一个更有创造力的模型 writer_llm = ChatOpenAI(model=“gpt-4“, temperature=0.7) def outline_node(state: Dict) -> Dict: """ 作家节点 - 步骤1：撰写大纲。 输入：state (包含 ‘topic‘, ‘research_summary‘) 输出：更新后的 state (包含 ‘article_outline‘) """ topic = state[‘topic‘] summary = state[‘research_summary‘] prompt = ChatPromptTemplate.from_messages([ (“system“, “你是一位经验丰富的技术作家。请根据研究摘要，为即将撰写的文章创建一份逻辑清晰、结构完整的大纲。“), (“human“, “文章主题：{topic}\n研究摘要：{summary}\n\n请输出文章大纲：“) ]) outline_chain = prompt | writer_llm | StrOutputParser() outline = outline_chain.invoke({“topic“: topic, “summary“: summary}) print(f“[作家] 文章大纲已创建“) return {“article_outline“: outline} def write_node(state: Dict) -> Dict: """ 作家节点 - 步骤2：撰写完整文章。 输入：state (包含 ‘topic‘, ‘research_summary‘, ‘article_outline‘) 输出：更新后的 state (包含 ‘final_article‘) """ topic = state[‘topic‘] summary = state[‘research_summary‘] outline = state[‘article_outline‘] prompt = ChatPromptTemplate.from_messages([ (“system“, “你是一位优秀的作家。请严格按照提供的大纲，并充分参考研究摘要，撰写一篇关于给定主题的完整、流畅、信息丰富的文章。文章应面向技术开发者。“), (“human“, “主题：{topic}\n研究摘要：{summary}\n文章大纲：{outline}\n\n请开始撰写文章：“) ]) write_chain = prompt | writer_llm | StrOutputParser() article = write_chain.invoke({“topic“: topic, “summary“: summary, “outline“: outline}) print(f“[作家] 文章撰写完成，长度：{len(article)} 字符“) return {“final_article“: article}

4. 组装 LangGraph 工作流

这是最关键的一步，我们将把分散的节点和状态组装成一个可执行的工作流图。在graph/workflow.py中操作。

4.1 创建图并添加节点

# graph/workflow.py from langgraph.graph import StateGraph, END from typing import Dict # 导入状态定义和节点函数 from ..state import AgentState from ..agents.researcher import research_node from ..agents.writer import outline_node, write_node def create_workflow() -> StateGraph: """ 创建并返回多智能体写作工作流图。 """ # 1. 初始化图，并指定状态结构为 AgentState workflow = StateGraph(AgentState) # 2. 添加节点 # 节点名是后续连接边时的标识符 workflow.add_node(“researcher“, research_node) workflow.add_node(“outline_writer“, outline_node) workflow.add_node(“article_writer“, write_node) # 3. 设置入口点：工作流从 ‘researcher‘ 节点开始 workflow.set_entry_point(“researcher“) # 4. 添加边，定义执行顺序 # 研究员完成后，进入大纲撰写 workflow.add_edge(“researcher“, “outline_writer“) # 大纲撰写完成后，进入文章撰写 workflow.add_edge(“outline_writer“, “article_writer“) # 文章撰写完成后，工作流结束 workflow.add_edge(“article_writer“, END) # 5. 编译图，得到一个可执行对象 compiled_workflow = workflow.compile() return compiled_workflow

目前，我们构建的是一个简单的线性工作流：研究员 -> 大纲作家 -> 文章作家 -> END。这已经是一个可运行的多智能体系统。

4.2 引入条件边与循环

为了让系统更智能，我们可以增加一个“评审”环节：让研究员评估作家生成的大纲，如果认为资料不足，则触发新一轮研究。这需要用到条件边。

首先，在agents/researcher.py中添加一个评审节点：

# agents/researcher.py (新增函数) def review_outline_node(state: Dict) -> Dict: """ 评审节点：评估大纲是否基于充分的研究。 输入：state (包含 ‘research_summary‘, ‘article_outline‘) 输出：更新后的 state (主要更新 ‘needs_more_research‘ 标志) """ summary = state[‘research_summary‘] outline = state[‘article_outline‘] prompt = ChatPromptTemplate.from_messages([ (“system“, “你是一个严格的评审员。请判断当前的文章大纲是否已经充分利用了已有的研究摘要。如果大纲中的关键点在研究摘要中缺乏足够依据，则判定为需要更多研究。只回答 ‘是‘ 或 ‘否‘。“), (“human“, “研究摘要：{summary}\n\n文章大纲：{outline}\n\n是否需要更多研究来支撑大纲？（是/否）：“) ]) review_chain = prompt | llm | StrOutputParser() decision = review_chain.invoke({“summary“: summary, “outline“: outline}) needs_more = decision.strip().lower() == ‘是‘ print(f“[评审员] 判定 ‘需要更多研究‘: {needs_more}“) return {“needs_more_research“: needs_more}

然后，修改graph/workflow.py，引入条件逻辑：

# graph/workflow.py (更新版) from langgraph.graph import StateGraph, END from langgraph.graph import START from typing import Dict, Literal # ... 其他导入 ... def create_workflow() -> StateGraph: workflow = StateGraph(AgentState) # 添加所有节点 workflow.add_node(“researcher“, research_node) workflow.add_node(“outline_writer“, outline_node) workflow.add_node(“reviewer“, review_outline_node) # 新增评审节点 workflow.add_node(“article_writer“, write_node) workflow.set_entry_point(“researcher“) # 修改边连接 # 研究员 -> 大纲作家 workflow.add_edge(“researcher“, “outline_writer“) # 大纲作家 -> 评审员 workflow.add_edge(“outline_writer“, “reviewer“) # 关键：从评审员出发的条件边 def decide_after_review(state: AgentState) -> Literal[“more_research“, “write_article“]: """根据评审结果决定下一步""" if state.get(“needs_more_research“, False): return “more_research“ # 需要跳回研究员节点 else: return “write_article“ # 可以继续写文章 workflow.add_conditional_edges( “reviewer“, # 源节点 decide_after_review, # 路由函数 { “more_research“: “researcher“, # 如果返回 “more_research“， 则前往 researcher 节点 “write_article“: “article_writer“, # 如果返回 “write_article“， 则前往 article_writer 节点 } ) # 文章作家 -> END workflow.add_edge(“article_writer“, END) # 注意：当流程跳回 ‘researcher‘ 时，它会再次经过 ‘outline_writer‘ -> ‘reviewer‘。 # 这形成了一个潜在的循环，直到评审通过。 # 为了避免无限循环，可以在 research_node 中增加逻辑，限制研究次数。 compiled_workflow = workflow.compile() return compiled_workflow

现在，工作流具备了基本的反馈循环能力：研究员 -> 大纲作家 -> 评审员 -> (若需要) 研究员 … -> 文章作家 -> END。

5. 运行、验证与结果分析

5.1 创建应用入口并运行

在main.py中，我们初始化工作流并传入初始状态。

# main.py from graph.workflow import create_workflow def main(): # 1. 编译工作流 print(“正在编译工作流...“) app = create_workflow() # 2. 定义初始状态 initial_state = { “topic“: “LangGraph 在多智能体系统中的应用与最佳实践“, “research_materials“: [], “research_summary“: ““, “article_outline“: ““, “final_article“: ““, “needs_more_research“: False, } # 3. 运行工作流 print(f“开始执行工作流，主题: {initial_state[‘topic‘]}“) print(“-“ * 50) final_state = app.invoke(initial_state) # 4. 输出结果 print(“\n“ + “=“ * 50) print(“工作流执行完成！“) print(“=“ * 50) print(f“\n最终生成的文章预览（前500字符）:\n{final_state[‘final_article‘][:500]}...“) print(f“\n文章总长度: {len(final_state[‘final_article‘])} 字符“) # 5. （可选）保存结果到文件 with open(‘output_article.md‘, ‘w‘, encoding=‘utf-8‘) as f: f.write(f“# {final_state[‘topic‘]}\n\n“) f.write(final_state[‘final_article‘]) print(“\n文章已保存至 ‘output_article.md‘“) if __name__ == “__main__“: main()

运行python main.py，观察控制台输出。你会看到类似下面的日志，清晰地展示了智能体间的协作过程：

正在编译工作流... 开始执行工作流，主题: LangGraph 在多智能体系统中的应用与最佳实践 -------------------------------------------------- [研究员] 正在搜索主题: LangGraph 在多智能体系统中的应用与最佳实践 [研究员] 研究摘要生成完成，长度：1200 字符 [作家] 文章大纲已创建 [评审员] 判定 ‘需要更多研究‘: False [作家] 文章撰写完成，长度：3200 字符 ================================================== 工作流执行完成！ ================================================== ...

5.2 状态流转可视化

LangGraph 的一个强大功能是可视化。你可以将编译后的图导出为图片，直观理解工作流。

# 在 main.py 中添加（需要安装 graphviz） try: # 显示图结构 from IPython.display import Image, display # 如果你在 Jupyter 环境中 display(Image(app.get_graph().draw_mermaid_png())) except: # 或者保存为文件 app.get_graph().draw_mermaid_png(output_file_path=“workflow_graph.png“) print(“工作流图已保存为 ‘workflow_graph.png‘“)

生成的图会清晰显示节点、普通边和条件边，是理解和调试复杂工作流的利器。

6. 常见问题与深度排查

在实际开发中，你可能会遇到以下典型问题。

6.1 状态（State）相关错误

6.2 图编译与执行错误

6.3 工具与模型调用错误

6.4 性能与成本优化

• 控制循环次数：对于可能循环的路径（如研究-评审循环），务必在状态中设置max_research_cycles并在路由函数中检查，防止因模型判断偏差导致无限循环和 API 费用激增。
• 选择性使用大模型：在成本敏感的场景下，可以将不同的节点分配给不同能力的模型。例如，研究摘要使用gpt-4o-mini，而创造性写作使用gpt-4。
• 状态序列化：如果状态对象很大（例如包含长文本列表），频繁在节点间传递会影响性能。考虑只传递必要的引用或 ID。

7. 生产环境最佳实践与扩展方向

将 LangGraph 工作流用于生产环境，需要考虑更多工程化因素。

7.1 配置与密钥管理

切勿将 API 密钥硬编码在代码中。使用环境变量或专业的配置管理工具（如python-dotenv,pydantic-settings）。

# config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str openai_base_url: str | None = None # 如需使用代理 model_for_research: str = “gpt-4o-mini“ model_for_writing: str = “gpt-4“ class Config: env_file = “.env“ settings = Settings()

然后在节点中从settings对象获取配置来初始化 LLM。

7.2 持久化与检查点

LangGraph 内置了检查点（Checkpoint）机制，可以持久化工作流状态，这对于长时间运行或可能中断的任务至关重要。

from langgraph.checkpoint import MemorySaver # 在编译图时传入检查点存储器 memory = MemorySaver() compiled_workflow = workflow.compile(checkpointer=memory) # 调用时传入一个 thread_id，用于标识会话 config = {“configurable“: {“thread_id“: “user_123_session_1“}} final_state = compiled_workflow.invoke(initial_state, config=config) # 后续可以从检查点恢复执行 # final_state = compiled_workflow.invoke(new_input, config=config)

生产环境应使用如Redis、PostgreSQL等外部存储的检查点实现。

7.3 日志、监控与可观测性

• 结构化日志：使用logging模块替代print，记录节点开始/结束、状态变化、工具调用结果和耗时。
• 链路追踪：集成 OpenTelemetry 等工具，追踪一次工作流调用在所有节点和外部服务（LLM API、工具 API）上的性能数据。
• 状态快照：在关键节点后，将重要状态字段记录到数据库或监控系统，便于事后分析和调试。

7.4 扩展方向

1. 动态工具调用：将 LangChain 的ToolExecutor和AgentExecutor作为节点，构建可以自主选择工具的智能体节点。
2. 并行执行：LangGraph 支持Pregel风格的并发执行。对于相互无依赖的节点，可以配置并行分支，提升整体效率。
3. 人工干预节点：在流程中插入一个“人工审核”节点，当模型置信度低或遇到敏感内容时，暂停工作流，等待人工输入后再继续。
4. 与 Web 框架集成：将编译好的app对象封装成 FastAPI 或 Django 的一个服务端点，提供 RESTful API 供前端调用。
5. 更复杂的状态管理：使用Pydantic模型替代TypedDict，获得更强大的数据验证和序列化能力。使用Annotation语法更精细地控制每个节点对状态的读写权限。

通过本文的步骤，你不仅搭建了一个可运行的多智能体系统，更重要的是掌握了 LangGraph 以状态为中心的编排思想。在实际项目中，应从最简单的线性流开始，逐步引入条件分支和循环，并始终关注状态的结构设计和节点的单一职责。当智能体数量和交互逻辑变得复杂时，清晰定义的图和状态将成为维护系统可理解性的最重要工具。