Skill、Expert、Connector三层架构

摘要

在AI应用从“演示可用”走向“生产可靠”的过程中，单靠一个模型加一段提示词已经远远不够。真正让AI从“能做”到“做对”，再到“接入外部系统”的，是一套清晰的能力调用栈。本文将围绕 Skill（技能层）、Expert（专家层）、Connector（连接器层） 三层架构展开，解释它们各自的职责、协作方式、落地代码与工程实践。读完本文，你将掌握如何设计一个可扩展、可维护、可观测的AI能力调用体系。

---

引言：为什么AI需要“能力调用栈”

很多团队在接入大模型时，最初的做法是这样的：

response = llm.chat("帮我查一下北京今天的天气，然后写一封提醒同事带伞的邮件")

这段代码在Demo里看起来没问题，但一旦进入生产环境，问题立刻暴露：

1. 模型不知道“查天气”具体调用哪个API，只能靠猜。
2. 写邮件的格式、语气、签名规则每次都不一致。
3. 外部系统（邮件服务、天气服务）的鉴权、重试、限流没人管。
4. 出错后无法定位是模型问题、工具问题，还是连接问题。

于是我们需要一个分层的能力调用栈：

• Skill层：定义“能做”——把原子能力封装成可调用单元。
• Expert层：保证“做对”——编排技能、校验结果、控制流程。
• Connector层：实现“接入外部系统”——统一协议、鉴权、容错与观测。

这三层不是简单的代码分层，而是AI从“会说话”到“能干活”的必经之路。

---

1. 三层架构总览

先看整体结构：

+--------------------------------------------------+
|                   用户 / 上游系统                 |
+--------------------------------------------------+
|                 Expert 专家层                     |
|   - 任务规划 / 技能编排 / 结果校验 / 兜底策略      |
+--------------------------------------------------+
|                 Skill 技能层                      |
|   - 原子能力封装 / 输入输出Schema / 幂等控制       |
+--------------------------------------------------+
|               Connector 连接器层                  |
|   - HTTP / DB / MQ / 文件 / 第三方SDK 统一适配     |
+--------------------------------------------------+
|              外部系统：天气、邮件、CRM、数据库      |
+--------------------------------------------------+

可以把它类比成一家公司：

• Connector 是行政与IT，负责把电话、网络、门禁、快递都接好。
• Skill 是员工，每个人会一项具体技能，比如“查天气”“发邮件”。
• Expert 是项目经理，接到需求后决定派谁、按什么顺序、结果是否合格。

下面逐层展开。

---

2. Skill层：把“能做”变成可调用能力

2.1 Skill的职责

Skill是最小可复用能力单元，通常具备以下特征：

• 单一职责：一个Skill只做一件事。
• Schema明确：输入输出都有结构化定义。
• 可独立测试：不依赖Expert也能单测。
• 幂等或可重试：明确是否允许重复执行。

例如“查天气”是一个Skill，“发邮件”也是一个Skill。

2.2 Skill的代码定义

我们用一个Python示例来定义Skill基类和具体实现：

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Dict
import requests


@dataclass
class SkillResult:
    success: bool
    data: Any = None
    error: str = None


class BaseSkill(ABC):
    """所有Skill的基类，统一输入输出协议"""

    name: str = "base_skill"
    description: str = ""

    @abstractmethod
    def run(self, payload: Dict[str, Any]) -> SkillResult:
        raise NotImplementedError


class WeatherSkill(BaseSkill):
    """查询天气的原子技能"""

    name = "weather_query"
    description = "根据城市查询当前天气"

    def __init__(self, api_key: str):
        self.api_key = api_key

    def run(self, payload: Dict[str, Any]) -> SkillResult:
        city = payload.get("city")
        if not city:
            return SkillResult(success=False, error="city is required")

        try:
            # 这里用伪API演示，真实场景替换为实际服务
            resp = requests.get(
                "https://api.example.com/weather",
                params={"city": city, "key": self.api_key},
                timeout=5,
            )
            resp.raise_for_status()
            data = resp.json()
            return SkillResult(success=True, data={
                "city": city,
                "condition": data.get("condition"),
                "temperature": data.get("temperature"),
            })
        except Exception as e:
            return SkillResult(success=False, error=str(e))

2.3 Skill的注册与发现

为了让Expert层能动态调用，需要一个注册中心：

class SkillRegistry:
    def __init__(self):
        self._skills: Dict[str, BaseSkill] = {}

    def register(self, skill: BaseSkill):
        self._skills[skill.name] = skill

    def get(self, name: str) -> BaseSkill:
        if name not in self._skills:
            raise KeyError(f"Skill not found: {name}")
        return self._skills[name]

    def list_skills(self):
        return [
            {"name": s.name, "description": s.description}
            for s in self._skills.values()
        ]


registry = SkillRegistry()
registry.register(WeatherSkill(api_key="demo-key"))
print(registry.list_skills())

