Python 开源聊天机器人在电商外贸智能客服中的实战应用与架构解析

关键组件为 DIETClassifier，双意图实体 Transformer。每种语言准备独立训练数据文件（如 nlu_en.yml），在 domain.yml 中统一管理。

# nlu_en.yml
version: "3.1"
nlu:
  - intent: track_order
    examples: |
      - Where is my order?
      - I want to track my package.
      - Has my order shipped yet?
      - [Order number](order_number) AB12345678, what's the status?
  - intent: return_item
    examples: |
      - I need to return a product.
      - How do I start a return?
      - This item is broken, I want a refund.

2. 电商订单查询 Action Server 实现

用户询问订单状态触发自定义 Action，运行于独立 Python 服务器，调用电商平台 API。

import logging
from typing import Any, Text, Dict, List
from rasa_sdk import Action, Tracker
from rasa_sdk.executor import CollectingDispatcher
import requests
import jwt
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

class ActionTrackOrder(Action):
    def name(self) -> Text:
        return "action_track_order"

    def _get_api_token(self) -> str:
        """生成 JWT 鉴权 Token"""
        secret_key = "your-secret-key"
        payload = {
            'iss': 'your-issuer',
            'exp': datetime.utcnow() + timedelta(minutes=5),
            'api_key': 'your-api-key'
        }
        token = jwt.encode(payload, secret_key, algorithm='HS256')
        if isinstance(token, bytes):
            token = token.decode('utf-8')
        return token

    async def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]:
        order_number = next(tracker.get_latest_entity_values("order_number"), None)
        if not order_number:
            dispatcher.utter_message(text="Please provide your order number.")
            return []
        api_url = f"https://api.your-ecommerce.com/orders/{order_number}"
        headers = {"Authorization": f"Bearer {self._get_api_token()}"}
        try:
            response = requests.get(api_url, headers=headers, timeout=10.0)
            response.raise_for_status()
            order_data = response.json()
            status = order_data.get('status', 'unknown')
            carrier = order_data.get('carrier', 'unknown')
            tracking_num = order_data.get('tracking_number', 'N/A')
            msg = f"Your order {order_number} is currently '{status}'. "
            if tracking_num != 'N/A':
                msg += f"Track it with {carrier}: {tracking_num}."
            dispatcher.utter_message(text=msg)
        except requests.exceptions.Timeout:
            logger.error(f"API timeout for order {order_number}")
            dispatcher.utter_message(text="The order system is busy, please try again later.")
        except requests.exceptions.RequestException as e:
            logger.error(f"API error for order {order_number}: {e}")
            dispatcher.utter_message(text="Sorry, I couldn't fetch your order details at the moment.")
        except KeyError as e:
            logger.error(f"Unexpected API response format for {order_number}: {e}")
            dispatcher.utter_message(text="Received an unexpected response from the order system.")
        return []

包含 JWT 鉴权、网络请求、超时处理及异常回复。

3. 对话策略与状态管理

多轮对话需状态管理。Rasa 使用 TEDPolicy 预测下一步。在 stories.yml 定义路径，domain.yml 定义表单收集信息。

# domain.yml 片段
forms:
  return_form:
    required_slots:
      - order_number
      - item_sku
      - return_reason
slots:
  order_number:
    type: text
    mappings:
      - type: from_entity
        entity: order_number
  item_sku:
    type: text
    mappings:
      - type: from_entity
        entity: item_sku

使用 FormAction 验证槽位填充值。

性能优化

1. 压力测试

使用 Locust 模拟高并发。

# locustfile.py
from locust import HttpUser, task, between
import random

class ChatbotUser(HttpUser):
    wait_time = between(1, 3)
    host = "http://your-rasa-server:5005"

    @task
    def send_message(self):
        messages = [
            "Hello",
            "Where is my order?",
            "I want to return something.",
            "What's your return policy?"
        ]
        msg = random.choice(messages)
        self.client.post("/webhooks/rest/webhook", json={"sender": f"user_{random.randint(1,10000)}", "message": msg})

目标 TPS 达到 2000+，P95 响应时间低于 2 秒。

2. 缓解冷启动延迟

引入 Redis 作为消息队列（RedisLockStore 和 RedisEventBroker）。

# endpoints.yml
action_endpoint:
  url: "http://action-server:5055/webhook"
lock_store:
  type: redis
  url: "redis://redis-host:6379"
  db: 0
  key_prefix: "rasa_lock:"
event_broker:
  type: redis
  url: "redis://redis-host:6379"
  db: 1

持久化对话状态，避免单点故障。

常见问题与解决方案

1. 多语言语料清洗

训练数据质量决定模型上限。

import re

def clean_text(text: str) -> str:
    """清洗用户输入文本"""
    cleaned = re.sub(r'[^�-￿]', ' ', text, flags=re.UNICODE)
    cleaned = re.sub(r'\s+', ' ', cleaned)
    cleaned = cleaned.strip()
    cleaned = re.sub(r'([!?.])\1+', r'\1', cleaned)
    return cleaned

注意清洗函数主要用于训练数据准备，线上预测时 Rasa 内置分词器会处理原始输入。

2. Django Channels 与 Rasa 的 Websocket 保活

前端用 Django Channels 做 WebSocket，需与 Rasa /socket.io 保持长连接。

• 前端：实现心跳机制，定期发送 ping，设置重连逻辑。
• Rasa 端：调整 sanic 服务器 WebSocket 配置（ping_timeout, ping_interval）。
• 基础设施：Nginx 调整 WebSocket 超时设置。

# consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer
import asyncio

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        # ... 连接 Rasa 逻辑 ...
        self.keepalive_task = asyncio.create_task(self._send_heartbeat())

    async def _send_heartbeat(self):
        while True:
            await asyncio.sleep(30)
            try:
                await self.send(text_data=json.dumps({"event": "ping"}))
            except:
                break

总结与展望

系统上线后，高峰期自动处理约 70% 常见咨询，整体响应效率提升超过 30%。 未来探索方向：结合大语言模型（LLM）增强'理解'和'生成'能力。

• 意图识别兜底：置信度低时抛给 LLM。
• 上下文润色：LLM 负责措辞自然、人性化。
• 知识库问答增强：Rasa 路由至 LLM 文档问答模块。 混合架构（Rasa 处理确定流程 + LLM 增强语义）是现阶段务实选择。