基于Agent+Skill机器人的智能对话系统架构设计与实战

1次阅读

没有评论

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

传统对话系统常面临两个核心问题：

扩展性差 ：每新增一个业务功能都需要修改主流程代码，形成典型的 ” 面条式 ” 架构。例如电商场景中同时需要物流查询和退换货功能时，代码耦合度会急剧上升。
技能复用率低 ：相同功能在不同渠道（如微信 /APP）需要重复开发。我们曾有个案例：天气查询功能在三个平台存在三套相似代码，维护成本增加 300%。

Agent 作为系统大脑，主要包含两个子系统：

对话管理引擎
维护对话状态机（DST）
处理超时 / 中断等异常流程
记录用户画像数据
上下文管理器
采用环形缓冲区存储最近 5 轮对话
支持基于 LRU 的缓存淘汰
实现跨技能上下文共享（如用户先说 ” 查天气 ” 再问 ” 会下雨吗 ”）

所有 Skill 必须实现以下接口：

class BaseSkill:
    @classmethod
    def get_intent(cls) -> str:
        """返回技能绑定的意图标识"""

    async def execute(self, context: dict) -> dict:
        """
        输入：标准化上下文（包含用户输入、历史记录等）输出：必须包含 response 字段
        """

采用动态加权决策流程：

各 Skill 上报意图置信度（0-1）
结合技能优先级权重（可在管理后台配置）
最终得分 = 置信度 × 权重 × 衰减系数
衰减系数：连续触发同技能时逐次减半

from abc import ABC, abstractmethod
import asyncio

class BaseSkill(ABC):
    """所有技能必须继承的基类"""

    @classmethod
    @abstractmethod
    def get_intent(cls) -> str:
        raise NotImplementedError

    @abstractmethod
    async def execute(self, context: dict) -> dict:
        """:param context: {'user_input': str,'session_id': str,'slots': dict   # 已填充的语义槽}
        :return: {
            'response': str,
            'new_slots': dict
        }
        """
        raise NotImplementedError

    async def safe_execute(self, context: dict) -> dict:
        """带异常处理的执行封装"""
        try:
            return await self.execute(context)
        except Exception as e:
            return {'response': f"技能执行出错: {str(e)}",
                'error': True
            }

import aiohttp
from datetime import datetime

class WeatherSkill(BaseSkill):
    API_URL = "https://api.weather.com/v3"

    @classmethod
    def get_intent(cls) -> str:
        return "query_weather"

    async def execute(self, context: dict) -> dict:
        # 从语义槽获取地点
        location = context['slots'].get('location')
        if not location:
            return {
                'response': "请问要查询哪个城市的天气?",
                'need_slot': 'location'
            }

        async with aiohttp.ClientSession() as session:
            async with session.get(f"{self.API_URL}/nowcast",
                params={"loc": location}
            ) as resp:
                data = await resp.json()

        return {
            'response': (f"{location} 当前天气: {data['condition']}\n"
                f"温度: {data['temp']}℃ 湿度: {data['humidity']}%"
            ),
            'new_slots': {'last_query': datetime.now().isoformat()}
        }