LangGraph多智能体系统实战：从架构到代码的完整指南

这次我们来看一个关于 LangGraph 多智能体实战的教程资源。这个标题指向的并非一个具体的开源项目，而是一套围绕 LangGraph 框架构建多智能体系统的实战教程内容。它的核心价值在于，将 LangGraph 这个用于编排复杂、有状态 Agent 工作流的强大框架，与“多智能体”这一热门方向结合，提供了从架构理解、核心组件拆解到代码落地的完整路径。对于想深入 Agent 开发，尤其是构建多个 Agent 协同工作的开发者来说，这类实战指南非常关键。

LangGraph 本身是 LangChain 生态的一部分，但它更专注于用“图”的思维来定义和控制 Agent 的执行流。与传统的线性链式调用不同，它允许你设计带条件分支、循环和状态管理的复杂工作流。当多个具备不同能力的 Agent（如分析、决策、执行）被组织进这样一个图中时，就构成了一个多智能体系统。本文将基于这一主题，为你梳理如何利用 LangGraph 搭建多智能体应用，涵盖从核心概念、环境搭建、基础组件编码到一个完整多智能体工作流实战的全过程。

1. 核心能力速览

2. 适用场景与使用边界

LangGraph 多智能体系统并非万能，理解其适用边界能帮助你更好地决策。

它非常适合以下场景：

1. 复杂任务自动化：例如，一个需求从“用户输入”开始，需要经历“需求分析”、“信息检索”、“代码生成”、“安全检查”、“结果汇总”等多个由不同 Agent 负责的阶段。
2. 模拟与仿真：构建多个角色（Agent）在特定规则下进行交互的环境，如市场模拟、游戏 NPC 决策等。
3. 决策支持系统：需要根据中间结果动态选择下一步执行路径的应用程序。
4. 长上下文/有状态对话：构建能记住历史交互、并根据上下文决定回复策略的聊天机器人或客服助手。

需要注意的使用边界：

1. 并非替代简单链：对于简单的“输入-处理-输出”线性任务，直接使用 LangChain 的 LCEL 或简单 Chain 更轻量高效。
2. 复杂性带来开销：图结构的设计、状态管理和多 Agent 间的通信会引入额外的开发和运行时开销。
3. 依赖底层模型能力：整个系统的智能上限取决于集成的各个大模型的能力。垃圾进，垃圾出。
4. 合规与授权：如果智能体处理用户数据、生成内容或做出影响现实的决策，必须考虑数据隐私、内容合规性和责任归属。

3. 环境准备与前置条件

在开始编码前，需要准备好基础的 Python 开发环境。

1. Python 版本：建议使用 Python 3.10 或 3.11，这是大多数 AI 框架兼容性较好的版本。
2. 包管理工具：使用pip或conda进行依赖管理。推荐创建独立的虚拟环境。
3. 核心依赖：核心库是langgraph。通常还会安装langchain以使用其丰富的组件（如提示模板、输出解析器、工具集）。如果需要调用 OpenAI 等模型，还需安装对应的 SDK。
4. 模型接入：准备好大模型的访问方式。可以是：
   • 云端 API：如 OpenAI、Anthropic、智谱 AI 等，需要相应的 API Key。
   • 本地模型：通过ollama、vLLM或transformers库部署，这会涉及 GPU 显存和本地推理服务。
5. 代码编辑器：任何你熟悉的 IDE，如 VS Code、PyCharm。

下面是一个基础环境配置清单：

# 创建并激活虚拟环境 (以 conda 为例) conda create -n langgraph-agent python=3.11 conda activate langgraph-agent # 安装核心库 pip install langgraph langchain # 按需安装模型 SDK 和工具库 pip install openai anthropic-ai # 例如使用 OpenAI 或 Anthropic # 或安装本地模型支持 # pip install ollama # pip install transformers torch # 安装开发常用工具 pip install jupyterlab ipython # 用于交互式实验 pip install pydantic # LangChain 常用作数据验证

4. LangGraph 核心组件与概念拆解

理解 LangGraph 的几个核心概念，是构建多智能体系统的基石。

4.1 状态（State）

状态是一个字典（或 Pydantic 模型），它随着工作流的执行而不断更新。它是在整个图中传递和修改的共享信息载体。通常，你会定义一个包含所有必要字段的 State 类型。

from typing import TypedDict, Annotated from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 消息历史，LangGraph 提供了便捷的注解来处理消息列表 messages: Annotated[list, add_messages] # 其他自定义状态字段 task: str current_step: str intermediate_results: dict

4.2 节点（Node）

