AI Skill系统架构选型：单进程注册表 vs 微服务网关，实测对比与避坑指南

先说结论：我这段时间把同一套AI Skill系统，用两种完全不同的架构路线各实现了一遍。一套是把所有Skill塞进同一个进程，靠注册表和调度器内部路由；另一套是把每个Skill拆成能独立部署的服务，再用网关统一收敛。两条路都跑通了，也都上了真实业务，最后沉淀下来的东西让我对"AI Skill系统到底该怎么搭"有了一个比较清醒的判断。

如果你正在做Agent类平台，或者想在项目里引入"可插拔技能/能力"这套设计，这篇文章应该能给你省不少试错时间。我会把两种架构的实现细节、调度链路的差异、实测数据以及迭代过程中踩过的坑全部摊开来讲，代码也已经整理成可直接复现的版本。下面直接进正题。

1. 两种架构的本质分歧在哪：调度方式和部署粒度决定一切

在聊代码之前，先把一个最容易被忽略的问题说清楚：什么是Skill？我代码里的Skill，指的是一个可以被外部统一调用的"能力单元"，它有自己的名字、功能描述、输入输出协议，以及真正的执行逻辑。跟传统插件体系最大的区别是，Skill在Agent场景里通常要跟大模型交互，所以它的输入输出不光是普通JSON，还可能涉及流式响应、工具调用、多轮上下文传递。

1.1 先给Skill下一个能写代码的定义

我在项目里把Skill抽象成四个要素，缺一个后面都会踩坑：

• 元信息：skill_id、版本号、能力描述、作者、依赖项，这部分是给调度器和管理后台看的；
• 输入Schema：声明这个Skill接受哪些参数，参数类型和校验规则，对应到代码里就是一个Pydantic模型；
• 执行体：真正干活的函数或者类方法，通常带async支持，因为大模型接口和外部API基本都是IO密集；
• 生命周期钩子：加载、卸载、健康检查，这个在单进程方案里看着多余，一旦拆成独立服务就会发现它是救命稻草。

1.2 方案A和方案B的分界线

两套方案的分界线不在"用没用FastAPI"，也不在"代码组织方式"，而在两个底层决策：

调度方式：方案A里，调度器拿到skill_id后直接在一个进程内部的注册表里查函数引用，然后调用它。整个过程没有网络跳转，相当于函数调用。方案B里，调度器拿到skill_id后要去注册中心查这个Skill目前部署在哪个IP和端口，然后用HTTP或者其他协议把请求转过去。调度从"本地调用"变成了"远程调用"。

部署粒度：方案A把所有Skill打包在一个进程里，你启动一个服务就等于启动了所有能力。方案B是每个Skill独立进程，各自启动、各自扩容、各自出问题各自重启。

这两条就是后来所有差异的根源，其余的一切，比如网关设计、注册中心、可观测性，都是被这两个决策逼出来的。

1.3 我为什么两个方案都做了

背景交代一下：我给某团队搭一个内部用的Agent能力平台，第一阶段要求快速跑通一个Demo，能给上面的人演示"一句话调用多个技能"的效果。这个阶段毫无疑问选了方案A，因为改动量小，代码量控制在几百行，一个人两天可以搞定。

第二阶段业务真上了，接到一个真实场景：不同技能要按部门独立部署，有的技能要给外部合作方调用，有的技能对稳定性要求极高，出了故障不能牵连主链路。这时候方案A的天花板就很明显了：所有Skill共享进程资源，一个技能出现死循环可能拖垮全部能力；新技能上线要全量发布；没法针对热点技能单独扩容。于是我把方案B完整实现了一遍。

这个过程不是"两个方案二选一"，而是先快速落地再根据瓶颈做演进。我建议所有刚接触Skill系统的人也用这个节奏，不要一上来就上微服务，后面我会解释为什么。

2. 方案A：单进程Skill Registry，用最少的代码跑通全链路

方案A的目标很明确：用最少的代码，把"注册Skill、调度Skill、跑通流式输出"这条主线走通。

2.1 Skill的抽象与协议

