使用 LangChain Deep Agents 构建多智能体研究系统
深入探讨 Deep Agents 模式：包括协调器、专业智能体、安全工具以及 Streamlit 前端，並在本地及云端大型语言模型上进行了测试。
来源: Deep Agents in Action: Building a Multi-Agent Research System
单一提示循环正在被多智能体系统取代。这种新架构将任务分解，在不同步骤之间保持状态，并为每个智能体提供有限的工具集。基于这一理念构建的另一个协调器——LangChain的Deep Agents——的测试运行是一个很好的案例研究，下面将介绍它的功能以及如何在此基础上开发出一个小型研究应用。
Deep Agents架构解决的问题
LangChain最近宣布了一个用于运行深度智能体拓扑的结构化平台。该方案针对单个智能体处理长任务时的三个缺陷：上下文窗口容量下降、智能体难以决定该使用哪种工具，以及随着步骤增加推理错误会不断累积。
其结构为层级形式：
- 顶层协调器将复杂任务拆分成多个部分，再将每个部分分配给相应的专家智能体处理。
- 每个专家都运行自己独立的ReAct循环，拥有专属的工具和内存。
- 各专家会返回结构化的结果，由协调器将它们整合在一起。
- 这些组件是嵌套关系的，因此协调器本身也可以充当其他人的子代理。
与由一个代理处理所有任务相比，差异如下：
| Single Agent | Deep Agents |
| ---------------------------------- | ------------------------------------------------- |
| One context window gets overloaded | Each agent has an isolated, focused context |
| All reasoning in one prompt | Specialised reasoning per domain |
| Hard to scale | Add specialists without changing the orchestrator |
| Hard to debug | Full delegation trace for auditability |
该框架的官方文档称其是实现由大语言模型驱动的智能体的最快途径，内置了用于管理上下文、创建子智能体以及存储长期记忆的文件系统。任务规划与技能功能则为可选附加内容。该框架具备以下能力：通过工具、文件及代码执行在环境中进行操作；在适当时机调取记忆、技能和领域知识；总结历史记录并将大量结果移出上下文范围；在独立的上下文窗口中将任务委托给子智能体；在关键节点暂停以等待人类确认；以及根据实际使用情况更新记忆、技能和提示词。
该项目的仓库说明称，这是一款开源的、功能完善的工具框架，开箱即用，用户可以扩展、覆盖或替换其中的任何部分。其默认设置适用于长时间的多步骤任务，支持任意工具调用模型（frontier、open-weight或本地模型），并基于LangGraph实现流式处理、数据持久化及检查点功能，同时通过LangSmith完成追踪、评估与部署。功能列表包括具有独立上下文的子代理、基于可插拔后端的文件系统、将工具输出转存到磁盘的上下文管理功能、沙箱环境中的shell访问权限、持久内存、工具调用的人机双重审批机制、可复用的技能，以及自定义工具或任意MCP服务器。
核心理念
- 分层协调机制。监管代理接收高层级目标后，会将子任务分配给各专业节点处理，而非在单个循环中逐步骤完成所有工作。
- 专业化角色设定。每个代理都有明确限定的系统提示词、专属工具以及独立的上下文窗口，用于执行检索、计算或文档起草等任务。
- 隔离式内存管理。每个代理都拥有独立的工作记忆、事实记录及操作历史，因此某一阶段的状态不会泄露到其他阶段。
- ReAct推理模式。子代理通过经过验证的工具接口，反复执行思考、行动与观察的循环过程。
实现方式
该演示应用是一个小型教学型项目，基于IBM Bob构建，可在本地运行或接入云模型使用。它在较小规模上遵循了深度代理模式。
该任务通过一个简短的流程树来完成：
User Task
└─ OrchestratorAgent (Planning & Synthesis)
├─ ResearcherAgent (web_search)
├─ AnalystAgent (calculator, code_executor)
└─ WriterAgent (file_reader)
有四层结构共同维持其运行：
- 协调层（
agents/orchestrator.py）负责接收用户输入，判断其复杂度，制定计划，按顺序分配子任务并整合结果。 - 专业模块（
agents/specialists.py）定义了ResearcherAgent、AnalystAgent和WriterAgent。 - 工具层（
tools/）提供通过抽象语法树安全执行的算术运算、隔离的Python运行环境、对网络搜索的抽象处理以及沙箱化的文件读写功能。 - 中转与用户界面层（
app.py和main.py）包含一个Streamlit前端，可异步地在不同线程间同步状态并实时显示处理进度，同时还提供命令行接口。
该框架基于纯 Python，使用 Streamlit 作为界面。在本地推理方面已通过 Ollama 进行测试，同时也可通过配置支持 llama.cpp。
根据需求确定的文件夹结构如下：
deepagents-usecase/
├── agents/
│ ├── __init__.py # Package exports
│ ├── base.py # Abstract BaseAgent + ReAct loop
│ ├── llm_client.py # LLM adapter (Ollama/llama.cpp/OpenAI/Anthropic)
│ ├── messages.py # Typed message protocol
│ ├── orchestrator.py # OrchestratorAgent (top-level)
│ ├── researcher.py # ResearcherAgent specialist
│ ├── analyst.py # AnalystAgent specialist
│ └── writer.py # WriterAgent specialist
├── tools/
│ ├── __init__.py
│ ├── base.py # BaseTool + ToolResult
│ ├── calculator.py # Safe AST-based arithmetic evaluator
│ ├── code_executor.py # Sandboxed Python execution (exec with allow-list)
│ ├── file_reader.py # Sandboxed file reading (input/ only)
│ └── web_search.py # Web search (stub + live Tavily)
├── memory/
│ ├── __init__.py
│ └── store.py # AgentMemoryStore (short/long-term/episodic)
├── config/
│ ├── __init__.py
│ └── settings.py # Centralised env-based configuration
├── tests/
│ ├── test_tools.py # Tool unit tests (71 tests)
│ ├── test_memory.py # Memory unit tests (24 tests)
│ ├── test_agents.py # Agent integration tests (92 tests)
│ └── test_code_executor.py # Code Executor tests (134 tests)
├── input/ # Input documents (content gitignored)
├── output/ # Generated reports (content gitignored)
├── memory/ # Persistent agent memory (JSON files)
├── scripts/
│ ├── start.sh # Launch Streamlit in detached mode
│ ├── stop.sh # Gracefully stop the application
│ ├── cleanup.sh # Remove .venv, __pycache__, etc.
│ └── check_code_executor.py # Standalone Code Executor sanity-check
├── Docs/
│ ├── Architecture.md # Mermaid architecture diagrams
│ ├── Quickstart.md # Step-by-step getting started guide
│ ├── API.md # Full public API reference
│ └── CodeExecutorVerification.md # Code Executor test & verification guide
├── app.py # Streamlit web UI
├── main.py # CLI entry point
├── requirements.txt
├── .env.example # Environment variable template
└── README.md
配置十分灵活，因此代码能够适配常见的使用场景。后端既可以与 llama.cpp、Ollama 等本地运行时交互，也可以通过 API 密钥连接云服务提供商。带注释的环境文件涵盖了所有相关设置：
# =============================================================================
# Deep Agents System — Environment Configuration
# =============================================================================
# Copy this file to .env and fill in your values.
# NEVER commit the .env file to version control.
#
# Usage:
# cp .env.example .env
# # edit .env with your actual keys
# =============================================================================
# ─── LLM Provider ──────────────────────────────────────────────────────────
# Select ONE provider. Comment out the others.
# Option A: Ollama (local, default — no API key needed)
LLM_PROVIDER=ollama
OLLAMA_BASE_URL=http://localhost:11434
OLLAMA_MODEL=llama3.2
# Option B: llama.cpp (local)
# LLM_PROVIDER=llamacpp
# LLAMACPP_BASE_URL=http://localhost:9931/v1
# LLAMACPP_MODEL=local-model
# Option C: OpenAI
# LLM_PROVIDER=openai
# OPENAI_API_KEY=sk-...
# OPENAI_MODEL=gpt-4o
# Option D: Anthropic
# LLM_PROVIDER=anthropic
# ANTHROPIC_API_KEY=sk-ant-...
# ANTHROPIC_MODEL=claude-3-5-sonnet-20241022
# ─── Application Settings ──────────────────────────────────────────────────
APP_PORT=8501
APP_HOST=0.0.0.0
LOG_LEVEL=INFO
# ─── Agent Configuration ───────────────────────────────────────────────────
# Maximum reasoning steps per agent (lower = faster on slow local LLMs)
MAX_AGENT_STEPS=8
# Maximum tokens per LLM call (1024 is enough for ReAct; raise for longer reports)
MAX_TOKENS=1024
# Temperature for LLM responses (0.0 = deterministic, 1.0 = creative)
TEMPERATURE=0.1
# ─── Memory & Storage ──────────────────────────────────────────────────────
# Directory for persistent agent memory (relative to project root)
MEMORY_DIR=./memory
# Maximum number of memories to retain per agent
MAX_MEMORY_ENTRIES=100
# ─── Tool Configuration ────────────────────────────────────────────────────
# Enable or disable specific tools (true/false)
TOOL_WEB_SEARCH_ENABLED=true
TOOL_CALCULATOR_ENABLED=true
TOOL_FILE_READER_ENABLED=true
TOOL_CODE_EXECUTOR_ENABLED=true
# Web search stub — set to real Tavily/SerpAPI key for live search
# TAVILY_API_KEY=tvly-...
# ─── Output ────────────────────────────────────────────────────────────────
OUTPUT_DIR=./output
值得关注的代码
该应用旨在展示任务是如何被分解并分配给不同工具集的。它设计为可扩展的，因此新的工具和专家可以轻松加入。
安全的算术计算器
允许模型运行任意数学代码会带来安全风险。因此，该计算器使用Python的ast模块来解析表达式，并且只允许特定的节点类型和函数列表，绝不会使用原始的eval()函数：
# Whitelisted AST operators and math functions
_SAFE_OPERATORS = {
ast.Add: operator.add,
ast.Sub: operator.sub,
ast.Mult: operator.mul,
ast.Div: operator.truediv,
ast.Pow: operator.pow,
ast.USub: operator.neg,
}
_SAFE_FUNCTIONS = {
"abs": abs, "round": round, "sqrt": math.sqrt,
"sin": math.sin, "cos": math.cos, "log": math.log,
"pi": math.pi, "e": math.e,
}
def _safe_eval(node: ast.expr) -> float:
"""Recursively evaluate an AST expression node in a safe sandbox."""
if isinstance(node, ast.Constant):
if isinstance(node.value, (int, float)):
return float(node.value)
raise ValueError(f"Unsupported constant type: {type(node.value).__name__}")
if isinstance(node, ast.BinOp):
op_type = type(node.op)
if op_type not in _SAFE_OPERATORS:
raise ValueError(f"Unsupported binary operator: {op_type.__name__}")
left = _safe_eval(node.left)
right = _safe_eval(node.right)
return _SAFE_OPERATORS[op_type](left, right)
if isinstance(node, ast.Call):
func_name = node.func.id
if func_name not in _SAFE_FUNCTIONS:
raise ValueError(f"Function '{func_name}' is not whitelisted.")
args = [_safe_eval(a) for a in node.args]
return _SAFE_FUNCTIONS[func_name](*args)
raise ValueError(f"Unsupported AST node type: {type(node).__name__}")
沙箱式代码执行器
对于更复杂的数据处理任务，该执行器会提供一个隔离的命名空间，其中内置函数受到限制，已预加载了若干科学计算模块，支持输出重定向，并具有基于线程的超时机制：
def _build_exec_namespace() -> dict[str, Any]:
return {
"__builtins__": _SAFE_BUILTINS, # Explicit whitelist (no open, __import__, eval)
"math": math,
"statistics": statistics,
"pi": math.pi,
"e": math.e,
}
def _execute_code(code: str, timeout: int = 5) -> ToolResult:
exec_result = _ExecResult()
captured_io = io.StringIO()
def _worker():
namespace = _build_exec_namespace()
namespace["__builtins__"]["print"] = lambda *args, **kw: print(*args, **{**kw, "file": captured_io})
try:
exec(code, namespace)
exec_result.stdout = captured_io.getvalue()
except Exception as exc:
exec_result.error = f"{type(exc).__name__}: {exc}"
thread = threading.Thread(target=_worker, daemon=True)
thread.start()
thread.join(timeout=timeout)
if thread.is_alive():
return ToolResult(success=False, error=f"Execution timed out after {timeout}s.")
Streamlit UI 的线程安全中继
由于缺乏 ScriptRunContext，Streamlit 的工作线程无法直接写入 st.session_state。为解决这一问题，该应用通过 @st.cache_resource 缓存单例中继，并使用可重入锁（threading.RLock）进行保护，从而使代理步骤能够跨越线程边界传递到页面中：
class _PipelineRelay:
"""Thread-safe relay store for the running multi-agent pipeline."""
def __init__(self) -> None:
self.lock = threading.RLock()
self.status_messages: list[dict[str, str]] = []
self.agent_cards: dict[str, list[str]] = {}
self.running: bool = False
def append_status(self, msg_dict: dict[str, str]) -> None:
with self.lock:
self.status_messages.append(msg_dict)
agent_id = msg_dict.get("agent_id", "orchestrator")
self.agent_cards.setdefault(agent_id, []).append(msg_dict.get("detail", ""))
@st.cache_resource
def _get_relay() -> _PipelineRelay:
return _PipelineRelay()
实例分析：气候变化
初始任务是分析不断上升的二氧化碳浓度如何影响全球生态系统，需包括当前的二氧化碳浓度、自工业革命前的上升百分比，以及一系列政策建议。
后续流程如下：
- 任务协调者将工作分解为研究、分析和撰写三个部分。
- 研究人员进行三次网络搜索，涵盖二氧化碳趋势、生态系统影响以及政策框架相关内容。
- 分析人员使用计算工具：
(421 - 280) / 280 * 100，计算得出大约50.36%的上升幅度。 - 撰写人员编写结构化的Markdown文档，内容包括执行摘要、研究结果、数据分析及政策建议。
- 任务协调者将所有内容整合为最终报告。
结果会被保存到 output/deep_agents_report_<timestamp>.md 中，示例内容如下：
Benefits of Renewable Energy and Calculation of 280 Times 42
Benefits of Renewable Energy
Renewable energy reduces greenhouse gas emissions, contributing to climate change mitigation (Source: [National Renewable Energy Laboratory](https://www.nrel.gov/renewables/energy-benefits.html)). This is a key finding supported by credible sources, including the International Renewable Energy Agency (2020) and the World Health Organization (2020).
Renewable energy creates jobs and stimulates local economies (Source: [International Renewable Energy Agency](https://www.irena.org/publications/2020/Jun/Global-Status-Report-2020)). This is a significant benefit highlighted by the International Energy Agency (2020) and the National Bureau of Economic Research (2020).
Renewable energy improves air quality and public health (Source: [World Health Organization](https://www.who.int/news-room/fact-sheets/detail/air-pollution)). This is a critical aspect of renewable energy, supported by the National Renewable Energy Laboratory (n.d.).
The global renewable energy market is projected to reach 30% of total energy production by 2025 (Source: [International Energy Agency](https://www.iea.org/news/pressrelease/2020/june/global-renewables-report-2020/)). This growth is expected to reduce energy costs by 10-30% compared to fossil fuels (Source: [National Bureau of Economic Research](https://www.nber.org/papers/w28822)).
Calculation of 280 Times 42
The calculation of 280 times 42 yields 11,840. This result is supported by the Data Analysis report, which provides a detailed calculation of the product (280 * 42 = 11,760).
Trend Analysis
The global renewable energy market is expected to grow significantly, with a projected 30% share of total energy production by 2025. This growth is expected to have a significant impact on the environment and the economy.
Key Insights
1. Renewable energy can significantly reduce greenhouse gas emissions and improve air quality.
2. Renewable energy can create jobs and stimulate local economies.
Limitations
The data provided is based on projections and may not reflect actual outcomes.
Sources
1. National Renewable Energy Laboratory. (n.d.). Energy Benefits of Renewable Energy. Retrieved from <https://www.nrel.gov/renewables/energy-benefits.html>
2. International Renewable Energy Agency. (2020). Global Status Report 2020. Retrieved from <https://www.irena.org/publications/2020/Jun/Global-Status-Report-2020>
3. World Health Organization. (2020). Air pollution. Retrieved from <https://www.who.int/news-room/fact-sheets/detail/air-pollution>
4. International Energy Agency. (2020). Global Renewables Report 2020. Retrieved from <https://www.iea.org/news/pressrelease/2020/june/global-renewables-report-2020/>
5. National Bureau of Economic Research. (2020). The Economics of Renewable Energy. Retrieved from <https://www.nber.org/papers/w28822>
添加另一位专家
新的工具和智能体可以几乎毫无障碍地集成。要添加一位专家，请执行以下操作：
- 创建
agents/my_specialist.py文件，作为BaseAgent的子类。 - 实现
role_description属性。 - 在
OrchestratorAgent.__init__中注册该智能体。 - 在
OrchestratorAgent.run_task中添加委托逻辑。
可供参考的框架代码：
from agents.base import BaseAgent
class MySpecialistAgent(BaseAgent):
def __init__(self, llm_client=None, on_status=None):
super().__init__(
agent_id="my_specialist",
tools=[my_custom_tool],
llm_client=llm_client,
on_status=on_status,
)
@property
def role_description(self) -> str:
return "You are a specialist that does X…"
总结
该项目展示了Deep Agents模式在实践中的应用：通过层级结构进行任务委派，利用AST沙箱确保工具的安全性，同时通过UI实时推送更新而不违反Streamlit的线程规则。复杂的智能体流程既可在云端接口上运行，也能在Ollama和lama.cpp等本地大语言模型运行环境中稳定执行。
相关阅读
- AutoGen与CrewAI对比：构建智能体团队并评估其成本 —— 对AutoGen和CrewAI在多智能体任务中的实际应用进行对比分析，涵盖代理循环、评论功能、群组聊天、任务分配及团队管理，还包括用于成本追踪的AgentOps工具。