做到第三个 Agent 项目的时候,我彻底被一件事逼疯了:所有能力都写在主循环里。搜索、读取网页、调数据库、生成报告,每一个功能都堆在同一个while True里,加一个新功能就要动主流程,改一个参数可能影响三个地方。后来我把agent-skills这套思路落进项目——把每一项能力都拆成独立、自描述、可插拔的“技能”,Agent 本身只剩调度逻辑。这篇文章就是这套方案的完整复盘:从技能层的设计动机、字段规范,到最小框架的注册、加载、调用,再到一个多技能协同的实战案例,以及我在真实项目里踩过的几个比较隐蔽的坑。适合正在做 Agent、或者想把现有 LLM 应用改造成技能化架构的开发者参考。
1. 技能层为何值得单独设计:从单体逻辑到可插拔能力
先说一个我自己的反面案例。第一个版本的 Agent,我管它叫“上帝对象”:一个run()方法里依次判断意图、调函数、拼 prompt、解析结果。一开始只有两三个功能的时候没什么感觉,等用户开始提各种需求,功能列表膨胀到十几个,主函数长到六百多行。每次新增功能,我要担心的不是新功能本身,而是它会不会踩到旧分支的判断逻辑。后面我切到技能化架构,才想明白这件事的本质。
1.1 第一版 Agent 是怎么变成一团乱麻的
那时候的代码结构大概是这样的:一个巨大的if/elif链,每个分支对应一类用户意图。搜索放一个分支,写文件放一个分支,查天气放一个分支。表面上看起来还挺有组织,但它有几个很难修补的问题。
第一,能力之间没有隔离。A 分支里不小心改了一个全局变量,B 分支的行为就变了。排查这种问题只能靠人肉 debug,效率很低。第二,模型调用和业务逻辑完全耦合。意图识别、参数抽取、函数执行全在一个上下文里,prompt 稍长一点,上下文窗口就被占掉大半。第三,能力复用基本靠复制粘贴。两个功能都要用 HTTP 请求的时候,拷贝一份代码再改改,一旦公共逻辑有 bug,每个副本都要修一遍。
这种结构在 Demo 阶段完全够用,但一旦进入真实业务,维护成本会明显上升。技能化的第一个价值,就是把“能力”从“主流程”里抽出来,变成独立存在的东西,Agent 只是执行者,不再被细节淹没。
1.2 技能层本质是把“能力”变成“资产”
把技能独立出来以后,能力就不再是散落在代码里的函数,而是一份有名字、有描述、有输入输出协议、可以被注册和发现的“资产”。你可以类比成给 Agent 装上了可拔插的肢体:需要搜索就插上搜索技能,需要读文件就挂载文件技能,不需要的功能直接从注册表下掉,主流程一行都不用改。
这种设计带来的直接收益是在扩展性上。团队里另一个同事写了一个新技能,只要他遵循同样的规范,放进技能目录就能被 Agent 加载,不需要我来改主循环。技能本身也可以被多个 Agent 共享,比如一个抓取网页的技能,数据分析 Agent 能用,问答 Agent 也能用。这种复用性在半年前那个单体代码里是完全没法想象的。
更重要的一个点,是技能层让“试错”变得便宜。新的思路先写成一个技能试跑,效果不好就下掉,不会污染主流程。我后来做实验经常同时挂十来个技能,白天调整 prompt,晚上只换技能配置,Agent 主体代码几乎没动过。
1.3 技能和工具、插件、工作流的边界划分
很多人会问:技能和工具(Tool)、插件(Plugin)、工作流(Workflow)有什么区别?我的理解比较务实,它们不是互斥概念,而是不同层级的抽象。
工具是技能的底层“动作”,比如一次 HTTP 请求、一次文件读写。技能则是围绕一个完整目标组织的“最小可用能力单元”,比如“搜索并返回摘要”本身可以理解为多个工具动作的组合,但对上层 Agent 来说它就是一个技能。插件更像是分发和打包的载体,把一组技能连同配置、依赖一起分发。工作流则是对多个技能的编排,定义它们按什么顺序执行、什么情况下跳转。
我在框架里不区分 tool 和 skill,统一叫 skill,但在技能内部可以调用底层工具函数。这样既保证上层接口统一,也不会把粒度搞得太碎。
2. 技能定义规范:元数据、参数契约与自描述
技能之所以能被 Agent 正确调用,靠的不是写代码的人自觉,而是它会“自我介绍”。我给每个技能配一份清单,包含名字、描述、参数 Schema、返回格式、超时时间。这份清单既给 LLM 看,也给代码看,两边用的同一个契约。
2.1 一份 Skill 的字段设计
下面是我实际在用的一份技能清单,用 JSON 表示:
{ "name": "web_search", "description": "在互联网上搜索给定的关键词,返回前 N 条结果的标题、链接和摘要。适合查询实时信息、查找资料、获取新闻等场景。", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词,建议使用具体短语而非自然语言长句" }, "top_k": { "type": "integer", "description": "返回结果条数,范围 1-10", "default": 5 } }, "required": ["query"] }, "returns": { "type": "array", "items": { "type": "object", "properties": { "title": {"type": "string"}, "url": {"type": "string"}, "snippet": {"type": "string"} } } }, "timeout": 15 }name是全局唯一标识,所有路由都靠它。description写清楚这个技能负责什么、适合哪些场景,它直接影响到大模型能不能在多个技能里选中它。parameters用的是 JSON Schema,既能给模型做参数生成约束,又能给运行时做校验。returns描述返回结构,模型拿到结果后可以基于结构做下一步推理。timeout是技能的最长执行时间,防止某个技能卡死把整个 Agent 拖停。
2.2 输入输出契约为什么是技能可用的前提
我自己一开始偷懒,参数只写了名字和类型,没有描述。结果模型在调用时经常把参数填反,比如把top_k填成搜索词,或者把日期格式传成2024-1-1,而技能内部等的是2024-01-01。加了描述和格式示例之后,错误率明显下降。
输入输出契约的另一个作用是把“错误拦截在边界上”。技能执行前先做一次参数校验,不符合 Schema 的直接拒绝,不让脏数据进到业务逻辑里。执行后的返回结果也尽量结构化成 JSON,而不是丢一段文本让模型自己猜。这样每一步的输入输出都可观测、可记录、可重放。调试时把技能调用日志打出来,一眼就能看出是哪一环出了问题。
2.3 描述文本里的玄机:写得越好,模型选得越准
我想强调一下description这个字段,它是我觉得性价比最高的调参位。同一个技能,描述写“执行搜索”和写“在互联网上搜索给定的关键词,返回前 N 条结果的标题、链接和摘要。适合查询实时信息、查找资料、获取新闻等场景”,模型选择它的准确率完全不是一个级别。
原因不难理解,大模型在选择技能时,本质上是在做语义匹配。描述里多给一些“触发场景词”,比如“新闻”“资料”“实时信息”,用户问题里出现这些词时,模型就更容易把这个技能排在候选前列。但描述也不是越长越好,写两到三句话足够,太长反而会稀释关键信息,模型可能抓不住重点。
我内部有个习惯:写完一个技能,先拿几个典型问题测试它会不会被选中。如果选不中,我优先怀疑描述写偏了,而不是怀疑模型能力。这跟给函数写 docstring 很像,但 docstring 是给程序员看的,技能描述是给大模型看的,服务对象变了,写法和侧重点也完全不一样。
3. agent-skills 最小框架:注册、加载、调用的三件套
理论说再多,不如直接跑起来。这一节我会带你从零搭一个最小可用的 agent-skills 框架,核心只有三个模块:注册中心、动态加载器、模型调用路由。代码量不大,但该有的设计都在。
3.1 注册中心:让 Agent 知道“我有什么”
注册中心是一个全局技能表,负责存储和索引所有技能。我在实现中使用了一个简单的字典,键是技能名,值是技能对象。技能对象至少包含清单和可调用函数两部分。
# skill_registry.py from typing import Dict, Optional, List class Skill: def __init__(self, name: str, description: str, parameters: dict, fn, timeout: int = 30, returns: Optional[dict] = None): self.name = name self.description = description self.parameters = parameters self.fn = fn self.timeout = timeout self.returns = returns def to_openai_tool(self) -> dict: return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": self.parameters } } class SkillRegistry: def __init__(self): self._skills: Dict[str, Skill] = {} def register(self, skill: Skill) -> Skill: if skill.name in self._skills: raise ValueError(f"skill name conflict: {skill.name}") self._skills[skill.name] = skill return skill def get(self, name: str) -> Optional[Skill]: return self._skills.get(name) def list_skills(self) -> List[Skill]: return list(self._skills.values()) registry = SkillRegistry()注意register里的重名检查,这个后面踩坑部分我会单独展开。注册中心同时承担“模型可见技能列表”的职责,把所有技能的清单转成 OpenAI 函数调用格式,模型就能看到这些技能。
3.2 动态加载器:import 之前先想清楚这三件事
注册中心提供的是“数据结构”,技能代码从哪来?最简单的方式是手动 import 再注册,但项目一大会变得很乱。我采用目录扫描的加载器,把skills/目录下的每个.py文件当作一个技能模块加载。
# skill_loader.py import importlib import inspect import pkgutil import skills from skill_registry import registry def load_skills_from_package(package=skills): for module_info in pkgutil.iter_modules(package.__path__): module = importlib.import_module(f"{package.__name__}.{module_info.name}") for _, obj in inspect.getmembers(module): if hasattr(obj, "__skill_manifest__"): skill = obj.__skill_manifest__ registry.register(skill)__skill_manifest__是绑定在函数上的技能清单。为了让装饰器写法更顺手,我还加了一个@skill装饰器,把清单和函数打包成 Skill 放进注册中心。
# skill_decorator.py from skill_registry import Skill, registry def skill(name, description, parameters, timeout=30, returns=None): def decorator(fn): skill_obj = Skill( name=name, description=description, parameters=parameters, fn=fn, timeout=timeout, returns=returns ) registry.register(skill_obj) return fn return decorator动态加载虽然方便,但有三个细节必须注意。第一是模块之间的依赖,技能用到第三方库时,要么把依赖写进项目配置,要么让技能内部做延迟 import,否则启动就会报错。第二是名称空间隔离,不同技能模块里尽量别定义同名全局变量,加载顺序不同会导致行为不一致。第三是热加载,生产环境不建议每次请求都重新扫描目录,一般启动时扫一次就够了,训练模型也好、调试也好,都依赖这份静态快照。
3.3 让 LLM 通过函数调用路由到技能
注册和加载做完,轮到核心的部分:怎么让模型决定调用哪个技能、传什么参数。现在主流做法是函数调用(Function Calling / Tool Use),我在示例里用 OpenAI 风格的接口,其他家的实现思路基本一致。
# agent.py import json from openai import OpenAI from skill_registry import registry from skill_loader import load_skills_from_package load_skills_from_package() client = OpenAI() tools = [skill.to_openai_tool() for skill in registry.list_skills()] def run_agent(user_message: str, max_rounds: int = 5): messages = [{"role": "user", "content": user_message}] for _ in range(max_rounds): response = client.chat.completions.create( model="gpt-4o", messages=messages, tools=tools, tool_choice="auto" ) message = response.choices[0].message if not message.tool_calls: return message.content messages.append(message) for tool_call in message.tool_calls: skill = registry.get(tool_call.function.name) if skill is None: messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps({"error": "skill not found"}) }) continue args = json.loads(tool_call.function.arguments) result = skill.fn(**args) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) return "达到最大轮次,任务未能完成。"流程其实很朴素:把技能清单和用户消息一起发给模型,模型如果决定调用技能,会返回技能名和参数;主循环执行技能函数,把结果作为tool消息放回去,让模型继续推理;直到模型认为已经完成任务,输出最终回复。max_rounds是必须加的控制项,防止 Agent 在多个技能之间无限循环。
4. 实战:多技能协同的研究助手
框架最小化之后,我用一个“研究助手”的例子验证整套设计。这个 Agent 的目标是:用户给出一个主题,它先搜索资料,再打开几个优质链接提取正文,然后生成摘要报告,最后把报告存成 Markdown 文件。整个流程涉及四个技能,串起来就是一条完整的工作链。
4.1 需求拆解与技能清单拆分
先把需求拆成技能,这个步骤我一般遵循“一技能一职责”的原则,避免把多个动作捆在一个技能里。
| 技能名 | 职责 | 输入 | 输出 |
|---|---|---|---|
| web_search | 搜索关键词,返回结果列表 | query, top_k | title, url, snippet 数组 |
| page_fetch | 抓取单个网页正文文本 | url, max_chars | 标题、正文、抓取状态 |
| summary_generate | 对长文本生成摘要 | text, max_words | 摘要文本 |
| report_save | 将内容写入本地 Markdown 文件 | filepath, content | 文件路径、写入状态 |
为什么拆成四个而不是合成一个“研究主题”技能?因为可组合性更强。web_search和page_fetch可以被其他 Agent 复用,summary_generate也可以单独用于文档总结场景。如果合成一个大技能,每次想调整其中一环,都得动整个技能,灵活性大打折扣。
4.2 两个代表性技能的实现细节
web_search是典型的“外部 API 封装型”技能,我直接调搜索接口,把结果清洗成统一结构:
# skills/web_search.py import requests from skill_decorator import skill @skill( name="web_search", description="在互联网上搜索给定的关键词,返回前 N 条结果的标题、链接和摘要。适合查询实时信息、查找资料、获取新闻等场景。", parameters={ "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词,建议使用具体短语" }, "top_k": { "type": "integer", "description": "返回结果条数,范围 1-10", "default": 5 } }, "required": ["query"] }, timeout=15 ) def web_search(query: str, top_k: int = 5): resp = requests.get(SEARCH_API_URL, params={"q": query, "count": top_k}, timeout=10) data = resp.json() results = [] for item in data.get("items", [])[:top_k]: results.append({ "title": item.get("title"), "url": item.get("link"), "snippet": item.get("snippet") }) return results这里有个小细节:技能内部尽量做异常捕获和兜底返回,不要让 HTTP 异常直接抛到 Agent 主流程。返回结构尽量固定,就算搜索失败也返回{"error": "..."}而不是抛异常,后面模型拿到错误信息可以自己决定下一步怎么办。
summary_generate则是纯 LLM 调用型技能,它内部调用一次大模型文本接口,跟 Agent 主调度逻辑完全解耦:
# skills/summary_generate.py from skill_decorator import skill @skill( name="summary_generate", description="对输入的长文本生成简洁摘要,保留关键事实和结论。适合需要快速理解长文内容的场景。", parameters={ "type": "object", "properties": { "text": {"type": "string", "description": "待摘要的正文文本"}, "max_words": {"type": "integer", "description": "摘要最大字数", "default": 300} }, "required": ["text"] }, timeout=30 ) def summary_generate(text: str, max_words: int = 300): resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": f"你是一个摘要助手,请将用户的文本压缩到 {max_words} 字以内,保留核心信息。"}, {"role": "user", "content": text[:8000]} ] ) return {"summary": resp.choices[0].message.content}技能内部自己调模型这件事,我一开始也有点纠结,觉得是不是绕了。后来想明白:技能是能力单元,它可以用任何手段完成职责,包括调用另一个模型。这样 Agent 主模型负责“决定做什么”,技能内部的小模型负责“把活干完”,职责非常清楚。
4.3 编排模式:串行、条件跳转和结果汇总
四个技能都有了,Agent 怎么把它们串起来?最简单的是让模型自己编排,主循环里拿到用户消息后模型会自动决定先搜、再抓、再摘要、再保存。但有些场景我们希望流程稳定,比如内部工具调用不想让模型临时发挥,这时可以在技能层做一个轻量编排器,写一个research_pipeline技能,内部显式调用其他技能的函数。
# skills/research_pipeline.py from skills.web_search import web_search from skills.page_fetch import page_fetch from skills.summary_generate import summary_generate from skills.report_save import report_save def research_pipeline(topic: str, top_k: int = 3): search_results = web_search(query=topic, top_k=top_k) if not search_results or "error" in search_results[0]: return {"error": "搜索阶段失败"} collected = [] for item in search_results: page = page_fetch(url=item["url"], max_chars=3000) if page.get("status") == "ok": collected.append({ "title": page["title"], "url": item["url"], "text": page["content"] }) if not collected: return {"error": "没有抓到任何网页正文"} combined_text = "\n\n".join([f"[{c['title']}]({c['url']})\n{c['text']}" for c in collected]) summary = summary_generate(text=combined_text, max_words=400) filepath = f"reports/{topic.replace(' ', '_')}.md" saved = report_save(filepath=filepath, content=f"# {topic}\n\n{summary['summary']}\n\n## 来源\n\n" + "\n".join([f"- [{c['title']}]({c['url']})" for c in collected])) return {"report_path": saved["path"], "summary": summary["summary"]}这里其实是两种编排模式的结合:模型自由编排适合探索性任务,代码显式编排适合稳定性要求高的任务。我在生产环境里大多数情况会让模型做高层决策、代码做底层时序,比如模型决定调用research_pipeline,剩下的步骤全部由技能内部按序完成。这样模型要决策的次数变少了,出错率也会降下来。
5. 真实项目里的踩坑记录与加固方案
框架看着简单,真正放到生产环境里跑,问题一套一套地来。我挑几个印象最深的:技能同名冲突、超时黑洞、模型幻觉参数,还有一个比较容易被忽略的降级问题。
5.1 技能同名冲突:注册中心里最隐蔽的地雷
有一次我把项目从单体改成技能化,加载时发现send_message这个技能被注册了两次,一个来自内部通知模块,一个来自外部客服模块。两个模块的行为完全不同,但名字一样,注册中心如果不加保护,后加载的会直接覆盖先加载的,Agent 调用时拿到的是哪个技能就完全取决于加载顺序。
这是为什么我在register里加了重名检查。如果你在跑多个技能包,我建议在命名上做一个约定,比如按领域加前缀:billing_query、crm_create_contact,降低跨团队重名的概率。注册中心启动时如果检测到冲突,直接报错而不是静默覆盖,这样问题能在集成阶段暴露,而不是跑到线上才暴露。
5.2 长耗时技能的默认两分钟黑洞
技能超时这个话题,我是被真实事故教育过的。有一个数据导出技能,正常情况几秒钟返回,但遇到超大数据集时能跑两分钟。Agent 主循环在等它返回时完全没有超时控制,结果就是整整两分钟卡死,用户的请求一直挂着,日志里什么都看不到。
后来我给技能执行包了一层带超时的执行器,Python 里可以用concurrent.futures实现。
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeoutError def execute_with_timeout(skill, args, timeout=None): timeout = timeout or skill.timeout with ThreadPoolExecutor(max_workers=1) as executor: future = executor.submit(skill.fn, **args) try: result = future.result(timeout=timeout) return {"status": "ok", "result": result} except FutureTimeoutError: return {"status": "timeout", "error": f"技能 {skill.name} 执行超过 {timeout} 秒"}返回timeout状态给模型后,模型会尝试换个方案,比如先返回部分结果,或者让用户缩小范围,而不是整个 Agent 卡死。超时时间也不能一刀切,搜索类给 15 秒,生成类给 60 秒,复杂报告可以放宽到 120 秒,尽量让技能把timeout写在清单里。
5.3 模型幻想参数:契约校验是你的安全带
大模型在生成函数参数时,偶尔会出现“一本正经地编参数”的情况。比如技能只需要两个字段,模型却传了五个;再比如top_k明明限定 1-10,它传了个 20。如果不做校验,技能内部可能因为异常参数直接崩溃,或者因为类型不对产生很难发现的隐性 bug。
我在调用技能前加了一层 JSON Schema 校验,Python 里用jsonschema库:
from jsonschema import validate, ValidationError def safe_call_skill(skill, raw_arguments: str): try: args = json.loads(raw_arguments) except json.JSONDecodeError: return {"status": "error", "error": "参数不是合法 JSON"} try: validate(instance=args, schema=skill.parameters) except ValidationError as e: return {"status": "error", "error": f"参数校验失败: {e.message}"} result = skill.fn(**args) return {"status": "ok", "result": result}校验失败时直接把错误信息抛给模型,模型看到“参数校验失败:20 大于 maximum 10”之后,通常会在下一轮自我纠正。比起默默让异常参数流入业务逻辑,这种“显式报错+让模型修正”的方式,在多次运行里明显更稳定。
5.4 技能失败不拖垮主流程:降级与局部重试
最后是降级意识。我早期的技能实现里,page_fetch一旦一个链接打不开,整个研究助手就失败了。后来我在脚本里做了两处调整:单个链接失败只跳过该链接、记录日志,不强抛异常;research_pipeline里对web_search和page_fetch各做了一次局部重试,第一次失败间隔 1 秒再试一次,重试仍失败才返回错误。
这些细节加起来,用户体验完全是两个等级。未加固的版本三天两头全链路报错,加固之后的版本即便部分来源失效,Agent 也能用剩余的资料完成任务,并且在报告里标注“部分来源抓取失败”,让用户知道信息不完整。
我在跑这套 agent-skills 架构大半年之后最大的体会是:技能化表面上是代码组织方式的改变,实际上是把 Agent 的可靠性问题从主流程内拆解到了每个技能边界上。每个技能做好自己的输入输出契约、超时控制、失败兜底,整个 Agent 才会在真实业务里立得住。这个方向后来我还在继续扩展:技能间共享内存怎么设计、技能版本如何灰度、哪些技能适合 GPU 推理加速,每一块都有不少值得写的东西。等实践再深一点,我回来继续总结。