节点是工作流中的一个执行单元，通常对应一个函数。这个函数接收当前的State，执行一些操作（如调用 LLM、运行工具），然后返回一个更新后的State。在多智能体系统中，每个节点可以视为一个具有特定职能的 Agent。

def analysis_agent(state: AgentState) -> AgentState: """分析 Agent：解析任务，决定下一步。""" from langchain_core.messages import HumanMessage # 从状态中获取最新消息或任务 task = state.get('task', '') # 这里模拟调用 LLM 进行分析 # analysis_result = llm.invoke(f"请分析任务：{task}") # 更新状态 new_state = state.copy() new_state['current_step'] = 'analysis_complete' new_state['intermediate_results']['analysis'] = '这是一个需要多步处理的任务。' # 也可以向消息历史中添加消息 new_state['messages'].append(HumanMessage(content=f"分析完成。结论：{new_state['intermediate_results']['analysis']}")) return new_state

4.3 边（Edge）

边定义了节点之间的流转关系。分为两种：

• 普通边：无条件地从上一个节点指向下一个节点。
• 条件边：根据State中的某个条件，决定下一步走向哪个节点。这通过一个路由函数（RunnableLambda）来实现。

4.4 图（Graph）与编译器

将节点和边组合起来，就形成了一个StateGraph。最后，需要调用compile()方法将其编译成一个可执行的Runnable对象，这个对象可以像函数一样被调用。

5. 构建一个基础的多智能体工作流

让我们构建一个简单的“任务处理”多智能体系统，包含三个 Agent：分析员、研究员、撰稿人。

5.1 定义状态与节点函数

from typing import TypedDict, Annotated from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, SystemMessage, AIMessage import operator # 1. 定义状态 class MultiAgentState(TypedDict): messages: Annotated[list, add_messages] # 消息历史 original_task: str # 原始任务 research_findings: str # 研究结果 final_report: str # 最终报告 # 2. 定义节点函数（模拟 Agent） def analysis_node(state: MultiAgentState) -> MultiAgentState: """分析节点：理解任务，规划步骤。""" task = state['original_task'] # 模拟分析逻辑 analysis_plan = f"任务『{task}』需要先进行研究，然后撰写报告。" new_messages = state['messages'] + [AIMessage(content=f"分析员：我已理解任务。计划：{analysis_plan}")] return {"messages": new_messages, "research_findings": "", "final_report": ""} def research_node(state: MultiAgentState) -> MultiAgentState: """研究节点：根据任务进行信息检索（模拟）。""" task = state['original_task'] # 模拟研究过程 findings = f"关于『{task}』的研究发现：这是一个热门技术话题，涉及图结构和状态管理。关键工具有LangGraph。" new_messages = state['messages'] + [AIMessage(content=f"研究员：研究完成。发现：{findings}")] return {"research_findings": findings} def write_node(state: MultiAgentState) -> MultiAgentState: """撰稿节点：综合分析和研究结果，生成报告。""" task = state['original_task'] findings = state['research_findings'] # 模拟撰写过程 report = f"任务报告：{task}\n\n基于研究（{findings}），我们建议采用LangGraph构建多智能体系统，其核心优势是...（此处省略详细内容）" new_messages = state['messages'] + [AIMessage(content=f"撰稿人：报告撰写完毕。")] return {"final_report": report, "messages": new_messages}

5.2 构建图并编译

from langgraph.graph import StateGraph, END # 3. 创建状态图 workflow = StateGraph(MultiAgentState) # 4. 添加节点 workflow.add_node("analyst", analysis_node) workflow.add_node("researcher", research_node) workflow.add_node("writer", write_node) # 5. 设置入口点 workflow.set_entry_point("analyst") # 6. 添加边，定义执行流：analyst -> researcher -> writer -> END workflow.add_edge("analyst", "researcher") workflow.add_edge("researcher", "writer") workflow.add_edge("writer", END) # 7. 编译图 app = workflow.compile()

5.3 执行工作流并查看结果

# 8. 定义初始状态并运行 initial_state = { "messages": [HumanMessage(content="请帮我写一份关于LangGraph多智能体系统的技术报告。")], "original_task": "撰写LangGraph多智能体系统技术报告", "research_findings": "", "final_report": "" } # 执行图 final_state = app.invoke(initial_state) # 查看最终状态 print("=== 最终报告 ===") print(final_state["final_report"]) print("\n=== 完整消息历史 ===") for msg in final_state["messages"]: print(f"{type(msg).__name__}: {msg.content}")

这个简单的线性流程展示了多智能体协作的基本形态。但 LangGraph 的强大之处在于支持更复杂的逻辑。

