企业官网建设流程全解析

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_id？

Agent 每次运行时都需要知道：

当前到底是哪一个会话？

只有拿到thread_id，才能构造：

/agent_files/{thread_id}

问题二：如何动态创建FilesystemBackend？

传统写法是：

backend = FilesystemBackend( root_dir="/agent_files" )

这里的root_dir是固定的。

但我们需要的是：

backend = FilesystemBackend( root_dir=f"/agent_files/{thread_id}" )

也就是说：

FilesystemBackend的创建必须能够感知当前 Agent 的运行上下文。

这两个问题解决之后，会话级文件隔离基本就完成了。

3. 获取当前会话的 thread_id

在 LangGraph / LangChain 的执行过程中，thread_id会随着运行上下文传递。

实际开发过程中，可以从不同的上下文入口获取它。

主要考虑两种方式：

1. Runtime

2. var_child_runnable_config

3.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_id

5. 为什么要封装 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(runtime=None): # 尝试从 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_ok=True ) return FilesystemBackend( root_dir=root_dir )

然后：

create_deep_agent( backend=create_backend )

看起来已经解决问题了。

但这里有一个非常关键的地方：

create_deep_agent的backend参数并不只能接收一个固定的 Backend 实例。

通过查看源码，可以看到它支持：

BackendProtocol | BackendFactory | None

这就提供了一个非常重要的扩展点。

7. BackendProtocol 与 BackendFactory

7.1 BackendProtocol

BackendProtocol可以理解为文件后端需要遵守的一套接口规范。

也就是说，只要一个对象实现了规定的文件操作方法，就可以作为 Backend 使用。

例如：

ls_info read write edit grep_raw glob_info

FilesystemBackend本身就是这一套协议的具体实现。

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_ok=True ) backend = FilesystemBackend( root_dir=root_dir ) return backend

然后：

agent = create_deep_agent( model=get_llm(), tools=[], backend=create_session_backend, system_prompt=prompt, )

执行流程变成：

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_ok=True )

这里。

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() ...

时，再真正创建：

FilesystemBackend

12. _ensure_backend：整个设计的核心

核心代码其实非常简单：

def _ensure_backend(self): if self._backend is None: os.makedirs( self._root_dir, exist_ok=True ) self._backend = FilesystemBackend( root_dir=self._root_dir, virtual_mode=True ) 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_ok=True ) # 创建真正的文件系统后端 self._backend = FilesystemBackend( root_dir=self._root_dir, virtual_mode=True ) 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( model=get_llm(), tools=[], backend=FilesystemBackend( root_dir=ROOT_PATH_AGENT ), system_prompt=prompt, )

现在修改为：

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( model=get_llm(), tools=[], backend=mybackend.create_session_backend, system_prompt=prompt, )

这里最关键的一行就是：

backend=mybackend.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 创建 ↓ 注入 BackendFactory

18. 这次改造真正解决了什么？

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。