最近在尝试构建一个复杂的AI应用时，你是否遇到过这样的困境：单个大语言模型（LLM）智能体在面对需要多步骤推理、调用多个工具或处理多个专业领域的任务时，显得力不从心？要么是上下文窗口迅速被填满，要么是工具调用决策混乱，最终导致任务失败或结果质量低下。这正是多智能体（Multi-Agent）系统要解决的核心问题。

本文将为你带来一份从零到一的AI Agent智能体保姆级实战教程。我们将以当前最流行的LangChain和LangGraph框架为核心，结合OpenClaw等工具，手把手教你搭建一个可运行、可扩展的多智能体协作系统。无论你是想入门AI应用开发，还是希望将现有单智能体项目升级为更强大的多智能体架构，这篇文章都将提供完整的代码示例、清晰的架构解析和避坑指南。

1. 背景与核心概念：为什么需要多智能体？

在深入代码之前，我们首先要理解多智能体系统（Multi-Agent System, MAS）的价值和核心思想。

1.1 什么是AI智能体（Agent）？

简单来说，一个AI智能体是一个能够感知环境、进行决策并执行动作以实现特定目标的系统。在LLM应用开发中，智能体通常指一个由大语言模型驱动的程序，它可以根据用户输入、上下文信息和可用工具（如搜索、计算、数据库查询）来决定下一步做什么。传统的ReAct（Reasoning + Acting）模式就是单智能体的典型代表。

然而，随着任务复杂度的提升，单智能体架构暴露出诸多局限性：

1. 工具过载：当智能体拥有数十个甚至上百个工具时，LLM在决定调用哪个工具时容易做出错误决策。
2. 上下文爆炸：长对话历史、中间思考步骤、工具调用结果全部塞进上下文，很快会触及模型token限制。
3. 缺乏专业化：一个“通才”智能体很难同时在代码生成、数学计算、文本总结等多个领域都表现出色。

1.2 多智能体系统的优势

多智能体系统通过将复杂任务分解，交由多个专业化、模块化的智能体协作完成，从而有效应对上述挑战。其核心优势在于：

• 模块化：每个智能体职责单一，易于开发、测试、维护和替换。
• 专业化：可以训练或提示（Prompt）不同的智能体成为特定领域的专家（如“代码专家”、“数据分析师”、“文案写手”），提升整体任务完成质量。
• 可控性：开发者可以显式地设计智能体之间的通信协议和控制流，而不是完全依赖LLM的函数调用，使系统行为更可预测、更可靠。

1.3 核心框架：LangChain 与 LangGraph

• LangChain：一个用于开发由LLM驱动的应用程序的框架。它提供了连接LLM、数据源（如向量数据库）和工具（如搜索引擎、API）的标准化接口，是构建智能体的基石。
• LangGraph：构建在LangChain之上的库，专门用于创建有状态、多步骤的应用程序，其核心抽象是图（Graph）。在LangGraph中，智能体被建模为图的节点（Node），节点之间的交互和状态流转被定义为边（Edge）。这使其成为构建复杂多智能体工作流的理想选择。

两者关系：你可以把LangChain看作是提供了“砖块”（模型、工具、记忆等），而LangGraph提供了将这些“砖块”组装成复杂“建筑”（工作流）的蓝图和脚手架。本文的实战将主要基于LangGraph来编排多智能体。

1.4 什么是OpenClaw？

OpenClaw（开源爪）是一个由火山引擎开源的AI智能体开发框架与平台。它提供了一套完整的工具链和技能（Skill）市场，旨在降低AI智能体开发的门槛。在本文的语境中，我们可以将OpenClaw视为一个强大的工具和技能库，其提供的技能（如网页搜索、文件处理、代码执行等）可以被我们的LangGraph多智能体系统所调用。后续实战中，我们会演示如何将其集成。

2. 环境准备与依赖安装

工欲善其事，必先利其器。开始编码前，请确保你的开发环境已就绪。

2.1 基础环境要求

• Python: 3.8 或更高版本。推荐使用 3.9+ 以获得更好的兼容性。
• 包管理工具:pip或conda。
• 代码编辑器: VS Code, PyCharm 等任选。
• LLM API Key: 你需要一个支持工具调用（Function Calling）的LLM服务API Key，例如：
  • OpenAI API Key(推荐，兼容性最好)
  • 通义千问、DeepSeek、智谱AI等国内主流模型的API Key。