输出：

[{'name': 'weather_query', 'description': '根据城市查询当前天气'}]

2.4 Skill层的工程要点

• 输入校验前置：不要等调用外部API才发现参数缺失。
• 超时与重试：Skill内部要设置合理超时，重试策略由Connector统一管理更佳。
• 结果结构化：返回统一SkillResult，方便Expert判断。
• 可观测：每个Skill执行都应记录耗时、成功率、错误码。

---

3. Expert层：让AI“做对”

3.1 Expert的职责

Skill只是“能做”，但用户需求往往是复合的。Expert层负责：

• 任务分解：把“提醒同事带伞”拆成“查天气 + 生成邮件 + 发送邮件”。
• 技能编排：决定调用顺序、并行或串行。
• 结果校验：检查Skill返回是否符合预期。
• 兜底与修正：失败时重试、降级或换技能。
• 与LLM协作：让模型参与规划，但由Expert执行与把关。

3.2 一个可运行的Expert示例

class WeatherReminderExpert:
    """专家：根据天气生成并发送提醒邮件"""

    name = "weather_reminder_expert"

    def __init__(self, registry: SkillRegistry):
        self.registry = registry

    def handle(self, user_input: Dict[str, Any]) -> SkillResult:
        city = user_input.get("city")
        email = user_input.get("email")

        # Step 1: 调用天气Skill
        weather_skill = self.registry.get("weather_query")
        weather_result = weather_skill.run({"city": city})
        if not weather_result.success:
            return SkillResult(False, error=f"天气查询失败: {weather_result.error}")

        condition = weather_result.data["condition"]
        temperature = weather_result.data["temperature"]

        # Step 2: 结果校验（Expert的核心价值）
        if condition is None or temperature is None:
            return SkillResult(False, error="天气数据不完整，拒绝生成邮件")

        # Step 3: 根据规则生成邮件内容
        if "雨" in condition:
            advice = "今天有雨，记得带伞。"
        elif temperature < 10:
            advice = "今天较冷，注意保暖。"
        else:
            advice = "今天天气不错，祝你愉快。"

        subject = f"{city}天气提醒"
        body = f"你好，\n\n{city}当前天气：{condition}，气温{temperature}℃。\n{advice}\n\n来自AI助手"

        # Step 4: 调用邮件Skill
        email_skill = self.registry.get("email_send")
        email_result = email_skill.run({
            "to": email,
            "subject": subject,
            "body": body,
        })
        if not email_result.success:
            return SkillResult(False, error=f"邮件发送失败: {email_result.error}")

        return SkillResult(True, data={
            "city": city,
            "condition": condition,
            "email_sent_to": email,
        })

3.3 Expert与LLM的协作模式

Expert并不排斥LLM，反而应该善用LLM做“软判断”，但把“硬执行”留给自己：

def plan_with_llm(self, user_input: str, available_skills: list) -> list:
    """让LLM生成调用计划，但由Expert校验计划合法性"""
    prompt = f"""
    用户需求：{user_input}
    可用技能：{available_skills}
    请输出JSON格式的调用计划，例如：
    [{{"skill": "weather_query", "payload": {{"city": "北京"}}}}]
    """
    plan = llm.chat(prompt)  # 假设返回JSON字符串
    plan = json.loads(plan)

    # Expert校验：技能必须存在，参数必须合法
    for step in plan:
        if step["skill"] not in [s["name"] for s in available_skills]:
            raise ValueError(f"非法技能: {step['skill']}")
    return plan

3.4 Expert层的工程要点

• 规则优先，模型辅助：关键校验必须用代码，不依赖模型“自觉”。
• 可回放：记录每一步输入输出，方便复盘。
• 失败可降级：天气查不到时，可以只发通用提醒，而不是整体失败。
• 权限与审计：Expert是审计入口，谁在什么场景调用了什么技能必须留痕。

---

4. Connector层：接入外部系统的统一入口

4.1 Connector的职责

Skill不应该直接写requests.get，否则每个Skill都要重复处理鉴权、重试、限流、日志。Connector层负责：

• 协议适配：HTTP、gRPC、数据库、消息队列、文件系统。
• 鉴权管理：API Key、OAuth、Token刷新。
• 容错机制：超时、重试、熔断、降级。
• 可观测性：统一日志、指标、链路追踪。

4.2 Connector接口设计

