免费获取学习方案
ARTICLE DETAIL

资讯详情

深耕编程基础知识与建站技术分享的一线实战洞察。

AI Agent技能管理器:从散装工具到集中注册与可视化调试

AI Agent技能管理器:从散装工具到集中注册与可视化调试 做 AI Agent 最头疼的其实不是把 Agent 跑通而是技能越挂越多以后整个项目变成一团乱麻。我手头维护过两套 Agent 项目一套跑在 LangGraph 上一套是偏 Java 工程化的 Spring AI 项目两边的工具函数加起来六十多个。每次新加一个技能要翻半天代码改一个函数签名不知道会影响哪些 Agent 场景线上出问题也只能靠日志里的蛛丝马迹去猜是哪一次工具调用歪了。后来我下定决心给所有 Agent 单独做一个可视化技能管理器把技能集中注册、集中管理、可视化调试。这大概是今年我做过回报率最高的一件事今天把完整的思路和落地过程整理出来给正在搭 Agent 平台的朋友参考。文章会覆盖几个偏实战的层面为什么分散的技能是隐患、技能注册表的数据模型怎么设计、可视化面板都做哪些事、怎么接入 LangChain / LangGraph / FastAPI以及并发和生产环境里必须处理的问题。适合已经在用工具调用、想让 Agent 技能变得可维护的人看也适合准备搭建 Agent 中台、想做内部技能市场的人参考。1. 为什么 Agent 需要技能管理器而不是一堆散装工具函数先说结论工具函数是 Agent 的肌肉技能管理器是神经系统。肌肉再多没有统一的调度和感知动作一定是混乱的。1.1 散装工具函数的四个典型痛点我接触过的项目里技能几乎都是长出来的不是设计出来的。刚开始只有一两个函数直接塞进 prompt 就完事后来函数越来越多、项目越来越复杂就会撞上几个典型问题。第一技能散落在各个角落。有的写在 prompt 的 system 里有的写在 chain 里面有的直接挂在 service 层还有的是其他项目通过 HTTP 暴露出的小接口。你想知道当前 Agent 到底有哪些技能没有一个统一答案。第二同名技能、不同实现冲突得莫名其妙。我在实际项目里见过三份get_stock_price一个在 A 项目返回字符串一个在 B 项目返回 JSON还有一个在 C 项目里已经废弃但没人删。LLM 根据描述选工具时很可能选中一个不是你期望的实现。第三上线下线全靠改代码。想停用一个坏掉的技能最快的办法是注释代码重新发布。一旦这个技能还被其他 Agent 引用那就不只是停用一个函数的问题而是动一发牵全身的发布操作。第四无法观测和复盘。用户问了一句帮我查下这个股票Agent 到底调了哪个技能、传了什么参数、返回了什么结果、耗时多少、成本多少散装状态下基本是一笔糊涂账。出问题只能让用户复现体验极其痛苦。1.2 技能管理器的三个角色定位我设计的技能管理器本质上承担三个角色。注册中心。所有技能必须在这里登记包括技能 ID、名称、用途描述、参数 Schema、版本号、所属领域、超时时间、限流策略、预估成本。任何 Agent 要使用技能只能通过注册中心获取技能清单不允许绕过。调度中心。所有技能调用的入口统一收口权限校验、参数校验、超时控制、限流熔断、调用日志全部在这里完成。Agent 侧只需要说我要调某个技能管理器负责把这件事安全地执行掉。监控台。每次调用都会沉淀一条结构化记录哪次会话、哪个 Agent、选了哪个技能、执行成功还是失败、耗时多少、返回了什么摘要。这一层数据是后续做技能健康度、成本分析、灰度发布的基础。1.3 和 RAG、MCP 怎么区分很多朋友会把技能管理器和 RAG、MCP 混在一起我简单捋一下关系。RAG 解决的是让 Agent 知道什么的问题技能管理器解决的是让 Agent 能做什么的问题。一个是被动检索的知识库一个是主动执行的动作集合两者可以共存但不该相互替代。MCP 则是这两年很火的工具通信协议。技能管理器完全可以兼容 MCP把 MCP Server 当作一种特殊的技能适配器注册进来Agent 调技能时管理器走 MCP 协议把请求转发出去。换句话说技能管理器是管理动作的载体MCP 是传输动作的语言两者是上下层关系。2. 技能注册表一个技能该怎么被描述Agent 才能准确调用可视化只是表象技能管理器真正的内核是一个设计良好的注册表。注册表里的每个字段都直接影响 LLM 能不能准确选对技能。2.1 技能声明的最小数据模型我先把最初的 SkillSpec 模型贴出来后面所有逻辑都围绕它展开。# skill_registry.py from pydantic import BaseModel, Field from typing import Callable, Optional class ParameterSpec(BaseModel): name: str Field(description参数名) type: str Field(description参数类型: string/number/integer/boolean) required: bool True default: Optional[str] None enum: Optional[list[str]] None description: str Field(description参数含义、格式要求、示例) class SkillSpec(BaseModel): skill_id: str Field(description全局唯一技能ID例如 stock.get_price) name: str Field(description技能短名) version: str 1.0.0 description: str Field(description给LLM看的技能说明决定工具选择准确率) parameters: list[ParameterSpec] [] handler: str Field(default, description对应handler的注册名) enabled: bool True timeout: float 10.0 tags: list[str] [] owner: str default fail_count: int 0不要小看skill_id的命名我统一采用领域.动作的格式比如stock.get_price、order.query_status、payment.trigger_refund。这么做的好处是技能树可以直接按领域前缀分组检索也方便。handler字段存的是 handler 的注册名不是函数对象本身。这样技能注册表可以序列化存储到数据库或 YAML 文件进程重启后可以重新加载而不是把所有函数都塞进内存里等 GC 回收。2.2 参数 Schema 的松紧程度直接决定幻觉率LLM 选完技能之后要填参数参数 Schema 写得太松模型就会自由发挥。我在参数约束上吃过亏现在坚持三条原则。能用enum锁死的值绝对不用自由文本。比如查询货币对直接给出enum: [USD, CNY, EUR, JPY]模型就不会臆造一个USDT出来。required要宁紧勿松。一个参数如果既可以不传又可以传模型大概率会尝试各种组合业务接口根本吃不消。上线前我会把所有参数过一遍能设默认值的设默认值不能缺的坚决 required。description里要写格式 示例。只说日期模型可能填2024年1月1日但接口要的是YYYY-MM-DD。正确写法是参数类型: string, 必填 格式: YYYY-MM-DD, 例如 2026-03-18 含义: 查询区间的起始日期2.3 技能描述的质量决定了 Agent 是精准调用还是抽签这是整篇博文里我最想强调的一点。LLM 选技能本质上是在做一次基于文本相似度的匹配。你给它的描述越具体它越能明确地判断这个场景该用我。看两个对比。差的描述获取天气信息好的描述获取指定位城市的当前天气与未来3天预报。 触发场景: 用户询问天气、温度、降雨概率、是否适合出门。 参数要求: city 必须为中文城市名, 例如北京、上海, 不接受拼音或英文。 返回结构: 包含 temperature(摄氏度)、humidity、condition(晴/多云/雨)。同样的技能描述改完之后Agent 在天气这个场景的选对率几乎是肉眼可见地提升。原因很简单LLM 不知道你背后代码长什么样它只能根据 description 和参数名来做决策。你把触发词、参数格式、返回结构都写清楚它就没有猜的空间。2.4 技能的生命周期状态注册表不是一张死表技能是有生命周期的。我设计了几个基础状态。状态含义对 Agent 的影响enabled正常可用会出现在工具清单里可以被 LLM 选择disabled手动停用从工具清单中剔除Agent 无法调用degraded部分降级保留在清单中但调用走兜底实现deprecated即将下线不在新会话中展示旧会话仍可用这个状态机看起来简单实际帮了大忙。有一次上游行情接口出了问题我直接把相关的行情技能置为 disabledAgent 立刻就不会再选它了整个过程五分钟搞定不用改代码、不用发布版本。这在散装方案下是做不到的。3. 可视化面板设计思路技能树、检索与调试台技能管理器如果只有 API那它只是一个高级工具类谈不上可视化。这一节讲前端交互层我做了哪些事情以及为什么每件事都有实际价值。3.1 技能按领域分组而不是按文件分组技能树是最直观的视图。我按skill_id的前缀把技能组织成树形结构stock股票领域 ├── get_price获取行情 ├── get_kline获取K线 └── get_market_status判断是否开市 order订单领域 ├── query_status查询订单状态 └── create_order创建订单这个树不需要递归维护注册的时候写清楚tags和skill_id前端按规则展开就行。分组的意义在于你可以一眼看出这个领域下到底有哪些能力而不是在一个平铺的 list 里翻来翻去。3.2 状态视图和检索过滤面板的右侧我做了一个过滤区支持按状态过滤启用/停用/异常、按领域过滤、按关键词搜索。页面同时展示每个技能的版本、近 24 小时调用次数、成功率、平均耗时。这几个指标很关键。你会发现有些技能调用次数特别少、但延迟特别高这种技能往往是听起来有用、用起来难的鸡肋可以考虑合并或优化描述。另外异常率高的技能会直接标红提醒你去检查。3.3 调试台不跑完整 Agent也能单独验证技能可视化面板给我带来的最大效率提升是调试台。以前测一个技能要先启动整个 Agent再拼一段 prompt 让 LLM 选中这个技能链路长、变量多而且经常分不清是 LLM 选错了还是技能本身有问题。调试台的做法是选中一个技能面板自动根据参数 Schema 生成表单你填好参数后点运行管理器直接调用 handler 并展示返回结果。这样技能本身的正确性可以独立验证和 Agent 的决策逻辑解耦。前端发起的请求就是调 FastAPI 的curl -X POST http://localhost:8000/api/skills/stock.get_price/invoke \ -H Content-Type: application/json \ -d {symbol: AAPL, exchange: NASDAQ}3.4 调用链路的可视化回放除了技能的静态信息我还会展示动态的调用链路。每次 Agent 会话结束管理器会把一次完整的调用过程沉淀为一条 trace{ session_id: sess_20260318_a1b2c3, agent_name: customer_service_bot, steps: [ {type: llm_decision, input_text: 用户问明天会不会下雨, selected: weather.get_forecast, reason: 询问降雨概率}, {type: skill_invoke, skill_id: weather.get_forecast, params: {city: 北京}, status: success, duration_ms: 320} ] }面板上按时间轴渲染这段链路每次 LLM 决策和技能调用都排成一条线。用户说了一句模糊的话Agent 为什么选了 A 技能而不是 B 技能可以看到模型给出的 reason技能失败时可以看到是参数配错还是上游超时。这个能力在排查线上问题时几乎不可替代。提示不要把完整的 prompt 和原始返回都塞进 trace数据量会爆掉。我一般只存 LLM 的最终决策摘要和技能执行的精简结果。4. 统一调用层如何接入 LangChain、LangGraph 和 FastAPI技能管理器不能只停留在管理层面它必须成为所有 Agent 的实际调用入口。这一节说清楚 ToolRegistry 的核心实现以及怎么跟主流框架对接。4.1 ToolRegistry 的核心实现整个管理器的心脏是一张技能注册表我封装成了ToolRegistry类。它负责三件事注册技能、生成给 LLM 的工具描述、统一执行调用。import asyncio import time from typing import Any, Callable, Optional class SkillDisabledError(Exception): pass class ToolRegistry: def __init__(self): self._specs: dict[str, SkillSpec] {} self._handlers: dict[str, Callable] {} self._call_logs: list[dict] [] def register(self, spec: SkillSpec, handler: Callable) - None: key spec.handler or spec.skill_id self._specs[key] spec self._handlers[key] handler def get_specs(self, enabled_only: bool True) - list[SkillSpec]: specs list(self._specs.values()) if enabled_only: specs [s for s in specs if s.enabled] return specs def to_openai_tools(self) - list[dict]: tools [] for spec in self.get_specs(enabled_onlyTrue): tools.append({ type: function, function: { name: spec.skill_id, description: spec.description, parameters: self._build_schema(spec), }, }) return tools def _build_schema(self, spec: SkillSpec) - dict: properties {} required [] for p in spec.parameters: prop: dict[str, Any] {type: p.type, description: p.description} if p.enum: prop[enum] p.enum if p.default is not None: prop[default] p.default properties[p.name] prop if p.required: required.append(p.name) return {type: object, properties: properties, required: required} async def invoke(self, skill_id: str, **params: Any) - Any: spec self._specs.get(skill_id) if not spec: raise KeyError(fskill not found: {skill_id}) if not spec.enabled: raise SkillDisabledError(fskill disabled: {skill_id}) handler self._handlers[skill_id] start time.monotonic() try: result await asyncio.wait_for(handler(**params), timeoutspec.timeout) status success note except Exception as exc: status error note str(exc)[:500] raise finally: self._call_logs.append({ skill_id: skill_id, status: status, note: note, duration_ms: int((time.monotonic() - start) * 1000), ts: time.time(), }) return result注意invoke里用了asyncio.wait_for这是技能级超时的最后一道防线。很多技能的 handler 是普通 Python 函数可能同步阻塞也可能内部慢到拖垮整个 Agent 会话。没有超时控制一个坏技能会让连带伤害辐射到所有用户。4.2 接入 LangChain直接复用 OpenAI Tools 格式LangChain 的工具调用体系基本兼容 OpenAI 的 function calling 格式所以注册表导出的to_openai_tools()可以直接喂给bind_tools。from langchain_openai import ChatOpenAI registry ToolRegistry() # 这里只做示例实际注册在启动时完成 # registry.register(stock_price_spec, stock_price_handler) llm ChatOpenAI(modelgpt-4o, temperature0) agent_tools registry.to_openai_tools() llm_with_tools llm.bind_tools(agent_tools)LangGraph 的玩法更灵活一点。你可以把选技能和执行技能拆成两个节点一个节点让 LLM 输出 tool_call另一个节点把参数传给registry.invoke去执行。这样 LangGraph 的图中流动的只是技能 ID 和参数具体技能实现交给注册表图结构会非常干净。4.3 FastAPI 统一入口顺便把面板的 API 一起做掉技能管理器本身我直接用 FastAPI 暴露成一组服务既给内部人当管理后台也给 Agent 做统一调用入口。from fastapi import FastAPI, HTTPException app FastAPI(titleSkill Manager) app.get(/api/skills) async def list_skills(): return [spec.model_dump() for spec in registry.get_specs(enabled_onlyFalse)] app.post(/api/skills/{skill_id}/invoke) async def invoke_skill(skill_id: str, payload: dict): try: result await registry.invoke(skill_id, **payload) except SkillDisabledError as e: raise HTTPException(status_code403, detailstr(e)) except Exception as e: raise HTTPException(status_code500, detailstr(e)) return {skill_id: skill_id, result: result} app.get(/api/trace) async def list_traces(limit: int 50): return list(reversed(registry._call_logs[-limit:]))这个/api/skills/{skill_id}/invoke就是所有 Agent 的最终调用入口。不管上层是 LangChain、LangGraph 还是自己写的 agent loop最终都走这个接口。好处是安全策略鉴权、限流、审计只要在管理器里做一次所有 Agent 就都能继承。4.4 与 Spring AI Agent 生态的对接思路现在很多团队核心业务用 JavaSpring AI 的 Agent 应用越来越多。我的实践是技能管理器作为独立的中心化服务部署Java 侧只写一个很薄的 Adapter。Spring AI 里有一个ToolCallback的抽象Adapter 做的事情就是把注册表返回的技能清单转换成ToolCallback调用时再把参数通过 HTTP 转发给管理器的 invoke 接口。思路和 LangChain 侧完全一致中心化注册、语言侧适配。这样即使同时存在 LangChain 项目和 Spring AI 项目技能只有一份不存在两边各写一套工具函数的问题。5. 并发与生产化Agent 到底怎么扛住高并发标题下面的热搜词里有个问题我一直觉得很有意思AI Agent 怎么扛并发。我的回答是Agent 框架本身并不天然是并发瓶颈真正被高并发打垮的往往是技能调用链路里的下游依赖。5.1 先搞清楚瓶颈在哪里Agent 的一次请求大概会经历LLM 生成决策、调用技能、LLM 再总结。技能管理器在这条链路里做的是编排和调度它自身是无状态的可以水平扩展。真正吃性能的是两件事LLM 调用本身这是大头。技能 handler 内部访问的第三方服务、数据库、缓存这些下游资源的并发上限决定了你 Agent 的天花板。所以技能管理器要做的是保护下游而不是简单地开更多线程。5.2 同步 handler 与 asyncio 的混用处理我见过不少团队在 FastAPI 里强行把所有 handler 写成 async结果技能内部调用的是requests照样阻塞事件循环。更合理的做法是区分两类技能纯 I/O 且支持异步的技能直接写 async handler。内部依赖同步库的技能通过asyncio.to_thread或独立线程池跑。import asyncio from concurrent.futures import ThreadPoolExecutor sync_executor ThreadPoolExecutor(max_workers16) async def run_sync_handler(handler, **params): loop asyncio.get_running_loop() return await loop.run_in_executor(sync_executor, lambda: handler(**params))在ToolRegistry.invoke里做一个判断如果 handler 是普通的同步函数就走线程池如果是 async 函数直接 await。这样 async 技能不会被同步技能拖累同步技能也不会堵死事件循环。5.3 按技能维度限流与熔断全局限流只能防住总流量超了但真正需要精细化管理的其实是单一技能。比如某个查询上游的接口每秒最多 20 次你总不能因为全局没超限就疯狂打它。我给每个技能配了一个令牌桶限流器。技能限流配置熔断阈值stock.get_price100 次/分钟连续失败 10 次order.create_order20 次/分钟连续失败 5 次weather.get_forecast60 次/分钟连续失败 8 次实现上不需要引入太重的外部依赖进程内做一个简单的滑动窗口计数就够了。一旦某个技能的失败率超过阈值自动把技能状态置为disabled同时发告警。这块的逻辑我直接放在invoke的入口处检查。class SkillCircuitBreaker: def __init__(self, threshold: int 8): self.threshold threshold self._failures [] def record_success(self): self._failures.clear() def record_failure(self): self._failures.append(time.time()) if len(self._failures) self.threshold: return True return False5.4 幂等读技能加缓存写技能必须穿透给读类技能做 TTL 缓存是性价比最高的优化。天气预报、股票行情、汇率这类数据几秒钟内的变化对用户决策没那么关键完全可以缓存 3-5 秒把上游压力降一个量级。需要注意缓存 key 的粒度。以stock.get_price为例缓存 key 应该是skill_id:symbol:exchange而不能只按 skill_id 缓存。同一个技能、不同参数返回结果完全不一致这算是一个低级的坑。写操作类技能下单、退款、发消息绝对不能缓存而且要保证同一用户的写操作串行化避免并发重复提交。管理器的做法是引入一个用户级锁同一个用户的写技能在并发时排队执行。5.5 日志采样别把监控做成新的故障源最开始我把每次技能调用的完整输入输出都写进日志跑了三天日志存储涨了几十 GB直接把磁盘打满了。后来改成两条策略默认只记录结构化的摘要包括 skill_id、耗时、状态、错误类型、返回结果的 size 和 JSON 摘要。重要的失败链路比如 LLM 决策失败、技能连续重试才记录完整输入输出用于复盘。这样既满足了可视化的需要又不至于让监控本身变成新的稳定性风险。6. 几个让我印象深刻的翻车现场最后分享几个真实踩过的坑。这些都不算什么高深理论但每一条都是花过时间、付出过线上事故代价才换来的经验。6.1 技能描述写得不到位Agent 在四个相似技能之间抽签有一次做金融类客服 Agent我注册了三个技能account.get_balance、account.list_accounts、account.query_transactions。结果用户问我账户里还有多少钱Agent 有时候选第一个有时候选第二个有时候甚至选第三个。我把描述拉出来一看三个描述都是获取账户信息LLM 根本分不清。后来我把描述改成这样account.get_balance查询账户总资产余额触发词余额、还有多少钱、总资产。account.list_accounts列出用户所有子账户列表触发词有哪些账户、账户列表。account.query_transactions查询账户交易明细流水触发词交易记录、明细、流水。示例返回包含time、amount、type字段。改完描述后这个问题的复现率几乎降为零。工具选择这个问题上prompt 工程的细节比任何算法都管用。6.2 Pydantic Schema 太宽松参数幻觉直接落到上游接口有一个查询物流的技能参数里tracking_no写的是 optional。模型经常不传或者传一个格式完全不对的运单号导致上游物流接口 4xx 报错。我排查很久才发现是参数 Schema 的问题。后来我把tracking_no改成 required并在 description 里写明运单号的具体格式数字 字母的组合长度 12-15 位。从那以后这个技能的参数幻觉率几乎降到零。可以理解为不给模型规则它就会自己编规则而编出来的规则基本不符合你的业务。6.3 灰度发布时忘了做会话粘性用户前后体验不一致技能管理器做了版本管理之后我开始尝试灰度切换技能的实现版本。结果发现一个问题同一个用户在一次会话里前半段用的是新版本技能后半段因为注册表 reload 变成了旧版本返回的数据格式都不一样用户直接懵了。解决方法是引入会话级技能快照。Agent 会话创建时把当前注册表里所有技能版本号快照一份会话期间调用技能时优先从快照里取版本对应的 handler。这样一次会话从头到尾看到的技能行为是一致的灰度只影响新会话不影响进行中的会话。6.4 别把所有技能都展示给 LLMOpenAI 的 function calling 对工具数量是有实际感知限度的。技能超过二三十个之后模型选错工具的概率会明显上升推理时间和 token 消耗也水涨船高。技能管理器要支持技能组的概念不同的 Agent 场景只挂载自己那一组技能而不是把所有技能一股脑塞给模型。比如客服 Agent 挂订单、售后、物流技能行情 Agent 只挂股票、汇率技能。这件事在可视化面板里就是一次勾选操作在注册表里就是一次skills_for_agent的过滤函数。做技能管理器这个项目我最大的体会是Agent 的能力边界不是模型决定的而是你给它准备的技能质量和调度能力决定的。工具可以越加越多但如果没有一套统一的注册、调度、观测机制加得越多系统越脆弱。如果你也在搭 Agent 平台我建议先把技能管理器做出来哪怕第一版只是一个很简陋的注册表和调试页它都会让你后续的开发效率高一个台阶。后面我会继续把这套方案往技能市场、技能编排的方向扩展有新的进展再来分享。
返回列表