2.2 创建虚拟环境与安装依赖

强烈建议使用虚拟环境来管理项目依赖，避免包冲突。

# 1. 创建并进入项目目录 mkdir multi-agent-tutorial && cd multi-agent-tutorial # 2. 创建Python虚拟环境 (以venv为例) python -m venv venv # 3. 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 4. 安装核心依赖 pip install langchain langchain-openai langgraph # 5. 安装可选依赖（用于示例中的工具） pip install duckduckgo-search # 用于网络搜索 # pip install openclaw-sdk # 如果使用OpenClaw SDK，请根据其官方文档安装

2.3 设置API密钥

在项目根目录创建一个.env文件来安全地存储你的API密钥，并使用python-dotenv加载。

pip install python-dotenv

.env文件内容：

OPENAI_API_KEY=sk-your-openai-api-key-here # 其他模型的API_KEY，如需要 # DASHSCOPE_API_KEY=your-dashscope-key # ZHIPUAI_API_KEY=your-zhipuai-key

在代码中加载环境变量：

# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") # 确保密钥已设置 if not OPENAI_API_KEY: raise ValueError("请在 .env 文件中设置 OPENAI_API_KEY")

3. LangGraph核心概念与多智能体架构拆解

在动手搭建之前，我们必须深入理解LangGraph的几个核心概念，这是构建任何工作流的基础。

3.1 状态（State）

状态是LangGraph工作流的“记忆”。它是一个字典（或Pydantic模型），在图的各个节点间传递和更新。对于多智能体系统，状态通常包含所有智能体需要共享的信息，最常见的是消息列表。

from typing import Annotated, List from typing_extensions import TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): """定义多智能体系统的共享状态""" # 消息历史，所有智能体都读写这个列表 messages: Annotated[List, add_messages] # 其他共享信息，例如当前任务描述、中间结果等 task: str final_answer: str

add_messages是一个归约函数，它能智能地将新旧消息列表合并，是处理对话历史的推荐方式。

3.2 节点（Node）与边（Edge）

• 节点：代表工作流中的一个步骤或一个计算单元。在多智能体系统中，每个智能体通常被实现为一个节点。节点函数接收当前状态，执行操作（如调用LLM、运行工具），并返回更新后的状态。
• 边：定义了节点之间的流转逻辑。分为两种：
  • 条件边（Conditional Edge）：根据节点返回的结果或状态中的某个值，动态决定下一个要执行的节点。这是实现智能体间“移交”（Handoff）的关键。
  • 普通边（Fixed Edge）：始终指向同一个下一个节点。

3.3 图（Graph）与编译

将定义好的节点和边组装起来，就形成了一个StateGraph。调用compile()方法后，会得到一个可执行的CompiledGraph对象，你可以像调用函数一样invoke它。

3.4 多智能体经典架构模式

根据LangGraph官方文档，多智能体系统主要有以下几种连接模式：

1. 网络（Network）模式：

   • 描述：每个智能体（节点）都可以直接与其他任何智能体通信，并自行决定下一步调用谁。这是一种扁平化、去中心化的结构。
   • 适用场景：智能体之间关系对等，任务没有固定顺序，需要高度动态协作。

2. 主管（Supervisor）模式：

   • 描述：引入一个专用的“主管”智能体。所有工作智能体只与主管通信。主管接收任务和状态，然后决定下一步应该由哪个（或哪些）工作智能体执行，并将任务分配出去。工作智能体完成任务后，将结果返回给主管。
   • 适用场景：任务需要集中调度和协调，是最常用、最实用的模式。

3. 主管（工具调用）模式：

   • 描述：主管模式的一种变体。工作智能体被“包装”成工具（Tool），主管智能体本身是一个标准的工具调用LLM。主管通过调用“智能体工具”来分配任务。这本质上利用了LLM原生函数调用的能力来管理流程。
   • 适用场景：希望复用标准ReAct智能体模式，将子智能体管理简化为工具调用。