先定义Skill基类，所有具体技能都继承它。我用了Python 3.10+的语法，配合Pydantic v2做输入校验。

# skill_base.py from abc import ABC, abstractmethod from typing import Any, AsyncIterator, Optional from pydantic import BaseModel, Field class SkillInput(BaseModel): """所有Skill输入参数必须继承这个基类，方便统一校验。""" request_id: str = Field(..., description="链路追踪ID") session_id: Optional[str] = Field(None, description="多轮会话ID") extra: dict[str, Any] = Field(default_factory=dict, description="扩展字段") class SkillOutput(BaseModel): """普通输出。流式输出单独走stream方法。""" status: str = "success" data: Any = None error: Optional[str] = None class BaseSkill(ABC): # 元信息 skill_id: str = "" version: str = "1.0.0" description: str = "" enabled: bool = True @abstractmethod async def execute(self, inputs: SkillInput) -> SkillOutput: """非流式执行入口。""" raise NotImplementedError async def stream(self, inputs: SkillInput) -> AsyncIterator[str]: """默认不支持流式，子类按需覆写。""" yield await self.execute(inputs) # 兜底实现 return

这里有个经验：不要把execute和stream混成一个方法。真实业务里，有的场景只要最终结果（比如调用一个工具），有的场景必须边生成边输出（比如聊天机器人逐字返回）。分开设计之后，调用方可以根据业务需要自由选择，互不干扰。

2.2 启动时的扫描注册

接下来是注册表。我用的是"装饰器+类扫描"的方式，这样新增Skill时不需要手动改注册表，只要把文件放进指定目录，类定义时加上装饰器即可。

# skill_registry.py import logging import importlib import pkgutil from typing import Type, Dict from skill_base import BaseSkill logger = logging.getLogger(__name__) class SkillRegistry: def __init__(self): self._skills: Dict[str, Type[BaseSkill]] = {} self._instances: Dict[str, BaseSkill] = {} def register(self, cls: Type[BaseSkill], skill_id: str = None): sid = skill_id or cls.skill_id if not sid: raise ValueError(f"Skill {cls.__name__} 缺少 skill_id") self._skills[sid] = cls logger.info(f"[Registry] 注册Skill: {sid} v{cls.version}") return cls def build_all(self): for sid, cls in self._skills.items(): if cls.enabled: self._instances[sid] = cls() logger.info(f"[Registry] 实例化完成，可用Skill数: {len(self._instances)}") def get(self, skill_id: str) -> BaseSkill: skill = self._instances.get(skill_id) if not skill: raise KeyError(f"Skill不存在或未启用: {skill_id}") return skill registry = SkillRegistry() def register_skill(skill_id: str = None): def decorator(cls: Type[BaseSkill]): registry.register(cls, skill_id) return cls return decorator def auto_load_skills(package_name: str): """自动扫描包内所有模块，触发装饰器注册。""" import importlib pkg = importlib.import_module(package_name) for mod_info in pkgutil.iter_modules(pkg.__path__): importlib.import_module(f"{package_name}.{mod_info.name}")

一个很常见的报错是：忘记调用auto_load_skills，导致注册表空空如也。我把自动扫描放在应用的启动事件里，启动时打印一行"当前可用Skill列表"，一眼就能看出哪些Skill没加载进来。

2.3 Dispatcher是怎么把请求路由到Skill的

调度器的本质是一张路由表：skill_id -> 实例 -> 执行方法。真正的复杂度在于流式响应、错误隔离和超时控制。

