基于 Rasa 框架构建跨境电商智能客服系统的实践。涵盖技术选型对比（Rasa vs ChatterBot）、核心模块实现（多语言 NLU、Action Server 对接电商 API、对话状态管理）、性能优化（Locust 压测、Redis 消息队列）以及常见问题处理（语料清洗、WebSocket 保活）。最后探讨了结合 LLM 增强语义理解的混合架构方向。系统上线后自动处理约 70% 常见咨询，显著提升响应效率。
本项目针对跨境电商场景，旨在解决多时区、多语言及高并发咨询难题。传统客服模式难以应对半夜订单咨询、小语种理解及手动查单效率低等问题。业务需求包括自动处理常见问题、对接电商后台查询订单，并强调响应速度与准确性。
团队对比了主流 Python 开源聊天机器人框架，核心诉求是意图识别准、扩展性强、社区活跃。
DIETClassifier 进行意图和实体识别，准确率超 92%。TED Policy 处理多轮对话，支持自定义 Action 集成复杂业务逻辑。支持中、英、西三种语言。Rasa 3.x 配置文件 config.yml 核心如下：
language: en
pipeline:
- name: WhitespaceTokenizer
- name: RegexFeaturizer
- name: LexicalSyntacticFeaturizer
- name: CountVectorsFeaturizer
analyzer: char_wb
min_ngram: 1
max_ngram: 4
- name: DIETClassifier
epochs: 100
constrain_similarities: true
- name: EntitySynonymMapper
- name: ResponseSelector
epochs: 50
关键组件为 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.
用户询问订单状态触发自定义 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 鉴权、网络请求、超时处理及异常回复。
多轮对话需状态管理。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 验证槽位填充值。
使用 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 秒。
引入 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
持久化对话状态，避免单点故障。
训练数据质量决定模型上限。
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 内置分词器会处理原始输入。
前端用 Django Channels 做 WebSocket，需与 Rasa /socket.io 保持长连接。
ping，设置重连逻辑。sanic 服务器 WebSocket 配置（ping_timeout, ping_interval）。# 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）增强'理解'和'生成'能力。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online