AI Skill开发实战：从零构建高可用智能对话系统

1次阅读

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

开发 AI Skill 时最常见的挑战集中在自然语言理解（NLU）和对话管理（DM）两个核心环节。根据我的项目经验，这些痛点会直接影响用户体验：

意图识别准确率低：当用户说 ” 明天会下雨吗 ” 和 ” 下周天气怎么样 ” 时，系统需要准确识别都是 weather_query 意图。但实际场景中，同义表达和口语化表述会造成大量误判
实体抽取不稳定：时间实体 ” 下周五下午 ” 需要被正确解析为 datetime 对象，但用户可能说 ” 大后天 ” 或 ” 国庆节第二天 ” 这类非结构化表达
多轮对话状态维护困难：用户可能在确认天气后突然问 ” 那空气质量呢？”，系统需要保持上下文而不重复询问城市信息
生产环境扩展性问题：当 Skill 需要同时处理数千并发对话时，如何保证状态不丢失、响应速度不下降

主流对话系统框架各有特点，这是我们团队的评估结论：

维度	Rasa	LUIS	DialogFlow
成本	开源免费	按调用量收费	免费版有限制
定制化	完全自主可控	部分参数可调	黑盒优化
部署复杂度	需自建服务	云端即用	云端即用
NLU 语言	支持中文需额外配置	中文支持较好	中文支持最好
学习曲线	陡峭（需理解全套架构）	中等	平缓

选择建议：

需要数据隐私和深度定制选 Rasa
追求快速上线且预算充足选 DialogFlow
微软生态集成选 LUIS

rasa init --no-prompt

intents:
  - weather_query: {examples: |
      - 今天天气
      - 明天会下雨吗
      - 查询 [北京](city) 的空气质量}
  - deny: {examples: |
      - 不需要
      - 暂时不用}

entities:
  - city
  - date

responses:
  utter_ask_city:
    - text: "您想查询哪个城市呢？"
  utter_weather_result:
    - text: "{city}{date}的天气是{description}, 气温{temp}℃"

创建actions/weather_action.py：

class FetchWeatherAction(Action):
    def name(self) -> Text:
        return "fetch_weather"

    async def run(
        self, 
        dispatcher: CollectingDispatcher,
        tracker: Tracker,
        domain: Dict[Text, Any]
    ) -> List[Dict[Text, Any]]:
        # 从对话状态获取参数
        city = tracker.get_slot("city")
        date = tracker.get_slot("date") or datetime.now().strftime("%Y-%m-%d")

        try:
            # 调用第三方天气 API（示例用伪代码）weather_data = requests.get(f"https://api.weather.com/v1?city={city}&date={date}",
                timeout=3
            ).json()

            dispatcher.utter_message(
                template="utter_weather_result",
                city=city,
                date=date,
                description=weather_data["condition"],
                temp=weather_data["temp"]
            )
        except Exception as e:
            logger.error(f"天气查询失败: {e}")
            dispatcher.utter_message("暂时无法获取天气信息，请稍后再试")

        return []

修改 endpoints.yml 启用 Redis 跟踪器：

tracker_store:
  type: redis
  url: "redis://localhost:6379"
  db: 0
  key_prefix: "rasa_tracker:"
  expiration: 86400  # 24 小时过期

在自定义通道（如微信接口）添加过滤层：

def sanitize_message(text):
    patterns = [(r"\b\d{11}\b", "[PHONE]"),  # 手机号
        (r"\b\d{18}\b", "[ID_CARD]") # 身份证
    ]
    for pattern, repl in patterns:
        text = re.sub(pattern, repl, text)
    return text

每个意图至少准备 30 个多样本
避免近义意图样本数差异过大（如 ” 查询 ” 和 ” 搜索 ”）
使用 Rasa 数据验证工具：
```
rasa data validate --fail-on-warnings
```

# 在自定义 Action 中添加重试逻辑
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def call_weather_api(params):
    async with timeout(5):
        return await aiohttp.request(...)