class BaseConnector(ABC):
    @abstractmethod
    def call(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        raise NotImplementedError


class HttpConnector(BaseConnector):
    """统一HTTP连接器，处理鉴权、重试、超时"""

    def __init__(self, base_url: str, token_provider=None, max_retries: int = 3):
        self.base_url = base_url
        self.token_provider = token_provider
        self.max_retries = max_retries

    def call(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        url = f"{self.base_url}/{action}"
        headers = {}
        if self.token_provider:
            headers["Authorization"] = f"Bearer {self.token_provider()}"

        last_error = None
        for attempt in range(self.max_retries):
            try:
                resp = requests.request(
                    method=params.pop("_method", "GET"),
                    url=url,
                    params=params if params.get("_method") != "POST" else None,
                    json=params if params.get("_method") == "POST" else None,
                    headers=headers,
                    timeout=5,
                )
                resp.raise_for_status()
                return resp.json()
            except Exception as e:
                last_error = e
                # 指数退避
                time.sleep(2 ** attempt)
        raise RuntimeError(f"Connector调用失败: {last_error}")

4.3 用Connector改造Skill

class WeatherSkillV2(BaseSkill):
    name = "weather_query"
    description = "根据城市查询当前天气"

    def __init__(self, connector: HttpConnector):
        self.connector = connector

    def run(self, payload: Dict[str, Any]) -> SkillResult:
        city = payload.get("city")
        if not city:
            return SkillResult(False, error="city is required")
        try:
            data = self.connector.call("weather", {"city": city})
            return SkillResult(True, data={
                "city": city,
                "condition": data.get("condition"),
                "temperature": data.get("temperature"),
            })
        except Exception as e:
            return SkillResult(False, error=str(e))

这样Skill只关心业务语义，重试、鉴权、超时全部下沉到Connector。

4.4 多类型Connector示例

class DbConnector(BaseConnector):
    def __init__(self, connection_pool):
        self.pool = connection_pool

    def call(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        with self.pool.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(params["sql"], params.get("args", ()))
            if action == "query":
                return {"rows": cursor.fetchall()}
            conn.commit()
            return {"affected": cursor.rowcount}


class MqConnector(BaseConnector):
    def __init__(self, producer):
        self.producer = producer

    def call(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        self.producer.send(params["topic"], params["message"])
        return {"status": "sent"}

4.5 Connector层的工程要点

• 统一错误模型：所有Connector抛出同一种异常，方便上层处理。
• 可插拔：测试时替换为MockConnector，无需改Skill代码。
• 限流与熔断：生产环境必须防止外部系统拖垮整个AI服务。
• 可观测：每次调用记录耗时、状态码、重试次数。

---

5. 三层协作的完整调用链

把三层串起来看一个完整流程：

def main():
    # 1. 初始化Connector
    http_conn = HttpConnector(
        base_url="https://api.example.com",
        token_provider=lambda: "dynamic-token",
    )

    # 2. 初始化Skill并注册
    registry = SkillRegistry()
    registry.register(WeatherSkillV2(connector=http_conn))
    registry.register(EmailSkill(connector=http_conn))  # 假设已实现

    # 3. 初始化Expert
    expert = WeatherReminderExpert(registry=registry)

    # 4. 处理用户请求
    result = expert.handle({
        "city": "北京",
        "email": "colleague@example.com",
    })

    if result.success:
        print("任务完成：", result.data)
    else:
        print("任务失败：", result.error)


if __name__ == "__main__":
    main()

调用链如下：

用户请求
  -> Expert 解析意图、编排技能
    -> Skill 执行原子能力
      -> Connector 调用外部系统
    <- Skill 返回结构化结果
  <- Expert 校验结果、决定下一步
<- 返回最终结果

---

6. 常见问题与最佳实践

6.1 三层边界如何划分？

• 如果逻辑涉及“怎么调外部系统”，放Connector。
• 如果逻辑是“一项具体能力”，放Skill。
• 如果逻辑是“多技能协作、校验、兜底”，放Expert。

6.2 如何测试？

• Connector：用Mock Server或录制回放。
• Skill：注入MockConnector，做单元测试。
• Expert：注入MockSkill，做编排与异常路径测试。

6.3 如何观测？

建议在三层都打点：

import time

def traced_run(skill, payload):
    start = time.time()
    result = skill.run(payload)
    duration = time.time() - start
    log.info({
        "skill": skill.name,
        "success": result.success,
        "duration_ms": int(duration * 1000),
        "error": result.error,
    })
    return result

6.4 如何避免“AI乱调技能”？

• Expert只暴露经过审核的技能列表给LLM。
• LLM只负责生成计划，不直接执行。
• 所有计划必须经过Schema校验和权限校验。

---

总结

Skill、Expert、Connector三层架构，本质上是在回答三个问题：

1. 能做吗？ —— Skill层把原子能力封装成可调用单元。
2. 做对了吗？ —— Expert层负责编排、校验、兜底。
3. 接得上吗？ —— Connector层统一接入外部系统，处理鉴权、重试与观测。

这三层缺一不可：

• 只有Skill，AI是个“会调用工具的新手”，容易出错。
• 只有Expert，AI是个“空想家”，没有落地能力。
• 只有Connector，AI是个“管道工”，不懂业务语义。

当三层协同工作时，AI才真正从“能说会道”变成“能干活、干对活、接得进系统”的生产级助手。