ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Agent技能管理框架:从工具调用到工作流编排的实战指南

Agent技能管理框架:从工具调用到工作流编排的实战指南 先说个题外话。我手上接过不少号称“大模型应用”的项目最后落地时十有七八都卡在同一个地方模型很聪明但它不知道该调用哪个工具、以什么顺序调用、参数怎么填。你可以让模型写一首诗、总结一份文档这些都很强可一旦要它去完成“查一下这个订单的物流状态如果三天没更新就自动发一封催件邮件”这种真实任务它就抓瞎了——不是模型不行是它的“技能”组织方式出了问题。我做的这个 agent-skills 项目说白了就是一套给智能体用的“能力管理框架”把 Agent 要执行的各种操作抽象成一个个标准化的“技能”Skill让模型能通过统一的协议去发现、选择、调用、编排它们。这篇内容我尽量讲得实在一点从技能设计的核心思路到代码层面的落地实现再到我把这套系统接进真实业务后踩过的坑都盘一遍。1. 技能设计的本质与核心思路1.1 为什么要给 Agent 单独做一层“技能系统”很多人第一次接触 Agent 开发时第一反应是“把工具描述写进 system prompt 不就行了”确实刚起步时这么干完全没问题。但等你接的工具超过二三十个时问题会像滚雪球一样涌过来提示词越来越长模型在长上下文里迷失反而开始忽略关键工具不同工具的调用格式五花八门有的走 REST API有的要连数据库有的是内部的 Python 函数模型根本搞不清哪个该用哪种协议更麻烦的是你加了新工具后得回去改 prompt一旦改得不小心把其他工具的语义描述挤掉了线上就出幺蛾子。我开发的这套 agent-skills 系统核心思路就是把这堆混乱的东西收编成一套“可注册、可发现、可编排”的技能体系。每个技能拥有独立的描述、参数结构、执行入口和错误处理策略模型不再需要阅读海量的工具说明而是通过一个结构化的“技能清单”按需查找。这个思路的本质就是把传统的“工具集成”升级成“能力治理”。1.2 技能粒度的设计原则刚开始设计技能时最容易犯的错误是粒度把握不好。我第一版设计里把“发送HTTP请求”做成了一个通用技能想着“一个技能搞定所有网络操作”。实际跑起来简直是一场灾难模型参数猜来猜去不是漏了 headers 就是搞错请求体格式而且模型完全不知道“这个接口需要什么鉴权方式”。后来我总结出一条经验技能粒度应该对标“业务动作”而不是“技术动作”。同样是“获取订单信息”如果只做“HTTP GET 请求”这个技能模型需要自己拼 URL、传 header、解析返回体这对模型的推理负担太重了。如果做“查询订单详情”这个技能把订单号的校验、接口鉴权、异常返回码的兜底都封装在技能内部模型只需要传一个订单号成功率就会大幅提升。不过粒度也不是越细越好。如果每个原子操作都做成技能技能数量会爆炸模型在发现阶段就懵了。我实践下来的平衡点是一个技能应该对应一个用户可感知的完整动作且该动作的执行最多需要 5~7 个参数。超过这个阈值就考虑拆分子技能低于这个阈值且场景雷同的合并成一个带自动路由的技能。2. 技能定义与注册中心2.1 技能的数据结构定义技能的核心数据结构我设计成了五个部分缺一不可字段说明设计理由name技能唯一标识用于注册与调用命名需具备语义清晰度description模型可读的能力描述写清楚能做什么、何时该用、何时不该用直接影响模型的选择准确率input_schema参数定义JSON Schema约束模型按规范生成参数减少执行期报错entrypoint实际执行入口可以是函数、API 封装或另一个 Agenterror_policy异常兜底策略define 参数校验失败怎么办、接口超时重试几次这个结构里最考究的是 description。不要写成“获取天气信息”要写成“根据城市名称和日期获取该城市的天气概况。当用户询问未来三天是否适合出游时应调用此技能”。模型是语义匹配的机器你把使用场景写清楚命中率能提升一大截。input_schema 一定要用严格的 JSON Schema 而不是写自然语言描述。我用过纯文本描述参数的做法结果模型传错类型是常态尤其在参数类型是数字但模型给成字符串的时候多数后端解析直接崩溃。用 JSON Schema 配合大模型的 function calling 能力可以把参数生成阶段的错误率降到极低。2.2 注册中心的实现与设计要点有了技能定义接下来需要一个地方把这些技能统一管理起来。我这里参考了插件化架构的思路做了一个轻量注册中心。核心代码如下class SkillRegistry: def __init__(self): self._skills {} def register(self, skill_cls: type) - None: instance skill_cls() self._skills[instance.name] instance def get(self, name: str) - Skill | None: return self._skills.get(name) def list_skills(self) - list[dict]: 为模型提供精简的技能清单 return [ { name: skill.name, description: skill.description, input_schema: skill.input_schema, } for skill in self._skills.values() ]这里的“为模型提供精简清单”是个很关键的设计——注册中心里可能存着技能的执行代码但给模型“看”的只是描述和参数结构。执行代码属于内部实现细节一旦全部暴露出去一是上下文长度不够二是给了模型太多无关干扰项。实际使用时我还会在注册中心上叠一层“能力分域”把技能打上 domain 标签比如“订单域”“营销域”“数据查询域”。在 Agent 处理某类请求时只向模型开放相关域的技能清单。这个做法把模型每次决策可选的范围从几十个缩小到七八个准确率上升非常明显。2.3 技能发现与路由策略技能注册好之后下一个问题就是模型如何选择正确的技能。最初我完全依赖模型的语义理解能力让它从清单里挑一个。测试下来的效果是常见场景命中率还行但一旦遇到描述相近的两个技能比如“查询订单物流”和“查询订单详情”模型经常选错。我的解法分两层第一层触发词预筛。每个技能定义里可以设置 trigger_keywords 列表模型拿到用户请求时先做一次轻量级关键词打分把明显不相关的技能过滤掉缩小候选集。第二层模型语义选择。在候选集基础上让模型用 function calling 的标准流程做二选一或三选一。叠加参数约束后最终选错率可以控制在极低水平。这套策略被我们内部称为“先粗筛再精挑”虽然看起来多了一步但实际把大模型的无效推理大量减少了整体响应反而更快。3. 从零搭建一套可用的技能系统3.1 基类设计与执行入口规范工具类项目的核心一定是基类设计得够不够稳。我花了不少时间打磨技能的抽象基类最终收敛成这样from abc import ABC, abstractmethod from typing import Any, Optional import json class Skill(ABC): name: str description: str input_schema: dict {} trigger_keywords: list[str] [] domain: str general abstractmethod def execute(self, params: dict[str, Any]) - dict: 执行技能具体逻辑统一返回包含 code/data/message 的结构 pass def validate_params(self, params: dict) - list[str]: 根据 input_schema 校验参数返回错误信息列表 # 简化版本实际可接入 jsonschema 库 errors [] required self.input_schema.get(required, []) for key in required: if key not in params or params[key] in (None, ): errors.append(fmissing required param: {key}) return errors def run(self, params: dict) - dict: errors self.validate_params(params) if errors: return {code: 400, data: None, message: ; .join(errors)} try: result self.execute(params) return {code: 200, data: result, message: ok} except Exception as exc: return {code: 500, data: None, message: str(exc)}要注意我要求所有技能返回统一的数据结构 code/data/message。没有这一层的时候每个技能各写各的有的直接抛异常有的返回字符串后续做结果分析时只能靠人肉看日志。统一包裹层之后调度器可以拿 code 字段做分支判断运维监控也能直接对异常码做统计报警省了非常多的心力。3.2 一个真实技能案例查询订单物流理论讲再多不如看一个实际的技能实现。我拿电商场景中最常用的“查询订单物流”举例class QueryLogisticsSkill(Skill): name query_logistics description ( 根据订单号查询物流流转信息。适用于用户询问包裹在哪、物流是否更新、 预计配送日期等场景。非订单类查询请勿使用。 ) input_schema { type: object, properties: { order_id: { type: string, description: 订单编号通常形如 SO20250101XXXX } }, required: [order_id] } trigger_keywords [物流, 快递, 包裹, 发货, 配送] domain order def execute(self, params: dict) - dict: order_id params[order_id] # 内部封装了鉴权、缓存、供应商 API 调用 logistics_data self._query_from_supplier(order_id) return {trace: logistics_data[trace], status: logistics_data[status]} def _query_from_supplier(self, order_id: str) - dict: # 模拟第三方物流查询接口 return { status: in_transit, trace: [ {time: 2025-01-10 09:00, location: 上海转运中心, desc: 已出库}, {time: 2025-01-10 15:30, location: 杭州分拨中心, desc: 运输中}, ] }这个技能的内部实现其实还可以继续深挖比如对订单号做前缀校验、对第三方接口做超时控制、对查不到的数据做降级返回。当初我第一版没做这些结果模型传了个不存在的订单号技能直接抛异常Agent 还一本正经地对用户说“您的订单已送达”——这就是没做好异常兜底的教训。3.3 接入大模型调度让 Agent“学会用技能”技能本身只是工具箱真正让 Agent 学会使用它们需要在调度层做整合。我使用 OpenAI 风格的 function calling 来建立双向连接import json from openai import OpenAI client OpenAI() def run_agent_with_skills(user_query: str, registry: SkillRegistry): # 1. 从注册中心获取技能清单 skills registry.list_skills() # 2. 将技能转换为 function calling 格式 tools [ { type: function, function: { name: skill[name], description: skill[description], parameters: skill[input_schema] } } for skill in skills ] # 3. 第一轮让模型决定是否调用技能 response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: user_query}], toolstools, tool_choiceauto ) # 4. 如果模型选择调用技能执行并回填结果 if response.choices[0].message.tool_calls: tool_results [] for tool_call in response.choices[0].message.tool_calls: skill_name tool_call.function.name params json.loads(tool_call.function.arguments) skill_instance registry.get(skill_name) result skill_instance.run(params) tool_results.append({ tool_call_id: tool_call.id, result: json.dumps(result) }) # 5. 将结果回传给模型生成最终回复 final_response client.chat.completions.create( modelgpt-4o-mini, messages[ {role: user, content: user_query}, response.choices[0].message, *[{role: tool, tool_call_id: tr[tool_call_id], content: tr[result]} for tr in tool_results] ], toolstools, tool_choicenone ) return final_response.choices[0].message.content return response.choices[0].message.content这段代码展示了完整的“模型决策-技能执行-结果回填”闭环。有几个细节值得专门强调一是工具描述得多场景化二是回填结果的结构必须稳定三是模型版本差异明显——不同模型对 function calling 的支持程度差异很大选型前务必实测。我自己做过的对比测试里在同一组技能定义下最新一代模型的工具选择准确率比上一代高出差不多二十个百分点。这也意味着如果你的 Agent 表现不好别急着改代码先换个更新的模型试试往往更省事。4. 技能编排从单技能到工作流4.1 顺序编排与状态传递单技能能解决的问题始终有限真实业务几乎都是多步流程。比如“退货退款流程”需要验证订单状态、检查退款资格、计算退款金额、执行退款、通知用户。这五个步骤如果靠模型一次调用全做完中间任何一步出错前面全部白干。我给系统设计了一套轻量级的顺序编排机制核心思路是让上一步的输出映射为下一步的输入from typing import Any class WorkflowEngine: def __init__(self, registry: SkillRegistry): self.registry registry def run_sequence(self, steps: list[dict], init_context: dict) - dict: context dict(init_context) for step in steps: skill_name step[skill] param_mapping step.get(param_mapping, {}) # 从 context 中映射参数 mapped_params {} for param_name, src_key in param_mapping.items(): mapped_params[param_name] context.get(src_key) # 执行技能 skill self.registry.get(skill_name) result skill.run(mapped_params) # 将结果写入上下文供后续步骤使用 output_keys step.get(output_keys, {}) for local_key, context_key in output_keys.items(): context[context_key] result[data].get(local_key) if result[code] ! 200: # 支持短路失败 return {success: False, failed_step: skill_name, context: context} return {success: True, context: context}这里最关键的抽象是 param_mapping 和 output_keys前者把上下文中的数据映射到技能的参数名后者把技能的返回结果写回上下文。通过这样一层映射技能与技能之间完全解耦我不需要改任何技能代码只调整工作流的 mapping 配置就能拼出新的业务流程。4.2 条件分支与并行执行流程引擎只有顺序是不够的业务里到处都是“如果今天是周末就走 A 流程否则走 B 流程”这种分支。我给 engine 加了一个简单的 branch 结构步骤可以携带 condition 字段内容是一段可序列化的断言表达式引擎在运行时对上下文做匹配。实现方式不复杂核心是控制条件表达式的语法边界不要引入任意代码执行。并行执行方面我的态度是慎重使用。Agent 场景下的并行和传统后端开发完全是两回事——大模型生成的参数质量不稳定两个并行技能如果共享同一个上下文且其中一个技能会修改上下文字段另一个读到的数据就可能是脏的。我现在只在两个技能完全互不依赖时才做并行且把上下文显式改成“不可变快照”传给每个并行分支避免隐性耦合。4.3 编排的另一种选择让模型自己编上面说的 workflow 是“预设好的剧本”模型没有发挥空间。但有些场景本身就是开放式的比如“帮我把下周的出差安排都处理好”你根本无法预写所有流程。这时我尝试了另一种模式把技能清单交给模型让其自己规划步骤并进行动作拆分逐步输出多个函数的调用序列。这个模式的容错率比固定工作流低很多但对任务的覆盖面广得多。我内部跑了两个月的测试后结论是开放式场景用模型自编稳定场景用预置工作流二者通过一个简单的意图路由开关切换是我目前见过的最实用的组合方式。5. 常见问题与排查技巧实录5.1 五大高频问题的排查速查表接入 agent-skills 这套体系跑了大半年我把团队踩过的高频问题整理了一份速查表问题现象根本原因解决方案模型总是调用错误的技能描述过于抽象或多个技能场景重叠重写 description加入具体业务场景动词给重叠技能加区分性关键词参数频繁缺 key 或类型错误input_schema 定义含糊或者未设置 required严格使用 JSON Schema给每个参数写清格式约束技能执行超时导致 Agent 假死技能内部调用外部接口无超时控制在技能基类 run() 中强制包裹超时装饰器并在错误策略里设置降级模型无视技能直接编答案工具提示词太长模型注意力被稀释压缩 description 长度使用触发词预筛缩小候选集新增技能后其他技能失灵技能之间名称或语义冲突上线前做语义碰撞测试将相同 domain 的技能做批量评测5.2 调试工具技能沙箱评测Agent 开发和传统后端开发最不一样的地方在于它的“正确性”不是确定性的你的技能写得再好模型也可能因为一个描述措辞就选错。这就要求我们必须有一个技能评测沙箱在每次改动后跑一遍回归。我的沙箱逻辑很简单准备一组覆盖所有技能的用户 query跑一次完整链路记录每个 query 的技能命中情况。再配上工具选择和参数质量的打分逻辑就能得到一份技能召回率和准确率的数字报表。当初我第一次跑这套评测时发现订单域的召回率低得离谱查日志才发现是 description 里只写了“查订单”没写“我的单子到哪了”这种口语化表达后来补上近义词表指标立马上来了。从那之后我养成了一个习惯每个技能的 description 必须包含业务侧真实出现过的口语表达。5.3 性能与成本的平衡实践技能体系还有一个常被忽视的隐性成本技能数量一旦增加给模型传的 function schema 会膨胀每轮对话的 token 消耗直线上升。为了控制成本我在 registry 里加了一个 profile 机制——按照当前对话所处的业务阶段只加载相关域的技能清单。比如在售前阶段只加载商品查询技能在售后阶段才加载退款/物流/评价相关技能。实测这个优化把单轮对话的 token 消耗大幅压缩同时因为可选技能少了模型的决策准确率反而提升了。这是典型的“减法优化”不是给模型更多选择而是让它该做什么时只有最合适的选项。写在最后回头来看agent-skills 这套体系最核心的价值不在于某一段代码多精巧而是它逼着我把“Agent 能力”当成一个产品去治理技能有定义、有注册、有评测、有编排每个环节都有标准可循。我个人的体感是这套东西理顺之后接新技能的效率高了很多Agent 在真实业务里翻车的次数也肉眼可见地变少了。最后分享一个小技巧在设计新技能之前先把用户真实说过的原始问题整理成几十条语料对着语料写 description而不是对着功能文档写。Agent 不是读功能文档的它读的是“用户会怎么问”。这一点想通了你的技能命中率自然就上去了。
返回列表