我做开源系列这些年,见过不少朋友拿着成熟的 AI 提示流编排器跑 Demo,跑完都会说一句:很顺滑,但然后呢?前八篇文章把节点调度、上下文管理、模板渲染这些骨架讲完了,可大家真正想要的是让大模型把事办成,而不是让它多说几句漂亮话。比如用户丢来一句"帮我把下载目录里的文件按类型整理好",以前的编排器只能让模型回一段操作建议,然后就没下文了。所以从第 9 篇开始,我的重心明确转向给大模型装上手和脚:在编排器里加入 Agent 节点,配合一套完整的 Tools 工具调用体系。这篇文章不绕弯子,直接讲清楚数据结构怎么设计、Agent 循环怎么跑、工具怎么注册和管控,以及一次端到端文件整理实战的完整过程。如果你正在折腾 Agent 开发,这篇应该能给你一份能直接抄作业的落地参考。
1. 为什么需要"手和脚":从对话模型到执行体的跃迁
1.1 大模型的能力边界:语言很强,行动为零
大模型从本质上说是"语言引擎",它做的是概率预测,输出的是下一段最合理的文本。这意味着它有三个硬伤:
- 知识截止到训练时间,拿不到实时数据。问它"上海现在气温多少",如果训练集里根本没有今天的数据,它只能凭记忆编一个数,而且大概率已经过时。
- 没有网络访问能力。"帮我打开某个网址获取标题",它做不到,因为模型本身不持有任何网络请求通道。
- 没有文件系统和执行环境。"统计我本机某个日志文件里 error 出现的次数",它根本看不到你的硬盘,更别提删改文件。
这是很多人在做 AI 应用时的第一课:模型再聪明,也只是个大脑。大脑负责思考,但真正要让任务落地,必须给它配肢体。我的做法就是在提示流编排器里引入两个东西:Agent 节点扮演"神经中枢",负责调度、决策、调用;Tools 工具调用体系扮演"手和脚",负责执行真实的操作。
可能有人会说,市面上不是已经有 Function Calling 了吗?这确实是一条路,但真正做业务落地时,它撑不起复杂场景。
1.2 为什么裸用 Function Calling 还不够
OpenAI 和国内几家主流模型都支持函数调用,模型会按我提供的参数描述输出一段结构化的调用意图。单看一次调用,它确实能用。但真实业务从来不是"一次调用"能解决的。
举个实际场景:根据订单号查询物流,再生成一段异常提醒。这个任务至少要两次工具调用:第一次查订单对应的物流单号,第二次用物流单号查轨迹。如果第一次调用返回的运单号需要作为第二次的输入,谁来保存这个中间状态?模型接口本身不负责这件事,调用方必须自己管理。
再往深一层看,复杂任务还需要条件分支:如果订单状态是"已签收",就不要再生成提醒;如果遇到超时异常,要不要重试?这些流程控制,函数调用接口天然不具备。
还有工程上的问题:工具统一注册谁来管?多个节点要用同一个工具,怎么避免重复实现?工具超时、失败重试、审计日志,这些系统级能力函数调用也不管。
所以我的结论是:函数调用只是"模型输出参数"的一小步,真正重要的是一整套编排体系。编排器要接管循环、状态、分支、容错,把模型的一次次思考变成一条可控的执行链。
1.3 编排器在 Agent 生态里的三层分工
我最终把编排器拆成三个层级:
- 流程编排层:Flow 把一次完整任务拆成节点,节点之间传参、跳转、合并都由编排器管。
- 模型调度层:Agent 节点内部完成"思考-行动-观察"循环,调用大模型生成决策和参数。
- 工具执行层:Tools 注册表维护所有可调用工具,执行前做参数校验和安全检查,执行后返回标准化结果。
如果把大模型比作大脑,Agent 节点就是连接大脑和四肢的运动神经,Tools 就是真正干活的肌肉。神经传递指令,肌肉执行动作,两者配合好坏,直接决定这套 Agent 系统是真的能干活,还是只是看起来很智能。
这三层既相互独立又彼此咬合。工具层不关心你用的是 GPT 还是自家微调模型,Agent 层不关心工具内部实现,只管按契约调用。这样设计的好处是,以后想换一个更好的模型,或者加一批新工具,都只需要动局部配置。
2. 核心对象设计:Flow、Node、ToolContract 三个主角
2.1 数据模型总览
做编排器之前,我先把数据模型定清楚。后面所有逻辑都围绕这三个对象展开:
@dataclass class FlowContext: """一次流程执行的全局上下文""" inputs: dict # 流程入口参数 variables: dict # 节点间共享变量 execution_id: str # 每次运行的唯一ID,用于审计日志 current_node: str = "" # 当前执行到哪个节点 @dataclass class BaseNode: id: str # 节点唯一ID,例如 "agent_main" name: str # 节点显示名 node_type: str # agent / llm / tool / condition @dataclass class AgentNode(BaseNode): model: str # 模型标识,例如 "qwen-plus" system_prompt: str # Agent 的系统提示词 tools: list[str] # 允许本节点调用的工具名列表 max_iterations: int = 5 # 思考-行动循环的最大轮数 @dataclass class ToolContract: name: str # 工具名,唯一 description: str # 工具描述,给模型看的说明书 parameters: dict # 参数 JSON Schema handler: Callable # 真正执行工具的函数 timeout: int = 30 # 执行超时,单位秒FlowContext 好理解,就是一次运行里所有节点共享的"公文包"。真正花了我最多时间设计的是 ToolContract,它决定了模型能不能准确调用工具、调用得对不对。
2.2 ToolContract:给工具写一份好说明书
工具名和描述是模型做决策时的核心依据。名字要短而准确,描述要足够详细。我见过不少项目把工具描述写成一两句废话,模型自然用错。来看我项目里的一个例子:
@tool_registry.register( name="file_list", description=( "列出指定目录下的所有文件及子目录,返回每个条目的名称、类型和大小。" "适用于任务开始前了解目录结构。注意:不接受相对路径,必须传入绝对路径。" ), parameters={ "type": "object", "properties": { "dir_path": { "type": "string", "description": "要列出的目录绝对路径,例如 /home/user/downloads" } }, "required": ["dir_path"] } ) def file_list(dir_path: str): ...一条好描述应该回答三个问题:工具做什么、什么时候用、失败条件是什么。一个坏描述和好描述的差距,直接影响模型的误调用率。我做过一组对比实验,同样一个 file_move 工具,描述从"移动文件"改成上面这种带场景和失败条件的描述后,模型在文件整理任务里选错工具的次数下降了一半以上。
2.3 节点状态与全局上下文流转
编排器和"调一次模型接口"最本质的区别,就是上下文可以在节点之间流动。在我的设计里,FlowContext.variables 是全局共享存储,上游节点把结果写进去,下游节点读取。
一个典型流程:
- 节点 A 是 tool 类型,调用 http_request 查询订单详情,把订单号写入 variables。
- 节点 B 是 agent 类型,它的 system_prompt 里注入"订单号是 {variables.order_id}",让模型基于真实订单号继续决策。
工具执行完的结果,同样会写回全局上下文。这样每一步的输出都可以被后续节点引用,也可以被审计模块记录。因为所有中间状态都在 FlowContext 里,出问题的时候我能直接导出一整条执行轨迹,而不是对着黑盒猜。
3. Agent 节点的内部循环:从"思考"到"行动"的完整链路
3.1 Agent 节点运行时状态机
Agent 节点的核心是一个状态机,我的实现里包含五个状态:
| 状态 | 动作 | 说明 |
|---|---|---|
| RECEIVE | 接收输入 | 从 FlowContext 拿到上游数据和工具列表 |
| THINK | 调用模型 | 把系统提示词、历史记录、工具清单发给模型 |
| ACT | 执行工具 | 解析模型输出,提取工具名和参数并调用 |
| OBSERVE | 观察结果 | 把工具返回结果追加到对话历史 |
| FINISH | 结束循环 | 模型输出结束标记或超出迭代上限 |
整个流程是按这个顺序循环的:RECEIVE 进来之后,THINK -> ACT -> OBSERVE 不断转圈,直到模型明确说"任务完成",或者轮数耗尽。这个循环就是大名鼎鼎的 ReAct 思想的一个具体落地。
3.2 工具调用参数提取的两条路线
模型思考之后,必须产出一个能被编排器解析的结果。我有两条路线:
第一条是走模型原生的 Function Calling。在请求参数里传 tools 数组:
{ "tools": [ { "type": "function", "function": { "name": "file_move", "description": "将文件从源路径移动到目标路径", "parameters": { "type": "object", "properties": { "src_path": {"type": "string"}, "dest_path": {"type": "string"} }, "required": ["src_path", "dest_path"] } } } ] }这种方式的优点是模型输出已经是结构化 JSON,解析成本低。缺点是它对模型有强依赖,不是所有模型都支持这么完整的 Function Calling 协议。
第二条是通用文本模型也能跑的方案:约定模型必须输出一个 json 代码块,我再用解析器提取。像这样:
import json, re def extract_tool_call(text: str): match = re.search(r"```json\n(.*?)\n```", text, re.DOTALL) if not match: return None try: return json.loads(match.group(1)) except json.JSONDecodeError: return None这条路看起来简陋,但胜在通用。我公司内部有一个基于开源模型微调出来的模型,不支持原生 Function Calling,我用这第二种方案照样让 Agent 跑起来了。所以我的建议是:优先支持方案一,同时保留方案二作为降级通道。
3.3 循环终止条件与最大迭代保护
给模型装上手脚之后,最怕的不是它不干活,而是它失控。一个 Agent 如果进入死循环,会一直调用工具,既烧钱又危险。我的编排器设置了四道保险:
- 正常结束:模型输出
<END>标记或"任务完成",表示它认为目标已达成。 - 最大迭代:默认 max_iterations=5,超过直接停止并返回当前状态。
- 结果收敛:连续两轮观察结果完全一致,且没有新的工具调用,说明模型陷入了重复,强制退出。
- 异常熔断:同一个工具和参数连续失败 2 次以上,不再允许重复执行,直接进入人工复核流程。
这些保护代码量不大,但价值极高。没有它们,Agent 就是一个没有刹车的车。我建议这些参数全部暴露成节点配置,不同任务给不同的上限。文件整理这种需要多次移动文件的场景,我会把 max_iterations 调到 10 到 15;而一些只查询不满意的简单场景,3 次就够。
4. Tools 工具调用体系的实战设计:插件化、安全与控制
4.1 内置工具与自定义工具的注册机制
工具体系我设计成插件式,核心是一个注册表,任何函数只要注册进去就能被 Agent 调用:
class ToolRegistry: def __init__(self): self._contracts = {} def register(self, contract: ToolContract): self._contracts[contract.name] = contract def get(self, name: str) -> ToolContract: return self._contracts[name] def call(self, name: str, params: dict) -> dict: contract = self.get(name) return execute_with_guard(contract, params)项目里我内置了这组工具,基本覆盖了大部分常见需求:
| 工具名 | 功能 | 备注 |
|---|---|---|
| http_request | 发起 HTTP 请求,返回状态码和响应体 | 支持 GET/POST,自动超时 |
| file_list | 列出目录内容 | 必须传绝对路径 |
| file_read | 读取文本文件 | 限定在沙箱目录内 |
| file_write | 写文本文件 | 限定在沙箱目录内 |
| file_move | 移动文件或目录 | 目标存在时失败,不覆盖 |
| file_delete | 删除文件或目录 | 单独工具,受沙箱限制 |
| md5_checksum | 计算文件哈希 | 用于重复文件识别 |
| calculator | 安全四则运算 | 只允许数字运算符 |
| now_datetime | 获取当前时间 | 解决模型时间盲区 |
自定义工具更简单,照着 2.2 节的装饰器写一个函数,注册进去,Agent 下轮就能用。工具体系的价值靠数量堆不出来,靠的是契约清晰。
4.2 工具执行器的安全边界
工具调用体系里安全是头等大事。给模型装上手脚,意味着它可以执行真实操作。如果没有任何防护,一个提示词注入就能让 Agent 删除服务器文件。我的安全设计分四层:
第一层是路径沙箱。所有文件类工具在真正执行前都要做路径校验,防止路径拼接越权:
def _validate_path(path: str, sandbox: str) -> str: abs_path = os.path.abspath(path) abs_sandbox = os.path.abspath(sandbox) if not abs_path.startswith(abs_sandbox + os.sep): raise PermissionError(f"路径越界,禁止访问沙箱外文件: {path}") return abs_path第二层是命令白名单。如果要有 shell 执行能力,绝不能开放任意命令。我做过一个 shell_exec 工具,但只允许 ls、df、du 这类只读命令,而且参数数量受限。像删除、格式化、重定向这类高风险操作,一律交给专门的工具函数处理,这样每一步都有日志可查。
第三层是网络出口限制。http_request 这类网络工具必须配置目标地址黑名单,至少封掉内网地址段。否则一旦提示词被注入,模型可能把请求发到内网的管理接口上。
第四层是超时控制。每个工具都有独立的超时上限,我用线程池做一个简单的带超时执行器:
from concurrent.futures import ThreadPoolExecutor, TimeoutError def execute_with_timeout(fn, timeout=30): with ThreadPoolExecutor(max_workers=1) as pool: future = pool.submit(fn) try: return future.result(timeout=timeout) except TimeoutError: return {"status": "error", "error": f"tool timeout after {timeout}s"}这四层必须全部开启,缺一不可。哪怕只是自己在本地玩,也建议都配上,因为模型的输出是不可预测的。
4.3 工具返回结果的标准化与错误处理
工具执行完,不是把原始返回值直接丢给模型就行,必须统一包装。我定的标准结构是:
{ "status": "ok", "action": "file_list", "params": {"dir_path": "/home/user/downloads"}, "result": "共发现 12 个文件,其中图片 4 个,文档 5 个……", "error": null, "duration_ms": 18, "truncated": false }这个结构里,params 和 duration_ms 是我后来加的。params 用于审计"模型到底传了什么参数",duration_ms 用于性能分析。真正重要的是 status 和 error 的语义:当 status 为 error 时,错误信息会被完整回传给模型,让模型看到失败原因,再决定下一步。
这一点是工具体系里最容易被忽略但最值钱的机制。举个例子,Agent 调用 file_move 想把文件移动到 images 目录,但目标位置上已经存在同名文件,handler 返回"eexist: destination path already exists"。模型看到这个错误后,有能力自行决策:要么换一个目标文件名,要么先调用 file_delete 再重试,要么直接告诉用户存在同名文件请求确认。这种自主纠错能力让整套系统显得非常"智能",但它的基础只是"错误信息可读"这一件事。所以我强烈建议:工具的错误信息别写"operation failed"这种废话,要写出具体原因和上下文。
5. 端到端实战:让 Agent 自己动手完成一次批量文件整理
5.1 场景设计与工具集准备
理论讲再多,不如跑一个真实任务。我选了一个特别常见的场景:整理下载目录。假设 /home/user/downloads 下面堆了 12 个文件,包括几张 jpg、一个 zip 压缩包、几个文档,还有两个重复的 PDF。任务要求是:
- 列出目录下所有文件。
- 按扩展名分类:图片进 images、文档进 docs、压缩包进 archives。
- 用 md5 找出重复文件,两个重复文件只保留体积较大的那个。
- 最后在目录下生成一份 report.txt,写明整理过程。
这个场景用到的工具有 file_list、file_move、file_delete、md5_checksum、file_write,全部来自内置注册表,不需要写一行新代码。
5.2 Flow 配置和 Agent 提示词
我直接在流程配置里定义一个 Agent 节点:
{ "id": "file-organizer-flow", "nodes": [ { "id": "agent_main", "type": "agent", "model": "gpt-4o-mini", "system_prompt": "你是一个文件整理助手。你的任务是整理指定目录,必须按以下步骤执行:1. 先调用 file_list 查看目录内容;2. 根据扩展名设计分类计划;3. 使用 file_move 移动文件到对应子目录;4. 使用 md5_checksum 检测重复文件,保留较大者;5. 调用 file_write 生成 report.txt。执行过程中,如果某一步失败,仔细观察错误信息并修正后重试。所有路径必须使用绝对路径。", "tools": ["file_list", "file_move", "file_delete", "md5_checksum", "file_write"], "max_iterations": 15 } ] }提示词里"按步骤执行"这句话很关键。模型在一堆工具面前容易自由发挥,给一个显式的执行顺序能显著降低它的迷茫感。我还有一个经验:不要用"你可以这样做"这种语气,要用"必须按以下步骤执行"。语气越明确,模型越不会发挥。
5.3 运行结果与调试过程
下面是其中一次运行的核心轨迹,我做了格式化处理:
[agent_main] 开始执行 [agent_main] THINK: 需要先了解目录结构 [tool] file_list(dir_path="/home/user/downloads") -> ok, 12 items [agent_main] THINK: 准备按类型分类,先创建三个目录 [tool] file_move(src_path="/home/user/downloads/photo1.jpg", dest_path="/home/user/downloads/images/photo1.jpg") -> ok [tool] file_move(src_path="/home/user/downloads/report_final.pdf", dest_path="/home/user/downloads/docs/report_final.pdf") -> ok [agent_main] THINK: 发现两个相同命名的pdf,需要比对md5 [tool] md5_checksum(file_path="/home/user/downloads/docs/report_final.pdf") -> ok [tool] md5_checksum(file_path="/home/user/downloads/docs/report_copy.pdf") -> ok [agent_main] THINK: 两份文件md5一致,删除其中一份 [tool] file_delete(file_path="/home/user/downloads/docs/report_copy.pdf") -> ok [agent_main] THINK: 生成整理报告 [tool] file_write(file_path="/home/user/downloads/report.txt", content="...") -> ok [agent_main] 输出结束标记,任务完成这次跑得挺顺畅,但第一次跑的时候根本不是这样。第一次运行里,Agent 试图把 images 目录的目标路径写成相对路径"images/photo1.jpg",结果是我的路径校验直接拒绝了。从那之后我把 file_move 的参数描述里加了一句"目标路径必须是绝对路径,包含完整目录前缀,例如 /home/user/downloads/images"。加了这句之后,同样的任务连续跑三次都没有再出现相对路径问题。经验就是:模型犯的运筹类错误,很多时候是工具描述给的信息不够,而不是模型不聪明。
6. 落地上线后最该记牢的四个坑
6.1 参数幻觉:模型会编造不存在的参数
Agent 在调用工具时,偶尔会输出参数 schema 里根本没定义的字段。比如我在文件整理任务里遇到的,模型调用 file_move 时传了一个"overwrite": true,但我的参数 schema 里根本没这个配置。如果不校验,这一步就会被静默忽略,模型还以为自己开了覆盖权限,后续决策全部建立在幻觉上。
解决办法是严格校验。在调用 handler 之前,用 jsonschema 校验参数:
import jsonschema def validate_params(contract, params): try: jsonschema.validate(instance=params, schema=contract.parameters) return None except jsonschema.ValidationError as e: return f"参数校验失败: {e.message}"一旦校验失败,直接给模型返回错误信息:"overwrite 不是合法参数,可用参数为 src_path、dest_path"。模型看到这个错误,下一轮通常就会修正。这一步把参数幻觉从"静默错误"变成了"可恢复错误"。
6.2 工具返回过长,把上下文撑爆
最让我头疼的坑之一是工具结果太长。文件读取类工具尤其危险,如果 Agent 好奇地读了一个几百行的日志文件,再配合多轮循环,上下文窗口很容易被撑爆。上下文一超,模型的表现是灾难性的,要么开始重复,要么输出格式变畸形。
我的方案是双管齐下。一方面,工具结果在写回对话历史之前做截断,超过 1500 字符的部分用"[已截断,剩余 N 字符]"替代。另一方面,针对超大文件提供细粒度读取工具 file_tail,支持按行号和行数读取,模型先用 file_tail 读取文件头部了解结构,再决定是否读后续行。这样就避免了一次性把整块内容塞进上下文。
6.3 循环失控:模型会在错误里打转
Agent 连续多次调用同一个工具、传同样的参数,每次结果都一样,但它还是要重试。最常见的原因是权限错误或者路径不存在。这种死不认错的 Agent 很让人崩溃。
我做了三层防御:
- 工具调用指纹记录。把"工具名 + 参数JSON"哈希成一个指纹,同一个指纹失败两次后,该指纹直接熔断。
- 连续无效动作检测。连续 3 轮没有产生状态变化(没有工具调用成功、没有变量更新),强制结束。
- 最后兜底就是 max_iterations。设置成多少要看场景,文件整理这种任务我放到 15,简单查询放到 5。
熔断机制上线之后,我再也没见过 Agent 卡死一个接口反复重试的情况。这比单纯限制迭代次数优雅得多,因为它在保护运行成本的同时,保留了正常的重试空间。
6.4 工具命名与描述的"措辞工程"
工具描述这种东西,看起来谁都能写,但写好非常难。我的经验是:工具描述是 Agent 的"操作手册",它的质量直接决定了模型的行为边界。命名上,我倾向于动词开头加明确对象,file_move、http_request 这种都比 do_action 强一百倍。描述上,一定要写清楚三个点:工具解决什么场景、什么时候不该用、失败条件是什么。
拿 file_move 举例,项目里这个工具的描述最后迭代成了这样:
将文件或目录从源路径移动到目标路径。用于文件分类、归档、整理目录等场景。目标路径必须位于沙箱目录内。如果 dest_path 已存在同名文件或目录,直接返回错误,不会覆盖。源文件移动后不再存在于原位置。
加了"不会覆盖"之后,模型就很少再试图用一个 move 来"顺便覆盖"掉旧文件,而是会主动考虑先用 file_delete。这种措辞细节,是模型实际跑出来后一点点调出来的。我的建议是,每新增一个工具,一定要做三组冒烟测试:正常的调用、参数错误时模型的纠错、以及被明确阻止的场景下模型的行为。测试通过之前,这个工具不要暴露给任何流程。
这个工具调用体系我前后迭代了四五轮,最后沉淀下来最重要的结论其实就一句话:大模型的手脚不是越长越多越好,而是每一个动作都清晰、可控、可回退。如果你也在做一个带工具的 Agent,我建议先把工具契约和错误回传这两件事做扎实,再回过头去调提示词。工具是手脚,但神经和肌肉的配合,得靠一遍遍实测磨出来。