1. 背景Agent 的文件为什么需要隔离在使用 DeepAgents 构建 Agent 时可以通过FilesystemBackend为 Agent 提供文件读写、编辑、搜索等能力。最初的实现比较简单直接给 Agent 指定一个固定的工作目录FilesystemBackend( root_dir/agent_files )这种方式在单用户、单会话场景下没有问题。但当系统存在多个聊天窗口甚至多个用户同时使用 Agent 时就会出现一个明显的问题/agent_files ├── report.docx ├── test.xlsx ├── result.md └── ...所有会话都在操作同一个目录。例如用户 A / 会话 1 └── report.docx 用户 B / 会话 2 └── report.docx两个会话都创建report.docx时就可能产生文件覆盖、读取错误等问题。因此文件系统至少应该按照会话进行隔离/agent_files ├── thread_001 │ ├── report.docx │ └── result.md │ ├── thread_002 │ ├── report.docx │ └── result.md │ └── thread_003 └── test.xlsx最终希望达到的效果是一个thread_id对应一个独立的 Agent 工作目录。即FilesystemBackend( root_dir/agent_files/{thread_id} )这样不同会话之间的文件天然隔离。2. 问题拆解整个问题实际上可以拆成两个子问题问题一如何获取当前会话的thread_idAgent 每次运行时都需要知道当前到底是哪一个会话只有拿到thread_id才能构造/agent_files/{thread_id}问题二如何动态创建FilesystemBackend传统写法是backend FilesystemBackend( root_dir/agent_files )这里的root_dir是固定的。但我们需要的是backend FilesystemBackend( root_dirf/agent_files/{thread_id} )也就是说FilesystemBackend的创建必须能够感知当前 Agent 的运行上下文。这两个问题解决之后会话级文件隔离基本就完成了。3. 获取当前会话的 thread_id在 LangGraph / LangChain 的执行过程中thread_id会随着运行上下文传递。实际开发过程中可以从不同的上下文入口获取它。主要考虑两种方式Runtimevar_child_runnable_config3.1 从 Runtime 获取在 LangGraph 中部分 Node、中间件等执行逻辑可以拿到Runtime。例如from langchain.agents.middleware import before_model, AgentState from langgraph.runtime import Runtime before_model() def do( state: AgentState, runtime: Runtime ): # 当前运行逻辑 return None这里的runtime就是当前执行过程中的运行时上下文。因此可以尝试从 Runtime 的配置中获取thread_id例如def get_thread_id_from_runtime(runtime): config getattr(runtime, config, None) if config: return config.get( configurable, {} ).get(thread_id) return None这里需要注意一个实际开发中的问题不同调用位置拿到的 Runtime 结构可能并不完全一样。所以实际项目中不应该过度依赖某一个固定属性而应该做一定的兼容处理。4. 从 var_child_runnable_config 获取另一种方式是使用 LangChain 提供的var_child_runnable_config它本质上是一个用于执行上下文传递的 Context Variable。当前 Runnable 执行链中的配置可以从这里获取。例如from langchain_core.runnables.config import var_child_runnable_config def get_thread_id_from_context(): config var_child_runnable_config.get() if not config: return None return config.get( configurable, {} ).get(thread_id)这里同样可以从configurable中获取thread_id5. 为什么要封装 get_thread_id既然存在多个获取入口那么直接在业务代码里写runtime.config...显然不太合适。因为以后其他模块也可能需要thread_id用户信息当前运行上下文其他 configurable 参数因此可以单独建立一个运行时工具模块content/ └── utils/ └── runtime_util.py将这些与 Runtime 相关的工具统一放进去。例如当前项目中的实现from langchain_core.runnables.config import var_child_runnable_config def get_thread_id(runtimeNone): # 尝试从 runtime 获取 config getattr(runtime, config, None) if config is None: configurable getattr(runtime, context, None) else: configurable config.get(configurable, None) # 如果 runtime 中没有再从上下文变量获取 if configurable is None: current_config var_child_runnable_config.get() if current_config is not None: configurable current_config.get( configurable, {} ) # 最终仍然无法获取 if configurable is None: return default return configurable.get( thread_id, default )这样业务代码就不需要关心thread_id到底是从 Runtime 还是 Context Variable 中拿到的。只需要thread_id get_thread_id(runtime)即可。6. 第二个问题如何动态创建 FilesystemBackend拿到thread_id后下一步自然会想到def create_backend(runtime): thread_id get_thread_id(runtime) root_dir os.path.join( ROOT_PATH_AGENT, thread_id ) os.makedirs( root_dir, exist_okTrue ) return FilesystemBackend( root_dirroot_dir )然后create_deep_agent( backendcreate_backend )看起来已经解决问题了。但这里有一个非常关键的地方create_deep_agent的backend参数并不只能接收一个固定的 Backend 实例。通过查看源码可以看到它支持BackendProtocol | BackendFactory | None这就提供了一个非常重要的扩展点。7. BackendProtocol 与 BackendFactory7.1 BackendProtocolBackendProtocol可以理解为文件后端需要遵守的一套接口规范。也就是说只要一个对象实现了规定的文件操作方法就可以作为 Backend 使用。例如ls_info read write edit grep_raw glob_infoFilesystemBackend本身就是这一套协议的具体实现。7.2 BackendFactory更关键的是BackendFactory。源码定义可以概括为BackendFactory Callable[ [ToolRuntime], BackendProtocol ]换句话说BackendFactory │ │ runtime ▼ 创建 Backend │ ▼ BackendProtocol也就是说backend不一定非要提前创建好backend FilesystemBackend(...)也可以传进去一个runtime - backend形式的可调用对象。这正好解决了我们的问题。8. 第一版方案BackendFactory 动态创建因此可以实现def create_session_backend(runtime): thread_id get_thread_id(runtime) root_dir os.path.join( ROOT_PATH_AGENT, thread_id ) os.makedirs( root_dir, exist_okTrue ) backend FilesystemBackend( root_dirroot_dir ) return backend然后agent create_deep_agent( modelget_llm(), tools[], backendcreate_session_backend, system_promptprompt, )执行流程变成Agent 执行 │ ▼ BackendFactory(runtime) │ ▼ 获取 thread_id │ ▼ /agent_files/{thread_id} │ ▼ 创建 FilesystemBackend │ ▼ Agent 使用该 Backend例如thread_001 ↓ /agent_files/thread_001 ↓ FilesystemBackend(root_dir/agent_files/thread_001)另一个会话thread_002 ↓ /agent_files/thread_002 ↓ FilesystemBackend(root_dir/agent_files/thread_002)这样就实现了会话隔离。9. 实际运行后发现的新问题但是第一版方案在实际运行时又出现了一个问题。问题出现在os.makedirs( root_dir, exist_okTrue )这里。BackendFactory的创建过程可能处在异步执行环境中。而os.makedirs()属于同步文件系统 I/O。如果在不合适的异步执行上下文中直接执行同步 I/O就可能造成事件循环阻塞。这时候问题就从“怎么动态创建 Backend”变成了“怎么动态创建 Backend同时避免在 BackendFactory 阶段执行不必要的同步 I/O”进一步分析后可以发现这里其实还存在一个设计上的浪费为什么一定要在创建 Backend 的时候就创建目录假设用户打开了一个新的聊天窗口用户打开 thread_001然后只是你好 今天天气怎么样 介绍一下你自己整个过程中根本没有进行文件操作。但我们的代码已经提前执行os.makedirs(/agent_files/thread_001)实际上没有必要。因此这里可以进一步优化。10. 第二版方案LazyFilesystemBackend解决方法就是Backend 可以先创建但真正的 FilesystemBackend 和工作目录延迟到第一次文件操作时再创建。这就是 Lazy Loading延迟加载的思路。整体执行过程变成创建 Backend │ ├── 获取 thread_id ├── 计算 root_dir └── 暂时不创建目录 │ ▼ Agent 是否操作文件 │ ┌───┴───┐ │ │ 否 是 │ │ ▼ ▼ 什么都不做 创建目录 │ ▼ FilesystemBackend │ ▼ 执行文件操作这样既解决了同步 I/O 时机的问题也避免了无意义的目录创建。11. LazyFilesystemBackend 的设计我们可以自己实现一个LazyFilesystemBackend它本身实现BackendProtocol但它并不直接负责真正的文件操作。它内部维护一个真正的FilesystemBackend结构可以理解为LazyFilesystemBackend │ ├── runtime ├── thread_id ├── root_dir └── _backend │ └── FilesystemBackend初始化的时候_backend None第一次执行read() write() edit() ls_info() ...时再真正创建FilesystemBackend12. _ensure_backend整个设计的核心核心代码其实非常简单def _ensure_backend(self): if self._backend is None: os.makedirs( self._root_dir, exist_okTrue ) self._backend FilesystemBackend( root_dirself._root_dir, virtual_modeTrue ) return self._backend这个方法负责保证只要真正需要文件操作就一定存在一个可用的 FilesystemBackend。第一次调用_backend None ↓ 创建目录 ↓ 创建 FilesystemBackend ↓ 保存到 _backend第二次调用_backend ! None ↓ 直接返回已有实例因此它实际上是延迟初始化 实例缓存。13. 文件操作如何转发有了_ensure_backend()后其他方法就非常简单。例如def read( self, file_path: str, offset: int 0, limit: int 2000 ): return self._ensure_backend().read( file_path, offset, limit )write()def write( self, file_path: str, content: str ): return self._ensure_backend().write( file_path, content )edit()def edit( self, file_path: str, old_string: str, new_string: str, replace_all: bool False ): return self._ensure_backend().edit( file_path, old_string, new_string, replace_all )其他方法同理。最终形成Agent │ ▼ LazyFilesystemBackend │ │ _ensure_backend() ▼ FilesystemBackend │ ▼ 真实文件系统这里的LazyFilesystemBackend实际上承担了一层代理/适配作用。Agent 并不知道后面是否已经初始化了真正的文件系统后端。14. 最终实现项目中最终实现的核心代码如下from deepagents.backends import ( FilesystemBackend, BackendProtocol ) from base.configs import ROOT_PATH_AGENT from content.utils import runtime_util as rt import os from typing import Optional class LazyFilesystemBackend(BackendProtocol): def __init__(self, runtime): self.runtime runtime # 真正的 FilesystemBackend self._backend: Optional[ FilesystemBackend ] None # 获取当前会话 self._thread_id rt.get_thread_id( runtime ) # 构造当前会话的工作目录 self._root_dir os.path.join( ROOT_PATH_AGENT, self._thread_id ) def _ensure_backend(self): if self._backend is None: # 第一次执行文件操作时才创建目录 os.makedirs( self._root_dir, exist_okTrue ) # 创建真正的文件系统后端 self._backend FilesystemBackend( root_dirself._root_dir, virtual_modeTrue ) return self._backend def ls_info(self, path: str): return self._ensure_backend().ls_info(path) def read( self, file_path: str, offset: int 0, limit: int 2000 ): return self._ensure_backend().read( file_path, offset, limit ) def write( self, file_path: str, content: str ): return self._ensure_backend().write( file_path, content ) def edit( self, file_path: str, old_string: str, new_string: str, replace_all: bool False ): return self._ensure_backend().edit( file_path, old_string, new_string, replace_all ) def grep_raw( self, pattern: str, path: Optional[str] None, glob: Optional[str] None ): return self._ensure_backend().grep_raw( pattern, path, glob ) def glob_info( self, pattern: str, path: str / ): return self._ensure_backend().glob_info( pattern, path ) def create_session_backend(runtime): return LazyFilesystemBackend(runtime)15. 接入 Agent原本 Agent 可能是self.agent create_deep_agent( modelget_llm(), tools[], backendFilesystemBackend( root_dirROOT_PATH_AGENT ), system_promptprompt, )现在修改为from deepagents import create_deep_agent from conn.llms import get_small_llm as get_llm from content.others import mybackend class AllAgent: def __init__(self): prompt 你是一个通用智能体 回答用户用中文。 self.agent create_deep_agent( modelget_llm(), tools[], backendmybackend.create_session_backend, system_promptprompt, )这里最关键的一行就是backendmybackend.create_session_backend注意这里传入的不是create_session_backend()而是create_session_backend因为这里需要把工厂函数本身交给框架。之后由框架在 Agent 执行过程中根据当前runtime调用它。16. 最终执行流程完整链路可以概括为用户打开聊天窗口 │ ▼ 生成 thread_id │ ▼ Agent 开始执行 │ ▼ BackendFactory(runtime) │ ▼ create_session_backend(runtime) │ ▼ LazyFilesystemBackend(runtime) │ ├── 获取 thread_id │ └── 计算 root_dir │ ▼ Agent 是否执行文件操作 │ ┌─────┴─────┐ │ │ 否 是 │ │ ▼ ▼ 不创建 _ensure_backend() 文件目录 │ ▼ 创建 thread 目录 │ ▼ FilesystemBackend │ ▼ 执行文件操作例如thread_id abc123最终得到/agent_files/abc123/另一个会话thread_id xyz456得到/agent_files/xyz456/两个 Agent 会话之间的文件完全分离。17. 项目目录结构最终相关代码结构content/ ├── others/ │ └── mybackend.py │ ├── utils/ │ └── runtime_util.py │ └── all_agent.py职责也比较清晰runtime_util.py ↓ 负责运行时信息获取 ↓ thread_id mybackend.py ↓ 负责文件系统后端 ↓ LazyFilesystemBackend ↓ create_session_backend all_agent.py ↓ Agent 创建 ↓ 注入 BackendFactory18. 这次改造真正解决了什么18.1 会话隔离从所有会话 ↓ /agent_files变成thread_001 ↓ /agent_files/thread_001 thread_002 ↓ /agent_files/thread_002不同会话拥有独立工作空间。18.2 解决同名文件冲突例如两个会话都生成report.docx现在实际对应/agent_files/thread_001/report.docx /agent_files/thread_002/report.docx不会互相覆盖。18.3 降低跨会话文件访问风险Agent 的文件操作始终发生在当前thread_id对应的工作目录下。因此从文件系统层面建立了基本的会话边界。需要注意的是这属于应用层面的文件隔离设计并不等同于完整的多租户安全体系。如果真正用于生产环境还需要进一步考虑权限控制、路径穿越、用户与 thread 的绑定关系、目录生命周期以及数据清理等问题。19. 为什么最后选择 Lazy Loading这个改造实际上经历了两个版本。第一版BackendFactory ↓ 获取 thread_id ↓ 创建目录 ↓ 创建 FilesystemBackend ↓ 返回优点是简单直接。但是存在两个问题第一Backend 创建阶段就执行了同步 I/O。这可能与异步执行环境产生冲突或造成事件循环阻塞。第二没有必要提前创建目录。用户可能只是聊天并没有执行任何文件操作。第二版改成BackendFactory ↓ 获取 thread_id ↓ 计算 root_dir ↓ 暂不创建目录 ↓ 等待真正的文件操作 ↓ _ensure_backend() ↓ 创建目录 FilesystemBackend这样Backend 初始化更加轻量文件目录按需创建避免初始化阶段执行不必要的 I/O文件操作逻辑集中在_ensure_backend()原有FilesystemBackend的能力仍然可以复用因此最终选择了第二种方案。20. 这次问题中比较值得记录的源码分析思路这次改造真正有价值的地方其实并不只是写了一个LazyFilesystemBackend而是如何从框架的约束中找到扩展点。最开始遇到的问题是FilesystemBackend的root_dir是固定的怎么动态设置如果直接从FilesystemBackend本身入手很容易陷入怎么修改 FilesystemBackend 怎么重新实现 FilesystemBackend但继续往上看create_deep_agent( backend... )发现BackendProtocol | BackendFactory再继续看BackendFactory Callable[ [ToolRuntime], BackendProtocol ]于是整个思路就发生了变化不是修改 FilesystemBackend ↓ 而是利用 BackendFactory ↓ 让 Backend 根据 runtime 动态生成 ↓ 再通过 Lazy Backend 延迟真正的文件系统初始化这也是使用成熟框架时比较重要的一种思路遇到框架无法直接满足的需求时先寻找框架提供的扩展点而不是马上绕开框架重写整个功能。21. 最终方案总结整个方案可以浓缩成四层① Runtime ↓ 获取 thread_id ② BackendFactory ↓ 根据 thread_id 创建会话级 Backend ③ LazyFilesystemBackend ↓ 延迟真正的文件系统初始化 ④ FilesystemBackend ↓ 实际执行 read / write / edit / grep / glob 等操作核心代码关系create_deep_agent │ │ backend ▼ create_session_backend │ │ runtime ▼ LazyFilesystemBackend │ │ 第一次文件操作 ▼ FilesystemBackend │ │ root_dir ▼ /agent_files/{thread_id}最终实现了基于thread_id的 Agent 会话级文件系统隔离并通过BackendFactory Lazy Loading在不修改 DeepAgents 原有文件系统实现的情况下实现动态工作目录。22. 一句话记录这次技术实践如果以后自己回头看实际上记住下面这句话就够了通过分析 DeepAgents 的BackendFactory扩展机制获取当前运行上下文中的thread_id动态构造会话级root_dir同时使用 Lazy Loading 延迟FilesystemBackend的实例化和目录创建从而实现 Agent 多会话文件隔离并避免在 Backend 初始化阶段执行不必要的同步 I/O。 基于 BackendFactory 实现 Agent 会话级文件系统隔离
1. 背景Agent 的文件为什么需要隔离在使用 DeepAgents 构建 Agent 时可以通过FilesystemBackend为 Agent 提供文件读写、编辑、搜索等能力。最初的实现比较简单直接给 Agent 指定一个固定的工作目录FilesystemBackend( root_dir/agent_files )这种方式在单用户、单会话场景下没有问题。但当系统存在多个聊天窗口甚至多个用户同时使用 Agent 时就会出现一个明显的问题/agent_files ├── report.docx ├── test.xlsx ├── result.md └── ...所有会话都在操作同一个目录。例如用户 A / 会话 1 └── report.docx 用户 B / 会话 2 └── report.docx两个会话都创建report.docx时就可能产生文件覆盖、读取错误等问题。因此文件系统至少应该按照会话进行隔离/agent_files ├── thread_001 │ ├── report.docx │ └── result.md │ ├── thread_002 │ ├── report.docx │ └── result.md │ └── thread_003 └── test.xlsx最终希望达到的效果是一个thread_id对应一个独立的 Agent 工作目录。即FilesystemBackend( root_dir/agent_files/{thread_id} )这样不同会话之间的文件天然隔离。2. 问题拆解整个问题实际上可以拆成两个子问题问题一如何获取当前会话的thread_idAgent 每次运行时都需要知道当前到底是哪一个会话只有拿到thread_id才能构造/agent_files/{thread_id}问题二如何动态创建FilesystemBackend传统写法是backend FilesystemBackend( root_dir/agent_files )这里的root_dir是固定的。但我们需要的是backend FilesystemBackend( root_dirf/agent_files/{thread_id} )也就是说FilesystemBackend的创建必须能够感知当前 Agent 的运行上下文。这两个问题解决之后会话级文件隔离基本就完成了。3. 获取当前会话的 thread_id在 LangGraph / LangChain 的执行过程中thread_id会随着运行上下文传递。实际开发过程中可以从不同的上下文入口获取它。主要考虑两种方式Runtimevar_child_runnable_config3.1 从 Runtime 获取在 LangGraph 中部分 Node、中间件等执行逻辑可以拿到Runtime。例如from langchain.agents.middleware import before_model, AgentState from langgraph.runtime import Runtime before_model() def do( state: AgentState, runtime: Runtime ): # 当前运行逻辑 return None这里的runtime就是当前执行过程中的运行时上下文。因此可以尝试从 Runtime 的配置中获取thread_id例如def get_thread_id_from_runtime(runtime): config getattr(runtime, config, None) if config: return config.get( configurable, {} ).get(thread_id) return None这里需要注意一个实际开发中的问题不同调用位置拿到的 Runtime 结构可能并不完全一样。所以实际项目中不应该过度依赖某一个固定属性而应该做一定的兼容处理。4. 从 var_child_runnable_config 获取另一种方式是使用 LangChain 提供的var_child_runnable_config它本质上是一个用于执行上下文传递的 Context Variable。当前 Runnable 执行链中的配置可以从这里获取。例如from langchain_core.runnables.config import var_child_runnable_config def get_thread_id_from_context(): config var_child_runnable_config.get() if not config: return None return config.get( configurable, {} ).get(thread_id)这里同样可以从configurable中获取thread_id5. 为什么要封装 get_thread_id既然存在多个获取入口那么直接在业务代码里写runtime.config...显然不太合适。因为以后其他模块也可能需要thread_id用户信息当前运行上下文其他 configurable 参数因此可以单独建立一个运行时工具模块content/ └── utils/ └── runtime_util.py将这些与 Runtime 相关的工具统一放进去。例如当前项目中的实现from langchain_core.runnables.config import var_child_runnable_config def get_thread_id(runtimeNone): # 尝试从 runtime 获取 config getattr(runtime, config, None) if config is None: configurable getattr(runtime, context, None) else: configurable config.get(configurable, None) # 如果 runtime 中没有再从上下文变量获取 if configurable is None: current_config var_child_runnable_config.get() if current_config is not None: configurable current_config.get( configurable, {} ) # 最终仍然无法获取 if configurable is None: return default return configurable.get( thread_id, default )这样业务代码就不需要关心thread_id到底是从 Runtime 还是 Context Variable 中拿到的。只需要thread_id get_thread_id(runtime)即可。6. 第二个问题如何动态创建 FilesystemBackend拿到thread_id后下一步自然会想到def create_backend(runtime): thread_id get_thread_id(runtime) root_dir os.path.join( ROOT_PATH_AGENT, thread_id ) os.makedirs( root_dir, exist_okTrue ) return FilesystemBackend( root_dirroot_dir )然后create_deep_agent( backendcreate_backend )看起来已经解决问题了。但这里有一个非常关键的地方create_deep_agent的backend参数并不只能接收一个固定的 Backend 实例。通过查看源码可以看到它支持BackendProtocol | BackendFactory | None这就提供了一个非常重要的扩展点。7. BackendProtocol 与 BackendFactory7.1 BackendProtocolBackendProtocol可以理解为文件后端需要遵守的一套接口规范。也就是说只要一个对象实现了规定的文件操作方法就可以作为 Backend 使用。例如ls_info read write edit grep_raw glob_infoFilesystemBackend本身就是这一套协议的具体实现。7.2 BackendFactory更关键的是BackendFactory。源码定义可以概括为BackendFactory Callable[ [ToolRuntime], BackendProtocol ]换句话说BackendFactory │ │ runtime ▼ 创建 Backend │ ▼ BackendProtocol也就是说backend不一定非要提前创建好backend FilesystemBackend(...)也可以传进去一个runtime - backend形式的可调用对象。这正好解决了我们的问题。8. 第一版方案BackendFactory 动态创建因此可以实现def create_session_backend(runtime): thread_id get_thread_id(runtime) root_dir os.path.join( ROOT_PATH_AGENT, thread_id ) os.makedirs( root_dir, exist_okTrue ) backend FilesystemBackend( root_dirroot_dir ) return backend然后agent create_deep_agent( modelget_llm(), tools[], backendcreate_session_backend, system_promptprompt, )执行流程变成Agent 执行 │ ▼ BackendFactory(runtime) │ ▼ 获取 thread_id │ ▼ /agent_files/{thread_id} │ ▼ 创建 FilesystemBackend │ ▼ Agent 使用该 Backend例如thread_001 ↓ /agent_files/thread_001 ↓ FilesystemBackend(root_dir/agent_files/thread_001)另一个会话thread_002 ↓ /agent_files/thread_002 ↓ FilesystemBackend(root_dir/agent_files/thread_002)这样就实现了会话隔离。9. 实际运行后发现的新问题但是第一版方案在实际运行时又出现了一个问题。问题出现在os.makedirs( root_dir, exist_okTrue )这里。BackendFactory的创建过程可能处在异步执行环境中。而os.makedirs()属于同步文件系统 I/O。如果在不合适的异步执行上下文中直接执行同步 I/O就可能造成事件循环阻塞。这时候问题就从“怎么动态创建 Backend”变成了“怎么动态创建 Backend同时避免在 BackendFactory 阶段执行不必要的同步 I/O”进一步分析后可以发现这里其实还存在一个设计上的浪费为什么一定要在创建 Backend 的时候就创建目录假设用户打开了一个新的聊天窗口用户打开 thread_001然后只是你好 今天天气怎么样 介绍一下你自己整个过程中根本没有进行文件操作。但我们的代码已经提前执行os.makedirs(/agent_files/thread_001)实际上没有必要。因此这里可以进一步优化。10. 第二版方案LazyFilesystemBackend解决方法就是Backend 可以先创建但真正的 FilesystemBackend 和工作目录延迟到第一次文件操作时再创建。这就是 Lazy Loading延迟加载的思路。整体执行过程变成创建 Backend │ ├── 获取 thread_id ├── 计算 root_dir └── 暂时不创建目录 │ ▼ Agent 是否操作文件 │ ┌───┴───┐ │ │ 否 是 │ │ ▼ ▼ 什么都不做 创建目录 │ ▼ FilesystemBackend │ ▼ 执行文件操作这样既解决了同步 I/O 时机的问题也避免了无意义的目录创建。11. LazyFilesystemBackend 的设计我们可以自己实现一个LazyFilesystemBackend它本身实现BackendProtocol但它并不直接负责真正的文件操作。它内部维护一个真正的FilesystemBackend结构可以理解为LazyFilesystemBackend │ ├── runtime ├── thread_id ├── root_dir └── _backend │ └── FilesystemBackend初始化的时候_backend None第一次执行read() write() edit() ls_info() ...时再真正创建FilesystemBackend12. _ensure_backend整个设计的核心核心代码其实非常简单def _ensure_backend(self): if self._backend is None: os.makedirs( self._root_dir, exist_okTrue ) self._backend FilesystemBackend( root_dirself._root_dir, virtual_modeTrue ) return self._backend这个方法负责保证只要真正需要文件操作就一定存在一个可用的 FilesystemBackend。第一次调用_backend None ↓ 创建目录 ↓ 创建 FilesystemBackend ↓ 保存到 _backend第二次调用_backend ! None ↓ 直接返回已有实例因此它实际上是延迟初始化 实例缓存。13. 文件操作如何转发有了_ensure_backend()后其他方法就非常简单。例如def read( self, file_path: str, offset: int 0, limit: int 2000 ): return self._ensure_backend().read( file_path, offset, limit )write()def write( self, file_path: str, content: str ): return self._ensure_backend().write( file_path, content )edit()def edit( self, file_path: str, old_string: str, new_string: str, replace_all: bool False ): return self._ensure_backend().edit( file_path, old_string, new_string, replace_all )其他方法同理。最终形成Agent │ ▼ LazyFilesystemBackend │ │ _ensure_backend() ▼ FilesystemBackend │ ▼ 真实文件系统这里的LazyFilesystemBackend实际上承担了一层代理/适配作用。Agent 并不知道后面是否已经初始化了真正的文件系统后端。14. 最终实现项目中最终实现的核心代码如下from deepagents.backends import ( FilesystemBackend, BackendProtocol ) from base.configs import ROOT_PATH_AGENT from content.utils import runtime_util as rt import os from typing import Optional class LazyFilesystemBackend(BackendProtocol): def __init__(self, runtime): self.runtime runtime # 真正的 FilesystemBackend self._backend: Optional[ FilesystemBackend ] None # 获取当前会话 self._thread_id rt.get_thread_id( runtime ) # 构造当前会话的工作目录 self._root_dir os.path.join( ROOT_PATH_AGENT, self._thread_id ) def _ensure_backend(self): if self._backend is None: # 第一次执行文件操作时才创建目录 os.makedirs( self._root_dir, exist_okTrue ) # 创建真正的文件系统后端 self._backend FilesystemBackend( root_dirself._root_dir, virtual_modeTrue ) return self._backend def ls_info(self, path: str): return self._ensure_backend().ls_info(path) def read( self, file_path: str, offset: int 0, limit: int 2000 ): return self._ensure_backend().read( file_path, offset, limit ) def write( self, file_path: str, content: str ): return self._ensure_backend().write( file_path, content ) def edit( self, file_path: str, old_string: str, new_string: str, replace_all: bool False ): return self._ensure_backend().edit( file_path, old_string, new_string, replace_all ) def grep_raw( self, pattern: str, path: Optional[str] None, glob: Optional[str] None ): return self._ensure_backend().grep_raw( pattern, path, glob ) def glob_info( self, pattern: str, path: str / ): return self._ensure_backend().glob_info( pattern, path ) def create_session_backend(runtime): return LazyFilesystemBackend(runtime)15. 接入 Agent原本 Agent 可能是self.agent create_deep_agent( modelget_llm(), tools[], backendFilesystemBackend( root_dirROOT_PATH_AGENT ), system_promptprompt, )现在修改为from deepagents import create_deep_agent from conn.llms import get_small_llm as get_llm from content.others import mybackend class AllAgent: def __init__(self): prompt 你是一个通用智能体 回答用户用中文。 self.agent create_deep_agent( modelget_llm(), tools[], backendmybackend.create_session_backend, system_promptprompt, )这里最关键的一行就是backendmybackend.create_session_backend注意这里传入的不是create_session_backend()而是create_session_backend因为这里需要把工厂函数本身交给框架。之后由框架在 Agent 执行过程中根据当前runtime调用它。16. 最终执行流程完整链路可以概括为用户打开聊天窗口 │ ▼ 生成 thread_id │ ▼ Agent 开始执行 │ ▼ BackendFactory(runtime) │ ▼ create_session_backend(runtime) │ ▼ LazyFilesystemBackend(runtime) │ ├── 获取 thread_id │ └── 计算 root_dir │ ▼ Agent 是否执行文件操作 │ ┌─────┴─────┐ │ │ 否 是 │ │ ▼ ▼ 不创建 _ensure_backend() 文件目录 │ ▼ 创建 thread 目录 │ ▼ FilesystemBackend │ ▼ 执行文件操作例如thread_id abc123最终得到/agent_files/abc123/另一个会话thread_id xyz456得到/agent_files/xyz456/两个 Agent 会话之间的文件完全分离。17. 项目目录结构最终相关代码结构content/ ├── others/ │ └── mybackend.py │ ├── utils/ │ └── runtime_util.py │ └── all_agent.py职责也比较清晰runtime_util.py ↓ 负责运行时信息获取 ↓ thread_id mybackend.py ↓ 负责文件系统后端 ↓ LazyFilesystemBackend ↓ create_session_backend all_agent.py ↓ Agent 创建 ↓ 注入 BackendFactory18. 这次改造真正解决了什么18.1 会话隔离从所有会话 ↓ /agent_files变成thread_001 ↓ /agent_files/thread_001 thread_002 ↓ /agent_files/thread_002不同会话拥有独立工作空间。18.2 解决同名文件冲突例如两个会话都生成report.docx现在实际对应/agent_files/thread_001/report.docx /agent_files/thread_002/report.docx不会互相覆盖。18.3 降低跨会话文件访问风险Agent 的文件操作始终发生在当前thread_id对应的工作目录下。因此从文件系统层面建立了基本的会话边界。需要注意的是这属于应用层面的文件隔离设计并不等同于完整的多租户安全体系。如果真正用于生产环境还需要进一步考虑权限控制、路径穿越、用户与 thread 的绑定关系、目录生命周期以及数据清理等问题。19. 为什么最后选择 Lazy Loading这个改造实际上经历了两个版本。第一版BackendFactory ↓ 获取 thread_id ↓ 创建目录 ↓ 创建 FilesystemBackend ↓ 返回优点是简单直接。但是存在两个问题第一Backend 创建阶段就执行了同步 I/O。这可能与异步执行环境产生冲突或造成事件循环阻塞。第二没有必要提前创建目录。用户可能只是聊天并没有执行任何文件操作。第二版改成BackendFactory ↓ 获取 thread_id ↓ 计算 root_dir ↓ 暂不创建目录 ↓ 等待真正的文件操作 ↓ _ensure_backend() ↓ 创建目录 FilesystemBackend这样Backend 初始化更加轻量文件目录按需创建避免初始化阶段执行不必要的 I/O文件操作逻辑集中在_ensure_backend()原有FilesystemBackend的能力仍然可以复用因此最终选择了第二种方案。20. 这次问题中比较值得记录的源码分析思路这次改造真正有价值的地方其实并不只是写了一个LazyFilesystemBackend而是如何从框架的约束中找到扩展点。最开始遇到的问题是FilesystemBackend的root_dir是固定的怎么动态设置如果直接从FilesystemBackend本身入手很容易陷入怎么修改 FilesystemBackend 怎么重新实现 FilesystemBackend但继续往上看create_deep_agent( backend... )发现BackendProtocol | BackendFactory再继续看BackendFactory Callable[ [ToolRuntime], BackendProtocol ]于是整个思路就发生了变化不是修改 FilesystemBackend ↓ 而是利用 BackendFactory ↓ 让 Backend 根据 runtime 动态生成 ↓ 再通过 Lazy Backend 延迟真正的文件系统初始化这也是使用成熟框架时比较重要的一种思路遇到框架无法直接满足的需求时先寻找框架提供的扩展点而不是马上绕开框架重写整个功能。21. 最终方案总结整个方案可以浓缩成四层① Runtime ↓ 获取 thread_id ② BackendFactory ↓ 根据 thread_id 创建会话级 Backend ③ LazyFilesystemBackend ↓ 延迟真正的文件系统初始化 ④ FilesystemBackend ↓ 实际执行 read / write / edit / grep / glob 等操作核心代码关系create_deep_agent │ │ backend ▼ create_session_backend │ │ runtime ▼ LazyFilesystemBackend │ │ 第一次文件操作 ▼ FilesystemBackend │ │ root_dir ▼ /agent_files/{thread_id}最终实现了基于thread_id的 Agent 会话级文件系统隔离并通过BackendFactory Lazy Loading在不修改 DeepAgents 原有文件系统实现的情况下实现动态工作目录。22. 一句话记录这次技术实践如果以后自己回头看实际上记住下面这句话就够了通过分析 DeepAgents 的BackendFactory扩展机制获取当前运行上下文中的thread_id动态构造会话级root_dir同时使用 Lazy Loading 延迟FilesystemBackend的实例化和目录创建从而实现 Agent 多会话文件隔离并避免在 Backend 初始化阶段执行不必要的同步 I/O。