先说结论SkillNet 不是又一个“套壳 Agent 框架”而是把智能体技能从“散装函数列表”升级成“可检索、可组合、可编排的技能网络”的一整套设计思路。它的核心价值是解决一个非常现实的工程问题——当 Agent 的技能从十个涨到一百个时LLM 还能不能准确选中技能、能不能把多个技能串成一个任务来执行。这篇文章会拆解 SkillNet 的模块结构、技能检索与编排链路并给出一套可以在本地沙盒里做最小验证的实现思路。如果你正在做 AI Agent 开发、多智能体系统、工具调用或工作流引擎这篇文章可以直接收藏。我们会先看它解决什么问题再看架构上怎么拆然后给出环境准备、核心代码结构、接口调用示例、性能观察和排错清单。1. SkillNet 核心能力速览先给一张总览表后面每一行都会展开讲。能力项说明项目定位面向 AI Agent 的可编排技能网络用于技能注册、检索、组合与调用解决的问题技能数量膨胀后LLM 选不准技能、编排链路复杂、调用可靠性下降核心功能技能注册、技能描述生成、语义检索、意图路由、技能组合编排、执行回放编排能力支持单技能调用、顺序编排、条件分支、并行组合、依赖声明检索机制基于技能描述向量的语义检索 规则匹配兜底接口形式参考实现可暴露 REST API支持技能列表查询、执行、编排任务提交批量任务支持通过任务队列批量提交执行请求推荐环境Linux / macOS / Windows 均可开发环境建议 Python 3.10检索部分可使用轻量向量库显存要求纯编排场景不依赖 GPU若接入本地 LLM 做意图识别则需要按模型实际测试适用场景客服多技能路由、自动化运维工具编排、RPA 技能管理、企业知识库问答这些能力项的描述是参考“技能网络”这一通用架构整理的最小集合。真实项目落地时应以实际代码仓库、模型版本和接口文档为准。2. 为什么要“技能网络”AI Agent 的技能管理困境先看一个现象。单机版的 Agent 想要“联网搜索”就注册一个search_web函数想要“查天气”就注册一个get_weather直接把函数列表塞进 Prompt交给 LLM 去选。这个方法在技能数少于十个时是可行的但一旦技能数量膨胀问题就出现了。第一Prompt 里塞不下。每个函数要写名称、描述、参数 Schema一个技能平均需要 200 到 500 个 token 的描述一百个技能就是两三万 token。即使上下文窗口能装下LLM 的注意力也会被稀释技能选择准确率明显下降。第二检索不精确。技能之间的描述可能有重叠比如“查订单”和“查物流”在很多系统里是两套接口但自然语言表达非常接近。纯粹靠 LLM 读函数列表做选择容易出现幻觉调用。第三组合缺乏约束。真实业务任务很少是单技能能完成的比如“查一下这个订单是否超时如果超时就自动发一条短信并登记理赔”。这涉及查询技能、判断逻辑、短信技能、登记技能四个步骤。传统函数列表只能让 LLM 自由发挥缺少可编排的执行顺序和依赖关系。第四维护成本高。每新增一个技能就要在所有调用点更新 Prompt改一个通用模板就导致全量 Prompt 重新生成。技能之间的依赖关系散落在业务代码里后期几乎无法维护。SkillNet 的思路就是针对这些问题把技能当成一种“一等公民”来管理。每个技能拥有独立的描述文档、参数 Schema、调用协议和依赖声明通过一个技能注册中心统一管理。调用方不是直接面对 LLM 的随机选择而是走“技能检索 - 路由决策 - 编排计划 - 执行器调度”这条链路。3. 核心概念拆解技能、描述、注册表与编排要把技能网络聊清楚先统一几个概念。下面的定义是理解 SkillNet 的基础。3.1 技能Skill技能是系统内可被 Agent 调用的最小功能单元。它不一定是单个函数可以是一个工具、一个 API、一段脚本甚至是一个子 Agent。一个标准化技能由四部分组成技能 ID全局唯一比如order_query、sms_send。功能描述用自然语言描述技能能做什么、不能做什么、适合什么场景。参数 Schema定义入参和出参的格式推荐使用 JSON Schema。调用协议实际执行方式可以是 HTTP 请求、本地函数调用、消息队列消息等。3.2 技能描述与元数据技能描述是检索和路由的核心输入。它要写清楚“什么时候用这个技能”而不只是“这个技能是什么”。比如一个订单查询技能描述应该是“当用户询问订单状态、物流进度、签收情况时使用”而不是“订单查询接口”。除了描述元数据还包括技能分类标签比如业务域交易、操作类型查询。依赖的技能 ID 列表。执行超时时间。需要的权限级别。3.3 技能注册表Registry)注册表是所有技能的元数据中心。每个技能在上线前必须注册。注册表负责技能 ID 的唯一性校验。描述与 Schema 的格式校验。依赖关系的完整性检查。为每个技能生成一份嵌入向量用于后续的语义检索。3.4 编排Orchestration)编排解决的是“多个技能怎么配合完成一个复杂任务”。SkillNet 的编排引擎会生成一个执行计划这个计划不是自由的函数调用序列而是有明确节点和边的工作流。一个典型的编排节点包含节点 ID。绑定的技能 ID。输入来源是用户输入、上一个节点输出还是外部数据。输出流向。条件分支逻辑。4. 总体架构SkillNet 的模块划分从宏观上看SkillNet 可以分成六个模块。模块职责关键输出技能注册中心技能上线、元数据存储、向量索引构建技能记录、技能索引技能检索器根据用户输入召回候选技能技能候选列表及相似度分数意图理解与路由确定用户的真实意图在候选技能中做最终选择主技能、辅助技能编排引擎生成多技能执行计划管理依赖关系可执行工作流执行器按计划调用技能处理参数映射和结果回传执行结果观测与回放记录日志、耗时、错误、性能指标可观测数据各个模块通过事件或 REST API 通信。下面是部署视角的架构示意。用户输入 | v [意图理解与路由] ---- [技能检索器] ---- 技能注册中心 | ^ v | [编排引擎] | | | v | [执行器] ------------------------- | v [观测与回放]这个架构的核心特点是检索和编排解耦。检索只用语义相关度和规则过滤不参与业务判断编排负责决定执行顺序不关心技能内部实现。这样两个模块可以独立扩展。5. 技能检索链路从自然语言到技能命中技能检索是 SkillNet 最关键的一条链路。它的目标是让用户输入先经过一个快速、廉价的检索过程缩小候选集再交给 LLM 做精细路由。这样做的好处是既能减少 Prompt 中技能描述的数量又能提升路由准确率。整个检索链路分四步。5.1 第一步输入预处理对用户输入先做一些基础处理包括去除无效符号、提取核心实体、识别领域关键词。这一步不一定要用 LLM用正则或词典也能完成。def preprocess_query(raw_text: str) - str: 简单预处理去空格、提取核心句。实际项目可按需扩展。 text raw_text.strip() text text.replace(, ,).replace(。, .) return text.lower()5.2 第二步语义召回将预处理后的文本编码成向量在技能索引里做最近邻搜索。这里有两个选择如果使用本地向量库比如 Chroma、FAISS、LanceDB可以把所有技能的描述向量化后建索引如果技能数量很少直接通过词法匹配也能达到目的。召回阶段的目标不是精准命中而是不要漏掉正确答案。所以 Top K 可以设置得宽一点比如召回 10 到 20 个候选。def recall_candidates(query: str, registry, top_k: int 10): 对技能注册表做语义召回返回候选技能列表和分数。 query_vec registry.embedding_model.encode(query) results registry.vector_store.search(query_vec, top_ktop_k) return [(r[skill_id], r[score]) for r in results]5.3 第三步规则过滤语义召回的结果可能包含一些不符合当前场景的技能。规则过滤层会做以下几件事技能是否启用已下线的技能直接排除。权限是否匹配当前调用方无权访问的技能排除。标签是否匹配用户输入中包含明确的业务域关键词时按标签过滤。依赖是否完整技能依赖的组件是否可用不可用则排除。filters: - type: enabled value: true - type: permission role: user - type: dependency check: all5.4 第四步LLM 路由候选技能被压缩到 10 个以内后再把这 10 个技能的描述和参数 Schema 组合成一份精简 Prompt交给 LLM 做最终选择。这一步的准确率远高于直接让 LLM 从一百个技能里选。{ query: 我的订单显示签收了但实际没收到货应该怎么办, candidate_skills: [ { skill_id: order_query, description: 查询订单状态、物流轨迹、签收信息, score: 0.89 }, { skill_id: logistics_complaint, description: 提交物流异常投诉, score: 0.82 } ] }LLM 的输出结构可以定义为{ selected_skill: order_query, reason: 用户需要查询物流签收详情, confirm_required: true }如果confirm_required为 true系统在调用前需要向用户确认。这一步能有效减少误调用。6. 技能组合与编排从单技能到多技能协作技能检索解决的是“选哪个技能”技能编排解决的是“多个技能怎么配合”。SkillNet 的编排引擎采用“声明式工作流 动态修正”的混合模式。6.1 声明式工作流编排引擎允许开发者预先定义一套工作流模板。模板描述了技能之间的顺序和条件关系。workflow_id: order_timeout_handle description: 订单超时处理流程 nodes: - id: query_order skill: order_query input_from: user_input - id: check_timeout skill: timeout_ruleset input_from: query_order.output - id: notify_user skill: sms_send input_from: check_timeout.output condition: check_timeout.is_timeout true - id: register_claim skill: claim_registration input_from: check_timeout.output condition: check_timeout.is_timeout true edges: - from: query_order to: check_timeout - from: check_timeout to: notify_user on_true: true - from: check_timeout to: register_claim on_true: true这个模板的价值在于把技能调用顺序和条件分支从 LLM 的“自由发挥”变成“可预期的执行计划”。LLM 只在condition字段的表达式求值上发挥作用。6.2 动态编排声明式模板覆盖常见场景但总会有没提前定义流程的新需求。SkillNet 的动态编排模块负责处理这种情况。动态编排的流程是获取用户输入和检索结果。让 LLM 生成一个执行图图中节点为技能调用。执行器按拓扑顺序执行节点每个节点执行后把结果回填。当某个节点的输出足够满足用户目标时提前终止。动态编排的风险是 LLM 生成非法流程。解决方法是加入校验器检查节点是否存在、参数是否完整、依赖是否满足。def validate_plan(plan: dict, registry) - bool: 校验动态编排计划是否合法。 for node in plan.get(nodes, []): if node[skill] not in registry.skill_ids(): return False if input_from not in node and input_value not in node: return False return True6.3 参数映射多技能编排最容易出问题的点是参数映射。上一个技能输出的字段名不一定匹配下一个技能需要的入参名。SkillNet 在编排引擎里增加了一层参数映射器。它负责把上游输出转换成下游输入。def map_input(source_output: dict, target_schema: dict) - dict: 将上游输出映射为下游技能入参字段名不一致时做转换。 mapped {} for field, spec in target_schema[properties].items(): if spec.get(source_field): mapped[field] source_output.get(spec[source_field]) else: mapped[field] source_output.get(field) return {k: v for k, v in mapped.items() if v is not None}7. 技能调用与执行参数校验、执行与反馈编排引擎生成执行计划后执行器开始工作。执行器的核心职责是把逻辑计划转成实际调用并保证每次调用的结果可观测、可回放。执行器的执行流程如下。7.1 参数校验在真正调用技能前执行器会使用 JSON Schema 校验参数。不经过这一步很多错误会在运行时才暴露出来排查成本很高。import jsonschema def validate_arguments(args: dict, schema: dict) - bool: 按 JSON Schema 校验技能入参。 try: jsonschema.validate(instanceargs, schemaschema) return True except jsonschema.ValidationError as e: print(f参数校验失败: {e.message}) return False7.2 技能执行技能执行有三种常见协议本地函数调用直接import技能模块调用指定函数。HTTP 调用向内部或外部服务发送请求。消息队列触发向队列发送任务消息由异步 Worker 消费。具体使用哪种协议由技能注册表里的protocol字段决定。执行器不关心协议细节只负责把参数按协议要求包装好。async def execute_skill(skill: dict, args: dict) - dict: 按技能的协议类型执行返回标准结果包装。 if skill[protocol] http: async with httpx.AsyncClient() as client: resp await client.post(skill[endpoint], jsonargs) return {status: success, data: resp.json()} elif skill[protocol] local: func load_local_function(skill[module], skill[function]) result await func(**args) return {status: success, data: result} else: return {status: error, error: f不支持的协议: {skill[protocol]}}7.3 结果标准化不同技能返回的数据结构差异很大。如果不做标准化编排引擎的下游节点就没法统一处理。标准化的目标是把结果统一成三个字段status执行成功或失败。data结构化的返回数据。trace_id用于链路追踪的日志 ID。{ status: success, data: { order_id: 20241017001, status: delivered, logistics: 已签收 }, trace_id: 8f3a9c1b2e }7.4 失败重试与回滚技能调用一定会遇到失败。执行器内置了策略可重试错误网络超时、服务暂时不可用按指数退避重试。不可重试错误参数错误、权限不足直接返回失败。成功节点补偿当多技能工作流中某个关键节点失败并希望撤销之前的节点操作时执行器调用技能注册表中的compensation_action。对于支付、库存这类高风险操作建议在技能注册时声明compensation_action否则出现半完成状态时很难恢复。8. 本地沙盒环境准备与最小实现验证SkillNet 的实际部署方式没有统一标准因为不同团队的代码实现差异很大。但无论后端代码怎么实现本地验证的核心步骤是一致的。这里给出一套通用沙盒流程。8.1 环境准备清单建议最小环境如下Python 3.10 或更高。一个向量数据库本地开发可以用 Chroma 或 FAISS。一个 LLM 接口用于意图路由可以是远程 API也可以是本地模型。一个测试技能建议先写两个简单的本地技能比如random_number和time_now用来验证链路是否打通。# 创建虚拟环境 python -m venv skillnet_env source skillnet_env/bin/activate # 安装基础依赖版本以实际项目 requirements 为准 pip install fastapi uvicorn chromadb sentence-transformers httpx8.2 最小技能注册示例下面是一个最小化的技能注册脚本目的是验证“注册中心 - 向量索引 - 检索”这条链路。# skill_registry.py from dataclasses import dataclass, field dataclass class SkillRecord: skill_id: str description: str parameters_schema: dict protocol: str endpoint: str tags: list field(default_factorylist) class SkillRegistry: def __init__(self): self.skills {} self.vectors [] def register(self, skill: SkillRecord): if skill.skill_id in self.skills: raise ValueError(f技能重复注册: {skill.skill_id}) self.skills[skill.skill_id] skill # 此处应将 description 向量化后写入向量库 # print(f注册技能: {skill.skill_id}) def get(self, skill_id: str): return self.skills.get(skill_id) registry SkillRegistry() registry.register(SkillRecord( skill_idtime_now, description获取服务器当前时间适合时间相关查询, parameters_schema{type: object, properties: {}, required: []}, protocollocal, tags[system, time] )) registry.register(SkillRecord( skill_idrandom_number, description生成随机数字适合抽奖、随机数生成, parameters_schema{type: object, properties: {min: {type: integer}, max: {type: integer}}, required: []}, protocollocal, tags[system, random] ))8.3 启动一个本地 API 服务验证链路最简单的方式是启动一个 FastAPI 服务提供两个接口一个查询技能列表一个执行技能。# api_server.py from fastapi import FastAPI from pydantic import BaseModel from skill_registry import registry, SkillRegistry app FastAPI() class ExecuteRequest(BaseModel): skill_id: str args: dict {} app.get(/skills) def list_skills(): return [{skill_id: s.skill_id, description: s.description} for s in registry.skills.values()] app.post(/execute) def execute_skill(req: ExecuteRequest): skill registry.get(req.skill_id) if not skill: return {status: error, error: 技能不存在} if skill.protocol local: # 实际应从模块加载函数 if req.skill_id time_now: import time return {status: success, data: {time: time.time()}} if req.skill_id random_number: import random lo req.args.get(min, 0) hi req.args.get(max, 100) return {status: success, data: {number: random.randint(lo, hi)}} return {status: error, error: 未实现} if __name__ __main__: import uvicorn uvicorn.run(api_server, host127.0.0.1, port8100)8.4 验证步骤启动服务后按顺序验证以下内容。使用curl查询技能列表curl http://127.0.0.1:8100/skills预期返回两个技能。调用时间技能curl -X POST http://127.0.0.1:8100/execute \ -H Content-Type: application/json \ -d {skill_id: time_now, args: {}}调用随机数技能curl -X POST http://127.0.0.1:8100/execute \ -H Content-Type: application/json \ -d {skill_id: random_number, args: {min: 1, max: 10}}每调用成功一次表示执行链路基本没问题。真正项目里的检索、路由和编排在这条链路上扩展即可。9. 接口 API 与批量调用设计当单个技能调用跑通后接下来关注的是接口的可用性和批量化能力。SkillNet 对外的接口设计可以参考以下三个层次。9.1 技能层 API技能层只关心单个技能的调用和查询。接口方法作用/skillsGET列出全部可调用技能/skills/{skill_id}GET查询单个技能详情/executePOST执行一个已注册技能/batch/executePOST批量执行多个技能9.2 编排层 API编排层负责提交多技能任务。{ workflow_id: order_timeout_handle, user_input: { order_id: 20241017001 }, trace_id: req_123456 }提交后返回{ task_id: task_8899, status: pending, workflow_id: order_timeout_handle }之后客户端通过任务 ID 轮询执行状态。9.3 批量任务队列批量任务适合离线处理比如晚上批量处理历史工单。实现方式是在编排层前面加一个任务队列用 Redis 或数据库表存储任务状态。# 批量任务伪代码提交多个任务的执行请求 import requests tasks [ {workflow_id: order_timeout_handle, user_input: {order_id: A001}}, {workflow_id: order_timeout_handle, user_input: {order_id: A002}}, ] for t in tasks: resp requests.post(http://127.0.0.1:8100/workflow/submit, jsont, timeout30) print(resp.json())批量任务的工程要点每个任务要有独立trace_id方便查失败原因。队列消费端要限流避免瞬间打爆下游技能服务。执行失败要落入重试队列并设置最大重试次数。批量执行结束后生成汇总报告统计成功率、平均耗时、失败原因分布。curl 重试示例配合接口层做基础验证curl -X POST http://127.0.0.1:8100/batch/execute \ -H Content-Type: application/json \ -d {tasks: [{skill_id: random_number, args: {min: 1, max: 100}}]}10. 资源占用与性能观察SkillNet 这类系统不是重计算型应用它的性能瓶颈通常不在推理算力而在检索速度、LLM 路由延迟和技能本身的外部依赖。下面列出需要观察的几个维度。10.1 显存占用如果只跑检索和编排逻辑纯 CPU 环境足够不占显存。只有当意图路由环节使用本地 LLM 时才需要考虑显存。具体占用取决于模型规模7B 量级的模型在常见量化方案下可能需要 6G 到 10G 显存。14B 或更大的模型可能需要 12G 以上显存。这些数字都是通用参考实际需要按模型版本和推理框架测试。SkillNet 本身并不要求一定用本地模型远程 API 同样可以。10.2 延迟构成一次完整技能调用的延迟可以拆解为输入预处理毫秒级。语义召回索引在百万级以下时通常几十毫秒。LLM 路由远程模型通常 500ms 到 2s本地模型取决于显存和批次。技能执行取决于技能本身数据库查询约几十毫秒外部 API 约几百毫秒。编排节点间传递毫秒级。如果对延迟敏感建议优先优化最后一环也就是技能本身的执行性能。很多场景下检索和路由的耗时占比并不高。10.3 调节批大小与超时批量任务中批大小直接决定资源占用。如果队列里有大量任务合理的做法是把批大小控制在 4 到 8 之间给每个技能留出超时时间。queue: batch_size: 8 max_retry: 3 timeout_seconds: 30 retry_backoff_seconds: 210.4 进程和端口规范本地调试多个服务时端口冲突是常见问题。建议对端口做统一规划。服务端口建议主 API 服务8100编排服务8200向量数据库采用默认端口本地推理服务按实际框架要求配置如果启动时报端口占用可以先用以下命令排查lsof -i :8100确认是本地残留进程可以直接结束进程或者换端口。11. 常见问题与排查方法技能网络类项目在落地中会遇到的问题大部分不在“技能”本身而在链路集成。下面给出一张排查表。问题现象可能原因排查方式解决方案系统反复选错技能技能描述模糊或候选集过大打开路由日志查看候选技能分数优化技能描述缩小语义召回 Top K技能调用报参数错误上游输出字段和下游入参不匹配查看参数映射日志完善参数映射器增加字段别名编排任务一直 pending队列消费进程挂了或任务卡在外部调用查看队列长度和 Worker 日志重启 Worker给任务加超时预判调用技能成功但用户反馈没生效调用的是测试环境技能或技能内部有错比对 trace_id 和实际技能日志确认技能路由到正确环境批量任务大量失败下游服务被限流查看失败错误码看是否为 429增加退避重试降低并发检索结果不稳定向量库中技能描述更新不彻底检查索引版本每次技能注册后强制增量更新索引LLM 路由输出非法 JSONPrompt 约束不够或模型版本不佳查看原始输出增加 JSON Schema 校验失败时走规则兜底新增技能后其他技能命中率下降技能描述之间语义重叠计算技能描述间相似度重写重复描述增加业务域标签以上排查思路适用于大多数技能网络实现。实际项目中日志规范化做得越早排查成本越低。12. 最佳实践与使用边界12.1 技能描述质量优先级最高技能网络的效果百分之七十取决于技能描述是否清晰。描述要写“什么时候用”以及“什么时候不能用”不要写成接口注释。新增技能前先检查描述里是否包含足够多的触发场景词汇。12.2 最小可运行配置要保留在开始写业务功能前先跑通一条最小链路注册两个技能 - 检索召回 - LLM 路由 - 执行器调用成功。这套最小配置能帮你快速判断问题出在上游检索还是下游执行而不是在几百行业务代码里找 bug。12.3 编排流程要可分步回放技能编排不能只记录最终结果要记录每个节点的输入、输出、耗时和状态。线上纠纷往往需要回溯到某一节点才能定位责任。12.4 合规与授权边界SkillNet 涉及的技能可能包括用户画像查询、订单数据读取、消息推送、自动化操作等敏感能力。在真实场景中务必注意技能调用前要确认当前请求是否有权限调用该技能。涉及用户个人信息、订单数据等要符合数据安全和隐私保护要求。自动回复消息、自动提交工单等操作需要设置操作确认或操作记录。使用第三方 API 或非自研模型时确认服务条款和使用范围。不要在未授权的情况下接入爬虫、破解类、绕过认证类技能。发布到公网的接口服务要加鉴权避免被未授权调用。12.5 接口访问范围本地开发时可以监听127.0.0.1不要直接暴露到公网。如果一定要提供局域网访问至少加上 Token 鉴权和 IP 白名单。# 本地启动不对外暴露 uvicorn api_server:app --host 127.0.0.1 --port 810013. 总结与下一步SkillNet 值得关注的核心不是某个“炫技”的 Agent 能力而是一套解决真实问题的工程化思路用注册中心管理技能元数据用语义检索缩小候选集用编排引擎把多技能协作变成可预期的工作流。如果你打算在自己的项目里实践建议按这个顺序推进。第一步把现有 Agent 代码里的所有函数列表整理成一份技能清单包括描述、参数和调用协议。第二步找一个轻量向量库把技能描述索引化实现语义召回。第三步写一个最小的编排引擎只支持顺序执行和条件分支应付一两个真实业务场景。第四步接入日志和链路追踪跑一批真实请求看技能选择准确率、执行成功率和平均耗时。最值得优先验证的是“技能检索 路由”这条链路。如果这一步能稳定跑通后面的编排、批量任务和 API 化改造都只是工作量问题。最容易踩的坑有两个。一是技能描述写得太模糊导致检索阶段就选错方向后面所有优化都白费二是编排计划缺少校验和回放能力上线后出了问题压根没法定位。后续扩展可以从两个方向切入一是给编排引擎接入流式输出和人工确认节点让复杂任务在执行前有干预机会二是把技能检索从纯文本描述扩展到“示例 异常案例”的多模态索引提高边缘场景的命中率。建议把这套“技能注册 - 检索 - 路由 - 编排 - 执行 - 观测”的链路保存在项目里作为基线架构后续无论是接本地模型还是远程 API都围绕这条链路延展。