从零搭建Agentic Skill架构：新手避坑指南与实践解析

2次阅读

没有评论

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

在构建智能体技能系统时，开发者常常会遇到几个典型问题：

技能耦合严重 ：不同技能之间直接调用，导致系统难以扩展和维护。
状态共享困难 ：技能之间需要共享状态时，往往通过全局变量或复杂的消息传递机制，增加了系统复杂度。
缺乏统一管理 ：技能的注册、发现和执行缺乏统一机制，导致系统混乱。

这些问题在项目初期可能不明显，但随着技能数量的增加，系统会变得越来越难以维护。

为了解决上述问题，我们采用分层架构设计：

接口层 ：负责与外部系统交互，接收请求并返回响应。
核心层 ：包含技能注册中心和消息总线，负责技能的调度和执行。
持久层 ：负责技能的状态存储和数据持久化。

技能注册中心是系统的核心组件，它允许技能动态注册和发现。每个技能在启动时向注册中心注册自己的元信息，包括技能名称、输入输出参数等。

消息总线用于解耦技能之间的直接调用。技能通过发布和订阅消息来交互，而不是直接调用其他技能的方法。这种设计提高了系统的灵活性和可扩展性。

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


class Skill(ABC):
    """技能基类，所有自定义技能需要继承此类并实现 execute 方法。"""

    @property
    @abstractmethod
    def name(self) -> str:
        """返回技能名称"""
        pass

    @abstractmethod
    def execute(self, params: Dict[str, Any]) -> Any:
        """
        执行技能
        :param params: 输入参数
        :return: 执行结果
        """
        pass

from typing import Dict, Type


class SkillRegistry:
    """技能注册中心"""

    def __init__(self):
        self._skills: Dict[str, Type[Skill]] = {}

    def register(self, skill_cls: Type[Skill]) -> None:
        """注册技能"""
        self._skills[skill_cls().name] = skill_cls

    def get_skill(self, name: str) -> Type[Skill]:
        """获取技能类"""
        return self._skills.get(name)

from typing import Any, Callable, Dict


class MessageBus:
    """简单消息总线实现"""

    def __init__(self):
        self._subscribers: Dict[str, list[Callable]] = {}

    def subscribe(self, topic: str, callback: Callable) -> None:
        """订阅消息"""
        if topic not in self._subscribers:
            self._subscribers[topic] = []
        self._subscribers[topic].append(callback)

    def publish(self, topic: str, message: Any) -> None:
        """发布消息"""
        if topic in self._subscribers:
            for callback in self._subscribers[topic]:
                callback(message)

在多线程环境下，技能注册中心和消息总线需要保证线程安全。可以使用锁机制来保护共享数据。

对于冷启动性能问题，可以考虑以下优化：