1. 制作基础 Tool在开始文件处理中间件之前先给 Agent 增加两个简单的 Tool方便后续测试获取当前时间执行终端命令Tool 放在content/mytools/globle_tools.py中通过get_tools()统一返回。其中获取时间比较简单主要用于处理一些和当前时间有关的任务。终端命令则不一样。对于 Agent 来说终端本身就是一个很通用的操作入口。例如cat file.txt可以读取文件echo hello file.txt可以写入文件python test.py可以执行 Agent 临时生成的 Python 代码。所以理论上只要 Agent 能执行终端命令就可以完成很多原本需要单独开发 Tool 的事情。但终端命令也存在明显的问题。首先是安全性Agent 生成的命令不一定可靠例如删除文件、修改系统配置等操作都可能造成实际影响。其次是环境依赖不同操作系统、不同运行环境下同一条命令可能有不同的结果。最后是返回结果不固定Agent 还需要自己理解命令输出。因此这里没有把run_command当成主要的文件操作方式而是作为一个兜底能力。后续如果有专门的 Tool 能完成某项工作优先使用专用 Tool实在没有合适的工具时再考虑通过终端处理。2. 文件处理中间件2.1 为什么需要前面的文件后端已经能够处理 Agent 的文件操作但实际使用过程中还有两个问题。第一个问题是Agent 对当前目录并不一定清楚。比如 Agent 前面刚生成了一个文件下一步又需要读取它。如果模型不知道当前目录里有哪些文件就可能重新创建文件或者使用错误的路径。第二个问题是工具返回结果可能包含系统绝对路径。例如系统实际使用D:/agent_files/thread_xxx/data/test.txt但对于 Agent 来说更合适的形式应该是data/test.txtAgent 没必要知道D:/agent_files/thread_xxx/这一层具体对应服务器上的什么位置。所以这里准备做一个文件处理中间件主要处理两件事模型调用前在必要的时候告诉 Agent 当前目录结构Tool 执行结束后将返回结果中的系统绝对路径转换成相对路径。这样文件后端负责真正的文件操作中间件负责处理 Agent 与文件系统之间的一些衔接问题。3. 实现思路3.1 目录结构首先需要一个方法根据真实工作目录生成类似下面的结构project/ ├── src/ │ ├── main.py │ └── utils.py ├── data/ └── README.md这个方法放在utils/doc_utils/os_util.py里面。实现上就是使用os.listdir()递归遍历目录同时处理好目录层级和├──、└──这些符号。这里没有直接调用系统的tree命令而是自己实现。主要还是考虑到项目需要兼容不同运行环境而且这个方法本身并不复杂自己实现之后也更容易控制输出格式。3.2 不需要每次都重新获取目录目录结构虽然有用但也没必要每次调用模型都重新发送。例如调用模型 → 调用 Tool → Tool 返回 → 再次调用模型如果中间没有创建、删除或者修改文件那么目录结构其实没有变化。因此这里增加一个简单的判断当前目录最新文件修改时间 ↓ 是否晚于上一次记录的时间 ↓ 是 → 重新获取目录结构 否 → 什么都不做为了保存这个状态对AgentState做一个简单扩展class CustomState(AgentState): start_work_time: float file_update_time: float其中file_update_time用来记录已经知道的文件更新时间。Agent 开始执行时记录一次当前时间async def abefore_agent(self, state, runtime): current_time time.time() return { start_work_time: current_time, file_update_time: current_time }之后在abefore_model中检查当前目录。文件更新时间的获取同样放在os_util.py中。通过os.walk()遍历目录下的文件获取每个文件的mtime最后取最大值。4. 路径处理4.1 系统路径和 Agent 路径这里需要区分两个概念。系统路径是程序真正访问文件时使用的路径例如D:/agent_files/thread_123/data/test.txtAgent 路径则只需要data/test.txt因此项目中增加ROOT_PATH_SYSTEM用于表示系统上的真实工作根目录。Windows 下需要带盘符例如D:/agent_filesLinux 下则可以直接使用/agent_files再和当前线程 ID 拼接就可以得到当前 Agent 的实际工作目录。4.2 获取当前线程目录在content/utils/runtime_util.py中增加def get_root_thread_dir(): raw_path os.path.join( gc.ROOT_PATH_SYSTEM, get_thread_id() ) return os.path.normpath(raw_path)以后需要获取当前线程真实工作目录时直接调用这个方法即可。例如D:/agent_files/thread_123这样中间件不需要自己处理线程 ID 和系统根目录。4.3 将绝对路径转换为相对路径Tool 执行完成后如果返回D:\agent_files\thread_123\data\test.txt需要转换成data\test.txt因此在runtime_util.py中再增加一个def get_out_path(file_path): ...它主要做的事情就是获取当前线程目录 ↓ 判断返回结果是否包含该目录 ↓ 如果包含去掉线程目录部分 ↓ 得到 Agent 使用的相对路径这样路径处理逻辑集中在一个地方后面其他地方如果也需要做相同转换可以直接复用。5. 文件处理中间件前面的几个方法准备好之后就可以把它们组合到中间件中。中间件放在content/middles/file_manager_middle.py这里使用AgentMiddleware并实现三个生命周期方法abefore_agent abefore_model awrap_tool_call分别对应 Agent 开始执行、模型调用前、Tool 调用前后这几个阶段。5.1abefore_agentAgent 开始执行时记录两个时间start_work_time file_update_time目前start_work_time主要是为了记录本次任务的开始时间后续如果需要统计 Agent 工作时间也可以直接使用。file_update_time则用于后面的文件变化检测。5.2abefore_model模型调用之前先获取当前线程的真实工作目录。然后检查最新文件修改时间 file_update_time如果没有变化就直接继续调用模型。如果有变化则重新生成目录结构并把目录结构作为一条ToolMessage加入消息中。逻辑大致如下async def abefore_model(self, state, runtime): dir_path rt.get_root_thread_dir() if await self._check_if_new_file( dir_path, state[file_update_time] ): content get_directory_tree(dir_path) return { messages: [ ToolMessage( contentf当前目录结构:\n{content}, tool_call_idget_uuid(), namesummery_file_paths ) ], file_update_time: time.time() }这里有两个值得注意的地方。第一目录遍历本身是同步 I/O而 Agent 当前运行环境是异步的因此不能直接在异步流程里做大量同步 I/O。这里使用await asyncio.to_thread(...)将同步操作放到线程中执行。第二这条ToolMessage并不是 Agent 真正调用某个 Tool 得到的因此没有真实的tool_call_id。这里生成一个 UUID 作为标识即可。5.3awrap_tool_call另一个问题发生在 Tool 执行完成以后。这里通过awrap_tool_call包住实际的 Tool 调用async def awrap_tool_call(self, request, handler): tool_result await handler(request) if isinstance(tool_result, ToolMessage): tool_result.content rt.get_out_path( tool_result.content ) return tool_result执行顺序就是Agent 调用 Tool ↓ handler(request) ↓ Tool 真正执行 ↓ 得到 ToolMessage ↓ 处理其中的绝对路径 ↓ 返回 Agent这样就不需要在每个 Tool 中分别处理路径。如果后面再增加新的文件 Tool只要它正常返回结果中间件就可以统一处理。6. 接入 Agent最后在all_agent.py中注册中间件self.agent create_deep_agent( modelget_llm(), toolsself._get_tools(), middlewareself._get_middlewares(), backendmybackend.create_session_backend, system_promptprompt, )中间件列表def _get_middlewares(self): return [ file_manager_middle.FileMiddleware() ]至此文件管理相关的处理逻辑就从具体 Tool 中独立出来了。7. 最终目录本次主要新增和修改以下文件├── base │ └── configs.py │ # 增加 ROOT_PATH_SYSTEM │ ├── content │ ├── all_agent.py │ │ # 注册文件处理中间件 │ │ │ ├── middles │ │ └── file_manager_middle.py │ │ # 文件处理中间件 │ │ │ ├── mytools │ │ └── globle_tools.py │ │ # 基础 Tool │ │ │ └── utils │ └── runtime_util.py │ # 工作目录、路径转换 │ └── utils ├── doc_utils │ └── os_util.py │ # 目录遍历、文件更新时间 │ └── general_utils └── globle_util.py # 系统判断、UUID等通用方法8. 小结这次主要做的其实不是一个新的文件 Tool而是把文件系统相关的公共处理逻辑放到了 Agent Middleware 中。最终处理流程Agent 开始 ↓ 记录文件状态 ↓ 模型调用前检查文件变化 ↓ 有变化 → 更新目录结构 ↓ Tool 执行 ↓ Tool 返回结果 ↓ 绝对路径转换为相对路径 ↓ 返回 Agent这样处理之后Agent 可以在文件发生变化后重新获得目录结构没有文件变化时不会重复发送目录信息Agent 看到的是相对路径不需要知道服务器上的真实目录文件路径处理不需要分散到每一个 Tool 中文件相关逻辑与具体业务 Tool 解耦。这也是这次使用 Middleware 比直接修改 Tool 更合适的地方文件变化检测和路径处理本身并不属于某一个具体 Tool而是 Agent 执行过程中的公共逻辑。
1. 制作基础 Tool在开始文件处理中间件之前先给 Agent 增加两个简单的 Tool方便后续测试获取当前时间执行终端命令Tool 放在content/mytools/globle_tools.py中通过get_tools()统一返回。其中获取时间比较简单主要用于处理一些和当前时间有关的任务。终端命令则不一样。对于 Agent 来说终端本身就是一个很通用的操作入口。例如cat file.txt可以读取文件echo hello file.txt可以写入文件python test.py可以执行 Agent 临时生成的 Python 代码。所以理论上只要 Agent 能执行终端命令就可以完成很多原本需要单独开发 Tool 的事情。但终端命令也存在明显的问题。首先是安全性Agent 生成的命令不一定可靠例如删除文件、修改系统配置等操作都可能造成实际影响。其次是环境依赖不同操作系统、不同运行环境下同一条命令可能有不同的结果。最后是返回结果不固定Agent 还需要自己理解命令输出。因此这里没有把run_command当成主要的文件操作方式而是作为一个兜底能力。后续如果有专门的 Tool 能完成某项工作优先使用专用 Tool实在没有合适的工具时再考虑通过终端处理。2. 文件处理中间件2.1 为什么需要前面的文件后端已经能够处理 Agent 的文件操作但实际使用过程中还有两个问题。第一个问题是Agent 对当前目录并不一定清楚。比如 Agent 前面刚生成了一个文件下一步又需要读取它。如果模型不知道当前目录里有哪些文件就可能重新创建文件或者使用错误的路径。第二个问题是工具返回结果可能包含系统绝对路径。例如系统实际使用D:/agent_files/thread_xxx/data/test.txt但对于 Agent 来说更合适的形式应该是data/test.txtAgent 没必要知道D:/agent_files/thread_xxx/这一层具体对应服务器上的什么位置。所以这里准备做一个文件处理中间件主要处理两件事模型调用前在必要的时候告诉 Agent 当前目录结构Tool 执行结束后将返回结果中的系统绝对路径转换成相对路径。这样文件后端负责真正的文件操作中间件负责处理 Agent 与文件系统之间的一些衔接问题。3. 实现思路3.1 目录结构首先需要一个方法根据真实工作目录生成类似下面的结构project/ ├── src/ │ ├── main.py │ └── utils.py ├── data/ └── README.md这个方法放在utils/doc_utils/os_util.py里面。实现上就是使用os.listdir()递归遍历目录同时处理好目录层级和├──、└──这些符号。这里没有直接调用系统的tree命令而是自己实现。主要还是考虑到项目需要兼容不同运行环境而且这个方法本身并不复杂自己实现之后也更容易控制输出格式。3.2 不需要每次都重新获取目录目录结构虽然有用但也没必要每次调用模型都重新发送。例如调用模型 → 调用 Tool → Tool 返回 → 再次调用模型如果中间没有创建、删除或者修改文件那么目录结构其实没有变化。因此这里增加一个简单的判断当前目录最新文件修改时间 ↓ 是否晚于上一次记录的时间 ↓ 是 → 重新获取目录结构 否 → 什么都不做为了保存这个状态对AgentState做一个简单扩展class CustomState(AgentState): start_work_time: float file_update_time: float其中file_update_time用来记录已经知道的文件更新时间。Agent 开始执行时记录一次当前时间async def abefore_agent(self, state, runtime): current_time time.time() return { start_work_time: current_time, file_update_time: current_time }之后在abefore_model中检查当前目录。文件更新时间的获取同样放在os_util.py中。通过os.walk()遍历目录下的文件获取每个文件的mtime最后取最大值。4. 路径处理4.1 系统路径和 Agent 路径这里需要区分两个概念。系统路径是程序真正访问文件时使用的路径例如D:/agent_files/thread_123/data/test.txtAgent 路径则只需要data/test.txt因此项目中增加ROOT_PATH_SYSTEM用于表示系统上的真实工作根目录。Windows 下需要带盘符例如D:/agent_filesLinux 下则可以直接使用/agent_files再和当前线程 ID 拼接就可以得到当前 Agent 的实际工作目录。4.2 获取当前线程目录在content/utils/runtime_util.py中增加def get_root_thread_dir(): raw_path os.path.join( gc.ROOT_PATH_SYSTEM, get_thread_id() ) return os.path.normpath(raw_path)以后需要获取当前线程真实工作目录时直接调用这个方法即可。例如D:/agent_files/thread_123这样中间件不需要自己处理线程 ID 和系统根目录。4.3 将绝对路径转换为相对路径Tool 执行完成后如果返回D:\agent_files\thread_123\data\test.txt需要转换成data\test.txt因此在runtime_util.py中再增加一个def get_out_path(file_path): ...它主要做的事情就是获取当前线程目录 ↓ 判断返回结果是否包含该目录 ↓ 如果包含去掉线程目录部分 ↓ 得到 Agent 使用的相对路径这样路径处理逻辑集中在一个地方后面其他地方如果也需要做相同转换可以直接复用。5. 文件处理中间件前面的几个方法准备好之后就可以把它们组合到中间件中。中间件放在content/middles/file_manager_middle.py这里使用AgentMiddleware并实现三个生命周期方法abefore_agent abefore_model awrap_tool_call分别对应 Agent 开始执行、模型调用前、Tool 调用前后这几个阶段。5.1abefore_agentAgent 开始执行时记录两个时间start_work_time file_update_time目前start_work_time主要是为了记录本次任务的开始时间后续如果需要统计 Agent 工作时间也可以直接使用。file_update_time则用于后面的文件变化检测。5.2abefore_model模型调用之前先获取当前线程的真实工作目录。然后检查最新文件修改时间 file_update_time如果没有变化就直接继续调用模型。如果有变化则重新生成目录结构并把目录结构作为一条ToolMessage加入消息中。逻辑大致如下async def abefore_model(self, state, runtime): dir_path rt.get_root_thread_dir() if await self._check_if_new_file( dir_path, state[file_update_time] ): content get_directory_tree(dir_path) return { messages: [ ToolMessage( contentf当前目录结构:\n{content}, tool_call_idget_uuid(), namesummery_file_paths ) ], file_update_time: time.time() }这里有两个值得注意的地方。第一目录遍历本身是同步 I/O而 Agent 当前运行环境是异步的因此不能直接在异步流程里做大量同步 I/O。这里使用await asyncio.to_thread(...)将同步操作放到线程中执行。第二这条ToolMessage并不是 Agent 真正调用某个 Tool 得到的因此没有真实的tool_call_id。这里生成一个 UUID 作为标识即可。5.3awrap_tool_call另一个问题发生在 Tool 执行完成以后。这里通过awrap_tool_call包住实际的 Tool 调用async def awrap_tool_call(self, request, handler): tool_result await handler(request) if isinstance(tool_result, ToolMessage): tool_result.content rt.get_out_path( tool_result.content ) return tool_result执行顺序就是Agent 调用 Tool ↓ handler(request) ↓ Tool 真正执行 ↓ 得到 ToolMessage ↓ 处理其中的绝对路径 ↓ 返回 Agent这样就不需要在每个 Tool 中分别处理路径。如果后面再增加新的文件 Tool只要它正常返回结果中间件就可以统一处理。6. 接入 Agent最后在all_agent.py中注册中间件self.agent create_deep_agent( modelget_llm(), toolsself._get_tools(), middlewareself._get_middlewares(), backendmybackend.create_session_backend, system_promptprompt, )中间件列表def _get_middlewares(self): return [ file_manager_middle.FileMiddleware() ]至此文件管理相关的处理逻辑就从具体 Tool 中独立出来了。7. 最终目录本次主要新增和修改以下文件├── base │ └── configs.py │ # 增加 ROOT_PATH_SYSTEM │ ├── content │ ├── all_agent.py │ │ # 注册文件处理中间件 │ │ │ ├── middles │ │ └── file_manager_middle.py │ │ # 文件处理中间件 │ │ │ ├── mytools │ │ └── globle_tools.py │ │ # 基础 Tool │ │ │ └── utils │ └── runtime_util.py │ # 工作目录、路径转换 │ └── utils ├── doc_utils │ └── os_util.py │ # 目录遍历、文件更新时间 │ └── general_utils └── globle_util.py # 系统判断、UUID等通用方法8. 小结这次主要做的其实不是一个新的文件 Tool而是把文件系统相关的公共处理逻辑放到了 Agent Middleware 中。最终处理流程Agent 开始 ↓ 记录文件状态 ↓ 模型调用前检查文件变化 ↓ 有变化 → 更新目录结构 ↓ Tool 执行 ↓ Tool 返回结果 ↓ 绝对路径转换为相对路径 ↓ 返回 Agent这样处理之后Agent 可以在文件发生变化后重新获得目录结构没有文件变化时不会重复发送目录信息Agent 看到的是相对路径不需要知道服务器上的真实目录文件路径处理不需要分散到每一个 Tool 中文件相关逻辑与具体业务 Tool 解耦。这也是这次使用 Middleware 比直接修改 Tool 更合适的地方文件变化检测和路径处理本身并不属于某一个具体 Tool而是 Agent 执行过程中的公共逻辑。
