ARTICLE DETAIL

资讯详情

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

从巨石到技能层:Agent架构的技能编排与工程实践

从巨石到技能层:Agent架构的技能编排与工程实践 去年年初开始我们的后端团队在慢慢把业务往 agent 架构上迁移。最早一批 agent 的代码写出来之后很快就遇到了一个很典型的问题每个 agent 都把自己要做的事、要调的接口、要处理的异常全揉在 prompt 和 print 语句里一个月之后再看一个 agent 就是一个没人敢动的巨石。后来我们参考了社区里 agent-skills 这个方向的设计思路把技能从 agent 的主逻辑里抽出来做成了一个独立的、可注册、可编排的技能层。这套改造做完之后效果非常明显——新需求从两三天压缩到半天老 agent 的 bug 率也降了一大截。这篇内容就是围绕 agent-skills 这个主题聊聊我们是怎么从混乱走向结构化的包括技能层为什么值得单独做、技能描述符该怎么设计、运行时编排有哪些坑以及我踩过的几个比较典型的故障。1. 为什么 agent 需要独立的技能层1.1 把技能从 prompt 里解放出来很多入门级的 agent 会把“做什么”和“怎么做”全部写进 system prompt。比如一个客服 agentprompt 里写着“当用户问退款时调用 refund_api 并传入 order_id”。这种方式在小规模 demo 里跑得通一旦技能数量超过十个prompt 就会变得臃肿不堪而且每加一个技能就得调整 prompttoken 消耗增加模型的理解精度反而下降。agent-skills 的核心思路是把技能从 prompt 里抽出来变成可以独立注册、独立调用、独立维护的代码模块。每个技能有自己的描述、参数定义、执行逻辑和返回值规范agent 不再靠“记住”技能而是靠“发现”技能。这个转变的实质是把模型从“记忆者”变成“决策者”。模型只需要根据用户意图去匹配技能而技能的具体实现细节由代码完成。这样 prompt 会大幅缩短模型的稳定性会提高技能本身也可以像普通代码一样做单元测试。1.2 为什么技能层要独立成模块在我们的实践中发现技能层如果不独立通常会面临三个问题职责混乱技能逻辑和 agent 的对话逻辑耦合在一起改技能可能影响对话行为改对话行为也可能误伤技能。复用困难两个 agent 可能都需要查询订单状态的技能但代码写在各自的逻辑里没法直接共享。测试成本高技能的输入输出没有统一规范测试只能端到端跑无法对单个技能单独验证。技能层独立之后这些问题基本迎刃而解。技能模块不关心对话上下文只关心输入参数和输出结果你可以像测试普通函数一样测试它。同时多个 agent 可以共享同一个技能注册表按需加载避免了重复开发。1.3 技能与工具、插件的边界这里需要澄清一个很容易混淆的概念skill、tool 和 plugin 到底有什么区别。在我们拆解 agent-skills 的过程中给这三者划了一条比较清晰的边界Tool工具最基础的原子操作比如“发送HTTP请求”“读写文件”“执行SQL查询”它们不包含业务语义。Skill技能在工具基础上封装了一层业务语义比如“查询订单状态”“生成退款单”“计算运费”它们通常需要组合多个工具并且包含一定的业务规则。Plugin插件是技能的集合通常对应一个完整的功能域比如“订单管理插件”包含了查询、退款、修改地址等多个技能。所以 agent-skills 关注的是中间这一层如何描述一个技能如何让 agent 理解并调用它如何编排多个技能完成复杂任务。2. 技能描述符的设计与调度机制2.1 技能描述符agent 理解技能的桥梁agent 要正确调用技能必须理解技能是干什么的、需要什么参数、返回什么结果。我们把这份描述性信息称之为技能描述符Skill Descriptor。一个合格的技能描述符至少应包含以下几项skill_id技能唯一标识agent 通过它引用技能。name人类可读的名称便于日志排查。description一段简洁的描述说明该技能适用的场景这部分会被注入 prompt 供模型匹配。parametersJSON Schema 格式的参数定义包括字段名、类型、是否必填、描述。returns返回值规范说明成功和失败时各自返回什么结构。我们的实践经验是description 写得好不好直接影响模型选技能的准确率。描述要突出“什么场景用这个技能”“和别的技能的区别在哪里”而不是泛泛地说“这是查询订单状态的技能”。同时描述里应写清楚典型的使用条件比如“仅当用户提供订单号时使用”这样模型在参数缺失时就会先追问用户而不是直接报错。字段说明示例skill_id唯一标识order.query_statusname可读名称查询订单状态description匹配用描述根据订单号查询订单当前物流与支付状态仅当用户已提供订单号时调用parametersJSON Schema{ order_id: { type: string, required: true } }returns返回值规范{ status: shipped }2.2 全局技能注册表与动态加载有了技能描述符之后下一步就是让这些技能可被发现、可被加载。我们用了“注册表 动态加载”的模型。注册表是一个全局的字典结构key 是 skill_idvalue 是技能对象。启动时我们扫描所有技能目录将技能注册进去运行时agent 根据模型匹配结果从注册表取出技能并执行。动态加载需要考虑版本问题。我们的做法是每次发布新技能时不覆盖旧版本而是以版本号区分注册表中同时保留多个版本agent 默认调用最新稳定版如果需要回滚可以通过配置指定版本。# registry.py class SkillRegistry: def __init__(self): self._skills {} def register(self, skill): if skill.skill_id in self._skills: raise ValueError(fskill {skill.skill_id} already registered) self._skills[skill.skill_id] skill def get(self, skill_id): if skill_id not in self._skills: raise KeyError(fskill {skill_id} not found) return self._skills[skill_id] def list_skills(self): return [ { skill_id: s.skill_id, name: s.name, description: s.description, parameters: s.parameters, } for s in self._skills.values() ]2.3 技能调度匹配、鉴权与执行调度是 agent 调用技能的核心链路我们将其拆为三步第一步是匹配。模型基于用户输入和技能描述符的 description 做语义匹配输出要调用的 skill_id。这里需要约束模型只能从注册表已有的 skill_id 中选避免模型编造不存在的技能。实践中我们用函数调用function calling来实现这一约束模型返回的调用参数会经过 schema 校验。第二步是鉴权。不是所有技能所有用户都有权限调用。我们的做法是为技能配置权限标签比如“仅管理员”“仅内部系统”调度层根据当前会话的权限上下文决定是否放行。第三步是执行。执行时把模型解析出的参数传给技能函数。行内有一个超时控制和重试机制超时时间默认设为 10 秒重试次数默认 2 次可调的参数都写在配置中心里。3. 实战从零实现一套 agent-skills 体系3.1 技能基类的抽象设计为了让技能代码保持统一风格我们定义了一个基础抽象类 BaseSkill。所有技能继承这个类并实现必需的接口这样调度层就可以用统一的方式调用所有技能。# base_skill.py from abc import ABC, abstractmethod from typing import Any, Dict class BaseSkill(ABC): skill_id: str name: str description: str parameters: Dict[str, Any] permissions: list[str] [] abstractmethod def execute(self, params: Dict[str, Any], context: Dict[str, Any]) - Dict[str, Any]: 执行技能逻辑返回结构化结果 passexecute 接收两个参数params 是模型解析出的参数context 是运行上下文包括用户身份、会话 ID 等。返回值统一用字典结构包含 status、data、message 三个字段方便调度层统一处理成功与失败。3.2 按目录组织技能实现自动注册我们采用目录即模块的组织方式每个技能一个目录目录名就是 skill_id目录内包含 main.py实现逻辑、schema.json参数定义、description.txt描述文本。启动时框架自动扫描所有技能目录并注册。skills/ ├── order_query_status/ │ ├── main.py │ ├── schema.json │ └── description.txt ├── order_refund/ │ ├── main.py │ ├── schema.json │ └── description.txt └── logistics_trace/ ├── main.py ├── schema.json └── description.txt自动注册的扫描逻辑很简单遍历 skills 目录找到每个包含 main.py 的子目录用 importlib 动态导入并实例化然后塞进注册表。这里有个值得注意的坑动态导入的模块名不能重复我们建议以技能目录名作为模块名的一部分比如skills.order_query_status.main。3.3 技能描述符注入 prompt 的格式设计技能描述符如何注入 prompt直接影响模型匹配的准确率。我们尝试过两种方式一是全部注入 system prompt二是只注入技能列表需要时再展开详情。实践下来第二种方式效果更好。我们采用的做法是把技能列表压缩成一行摘要注入系统消息摘要格式为“skill_id: 简短描述”模型通过摘要缩小候选范围再通过 function calling 的 schema 完成参数绑定。可用技能列表 - order.query_status: 查询订单状态需要订单号 - order.refund: 发起订单退款需要订单号和退款原因 - logistics.trace: 查询物流轨迹需要运单号这种方式既削减了 token 消耗又保证了模型能掌握技能的全貌。模型只负责输出 skill_id 和参数不负责拼接逻辑大大降低了出错的概率。3.4 技能的编排与组合当单个技能无法满足用户需求时我们需要把多个技能编排起来。比如用户问“我的订单到哪了顺便退款”这需要先查询订单号对应的运单号再查询物流轨迹最后发起退款。我们实现了一个简单的编排引擎支持顺序执行和有条件执行。顺序执行就是按 skill_id 列表依次调用的流水线前一个技能的输出可以作为后一个技能的输入映射。有条件执行则是根据某个技能返回的字段决定是否执行下一个技能。# pipeline.py class SkillPipeline: def __init__(self, steps): self.steps steps async def run(self, initial_params, context): current_params initial_params for step in self.steps: skill registry.get(step.skill_id) result await skill.execute(current_params, context) if step.condition and not step.condition(result): return result current_params step.output_mapper(result, current_params) return current_params这个引擎看似简单但把技能之间的依赖关系显式化了。每条流水线定义在配置文件里新业务只需要新增配置和技能代码不需要改主逻辑。3.5 技能执行的超时、重试与降级技能执行过程中超时和失败是最常见的问题。我们为每个技能配置了超时时间、重试次数和降级策略这些配置项统一放在配置中心支持热更新。超时时间的设置需要根据技能类型区分内部函数调用通常 3 秒足够外部 HTTP 调用建议放宽到 10 秒。我们最初统一用 5 秒结果外部接口稍慢就触发超时之后改为分技能配置问题就消失了。重试要特别注意幂等性。查询类技能可以安全重试但退款、下单等写操作如果重试前没有做幂等检查很容易重复提交。我们要求所有写操作类技能在参数中带上 idempotency_key服务端根据这个 key 去重。4. 踩坑实录常见问题与排查技巧4.1 模型乱编 skill_id上线初期模型偶尔会输出一个注册表里不存在的 skill_id对话直接崩溃。排查后发现原因在于 prompt 里技能列表格式太松散没有强调必须从列表中选择。修复方案有两个一是结构化约束用 function calling 的方式让模型只能从给定的函数列表中选择二是增加校验调度层拿模型返回的 skill_id 去注册表检查找不到就返回一条清晰提示让模型重新选择。我们最终两层都做了效果很好。现在即便模型输出了非法 ID调度层也会优雅地提示“技能不存在请从以下列表中选择”而不是直接抛异常。4.2 参数校验不一致另一个高频问题技能内部的参数校验和描述符里的 JSON Schema 校验不一致导致描述符说订单号必填代码里却允许空值或者反过来模型按 Schema 传了参数代码却因为类型不匹配报错。我们的解决办法是强制代码执行前统一走 Schema 校验。所有技能在 execute 开头先校验 params 是否符合 schema.json 定义不符合就返回参数错误。这样代码里就不用再重复写参数检查逻辑也保证了两处的校验规则永远一致。def execute(self, params, context): errors validate_schema(params, self.parameters) if errors: return {status: error, message: f参数校验失败: {errors}, data: None} # 业务逻辑4.3 技能会话状态丢失第三个值得一提的问题是状态管理的陷阱。我们的订单技能需要登录态最初把登录态存在技能内部结果换一个会话就丢了。排查之后发现技能应该是无状态的所有状态必须放在 context 里由调度层维护。我们把技能接口约束为无状态后技能的复用性大幅提升。状态信息用户身份、会话、租户 ID统一从 context 传入技能内部不维护任何会话数据。4.4 技能冲突与版本回滚多个技能同时依赖同一个底层服务时很容易出现版本冲突。比如查询订单技能和查询物流技能都依赖订单服务 SDK升级 SDK 后物流技能暂时不兼容。目前我们的做法是技能与 SDK 版本一起打包每个技能独立声明依赖。上线新技能前跑一套自动测试如果某个技能的依赖与全局配置冲突就暂时保留旧版本等依赖更新后再切换。我在实际把 agent-skills 这套体系落地到生产环境之后最深的体会是技能层解决的不是“能不能调用”的问题而是“能不能规模化管理”的问题。单个 agent 手写技能调用没问题但当你有二十个 agent、上百个技能的时候没有统一的注册、调度、校验和版本管理系统迟早会在某个半夜被一个参数错误打崩。把技能当成一等公民来设计短时间看好像多写了一些基础设施代码但后面每次新增功能、每次排查线上问题你都会发现这套投入是值得的。
返回列表