ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

Agent技能系统设计实战:从函数调用到可复用能力单元

Agent技能系统设计实战:从函数调用到可复用能力单元 前几天和一个做AI应用的朋友聊天他吐槽说现在接大模型接口写Agent最头疼的不是模型能力不够而是把“让模型干活”这件事做得可靠。他团队里十几个Agent每个都挂了一堆函数有的叫get_weather有的叫fetch_user_info但换个场景、换个模型这些“技能”就很难复用。我听完第一反应是这哥们儿缺的其实不是更多提示词而是一套正经的agent-skills工程化思路。“agent-skills”这个词这两年从Anthropic提出Agent Skills概念之后基本成了Agent工程化绕不开的关键词。说白了它不是让你去调某个现成的API而是指给Agent设计、封装、管理一套“可复用能力单元”的方法论。你可以把它理解成给AI助手装上一个个插件式的“专业技能包”让它能根据具体任务自动调用对应的能力而不是每次都在系统提示词里塞一大坨互相纠缠的指令。这篇文章我就把这些年在Agent技能系统上踩过的坑、沉淀下来的设计思路连同可以直接抄作业的框架代码一次讲清楚。不管你是刚接触Agent开发还是已经在生产环境里磨了好几个版本这篇文章应该都能给你一些有用的东西。1. 为什么Agent需要一套“技能系统”而不是一堆工具函数先聊一个很本质的问题既然模型本身就支持function calling我们直接把函数注册给模型不就行了吗为什么还要搞一套“技能系统”实际做过的朋友应该都知道函数调用和技能系统之间隔着一层“工程化”的距离。1.1 从“能用”到“好用”缺的是技能抽象层单纯的函数调用解决的是“模型知道有哪些函数、参数怎么填”的问题。但落到真实业务里你很快就会遇到几个尴尬场面一个技能往往不是一个函数能搞定的需要多个函数按顺序配合。比如“帮用户分析一份PDF财报”至少涉及文件解析、数据抽取、指标计算、报告生成四步。如果全拆成函数丢给模型它自己编排出来的流程大概率不稳定。技能应该有“记忆”和“配置”。同一个报表生成技能给财务部门和给运营部门用指标口径、输出格式完全不一样。函数做不到这种上下文感知。技能需要被复用、被分享、被版本管理。一个团队里不同Agent可能都要用到“发送邮件”这个能力。如果每个人各自写一个函数改一个逻辑就得全网通知完全是一团乱麻。所以agent-skills的核心理念是把“模型与外部世界交互的最小能力单元”标准化、模块化、可管理化。它不是函数的上位替代而是在函数和服务之上加了一层面向任务的“能力壳”。1.2 技能与大模型协作时的边界划分设计技能系统时我们得时刻想清楚哪些逻辑放模型那边哪些逻辑放技能这边。放错了结果就是要么模型瞎发挥要么技能臃肿得像个单体应用。我的经验是遵循一个原则凡是规则明确、有固定执行路径的尽量沉淀到技能内部凡是需要理解、判断、生成策略的留给模型去决策。拿“文件总结”举例。文件格式识别、内容分段提取、超长文本切片这些是规则明确的必须在技能里做不能指望模型自己处理2万字的洗稿文件还能不丢上下文。但“从文件中找出核心观点并输出报告”这种就需要模型来判断什么算“核心”技能把素材准备好把决策权交给模型。再比如“联网搜索”。技能内部要处理搜索词重构、网页抓取、正文提取、去重过滤这些统统是标准动作。但“搜什么关键词、看哪些链接、如何判断信息可信度”就应该让模型基于当前对话上下文来完成。边界划清楚了技能才会既稳定又灵活。2. 技能系统的四大核心设计维度聊完“为什么”我们进入正题看看一个生产可用的agent-skills框架到底要关注哪些设计维度。这里我会结合自己实际写过的框架来讲尽量不说虚的。2.1 接口标准让技能像USB-C一样即插即用技能系统的第一个核心问题是接口怎么定。业界目前还没有一个统一的Agent技能协议但一个合理的最小接口集通常包含这么几个部分技能元信息名称、描述、版本、作者、依赖环境这些描述要够清晰让模型能准确判断“什么时候该用这个技能”。输入/输出契约结构化定义技能接收什么参数、返回什么格式。这里建议用JSON Schema做输入校验而不是相信任何调用方包括模型会规规矩矩传参。执行入口所有技能暴露一个统一的执行方法比如Python里的async def run(context)或者def execute(params)这样上层调度器可以统一调用不需要为每个技能写特判。生命周期钩子初始化、校验、执行、清理、错误处理。这五个钩子看着不起眼实际用起来能解决大量脏活累活。比如初始化时加载模型、连接数据库清理时释放资源错误处理时做日志采集和降级。我自己在项目里用Pydantic定了一套基础类所有技能都继承同一个基类接口长得像这样简化版from pydantic import BaseModel, Field from abc import ABC, abstractmethod class SkillInput(BaseModel): pass class SkillOutput(BaseModel): pass class BaseSkill(ABC): name: str base_skill description: str Base skill description version: str 1.0.0 input_schema: type[SkillInput] SkillInput output_schema: type[SkillOutput] SkillOutput abstractmethod async def execute(self, params: SkillInput) - SkillOutput: ...这样统一之后往下接Agent框架、往上挂服务层都很顺因为外层根本不在乎你内部干了啥只在乎你是不是实现了execute。2.2 技能注册与发现模型怎么知道该用哪个技能接口标准化解决的是“怎么调用”接下来要解决“调用哪个”。模型在跑任务时不可能从几百个技能里逐一筛选。所以技能系统需要一套注册与发现机制。常见的做法是技能注册中心Skill Registry。每个技能启动时向注册中心登记自己的元信息注册中心负责三件事维护一个“技能索引表”内置技能ID、触发条件、适用场景、输入要求。根据Agent当前的任务上下文做一次粗粒度的技能候选筛选。这一步可以用关键词匹配也可以直接向量化后用语义检索。将候选技能的名称和描述拼进模型可访问的上下文里指导它精准选择。筛选这一步极其重要。如果不做粗筛把所有技能描述都塞给模型既浪费token又会让模型选择困难甚至错误使用。我见过一个案例Agent要执行“查询天气”结果模型选了一个“股票行情分析”技能就是因为候选太多描述互相干扰。还有一个容易被忽略的点技能描述别写得太“文学化”。模型的语义理解能力再强你写“此技能旨在为用户提供多元化全方位的信息获取解决方案”它大概率搞不懂你在说啥。直接写“按城市名查询实时天气支持国内主要城市”这种大白话命中率反而高得多。2.3 技能上下文与状态管理让技能带着“记忆”干活技能不能是无状态的“纯函数”否则在真实业务里会非常难用。比如一个“写周报”技能它需要知道用户是谁、过去七天做了什么、周报格式偏好是什么。这些信息从哪来不是每次调用都重新让用户填一遍而是技能系统要和Agent的上下文系统打通。这一块我的设计思路是分三层全局上下文用户身份、租户配置、跨技能共享的偏好设置所有技能都能读取。会话上下文当前Agent会话内的状态比如对话历史摘要、已经执行过的技能列表、中间产出物。技能私有状态技能自己缓存的临时数据比如一次长任务中途的计算结果。实现上我通常会封装一个SkillContext对象把它传给每个技能的execute方法。这个对象内部维护一个线程安全的内存存储同时支持序列化到Redis等外部存储以便多实例部署时共享状态。这里有一个经验之谈技能执行过程中产生的中间数据尽量显式写入上下文而不是藏在技能内部的全局变量里。藏内部变量的坏处是进程一重启就丢了而且多个技能并发跑的时候容易串数据。显式写入上下文逻辑清晰还不容易踩并发坑。2.4 技能的可观测性出问题时别靠猜Agent系统是个天生的“黑盒”模型的行为有随机性技能链路较长一旦出问题定位成本极高。所以技能系统从第一天起就要把可观测性设计进去等上线了再加通常已经晚了。可观测性至少要覆盖四个维度日志每个技能执行时记录入参、出参、耗时、错误堆栈。这个看起来基础但很多团队连这个都没做到出了问题只能人肉翻日志。追踪一条Agent任务会串起多个技能需要一个全局Trace ID贯穿始终把技能调用链串起来看。指标技能的调用次数、成功率、平均耗时、p99耗时这些数据要持续采集用来判断哪个技能需要优化。评估技能产出的质量评估。有些技能是纯代码逻辑输出是确定性的好评估有些技能依赖模型生成输出质量波动大就需要引入额外的评估机制比如用户反馈打分、规则校验、定期抽样人工评估。追踪这块我强烈建议选一个开源方案比如OpenTelemetry的标准或者LangSmith这类产品化的工具别自己造轮子。造轮子的坑在于初期看着省事等技能数量超过20个你会后悔的。3. 从零搭建一个最小可用的agent-skills框架讲完了设计维度我们直接上手撸一个最小但五脏俱全的skils框架。这个框架我实际在项目里用过简化版本地跑通后你完全可以往里面加数据库、消息队列、分布式调度把它扩展成生产级。3.1 技能仓库结构与基础抽象先看仓库结构。核心理念是一个技能一个目录目录内自带元信息、执行逻辑和依赖声明。这样技能天然可复用、可分享新同学参与开发也容易上手agent-skills/ ├── core/ │ ├── __init__.py │ ├── base.py # 技能基类定义接口标准 │ ├── registry.py # 技能注册中心 │ ├── context.py # 技能上下文管理 │ └── exceptions.py # 统一异常体系 ├── skills/ │ ├── __init__.py │ ├── email_summary/ │ │ ├── skill.py # 技能主逻辑 │ │ ├── config.yaml # 技能配置 │ │ └── requirements.txt │ └── weather_query/ │ ├── skill.py │ ├── config.yaml │ └── requirements.txt ├── agent/ │ └── runner.py # 与模型交互的调度器 └── tests/ ├── test_registry.py └── test_email_summary.pybase.py里除了上面写过的基类我还会加一个validate_input方法和一个preprocess钩子。validate_input用JSON Schema做参数校验preprocess用来统一做鉴权、限流、资源准备。这两个方法非常实用能让业务技能专注于自己的核心动作class BaseSkill(ABC): 所有技能的抽象基类定义生命周期和接口契约 name: str base_skill description: str version: str 1.0.0 tags: list[str] [] abstractmethod async def execute(self, params: SkillInput, context: SkillContext) - SkillOutput: 技能核心执行逻辑子类必须实现 ... async def preprocess(self, params: SkillInput, context: SkillContext) - SkillInput: 执行前的统一预处理如鉴权、限流 return params async def postprocess(self, result: SkillOutput, context: SkillContext) - SkillOutput: 执行后的统一后处理如脱敏、格式化 return result async def cleanup(self, context: SkillContext) - None: 资源清理异常退出时也会被调用 ...3.2 技能注册中心实现注册中心是技能系统的“大脑”。我的实现里它维护两套索引一套是按技能ID映射具体对象的字典一套是面向模型查询的语义索引。后者我直接用了一个轻量级的词向量匹配够用不用动不动就上向量数据库from typing import Optional, Type import inspect class SkillRegistry: 技能注册中心负责登记、检索和生命周期管理 def __init__(self): self._skills: dict[str, BaseSkill] {} self._semantic_index: dict[str, list[str]] {} # 技能ID - 关键词列表 def register(self, skill: BaseSkill, keywords: list[str] | None None) - None: 注册一个技能keywords是可选的触发词扩展 if skill.name in self._skills: raise SkillConflictError(fskill {skill.name} already registered) self._skills[skill.name] skill self._semantic_index[skill.name] keywords or [] logger.info(fregistered skill: {skill.name} v{skill.version}) def discover(self, task_description: str, top_k: int 5) - list[BaseSkill]: 根据任务描述粗筛候选技能 candidates [] for name, skill in self._skills.items(): # 先用描述做粗过滤 score self._calc_match_score(task_description, skill.description) # 再加上关键词扩展命中 for kw in self._semantic_index[name]: if kw.lower() in task_description.lower(): score 0.5 candidates.append((score, skill)) candidates.sort(keylambda x: x[0], reverseTrue) return [s for _, s in candidates[:top_k] if _ 0] def get(self, skill_name: str) - Optional[BaseSkill]: return self._skills.get(skill_name) def _calc_match_score(self, task: str, description: str) - float: 轻量文本相似度实际项目里可换成embedding方案 task_set set(task.lower().replace(., ).split()) desc_set set(description.lower().split()) if not task_set or not desc_set: return 0.0 overlap len(task_set desc_set) return overlap * 2.0 / (len(task_set) len(desc_set))注册中心的discover这里_calc_match_score是一个非常简化的词重叠方案。实际生产环境我会推荐用sentence-transformer或者OpenAI的embedding接口生成描述向量再用余弦相似度检索。不过本地demo阶段词重叠方案已经能让你看到完整流程的样子了。3.3 Agent调度器让模型在技能中间做决策有了注册中心接下来写Agent调度器。这块的核心是一个循环模型看任务决定调哪个技能技能执行结果喂回模型模型决定下一步动作。整个过程有点像带工具的人类员工干活想到什么查一下得到结果继续想直到任务完成。from typing import Optional import json class AgentRunner: Agent调度器负责编排模型与技能的交互流程 def __init__(self, model_api, registry: SkillRegistry, max_steps: int 8): self.model_api model_api self.registry registry self.max_steps max_steps async def run(self, user_query: str) - str: messages [{role: user, content: user_query}] for step in range(self.max_steps): response await self.model_api.chat(messages) # 解析模型输出判断是要求调用技能还是已生成最终回复 action self._parse_action(response) if action[type] reply: return action[content] if action[type] call_skill: skill_name action[skill_name] params action[params] skill self.registry.get(skill_name) if not skill: messages.append({ role: tool, name: skill_name, content: json.dumps({error: fskill {skill_name} not found}) }) continue result await skill.execute( paramsskill.input_schema(**params), contextSkillContext() ) # 把执行结果塞回对话历史 messages.append({ role: tool, name: skill_name, content: result.model_dump_json() }) return 达到最大步数任务未完成。也许需要拆分任务或调整技能配置。实际接入时模型API返回的action格式需要根据你用的模型调整。用原生tool-calling接口比如OpenAI的tools参数会更规范内部逻辑从“解析模型自然语言输出”变成“解析结构化tool_call”省去很多字符串解析的麻烦。上面这个示例保留了“模型输出-解析-执行”的流程方便你理解全局。3.4 技能上下文与配置文件实现最后是context.py和技能配置文件。SkillContext要能在整个技能链路里透明传递。为了演示简单这里用一个进程内的数据存储生产环境换成Redis或数据库连接也一样from dataclasses import dataclass, field from datetime import datetime from typing import Any import threading, uuid dataclass class SkillContext: 技能执行的全局上下文承载状态传递 trace_id: str field(default_factorylambda: uuid.uuid4().hex) session_id: str user_id: str created_at: str field(default_factorylambda: datetime.utcnow().isoformat()) def __post_init__(self): self._state: dict[str, Any] {} self._lock threading.Lock() def set(self, key: str, value: Any) - None: with self._lock: self._state[key] value def get(self, key: str, default: Any None) - Any: with self._lock: return self._state.get(key, default) def to_dict(self) - dict: 序列化用便于日志和追踪 return { trace_id: self.trace_id, session_id: self.session_id, user_id: self.user_id, created_at: self.created_at, state: self._state, }到这里一个最小可用的技能框架就能跑起来了。我本地测试时用上面这套结构注册了3个技能然后接一个开源模型API让它执行“查询北京的天气并总结成一句提醒带伞的话”这种复合任务。模型先调weather_query拿到结果后又调用text_summary生成最终回复整个链路走得相当顺。你可以试着在这个基础上加技能进去感受一下“能力单元被复用”的快乐。4. 实战中的坑与排查技巧框架写出来是一回事跑起来不出问题又是另一回事。这里我把自己在技能系统上踩过的坑整理成了一份速查希望能帮你少走弯路。4.1 技能参数校验不可靠模型传参容易“自由发挥”模型生成JSON参数时偶尔会出现字段缺失、类型错误、枚举值乱填的情况。如果你在技能内部不校验直接取参数运算轻则白跑一次重则把脏数据写进数据库。我的做法是三层防线第一层模型出口做一次结构校验用tools接口自带的strict模式或JSON Schema校验。第二层技能入口再校验一遍复用SkillInput的Pydantic能力。第三层技能内部访问关键字段时用context.get加默认值不要直接下标访问。三个地方重复写校验确实有点繁琐但它能挡住90%的异常让技能的稳定性上一个台阶。生产系统里出错成本的次序是运行时拿脏数据最贵执行前发现错误较贵模型侧拦截最便宜。所以不管你多懒第二层一定要保留。4.2 模型诊断技能调用失败时别急着怀疑模型先查技能本身遇到Agent技能执行报错第一反应不应该是“大模型是不是抽风了”而是先看技能日志和追踪信息。大概率是这几个原因技能入参不符合预期、依赖的第三方服务超时、技能内部状态被并发写坏了、技能描述与实际行为不一致导致模型错误调用。我特别想提“技能描述与实际行为不一致”这个坑。有一次我写了一个“search_knowledge_base”技能内部实现其实只能查文档库但描述里没写清楚结果模型拿它去查用户权限返回一堆无关内容整个任务链路都偏了。后来我把描述改成“query internal documentation by keywords, returns document titles and snippets”行为才正常。你写技能描述时要站在模型的“阅读理解视角”去审视而不是站在开发者视角觉得“这不显然吗”。4.3 状态泄漏多个Agent会话互相串数据技能里如果用了类级变量或者模块级单例来存临时状态在并发场景下A用户的请求很可能会读到B用户的数据。这类问题最难排查因为报错可能完全随机。排查思路一旦出现这种偶发数据错乱先检查技能里有没有非必要的全局可变状态再把SkillContext的set/get打点追踪数据写入方和读取方的会话ID。根治办法是技能内部尽量无状态所有可变数据统一放上下文并且上下文要绑定会话ID。这里附一张我实际踩坑时整理的排查表症状最可能原因排查入口推荐做法技能偶尔返回空结果入参校验不严模型传入空值查看日志里的入参快照SkillInput加必填校验和默认值并发时数据错乱技能内部存在全局可变状态检查类变量和模块级变量状态全部移入SkillContext模型频繁调用错误技能技能描述含糊或过于宽泛查看discover阶段的候选排名重写描述增加触发词缩小边界技能报超时依赖服务慢或技能无超时控制查看追踪中的依赖耗时为外部调用统一设置超时和重试技能链路过长模型记不住中间结果没有把中间结果及时压回对话上下文查看最终推理消息的上下文长度适时对中间结果做摘要清理无用内容4.4 超时与重试策略不能一刀切技能里调外部API、数据库、模型接口每一个外部依赖都要有独立的超时配置。千万不要只设一个全局默认超时理由是不同依赖的健康状态差异很大。比如连接本地Redis超时100毫秒都嫌多调第三方网页搜索API3秒都可能不够。我的习惯是在技能配置里加一个timeout字段每个技能按自身依赖特性声明。重试策略同理非幂等操作比如“发送邮件”这种只允许发一次的千万别自动重试否则用户会收到三封一模一样的邮件。区分幂等和非幂等操作是设计重试逻辑时的基本原则。4.5 日志别只打“成功”失败时要留全现场最怕看到这样的日志2025-01-10 11:22:33,456 ERROR skill failed完了没参数、没异常类型、没traceback等于啥也没记。建议错误日志至少包含技能名称、版本、本次执行的trace_id、完整入参脱敏后、异常堆栈。入参里如果涉及个人敏感信息要先做脱敏再落日志。这个补丁看起来简单排查问题时能救命的。5. 技能系统的进阶扩展方向如果你已经跑通了基础框架接下来想往生产级走有这么几个方向值得投入。5.1 技能自省与自动生成让Agent自己丰富技能库高级玩家已经在尝试让Agent在运行过程中发现“能力缺口”然后自动生成新技能来补上。实现路径大致是当Agent发现没有技能能处理当前任务时把任务保存到一个“未覆盖需求池”定期用大模型分析这批需求生成候选技能的骨架代码和描述再交给人工审核通过后入库。这样技能库会像一个活体组织一样随业务需要自动生长。不过这条路的坑在于自动生成的代码质量和安全性不可控需要非常严格的审核机制和沙箱隔离不是小团队能轻易驾驭的。5.2 技能组合编排把基础技能拼成复杂工作流单个技能解决单一问题复杂任务需要多个技能协同。进阶做法是引入“技能编排层”用DSL或图结构定义技能之间的依赖和执行顺序。比如“生成经营分析日报”技能内部会编排拉取数据data_fetch- 数据清洗data_clean- 指标计算metric_calc- 报告生成report_gen。有了编排层我们就能对整条链路做更精细的控制比如某一步失败时是重试还是走降级方案某两步之间要不要做数据一致性校验。编排层的设计可以借鉴工作流引擎的成熟方案比如Temporal、Airflow但注意别做得过重Agent的技能编排要快进快出。5.3 技能评测与持续优化把技能当成产品来运营技能上线不是终点而是要持续追踪它的真实效果定期迭代。我建议每个技能上线时都跑一轮离线基准测试用固定的测试集记录技能的正确率、耗时、成本线上运行后每天看指标波动每周选几个低分案例人工复盘分析是模型问题还是技能问题还是数据问题。这块可以参考传统MLOps的思路数据版本管理、模型版本管理、AB实验、灰度发布全都迁移到技能系统上来。技能系统的生命周期管理能力和模型一样重要只是很多人意识得比较晚。我之前在一个客户项目里就靠这个思路把一个文档解析技能从最初的72%准确率迭代到第三个月稳定在94%靠的就是持续采集失败样本、补充边界case、优化技能内部的预处理逻辑。这已经不是在“写代码”了而是在“运营技能资产”。写在最后的实际操作体会技能系统这个东西看着技术门槛好像不高但真正把一个技能体系从无到有搭起来、再跑稳过程中会遇到特别多书本上不讲的细节。我复盘了一下最值钱的体验有三件事第一接口规范化越早做越好。宁可一开始多花两天把技能基类和注册机制设计好也别图快先写业务技能。一旦业务技能多了再回头统一接口那个成本会是初期的好几倍。第二技能描述就是给模型看的“产品说明书”值得花时间打磨。同一个技能描述写得好和写得差在Agent任务上的成功率能有两位数的差距。写完之后拿几个典型任务跑一遍测试看看模型到底会不会选对这比看文档写得多漂亮重要得多。第三可观测性是对Agent系统最大的善意。Agent任务天然带随机性和长链路没有日志、追踪和指标这三件套出问题时你会变成盲人摸象。所以每次在企业里推行Agent方案我都会先说一句先把观测铺好再谈业务指标。现在这个框架我已经在几个线下项目里跑过企业内部知识库问答、工单自动分拣、报表生成助手都从这套技能系统里获益不少。如果你也在设计自己的Agent技能体系或者已经在用类似方案欢迎在评论区聊聊你的实践心得一起把这块的方法论再打磨得更扎实一点。
返回列表