現(xiàn) Agent 會話級文件系統(tǒng)隔離)
1. 背景Agent 的文件為什么需要隔離在使用 DeepAgents 構(gòu)建 Agent 時可以通過FilesystemBackend為 Agent 提供文件讀寫、編輯、搜索等能力。最初的實(shí)現(xiàn)比較簡單直接給 Agent 指定一個固定的工作目錄FilesystemBackend( root_dir/agent_files )這種方式在單用戶、單會話場景下沒有問題。但當(dāng)系統(tǒng)存在多個聊天窗口甚至多個用戶同時使用 Agent 時就會出現(xiàn)一個明顯的問題/agent_files ├── report.docx ├── test.xlsx ├── result.md └── ...所有會話都在操作同一個目錄。例如用戶 A / 會話 1 └── report.docx 用戶 B / 會話 2 └── report.docx兩個會話都創(chuàng)建report.docx時就可能產(chǎn)生文件覆蓋、讀取錯誤等問題。因此文件系統(tǒng)至少應(yīng)該按照會話進(jìn)行隔離/agent_files ├── thread_001 │ ├── report.docx │ └── result.md │ ├── thread_002 │ ├── report.docx │ └── result.md │ └── thread_003 └── test.xlsx最終希望達(dá)到的效果是一個thread_id對應(yīng)一個獨(dú)立的 Agent 工作目錄。即FilesystemBackend( root_dir/agent_files/{thread_id} )這樣不同會話之間的文件天然隔離。2. 問題拆解整個問題實(shí)際上可以拆成兩個子問題問題一如何獲取當(dāng)前會話的thread_idAgent 每次運(yùn)行時都需要知道當(dāng)前到底是哪一個會話只有拿到thread_id才能構(gòu)造/agent_files/{thread_id}問題二如何動態(tài)創(chuàng)建FilesystemBackend傳統(tǒng)寫法是backend FilesystemBackend( root_dir/agent_files )這里的root_dir是固定的。但我們需要的是backend FilesystemBackend( root_dirf/agent_files/{thread_id} )也就是說FilesystemBackend的創(chuàng)建必須能夠感知當(dāng)前 Agent 的運(yùn)行上下文。這兩個問題解決之后會話級文件隔離基本就完成了。3. 獲取當(dāng)前會話的 thread_id在 LangGraph / LangChain 的執(zhí)行過程中thread_id會隨著運(yùn)行上下文傳遞。實(shí)際開發(fā)過程中可以從不同的上下文入口獲取它。主要考慮兩種方式Runtimevar_child_runnable_config3.1 從 Runtime 獲取在 LangGraph 中部分 Node、中間件等執(zhí)行邏輯可以拿到Runtime。例如from langchain.agents.middleware import before_model, AgentState from langgraph.runtime import Runtime before_model() def do( state: AgentState, runtime: Runtime ): # 當(dāng)前運(yùn)行邏輯 return None這里的runtime就是當(dāng)前執(zhí)行過程中的運(yùn)行時上下文。因此可以嘗試從 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這里需要注意一個實(shí)際開發(fā)中的問題不同調(diào)用位置拿到的 Runtime 結(jié)構(gòu)可能并不完全一樣。所以實(shí)際項(xiàng)目中不應(yīng)該過度依賴某一個固定屬性而應(yīng)該做一定的兼容處理。4. 從 var_child_runnable_config 獲取另一種方式是使用 LangChain 提供的var_child_runnable_config它本質(zhì)上是一個用于執(zhí)行上下文傳遞的 Context Variable。當(dāng)前 Runnable 執(zhí)行鏈中的配置可以從這里獲取。例如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既然存在多個獲取入口那么直接在業(yè)務(wù)代碼里寫runtime.config...顯然不太合適。因?yàn)橐院笃渌K也可能需要thread_id用戶信息當(dāng)前運(yùn)行上下文其他 configurable 參數(shù)因此可以單獨(dú)建立一個運(yùn)行時工具模塊content/ └── utils/ └── runtime_util.py將這些與 Runtime 相關(guān)的工具統(tǒng)一放進(jìn)去。例如當(dāng)前項(xiàng)目中的實(shí)現(xiàn)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 )這樣業(yè)務(wù)代碼就不需要關(guān)心thread_id到底是從 Runtime 還是 Context Variable 中拿到的。只需要thread_id get_thread_id(runtime)即可。6. 第二個問題如何動態(tài)創(chuàng)建 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 )看起來已經(jīng)解決問題了。但這里有一個非常關(guān)鍵的地方create_deep_agent的backend參數(shù)并不只能接收一個固定的 Backend 實(shí)例。通過查看源碼可以看到它支持BackendProtocol | BackendFactory | None這就提供了一個非常重要的擴(kuò)展點(diǎn)。7. BackendProtocol 與 BackendFactory7.1 BackendProtocolBackendProtocol可以理解為文件后端需要遵守的一套接口規(guī)范。也就是說只要一個對象實(shí)現(xiàn)了規(guī)定的文件操作方法就可以作為 Backend 使用。例如ls_info read write edit grep_raw glob_infoFilesystemBackend本身就是這一套協(xié)議的具體實(shí)現(xiàn)。7.2 BackendFactory更關(guān)鍵的是BackendFactory。源碼定義可以概括為BackendFactory Callable[ [ToolRuntime], BackendProtocol ]換句話說BackendFactory │ │ runtime ▼ 創(chuàng)建 Backend │ ▼ BackendProtocol也就是說backend不一定非要提前創(chuàng)建好backend FilesystemBackend(...)也可以傳進(jìn)去一個runtime - backend形式的可調(diào)用對象。這正好解決了我們的問題。8. 第一版方案BackendFactory 動態(tài)創(chuàng)建因此可以實(shí)現(xiàn)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, )執(zhí)行流程變成Agent 執(zhí)行 │ ▼ BackendFactory(runtime) │ ▼ 獲取 thread_id │ ▼ /agent_files/{thread_id} │ ▼ 創(chuàng)建 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)這樣就實(shí)現(xiàn)了會話隔離。9. 實(shí)際運(yùn)行后發(fā)現(xiàn)的新問題但是第一版方案在實(shí)際運(yùn)行時又出現(xiàn)了一個問題。問題出現(xiàn)在os.makedirs( root_dir, exist_okTrue )這里。BackendFactory的創(chuàng)建過程可能處在異步執(zhí)行環(huán)境中。而os.makedirs()屬于同步文件系統(tǒng) I/O。如果在不合適的異步執(zhí)行上下文中直接執(zhí)行同步 I/O就可能造成事件循環(huán)阻塞。這時候問題就從“怎么動態(tài)創(chuàng)建 Backend”變成了“怎么動態(tài)創(chuàng)建 Backend同時避免在 BackendFactory 階段執(zhí)行不必要的同步 I/O”進(jìn)一步分析后可以發(fā)現(xiàn)這里其實(shí)還存在一個設(shè)計(jì)上的浪費(fèi)為什么一定要在創(chuàng)建 Backend 的時候就創(chuàng)建目錄假設(shè)用戶打開了一個新的聊天窗口用戶打開 thread_001然后只是你好 今天天氣怎么樣 介紹一下你自己整個過程中根本沒有進(jìn)行文件操作。但我們的代碼已經(jīng)提前執(zhí)行os.makedirs(/agent_files/thread_001)實(shí)際上沒有必要。因此這里可以進(jìn)一步優(yōu)化。10. 第二版方案LazyFilesystemBackend解決方法就是Backend 可以先創(chuàng)建但真正的 FilesystemBackend 和工作目錄延遲到第一次文件操作時再創(chuàng)建。這就是 Lazy Loading延遲加載的思路。整體執(zhí)行過程變成創(chuàng)建 Backend │ ├── 獲取 thread_id ├── 計(jì)算 root_dir └── 暫時不創(chuàng)建目錄 │ ▼ Agent 是否操作文件 │ ┌───┴───┐ │ │ 否 是 │ │ ▼ ▼ 什么都不做 創(chuàng)建目錄 │ ▼ FilesystemBackend │ ▼ 執(zhí)行文件操作這樣既解決了同步 I/O 時機(jī)的問題也避免了無意義的目錄創(chuàng)建。11. LazyFilesystemBackend 的設(shè)計(jì)我們可以自己實(shí)現(xiàn)一個LazyFilesystemBackend它本身實(shí)現(xiàn)BackendProtocol但它并不直接負(fù)責(zé)真正的文件操作。它內(nèi)部維護(hù)一個真正的FilesystemBackend結(jié)構(gòu)可以理解為LazyFilesystemBackend │ ├── runtime ├── thread_id ├── root_dir └── _backend │ └── FilesystemBackend初始化的時候_backend None第一次執(zhí)行read() write() edit() ls_info() ...時再真正創(chuàng)建FilesystemBackend12. _ensure_backend整個設(shè)計(jì)的核心核心代碼其實(shí)非常簡單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這個方法負(fù)責(zé)保證只要真正需要文件操作就一定存在一個可用的 FilesystemBackend。第一次調(diào)用_backend None ↓ 創(chuàng)建目錄 ↓ 創(chuàng)建 FilesystemBackend ↓ 保存到 _backend第二次調(diào)用_backend ! None ↓ 直接返回已有實(shí)例因此它實(shí)際上是延遲初始化 實(shí)例緩存。13. 文件操作如何轉(zhuǎn)發(fā)有了_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 │ ▼ 真實(shí)文件系統(tǒng)這里的LazyFilesystemBackend實(shí)際上承擔(dān)了一層代理/適配作用。Agent 并不知道后面是否已經(jīng)初始化了真正的文件系統(tǒng)后端。14. 最終實(shí)現(xiàn)項(xiàng)目中最終實(shí)現(xiàn)的核心代碼如下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 # 獲取當(dāng)前會話 self._thread_id rt.get_thread_id( runtime ) # 構(gòu)造當(dāng)前會話的工作目錄 self._root_dir os.path.join( ROOT_PATH_AGENT, self._thread_id ) def _ensure_backend(self): if self._backend is None: # 第一次執(zhí)行文件操作時才創(chuàng)建目錄 os.makedirs( self._root_dir, exist_okTrue ) # 創(chuàng)建真正的文件系統(tǒng)后端 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, )現(xiàn)在修改為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, )這里最關(guān)鍵的一行就是backendmybackend.create_session_backend注意這里傳入的不是create_session_backend()而是create_session_backend因?yàn)檫@里需要把工廠函數(shù)本身交給框架。之后由框架在 Agent 執(zhí)行過程中根據(jù)當(dāng)前runtime調(diào)用它。16. 最終執(zhí)行流程完整鏈路可以概括為用戶打開聊天窗口 │ ▼ 生成 thread_id │ ▼ Agent 開始執(zhí)行 │ ▼ BackendFactory(runtime) │ ▼ create_session_backend(runtime) │ ▼ LazyFilesystemBackend(runtime) │ ├── 獲取 thread_id │ └── 計(jì)算 root_dir │ ▼ Agent 是否執(zhí)行文件操作 │ ┌─────┴─────┐ │ │ 否 是 │ │ ▼ ▼ 不創(chuàng)建 _ensure_backend() 文件目錄 │ ▼ 創(chuàng)建 thread 目錄 │ ▼ FilesystemBackend │ ▼ 執(zhí)行文件操作例如thread_id abc123最終得到/agent_files/abc123/另一個會話thread_id xyz456得到/agent_files/xyz456/兩個 Agent 會話之間的文件完全分離。17. 項(xiàng)目目錄結(jié)構(gòu)最終相關(guān)代碼結(jié)構(gòu)content/ ├── others/ │ └── mybackend.py │ ├── utils/ │ └── runtime_util.py │ └── all_agent.py職責(zé)也比較清晰runtime_util.py ↓ 負(fù)責(zé)運(yùn)行時信息獲取 ↓ thread_id mybackend.py ↓ 負(fù)責(zé)文件系統(tǒng)后端 ↓ LazyFilesystemBackend ↓ create_session_backend all_agent.py ↓ Agent 創(chuàng)建 ↓ 注入 BackendFactory18. 這次改造真正解決了什么18.1 會話隔離從所有會話 ↓ /agent_files變成thread_001 ↓ /agent_files/thread_001 thread_002 ↓ /agent_files/thread_002不同會話擁有獨(dú)立工作空間。18.2 解決同名文件沖突例如兩個會話都生成report.docx現(xiàn)在實(shí)際對應(yīng)/agent_files/thread_001/report.docx /agent_files/thread_002/report.docx不會互相覆蓋。18.3 降低跨會話文件訪問風(fēng)險(xiǎn)Agent 的文件操作始終發(fā)生在當(dāng)前thread_id對應(yīng)的工作目錄下。因此從文件系統(tǒng)層面建立了基本的會話邊界。需要注意的是這屬于應(yīng)用層面的文件隔離設(shè)計(jì)并不等同于完整的多租戶安全體系。如果真正用于生產(chǎn)環(huán)境還需要進(jìn)一步考慮權(quán)限控制、路徑穿越、用戶與 thread 的綁定關(guān)系、目錄生命周期以及數(shù)據(jù)清理等問題。19. 為什么最后選擇 Lazy Loading這個改造實(shí)際上經(jīng)歷了兩個版本。第一版BackendFactory ↓ 獲取 thread_id ↓ 創(chuàng)建目錄 ↓ 創(chuàng)建 FilesystemBackend ↓ 返回優(yōu)點(diǎn)是簡單直接。但是存在兩個問題第一Backend 創(chuàng)建階段就執(zhí)行了同步 I/O。這可能與異步執(zhí)行環(huán)境產(chǎn)生沖突或造成事件循環(huán)阻塞。第二沒有必要提前創(chuàng)建目錄。用戶可能只是聊天并沒有執(zhí)行任何文件操作。第二版改成BackendFactory ↓ 獲取 thread_id ↓ 計(jì)算 root_dir ↓ 暫不創(chuàng)建目錄 ↓ 等待真正的文件操作 ↓ _ensure_backend() ↓ 創(chuàng)建目錄 FilesystemBackend這樣Backend 初始化更加輕量文件目錄按需創(chuàng)建避免初始化階段執(zhí)行不必要的 I/O文件操作邏輯集中在_ensure_backend()原有FilesystemBackend的能力仍然可以復(fù)用因此最終選擇了第二種方案。20. 這次問題中比較值得記錄的源碼分析思路這次改造真正有價(jià)值的地方其實(shí)并不只是寫了一個LazyFilesystemBackend而是如何從框架的約束中找到擴(kuò)展點(diǎn)。最開始遇到的問題是FilesystemBackend的root_dir是固定的怎么動態(tài)設(shè)置如果直接從FilesystemBackend本身入手很容易陷入怎么修改 FilesystemBackend 怎么重新實(shí)現(xiàn) FilesystemBackend但繼續(xù)往上看create_deep_agent( backend... )發(fā)現(xiàn)BackendProtocol | BackendFactory再繼續(xù)看BackendFactory Callable[ [ToolRuntime], BackendProtocol ]于是整個思路就發(fā)生了變化不是修改 FilesystemBackend ↓ 而是利用 BackendFactory ↓ 讓 Backend 根據(jù) runtime 動態(tài)生成 ↓ 再通過 Lazy Backend 延遲真正的文件系統(tǒng)初始化這也是使用成熟框架時比較重要的一種思路遇到框架無法直接滿足的需求時先尋找框架提供的擴(kuò)展點(diǎn)而不是馬上繞開框架重寫整個功能。21. 最終方案總結(jié)整個方案可以濃縮成四層① Runtime ↓ 獲取 thread_id ② BackendFactory ↓ 根據(jù) thread_id 創(chuàng)建會話級 Backend ③ LazyFilesystemBackend ↓ 延遲真正的文件系統(tǒng)初始化 ④ FilesystemBackend ↓ 實(shí)際執(zhí)行 read / write / edit / grep / glob 等操作核心代碼關(guān)系create_deep_agent │ │ backend ▼ create_session_backend │ │ runtime ▼ LazyFilesystemBackend │ │ 第一次文件操作 ▼ FilesystemBackend │ │ root_dir ▼ /agent_files/{thread_id}最終實(shí)現(xiàn)了基于thread_id的 Agent 會話級文件系統(tǒng)隔離并通過BackendFactory Lazy Loading在不修改 DeepAgents 原有文件系統(tǒng)實(shí)現(xiàn)的情況下實(shí)現(xiàn)動態(tài)工作目錄。22. 一句話記錄這次技術(shù)實(shí)踐如果以后自己回頭看實(shí)際上記住下面這句話就夠了通過分析 DeepAgents 的BackendFactory擴(kuò)展機(jī)制獲取當(dāng)前運(yùn)行上下文中的thread_id動態(tài)構(gòu)造會話級root_dir同時使用 Lazy Loading 延遲FilesystemBackend的實(shí)例化和目錄創(chuàng)建從而實(shí)現(xiàn) Agent 多會話文件隔離并避免在 Backend 初始化階段執(zhí)行不必要的同步 I/O。 基于 BackendFactory 實(shí)現(xiàn) Agent 會話級文件系統(tǒng)隔離
1. 背景Agent 的文件為什么需要隔離在使用 DeepAgents 構(gòu)建 Agent 時可以通過FilesystemBackend為 Agent 提供文件讀寫、編輯、搜索等能力。最初的實(shí)現(xiàn)比較簡單直接給 Agent 指定一個固定的工作目錄FilesystemBackend( root_dir/agent_files )這種方式在單用戶、單會話場景下沒有問題。但當(dāng)系統(tǒng)存在多個聊天窗口甚至多個用戶同時使用 Agent 時就會出現(xiàn)一個明顯的問題/agent_files ├── report.docx ├── test.xlsx ├── result.md └── ...所有會話都在操作同一個目錄。例如用戶 A / 會話 1 └── report.docx 用戶 B / 會話 2 └── report.docx兩個會話都創(chuàng)建report.docx時就可能產(chǎn)生文件覆蓋、讀取錯誤等問題。因此文件系統(tǒng)至少應(yīng)該按照會話進(jìn)行隔離/agent_files ├── thread_001 │ ├── report.docx │ └── result.md │ ├── thread_002 │ ├── report.docx │ └── result.md │ └── thread_003 └── test.xlsx最終希望達(dá)到的效果是一個thread_id對應(yīng)一個獨(dú)立的 Agent 工作目錄。即FilesystemBackend( root_dir/agent_files/{thread_id} )這樣不同會話之間的文件天然隔離。2. 問題拆解整個問題實(shí)際上可以拆成兩個子問題問題一如何獲取當(dāng)前會話的thread_idAgent 每次運(yùn)行時都需要知道當(dāng)前到底是哪一個會話只有拿到thread_id才能構(gòu)造/agent_files/{thread_id}問題二如何動態(tài)創(chuàng)建FilesystemBackend傳統(tǒng)寫法是backend FilesystemBackend( root_dir/agent_files )這里的root_dir是固定的。但我們需要的是backend FilesystemBackend( root_dirf/agent_files/{thread_id} )也就是說FilesystemBackend的創(chuàng)建必須能夠感知當(dāng)前 Agent 的運(yùn)行上下文。這兩個問題解決之后會話級文件隔離基本就完成了。3. 獲取當(dāng)前會話的 thread_id在 LangGraph / LangChain 的執(zhí)行過程中thread_id會隨著運(yùn)行上下文傳遞。實(shí)際開發(fā)過程中可以從不同的上下文入口獲取它。主要考慮兩種方式Runtimevar_child_runnable_config3.1 從 Runtime 獲取在 LangGraph 中部分 Node、中間件等執(zhí)行邏輯可以拿到Runtime。例如from langchain.agents.middleware import before_model, AgentState from langgraph.runtime import Runtime before_model() def do( state: AgentState, runtime: Runtime ): # 當(dāng)前運(yùn)行邏輯 return None這里的runtime就是當(dāng)前執(zhí)行過程中的運(yùn)行時上下文。因此可以嘗試從 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這里需要注意一個實(shí)際開發(fā)中的問題不同調(diào)用位置拿到的 Runtime 結(jié)構(gòu)可能并不完全一樣。所以實(shí)際項(xiàng)目中不應(yīng)該過度依賴某一個固定屬性而應(yīng)該做一定的兼容處理。4. 從 var_child_runnable_config 獲取另一種方式是使用 LangChain 提供的var_child_runnable_config它本質(zhì)上是一個用于執(zhí)行上下文傳遞的 Context Variable。當(dāng)前 Runnable 執(zhí)行鏈中的配置可以從這里獲取。例如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既然存在多個獲取入口那么直接在業(yè)務(wù)代碼里寫runtime.config...顯然不太合適。因?yàn)橐院笃渌K也可能需要thread_id用戶信息當(dāng)前運(yùn)行上下文其他 configurable 參數(shù)因此可以單獨(dú)建立一個運(yùn)行時工具模塊content/ └── utils/ └── runtime_util.py將這些與 Runtime 相關(guān)的工具統(tǒng)一放進(jìn)去。例如當(dāng)前項(xiàng)目中的實(shí)現(xiàn)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 )這樣業(yè)務(wù)代碼就不需要關(guān)心thread_id到底是從 Runtime 還是 Context Variable 中拿到的。只需要thread_id get_thread_id(runtime)即可。6. 第二個問題如何動態(tài)創(chuàng)建 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 )看起來已經(jīng)解決問題了。但這里有一個非常關(guān)鍵的地方create_deep_agent的backend參數(shù)并不只能接收一個固定的 Backend 實(shí)例。通過查看源碼可以看到它支持BackendProtocol | BackendFactory | None這就提供了一個非常重要的擴(kuò)展點(diǎn)。7. BackendProtocol 與 BackendFactory7.1 BackendProtocolBackendProtocol可以理解為文件后端需要遵守的一套接口規(guī)范。也就是說只要一個對象實(shí)現(xiàn)了規(guī)定的文件操作方法就可以作為 Backend 使用。例如ls_info read write edit grep_raw glob_infoFilesystemBackend本身就是這一套協(xié)議的具體實(shí)現(xiàn)。7.2 BackendFactory更關(guān)鍵的是BackendFactory。源碼定義可以概括為BackendFactory Callable[ [ToolRuntime], BackendProtocol ]換句話說BackendFactory │ │ runtime ▼ 創(chuàng)建 Backend │ ▼ BackendProtocol也就是說backend不一定非要提前創(chuàng)建好backend FilesystemBackend(...)也可以傳進(jìn)去一個runtime - backend形式的可調(diào)用對象。這正好解決了我們的問題。8. 第一版方案BackendFactory 動態(tài)創(chuàng)建因此可以實(shí)現(xiàn)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, )執(zhí)行流程變成Agent 執(zhí)行 │ ▼ BackendFactory(runtime) │ ▼ 獲取 thread_id │ ▼ /agent_files/{thread_id} │ ▼ 創(chuàng)建 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)這樣就實(shí)現(xiàn)了會話隔離。9. 實(shí)際運(yùn)行后發(fā)現(xiàn)的新問題但是第一版方案在實(shí)際運(yùn)行時又出現(xiàn)了一個問題。問題出現(xiàn)在os.makedirs( root_dir, exist_okTrue )這里。BackendFactory的創(chuàng)建過程可能處在異步執(zhí)行環(huán)境中。而os.makedirs()屬于同步文件系統(tǒng) I/O。如果在不合適的異步執(zhí)行上下文中直接執(zhí)行同步 I/O就可能造成事件循環(huán)阻塞。這時候問題就從“怎么動態(tài)創(chuàng)建 Backend”變成了“怎么動態(tài)創(chuàng)建 Backend同時避免在 BackendFactory 階段執(zhí)行不必要的同步 I/O”進(jìn)一步分析后可以發(fā)現(xiàn)這里其實(shí)還存在一個設(shè)計(jì)上的浪費(fèi)為什么一定要在創(chuàng)建 Backend 的時候就創(chuàng)建目錄假設(shè)用戶打開了一個新的聊天窗口用戶打開 thread_001然后只是你好 今天天氣怎么樣 介紹一下你自己整個過程中根本沒有進(jìn)行文件操作。但我們的代碼已經(jīng)提前執(zhí)行os.makedirs(/agent_files/thread_001)實(shí)際上沒有必要。因此這里可以進(jìn)一步優(yōu)化。10. 第二版方案LazyFilesystemBackend解決方法就是Backend 可以先創(chuàng)建但真正的 FilesystemBackend 和工作目錄延遲到第一次文件操作時再創(chuàng)建。這就是 Lazy Loading延遲加載的思路。整體執(zhí)行過程變成創(chuàng)建 Backend │ ├── 獲取 thread_id ├── 計(jì)算 root_dir └── 暫時不創(chuàng)建目錄 │ ▼ Agent 是否操作文件 │ ┌───┴───┐ │ │ 否 是 │ │ ▼ ▼ 什么都不做 創(chuàng)建目錄 │ ▼ FilesystemBackend │ ▼ 執(zhí)行文件操作這樣既解決了同步 I/O 時機(jī)的問題也避免了無意義的目錄創(chuàng)建。11. LazyFilesystemBackend 的設(shè)計(jì)我們可以自己實(shí)現(xiàn)一個LazyFilesystemBackend它本身實(shí)現(xiàn)BackendProtocol但它并不直接負(fù)責(zé)真正的文件操作。它內(nèi)部維護(hù)一個真正的FilesystemBackend結(jié)構(gòu)可以理解為LazyFilesystemBackend │ ├── runtime ├── thread_id ├── root_dir └── _backend │ └── FilesystemBackend初始化的時候_backend None第一次執(zhí)行read() write() edit() ls_info() ...時再真正創(chuàng)建FilesystemBackend12. _ensure_backend整個設(shè)計(jì)的核心核心代碼其實(shí)非常簡單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這個方法負(fù)責(zé)保證只要真正需要文件操作就一定存在一個可用的 FilesystemBackend。第一次調(diào)用_backend None ↓ 創(chuàng)建目錄 ↓ 創(chuàng)建 FilesystemBackend ↓ 保存到 _backend第二次調(diào)用_backend ! None ↓ 直接返回已有實(shí)例因此它實(shí)際上是延遲初始化 實(shí)例緩存。13. 文件操作如何轉(zhuǎn)發(fā)有了_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 │ ▼ 真實(shí)文件系統(tǒng)這里的LazyFilesystemBackend實(shí)際上承擔(dān)了一層代理/適配作用。Agent 并不知道后面是否已經(jīng)初始化了真正的文件系統(tǒng)后端。14. 最終實(shí)現(xiàn)項(xiàng)目中最終實(shí)現(xiàn)的核心代碼如下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 # 獲取當(dāng)前會話 self._thread_id rt.get_thread_id( runtime ) # 構(gòu)造當(dāng)前會話的工作目錄 self._root_dir os.path.join( ROOT_PATH_AGENT, self._thread_id ) def _ensure_backend(self): if self._backend is None: # 第一次執(zhí)行文件操作時才創(chuàng)建目錄 os.makedirs( self._root_dir, exist_okTrue ) # 創(chuàng)建真正的文件系統(tǒng)后端 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, )現(xiàn)在修改為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, )這里最關(guān)鍵的一行就是backendmybackend.create_session_backend注意這里傳入的不是create_session_backend()而是create_session_backend因?yàn)檫@里需要把工廠函數(shù)本身交給框架。之后由框架在 Agent 執(zhí)行過程中根據(jù)當(dāng)前runtime調(diào)用它。16. 最終執(zhí)行流程完整鏈路可以概括為用戶打開聊天窗口 │ ▼ 生成 thread_id │ ▼ Agent 開始執(zhí)行 │ ▼ BackendFactory(runtime) │ ▼ create_session_backend(runtime) │ ▼ LazyFilesystemBackend(runtime) │ ├── 獲取 thread_id │ └── 計(jì)算 root_dir │ ▼ Agent 是否執(zhí)行文件操作 │ ┌─────┴─────┐ │ │ 否 是 │ │ ▼ ▼ 不創(chuàng)建 _ensure_backend() 文件目錄 │ ▼ 創(chuàng)建 thread 目錄 │ ▼ FilesystemBackend │ ▼ 執(zhí)行文件操作例如thread_id abc123最終得到/agent_files/abc123/另一個會話thread_id xyz456得到/agent_files/xyz456/兩個 Agent 會話之間的文件完全分離。17. 項(xiàng)目目錄結(jié)構(gòu)最終相關(guān)代碼結(jié)構(gòu)content/ ├── others/ │ └── mybackend.py │ ├── utils/ │ └── runtime_util.py │ └── all_agent.py職責(zé)也比較清晰runtime_util.py ↓ 負(fù)責(zé)運(yùn)行時信息獲取 ↓ thread_id mybackend.py ↓ 負(fù)責(zé)文件系統(tǒng)后端 ↓ LazyFilesystemBackend ↓ create_session_backend all_agent.py ↓ Agent 創(chuàng)建 ↓ 注入 BackendFactory18. 這次改造真正解決了什么18.1 會話隔離從所有會話 ↓ /agent_files變成thread_001 ↓ /agent_files/thread_001 thread_002 ↓ /agent_files/thread_002不同會話擁有獨(dú)立工作空間。18.2 解決同名文件沖突例如兩個會話都生成report.docx現(xiàn)在實(shí)際對應(yīng)/agent_files/thread_001/report.docx /agent_files/thread_002/report.docx不會互相覆蓋。18.3 降低跨會話文件訪問風(fēng)險(xiǎn)Agent 的文件操作始終發(fā)生在當(dāng)前thread_id對應(yīng)的工作目錄下。因此從文件系統(tǒng)層面建立了基本的會話邊界。需要注意的是這屬于應(yīng)用層面的文件隔離設(shè)計(jì)并不等同于完整的多租戶安全體系。如果真正用于生產(chǎn)環(huán)境還需要進(jìn)一步考慮權(quán)限控制、路徑穿越、用戶與 thread 的綁定關(guān)系、目錄生命周期以及數(shù)據(jù)清理等問題。19. 為什么最后選擇 Lazy Loading這個改造實(shí)際上經(jīng)歷了兩個版本。第一版BackendFactory ↓ 獲取 thread_id ↓ 創(chuàng)建目錄 ↓ 創(chuàng)建 FilesystemBackend ↓ 返回優(yōu)點(diǎn)是簡單直接。但是存在兩個問題第一Backend 創(chuàng)建階段就執(zhí)行了同步 I/O。這可能與異步執(zhí)行環(huán)境產(chǎn)生沖突或造成事件循環(huán)阻塞。第二沒有必要提前創(chuàng)建目錄。用戶可能只是聊天并沒有執(zhí)行任何文件操作。第二版改成BackendFactory ↓ 獲取 thread_id ↓ 計(jì)算 root_dir ↓ 暫不創(chuàng)建目錄 ↓ 等待真正的文件操作 ↓ _ensure_backend() ↓ 創(chuàng)建目錄 FilesystemBackend這樣Backend 初始化更加輕量文件目錄按需創(chuàng)建避免初始化階段執(zhí)行不必要的 I/O文件操作邏輯集中在_ensure_backend()原有FilesystemBackend的能力仍然可以復(fù)用因此最終選擇了第二種方案。20. 這次問題中比較值得記錄的源碼分析思路這次改造真正有價(jià)值的地方其實(shí)并不只是寫了一個LazyFilesystemBackend而是如何從框架的約束中找到擴(kuò)展點(diǎn)。最開始遇到的問題是FilesystemBackend的root_dir是固定的怎么動態(tài)設(shè)置如果直接從FilesystemBackend本身入手很容易陷入怎么修改 FilesystemBackend 怎么重新實(shí)現(xiàn) FilesystemBackend但繼續(xù)往上看create_deep_agent( backend... )發(fā)現(xiàn)BackendProtocol | BackendFactory再繼續(xù)看BackendFactory Callable[ [ToolRuntime], BackendProtocol ]于是整個思路就發(fā)生了變化不是修改 FilesystemBackend ↓ 而是利用 BackendFactory ↓ 讓 Backend 根據(jù) runtime 動態(tài)生成 ↓ 再通過 Lazy Backend 延遲真正的文件系統(tǒng)初始化這也是使用成熟框架時比較重要的一種思路遇到框架無法直接滿足的需求時先尋找框架提供的擴(kuò)展點(diǎn)而不是馬上繞開框架重寫整個功能。21. 最終方案總結(jié)整個方案可以濃縮成四層① Runtime ↓ 獲取 thread_id ② BackendFactory ↓ 根據(jù) thread_id 創(chuàng)建會話級 Backend ③ LazyFilesystemBackend ↓ 延遲真正的文件系統(tǒng)初始化 ④ FilesystemBackend ↓ 實(shí)際執(zhí)行 read / write / edit / grep / glob 等操作核心代碼關(guān)系create_deep_agent │ │ backend ▼ create_session_backend │ │ runtime ▼ LazyFilesystemBackend │ │ 第一次文件操作 ▼ FilesystemBackend │ │ root_dir ▼ /agent_files/{thread_id}最終實(shí)現(xiàn)了基于thread_id的 Agent 會話級文件系統(tǒng)隔離并通過BackendFactory Lazy Loading在不修改 DeepAgents 原有文件系統(tǒng)實(shí)現(xiàn)的情況下實(shí)現(xiàn)動態(tài)工作目錄。22. 一句話記錄這次技術(shù)實(shí)踐如果以后自己回頭看實(shí)際上記住下面這句話就夠了通過分析 DeepAgents 的BackendFactory擴(kuò)展機(jī)制獲取當(dāng)前運(yùn)行上下文中的thread_id動態(tài)構(gòu)造會話級root_dir同時使用 Lazy Loading 延遲FilesystemBackend的實(shí)例化和目錄創(chuàng)建從而實(shí)現(xiàn) Agent 多會話文件隔離并避免在 Backend 初始化階段執(zhí)行不必要的同步 I/O。