基于LangChain与LangGraph的多智能体系统实战：DeepAgents框架开发指南

这次我们来看一个名为 DeepAgents 的多智能体开发框架实战教程。这个项目不是某个具体的模型，而是一个基于 LangChain 和 LangGraph 构建多智能体系统的实战指南。它的核心价值在于，将复杂、前沿的多智能体协作开发流程，拆解成一套可落地、可复现的工程化实践，目标是让你在构建自己的智能体应用时，能直接上手，避开那些文档里没写的“坑”。

对于开发者而言，多智能体系统听起来很酷，但实际开发中常会遇到智能体间通信混乱、状态管理复杂、任务流程难以编排等问题。DeepAgents 这个实战教程，就是针对这些痛点，提供了一套从环境搭建、智能体定义、工作流编排到实际任务执行的完整解决方案。它不只是一个概念讲解，更侧重于“怎么用代码实现”。

本文将带你快速梳理 DeepAgents 实战教程的核心内容，并基于 LangChain/LangGraph 生态，完成一个从零开始的多智能体项目搭建。我们会重点关注：环境如何一键配置、智能体角色如何定义、LangGraph 状态图如何设计、任务如何分解与协作，以及最终如何运行和调试。无论你是想了解多智能体开发，还是急需一个可运行的参考项目，这篇文章都能提供直接的路径。

1. 核心能力速览

2. 适用场景与使用边界

DeepAgents 实战教程主要适用于以下几类开发者和场景：

适用场景：

1. AI 应用开发者：希望在自己的产品中引入多个 AI 角色分工协作，例如一个客服系统包含“接待员”、“技术专家”、“质检员”等多个智能体。
2. 自动化流程工程师：需要将复杂的、多步骤的业务流程（如报告生成、数据审核、内容创作）自动化，且步骤间存在依赖和决策。
3. 技术学习者与研究者：希望深入理解 LangChain 和 LangGraph 如何在实际项目中结合，掌握多智能体系统的工程化实现方法。
4. 原型验证：快速搭建一个多智能体概念验证（PoC）项目，验证想法的可行性。

使用边界与注意事项：

1. 非开箱即用产品：这是一个教程和代码框架，你需要理解代码并根据自己的业务逻辑进行修改和扩展。
2. 依赖外部大模型：智能体的“大脑”通常依赖 OpenAI GPT、Anthropic Claude 或国内如智谱、DeepSeek 等大模型 API。你需要自行申请并配置 API Key，并承担相应的调用费用。
3. 复杂度随智能体数量增长：智能体越多，它们之间的交互和状态管理就越复杂。教程提供的是基础模式，复杂场景需要你具备良好的系统设计能力。
4. 合规与安全：当智能体处理用户数据、生成内容或做出决策时，必须考虑数据隐私、内容安全性和责任归属。确保你的应用符合相关法律法规。
5. 性能与成本：频繁调用大模型 API 会产生延迟和成本。在设计工作流时，需考虑缓存、异步调用和成本控制策略。

3. 环境准备与前置条件

在开始编码之前，请确保你的开发环境满足以下要求。这是能顺利跑通教程代码的基础。

基础环境检查清单：

• 操作系统：Windows 10/11, macOS, 或 Linux (如 Ubuntu)。教程代码通常是跨平台的。
• Python 版本：Python 3.8 或更高版本。推荐使用 Python 3.10 以获得最佳兼容性。
• 包管理工具：pip是最低要求。强烈推荐使用venv或conda创建独立的虚拟环境，避免包冲突。
• 网络连接：能够稳定访问你所选用的大模型供应商的 API 服务器（如api.openai.com或国内对应服务地址）。
• 代码编辑器/IDE：VS Code, PyCharm 等任选，具备 Python 开发支持即可。
• 版本控制：Git，用于克隆教程代码仓库。

关键账户与密钥准备：