4. 层级式（Hierarchical）模式：

   • 描述：在主管模式上的扩展，形成树状结构。顶层主管管理多个团队主管，每个团队主管再管理自己的一组工作智能体。适用于超大型、模块化的智能体组织。
   • 适用场景：智能体数量众多，且可以按功能或领域自然分组。

5. 自定义工作流：

   • 描述：开发者显式地定义智能体之间的调用顺序和条件，混合使用固定边和条件边。提供最大的灵活性。
   • 适用场景：流程有部分确定性步骤，部分步骤需要LLM动态决策。

在接下来的实战中，我们将以主管模式为例，因为它结构清晰，易于理解和实现，是大多数场景下的最佳起点。

4. 实战：手把手搭建一个多智能体协作系统

我们将构建一个“技术问答助手”系统。用户提出一个复杂的技术问题（例如，“如何在Docker中部署一个带有Redis缓存的Django应用？”），系统将协同多个智能体来解答。

系统设计：

• 主管智能体（Supervisor）：分析用户问题，拆解任务，并调度其他智能体。
• 研究智能体（Researcher）：负责利用网络搜索工具（如DuckDuckGo）查找最新的外部信息。
• 代码智能体（Coder）：负责生成、解释或审查代码片段。
• 总结智能体（Summarizer）：负责整合各智能体的输出，生成最终友好、全面的回答。

4.1 定义共享状态与工具

首先，我们定义系统共享的状态和智能体将要使用的工具。

# agents/state.py from typing import Annotated, List, Optional from typing_extensions import TypedDict from langgraph.graph.message import add_messages class MultiAgentState(TypedDict): """ 多智能体系统的共享状态。 messages: 所有智能体间的对话历史。 task: 原始用户任务。 research_findings: 研究智能体的发现。 code_snippets: 代码智能体生成的代码。 final_output: 总结智能体生成的最终答案。 next_agent: 主管决定的下一个要执行的智能体名称。 """ messages: Annotated[List, add_messages] task: str research_findings: Optional[str] code_snippets: Optional[str] final_output: Optional[str] next_agent: Optional[str] # 用于主管路由

# agents/tools.py from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain_core.tools import tool # 1. 网络搜索工具 (使用 DuckDuckGo) search_tool = DuckDuckGoSearchRun() # 2. 自定义工具示例：代码格式化/检查（这里简化为一个占位符） @tool def code_linter(code: str) -> str: """对提供的代码进行简单的格式化和基础检查。""" # 这里可以集成真实的linter，如flake8, black等 # 此处仅作示例 print(f"[Linter] 收到代码片段，长度：{len(code)}") # 模拟一些检查 if "import os" in code and "os.getenv" in code: return "代码检查通过：使用了os模块读取环境变量，建议确保环境变量已设置。" return "代码格式检查完成，未发现明显问题（示例）。" # 将所有工具放入一个字典，方便按名称调用 TOOLS = { "search_web": search_tool, "lint_code": code_linter, }

4.2 实现各个智能体节点

每个智能体都是一个独立的函数，它接收状态，执行逻辑，并返回更新后的状态。

# agents/nodes.py from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langgraph.types import Command from .state import MultiAgentState from .tools import TOOLS import json # 初始化LLM模型 model = ChatOpenAI(model="gpt-4o", temperature=0) # 使用gpt-4o以获得更好的推理和工具调用能力 def supervisor_node(state: MultiAgentState) -> Command: """ 主管智能体节点。 职责：分析当前对话和任务，决定下一步该调用哪个智能体，或结束流程。 """ messages = state.get("messages", []) task = state.get("task", "") # 构建给主管的提示词 system_prompt = """你是一个多智能体系统的主管。你的任务是分析用户的问题和当前对话状态，然后决定下一步应该由哪个专家智能体来处理，或者直接给出最终答案。 可用的专家智能体有： - `researcher`: 当问题需要查找最新资料、文档或外部信息时调用。 - `coder`: 当问题涉及代码生成、代码解释、代码审查或技术实现细节时调用。 - `summarizer`: 当已有足够信息（research_findings, code_snippets）需要整合成最终答案时调用。 - `__end__`: 如果认为问题已得到圆满解答，或者无法处理，则结束流程。 请只输出一个JSON对象，格式如下： ```json { "next_agent": "agent_name", "reason": "你的决策理由" }