# dispatcher.py import asyncio import time import uuid from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from skill_registry import registry from skill_base import SkillInput app = FastAPI(title="AI Skill System - Single Process") class DispatchRequest(BaseModel): skill_id: str payload: dict = {} session_id: str = None use_stream: bool = False @app.post("/v1/skill/dispatch") async def dispatch(req: DispatchRequest): request_id = str(uuid.uuid4()) try: skill = registry.get(req.skill_id) except KeyError as e: raise HTTPException(status_code=404, detail=str(e)) inputs = SkillInput( request_id=request_id, session_id=req.session_id, extra=req.payload, ) if not req.use_stream: start = time.perf_counter() try: result = await asyncio.wait_for(skill.execute(inputs), timeout=60.0) except asyncio.TimeoutError: raise HTTPException(status_code=504, detail="Skill执行超时") return { "request_id": request_id, "skill_id": req.skill_id, "elapsed_ms": round((time.perf_counter() - start) * 1000, 2), "result": result, } async def event_generator(): yield f"event: start\ndata: {request_id}\n\n" try: async for chunk in skill.stream(inputs): yield f"event: delta\ndata: {chunk}\n\n" yield "event: done\ndata: [DONE]\n\n" except Exception as e: yield f"event: error\ndata: {str(e)}\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream")

这个版本我已经把超时控制、流式SSE协议、请求ID追踪都补上了。方案A看起来简单，但它是后面方案B的"最小可行原型"，很多协议设计都是在这版里敲定的。

2.4 方案A的优势和天花板

方案A在实际跑通的业务里，优点非常突出：

• 代码量少，一个中等水平的开发者半天就能看懂全部调度逻辑；
• 调试方便，断点可以打在进程里任何位置，不需要跨服务追踪；
• 延迟最低，本地函数调用没有序列化和网络开销，实测P50延迟比方案B低大概30%-40%；
• 发布简单，一个构建产物打包所有技能，版本不容易漂移。

但它的天花板也很硬：进程内全部技能共享GIL和内存，一个技能里的重度计算或者死循环，会让所有技能一起遭殃。我带过一次线上事故：某个技能在解析用户上传的大文件时OOM，整个主进程直接挂了，所有Agent功能全部不可用。这个事故直接促使我把方案B的优先级提到最高。

另外，扩张性上方案A几乎为零。你想给A技能单独加两个实例扛流量，做不到，因为所有的Skill都在同一个进程里。想用不同技术栈写技能（比如某个技能团队只会Java），方案A也直接卡死。这些是方案B要解决的。

3. 方案B：把Skill拆成独立服务，靠网关做统一收敛

方案B的设计目标：Skill可以独立部署、独立扩容，主链路不再被某个技能拖垮。调用方仍然只需要面对一个统一入口，不需要关心Skill部署在哪个机器上。

3.1 架构容不得"想当然"，先画调度链

方案B我从一开始就画清楚了请求链路，建议大家也先画完再写代码：

客户端 -> 统一网关(Gateway) -> 注册中心(Redis) 查路由 -> 目标Skill服务(HTTP/SSE) -> 回传结果 -> 客户端

这里面有三个必须实现的核心节点：

• 注册中心：存放所有在线Skill的路由信息，包括skill_id、服务地址、版本、健康状态；
• Skill服务：每个Skill是一个独立的FastAPI应用，启动时向注册中心上报自己的元数据和地址，并提供一个标准的执行接口；
• 统一网关：对外提供跟方案A几乎一致的API，内部根据请求里的skill_id去注册中心查地址，然后把请求转发到对应的Skill服务。

3.2 注册中心与Skill元数据

我用Redis做注册中心，key为skill:{skill_id}，value是一个JSON字符串，里面记录地址、版本、协议。同时用Redis的TTL实现心跳超时淘汰：每个Skill服务每10秒上报一次，如果TTL到期，网关就不再往这个地址转发。

# skill_register.py import json import time import redis class SkillRegistryClient: def __init__(self, redis_url: str, ttl: int = 30): self.r = redis.Redis.from_url(redis_url, decode_responses=True) self.ttl = ttl def register(self, skill_id: str, host: str, port: int, version: str, schema: dict): meta = { "skill_id": skill_id, "host": host, "port": port, "version": version, "ts": int(time.time()), "schema": schema, } key = f"skill:{skill_id}" self.r.set(key, json.dumps(meta), ex=self.ttl * 3) def heartbeat(self, skill_id: str): key = f"skill:{skill_id}" if self.r.exists(key): self.r.expire(key, self.ttl * 3) def lookup(self, skill_id: str) -> dict | None: key = f"skill:{skill_id}" raw = self.r.get(key) if not raw: return None return json.loads(raw) def unregister(self, skill_id: str): key = f"skill:{skill_id}" self.r.delete(key)

