跳到主要内容

存储后端

后端的基本概念

存储后端为 DeepAgents 定义了文件系统的抽象。虽然代理看到的是标准的文件操作接口(read_filewrite_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

检查

  1. 路径是否正确
  2. 后端是否有权限读取
  3. 虚拟模式是否阻止了访问

权限拒绝

症状PermissionError

解决

  • 检查文件系统权限
  • 对 FilesystemBackend 使用虚拟模式
  • 检查 SELinux 或 AppArmor 限制

Store 连接失败

症状Connection refused

解决

  • 确保 LangSmith 部署可访问
  • 检查认证是否正确
  • 查看网络配置

下一步