""" prompt = f"用户原始任务：{task}\n\n当前对话历史：{messages[-5:] if len(messages) > 5 else messages}" # 只取最近几条消息

full_messages = [SystemMessage(content=system_prompt), HumanMessage(content=prompt)] try: # 使用LLM进行结构化输出，强制其返回JSON response = model.with_structured_output( { "type": "object", "properties": { "next_agent": {"type": "string"}, "reason": {"type": "string"} }, "required": ["next_agent", "reason"] } ).invoke(full_messages) except: # 如果模型不支持structured_output，使用普通调用并解析 response_raw = model.invoke(full_messages) # 简单解析，实际项目需要更健壮的解析逻辑 import re json_match = re.search(r'\{.*\}', response_raw.content, re.DOTALL) if json_match: response = json.loads(json_match.group()) else: response = {"next_agent": "__end__", "reason": "无法解析主管决策"} next_agent = response.get("next_agent", "__end__") reason = response.get("reason", "") print(f"[Supervisor] 决定下一步调用: {next_agent}, 理由: {reason}") # 更新状态，并指示下一个节点 new_messages = messages + [AIMessage(content=f"主管决策：下一步调用 {next_agent}。理由：{reason}")] return Command( goto=next_agent, # 关键：使用Command对象进行路由 update={ "messages": new_messages, "next_agent": next_agent } )

def researcher_node(state: MultiAgentState): """研究智能体节点。负责搜索网络信息。""" messages = state.get("messages", []) task = state.get("task", "")

# 构建搜索查询。可以基于最近的消息或原始任务。 # 这里简单地将任务作为搜索词 search_query = task print(f"[Researcher] 正在搜索: {search_query}") try: search_result = TOOLS["search_web"].invoke(search_query) findings = f"针对查询 '{search_query}' 的搜索结果：\n{search_result[:2000]}" # 限制长度 except Exception as e: findings = f"搜索过程中出现错误：{e}" # 更新状态 new_messages = messages + [ AIMessage(content=f"我是研究智能体。我已经完成了信息搜索。"), AIMessage(content=findings) ] return { "messages": new_messages, "research_findings": findings }

def coder_node(state: MultiAgentState): """代码智能体节点。负责处理代码相关任务。""" messages = state.get("messages", []) task = state.get("task", "") research = state.get("research_findings", "")

# 构建给代码智能体的提示词 prompt = f""" 用户任务：{task} {f'相关研究发现：{research}' if research else ''} 请根据以上信息，生成或解释相关的代码。如果你是生成代码，请确保代码是正确、可运行的，并附上简要说明。 """ system_msg = SystemMessage(content="你是一个专业的软件开发工程师，擅长生成清晰、正确、高效的代码。") response = model.invoke([system_msg, HumanMessage(content=prompt)]) code_output = response.content # 可选：调用代码检查工具 lint_feedback = TOOLS["lint_code"].invoke(code_output) final_output = f"{code_output}\n\n---\n代码检查反馈：{lint_feedback}" new_messages = messages + [ AIMessage(content="我是代码智能体。我已经生成了相关代码。"), AIMessage(content=final_output) ] return { "messages": new_messages, "code_snippets": final_output }

def summarizer_node(state: MultiAgentState): """总结智能体节点。整合信息，生成最终答案。""" messages = state.get("messages", []) task = state.get("task", "") research = state.get("research_findings", "") code = state.get("code_snippets", "")

prompt = f""" 原始任务：{task} 以下是收集到的信息： 【研究结果】 {research if research else '暂无研究结果'} 【代码相关】 {code if code else '暂无代码生成'} 请基于以上所有信息，生成一个最终、全面、用户友好的回答。回答应该结构化、清晰，并直接解决用户的任务。 """ system_msg = SystemMessage(content="你是一个技术文档作家和总结者，擅长将复杂的技术信息整合成清晰、易懂的答案。") response = model.invoke([system_msg, HumanMessage(content=prompt)]) final_answer = response.content new_messages = messages + [ AIMessage(content="我是总结智能体。我已整合所有信息，生成最终答案。"), AIMessage(content=final_answer) ] return { "messages": new_messages, "final_output": final_answer }

**关键点解析**： 1. **主管节点返回`Command`**：这是实现动态路由的核心。`Command(goto=...)` 告诉LangGraph下一步执行哪个节点。 2. **其他节点返回状态字典**：它们只负责更新状态，不决定流程。流程由主管控制。 3. **工具调用**：智能体通过 `TOOLS["tool_name"].invoke(...)` 来使用工具。 ### 4.3 构建并编译LangGraph图 现在，我们将节点组装成完整的工作流。 ```python # graph_builder.py from langgraph.graph import StateGraph, START, END from agents.state import MultiAgentState from agents.nodes import supervisor_node, researcher_node, coder_node, summarizer_node def build_multi_agent_graph(): """构建并编译多智能体图""" # 1. 创建图构建器，并指定状态模式 builder = StateGraph(MultiAgentState) # 2. 添加节点 builder.add_node("supervisor", supervisor_node) builder.add_node("researcher", researcher_node) builder.add_node("coder", coder_node) builder.add_node("summarizer", summarizer_node) # 3. 设置入口点：从主管开始 builder.add_edge(START, "supervisor") # 4. 定义主管节点的条件边 # 主管节点的 `Command.goto` 字段决定了下一个节点 # 我们使用一个路由函数来实现 def route_after_supervisor(state: MultiAgentState): """根据主管决策的路由函数""" next_agent = state.get("next_agent") if next_agent == "__end__": return END # 确保 next_agent 是已定义的节点名称 if next_agent in ["researcher", "coder", "summarizer"]: return next_agent else: # 如果返回了未知节点，默认回到主管重新决策 print(f"[Warning] 未知的下一个智能体: {next_agent}, 返回主管。") return "supervisor" # 将主管节点连接到路由函数 builder.add_conditional_edges( "supervisor", route_after_supervisor, # 可选：指定可能的目的地，用于图可视化 { "researcher": "researcher", "coder": "coder", "summarizer": "summarizer", END: END, } ) # 5. 定义工作智能体执行后的流转：总是回到主管进行下一轮调度 builder.add_edge("researcher", "supervisor") builder.add_edge("coder", "supervisor") builder.add_edge("summarizer", "supervisor") # 6. 编译图 graph = builder.compile() return graph if __name__ == "__main__": graph = build_multi_agent_graph() # 可视化图（需要安装graphviz） try: from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png())) except: print("无法显示图形，但图已成功构建。") # 打印图的结构 print(graph.get_graph().draw_ascii())

4.4 运行与测试多智能体系统

让我们编写一个主程序来运行这个系统。

# main.py from graph_builder import build_multi_agent_graph from agents.state import MultiAgentState from langchain_core.messages import HumanMessage import asyncio async def run_agent_system(query: str): """运行多智能体系统处理用户查询""" print(f"\n{'='*50}") print(f"开始处理查询: {query}") print(f"{'='*50}") # 1. 构建图 graph = build_multi_agent_graph() # 2. 初始化状态 initial_state: MultiAgentState = { "messages": [HumanMessage(content=query)], "task": query, "research_findings": None, "code_snippets": None, "final_output": None, "next_agent": None, } # 3. 运行图 # 设置最大步数防止无限循环 max_steps = 10 current_state = initial_state for step in range(max_steps): print(f"\n--- 步骤 {step+1} ---") # 调用图，传入当前状态 result = graph.invoke(current_state) current_state = result # 检查是否结束（状态中包含 `final_output` 且主管决定结束） if current_state.get("final_output") and current_state.get("next_agent") == "__end__": print("\n✅ 任务完成！") break if current_state.get("next_agent") == "__end__": print("\n⏹️ 主管决定结束流程。") break else: print(f"\n⚠️ 达到最大步数 ({max_steps})，强制结束。") # 4. 输出最终结果 print(f"\n{'='*50}") print("最终答案：") print(f"{'='*50}") final_answer = current_state.get("final_output") if final_answer: print(final_answer) else: print("未生成最终答案。最终消息历史：") for msg in current_state.get("messages", [])[-5:]: # 打印最后几条消息 print(f"{type(msg).__name__}: {msg.content[:200]}...") return current_state if __name__ == "__main__": # 测试查询 test_queries = [ "Python中如何使用异步IO？", "给我一个用FastAPI创建简单API的示例代码。", "解释一下Docker容器和虚拟机的区别。", ] # 运行第一个查询 import asyncio final_state = asyncio.run(run_agent_system(test_queries[1]))

运行结果示例：

================================================== 开始处理查询: 给我一个用FastAPI创建简单API的示例代码。 ================================================== --- 步骤 1 --- [Supervisor] 决定下一步调用: coder, 理由: 用户请求直接涉及代码生成，应调用代码智能体。 --- 步骤 2 --- [Coder] 正在生成代码... [Linter] 收到代码片段，长度：832 --- 步骤 3 --- [Supervisor] 决定下一步调用: summarizer, 理由: 代码智能体已生成代码片段，现在需要总结智能体将其整合成完整的、带有解释的答案。 --- 步骤 4 --- [Summarizer] 正在整合信息... ✅ 任务完成！ ================================================== 最终答案： ================================================== 以下是一个使用 FastAPI 创建简单 API 的完整示例，包含设置、运行和测试步骤。 1. 环境准备与安装 首先，确保已安装 Python (3.7+)，然后使用 pip 安装 FastAPI 和 Uvicorn（一个 ASGI 服务器）： ```bash pip install fastapi uvicorn

2. 创建主应用文件main.py

from fastapi import FastAPI from pydantic import BaseModel # 创建 FastAPI 应用实例 app = FastAPI(title="简单示例 API", description="一个演示用的 FastAPI 应用") # 定义一个 Pydantic 模型用于请求体验证 class Item(BaseModel): name: str price: float is_offer: bool = False # 根路径，返回欢迎信息 @app.get("/") def read_root(): return {"message": "欢迎使用 FastAPI 示例 API！"} # 带路径参数的 GET 端点 @app.get("/items/{item_id}") def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q} # 带请求体的 POST 端点 @app.post("/items/") def create_item(item: Item): return {"received_item": item, "message": "物品创建成功！"} # 更新物品的 PUT 端点 @app.put("/items/{item_id}") def update_item(item_id: int, item: Item): return {"item_id": item_id, "updated_item": item}

3. 运行应用 在终端中，进入main.py所在目录，运行：

uvicorn main:app --reload

• --reload参数使得代码修改后服务器自动重启，仅用于开发。

4. 测试 API 服务器启动后（默认 http://127.0.0.1:8000），你可以：

• 访问 http://127.0.0.1:8000/docs 查看自动生成的交互式 API 文档（Swagger UI）。
• 访问 http://127.0.0.1:8000/redoc 查看 ReDoc 格式的文档。
• 使用 curl 或 Postman 测试端点：

  # GET 请求 curl http://127.0.0.1:8000/items/42?q=test # POST 请求 curl -X POST http://127.0.0.1:8000/items/ \ -H "Content-Type: application/json" \ -d '{"name":"笔记本电脑","price":5999.99,"is_offer":true}'

代码检查反馈：代码检查通过：结构清晰，包含了基本的 GET、POST、PUT 端点以及 Pydantic 模型验证，是标准的 FastAPI 入门示例。

总结：这个示例涵盖了 FastAPI 的核心功能：快速创建路由、路径/查询参数处理、请求体验证（通过 Pydantic）以及自动 API 文档生成。你可以以此为基础，添加数据库连接、身份验证等更多功能。

## 5. 进阶：集成OpenClaw技能与状态持久化 ### 5.1 集成OpenClaw作为工具源 OpenClaw提供了丰富的预制技能（Skill）。我们可以将这些技能封装成LangChain Tool，供我们的智能体调用。假设我们想使用一个“天气查询”技能。 ```python # agents/openclaw_tools.py from langchain.tools import BaseTool from typing import Optional import requests import os class OpenClawWeatherTool(BaseTool): name = "openclaw_weather" description = "查询指定城市的当前天气情况。" # 假设OpenClaw技能通过一个API端点调用 openclaw_api_base: str = os.getenv("OPENCLAW_API_BASE", "http://localhost:8080") def _run(self, city: str) -> str: """执行工具调用""" # 这里是模拟调用，实际需要根据OpenClaw的API文档调整 try: # 示例：调用OpenClaw的天气技能 # response = requests.post(f"{self.openclaw_api_base}/skill/weather", json={"city": city}) # data = response.json() # return f"{city}的天气：{data['weather']}，温度：{data['temp']}°C" return f"[模拟] 调用OpenClaw天气技能查询{city}。结果：晴朗，25°C。" except Exception as e: return f"调用OpenClaw天气技能失败：{e}" async def _arun(self, city: str) -> str: """异步执行工具调用""" # 实现异步版本 return self._run(city) # 将新工具注册到工具字典中 from .tools import TOOLS TOOLS["openclaw_weather"] = OpenClawWeatherTool()

然后，你可以在researcher_node或其他智能体中像使用普通工具一样使用它：weather_info = TOOLS["openclaw_weather"].invoke("北京")。

5.2 状态持久化与检查点

对于长时间运行或需要中断恢复的智能体工作流，状态持久化至关重要。LangGraph内置了检查点（Checkpoint）机制。

# 使用内存存储的简单示例 from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() # 在编译图时加入持久化 graph = builder.compile( checkpointer=memory, # 配置中断后是否从检查点恢复 interrupt_before=["supervisor"], # 例如，在每次主管决策前允许中断 ) # 使用线程ID（或用户ID、会话ID）来区分不同的对话流 config = {"configurable": {"thread_id": "user_123_session_1"}} # 首次调用 initial_state = {...} result1 = graph.invoke(initial_state, config=config) # 模拟中断后，再次调用会从上次的检查点继续 # 例如，用户提供了更多信息 new_state_with_update = {"messages": [HumanMessage(content="我还需要知道湿度")]} result2 = graph.invoke(new_state_with_update, config=config) # 会从上次中断的`supervisor`节点之前继续

6. 常见问题与排查思路

在开发多智能体系统时，你可能会遇到以下典型问题：

7. 最佳实践与工程建议

1. 始于简单，逐步复杂：不要一开始就设计包含10个智能体的复杂系统。从2-3个智能体的主管模式开始，验证流程跑通，再逐步增加新的智能体或引入更复杂的架构（如层级式）。
2. 设计清晰的状态模式：花时间精心设计State。明确哪些信息是全局共享的，哪些是智能体私有的。良好的状态设计是系统可维护性的基础。
3. 为智能体编写明确的“岗位描述”：每个智能体的系统提示词（System Prompt）就是它的岗位描述。要清晰定义其职责、输入输出格式以及可用的工具。模糊的提示词会导致智能体行为不稳定。
4. 实现健壮的错误处理：在每个智能体节点和工具调用周围添加try...except。当某个智能体或工具失败时，应有备选路径（例如，返回错误信息给主管，由主管决定重试或换一种方式）。
5. 日志与可观测性：在关键节点（如智能体调用开始/结束、工具调用、主管决策）添加详细的日志。这对于调试复杂的工作流至关重要。考虑使用LangSmith进行追踪和监控。
6. 测试策略：
   • 单元测试：单独测试每个智能体节点和工具函数。
   • 集成测试：测试两个智能体之间的交互（如主管->研究员）。
   • 端到端测试：用一系列有代表性的用户查询测试整个图的工作流和最终输出质量。
7. 生产环境部署：
   • 配置管理：将模型API密钥、工具端点等配置信息通过环境变量或配置中心管理。
   • 限流与降级：对LLM API和外部工具调用实施限流，并设计降级方案（例如，搜索失败时使用本地知识库）。
   • 异步处理：对于耗时任务，考虑使用LangGraph的异步支持（ainvoke）或将工作流放入任务队列（如Celery）。

通过本教程，你已经掌握了使用LangChain和LangGraph构建多智能体系统的核心方法。从理解状态、节点、图的概念，到实现一个完整的主管模式协作系统，再到集成外部工具和考虑生产实践，这套方法论可以应用于客服助手、内容创作、数据分析、自动化运维等众多场景。记住，多智能体系统的核心优势在于“分而治之”和“专业协作”，合理的架构设计比单纯追求智能体数量更重要。