ARTICLE DETAIL

资讯详情

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

Agent技能化架构实践:从单体逻辑到可插拔能力

Agent技能化架构实践:从单体逻辑到可插拔能力 做到第三个 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(fskill 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(packageskills): 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, timeout30, returnsNone): def decorator(fn): skill_obj Skill( namename, descriptiondescription, parametersparameters, fnfn, timeouttimeout, returnsreturns ) 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( modelgpt-4o, messagesmessages, toolstools, tool_choiceauto ) 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_asciiFalse) }) return 达到最大轮次任务未能完成。流程其实很朴素把技能清单和用户消息一起发给模型模型如果决定调用技能会返回技能名和参数主循环执行技能函数把结果作为tool消息放回去让模型继续推理直到模型认为已经完成任务输出最终回复。max_rounds是必须加的控制项防止 Agent 在多个技能之间无限循环。4. 实战多技能协同的研究助手框架最小化之后我用一个“研究助手”的例子验证整套设计。这个 Agent 的目标是用户给出一个主题它先搜索资料再打开几个优质链接提取正文然后生成摘要报告最后把报告存成 Markdown 文件。整个流程涉及四个技能串起来就是一条完整的工作链。4.1 需求拆解与技能清单拆分先把需求拆成技能这个步骤我一般遵循“一技能一职责”的原则避免把多个动作捆在一个技能里。技能名职责输入输出web_search搜索关键词返回结果列表query, top_ktitle, 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( nameweb_search, description在互联网上搜索给定的关键词返回前 N 条结果的标题、链接和摘要。适合查询实时信息、查找资料、获取新闻等场景。, parameters{ type: object, properties: { query: { type: string, description: 搜索关键词建议使用具体短语 }, top_k: { type: integer, description: 返回结果条数范围 1-10, default: 5 } }, required: [query] }, timeout15 ) def web_search(query: str, top_k: int 5): resp requests.get(SEARCH_API_URL, params{q: query, count: top_k}, timeout10) 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( namesummary_generate, description对输入的长文本生成简洁摘要保留关键事实和结论。适合需要快速理解长文内容的场景。, parameters{ type: object, properties: { text: {type: string, description: 待摘要的正文文本}, max_words: {type: integer, description: 摘要最大字数, default: 300} }, required: [text] }, timeout30 ) def summary_generate(text: str, max_words: int 300): resp client.chat.completions.create( modelgpt-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(querytopic, top_ktop_k) if not search_results or error in search_results[0]: return {error: 搜索阶段失败} collected [] for item in search_results: page page_fetch(urlitem[url], max_chars3000) 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(textcombined_text, max_words400) filepath freports/{topic.replace( , _)}.md saved report_save(filepathfilepath, contentf# {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, timeoutNone): timeout timeout or skill.timeout with ThreadPoolExecutor(max_workers1) as executor: future executor.submit(skill.fn, **args) try: result future.result(timeouttimeout) 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(instanceargs, schemaskill.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 推理加速每一块都有不少值得写的东西。等实践再深一点我回来继续总结。
返回列表