提示流编排器这个开源项目做到第九期前面几期我们已经把提示词模板、流程编排、变量串联这些基础能力铺得差不多了。但这段时间越用越觉得不对劲只靠“写提示词 拼流程”大模型本质上还是个“只会动嘴”的组件。你让它算一道复杂的数学题它可能会一本正经地给你一个错误答案你让它查昨天线上报错日志的统计它连日志文件在哪儿都不知道。所以这一期我决定动真格的了——给编排器加上 Agent 节点和 Tools 工具调用体系说白了就是给大模型装上手和脚让它不只会“说”还能“做”。这篇文章会从整体设计思路开始把工具注册、函数调用协议、Agent 循环、上下文回填、常见坑全部过一遍适合正在折腾 AI 应用编排、想给自家项目接工具调用的开发者参考。1. 整体设计与思路拆解1.1 为什么我决定在编排器里加 Agent 节点先理清一个概念普通的大模型调用是你问一句、它答一句整个过程中模型没有任何“自主行为”的能力。哪怕你提示词写得再花哨它也只能在你给定的文本上下文里生成回复没法主动去查数据库、调接口、跑命令。这就是所谓的“嘴强王者”——嘴上什么都会实际一步也动不了。而 Agent 节点要解决的核心问题就是把“行动能力”注入到这个闭环里。它的运行逻辑不是一次性生成而是一个循环模型观察当前情况 - 决定要调用哪个工具 - 执行工具拿到结果 - 把结果喂回去 - 模型继续推理直到它认为不需要再调用任何工具给出最终回答。你可以把这个过程理解成真人处理工作老板问“这周线上服务为什么变慢”你不会直接拍脑袋回答而是先看监控、查日志、问同事综合所有信息之后才给出结论。Agent 节点就是让大模型走一遍这个“先调研再发言”的流程。在编排器里加上它之后原来那些只能做文本转换的静态流程就变成了能根据实际输入动态决策的智能流程。这个能力在真实场景里特别重要。比如做一个“客服工单自动分类 舆情摘要”的编排用户提交一个工单Agent 先调用检索工具查历史相似工单再调用摘要工具压缩内容最后根据分类规则库判断优先级。这一整套动作如果用传统 DAG 流程去写每一分支都要人肉判断写出来又长又脆用 Agent 节点大模型自己判断该走哪条路流程天然就简洁了。1.2 Tools 工具调用体系设计的边界动手写代码之前我先把 Tools 体系的设计边界定下来。所谓边界就是回答清楚三个问题哪些能力可以做成工具、工具之间怎么隔离、Agent 节点如何知道该用哪个工具。第一能做成工具的能力一定要满足“边界清晰、输入输出可结构化”。比如“根据城市名查天气”“计算两个日期之间的工作日天数”“查询 ES 中最近一小时的错误日志数量”这种输入输出都很明确的能力适合做工具。反过来“写一段打动人的文案”这种模糊任务就不适合因为它太依赖模型本身做成工具反而限制发挥。第二工具之间必须隔离。每个工具是一个独立函数有自己的参数校验、异常处理、超时控制。工具 A 崩了不能影响工具 B更不能拖垮整个 Agent 循环。我在实现里给每个工具包了一层 executor统一处理超时和重试这样工具作者只关心业务逻辑不用操心这些基础设施问题。第三Agent 节点要按需选择工具。不是所有工具都能给同一个 Agent 用比如涉及删除数据库记录的危险工具肯定不能和普通查询工具放在同一个白名单里。所以我在 Agent 节点的配置里加了一个tools字段只挂载指定的工具集合从机制上防止模型乱调。选定这些边界之后整体架构就清晰了提示流编排器负责流程调度Agent 节点内部跑模型决策循环ToolRegistry 统一管理工具注册和分发每个工具都是独立模块。这个分层的好处是将来新增工具不用改编排器核心代码注册一下就能用。1.3 核心设计目标可控、可观测、可复用设计这套体系时我给自己定了三条硬指标也就是可控、可观测、可复用现在回头看这三条监督了整个实现过程。可控是第一位的。Agent 本质上是不可预测的模型下一步调什么工具说实话没人能 100% 预判。所以不能让它变成脱缰野马。我通过白名单工具列表、最大迭代次数、每一步的工具结果都经过执行器返回这三个手段把不确定性关在笼子里。白名单解决“能用什么”最大迭代解决“跑多久”执行器解决“结果可不可信”。可观测也很关键。Agent 循环中间发生的事如果不记录下来出了问题只能干瞪眼。我在节点里维护了一个steps数组每一步存下模型思考过程如果有、调用的工具名、传入参数、工具返回结果。这样 UI 上可以完整展示 Agent 的“心路历程”调试时也能还原现场。可复用则是从工程角度考虑的。ToolRegistry 设计成全局单例工具注册后可以被任意 Agent 节点引用Agent 节点配置好的 system_prompt、工具集合、迭代上限也能作为模板被多个流程共用。这套做法跟写服务类似先定义好接口再各自实现最后自由组装。2. 核心细节解析与实操要点2.1 Tools Schema大模型唯一能读懂的“接口文档”工具调用体系里最容易被低估的环节是工具定义本身也就是我们常说的 Tools Schema。模型不像人它不会去看你的函数注释或 README它唯一能理解的就是你通过 API 传给它的 JSON 结构。这个结构写得不清楚后面全都白搭。我沿用的是 OpenAI 引入的函数调用Function Calling格式现在国产和开源的模型大多也兼容这个协议。一个工具定义包含name、description、parameters三个核心字段其中parameters是标准的 JSON Schema。看起来简单实际写起来踩坑不少我整理了下面这张表Schema 字段作用我的建议name工具唯一标识全部小写用下划线分隔单词不要冒号、空格description工具行为说明写清楚“何时用、用它干嘛、结果长什么样”越具体越好parameters.type参数结构类型几乎都是 object别整花活properties每个参数的详细定义每个参数都要写 type descriptionrequired必填字段列表缺一个都可能导致模型不调用这个工具描述中的示例值帮助模型理解格式在 description 里直接举例比如日期写成 “2024-06-01”按理说 description 越长对模型越友好但太长也不行。模型对超长描述的注意力会衰减重要信息反而会被淹没。我实践下来觉得一个工具的 description 控制在两到四句话最合适第一句说明工具能力第二到三句说明适用场景最后一句如果有特殊格式要求就明确给出示例。这里有个很容易翻车的细节参数描述里必须带上单位、格式、边界。比如写一个“查询天气”的工具参数是城市名description 里如果不写“城市名需为中文城市全称如北京市”而不是“北京”或“BJ”模型传给你的可能就是花式缩写。同理数值型参数一定要写单位不然模型可能把“5 公里”传成 5 米。2.2 工具注册机制与运行时隔离工具定义写完就要解决注册和调度问题。我实现了一个ToolRegistry类核心是一个装饰器让开发者加一个工具只需要写一个普通函数然后在上面挂一个registry.tool装饰器。装饰器背后做这些事解析函数签名和 Schema 做一次一致性检查把函数对象存入注册表同时生成一个内部可调用的函数句柄。这个句柄做的事情包括把 JSON 参数转换为 Python 实参、执行函数、捕捉异常并规范化为统一的返回结构。这么设计的好处是工具作者永远不需要关心 JSON 和 Python 类型之间的转换写业务逻辑就完了。运行时隔离我通过两层实现。第一层是进程内的 Futures 超时控制每个工具调用最多执行 N 秒超时就抛一个标准化错误。第二层是参数校验沙箱工具的入参先经过 JSON Schema validate再进业务函数从源头挡住模型传过来的非法参数。有人可能会问是不是每次都要做这么重的隔离我建议至少要保留超时控制。因为模型可能会对一个工具反复调用某个第三方 API 万一卡住了超时机制能保证整个 Agent 循环不是无限等下去。为了省事不做超时结果就是生产环境里一个慢 API 挂起整个编排流程这个亏我吃过印象太深了。2.3 同步工具与异步工具的取舍工具代码写法没有统一标准你的工具可能是同步函数直接请求一次 HTTP 接口也可能是异步函数要走 asyncio 的协程。提示流编排器本身是异步框架所以我在执行层做了兼容。对于同步工具直接用asyncio.to_thread扔进线程池避免阻塞事件循环。对于异步工具直接放到当前事件循环里await。ToolRegistry 里用一个is_async标志区分这两类执行器拿到工具后自动选择路径开发者不用自己判断。不过这里有个坑我先提前说一下不要在一个异步流程里混合大量同步阻塞调用。每个同步工具调用都占一个线程池的线程如果 Agent 一轮循环里调了五个同步工具线程池配额可能被打满反而拖慢整体。我的做法是让工具尽量别做重 I/O能用异步就用异步实在改不了同步的就把线程池的 max_workers 调大一点并且在编排器配置里留好提示。还有超时也要分层。我分成三层网络层超时HTTP client 的 timeout、执行层超时Future 的 timeout、整体 Agent 循环超时节点级别。三层各管一段层级越往上阈值越大。比如 HTTP 超时 5 秒、工具执行超时 10 秒、整个 Agent 循环最多 60 秒。这个分层设计看着繁琐但真正排查线上问题时它能帮你快速定位到底卡在哪一层。2.4 工具调用循环模型与工具的“对话协议”理解 Agent 节点的运行机制最核心的是搞清楚“模型与工具怎么对话”。我直接用生活例子说明你跟助手说“帮我看看上海明天会不会下雨如果不下的的话帮我订个户外餐厅”。助手不会一步到位它会先调天气工具得到“明天下雨”的结果然后改口告诉你“明天下雨不适合户外建议室内”。这个过程中天气工具的结果就是模型下一步推理的“事实依据”。用协议的语言描述就是这样的循环把用户消息拼上 system prompt整体发给大模型。大模型返回两种可能之一要么是一个文本回复表示它已经能回答要么是若干tool_calls表示它要调用工具。如果是文本回复循环结束。如果是tool_calls对每个调用去 ToolRegistry 里找到对应工具经过参数校验后执行拿到结果。把工具调用的原始请求和工具结果作为一条新的消息追加进对话历史再回到第 1 步。这个循环会一直反复直到模型不再返回tool_calls或者循环次数触顶。很多人第一次写的时候容易漏掉第 5 步也就是“把工具结果追加回对话历史”。如果不做这一步后续的模型请求根本看不到工具执行结果它就只能凭空推理结果必然跑偏。还有一个细节工具调用消息的 role。按 OpenAI 协议的写法模型返回的 tool_calls 跟着一条assistant消息而工具执行结果要包在role: tool的消息里并且每条工具消息要带上对应的tool_call_id。这个 ID 不能随便填模型是靠它把工具结果和之前的调用对应起来的。一旦 ID 对不上或缺失整个对话上下文就乱了。这也算是我踩过的坑后面会在问题排查部分细说。3. 实操过程与核心环节实现3.1 先跑通一个最小可用的 Agent 循环理论讲完直接进入代码。我先给一个最小可复现的 Agent 循环不依赖我的编排器框架纯粹展示核心逻辑。这样大家可以先跑通这个最小闭环再理解我后续在编排器里的封装。# minimal_agent_loop.py import json from typing import Callable, Any # 简化版 ToolRegistry class ToolRegistry: def __init__(self): self._tools {} def tool(self, name: str, description: str, parameters: dict): def decorator(func: Callable): self._tools[name] { name: name, description: description, parameters: parameters, func: func, } return func return decorator def get(self, name: str): return self._tools[name] def schemas(self): return [{name: t[name], description: t[description], parameters: t[parameters]} for t in self._tools.values()] registry ToolRegistry() # 注册一个加法工具示例 registry.tool( nameadd_numbers, description计算两个数字的和参数必须是整数或浮点数。, parameters{ type: object, properties: { a: {type: number, description: 第一个加数}, b: {type: number, description: 第二个加数}, }, required: [a, b], }, ) def add_numbers(a: float, b: float) - float: return a b # OpenAI SDK 兼容的消息格式 def run_agent_loop(user_message: str): messages [{role: user, content: user_message}] max_iterations 5 for step in range(max_iterations): # 1. 请求大模型这里用占位需要替换成实际模型调用 response fake_chat_completion( messagesmessages, toolsregistry.schemas(), ) # 2. 解析返回 if response[finish_reason] stop: return response[content] # 3. 处理 tool_calls assistant_msg {role: assistant, content: None, tool_calls: []} tool_results [] for call in response[tool_calls]: assistant_msg[tool_calls].append({ id: call[id], type: function, function: {name: call[function][name], arguments: call[function][arguments]}, }) # 4. 执行工具 tool registry.get(call[function][name]) args json.loads(call[function][arguments] or {}) try: output json.dumps(tool[func](**args), ensure_asciiFalse) except Exception as exc: output json.dumps({error: str(exc)}, ensure_asciiFalse) tool_results.append({ tool_call_id: call[id], role: tool, content: output, }) # 5. 追加消息并进入下一轮 messages.append(assistant_msg) messages.extend(tool_results) return 达到最大迭代次数强制终止这段代码就是 Agent 节点最核心的骨架。fake_chat_completion可以替换成任何兼容 Function Calling 的模型客户端核心是 messages 的累积和 tool_calls 的解析。先在本地把跑通的感觉找到再往上加工程能力比直接怼一个大而全的框架要舒服得多。3.2 ToolRegistry 与两个真实工具上面的骨架里已经出现了简化版 ToolRegistry下面补充两个真实工具。一个是计算器一个是天气查询。计算器不适合用eval原因太容易注入恶意表达式安全问题很严重。我采用一个只支持四则运算和幂运算的极小解析器或者干脆用正则限制表达式只能包含数字和加减乘除符号。# calculator_tool.py import re registry.tool( namesafe_calculate, description计算数学表达式支持 - * / 和括号。例如 3.5 * (2 4)。, parameters{ type: object, properties: { expression: {type: string, description: 数学表达式仅支持数字、括号和 - * / 符号}, }, required: [expression], }, ) def safe_calculate(expression: str) - float: cleaned expression.replace( , ) if not re.fullmatch(r[0-9\-*/().], cleaned): raise ValueError(表达式包含非法字符) # 不允许连续运算符等情况简单校验后按 AST 执行更安全 import ast node ast.parse(cleaned, modeeval) if not all(isinstance(n, (ast.Expression, ast.BinOp, ast.Num, ast.UnaryOp, ast.Load, ast.Constant, ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Paren)) for n in ast.walk(node)): raise ValueError(表达式含不支持的操作) result eval(compile(node, string, eval), {__builtins__: {}}, {}) return float(result)# weather_tool.py # 简化版以同步 HTTP 请求为例实际应替换为真实天气 API import httpx registry.tool( nameget_weather, description查询中国城市的实时天气和温度。城市名需要中文全称例如传入「北京市」而不是「北京」或「BJ」。, parameters{ type: object, properties: { city: {type: string, description: 城市中文全称如「上海市」}, date: {type: string, description: 查询日期格式 YYYY-MM-DD默认今天}, }, required: [city], }, ) def get_weather(city: str, date: str ) - dict: # 这里用示例接口真实项目中替换为你的天气服务 url https://api.example.com/weather resp httpx.get(url, params{city: city, date: date or }, timeout5.0) resp.raise_for_status() data resp.json() return {city: city, temperature: data[temp], condition: data[condition]}两个工具注册完毕后通过registry.schemas()就能拿到模型需要的工具定义列表。要注意get_weather的说明里我特意写了城市名示例这种“描述里带示例”的做法能显著降低模型传参的错误率比只写一个干巴巴的“城市名称”有效得多这是跟模型对齐数据的微妙之处。3.3 编排器里的 Agent 节点实现骨架跑通、工具就位后就要把这些能力收拢进提示流编排器做成一个正式的 Agent 节点。这个节点跟其他节点比如 Prompt 节点、条件分支节点的对外接口要保持一致输入一个消息对象输出一个消息对象中间过程记到 context 里。# agent_node.py class AgentNode(BaseNode): Agent 节点配置项 - model: 模型标识 - tools: 工具名列表留空代表使用全部已注册工具 - system_prompt: 给 Agent 的系统提示 - max_iterations: 最大循环次数默认 5 - temperature: 采样温度默认 0 run_type agent def __init__(self, node_id, model, toolsNone, system_prompt你是一个智能助手。, max_iterations5, temperature0): super().__init__(node_idnode_id) self.model model self.tools tools or [] self.system_prompt system_prompt self.max_iterations max_iterations self.temperature temperature def build_messages(self, payload: str) - list: messages [{role: system, content: self.system_prompt}] if payload: messages.append({role: user, content: payload}) return messages async def run(self, payload: str, context: dict) - dict: allowed_tools self.tools if self.tools else list(registry._tools.keys()) tools_schema [registry.get(name)[schema] for name in allowed_tools] messages self.build_messages(payload) steps [] for step_idx in range(self.max_iterations): resp await chat_completion( messagesmessages, toolstools_schema, temperatureself.temperature, ) if not resp.get(tool_calls): steps.append({type: final, content: resp[content]}) break assistant_msg {role: assistant, content: resp[content], tool_calls: []} tool_msgs [] for call in resp[tool_calls]: fn_name call[function][name] fn_args json.loads(call[function][arguments] or {}) steps.append({ type: tool_call, step: step_idx, tool: fn_name, args: fn_args, }) # 真正执行时走 executor 的队列这里为演示而简化 tool_exec_result await run_tool_with_timeout(fn_name, fn_args) tool_msgs.append({ role: tool, tool_call_id: call[id], content: tool_exec_result.serialized(), }) steps.append({type: tool_result, tool: fn_name, result: tool_exec_result.data}) assistant_msg[tool_calls] [ {id: c[id], type: function, function: {name: c[function][name], arguments: c[function][arguments]}} for c in resp[tool_calls] ] messages.extend([assistant_msg] tool_msgs) else: steps.append({type: force_stop, reason: max_iterations}) context[agent_steps] steps last_content steps[-1].get(content) if steps else 抱歉未能处理该请求 return {output: last_content, trace: steps}这个节点的run方法是一个完整的 Agent 循环封装其中run_tool_with_timeout就是前面说的执行器负责超时控制、错误规范化和重试。值得一提的是我把每一步的工具名、参数、结果都放进了steps数组并写进 context。这个设计在排查问题的时候帮助特别大哪个工具参数传错了、哪一步结果被截断了打开 trace 一目了然比对着日志猜要高效太多。3.4 工具结果回填与上下文控制工具结果回填是整个 Agent 节点里最容易失控的一环。模型每轮对话携带的上下文是有限的如果工具返回了一个超大的结果比如某次查询返回了几百条日志直接原封不动塞进 messages下一轮请求的 token 数可能就爆了。我的处理思路是三层。第一层是工具返回结果在源头做精简让工具开发者遵循“返回结论和必要明细不返回原始流水”的原则。第二层是在执行器层面对超大结果做截断单个工具结果超过设定阈值我通常设置为 2000 tokens就只保留摘要和数量统计。第三层是整个 Agent 上下文超过预算时把比较早的工具结果消息压缩成摘要文本再放入下一轮请求。这里有一个容易忽略的前提截断可以在工具执行器层做但绝不能篡改已经记录到 trace 里的原始结果。trace 是为了调试和审计原始数据必须保留而喂给模型的消息可以精简。这两份数据要分开存不能图省事合并到一个变量里。我在一开始就把 trace 和 model_messages 分成了两个数据结构后面改策略的时候省了很大力气。4. 常见问题与排查技巧实录4.1 大模型死活不调用工具写好的工具模型就是不调这是最让人血压升高的问题之一。我排查过很多次总结下来就这么几个原因按概率从高到低排。第一个是 tools 参数没真正传进模型请求。很多框架封装了 Chat 接口但忘了透传 tools 字段。检查方式很简单把实际发给模型的消息和参数打印出来看 tools 数组在不在。第二个是工具描述写得让模型觉得“这个问题不需要工具”。比如你明明提供了天气工具但描述写得含含糊糊模型就没把用户问题和工具关联起来。解决方式是对着真实用户问题逐个审视工具描述如果你是模型看到这个描述你会不会想到用这个工具第三个是模型本身不支持 Function Calling。有些只支持文本补全的模型压根不理解 tools 参数它会忽略这个字段直接回复。换成支持函数调用的模型一般立竿见影。第四个是 temperature 太高模型有时候“发挥”太自由不去走工具调用流程。把 temperature 降到 0 能明显提升调用工具的稳定性。4.2 参数频繁传错怎么办模型调用工具的请求格式合规但传的参数就是不对明明是数字类型传成字符串、日期格式五花八门、城市名缩写花式出现。这个问题的根源多半不在模型而在你的 Schema 不够严谨。最有效的改进是把“格式示例”写进参数 description而不是指望模型看过你的类型定义就能猜对。例如日期参数只在类型里写string基本没用要写“格式为 YYYY-MM-DD例如 2024-06-01”。另外在工具函数内部做二次校验是兜底方案校验不通过时返回一个包含具体错误原因的结果让模型自己看错误信息后修正重试。这里要记住Agent 模型有自我纠错能力只要错误信息足够明确它下一次调用往往就能改对。所以不要一旦参数错误就让整个流程崩掉而是要做一个“容错 重试”的闭环。4.3 Agent 死循环的应急处理Agent 循环跑起来了但停不下来或者反复调用同一个工具不推进。前面设计里提过的max_iterations在这里派上用场但光有上限还不够我还会加一个“连续 N 次相同工具调用则强制终止”的规则。具体实现是维护一个最近几步的调用了列表如果连续三步调用的工具名和参数完全一致那基本可以断定模型在空转。这种空转常见于两种情况一是工具返回的结果不理想模型反复尝试同一种方式想拿到不同结果二是模型的 reasoning 能力不足找不到其他可行路径就开始复读。处理方式是提前中断并返回当前已收集的信息同时把“为什么中断”写进 trace。这样用户至少能拿到一个部分结果而不至于整个编排超时。4.4 工具结果太大撑爆上下文前面说了一堆预防措施这里再给一个硬性规则任何工具的结果回填到对话历史之前都要先走一遍 token 估算。估算可以简单按字符数除以 4 近似也可以用分词器精确算总之要有这么一步。如果估算超限最简单实用的策略是把工具结果截断到前面 N 个字符再附上一句“……结果过长已截断需要完整数据请调用查询接口获取”。这样模型至少能判断结果的大致方向不至于完全瞎猜。更进阶一点的做法是在截断前让模型层面的一个小模型生成结构化摘要然后只把摘要回填。我在项目里两种都实现了但默认用的是截断简单、可靠、不引入额外延迟。说到底工具结果回填的原则就一句话给模型刚刚好的信息量既够它推理又不至于淹死。4.5 并行工具调用的坑现在不少新模型支持在一次响应里返回多个 tool_calls也就是并行调用多个工具。这听起来很美好但实现时有几个暗坑。第一个坑是并行执行时的资源竞争。比如同时调两个工具一个特别快一个特别慢如果统一等最慢的会让整个循环卡在那一步。我的做法是对每个工具单独设超时快的结果先记录慢的要么等待属于它的超时阈值要么提前降级为“超时结果”返回给模型不让它阻塞整体节奏。第二个坑是错误隔离。多个工具并行时如果其中一个抛异常不能让它影响其他工具结果的回填。每个 tool_call 的执行结果要独立 catch独立序列化然后分别追加到消息列表里。第三个坑是模型的 tool_call 消息和对应的 tool 结果消息顺序要保持一致ID 必须一一对应。如果并行结果拼接时顺序错乱模型就会混乱。我自己的经验是功能迭代初期先强制串行把所有逻辑跑通之后再放开并行这样排查问题的复杂度会低一个量级。结尾这里我就聊点个人实操感受。做到现在这套 Agent 节点和 Tools 体系我最深的感触是真正难的不是让模型会调用工具而是你如何把工具描述得让模型“一看到就知道怎么用、什么时候用”。在这个系统里工具描述和 schema 的工程质量决定了 Agent 能力的上限。你与其花时间去调各种推理参数不如先把工具描述里有歧义的地方全抠干净。另外建议所有做类似编排器的朋友先从“一个模型 一个工具 一次回填”的最小闭环跑起让它能稳定解决一个最朴素的真实问题再往上面堆工具和分支。越早形成闭环你对 Agent 系统的直觉就越准后面加再多工具也不会乱。