手把手学习Agent 及使用-4.2(File I/O 文件读写工具 )

h5打开以查看上一关我们完成了 Agent 底层架构的全面重构搭建了标准化、可扩展的工具注册中心彻底解决了老旧架构耦合臃肿、难以拓展高阶工具的问题为后续所有工具开发统一了技术规范。有了稳定的架构底座我们正式开启 其他进阶工具的落地迭代。在真实落地场景中Agent 不仅需要检索信息、联网查询更需要具备信息沉淀与文档输出能力能够自主整理对话结果、汇总研究内容、生成标准化文档报告。本篇将依托全新工具注册中心从零实现一套高安全、多格式适配的 File I/O 文件读写工具解决路径穿越、编码冲突、中文乱码、超长文本拆分等核心问题让 Agent 拥有专属“书写能力”补齐智能体内容落地的关键短板。一、阶段一File I/O 文件读写工具这是最简单的一个工具也是生成“最终研究报告”的必备工具。我们现在最开始的工具描述 Schema 中加入新工具介绍{ type: function, function: { name: save_document, description: 将整理好的研究报告、长文本保存至本地固定工作目录支持txt/md/pdf三种格式可选择追加写入已有文件, parameters: { type: object, required: [title, content, format], properties: { title: {type: string, description: 文档文件名无需后缀系统自动补充}, content: {type: string, description: 需要写入文档的完整文本内容}, format: {type: string, enum: [txt, md, pdf], description: 文件格式仅支持txt/md/pdf}, append: {type: boolean, default: False, description: 是否追加写入已有文件False则覆盖原有内容} } } } }并在提示词中加入这个工具3. save_document将整理后的长报告保存到本地文档支持txt/md/pdf支持追加写入。1. 思路在构建这个函数的过程中我们需要考虑写一个安全前置函数去防御路径穿越即避免大模型把文件位置乱写乱存所以我们计划在代码中写死文件存放路径。同时我们考虑到中文环境下的 GBK 与 UTF-8 的冲突。以及我们需要考虑到一个关键问题 大模型在生成超长文本时可能会因为 token 限制中断文件工具是否需要支持append追加模式2. 文件工具主流程固定文件读写根目录在正式开始撰写工具代码时从一开始固定 Agent 的执行目录要求 Agent 把文件存放在./agent_workspace下。如果没有这个文件夹则新建文件夹。WORKSPACE_ROOT Path(./agent_workspace) WORKSPACE_ROOT.mkdir(exist_okTrue, parentsTrue)注意./agent_workspace的.指命令行启动程序时的工作目录不是 py 脚本文件所在目录。如果你在别的文件夹运行脚本文件夹会生成到你敲命令的位置而不是代码文件旁边。如果想要文件夹固定生成在 py 脚本同目录改成WORKSPACE_ROOT Path(__file__).parent / agent_workspace WORKSPACE_ROOT.mkdir(exist_okTrue, parentsTrue)一般选择将其固定到代码文件旁边不管是人找起来还是 Agent 找起来都比较方便。安全校验函数因为这个工具需要我们存储和修改本地文件因此加入专属安全校验函数去检测路径穿越、非法文件名、非法格式等问题。def file_security_check(title: str, format: str, **kwargs) - Optional[str]: # 1. 校验文件格式 allow_formats {txt, md, pdf} if format not in allow_formats: return f不支持的文件格式 {format}仅允许 txt/md/pdf # 2. 清洗文件名过滤非法字符禁止../ / \等逃逸字符 unsafe_pattern re.compile(r[\\/:*?|]|\.\.) clean_title unsafe_pattern.sub(_, title) if clean_title ! title: return f文件名包含非法字符/路径逃逸字符已自动清洗为 {clean_title}请重新调用工具使用安全文件名 # 3. 拼接完整路径强制校验文件在WORKSPACE_ROOT内 file_name f{clean_title}.{format} target_path (WORKSPACE_ROOT / file_name).resolve() root_abs WORKSPACE_ROOT.resolve() # 核心防路径穿越目标路径必须是工作目录的子文件 if not str(target_path).startswith(str(root_abs)): return f路径逃逸拦截禁止写入工作目录外路径 {target_path}仅可保存至 {WORKSPACE_ROOT} # 校验通过无报错返回None return None这就是绑定给 save_document 的前置安全校验函数。当通过run_tool(save_document, ...)调用工具时run_tool 会先自动执行这个函数。返回None代表校验放行返回字符串代表校验失败直接拦截不会执行真正的保存函数。安装 PDF 依赖如果想让程序有导出 PDF 的能力需要安装第三方 PDF 生成库。如果不安装程序也可以用只是不能导出 PDF。pip install reportlabtry: from reportlab.lib.pagesizes import A4 from reportlab.pdfgen import canvas PDF_AVAILABLE True except ImportError: PDF_AVAILABLE False print(警告未安装reportlabpdf格式导出功能不可用执行 pip install reportlab 开启)同时由于我们希望导出的 PDF 是中文版本因此需要安装字体库这里采用 Windows 宋体字体来源于 Windows 系统字体路径不需要额外下载字体。# 注册中文字体Windows 宋体 try: # Windows 系统字体路径 font_path C:/Windows/Fonts/simsun.ttc if os.path.exists(font_path): pdfmetrics.registerFont(TTFont(SimSun, font_path)) CHINESE_FONT_AVAILABLE True print(已加载系统宋体PDF 中文导出可用) else: # 备选尝试黑体微软雅黑 font_path C:/Windows/Fonts/msyh.ttc if os.path.exists(font_path): pdfmetrics.registerFont(TTFont(SimSun, font_path)) # 仍注册为 SimSun 名字方便调用 CHINESE_FONT_AVAILABLE True print(已加载微软雅黑作为 PDF 中文字体) else: CHINESE_FONT_AVAILABLE False print(警告未找到中文字体文件PDF 中文可能显示为空白或乱码) except Exception as e: CHINESE_FONT_AVAILABLE False print(f字体注册失败{e})存储文件的主函数def save_document(title: str, content: str, format: str, append: bool False) - str: [WARNING] 禁止直接调用此函数 外部调用必须通过 tool_registry.run_tool(save_document, ...) 执行否则会绕过文件安全校验导致路径穿越风险 # 清洗文件名 unsafe_pattern re.compile(r[\\/:*?|]|\.\.) clean_title unsafe_pattern.sub(_, title) file_path WORKSPACE_ROOT / f{clean_title}.{format} # 文本文件 md / txt 统一UTF-8编码写入 if format in (md, txt): mode a if append else w with open(file_path, modemode, encodingutf-8) as f: if append: f.write( 新增追加内容 ) f.write(content) return f文档保存成功路径{file_path.resolve()}\n格式{format}追加模式{append} # PDF 生成逻辑 elif format pdf: if not PDF_AVAILABLE: raise Exception(PDF导出不可用请安装reportlabpip install reportlab) if not CHINESE_FONT_AVAILABLE: print(警告中文字体未注册PDF将使用默认英文后备字体中文会显示为乱码/方块) font_name Helvetica # 后备方案不报错但中文乱码 else: font_name SimSun # 已注册成功正常显示中文 c canvas.Canvas(str(file_path), pagesizeA4) c.setFont(font_name, 12) _, height A4 # PDF文本分页简单处理 x, y 50, height - 50 line_height 18 # PDF文本换行处理 max_chars_per_line 40 lines [] for paragraph in content.split(\n): wrapped textwrap.wrap(paragraph, widthmax_chars_per_line) lines.extend(wrapped if wrapped else []) for line in lines: if y 50: c.showPage() y height - 50 c.drawString(x, y, line) y - line_height c.save() return fPDF文档保存成功路径{file_path.resolve()}将文件工具注册到注册中心tool_registry.register(save_document, save_document, file_security_check)3. 改造 generate_response 主循环我们之前增添工具注册中心后为了让后续调用都通过工具注册中心还必须调整原本的generate_response主循环核心改动如下删除原硬编码 available_tools 字典原有availabe_tools字典删除# 定义可用工具映射 available_tools { query_knowledge_base: query_knowledge_base, web_search: web_search }工具执行统一调用 tool_registry.run_tool()要求 Agent 工具执行统一调用tool_registry.run_tool()且简化工具权限拦截逻辑复用注册中心安全校验。更改代码块判断工具是否在可用列表中如果仍在要如何进行安全校验更改后如下if not is_tool_allowed: tool_result f【工具调用失败】你没有权限调用 {func_name} 工具。 else: # 统一走注册中心执行工具自动安全校验异常捕获 tool_result tool_registry.run_tool(func_name, **func_args) print(f工具执行结果{tool_result})然后将之前的web_search的 403 熔断代码继续写在上一段代码的后面chat_history.append之前if func_name web_search: if (403 in tool_result) or (百度反爬拦截 in tool_result): current_tools [t for t in current_tools if t[function][name] ! web_search] print(⚠️ [系统干预] web_search 工具调用失败疑似反爬已没收 web_search 权限。)4. 测试测试一下我们的改动和新添加的工具是否好用。可以依次测试是否仍能调用 RAG 和 web_search 工具请先调用 query_knowledge 和 web_search 工具帮我查询……调用 save_document 后是否能存储基础 txt/md 文件请将刚才的结果总结为报告并分别存储为 txt/md 格式的文件。调用 save_document 后是否能在已有文件的基础上追加新内容再帮我查询……并将结果添加到刚才的报告中。调用 save_document 后是否能生成 PDF 文件请把刚才的报告存储为 PDF 格式。理论上到这里应该都是 OK 的祝大家好运二、本篇总结 下期预告本篇我们基于工具注册中心完整落地了 File I/O 文件工具。通过固定工作目录、正则非法字符过滤、路径溯源校验彻底防御了路径穿越安全风险同时适配 TXT、MD、PDF 三种主流文档格式支持文本覆盖与追加写入解决了中文编码、PDF 中文乱码、超长文本分页等实操问题让 Agent 可以自主完成信息整理、报告生成、本地持久化存储具备了完整的内容输出能力。目前 Agent 已具备检索、联网、文档输出能力但仍无法对接结构化数据资源。下一关我们将继续拓展工具生态落地 Text-to-SQL 数据库查询工具让 Agent 拥有读取、解析、查询业务数据库的能力打通智能体与结构化业务数据的壁垒。h5打开以查看