上一关我们完成了 Agent 底层架构的全面重构,搭建了标准化、可扩展的工具注册中心,彻底解决了老旧架构耦合臃肿、难以拓展高阶工具的问题,为后续所有工具开发统一了技术规范。有了稳定的架构底座,我们正式开启 Stage 2 其他进阶工具的落地迭代。
在真实落地场景中,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_ok=True,parents=True)注意:"./agent_workspace"的.指命令行启动程序时的工作目录,不是 py 脚本文件所在目录。如果你在别的文件夹运行脚本,文件夹会生成到你敲命令的位置,而不是代码文件旁边。
如果想要文件夹固定生成在 py 脚本同目录,改成:
WORKSPACE_ROOT=Path(__file__).parent/"agent_workspace"WORKSPACE_ROOT.mkdir(exist_ok=True,parents=True)一般选择将其固定到代码文件旁边,不管是人找起来还是 Agent 找起来都比较方便。
安全校验函数
因为这个工具需要我们存储和修改本地文件,因此加入专属安全校验函数,去检测路径穿越、非法文件名、非法格式等问题。
deffile_security_check(title:str,format:str,**kwargs)->Optional[str]:# 1. 校验文件格式allow_formats={"txt","md","pdf"}ifformatnotinallow_formats:returnf"不支持的文件格式{format},仅允许 txt/md/pdf"# 2. 清洗文件名:过滤非法字符,禁止../ / \等逃逸字符unsafe_pattern=re.compile(r'[\\/:*?"<>|]|\.\.')clean_title=unsafe_pattern.sub("_",title)ifclean_title!=title:returnf"文件名包含非法字符/路径逃逸字符,已自动清洗为{clean_title},请重新调用工具使用安全文件名"# 3. 拼接完整路径,强制校验文件在WORKSPACE_ROOT内file_name=f"{clean_title}.{format}"target_path=(WORKSPACE_ROOT/file_name).resolve()root_abs=WORKSPACE_ROOT.resolve()# 核心防路径穿越:目标路径必须是工作目录的子文件ifnotstr(target_path).startswith(str(root_abs)):returnf"路径逃逸拦截:禁止写入工作目录外路径{target_path},仅可保存至{WORKSPACE_ROOT}"# 校验通过,无报错返回NonereturnNone这就是绑定给 save_document 的前置安全校验函数。
当通过run_tool("save_document", ...)调用工具时,run_tool 会先自动执行这个函数。返回None代表校验放行;返回字符串代表校验失败,直接拦截,不会执行真正的保存函数。
安装 PDF 依赖
如果想让程序有导出 PDF 的能力,需要安装第三方 PDF 生成库。如果不安装程序也可以用,只是不能导出 PDF。
pipinstallreportlabtry:fromreportlab.lib.pagesizesimportA4fromreportlab.pdfgenimportcanvas PDF_AVAILABLE=TrueexceptImportError:PDF_AVAILABLE=Falseprint("警告:未安装reportlab,pdf格式导出功能不可用,执行 pip install reportlab 开启")同时,由于我们希望导出的 PDF 是中文版本,因此需要安装字体库,这里采用 Windows 宋体,字体来源于 Windows 系统字体路径,不需要额外下载字体。
# 注册中文字体(Windows 宋体)try:# Windows 系统字体路径font_path="C:/Windows/Fonts/simsun.ttc"ifos.path.exists(font_path):pdfmetrics.registerFont(TTFont('SimSun',font_path))CHINESE_FONT_AVAILABLE=Trueprint("已加载系统宋体,PDF 中文导出可用")else:# 备选:尝试黑体(微软雅黑)font_path="C:/Windows/Fonts/msyh.ttc"ifos.path.exists(font_path):pdfmetrics.registerFont(TTFont('SimSun',font_path))# 仍注册为 SimSun 名字方便调用CHINESE_FONT_AVAILABLE=Trueprint("已加载微软雅黑作为 PDF 中文字体")else:CHINESE_FONT_AVAILABLE=Falseprint("警告:未找到中文字体文件,PDF 中文可能显示为空白或乱码")exceptExceptionase:CHINESE_FONT_AVAILABLE=Falseprint(f"字体注册失败:{e}")存储文件的主函数
defsave_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编码写入ifformatin("md","txt"):mode="a"ifappendelse"w"withopen(file_path,mode=mode,encoding="utf-8")asf:ifappend:f.write("===== 新增追加内容 =====")f.write(content)returnf"文档保存成功!路径:{file_path.resolve()}\n格式:{format},追加模式:{append}"# PDF 生成逻辑elifformat=="pdf":ifnotPDF_AVAILABLE:raiseException("PDF导出不可用,请安装reportlab:pip install reportlab")ifnotCHINESE_FONT_AVAILABLE:print("警告:中文字体未注册,PDF将使用默认英文后备字体,中文会显示为乱码/方块")font_name="Helvetica"# 后备方案,不报错但中文乱码else:font_name="SimSun"# 已注册成功,正常显示中文c=canvas.Canvas(str(file_path),pagesize=A4)c.setFont(font_name,12)_,height=A4# PDF文本分页简单处理x,y=50,height-50line_height=18# PDF文本换行处理max_chars_per_line=40lines=[]forparagraphincontent.split("\n"):wrapped=textwrap.wrap(paragraph,width=max_chars_per_line)lines.extend(wrappedifwrappedelse[""])forlineinlines:ify<50:c.showPage()y=height-50c.drawString(x,y,line)y-=line_height c.save()returnf"PDF文档保存成功!路径:{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(),且简化工具权限拦截逻辑,复用注册中心安全校验。
更改代码块:判断工具是否在可用列表中,如果仍在要如何进行安全校验,更改后如下:
ifnotis_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之前:
iffunc_name=="web_search":if("403"intool_result)or("百度反爬拦截"intool_result):current_tools=[tfortincurrent_toolsift['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 拥有读取、解析、查询业务数据库的能力,打通智能体与结构化业务数据的壁垒。