6. 实现高级特性：条件路由与循环

让我们的系统更智能：如果研究结果不够充分，则循环回去继续研究。

6.1 修改研究节点和状态

首先，为状态增加一个“研究轮次”计数器。

class AdvancedAgentState(TypedDict): messages: Annotated[list, add_messages] original_task: str research_findings: list # 改为列表，存储多轮研究结果 research_round: int # 研究轮次 final_report: str

6.2 创建带条件判断的研究节点

def research_node_with_check(state: AdvancedAgentState) -> AdvancedAgentState: """研究节点：模拟研究，并判断结果是否充分。""" task = state['original_task'] round_num = state.get('research_round', 0) + 1 # 模拟研究，轮次越多，“发现”越多 finding = f"第{round_num}轮研究：发现要点{round_num}..." new_findings = state['research_findings'] + [finding] new_messages = state['messages'] + [AIMessage(content=f"研究员（第{round_num}轮）：{finding}")] return { "research_findings": new_findings, "research_round": round_num, "messages": new_messages } def should_continue_research(state: AdvancedAgentState) -> str: """条件路由函数：判断是否继续研究。""" # 简单逻辑：如果研究轮次少于2轮，就继续研究，否则去撰写 if state.get('research_round', 0) < 2: return "continue_research" else: return "proceed_to_write"

6.3 构建带循环的图

from langgraph.graph import StateGraph, END workflow = StateGraph(AdvancedAgentState) workflow.add_node("analyst", analysis_node) # 复用之前的分析节点，需适配状态 workflow.add_node("researcher", research_node_with_check) workflow.add_node("writer", write_node) # 复用之前的撰写节点，需适配状态 workflow.set_entry_point("analyst") workflow.add_edge("analyst", "researcher") # 关键：添加条件边 workflow.add_conditional_edges( "researcher", should_continue_research, # 条件判断函数 { "continue_research": "researcher", # 返回 `continue_research` 则循环回 researcher 节点 "proceed_to_write": "writer" # 返回 `proceed_to_write` 则前往 writer 节点 } ) workflow.add_edge("writer", END) app = workflow.compile()

执行这个图，你会看到“研究员”节点被执行了两次，然后才进入“撰稿人”节点。这实现了基于状态的动态工作流控制。

7. 集成真实 LLM 与工具

前面的例子模拟了 LLM 调用。现在，我们集成真实的 OpenAI API 和一个搜索工具。

7.1 环境变量与模型设置

# 在终端设置你的 OpenAI API Key # export OPENAI_API_KEY='your-api-key-here'

import os from langchain_openai import ChatOpenAI from langchain_community.tools import DuckDuckGoSearchRun # 初始化 LLM 和工具 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 使用 gpt-4o-mini 控制成本 search_tool = DuckDuckGoSearchRun() # 为不同 Agent 定义提示词 analyst_system_prompt = """你是一个任务分析专家。请根据用户的任务，将其分解为清晰的研究子问题。输出格式：以列表形式给出2-3个关键研究问题。""" researcher_system_prompt = """你是一个网络研究员。请基于给定的研究问题，使用搜索工具获取最新、最相关的信息，并总结成要点。""" writer_system_prompt = """你是一个技术文档撰稿人。请根据任务描述和研究发现，撰写一份结构清晰、内容详实的技术报告大纲。"""

7.2 构建集成真实组件的节点

from langchain_core.prompts import ChatPromptTemplate from langgraph.prebuilt import ToolExecutor, ToolInvocation import json tool_executor = ToolExecutor([search_tool]) def analysis_agent_llm(state: AdvancedAgentState): """使用 LLM 进行分析的节点""" prompt = ChatPromptTemplate.from_messages([ ("system", analyst_system_prompt), ("user", "原始任务：{task}") ]) chain = prompt | llm task = state['original_task'] response = chain.invoke({"task": task}) # 更新状态，例如将分析结果存入消息或特定字段 new_messages = state['messages'] + [AIMessage(content=f"分析完成。分解的研究问题：{response.content}")] return {"messages": new_messages, "research_questions": response.content} def research_agent_llm(state: AdvancedAgentState): """使用 LLM 和工具进行研究。这是一个简化示例，实际需要处理工具调用循环。""" research_question = state.get('research_questions', '').split('\n')[0] # 取第一个问题 # 让 LLM 决定是否调用工具 prompt = ChatPromptTemplate.from_messages([ ("system", researcher_system_prompt), ("human", f"请研究：{research_question}") ]) # 此处应为更复杂的、支持工具调用的 Agent 执行器，例如使用 `create_react_agent` # 为简化，我们直接调用搜索工具 search_result = search_tool.invoke(research_question) summary = llm.invoke(f"请总结以下搜索结果：\n\n{search_result}") new_findings = state['research_findings'] + [f"Q: {research_question}\nA: {summary.content}"] new_messages = state['messages'] + [AIMessage(content=f"研究员：关于『{research_question}』的研究总结完成。")] return {"research_findings": new_findings, "messages": new_messages}