1. 大模型 API 账户：根据教程或你的选择，注册并获取以下至少一项：
   • OpenAI API Key
   • Anthropic API Key
   • 智谱 AI API Key
   • 其他兼容 OpenAI 格式的 API 服务密钥
2. 环境变量管理：准备好将 API Key 设置为环境变量，这是安全且通用的做法。

4. 安装部署与启动方式

假设你已经从 GitHub 或教程提供的地址获取了 DeepAgents 项目的代码。以下是通用的部署启动步骤。

步骤 1：克隆项目与创建环境

# 1. 克隆项目代码 (假设仓库地址为 placeholder，请替换为实际地址) git clone <deepagents-tutorial-repo-url> cd deepagents-tutorial # 2. 创建并激活 Python 虚拟环境 python -m venv venv # Windows: venv\Scripts\activate # Linux/macOS: source venv/bin/activate # 3. 升级 pip 和安装基础依赖 pip install --upgrade pip

步骤 2：安装项目依赖通常项目根目录会有一个requirements.txt或pyproject.toml文件。

# 安装所有依赖 pip install -r requirements.txt

如果项目没有提供requirements.txt，核心依赖通常包括：

pip install langchain langgraph langchain-openai langchain-anthropic langchain-community # 以及其他可能需要的工具库，如 requests, pydantic, python-dotenv 等

步骤 3：配置 API 密钥在项目根目录创建.env文件，用于存储敏感信息。

# .env 文件内容示例 OPENAI_API_KEY=sk-your-openai-api-key-here ANTHROPIC_API_KEY=your-anthropic-api-key-here ZHIPUAI_API_KEY=your-zhipuai-api-key-here

然后在你的主 Python 脚本或app.py中，通过dotenv加载：

from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 import os openai_api_key = os.getenv(“OPENAI_API_KEY”)

步骤 4：理解项目结构并启动一个典型的 DeepAgents 教程项目结构可能如下：

deepagents-tutorial/ ├── agents/ # 存放各个智能体的定义 │ ├── researcher.py │ ├── writer.py │ └── reviewer.py ├── graphs/ # 存放 LangGraph 状态图定义 │ └── workflow_graph.py ├── state.py # 定义共享的状态对象 ├── main.py # 主程序入口 ├── requirements.txt └── .env.example

启动服务或运行示例，通常是执行主入口文件：

# 直接运行一个示例工作流 python main.py # 或者，如果项目提供了 Web 界面或 API 服务 python app.py

运行后，控制台会输出智能体间的对话和工作流执行步骤。

5. 功能测试与效果验证

现在，我们来验证一个多智能体系统是否按预期工作。我们将设计一个简单的测试场景：一个包含“研究员”、“写手”、“评审员”三个智能体的内容创作流水线。

测试目标：验证智能体能根据主题协同工作，最终产出一份合格的大纲。

5.1 定义智能体角色与工具

首先，查看agents/目录下的文件，理解每个智能体的定义。一个智能体通常包括：

• 角色描述：告诉模型它扮演谁。
• 能力指令：它擅长做什么。
• 可用工具：它可以调用哪些函数（如网络搜索、计算器、数据库查询）。

# agents/researcher.py 示例片段 from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.tools import Tool from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder def get_researcher_agent(): llm = ChatOpenAI(model=“gpt-4-turbo-preview”, temperature=0.2) # 假设有一个模拟的搜索工具 search_tool = Tool( name=“WebSearch”, func=lambda query: f“Search results for ‘{query}‘: [Simulated Data]”, description=“Useful for searching the web for current information.” ) prompt = ChatPromptTemplate.from_messages([ (“system”, “You are a meticulous researcher. Your job is to find accurate and relevant information on a given topic.”), MessagesPlaceholder(variable_name=“chat_history”), (“human”, “{input}”), MessagesPlaceholder(variable_name=“agent_scratchpad”) ]) agent = create_openai_tools_agent(llm, [search_tool], prompt) return AgentExecutor(agent=agent, tools=[search_tool], verbose=True)

