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_ok=True, parents=True)注意:"./agent_workspace"的.指命令行启动程序时的工作目录,不是 py 脚本文件所在目录。如果你在别的文件夹运行脚本,文件夹会生成到你敲命令的位置,而不是代码文件旁边。
如果想要文件夹固定生成在 py 脚本同目录,改成:
WORKSPACE_ROOT = Path(__file__).parent / "agent_workspace" WORKSPACE_ROOT.mkdir(exist_ok=True, parents=True)一般选择将其固定到代码文件旁边,不管是人找起来还是 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("警告:未安装reportlab,pdf格式导出功能不可用,执行 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, mode=mode, encoding="utf-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导出不可用,请安装reportlab:pip install reportlab") if not CHINESE_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 - 50 line_height = 18 # PDF文本换行处理 max_chars_per_line = 40 lines = [] for paragraph in content.split("\n"): wrapped = textwrap.wrap(paragraph, width=max_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 f"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(),且简化工具权限拦截逻辑,复用注册中心安全校验。
更改代码块:判断工具是否在可用列表中,如果仍在要如何进行安全校验,更改后如下:
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打开以查看