这里有个细节，我故意把heartbeat和register分成两个方法。Skill服务的启动流程是：先register写入完整元数据，之后每10秒调用一次heartbeat续期。这样网关读到元数据后就知道该走HTTP还是SSE，也能拿到schema做参数校验。

3.3 网关转发与流式适配

网关是方案B里最容易被低估的组件。新手往往以为它就是"收到请求转发一下"，实际上要做好三件事：服务发现缓存、流式协议转换、故障熔断。

# gateway.py import httpx import json from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse from skill_register import SkillRegistryClient app = FastAPI(title="AI Skill System - Gateway") registry = SkillRegistryClient(redis_url="redis://localhost:6379/0") client = httpx.AsyncClient(timeout=60.0) class DispatchRequest(BaseModel): skill_id: str payload: dict = {} session_id: str = None use_stream: bool = False @app.post("/v1/skill/dispatch") async def dispatch(req: DispatchRequest): meta = registry.lookup(req.skill_id) if not meta: raise HTTPException(status_code=404, detail=f"Skill {req.skill_id} 不在线") target = f"http://{meta['host']}:{meta['port']}" body = { "payload": req.payload, "session_id": req.session_id, } if not req.use_stream: resp = await client.post(f"{target}/skill/execute", json=body) if resp.status_code >= 400: raise HTTPException(status_code=502, detail=resp.text) return resp.json() async def stream_forward(): async with client.stream("POST", f"{target}/skill/execute_stream", json=body) as resp: if resp.status_code >= 400: yield f"event: error\ndata: {resp.text}\n\n" return async for line in resp.aiter_lines(): if line: yield line + "\n" return StreamingResponse(stream_forward(), media_type="text/event-stream")

网关层我只做了一个很轻的转发，没有做复杂的路由规则。真实项目里如果你有几十个Skill，建议在这里加一个本地路由缓存，避免每次请求都打一次Redis；另外要做熔断，当某个Skill连续报错超过N次，网关直接返回"该技能暂不可用"，而不是傻傻地继续吞超时。

3.4 独立部署后的真实收益

方案B上线后收益是立竿见影的。第一件事，我把那个有OOM风险的文档解析技能拆成独立服务，并给它的容器设置了内存上限和自动重启策略。它之后又挂过两次，但主链路纹丝不动，这个"故障隔离"就是方案B存在的全部意义。

第二件事，团队里另一个小组用Java写了一个技能，我只需要让它按协议上报到注册中心、实现两个HTTP接口，就能无缝接入系统。技术栈边界被彻底打开，这是方案A做不到的。

第三件事是扩容，某技能在活动期间流量翻了三倍，我只把它单独扩到5个实例，其他技能保持原样。这在方案A里要被迫全量扩容，成本完全不同。

当然，方案B的代价也很明显：链路变长，每次请求多一次网络跳转和一次Redis查询，延迟高了不少；部署运维复杂度直线上升；排查问题时要同时看网关日志、注册中心状态、目标服务日志。这些坑我在第5节详细讲。

4. 同一批测试用例下的实测数据：延迟、吞吐和维护成本

两套方案都写完以后，我没有凭感觉下结论，而是拿同一批业务用例做了压测和软性指标统计。这里包括真实业务重建的模拟场景和压力测试数据。

4.1 测试口径与仿真用例

测试环境是两台8核16G的云主机，服务都跑在Docker里。测试用例选择了我线上最具代表性的三类：

• 简单查询型：入参两个字段，调用一次模拟外部API返回结果，模拟调用第三方库存查询；
• 大模型对话型：流式输出，每轮约300个token，走SSE协议；
• 重计算型：读取一段文本，做解析和关键词抽取，CPU密集，模拟文档分析类技能。

压测工具用的wrk和自研的异步脚本，并发数从10到200递增，每个档位跑5分钟取稳定值。