5.2 构建 LangGraph 工作流

接下来，在graphs/workflow_graph.py中，定义智能体如何协作。LangGraph 的核心是“状态”和“边”。

# graphs/workflow_graph.py 示例片段 from langgraph.graph import StateGraph, END from .state import ContentCreationState # 假设有一个自定义状态类 from ..agents.researcher import get_researcher_agent from ..agents.writer import get_writer_agent from ..agents.reviewer import get_reviewer_agent def create_workflow_graph(): workflow = StateGraph(ContentCreationState) # 1. 添加节点（每个智能体或步骤都是一个节点） workflow.add_node(“research”, research_node) workflow.add_node(“write”, write_node) workflow.add_node(“review”, review_node) # 2. 设置入口点 workflow.set_entry_point(“research”) # 3. 定义边（控制流） workflow.add_edge(“research”, “write”) workflow.add_edge(“write”, “review”) # 评审后可能返回修改或结束 workflow.add_conditional_edges( “review”, decide_to_finish, {“revise”: “write”, “approve”: END} ) return workflow.compile() def research_node(state: ContentCreationState): agent = get_researcher_agent() result = agent.invoke({“input”: state[“topic”]}) state[“research_materials”] = result[“output”] return state def write_node(state: ContentCreationState): # 写手基于研究材料创作 agent = get_writer_agent() input_text = f“Topic: {state[‘topic’]}. Research: {state[‘research_materials’]}” result = agent.invoke({“input”: input_text}) state[“draft”] = result[“output”] return state def review_node(state: ContentCreationState): # 评审员审核草稿 agent = get_reviewer_agent() result = agent.invoke({“input”: f“Please review this draft: {state[‘draft’]}”}) state[“feedback”] = result[“output”] return state def decide_to_finish(state: ContentCreationState): # 根据评审反馈决定下一步 if “needs major revision” in state[“feedback”].lower(): return “revise” else: return “approve”

5.3 运行与验证工作流

最后，在main.py中初始化并运行这个图。

# main.py from graphs.workflow_graph import create_workflow_graph from state import ContentCreationState def main(): # 初始化工作流 app = create_workflow_graph() # 定义初始状态 initial_state = ContentCreationState(topic=“The impact of AI on software development in 2024”) # 运行工作流 print(“Starting multi-agent workflow...\n”) final_state = app.invoke(initial_state) # 输出结果 print(“\n=== Workflow Completed ===") print(f“Topic: {final_state[‘topic’]}”) print(f“\nFinal Draft:\n{final_state.get(‘draft’, ‘N/A’)}”) print(f“\nReview Feedback:\n{final_state.get(‘feedback’, ‘N/A’)}”) if __name__ == “__main__”: main()

预期结果与成功标准：

• 控制台输出：应能看到清晰的步骤日志，如 “Researching topic…”， “Writing draft…”， “Reviewing…”。
• 状态流转：最终状态 (final_state) 中应包含research_materials、draft、feedback等字段，且内容与主题相关。
• 协作逻辑：如果评审反馈要求修改，工作流应能正确跳回write节点，形成循环。
• 内容质量：生成的草稿和反馈应具备一定的逻辑性和专业性，符合各智能体的角色设定。

常见失败原因：

1. API 密钥错误或网络问题：导致无法调用大模型。检查.env文件和控制台错误信息。
2. 依赖包版本冲突：LangChain 和 LangGraph 版本迭代快。尝试固定版本，如pip install langchain==0.1.0 langgraph==0.0.22。
3. 状态对象定义错误：ContentCreationState的字段与节点中存取的字段名不匹配。确保状态类使用TypedDict或Pydantic BaseModel明确定义。
4. 图结构定义错误：节点未正确连接，或条件边函数返回了不存在的节点名。仔细检查add_edge和add_conditional_edges部分。

6. 接口 API 与批量任务

