DeepAgent中的BackendProtocol协议学习
·
BackendProtocol 深度解析
一、什么是 BackendProtocol?
BackendProtocol 是 deepagents 定义的统一后端协议,用于抽象文件存储和执行环境。它是一个抽象基类(ABC),所有后端实现都必须遵循这个协议。
┌─────────────────────────────────────────────────────────┐
│ BackendProtocol │
│ (统一文件操作接口) │
├─────────────────────────────────────────────────────────┤
│ • ls_info() - 列出目录文件及元数据 │
│ • read() - 读取文件内容 │
│ • grep_raw() - 文本搜索 │
│ • glob_info() - 文件模式匹配 │
│ • write() - 写入新文件 │
│ • edit() - 编辑现有文件 │
│ • upload_files() - 上传文件 │
│ • download_files() - 下载文件 │
└─────────────────────────────────────────────────────────┘
▲
│ 扩展
│
┌─────────────────────────────────────────────────────────┐
│ SandboxBackendProtocol │
│ (增加沙箱执行能力) │
├─────────────────────────────────────────────────────────┤
│ • id - 沙箱唯一标识 │
│ • execute() - 执行 Shell 命令 │
│ • aexecute() - 异步执行 Shell 命令 │
└─────────────────────────────────────────────────────────┘
二、核心设计目标
2.1 存储后端解耦
# 所有文件数据统一表示为:
{
"content": list[str], # 文本行
"created_at": str, # ISO 时间戳
"modified_at": str, # ISO 时间戳
}
2.2 可切换的执行环境
┌─────────────────────────────────────────────────────────┐
│ 开发/测试环境 │
│ StateBackend │
│ (内存模拟,安全) │
└─────────────────────────────────────────────────────────┘
│
│ 切换 backend 参数
▼
┌─────────────────────────────────────────────────────────┐
│ 生产环境 │
│ FilesystemBackend │
│ (真实文件系统/S3/数据库) │
└─────────────────────────────────────────────────────────┘
三、应用场景
3.1 场景一:开发环境 vs 生产环境
from deepagents import create_deep_agent
from deepagents.backends import StateBackend, FilesystemBackend
# 开发环境 - 内存模拟(安全,不会真实写文件)
dev_agent = create_deep_agent(
backend=StateBackend,
)
# 生产环境 - 真实文件系统
prod_agent = create_deep_agent(
backend=FilesystemBackend(root_dir="/workspace"),
)
优势:
- 开发测试:用
StateBackend,操作只在内存中,不会影响真实文件系统 - 生产部署:用
FilesystemBackend,操作真实持久化
3.2 场景二:沙箱隔离执行
from deepagents.backends import SandboxBackend
# 容器化沙箱环境 - 执行命令安全隔离
sandbox_agent = create_deep_agent(
backend=SandboxBackend(
container_image="python:3.12",
timeout=300,
),
)
# 可以安全执行用户提供的代码
# 即使在容器内 rm -rf / 也不会影响宿主机
适用:
- 代码执行平台
- 用户自定义脚本运行
- 需要安全隔离的场景
3.3 场景三:云端文件存储
from deepagents.backends import S3Backend
# 使用 S3 作为文件存储后端
cloud_agent = create_deep_agent(
backend=S3Backend(
bucket="my-agent-files",
region="us-east-1",
),
)
# 所有文件操作都持久化到 S3
# agent 重启后文件依然存在
适用:
- 需要持久化存储
- 多实例共享文件
- 分布式 Agent 系统
四、使用案例
4.1 案例一:自定义 Backend 实现
from deepagents.backends.protocol import BackendProtocol, FileInfo, WriteResult
class DatabaseBackend(BackendProtocol):
"""将文件存储在数据库中的自定义后端"""
def __init__(self, db_connection):
self.db = db_connection
def ls_info(self, path: str) -> list[FileInfo]:
# 从数据库查询文件列表
files = self.db.query("SELECT * FROM files WHERE parent_path = ?", path)
return [
FileInfo(
path=f["path"],
is_dir=f["is_dir"],
size=f["size"],
modified_at=f["modified_at"]
)
for f in files
]
def read(self, file_path: str, offset: int = 0, limit: int = 2000) -> str:
# 从数据库读取文件内容
file_data = self.db.query_one(
"SELECT content FROM files WHERE path = ?",
file_path
)
if not file_data:
return "Error: File not found"
lines = file_data["content"].split("\n")[offset:offset+limit]
return "\n".join(f"{i+1}|{line}" for i, line in enumerate(lines))
def write(self, file_path: str, content: str) -> WriteResult:
# 写入文件到数据库
try:
self.db.execute(
"INSERT INTO files (path, content) VALUES (?, ?)",
file_path,
content
)
return WriteResult(path=file_path, files_update=None)
except Exception as e:
return WriteResult(error=str(e))
# ... 实现其他方法
# 使用自定义后端
agent = create_deep_agent(
backend=DatabaseBackend(my_db_connection),
)
五、协议方法详解
5.1 核心文件操作
| 方法 | 功能 | 典型用例 |
|---|---|---|
ls_info(path) | 列出目录内容 | 探索项目结构 |
read(path, offset, limit) | 分页读取文件 | 大文件分段读取 |
write(path, content) | 创建新文件 | 生成代码文件 |
edit(path, old, new) | 精确替换文本 | 修改代码 |
glob_info(pattern, path) | 模式匹配文件 | 查找所有 .py 文件 |
grep_raw(pattern, path, glob) | 文本搜索 | 搜索 TODO 注释 |
5.2 批量文件传输
# 批量上传
responses = backend.upload_files([
("/app/config.json", b'{"key": "value"}'),
("/app/data.txt", b"content"),
])
# 批量下载
responses = backend.download_files([
"/app/config.json",
"/app/data.txt",
])
六、架构优势总结
┌─────────────────────────────────────────────────────────┐
│ Agent 应用层 │
│ create_deep_agent() │
└─────────────────────────────────────────────────────────┘
│
│ 使用 BackendProtocol
▼
┌─────────────────────────────────────────────────────────┐
│ BackendProtocol │
│ (统一抽象接口层) │
└─────────────────────────────────────────────────────────┘
│ │ │
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ StateBackend│ │FilesystemBE │ │ SandboxBE │
│ (内存) │ │ (磁盘) │ │ (容器) │
└─────────────┘ └─────────────┘ └─────────────┘
核心优势:
- 可测试性 - 用
StateBackend单元测试,不影响真实文件系统 - 可替换性 - 轻松切换存储后端(本地/云端/数据库)
- 安全性 - 沙箱环境隔离危险操作
- 可扩展性 - 自定义 Backend 满足特殊需求
- 统一接口 - 上层代码无需关心底层实现
七、Backend 在 create_deep_agent 中的使用方式
使用流程图
┌─────────────────────────────────────────────────────────┐
│ create_deep_agent(backend=xxx) │
│ 用户传入 backend │
└─────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Line 193: backend = backend if backend is not None │
│ else (StateBackend) │
│ 默认值:StateBackend(内存模拟) │
└─────────────────────────────────────────────────────────┘
│
▼
┌───────────────┴───────────────┐
│ │
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ 子代理中间件栈 │ │ 主代理中间件栈 │
│ (gp_middleware) │ │(deepagent_middleware)
└─────────────────┘ └─────────────────┘
│ │
注入 backend 注入 backend
│ │
▼ ▼
FilesystemMiddleware(backend) FilesystemMiddleware(backend)
SummarizationMiddleware(backend) SummarizationMiddleware(backend)
SkillsMiddleware(backend) SkillsMiddleware(backend)
MemoryMiddleware(backend)
SubAgentMiddleware(backend)
具体注入位置
1️⃣ 通用子代理的中间件栈(Line 196-208)
gp_middleware = [
TodoListMiddleware(),
FilesystemMiddleware(backend=backend), # ← 注入
SummarizationMiddleware(
model=model,
backend=backend, # ← 注入
# ...
),
AnthropicPromptCachingMiddleware(),
PatchToolCallsMiddleware(),
]
if skills is not None:
gp_middleware.append(SkillsMiddleware(backend=backend, sources=skills)) # ← 注入
2️⃣ 用户自定义子代理的中间件栈(Line 236-252)
subagent_middleware = [
TodoListMiddleware(),
FilesystemMiddleware(backend=backend), # ← 注入
SummarizationMiddleware(
model=subagent_model,
backend=backend, # ← 注入
# ...
),
AnthropicPromptCachingMiddleware(),
PatchToolCallsMiddleware(),
]
if subagent_skills:
subagent_middleware.append(SkillsMiddleware(backend=backend, sources=subagent_skills)) # ← 注入
3️⃣ 主代理的中间件栈(Line 267-292)
deepagent_middleware = [
TodoListMiddleware(),
]
if memory is not None:
deepagent_middleware.append(MemoryMiddleware(backend=backend, sources=memory)) # ← 注入
if skills is not None:
deepagent_middleware.append(SkillsMiddleware(backend=backend, sources=skills)) # ← 注入
deepagent_middleware.extend([
FilesystemMiddleware(backend=backend), # ← 注入
SubAgentMiddleware(
backend=backend, # ← 注入
subagents=all_subagents,
),
SummarizationMiddleware(
model=model,
backend=backend, # ← 注入
# ...
),
AnthropicPromptCachingMiddleware(),
PatchToolCallsMiddleware(),
])
最终传递链
┌─────────────────────────────────────────────────────────┐
│ deepagent_middleware (7-8 个中间件) │
│ 每个中间件都持有 backend 引用 │
└─────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ create_agent(middleware=deepagent_middleware) │
│ 传递给 LangChain 基础框架 │
└─────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ 运行时:Agent 调用工具 │
│ → FilesystemMiddleware 使用 backend.read/write() │
│ → SkillsMiddleware 使用 backend 加载 skill 文件 │
│ → MemoryMiddleware 使用 backend 读取记忆文件 │
│ → SummarizationMiddleware 使用 backend 存储摘要 │
└─────────────────────────────────────────────────────────┘
关键总结
| 中间件 | 使用 backend 做什么 |
|---|---|
| FilesystemMiddleware | 文件操作(ls, read, write, edit, grep, glob) |
| SkillsMiddleware | 从 backend 加载 skill 文件(SKILL.md) |
| MemoryMiddleware | 从 backend 读取记忆文件(AGENTS.md) |
| SummarizationMiddleware | 将消息摘要存储到 backend |
| SubAgentMiddleware | 传递给子代理,子代理也使用相同的 backend |
一句话总结:backend 被注入到所有需要文件系统能力的中间件中,Agent 运行时通过这些中间件调用 backend 的方法来实现文件操作、技能加载、记忆读取等功能。
参考
[1]. deepagents-0.4.4.dist-info
更多推荐



所有评论(0)