4.2 关键指标对比

先看延迟和吞吐的直观对比：

我说几个关键读数。第一，简单查询型方案B的P50延迟比方案A翻了一倍还要多，原因很明确：一次网关到Skill服务的HTTP往返，加上Redis查询，这几十毫秒在低延迟场景里非常扎眼。第二，重计算场景反而两者差距不大，因为CPU耗时掩盖了网络开销。第三，吞吐方面方案B在并发到150之后开始出现大量连接复用瓶颈，我后面通过给httpx.AsyncClient加连接池参数解决了。

这里绝不意味着方案B"更差"。延迟高的几十毫秒，在Agent平台里通常是完全可接受的；换来的是故障隔离和独立扩容能力。关键是你要清楚自己的业务对延迟有多敏感。

4.3 隐藏成本：开发和排错的时间账

比压测数据更能说明问题的，是维护成本。我把两套方案从零实现到跑通全部用例的时间做了记录：

这个表是我真实体会的浓缩。方案B的新增Skill成本高在企业级流程：你要写Schema、起服务、配注册中心、配置部署流水线。如果团队只有两三个人，这些全是负担；如果团队有多个部门协作，这些全是必需品。

5. 从启动卡死到断流，迭代过程踩过的四个真坑

这部分我单独拎出来，是因为两个方案实现过程中踩的四个坑都非常隐蔽，网上很多文章不会写，但真遇到了能卡你整整一天。

5.1 坑一：自动扫描模块触发了循环导入，启动即卡死

方案A的auto_load_skills里，我用importlib导入包内所有模块，结果某个Skill模块在导入时又反向引用了dispatcher里的内容，造成循环导入。现象很迷惑：进程既不报错也不退出，就停在启动界面，看起来像"慢"，实际上是卡死。

排查思路：先加-X importtime参数查看导入耗时，锁定在某个模块后，发现是模块A导入模块B，模块B又导入模块A。解决办法有两个，第一是约定所有Skill模块不允许反向导入任何应用层代码，仅允许依赖基类和注册表；第二是把自动扫描改成"懒加载"，第一次被调用到才导入模块，避免启动时全部加载。

5.2 坑二：流式转发的背压问题，前端出现断流

方案B的网关做SSE转发时，最初写法是：

async for line in resp.aiter_lines(): yield line + "\n"

看起来没问题，但实际并发高的时候，前面几十行输出正常，流到某个位置就中断了，没有任何报错。查了很久发现是背压问题：上游Skill服务不断生成数据，网关来不及通过aiter_lines消费，HTTP连接被对端中止。更隐蔽的是某些网关库会静默丢弃缓冲区溢出的块。

解决方案是在Skill服务端和网关端都做限速：服务端每产出一个chunk后做一个很小的await asyncio.sleep(0)，把控制权交还给事件循环；网关端使用httpx.AsyncClient时限制max_connections，不要无限开连接。改完之后连续跑了一个小时压测，再没出现过断流。

5.3 坑三：注册中心里残留了状态，网关往死服务转发

方案B上线第三天遇到一个情况：某个Skill服务被手动停止，但网关依然把请求转发到那个死地址，导致大量502。原因是那个Skill实例停掉之前没有调用unregister，Redis里的key没有消失，TTL又还没到期。

解决方法是两层：注册中心在Skill服务注册时写上ts时间戳，网关转发前检查这个服务最近一次心跳时间有没有超过阈值；另外在运维侧把"下线前自动注销"写进Skill服务的生命周期钩子，保证收到SIGTERM信号时主动清注册信息。不要迷信TTL，TTL只是兜底，主动注销才是第一道防线。

5.4 坑四：开源代码里容易忽略的目录结构问题

我整理开源代码时发现，两个方案的代码如果放在同一个仓库里，特别容易把运行时依赖搞混。方案A只需要FastAPI和Pydantic，方案B多了Redis和httpx。别人clone下来如果直接跑方案B的示例，会报redis模块不存在。

