从架构到实现：如何设计高可用的Skill系统

2次阅读

没有评论

共计 2184 个字符，预计需要花费 6 分钟才能阅读完成。

我曾维护过一个智能客服系统，初期将所有业务逻辑硬编码在单个 Python 文件中。随着功能增加，出现了两个典型问题：

圣诞树问题 ：每次新增需求都要修改核心调度类，代码逐渐变成 2000 行的if-else 森林。某次修改天气查询逻辑时意外触发了订单查询的 bug
启动雪崩：系统启动时需要初始化所有 Skill，当技能数超过 50 个时，服务冷启动时间长达 12 秒，导致 Kubernetes 健康检查失败

优点：开发简单，所有逻辑一目了然
缺点：
单文件超过 3000 行后难以维护
测试用例相互干扰
资源加载无法按需进行

我们最终选择基于事件总线（Event Bus）的解决方案，关键考虑因素：

水平扩展：每个 Skill 可以独立部署
松耦合 ：通过SkillRegister 事件实现动态注册
资源隔离：崩溃的 Skill 不会影响主系统

# 事件总线伪代码示例
class EventBus:
    def __init__(self):
        self._handlers = defaultdict(list)

    def subscribe(self, event_type: Type[Event], handler: Callable):
        self._handlers[event_type].append(handler)

    def publish(self, event: Event):
        for handler in self._handlers[type(event)]:
            handler(event)

所有 Skill 必须实现基础接口，包含三个核心方法：

from abc import ABC, abstractmethod
from typing import Dict, Any

class BaseSkill(ABC):
    @property
    @abstractmethod
    def name(self) -> str:
        """返回技能唯一标识"""

    @abstractmethod
    def can_handle(self, user_input: str, context: Dict[str, Any]) -> bool:
        """判断是否响应当前输入"""

    @abstractmethod
    def execute(self, user_input: str, context: Dict[str, Any]) -> str:
        """执行核心逻辑"""

采用两级路由机制（时间复杂度 O(n)）：

第一级：根据意图识别结果快速过滤
第二级：按优先级排序（数值越小优先级越高）

def route(input_text: str, context: dict) -> Optional[BaseSkill]:
    candidates = [s for s in registered_skills if s.can_handle(input_text, context)]

    if not candidates:
        return None

    # 取优先级最高的技能（稳定排序）return min(candidates, key=lambda x: x.priority)

为解决并发问题，采用线程安全的 ContextManager：

from threading import Lock

class ContextManager:
    def __init__(self):
        self._contexts = {}
        self._lock = Lock()

    def get_context(self, session_id: str) -> dict:
        with self._lock:
            if session_id not in self._contexts:
                self._contexts[session_id] = {'created_at': time.time(),
                    'last_used': time.time()}
            return self._contexts[session_id]

懒加载：首次调用时才初始化 Skill
预加载：对高频 Skill 提前加载
资源分级：将词典等大文件放在 CDN

每个 Skill 运行在独立线程池
设置超时熔断机制（示例配置）：

from concurrent.futures import ThreadPoolExecutor

EXECUTOR = ThreadPoolExecutor(
    max_workers=50,
    thread_name_prefix='skill_worker',
)

# 带超时的执行封装
def safe_execute(skill: BaseSkill, timeout=3):
    future = EXECUTOR.submit(skill.execute, ...)
    return future.result(timeout=timeout)

关键指标维度：