虽然教程本身可能是一个脚本，但你可以轻松地将其封装成 API 服务，以支持批量任务。

6.1 封装为 FastAPI 服务

将编译好的 LangGraph 应用 (app) 包装成一个 HTTP 端点。

# api_server.py from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from typing import List, Optional from graphs.workflow_graph import create_workflow_graph from state import ContentCreationState app = FastAPI(title=“DeepAgents API”) workflow_app = create_workflow_graph() # 预加载工作流 class WorkflowRequest(BaseModel): topic: str callback_url: Optional[str] = None # 用于异步回调 class BatchRequest(BaseModel): tasks: List[WorkflowRequest] @app.post(“/run”) async def run_workflow(request: WorkflowRequest): “”“同步执行单个任务”“” initial_state = ContentCreationState(topic=request.topic) final_state = workflow_app.invoke(initial_state) return { “status”: “success”, “topic”: final_state[“topic”], “draft”: final_state.get(“draft”), “feedback”: final_state.get(“feedback”) } @app.post(“/run_batch”) async def run_batch(request: BatchRequest, background_tasks: BackgroundTasks): “”“异步执行批量任务”“” task_ids = [] for task in request.tasks: # 为每个任务生成一个唯一ID，并提交到后台任务队列 task_id = f“task_{len(task_ids)}” background_tasks.add_task(process_single_task, task_id, task) task_ids.append(task_id) return {“status”: “batch_started”, “task_ids”: task_ids} def process_single_task(task_id: str, task: WorkflowRequest): “”“后台任务处理函数”“” try: initial_state = ContentCreationState(topic=task.topic) final_state = workflow_app.invoke(initial_state) # 这里可以将结果保存到数据库或发送到 callback_url print(f“Task {task_id} completed: {task.topic}”) except Exception as e: print(f“Task {task_id} failed: {e}”) if __name__ == “__main__”: import uvicorn uvicorn.run(app, host=“0.0.0.0”, port=8000)

启动服务：python api_server.py。访问http://127.0.0.1:8000/docs查看自动生成的 API 文档。

6.2 调用 API 示例

使用curl或 Pythonrequests库调用服务。

# 同步调用单个任务 curl -X POST “http://127.0.0.1:8000/run" \ -H “Content-Type: application/json” \ -d ‘{“topic”: “Future of renewable energy”}’

# Python 异步批量调用示例 import asyncio import aiohttp import json async def send_batch(): async with aiohttp.ClientSession() as session: batch_data = { “tasks”: [ {“topic”: “Topic 1”}, {“topic”: “Topic 2”}, # … 更多任务 ] } async with session.post(‘http://127.0.0.1:8000/run_batch', json=batch_data) as resp: result = await resp.json() print(f“Batch started: {result}”) asyncio.run(send_batch())

6.3 批量任务管理建议

• 队列与去重：对于大规模批量任务，建议引入 Redis 或 RabbitMQ 等消息队列，而非简单的后台任务。
• 状态持久化：将每个任务的状态（如task_id,topic,status,result,error）存入数据库（如 SQLite, PostgreSQL），便于查询和重试。
• 限流与重试：大模型 API 有速率限制。在批量调用时，需要添加限流逻辑和失败重试机制。
• 结果回调：如果任务处理时间长，采用异步模式并通过callback_url通知调用方，是更友好的设计。

7. 资源占用与性能观察

由于 DeepAgents 框架主要协调对大模型 API 的调用，其本地资源占用主要体现在 CPU、内存和网络 I/O 上。

关键性能观察点：

1. API 调用延迟：这是最主要的性能瓶颈。每个智能体的每次“思考”都是一次或多次网络请求。可以在代码中添加计时逻辑来监控。

   import time start = time.time() result = agent.invoke({“input”: “…”}) latency = time.time() - start print(f“Agent invocation took {latency:.2f} seconds”)

