DeepAgents:基于LangChain的下一代智能代理框架

https://docs.langchain.com/oss/python/deepagents/overview

概述

DeepAgents是一个构建在LangChain之上的高级智能代理框架,旨在提供更强大、更灵活的代理构建能力。该框架通过模块化设计,为开发者提供了文件系统操作、子代理管理、可插拔后端存储等核心功能,使得构建复杂的多步骤AI代理变得更加简单和高效。

DeepAgents构建在以下基础之上:

核心架构

DeepAgents采用分层架构设计,主要包含以下核心模块:

1. 核心入口模块(graph.py

  • create_deep_agent():框架的主要入口函数,用于创建深度代理
  • 默认使用Claude Sonnet 4作为模型
  • 内置多种中间件,包括待办事项、文件系统、子代理等

2. 后端存储模块(backends/

提供可插拔的文件存储后端系统:

后端协议(protocol.py

  • BackendProtocol:定义统一的后端接口
  • SandboxBackendProtocol:支持命令执行的沙箱后端接口
  • 支持文件上传/下载、读写、编辑、搜索等操作

具体实现:

  • StateBackend:将文件存储在LangGraph代理状态中(临时存储)

    • 文件存在于代理的状态中(通过对话进行检查点)
    • 在线程内持久存在,但不会跨线程持久存在。
    • 适用于临时工作文件
  • FilesystemBackend:直接读写文件系统

    • 从实际磁盘进行读/写操作
    • 支持虚拟模式(沙盒化到根目录)
    • 与系统工具集成(ripgrep 用于 grep)
    • 安全特性:路径验证、大小限制、防止符号链接
  • StoreBackend:持久化存储后端

    • 使用 LangGraph 的 BaseStore 来实现持久化
    • 按 assistant_id 命名空间
    • 文件会在对话之间保留。
    • 有助于长期记忆或知识库的建立
  • CompositeBackend:组合多个后端,支持路由规则

    • 示例:/→ StateBackend、/memories/→ StoreBackend
    • 最长前缀匹配路由
    • 支持混合存储策略

3. 中间件模块(middleware/

提供增强代理功能的中间件:

文件系统中间件(filesystem.py

  • FilesystemMiddleware:为代理添加文件系统工具

  • 提供lsread_filewrite_fileedit_fileglobgrepexecute等工具

    • 工具 描述
      ls 列出目录中的文件及其元数据(大小、修改时间)
      read_file 读取文件内容并显示行号,支持大文件的偏移量/限制值。
      write_file 创建新文件
      edit_file 在文件中执行精确字符串替换(使用全局替换模式)
      glob 查找符合模式的文件(例如,**/*.py
      grep 支持多种输出模式(仅文件、包含上下文的内容或计数)搜索文件内容
  • 支持大文件结果自动转存到文件系统

    • 当工具结果超过令牌阈值时,该工具会自动将大型工具结果转储到文件系统中,从而防止上下文窗口饱和。工作原理:
      • 监控工具调用结果的大小(默认阈值:20,000 个令牌)
      • 当超出限制时,将结果写入文件。
      • 将工具结果替换为对文件的简洁引用
      • 如有需要,代理稍后可以读取该文件。

子代理中间件(subagents.py

  • SubAgentMiddleware:提供task工具来启动子代理
  • 支持通用目的代理和自定义专业代理
  • 子代理可以独立运行复杂任务,返回简洁结果

工具调用补丁中间件(patch_tool_calls.py

  • PatchToolCallsMiddleware:修复悬空工具调用

中间件如何作为工具发挥作用

DeepAgents中间件的核心设计理念是通过中间件向代理暴露工具。这是通过LangChain的中间件系统实现的,具体机制如下:

1. 工具生成机制

每个中间件都包含一个tools属性,该属性是一个工具列表。当中间件被添加到代理时,这些工具会被自动注入到代理的工具集中。

FilesystemMiddleware为例:

1
2
3
4
class FilesystemMiddleware(AgentMiddleware):
def __init__(self, *, backend=None, ...):
# ... 初始化代码
self.tools = _get_filesystem_tools(self.backend, custom_tool_descriptions)

_get_filesystem_tools()函数使用工具生成器创建具体的工具:

1
2
3
4
5
6
7
8
9
TOOL_GENERATORS = {
"ls": _ls_tool_generator,
"read_file": _read_file_tool_generator,
"write_file": _write_file_tool_generator,
"edit_file": _edit_file_tool_generator,
"glob": _glob_tool_generator,
"grep": _grep_tool_generator,
"execute": _execute_tool_generator,
}

每个工具生成器都创建一个具体的BaseTool实例,这些工具通过@tool装饰器定义,并与后端系统交互。

2. 工具与后端的连接

工具通过后端协议与存储系统交互。例如,read_file工具:

1
2
3
4
5
@tool(description=READ_FILE_TOOL_DESCRIPTION)
def read_file(file_path: str, runtime: ToolRuntime, offset: int = 0, limit: int = 500) -> str:
resolved_backend = _get_backend(backend, runtime)
file_path = _validate_path(file_path)
return resolved_backend.read(file_path, offset=offset, limit=limit)

工具接收运行时上下文,通过后端执行实际的文件操作,并返回格式化的结果。

3. 系统提示注入

中间件还负责向代理注入系统提示,指导代理如何使用这些工具:

1
2
3
4
5
6
7
8
9
def wrap_model_call(self, request: ModelRequest, handler):
# 构建动态系统提示
prompt_parts = [FILESYSTEM_SYSTEM_PROMPT]
if has_execute_tool and backend_supports_execution:
prompt_parts.append(EXECUTION_SYSTEM_PROMPT)

system_prompt = "\n\n".join(prompt_parts)
request = request.override(system_prompt=request.system_prompt + "\n\n" + system_prompt)
return handler(request)

4. 状态管理集成

中间件工具可以更新代理状态。例如,文件写入操作会返回Command对象来更新状态:

1
2
3
4
5
6
7
8
9
10
def write_file(file_path: str, content: str, runtime: ToolRuntime) -> Command | str:
res: WriteResult = resolved_backend.write(file_path, content)
if res.files_update is not None:
return Command(
update={
"files": res.files_update,
"messages": [ToolMessage(...)]
}
)
return f"Updated file {res.path}"

5. 子代理中间件的工作机制

SubAgentMiddleware提供了一个特殊的task工具,允许主代理启动子代理:

1
2
3
4
5
6
7
def task(description: str, subagent_type: str, runtime: ToolRuntime) -> str | Command:
if subagent_type not in subagent_graphs:
return f"We cannot invoke subagent {subagent_type}..."

subagent, subagent_state = _validate_and_prepare_state(subagent_type, description, runtime)
result = subagent.invoke(subagent_state)
return _return_command_with_state_update(result, runtime.tool_call_id)

子代理中间件:

  1. 管理一组预定义的子代理(包括通用目的代理和自定义代理)
  2. 提供task工具让主代理可以调用这些子代理
  3. 处理子代理的状态隔离和结果返回

6. 大文件处理机制

中间件还包含智能的大文件处理功能:

1
2
3
4
5
6
7
def _intercept_large_tool_result(self, tool_result: ToolMessage | Command, runtime: ToolRuntime):
if isinstance(tool_result, ToolMessage) and len(tool_result.content) > threshold:
# 将大文件结果保存到文件系统
file_path = f"/large_tool_results/{sanitized_id}"
result = resolved_backend.write(file_path, content)
# 返回指向文件的提示消息
return ToolMessage(f"Tool result too large, saved at {file_path}...")

关键特性

1. 可插拔后端系统

1
2
3
4
5
6
7
8
9
10
11
# 使用StateBackend(临时存储)
backend = StateBackend(runtime)

# 使用FilesystemBackend(文件系统存储)
backend = FilesystemBackend(root_dir="/path/to/root")

# 使用CompositeBackend(混合存储)
backend = CompositeBackend(
default=StateBackend(),
routes={"/memories/": StoreBackend()}
)

2. 智能文件系统工具

  • 安全路径验证:防止目录遍历攻击
  • 大文件处理:自动分页读取,避免上下文溢出
  • 智能搜索:支持ripgrep和Python正则表达式搜索
  • 批量操作:支持多文件上传/下载

3. 子代理系统

该组件允许主代理创建临时的“子代理”来执行孤立的多步骤任务。

它的优势在于:

  • 上下文隔离——子代理的工作不会干扰主代理的上下文。
  • 并行执行——多个子代理可以同时运行
  • 专业化——子代理可以拥有不同的工具/配置
  • 令牌效率——将大型子任务上下文压缩成单个结果

工作原理:

  • 主代理人拥有一个task工具
  • 调用时,会创建一个具有自身上下文的全新代理实例。
  • 子代理自主执行直至完成
  • 向主代理返回一份最终报告
  • 子代理是无状态的(不能发送多条消息)

默认子代理:

  • “通用”子代理自动可用
  • 默认包含文件系统工具
  • 可通过附加工具/中间件进行定制

自定义子代理:

  • 定义具有特定工具的专门子代理
  • 例如:代码审查员、网络研究员、测试运行员
  • 通过subagents参数配置

子代理解决了上下文膨胀问题。当代理使用输出量大的工具(例如网页搜索、文件读取、数据库查询)时,上下文窗口会迅速被中间结果填满。子代理将这些细节工作隔离出来——主代理只接收最终结果,而不是生成该结果的数十个工具调用。何时使用子代理:

  • ✅ 多步骤任务会使主代理的上下文变得混乱
  • ✅ 需要定制说明或工具的专业领域
  • ✅ 需要不同模型功能的任务
  • ✅ 当您希望主要代理人专注于高层协调工作时

何时不应使用子代理:

  • ❌ 简单的单步任务
  • ❌ 当您需要维护中间语境时
  • ❌ 当成本超过收益时
1
2
3
4
5
6
7
8
9
10
11
# 创建自定义子代理
subagents = [{
"name": "code-reviewer",
"description": "用于代码审查的专业代理",
"system_prompt": "你是一个专业的代码审查助手...",
"tools": [code_review_tool],
"model": "claude-sonnet-4"
}]

# 在深度代理中使用
agent = create_deep_agent(subagents=subagents)

4. 与LangChain的深度集成

  • 完全兼容LangChain的代理系统
  • 使用LangGraph进行状态管理
  • 支持LangChain的所有工具和模型
  • 继承LangChain的中间件系统

5. 对话历史摘要(上下文压缩)

当token使用量过大时,该机制会自动压缩旧的对话历史记录。配置:

  • 触发条件:token数量达到 17 万
  • 保留最近的 6 条消息
  • 模型会对较早的消息进行总结。

它的优势在于:

  • 支持超长时间对话而不会超出上下文限制
  • 既保留了近期历史的背景,又压缩了久远历史的内容。
  • 对代理透明(显示为特殊系统消息)

6. dangling-tool-call-repair

当工具调用在接收结果之前被中断或取消时,该机制可以修复消息历史记录。问题:

  • 代理请求工具调用:“请运行 X”
  • 工具调用被中断(用户取消、出错等)
  • 代理在 AIMessage 中看到了 tool_call,但没有看到对应的 ToolMessage。
  • 这将创建无效的消息序列

解决方案:

  • 检测没有结果的包含 tool_call 的 AIMessages。
  • 创建合成的 ToolMessage 响应,表明呼叫已取消
  • 在代理执行之前修复消息历史记录

它的优势在于:

  • 防止因消息链不完整而导致代理混淆
  • 优雅地处理中断和错误
  • 保持对话连贯性

7. 待办事项清单跟踪

该工具框架为write_todos代理提供了一个工具,可以用来维护结构化的任务列表。特征:

  • 跟踪多个任务及其状态(待处理、进行中、已完成)
  • 持续存在于代理状态
  • 帮助代理人组织复杂的多步骤工作
  • 适用于长时间运行的任务和计划

8. 人机交互

该安全机制会在指定工具调用时暂停代理执行,以便人工审批/修改。配置:

  • 将工具名称映射到中断配置
  • 例如:{"edit_file": True}每次编辑前暂停
  • 可以提供审批消息或修改工具输入

它的优势在于:

  • 破坏性作业的安全门
  • 在进行昂贵的 API 调用之前,需要进行用户验证。
  • 交互式调试和指导

通过 LangGraph 的中断功能支持人机协作工作流程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

@tool
def delete_file(path: str) -> str:
"""Delete a file from the filesystem."""
return f"Deleted {path}"

@tool
def read_file(path: str) -> str:
"""Read a file from the filesystem."""
return f"Contents of {path}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Sent email to {to}"

# Checkpointer is REQUIRED for human-in-the-loop
checkpointer = MemorySaver()

agent = create_deep_agent(
model="claude-sonnet-4-5-20250929",
tools=[delete_file, read_file, send_email],
interrupt_on={
"delete_file": True, # Default: approve, edit, reject
"read_file": False, # No interrupts needed
"send_email": {"allowed_decisions": ["approve", "reject"]}, # No editing
},
checkpointer=checkpointer # Required!
)

子代理中断也有自己的interrupt_on配置,这个配置可以覆盖主代理的配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
agent = create_deep_agent(
tools=[delete_file, read_file],
interrupt_on={
"delete_file": True,
"read_file": False,
},
subagents=[{
"name": "file-manager",
"description": "Manages file operations",
"system_prompt": "You are a file management assistant.",
"tools": [delete_file, read_file],
"interrupt_on": {
# Override: require approval for reads in this subagent
"delete_file": True,
"read_file": True, # Different from main agent!
}
}],
checkpointer=checkpointer
)

使用示例

基本使用

1
2
3
4
5
6
7
8
9
10
11
from deepagents import create_deep_agent

# 创建深度代理
agent = create_deep_agent(
model="claude-sonnet-4",
system_prompt="你是一个编程助手",
tools=[custom_tool1, custom_tool2]
)

# 运行代理
result = agent.invoke({"messages": [HumanMessage(content="写一个Python函数")]})

高级配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from deepagents.backends import FilesystemBackend
from deepagents.middleware import FilesystemMiddleware

# 自定义后端和中间件
backend = FilesystemBackend(root_dir="/workspace", virtual_mode=True)
agent = create_deep_agent(
backend=backend,
subagents=[{
"name": "research-analyst",
"description": "研究分析专家",
"system_prompt": "你是一个研究分析专家...",
"tools": [research_tools]
}],
interrupt_on={"write_file": True} # 写文件时中断等待人工确认
)

架构优势

1. 模块化设计

  • 每个组件都可以独立替换或扩展
  • 清晰的接口定义,便于自定义实现
  • 松耦合的架构设计

2. 生产就绪特性

  • 安全性:路径验证、权限控制、防注入
  • 可扩展性:支持自定义后端和中间件
  • 性能优化:大文件处理、并行子代理

3. 开发者友好

  • 详细的错误信息和文档
  • 丰富的配置选项
  • 与现有LangChain生态无缝集成

适用场景

  1. 复杂任务自动化:需要多步骤、多工具协作的任务
  2. 代码生成与审查:软件开发、代码重构、安全审查
  3. 研究分析:数据收集、分析、报告生成
  4. 文件操作密集型任务:文档处理、代码库管理
  5. 多代理协作:需要多个专业代理协同工作的场景

总结

DeepAgents作为LangChain生态系统的重要补充,为构建复杂的AI代理提供了强大的基础设施。其核心创新在于通过中间件系统将复杂功能(如文件操作、子代理管理)封装为工具,使得代理可以自然地使用这些高级功能。

中间件在DeepAgents中扮演着双重角色:

  1. 工具提供者:向代理暴露具体的功能工具
  2. 系统增强者:注入系统提示、管理状态、处理边缘情况

这种设计使得DeepAgents既保持了与LangChain的兼容性,又提供了更高级的抽象和更强大的功能。无论是简单的自动化任务还是复杂的多代理协作系统,DeepAgents都能提供可靠的技术支持,是下一代智能代理开发的重要工具。


注:本文基于对DeepAgents框架代码的深入分析,框架版本为当前工作空间中的安装版本。具体实现细节可能随版本更新而变化。