Backend系统概述
后端是智能体（Agent）文件系统工具的底层实现，它决定了Agent可以访问哪些存储介质（如内存、本地磁盘、数据库等）以及如何访问。
DeepAgents 通过一套文件系统工具（如 ls, read_file, write_file等）为Agent提供与外界存储交互的能力。这些工具并不直接操作存储，而是通过一个可插拔的后端（Backend）来执行实际的操作。后端是一个遵循特定协议（BackendProtocol）的组件。
Agent的文件系统工具调用后端，后端再根据配置将操作路由到不同的具体存储实现，DeepAgents 提供了多种开箱即用的后端，适用于不同场景。例如：
- 状态存储 (
StateBackend): 存储在 LangGraph 状态中，单次会话有效。 - 本地磁盘 (
FilesystemBackend): 访问宿主机的真实文件系统。 - 持久化存储 (
StoreBackend): 使用 LangGraph 的BaseStore实现跨会话持久化。 - 沙盒/本地Shell (
Sandbox,LocalShellBackend): 提供隔离或非隔离的执行环境。 - 复合后端 (
CompositeBackend): 作为路由器，将不同路径的请求分发到不同的后端。
默认后端是 StateBackend，它将文件存储在Agent的运行时状态中，仅在一次执行线程内有效。
# Agent在一个临时的、内存式的“文件系统”中工作
result = await agent.ainvoke({
"messages": [{
"role": "user",
"content": "请创建一个文件 /plan.txt， 内容为‘项目启动计划’，然后列出根目录 / 下的所有内容。"
}]
})
StateBackend (临时状态后端)
- 工作原理：将文件数据存储在 LangGraph 的Agent状态（
runtime.state）中。这些数据在同一个执行会话（thread）的多次调用间是持久的，但线程（会话）结束后即消失。当然：如果有Checkpointer快照机制，那么也能恢复。 - 最佳用途：作为Agent的草稿纸，用于暂存中间结果。也用于上下文的自动剪裁（offload）大文件输出。
注意：必须提供具体的Checkpointer。部署到LangSmith时可省略，平台会自动提供。
注意：在没有Checkpointer的情况下，是看不到刚刚创建的文件的。 子Agent与主Agent共享此状态后端，子Agent创建的文件对主Agent可见。
FilesystemBackend (本地磁盘后端)
- 工作原理：允许Agent读写本地机器上指定根目录（
root_dir）下的真实文件。 - ⚠️ 重要安全警告：此后端赋予Agent直接的文件系统访问权限。切勿在Web服务器、API等多租户生产环境使用。仅在可信的本地开发中使用。
安全建议：
始终设置
virtual_mode=True以启用路径沙盒，阻止Agent使用..或~访问根目录之外的路径。从可访问路径中排除包含密钥、密码的敏感文件（如
.env）。对于需要高安全的生产环境，考虑使用沙盒后端。
import os from deepagents import create_deep_agent from deepagents.backends import FilesystemBackend from langchain_core.stores import InMemoryStore from utils.llm_util import llm_zhipu from utils.tools_util import web_search # 创建一个临时目录作为Agent的“沙盒” temp_workspace = "./agent_workspace" os.makedirs(temp_workspace, exist_ok=True) checkpoint = InMemoryStore() agent = create_deep_agent( model=llm_zhipu, tools=[web_search], checkpoint=checkpoint, backend=FilesystemBackend( root_dir=temp_workspace, virtual_mode=True, ), system_prompt="你是一位财经分析师。请用简洁的语言总结网络搜索结果中的关键信息，按重要性排序并引用来源日期。今天的日期是2026年6月29日。", )
LocalShellBackend (本地Shell后端)
工作原理：在
FilesystemBackend的基础上，额外提供了一个execute工具，允许Agent在主机上执行任意的 Shell 命令。⚠️ 极度危险警告：此后端赋予Agent在您的主机上执行任意命令的最高权限。仅限在您完全信任Agent代码的本地开发环境中使用。禁止用于生产环境或处理不可信输入。
# 创建一个临时目录作为Agent的“沙盒” temp_workspace = "./agent_workspace" os.makedirs(temp_workspace, exist_ok=True) # 本地沙箱 backend = LocalShellBackend( root_dir=temp_workspace, virtual_mode=True, timeout=30, # 命令执行超时时间（秒） max_output_bytes=50000, # 命令输出最大字节数 # 设置环境变量，包含编码相关的配置 env={ # 获取当前Python解释器的完整路径 "PATH": f"{os.path.dirname(sys.executable)};{os.environ.get('PATH', '')}", }, ) agent = create_deep_agent( model=llm, tools=[web_search], checkpointer=checkpointer, backend=backend, system_prompt='你是一个助手，请根据用户输入的指令，进行相应的操作。' )
StoreBackend (LangGraph 存储后端)
- 工作原理：使用 LangGraph 的
BaseStore抽象（支持 Redis、Postgres、内存等实现）来存储文件，从而实现跨不同执行线程的持久化存储。 - 最佳用途：存储需要长期记忆的数据，例如用户偏好、跨对话知识库。
注意：必须提供具体的存储实现。部署到LangSmith时可省略，平台会自动提供。
# 案例2-4：配置一个具有持久化记忆的Agent
from deepagents.backends import StoreBackend
from deepagents import create_deep_agent
from langgraph.store.memory import InMemoryStore
agent = create_deep_agent(
model="gpt-4o",
backend=lambda rt: StoreBackend(rt),
store=InMemoryStore() # 使用内存存储，进程重启后丢失。生产环境可用RedisStore等。
)
# Agent写入 /memories/ 下的文件，在后续的新对话中仍可读取。
CompositeBackend (复合/路由后端)
工作原理：作为后端路由器，根据文件路径的前缀，将操作定向到不同的底层后端。
最佳用途：实现混合存储策略。例如，临时工作文件用
StateBackend，长期记忆用StoreBackend，特定目录映射到本地磁盘。# 定义复合后端：默认用StateBackend，但`/memories/`路径下的操作路由到StoreBackend composite_backend = lambda rt: CompositeBackend( default=StateBackend(rt), # 默认后端，用于临时文件 routes={ "/memories/": StoreBackend(rt), # 持久化路径 } ) agent = create_deep_agent( model="gpt-4o", backend=composite_backend, store=InMemoryStore() ) # 现在，Agent对`/scratch/plan.md`的读写是临时的，而对`/memories/user_pref.json`的读写是持久的。 composite_backend = lambda rt: CompositeBackend( default=StateBackend(rt), # 默认临时存储 routes={ "/memories/": FilesystemBackend(root_dir="/data/agent_memories", virtual_mode=True), "/shared_docs/": FilesystemBackend(root_dir="/company/docs", virtual_mode=True), }, ) agent = create_deep_agent(backend=composite_backend) # 路径解析： # `/scratch/temp.txt` -> StateBackend # `/memories/2025/note.md` -> 本地文件 `/data/agent_memories/2025/note.md` # `/shared_docs/api.md` -> 本地文件 `/company/docs/api.md` # `ls /` 命令会聚合来自所有后端的结果。
自定义后端
你可以实现 BackendProtocol接口，将任何存储系统（如云存储S3、数据库）暴露给Agent。
需要重写的核心方法的简要说明：
ls_info(path: str) -> list[FileInfo]
- •功能：列出指定路径下的文件和目录。
- •要求：返回的 FileInfo列表必须至少包含 path字段。应对结果按 path排序以保证输出确定性。
read(file_path: str, offset: int = 0, limit: int = 2000) -> str
- •功能：读取文件内容，支持从某行开始（offset）并限制行数（limit）。
- •要求：返回的字符串必须包含行号（如 “1: first line\n2: second line”）。文件不存在时返回错误字符串。
write(file_path: str, content: str) -> WriteResult
- •功能：创建新文件。默认应是“仅创建”（Create-only），即文件已存在时应返回错误。
- •要求：操作成功返回 WriteResult(path=…, files_update=…)。对于外部后端（如S3），files_update应为 None。
edit(file_path: str, old_string: str, new_string: str, replace_all: bool = False) -> EditResult
- •功能：替换文件中的文本。当 replace_all=False时，old_string必须在文件中精确出现一次，否则返回错误。这是为了防止意外替换。
grep_raw(pattern: str, path: str | None = None, glob: str | None = None) -> list[GrepMatch] | str
- •功能：在文件中搜索正则表达式 pattern。可以限制在特定 path（目录）下，或使用 glob模式过滤文件。
- •要求：正则表达式无效时，返回错误字符串而非抛出异常。
glob_info(pattern: str, path: str = “/“) -> list[FileInfo]
- •功能：使用 glob 模式（如 *.txt）在指定路径下搜索文件。
设计一个极简的“内存字典”后端示例：
from deepagents.backends.protocol import BackendProtocol, WriteResult, EditResult
from deepagents.backends.utils import FileInfo, GrepMatch
from datetime import datetime
import re
class DictBackend(BackendProtocol):
"""一个将文件存储在内存字典中的简单后端。"""
def __init__(self):
self.files = {} # 路径 -> 内容
self.metadata = {} # 路径 -> 元数据（大小，修改时间）
def ls_info(self, path: str) -> list[FileInfo]:
# 列出以`path`为前缀的“文件”和“目录”
result = []
seen_dirs = set()
for file_path in self.files.keys():
if file_path.startswith(path):
# 处理目录项
remaining = file_path[len(path):]
if '/' in remaining:
dir_name = path + remaining.split('/')[0] + '/'
if dir_name not in seen_dirs:
seen_dirs.add(dir_name)
result.append(FileInfo(path=dir_name, is_dir=True))
else:
# 文件项
meta = self.metadata.get(file_path, {})
result.append(FileInfo(
path=file_path,
is_dir=False,
size=meta.get('size', 0),
modified_at=meta.get('modified_at')
))
# result.sort(key=lambda x: x.path)
return result
def read(self, file_path: str, offset: int = 0, limit: int = 2000) -> str:
if file_path not in self.files:
return f"Error: File '{file_path}' not found"
content = self.files[file_path]
# 简单模拟分页：按行处理
lines = content.splitlines(keepends=True)
start_line = offset
end_line = start_line + limit if limit > 0 else len(lines)
selected_lines = lines[start_line:end_line]
result = ''.join(f"{start_line + i + 1}: {line}" for i, line in enumerate(selected_lines))
return result if result else "(end of file)"
def write(self, file_path: str, content: str) -> WriteResult:
if file_path in self.files:
return WriteResult(error=f"File '{file_path}' already exists (create-only).")
self.files[file_path] = content
self.metadata[file_path] = {
'size': len(content),
'modified_at': datetime.now().isoformat()
}
# 对于自定义后端，files_update 通常为 None
return WriteResult(path=file_path, files_update=None)
# 注意：为简洁起见，省略了 grep_raw, glob_info, edit 的完整实现。
# 一个完整的实现需要填充这些方法。
def grep_raw(self, pattern: str, path: str | None = None, glob: str | None = None) -> list[GrepMatch] | str:
# 简化实现：仅在全文件搜索
try:
re.compile(pattern)
except re.error as e:
return f"Invalid regex pattern: {e}"
matches = []
for file_path, content in self.files.items():
for i, line in enumerate(content.splitlines()):
if re.search(pattern, line):
matches.append(GrepMatch(path=file_path, line=i+1, text=line))
return matches
def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
# 简化实现：使用 fnmatch
import fnmatch
all_files = [FileInfo(path=p, is_dir=False, size=self.metadata[p]['size'], modified_at=self.metadata[p]['modified_at']) for p in self.files.keys() if p.startswith(path)]
matched = [fi for fi in all_files if fnmatch.fnmatch(fi.path, path.rstrip('/') + '/' + pattern)]
return matched
def edit(self, file_path: str, old_string: str, new_string: str, replace_all: bool = False) -> EditResult:
if file_path not in self.files:
return EditResult(error=f"File '{file_path}' not found")
content = self.files[file_path]
if replace_all:
new_content = content.replace(old_string, new_string)
occurrences = content.count(old_string)
else:
if content.count(old_string) != 1:
return EditResult(error=f"Found {content.count(old_string)} occurrences of '{old_string}'. For safety, edit requires exactly one match unless replace_all=True.")
new_content = content.replace(old_string, new_string, 1)
occurrences = 1
if new_content == content:
return EditResult(error=f"String '{old_string}' not found.")
self.files[file_path] = new_content
self.metadata[file_path]['size'] = len(new_content)
self.metadata[file_path]['modified_at'] = datetime.now().isoformat()
return EditResult(path=file_path, files_update=None, occurrences=occurrences)
文件后端和checkpointer、store的关系
- DeepAgents 的 Backend
- 设计目标：为 Agent 提供文件系统语义的抽象。它的核心接口是
ls_info,read,write,edit等，让Agent感觉自己在一个真实的目录树下操作文件。 - 数据模型：文件/目录树。操作对象是文件路径和文件内容。
- 持久化范围：由具体实现决定。
StateBackend是线程内，FilesystemBackend是进程/机器内，StoreBackend可以跨进程/会话。 - 典型用例：Agent的工作空间、长期记忆存储 (
/memories/)、技能和文档的加载来源。
- 设计目标：为 Agent 提供文件系统语义的抽象。它的核心接口是
- LangGraph 的 Checkpointer
- 设计目标：为 Agent 或者 StateGraph 提供状态快照与恢复机制。用于实现对话的持久化、暂停/继续、回溯以及人类介入审核（Human-in-the-loop）。
- 数据模型：序列化的工作流状态。它保存的是整个
State对象的检查点（checkpoint），包括所有通道的消息、变量值等。 - 持久化范围：跨会话持久化工作流状态。允许用户离开后，稍后从完全相同的地方继续对话。
- 典型用例：聊天机器人记住之前的对话上下文；一个长时间运行的任务支持暂停和继续；需要人工审批节点的多步骤工作流。
- LangChain 的 Store (BaseStore)
- 设计目标：一个通用的、需要自定义操作的 长期存储。它是 LangGraph 存储层的底层抽象，非常简单。
- 数据模型：命名空间下的键值对。基本操作是
get,set,delete,list。 - 持久化范围：由具体实现决定（如
InMemoryStore,RedisStore,PostgresStore）。 - 典型用例：为
Checkpointer或StoreBackend提供底层存储驱动。也可直接用于缓存、会话存储等任何需要简单KV存储的场景。