存储后端
后端的基本概念
存储后端为 DeepAgents 定义了文件系统的抽象。虽然代理看到的是标准的文件操作接口(read_file、write_file 等),但数据实际存储位置由后端决定。
后端的作用
代理代码
↓
文件系统工具接口
↓
后端抽象层 ← 决定数据存储位置和方式
↓
具体存储(本地磁盘、内存、数据库等)
可用的后端类型
1. StateBackend(默认,临时存储)
特点:
- 存储在 LangGraph 状态中
- 仅在单个线程内持久化
- 自动清理和淘汰
适用场景:
- 单次对话的临时文件
- 开发和测试
- 代理的工作区
配置:
from deepagents import create_deep_agent
from deepagents.backends import StateBackend
agent = create_deep_agent(
backend=StateBackend(), # 显式指定(其实是默认值)
)
2. FilesystemBackend(本地磁盘)
特点:
- 读写真实的文件系统
- 完全控制磁盘访问
- 需要仔细的安全考虑
适用场景:
- 本地开发工具
- CLI 应用
- CI/CD 集成
配置:
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend
agent = create_deep_agent(
backend=FilesystemBackend(
root_dir="/home/user/agent-workspace/", # 根目录
virtual_mode=True, # 启用路径沙盒化
),
)
安全警告:
- 启用
virtual_mode=True防止目录遍历攻击 - 限制根目录到安全位置
- 评估与网络工具结合的风险
3. LocalShellBackend(本地Shell)
特点:
- 包含文件系统操作
- 加上
execute工具允许shell命令执行 - 最高权限,最高风险
适用场景:
- 本地开发和自动化
- 受信的开发环境
- 绝不用于生产
配置:
from deepagents import create_deep_agent
from deepagents.backends import LocalShellBackend
agent = create_deep_agent(
backend=LocalShellBackend(
root_dir="/tmp/safe-workspace/",
timeout=120, # 命令执行超时(秒)
max_output_bytes=100000, # 限制输出大小
env={
"PATH": "/usr/bin:/bin",
"HOME": "/tmp/safe-workspace/"
}
),
)
使用示例:
result = agent.invoke({
"messages": [{
"role": "user",
"content": "编译这个 C 程序"
}]
})
# 代理会调用 execute 工具来运行 gcc 命令
4. StoreBackend(LangGraph Store,持久化)
特点:
- 数据持久化到 LangGraph Store
- 跨多个线程可访问
- 支持命名空间隔离
适用场景:
- 长期记忆存储
- 多用户系统中的用户数据隔离
- 生产环境中的可靠存储
配置:
from deepagents import create_deep_agent
from deepagents.backends import StoreBackend
from langgraph.store.memory import InMemoryStore
# 开发环境使用内存存储
store = InMemoryStore()
agent = create_deep_agent(
backend=StoreBackend(store=store),
store=store, # 传递相同的 store 实例
)
生产环境:
# 生产环境会自动使用 LangSmith 的存储
# 无需手动创建 store 实例
agent = create_deep_agent(
backend=StoreBackend(), # 自动使用平台存储
)
5. CompositeBackend(复合后端,路由)
特点:
- 为不同的路径指定不同的后端
- 灵活组合多种存储策略
- 实现精细的数据管理
适用场景:
- 临时数据使用内存,重要数据使用持久化存储
- 不同类型数据使用不同存储
- 混合部署环境
配置:
from deepagents import create_deep_agent
from deepagents.backends import (
CompositeBackend,
StateBackend,
StoreBackend,
FilesystemBackend,
)
agent = create_deep_agent(
backend=CompositeBackend(
default=StateBackend(), # 默认后端
routes={
"/memories/": StoreBackend(), # 记忆存储在 Store
"/workspace/": StateBackend(), # 工作区在状态中
"/uploads/": FilesystemBackend( # 上传文件在本地
root_dir="/data/uploads/"
),
"/temp/": StateBackend(), # 临时数据在状态中
}
),
)
实际应用示例
例1:单线程应用(开发工具)
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend
# 用于单个用户的桌面应用
agent = create_deep_agent(
backend=FilesystemBackend(
root_dir="./workspace/",
virtual_mode=True,
),
system_prompt="你是一个Python代码助手。你可以编辑和运行本地文件。",
)
result = agent.invoke({
"messages": [{
"role": "user",
"content": "读取并分析 app.py 文件"
}]
})
例2:多用户系统(SaaS)
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from langgraph.store.memory import InMemoryStore
class UserAgentFactory:
def __init__(self):
self.store = InMemoryStore()
def create_user_agent(self, user_id: str):
"""为特定用户创建代理"""
# 每个用户的数据存储在独立的命名空间
return create_deep_agent(
backend=CompositeBackend(
default=StateBackend(),
routes={
"/memories/": StoreBackend(
store=self.store,
namespace=lambda ctx: (user_id,), # 按用户隔离
),
"/uploads/": FilesystemBackend(
root_dir=f"/data/users/{user_id}/",
virtual_mode=True,
),
}
),
store=self.store,
)
# 使用
factory = UserAgentFactory()
user1_agent = factory.create_user_agent("user-123")
user2_agent = factory.create_user_agent("user-456")
# 每个代理有隔离的数据存储
result1 = user1_agent.invoke(...)
result2 = user2_agent.invoke(...)
例3:混合存储策略
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
agent = create_deep_agent(
backend=CompositeBackend(
default=StateBackend(), # 会话临时数据
routes={
# 长期数据
"/memories/agent-knowledge/": StoreBackend(),
"/memories/user-preferences/": StoreBackend(),
# 工作区
"/workspace/current/": StateBackend(),
# 永久数据
"/archive/": FilesystemBackend(
root_dir="/secure/archive/",
virtual_mode=True,
),
}
)
)
后端的选择决策树
需要什么存储?
├─ 单次会话临时文件?
│ └─→ StateBackend
│
├─ 本地磁盘访问?
│ └─→ FilesystemBackend
│
├─ 本地Shell执行?
│ └─→ LocalShellBackend
│
├─ 跨会话持久化?
│ └─→ StoreBackend
│
└─ 混合需求?
└─→ CompositeBackend
协议参考
所有后端实现都必须遵循 DeepAgents 后端协议:
from abc import ABC, abstractmethod
class Backend(ABC):
@abstractmethod
def ls(self, path: str) -> list[str]:
"""列出目录内容"""
pass
@abstractmethod
def read_file(self, path: str) -> str:
"""读取文件内容"""
pass
@abstractmethod
def write_file(self, path: str, content: str) -> None:
"""写入文件"""
pass
@abstractmethod
def edit_file(self, path: str, edits: list[tuple]) -> None:
"""编辑文件(替换操作)"""
pass
@abstractmethod
def rm(self, path: str) -> None:
"""删除文件或目录"""
pass
@abstractmethod
def glob(self, pattern: str) -> list[str]:
"""按模式查找文件"""
pass
最佳实践
1. 正确使用虚拟模式
# ✅ 始终启用虚拟模式防止目录遍历
backend = FilesystemBackend(
root_dir="/safe/path/",
virtual_mode=True, # 重要!
)
# ❌ 危险:禁用虚拟模式
backend = FilesystemBackend(
root_dir="/",
virtual_mode=False, # 代理可以访问整个文件系统!
)
2. 路径隔离
# ✅ 按用户隔离存储
namespace_func = lambda ctx: (ctx.runtime.server_info.user_id,)
# ✅ 按助手隔离存储
namespace_func = lambda ctx: (ctx.runtime.server_info.assistant_id,)
# ✅ 多级隔离
namespace_func = lambda ctx: (
ctx.runtime.server_info.organization_id,
ctx.runtime.server_info.user_id,
)
3. 监控和日志
from deepagents.backends import StateBackend
from deepagents import create_deep_agent
class LoggingBackend(StateBackend):
def write_file(self, path: str, content: str):
print(f"[WRITE] {path} ({len(content)} bytes)")
super().write_file(path, content)
def read_file(self, path: str):
print(f"[READ] {path}")
return super().read_file(path)
agent = create_deep_agent(
backend=LoggingBackend(),
)
故障排查
文件不存在错误
症状:FileNotFoundError: /path/to/file
检查:
- 路径是否正确
- 后端是否有权限读取
- 虚拟模式是否阻止了访问
权限拒绝
症状:PermissionError
解决:
- 检查文件系统权限
- 对 FilesystemBackend 使用虚拟模式
- 检查 SELinux 或 AppArmor 限制
Store 连接失败
症状:Connection refused
解决:
- 确保 LangSmith 部署可访问
- 检查认证是否正确
- 查看网络配置