2. 内存占用：LangChain/LangGraph 本身内存占用不高。但如果处理大量上下文（长对话历史、大文档），或同时运行大量智能体实例，内存使用会增长。使用工具如psutil监控进程内存。
3. Token 消耗与成本：智能体间的对话、工具调用结果都会计入上下文，增加 Token 消耗。务必在代码中或通过大模型供应商的控制台监控 Token 使用量，以控制成本。
4. 工作流复杂度：图中的节点和边越多，条件判断越复杂，状态流转的逻辑开销就越大。对于超复杂工作流，考虑将其拆分为多个子图，或优化状态结构。

优化建议：

• 缓存：对频繁查询且结果不变的内容（如某些工具调用结果）进行缓存。
• 异步调用：如果多个智能体可以并行工作，使用asyncio进行异步调用，减少总等待时间。
• 精简上下文：设计工作流时，只将必要的信息传递给下一个节点，避免上下文膨胀。
• 选择合适模型：对于不需要顶级推理能力的环节，可以使用更小、更快的模型（如 GPT-3.5-turbo），以平衡速度、效果和成本。

8. 常见问题与排查方法

在开发和运行 DeepAgents 多智能体系统时，你可能会遇到以下典型问题。

9. 最佳实践与使用建议

基于 DeepAgents 框架进行多智能体开发，遵循以下实践能让项目更稳健、易维护。

1. 从简单开始，逐步复杂化

   • 不要一开始就设计包含 10 个智能体的复杂系统。先从 2-3 个智能体的最小可行工作流开始，确保基础通信和状态流转正常。
   • 验证通过后，再逐步添加新的智能体角色和更复杂的条件分支。

2. 明确定义状态结构

   • 使用Pydantic的BaseModel来定义状态类。这能提供类型提示、自动验证和清晰的文档。
   • 将状态划分为不同的命名空间，避免所有数据都堆在一个扁平字典里。

3. 为智能体编写清晰的“岗位说明书”

   • 系统提示词（System Prompt）是智能体的灵魂。用清晰、无歧义的语言描述其角色、职责、行为边界和输出格式。
   • 示例：“你是一名严谨的代码评审员。你的任务是检查 Python 代码中的 bug 和风格问题。请按‘问题描述：… 建议修改：…’的格式逐条列出。”

4. 实现有效的工具（Tools）

   • 工具是智能体与外部世界交互的桥梁。确保每个工具功能单一、可靠，并配有准确的description，以便智能体理解何时使用它。
   • 对于可能失败的工具调用（如网络请求），做好异常处理，并返回友好的错误信息给智能体。

5. 日志与可观测性

   • 启用 LangChain 的verbose=True来查看详细的思考链。
   • 在关键节点（如图的每个节点开始/结束时）记录状态快照和耗时，便于调试和性能分析。
   • 考虑将运行日志结构化输出到文件或监控系统。

6. 成本与性能监控

   • 在调用大模型 API 的代码前后记录 Token 使用量（如果 API 返回）。
   • 为你的应用设置预算告警，避免意外的高额账单。
   • 对于非关键路径或简单任务，评估是否可以使用更便宜的模型。

7. 安全与合规

   • 智能体生成的内容（文本、代码、建议）必须经过人工审核或设置安全护栏（如 LangChain 的Guardrails），才能对外发布或执行。
   • 如果智能体处理用户数据，确保符合数据隐私法规（如 GDPR）。
   • 明确告知用户正在与 AI 交互，并对 AI 可能产生的错误或误导性内容进行免责声明。

通过 DeepAgents 实战教程，你获得的不只是一套可运行的代码，更是一套构建复杂 AI 协作系统的思维模式和工程方法。它的价值在于将 LangChain/LangGraph 的理论转化为肌肉记忆。接下来，你可以尝试修改智能体的角色和工具，设计更复杂的工作流（如带循环审批的流程），或者将其集成到现有的 Web 应用或自动化平台中。多智能体开发的坑，很多都在状态管理和异常处理上，动手踩一遍，印象最深刻。