Agent Skill 开发实战：从零构建智能对话系统的避坑指南

4次阅读

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

智能对话系统的核心能力来源于一个个独立的 Agent Skill（对话技能）。就像手机里的 APP 一样，每个 Skill 负责处理特定领域的用户请求，比如查天气、订餐、控制智能家居等。当用户说 ” 明天上海会下雨吗 ” 时，系统需要准确调用天气查询 Skill，而不是误触发音乐播放 Skill。

上下文丢失 ：用户问 ” 那北京呢？” 时，系统忘记前文在讨论天气
意图冲突 ：” 订披萨 ” 同时触发外卖 Skill 和菜谱 Skill
状态混乱 ：多轮对话中用户突然切换话题导致状态机崩溃
异常处理缺失 ：API 调用失败时直接抛出代码错误
性能瓶颈 ：每次请求都重新初始化整个 Skill

推荐使用 Rasa 3.x 的 ” 模块化技能 ” 方案，主要组件包括：

技能路由 ：通过 DIETClassifier 识别意图并路由到对应 Skill
对话状态机 ：每个 Skill 维护独立的 Tracker 对象
上下文缓存 ：Redis 存储跨轮次对话数据
异常处理层 ：统一处理 API 超时 / 格式错误

# 技能基类示例（抽象类）from abc import ABC, abstractmethod
from typing import Dict, Any

class BaseSkill(ABC):
    @property
    @abstractmethod
    def skill_name(self) -> str:
        pass

    @abstractmethod
    def execute(self, tracker: "Tracker") -> Dict[str, Any]:
        """
        :param tracker: 当前对话状态跟踪器
        :return: 包含 text/image 等响应内容的字典
        """

# 单元测试示例
import unittest
from unittest.mock import MagicMock

class TestBaseSkill(unittest.TestCase):
    def test_abstract_methods(self):
        with self.assertRaises(TypeError):
            BaseSkill()  # 抽象类不能实例化

class WeatherSkill(BaseSkill):
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.cache = {}  # 简单内存缓存

    @property
    def skill_name(self) -> str:
        return "weather"

    def execute(self, tracker: "Tracker") -> Dict[str, Any]:
        # 获取最后一条用户消息
        latest_message = tracker.latest_message

        # 实体抽取（城市 + 时间）entities = {e["entity"]: e["value"] for e in latest_message["entities"]}
        city = entities.get("city", "北京")  # 默认北京
        date = entities.get("time", "今天")

        # 检查缓存
        cache_key = f"{city}_{date}"
        if cache_key in self.cache:
            return {"text": f"（缓存）{city}{date} 天气: {self.cache[cache_key]}"}

        # 调用天气 API（模拟）try:
            weather = self._call_weather_api(city, date)
            self.cache[cache_key] = weather  # 更新缓存
            return {"text": f"{city}{date} 天气: {weather}",
                "image": f"weather_icons/{weather}.png"  # 附加天气图标
            }
        except Exception as e:
            return {
                "text": "天气服务暂时不可用",
                "error": str(e)
            }

    def _call_weather_api(self, city: str, date: str) -> str:
        """模拟 API 调用，实际项目替换为真实请求"""
        # 这里应该有重试机制和超时设置
        weather_map = {"今天": "晴", "明天": "多云", "后天": "小雨"}
        return weather_map.get(date, "未知")

对话缓存策略
短期缓存：用 Redis 存 1 小时内的对话上下文
长期记忆：重要信息存入用户数据库
本地缓存：高频数据保存在 Skill 内存中

懒加载机制

# 在__init__.py 中动态注册 Skill
def get_skill(skill_name: str) -> BaseSkill:
    if skill_name == "weather":
        from .weather import WeatherSkill
        return WeatherSkill(api_key=os.getenv("WEATHER_API_KEY"))
    # 其他技能...

异步处理

async def async_execute(self, tracker):
    # 使用 aiohttp 代替 requests
    async with aiohttp.ClientSession() as session:
        async with session.get(api_url) as resp:
            return await resp.json()