Agent+Skill机器人开发实战：从零搭建智能对话系统的避坑指南

1次阅读

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

传统聊天机器人最让人头疼的就是两件事：要么听不懂人话（意图识别不准），要么聊着聊着就失忆（上下文丢失）。我做过一个电商客服机器人，用户问完 ” 这件衣服有红色吗 ” 之后，接着说 ”M 码的呢 ”，结果系统完全不明白这个 ”M 码 ” 指的是刚才那件红衣服。

Agent+Skill 架构就像乐高积木：

Agent 是大脑，负责记忆对话状态和调度能力
Skill 是各种技能模块，每个只专注做一件事
通过标准化接口组合，就像给机器人安装可拆卸的 APP

对比了市面上三大方案后，我的选择依据是：

方案	优点	缺点
Rasa	完整的 NLU 管道	学习曲线陡峭
Dialogflow	谷歌的 NLU 能力强大	黑箱操作，定制困难
自建框架	完全可控，轻量	需要造轮子

选择 Python+Flask 是因为：

团队已有 Python 技术栈
Flask 的轻量级特性适合快速迭代
丰富的 NLP 生态（spaCy、Transformers 等）

from typing import Dict, Any

class DialogState:
    def __init__(self):
        self.current_skill: str = None
        self.slots: Dict[str, Any] = {}

class BaseAgent:
    def __init__(self):
        self.skills = {}
        self.state = DialogState()

    def register_skill(self, skill_name: str, skill):
        self.skills[skill_name] = skill

关键设计点：

使用类型注解提高代码可维护性
对话状态与业务逻辑分离
采用注册模式而非继承

from abc import ABC, abstractmethod

class BaseSkill(ABC):
    @abstractmethod
    def can_handle(self, user_input: str) -> bool:
        pass

    @abstractmethod
    def execute(self, agent: BaseAgent) -> str:
        pass

import requests
from urllib.parse import quote

class WeatherSkill(BaseSkill):
    def can_handle(self, user_input: str) -> bool:
        return "天气" in user_input

    def execute(self, agent: BaseAgent) -> str:
        city = self._extract_city(agent.state.slots)
        try:
            url = f"https://api.weather.com/v1?city={quote(city)}"
            resp = requests.get(url, timeout=3)
            resp.raise_for_status()
            return self._format_response(resp.json())
        except Exception as e:
            return f"查询失败：{str(e)}"

异常处理要点：

网络请求必须设置超时
城市参数需要 URL 编码
错误信息要友好但不过于详细

def handle_message(agent: BaseAgent, user_input: str) -> str:
    # 当前活跃 skill 优先
    if agent.state.current_skill:
        skill = agent.skills[agent.state.current_skill]
        if skill.can_handle(user_input):
            return skill.execute(agent)

    # 全局匹配
    for name, skill in agent.skills.items():
        if skill.can_handle(user_input):
            agent.state.current_skill = name
            return skill.execute(agent)

    return "抱歉，我不明白您的意思"

存储方案	写入速度	查询效率	适合场景
SQLite	中等	中等	小型单机部署
Redis	极快	极快	高并发生产环境
MySQL	较慢	较慢	需要复杂查询分析

使用 Locust 模拟用户并发请求：

安装：pip install locust
创建 locustfile.py:

from locust import HttpUser, task

class ChatbotUser(HttpUser):
    @task
    def ask_weather(self):
        self.client.post("/chat", json={"text":"北京天气"})

运行测试：locust -f locustfile.py

推荐使用 DFA 算法而非简单关键词替换：

class SensitiveFilter:
    def __init__(self):
        self.keywords = set()

    def add_word(self, word: str):
        self.keywords.add(word)

    def filter(self, text: str) -> str:
        for word in self.keywords:
            if word in text:
                return "[内容已屏蔽]"
        return text

常见问题：天气 Skill 修改了全局状态导致购物车信息丢失

解决方案：

class DialogState:
    def __init__(self):
        self.global_slots = {}
        self.skill_states = {}  # 每个 skill 独立命名空间

错误示范：

async def execute(self):
    self.state.current_city = "北京"  # 可能被其他请求覆盖
    await call_api()

正确做法：

async def execute(self):
    local_state = self.state.copy()
    local_state.current_city = "北京"
    await call_api()
    self.state.update(local_state)