企业官网建设流程全解析

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.txt

Agent 没必要知道D:/agent_files/thread_xxx/这一层具体对应服务器上的什么位置。

所以这里准备做一个文件处理中间件，主要处理两件事：

1. 模型调用前，在必要的时候告诉 Agent 当前目录结构；

2. 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.txt

Agent 路径则只需要：

data/test.txt

因此项目中增加：

ROOT_PATH_SYSTEM

用于表示系统上的真实工作根目录。

Windows 下需要带盘符，例如：

D:/agent_files

Linux 下则可以直接使用：

/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_agent

Agent 开始执行时，记录两个时间：

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( content=f"当前目录结构:\n{content}", tool_call_id=get_uuid(), name="summery_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( model=get_llm(), tools=self._get_tools(), middleware=self._get_middlewares(), backend=mybackend.create_session_backend, system_prompt=prompt, )

中间件列表：

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 执行过程中的公共逻辑。