所以我在代码仓库里做了分离：arch_a/和arch_b/两个目录，各自有独立的requirements.txt和README。启动命令也在文档里写清楚，方案A一条uvicorn main:app就能跑；方案B要先启动Redis、再启动Skill服务、最后启动网关。这样复现成本最低。

仓库结构和关键文件如下：

ai-skill-system/ ├── arch_a/ │ ├── skill_base.py # Skill抽象基类 │ ├── skill_registry.py # 注册表与自动扫描 │ ├── dispatcher.py # 单进程调度入口 │ ├── skills/ │ │ ├── weather_skill.py # 示例技能：天气查询 │ │ └── doc_qa_skill.py # 示例技能：文档问答 │ └── requirements.txt ├── arch_b/ │ ├── gateway.py # 统一网关 │ ├── skill_register.py # Redis注册中心客户端 │ ├── skill_service.py # Skill服务模板 │ └── skills/ │ └── http_echo_skill.py # 示例技能 ├── docs/ │ ├── protocol.md # Skill协议设计文档 │ └── benchmark.md # 压测数据与分析方法 └── README.md

docs/protocol.md里面我写了完整的Skill协议定义：请求格式、SSE事件类型、错误码、心跳规范。如果你准备在公司里推广这个体系，建议最早把这个文档固定下来，因为它直接影响各个团队实现的兼容性。

6. 架构取舍的结论：没有银弹，只有边界

所有架构对比最后都要回归到业务边界。我现在对两套方案的使用边界有非常明确的判断，直接分享给大家。

6.1 什么场景必须选方案A

团队很小，一到三个人，Skill数量不超过五个，所有技能都是内部使用，对延迟敏感，技术上统一用Python。这时候不要有任何犹豫，直接方案A。它的价值是让你用最小成本验证整个Skill体系的流程，把精力放在"技能本身的质量"而不是"服务怎么拆"上。

6.2 什么场景必须选方案B

你的系统要支撑第三方接入，或者不同技能由不同团队维护，或者某个技能的故障绝对不能影响主链路，再或者你可能用不同技术栈开发不同技能。满足其中任意一条，就要上方案B。我见过很多团队在这个阶段还硬扛着方案A，最后都在某个凌晨的线上事故里被迫重构。

6.3 一条可以同时吃的红利：核心能力留进程内，外围能力走独立服务

最后的落地形态，我其实是"混合"的：核心调度链路保持方案A的单进程模式，几个最关键、调用最频繁的Skill留在主进程内，享受最低延迟；其余技能全部走方案B，注册中心照上、网关照做。网关在发现某个skill_id在主进程本地注册表里存在时，直接走本地调用，否则才走远程转发。

这个"本地优先"的调度策略，让我同时拿到了两个方案最想要的那部分收益：核心路径的延迟几乎等于方案A，扩展路径的隔离性等同于方案B。如果你的系统已经在方案A上跑了一段时间，又不想推倒重来，强烈建议用这种方式平滑过渡。

我当时在代码里就加了一个判断：

# 网关里的本地优先路由逻辑（伪代码） def route(skill_id): if skill_id in local_registry: return "local", local_registry[skill_id] meta = redis_lookup(skill_id) if meta: return "remote", meta return None, None

代码实现本身不难，难的是你要想清楚：到底哪些技能值得留在进程内，哪些该拆出去。我的标准是看两个维度，"调用频率"和"故障代价"。调用频率高且故障代价低的，留在本地；调用频率低或者故障代价高的，拆出去。按这个标准，我的核心编排技能和短文本处理技能留在本地，文档解析、网页抓取、外部API聚合类技能全部拆成独立服务。

我个人在实际操作中的体会是：架构的选择不要被"微服务"这三个字绑架。单一进程架构不是low，独立服务架构也不一定高大上，关键看你的故障半径、团队协作模式和迭代节奏匹配哪一种。先跑通再拆分，是我这次做完两套方案之后最想分享的一句话。代码已经放出来了，按照README里的步骤，两套架构你都能在半小时内跑起来，到时候对照着压测一下，你对"该选哪条路"会有比我更实在的判断。