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。