在实际项目中，你会使用langgraph.prebuilt.create_react_agent或ToolNode来更优雅地处理工具调用循环。这里展示了集成的核心思路。

8. 持久化与并发：Checkpointer 与多线程

对于生产环境，状态持久化和并发处理很重要。

8.1 使用 Checkpointer 持久化状态

Checkpointer 允许工作流在中断后从上次的状态恢复。

from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph # 创建 SQLite 检查点存储器 memory = SqliteSaver.from_conn_string(":memory:") # 内存数据库，也可用文件路径 workflow = StateGraph(AdvancedAgentState, config_schema=...) # ... 添加节点和边 ... # 编译时传入 checkpointer app = workflow.compile(checkpointer=memory) # 使用 config 中的 `thread_id` 来区分不同的会话/线程 config = {"configurable": {"thread_id": "user_session_123"}} initial_state = {...} # 第一次调用 result1 = app.invoke(initial_state, config=config) # 模拟中断后，再次调用，可以从上次的断点继续（取决于图的设计） # result2 = app.invoke({}, config=config) # 传入空状态或部分状态

8.2 并发执行考虑

LangGraph 编译后的app是一个Runnable，可以集成到像FastAPI这样的 Web 框架中，天然支持并发请求。每个请求使用不同的thread_id，其状态通过 Checkpointer 隔离。

from fastapi import FastAPI from pydantic import BaseModel app_fastapi = FastAPI() class TaskRequest(BaseModel): task_description: str user_id: str @app_fastapi.post("/process") async def process_task(request: TaskRequest): config = {"configurable": {"thread_id": request.user_id}} initial_state = { "messages": [HumanMessage(content=request.task_description)], "original_task": request.task_description, # ... 其他初始字段 } # 注意：在异步环境中使用适当的调用方式 final_state = await app.ainvoke(initial_state, config=config) return {"report": final_state.get("final_report")}

9. 常见问题与排查方法

在开发和运行 LangGraph 多智能体应用时，你可能会遇到以下问题：

10. 最佳实践与使用建议

1. 始于简单，迭代复杂：不要一开始就设计庞大的智能体网络。从一个线性流程开始，验证每个节点工作正常，再逐步添加条件分支、循环和更多 Agent。
2. 明确状态设计：花时间精心设计State的结构。它是智能体之间通信的唯一渠道。字段应清晰、必要，避免冗余。
3. 为节点函数编写清晰的文档：每个节点函数（Agent）应该像微服务一样，有明确的输入（State 的哪些字段）、处理逻辑和输出（更新哪些 State 字段）。这有助于团队协作和后期维护。
4. 充分利用预构建组件：LangGraph 和 LangChain 提供了ToolNode、create_react_agent、messages_state_modifier等高级抽象。在理解基础原理后，尽量使用这些经过测试的组件来提高开发效率和稳定性。
5. 实施全面的日志记录：在图执行的关键点（如进入/退出节点、条件判断结果）添加日志。这对于调试复杂的工作流至关重要。可以考虑使用langgraph的Events机制。
6. 进行彻底的测试：为每个节点函数编写单元测试。为整个图编写集成测试，模拟不同的输入和状态路径。测试条件分支和循环的边界情况。
7. 安全性考量：如果智能体可以执行网络搜索、代码执行、文件操作等工具，必须实施严格的输入验证和权限控制，防止恶意指令或提示词注入攻击。
8. 成本监控：当使用付费 API 时，在多智能体系统中成本可能快速增长。在代码中集成 token 计数和成本估算，为工作流设置预算或中断机制。

构建 LangGraph 多智能体系统是一个将复杂业务逻辑清晰化的过程。它强迫你将问题分解成模块化的、可管理的步骤（节点），并明确它们之间的依赖和规则（边）。虽然学习曲线存在，但一旦掌握，你将拥有一个极其灵活和强大的框架，来应对各类需要自动化智能协作的场景。从今天介绍的基础架构和代码实战出发，你可以尝试将其应用到你的具体领域，如自动化客服、智能编码助手、游戏 AI 或数据分析流水线中。