1. 问题目前 Agent 已经可以在工作目录中创建和处理文件但这些文件都保存在后端服务器上前端无法直接访问。因此需要增加一个中间环节把 Agent 生成的文件交给用户。这里采用对象存储的方式Agent生成文件 ↓ 工作目录 ↓ 打包 ↓ MinIO ↓ 生成下载链接 ↓ 前端下载这样 Agent 仍然只负责处理自己的工作目录文件交付由后续流程完成。2. 对象存储**对象存储Object Storage**主要用于保存图片、视频、压缩包等非结构化文件。和普通文件系统相比对象存储更适合通过程序进行文件上传和访问。可以简单理解成一个可以通过 API 操作的网盘。本项目使用MinIO。MinIO 是一个开源对象存储服务可以直接部署在自己的服务器上并提供 Python SDK。这里使用 MinIO 主要是因为项目本身可以本地部署没有必要为了文件存储再依赖第三方云存储。3. MinIO 部署项目使用 Docker 部署 MinIO。docker pull minio/minio mkdir -p /usr/local/minio-data docker run -p 9000:9000 -p 9190:9090 \ --name minio \ -d --restartalways \ -e MINIO_ROOT_USERminio \ -e MINIO_ROOT_PASSWORDminio123 \ -v /usr/local/minio-data:/data \ minio/minio server /data --console-address :9090启动后可以通过http://服务器IP:9190进入 MinIO 管理页面。如果是在本机部署则使用http://localhost:9190登录后即可看到 MinIO 的管理界面。4. Python SDK安装 MinIO 的 Python SDKpip install minio最基本的操作就是连接 MinIO、创建 Bucket、上传文件以及生成下载链接。from minio import Minio client Minio( localhost:9000, access_keyminio, secret_keyminio123, secureFalse )项目实际只需要用到其中两个操作上传文件 生成下载链接下载链接使用presigned_get_object()生成。它返回的是一个带有效期的 URL用户拿到链接后可以直接下载文件不需要知道 MinIO 的账号密码。5. 文件传递方案5.1 为什么需要打包一个 Agent 任务不一定只生成一个文件。例如工作目录/ ├── result.docx ├── data.xlsx ├── image.png └── output.pdf如果逐个上传并生成链接前端还需要处理多个文件。因此这里直接将整个工作目录打包成一个 ZIP工作目录 ↓ output.zip ↓ MinIO ↓ 一个下载链接用户只需要下载一个文件即可。5.2 为什么放到 Agent 结束之后处理文件上传不属于具体工具的功能。例如create_file工具只负责创建文件如果让它同时负责上传就会让工具本身承担文件交付的工作。因此将这部分放到FileMiddleware中在 Agent 执行完成后统一处理。6. 开始实现6.1 添加 ZIP 压缩方法在utils/doc_utils下增加zip_files.py。使用 Python 自带的shutil直接压缩目录import os import shutil def compress_dir(folder_path): 将目录压缩成 zip output_path shutil.make_archive( base_namefolder_path, formatzip, root_diros.path.dirname(folder_path), base_diros.path.basename(folder_path) ) return output_path这里没有额外引入压缩库项目目前的需求比较简单标准库就够用了。6.2 添加 MinIO 配置MinIO 的地址和认证信息放到.env中HOST_IPlocalhost MINIO_PORT9000 MINIO_ACCESS_KEYminio MINIO_SECRET_KEYminio123然后在配置文件中统一读取HOST_IP os.getenv(HOST_IP) MINIO_PORT os.getenv(MINIO_PORT) MINIO_ENDPOINT f{HOST_IP}:{MINIO_PORT} MINIO_ACCESS_KEY os.getenv(MINIO_ACCESS_KEY) MINIO_SECRET_KEY os.getenv(MINIO_SECRET_KEY)这样代码中不需要直接写 MinIO 的连接信息部署到其他环境时只需要修改环境变量。6.3 封装 MinIO 连接在conn下新建minio_conn.py。不直接在 Middleware 中操作 MinIO SDK而是简单封装一层from minio import Minio from base import configs as cfg class MinioConn: def __init__(self): self.client self.create_client() self.bucket_name stores self._init_bucket() def create_client(self): return Minio( cfg.MINIO_ENDPOINT, access_keycfg.MINIO_ACCESS_KEY, secret_keycfg.MINIO_SECRET_KEY, secureFalse ) def _init_bucket(self): if not self.client.bucket_exists(self.bucket_name): self.client.make_bucket(self.bucket_name) def upload_obj(self, object_name, file_path): self.client.fput_object( self.bucket_name, object_name, file_path ) def gen_presigned_url(self, object_name): return self.client.presigned_get_object( self.bucket_name, object_name )这里暂时只保留项目需要的两个方法upload_obj() gen_presigned_url()Bucket 在初始化时检查不存在就自动创建。7. 接入 FileMiddlewareMinIO 部分准备好之后最后需要把它接到 Agent 的执行流程中。首先在FileMiddleware初始化时创建 MinIO 连接class FileMiddleware(AgentMiddleware[CustomState]): def __init__(self): super().__init__() self.mc MinioConn()然后利用before_agent和aafter_agent两个生命周期节点完成文件处理。7.1 记录开始时间在before_agent中记录当前任务的开始时间。原因是工作目录可能本身就存在一些文件不能每次 Agent 执行结束都把整个目录当成新文件。因此需要知道Agent 开始执行之后有没有产生新的文件。7.2 Agent 执行结束后处理文件主要逻辑放在aafter_agentasync def aafter_agent(self, state, runtime): dir_path rt.get_root_thread_dir() # 判断任务期间是否产生新文件 if not await self._check_if_new_file( dir_path, state[start_work_time] ): return None # 压缩目录 local_path await asyncio.to_thread( compress_dir, dir_path ) # 上传 MinIO object_name os.path.basename(local_path) await asyncio.to_thread( self.mc.upload_obj, object_name, local_path ) # 生成下载链接 url self.mc.gen_presigned_url(object_name) return { messages: [ AIMessage(content可通过如下地址下载), AIMessage(contenturl) ] }这里压缩和上传都属于比较明显的文件 I/O 操作因此使用asyncio.to_thread()执行避免直接阻塞当前异步流程。如果任务期间没有产生文件则直接返回不进行压缩和上传。8. 测试可以直接在 Agent Chat UI 中测试例如做个 a.txt随便写点什么Agent 完成后可以看到返回结果中多了一条下载链接。整个过程实际上就是创建 a.txt ↓ Agent 执行结束 ↓ 检测到新文件 ↓ 工作目录压缩成 ZIP ↓ 上传 MinIO ↓ 生成 Presigned URL ↓ 返回给前端点击链接即可下载 Agent 生成的文件。9. 修改后的目录本次主要涉及以下几个文件.env # MinIO 配置 base/ └── configs.py # 读取 MinIO 配置 conn/ └── minio_conn.py # MinIO 连接、上传、生成下载链接 content/ └── middles/ └── file_manager_middle.py # Agent 执行结束后的文件处理 utils/ └── doc_utils/ └── zip_files.py # 工作目录压缩10. 小结这一部分实际上没有改变 Agent 原来的文件处理方式只是在原来的工作目录后面增加了一个文件交付流程Agent工作目录 ↓ 检测新文件 ↓ ZIP ↓ MinIO ↓ Presigned URL ↓ 用户Agent 继续在自己的工作目录中操作文件Middleware 负责在任务结束后把文件整理并交给用户。这样文件生成和文件传递就分开了。 
长河君 · 2026/8/30 18:35:53 · 阅读 1.2k Agent 文件传递
1. 问题目前 Agent 已经可以在工作目录中创建和处理文件但这些文件都保存在后端服务器上前端无法直接访问。因此需要增加一个中间环节把 Agent 生成的文件交给用户。这里采用对象存储的方式Agent生成文件 ↓ 工作目录 ↓ 打包 ↓ MinIO ↓ 生成下载链接 ↓ 前端下载这样 Agent 仍然只负责处理自己的工作目录文件交付由后续流程完成。2. 对象存储**对象存储Object Storage**主要用于保存图片、视频、压缩包等非结构化文件。和普通文件系统相比对象存储更适合通过程序进行文件上传和访问。可以简单理解成一个可以通过 API 操作的网盘。本项目使用MinIO。MinIO 是一个开源对象存储服务可以直接部署在自己的服务器上并提供 Python SDK。这里使用 MinIO 主要是因为项目本身可以本地部署没有必要为了文件存储再依赖第三方云存储。3. MinIO 部署项目使用 Docker 部署 MinIO。docker pull minio/minio mkdir -p /usr/local/minio-data docker run -p 9000:9000 -p 9190:9090 \ --name minio \ -d --restartalways \ -e MINIO_ROOT_USERminio \ -e MINIO_ROOT_PASSWORDminio123 \ -v /usr/local/minio-data:/data \ minio/minio server /data --console-address :9090启动后可以通过http://服务器IP:9190进入 MinIO 管理页面。如果是在本机部署则使用http://localhost:9190登录后即可看到 MinIO 的管理界面。4. Python SDK安装 MinIO 的 Python SDKpip install minio最基本的操作就是连接 MinIO、创建 Bucket、上传文件以及生成下载链接。from minio import Minio client Minio( localhost:9000, access_keyminio, secret_keyminio123, secureFalse )项目实际只需要用到其中两个操作上传文件 生成下载链接下载链接使用presigned_get_object()生成。它返回的是一个带有效期的 URL用户拿到链接后可以直接下载文件不需要知道 MinIO 的账号密码。5. 文件传递方案5.1 为什么需要打包一个 Agent 任务不一定只生成一个文件。例如工作目录/ ├── result.docx ├── data.xlsx ├── image.png └── output.pdf如果逐个上传并生成链接前端还需要处理多个文件。因此这里直接将整个工作目录打包成一个 ZIP工作目录 ↓ output.zip ↓ MinIO ↓ 一个下载链接用户只需要下载一个文件即可。5.2 为什么放到 Agent 结束之后处理文件上传不属于具体工具的功能。例如create_file工具只负责创建文件如果让它同时负责上传就会让工具本身承担文件交付的工作。因此将这部分放到FileMiddleware中在 Agent 执行完成后统一处理。6. 开始实现6.1 添加 ZIP 压缩方法在utils/doc_utils下增加zip_files.py。使用 Python 自带的shutil直接压缩目录import os import shutil def compress_dir(folder_path): 将目录压缩成 zip output_path shutil.make_archive( base_namefolder_path, formatzip, root_diros.path.dirname(folder_path), base_diros.path.basename(folder_path) ) return output_path这里没有额外引入压缩库项目目前的需求比较简单标准库就够用了。6.2 添加 MinIO 配置MinIO 的地址和认证信息放到.env中HOST_IPlocalhost MINIO_PORT9000 MINIO_ACCESS_KEYminio MINIO_SECRET_KEYminio123然后在配置文件中统一读取HOST_IP os.getenv(HOST_IP) MINIO_PORT os.getenv(MINIO_PORT) MINIO_ENDPOINT f{HOST_IP}:{MINIO_PORT} MINIO_ACCESS_KEY os.getenv(MINIO_ACCESS_KEY) MINIO_SECRET_KEY os.getenv(MINIO_SECRET_KEY)这样代码中不需要直接写 MinIO 的连接信息部署到其他环境时只需要修改环境变量。6.3 封装 MinIO 连接在conn下新建minio_conn.py。不直接在 Middleware 中操作 MinIO SDK而是简单封装一层from minio import Minio from base import configs as cfg class MinioConn: def __init__(self): self.client self.create_client() self.bucket_name stores self._init_bucket() def create_client(self): return Minio( cfg.MINIO_ENDPOINT, access_keycfg.MINIO_ACCESS_KEY, secret_keycfg.MINIO_SECRET_KEY, secureFalse ) def _init_bucket(self): if not self.client.bucket_exists(self.bucket_name): self.client.make_bucket(self.bucket_name) def upload_obj(self, object_name, file_path): self.client.fput_object( self.bucket_name, object_name, file_path ) def gen_presigned_url(self, object_name): return self.client.presigned_get_object( self.bucket_name, object_name )这里暂时只保留项目需要的两个方法upload_obj() gen_presigned_url()Bucket 在初始化时检查不存在就自动创建。7. 接入 FileMiddlewareMinIO 部分准备好之后最后需要把它接到 Agent 的执行流程中。首先在FileMiddleware初始化时创建 MinIO 连接class FileMiddleware(AgentMiddleware[CustomState]): def __init__(self): super().__init__() self.mc MinioConn()然后利用before_agent和aafter_agent两个生命周期节点完成文件处理。7.1 记录开始时间在before_agent中记录当前任务的开始时间。原因是工作目录可能本身就存在一些文件不能每次 Agent 执行结束都把整个目录当成新文件。因此需要知道Agent 开始执行之后有没有产生新的文件。7.2 Agent 执行结束后处理文件主要逻辑放在aafter_agentasync def aafter_agent(self, state, runtime): dir_path rt.get_root_thread_dir() # 判断任务期间是否产生新文件 if not await self._check_if_new_file( dir_path, state[start_work_time] ): return None # 压缩目录 local_path await asyncio.to_thread( compress_dir, dir_path ) # 上传 MinIO object_name os.path.basename(local_path) await asyncio.to_thread( self.mc.upload_obj, object_name, local_path ) # 生成下载链接 url self.mc.gen_presigned_url(object_name) return { messages: [ AIMessage(content可通过如下地址下载), AIMessage(contenturl) ] }这里压缩和上传都属于比较明显的文件 I/O 操作因此使用asyncio.to_thread()执行避免直接阻塞当前异步流程。如果任务期间没有产生文件则直接返回不进行压缩和上传。8. 测试可以直接在 Agent Chat UI 中测试例如做个 a.txt随便写点什么Agent 完成后可以看到返回结果中多了一条下载链接。整个过程实际上就是创建 a.txt ↓ Agent 执行结束 ↓ 检测到新文件 ↓ 工作目录压缩成 ZIP ↓ 上传 MinIO ↓ 生成 Presigned URL ↓ 返回给前端点击链接即可下载 Agent 生成的文件。9. 修改后的目录本次主要涉及以下几个文件.env # MinIO 配置 base/ └── configs.py # 读取 MinIO 配置 conn/ └── minio_conn.py # MinIO 连接、上传、生成下载链接 content/ └── middles/ └── file_manager_middle.py # Agent 执行结束后的文件处理 utils/ └── doc_utils/ └── zip_files.py # 工作目录压缩10. 小结这一部分实际上没有改变 Agent 原来的文件处理方式只是在原来的工作目录后面增加了一个文件交付流程Agent工作目录 ↓ 检测新文件 ↓ ZIP ↓ MinIO ↓ Presigned URL ↓ 用户Agent 继续在自己的工作目录中操作文件Middleware 负责在任务结束后把文件整理并交给用户。这样文件生成和文件传递就分开了。