介绍基于 Rasa、Transformers 和 FastAPI 构建 AI 智能客服系统的完整方案。对比了 Rasa、Dialogflow 和 LangChain，选定 Rasa 处理复杂多轮对话。详解了对话状态机构建、BERT 意图识别微调及 FastAPI 接口封装。涵盖生产环境避坑指南（数据闭环、会话隔离、模型热更新）及性能压测结果。强调规则引擎与深度学习混合策略，为开发者提供从选型到落地的实战参考。
最近在做一个 AI 智能客服项目，从零开始摸索，踩了不少坑，也积累了一些经验。今天就把这个基于开源技术栈的完整方案整理出来，希望能帮到同样想自己动手搭建的开发者朋友们。这个方案的核心是 Rasa、Transformers 和 FastAPI，目标是构建一个既能快速上手，又能应对生产环境挑战的智能客服系统。
一开始你可能觉得，用现成的 SaaS 服务多省事。但真到了业务复杂、数据敏感或者需要深度定制的时候，自建就成了刚需。不过，这条路并不平坦，我遇到的几个典型挑战是：
面对众多框架，我做了一个简单的横向对比，主要聚焦在意图识别和实体抽取这两个 NLU 核心任务上。
我的选择：考虑到我们需要处理复杂的、带状态的业务对话（如售后投诉流程），且对数据本地化有要求，最终选择了 Rasa 作为对话引擎的核心。对于需要深度语义理解或知识问答的部分，可以后期集成 LangChain 调用 LLM 作为补充。
确定了 Rasa 作为主干，我们开始搭建三个核心模块。
Rasa 的对话管理由 domain.yml、stories.yml 和 rules.yml 等文件定义。这里展示一个简单的'天气查询'多轮对话状态机实现。
首先，定义领域 (domain.yml)：
intents:
- greet
- inquire_weather
- inform_city
- deny
- affirm
entities:
- city
slots:
city:
type: text
mappings:
- type: from_entity
entity: city
responses:
utter_greet:
- text: "你好！我是天气助手。"
utter_ask_city:
- text: "请问你想查询哪个城市的天气？"
utter_weather_info:
- text: "{city}的天气是晴天，温度 25 度。" # 实际中这里会调用 API
utter_default:
- text: "抱歉，我没听明白。"
actions:
- utter_greet
- utter_ask_city
- utter_weather_info
- utter_default
- action_validate_city # 自定义的验证动作
然后，用故事 (stories.yml) 描述成功的对话路径：
- story: happy path weather inquiry
steps:
- intent: greet
- action: utter_greet
- intent: inquire_weather
- action: utter_ask_city
- intent: inform_city
entities:
- city: "北京"
- slot_was_set:
- city: "北京"
- action: action_validate_city
- action: utter_weather_info
- story: user provides city immediately
steps:
- intent: inquire_weather
entities:
- city: "上海"
- slot_was_set:
- city: "上海"
- action: action_validate_city
- action: utter_weather_info
最后，用规则 (rules.yml) 处理固定模式的对话，优先级高于故事：
- rule: Say `hello` when user sends a greet message
steps:
- intent: greet
- action: utter_greet
关键点：slots 用于存储对话状态（如用户提到的城市）。stories 用于训练一个预测下一动作的机器学习模型，而 rules 是硬编码的、确定性的对话路径。自定义动作（如 action_validate_city）用 Python 编写，可以连接数据库或外部 API。
虽然 Rasa 内置的 DIET 分类器不错，但在特定领域，用预训练模型如 BERT 微调，效果往往更好。这里给出一个用 PyTorch 和 Hugging Face transformers 库微调 BERT 进行意图分类的简化示例。
import torch
from torch.utils.data import Dataset, DataLoader
from transformers import BertTokenizer, BertForSequenceClassification, AdamW
from sklearn.model_selection import train_test_split
# 1. 准备数据
class IntentDataset(Dataset):
def __init__(self, texts, labels, tokenizer, max_len):
self.texts = texts
self.labels = labels
self.tokenizer = tokenizer
self.max_len = max_len
def __len__(self):
return len(self.texts)
def __getitem__(self, idx):
text = str(self.texts[idx])
label = self.labels[idx]
encoding = self.tokenizer.encode_plus(
text,
add_special_tokens=True,
max_length=self.max_len,
return_token_type_ids=False,
truncation=True,
return_attention_mask=True,
return_tensors='pt',
)
return {
'input_ids': encoding['input_ids'].flatten(),
'attention_mask': encoding['attention_mask'].flatten(),
'labels': torch.tensor(label, dtype=torch.long)
}
# 假设 texts_list 和 labels_list 是你的训练数据和标签
train_texts, val_texts, train_labels, val_labels = train_test_split(texts_list, labels_list, test_size=0.2)
# 2. 初始化模型和分词器
MODEL_NAME = 'bert-base-chinese' # 中文任务
tokenizer = BertTokenizer.from_pretrained(MODEL_NAME)
model = BertForSequenceClassification.from_pretrained(MODEL_NAME, num_labels=NUM_INTENTS)
# 3. 创建数据加载器
train_dataset = IntentDataset(train_texts, train_labels, tokenizer, max_len=128)
val_dataset = IntentDataset(val_texts, val_labels, tokenizer, max_len=128)
train_loader = DataLoader(train_dataset, batch_size=16, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=16)
# 4. 训练循环（简化版）
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = model.to(device)
optimizer = AdamW(model.parameters(), lr=2e-5)
for epoch in range(3): # 训练 3 轮
model.train()
for batch in train_loader:
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
labels = batch['labels'].to(device)
outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
loss = outputs.loss
loss.backward()
optimizer.step()
optimizer.zero_grad()
# 在验证集上评估...
# print(f"Epoch {epoch}, Loss: {loss.item()}")
时间复杂度分析：BERT 模型的前向传播复杂度大致为 O(L * H^2)，其中 L 是序列长度，H 是隐藏层维度。微调阶段，主要开销在于矩阵运算，在 GPU 上可以并行加速。对于在线推理，单条请求的延迟通常在几十到几百毫秒，取决于模型大小和硬件。
Rasa 本身提供 HTTP API，但为了更好的工程控制（如鉴权、限流、监控），我们常用 FastAPI 再封装一层。
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
import uvicorn
from typing import Optional
import asyncio
from your_rasa_agent import agent # 假设你已经加载了 Rasa 智能体
app = FastAPI(title="AI 智能客服 API")
class UserMessage(BaseModel):
sender_id: str # 用户会话 ID，用于隔离对话
message: str # 用户消息文本
session_id: Optional[str] = None # 可选，用于更细粒度会话
@app.post("/chat")
async def chat_endpoint(user_msg: UserMessage):
"""处理用户消息并返回机器人回复"""
try:
# 将消息发送给 Rasa 智能体处理
# 这里演示异步处理，避免阻塞
responses = await asyncio.to_thread(
agent.handle_text,
text_message=user_msg.message,
sender_id=user_msg.sender_id
)
# 提取回复文本
bot_replies = [r.get("text", "") for r in responses if "text" in r]
return {
"sender_id": user_msg.sender_id,
"responses": bot_replies,
"status": "success"
}
except Exception as e:
raise HTTPException(status_code=500, detail=f"处理消息时出错：{str(e)}")
@app.on_event("startup")
async def startup_event():
"""服务启动时，可以在这里加载模型"""
# await load_models()
print("AI 客服服务已启动。")
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
工程化要点：
async/await 和 asyncio.to_thread 处理可能阻塞的 Rasa 调用，提高并发能力。sender_id 确保不同用户的对话状态独立。HTTPException 返回清晰的错误信息。Depends 添加认证等中间件。代码跑起来只是第一步，要稳定服务，还得解决下面几个问题。
nlu_fallback（低置信度）或导致用户不满的对话。定期（如每周）将这些'bad cases'整理出来，标注正确的意图和实体，加入到训练数据中重新训练模型。这个'收集 - 标注 - 训练 - 部署'的循环是提升系统效果的生命线。InMemoryTrackerStore）在内存中，重启即丢失，且无法分布式扩展。生产环境必须使用外部存储，如 Redis 或 SQL 数据库。
RedisTrackerStore 或 SQLTrackerStore。这样，不同服务实例都能访问到同一份对话状态，支持水平扩展。同时，为每个 sender_id 设置合理的会话过期时间（TTL），定期清理僵尸会话。POST /model) 动态加载新模型。你可以训练好新模型后，通过 API 通知运行中的服务加载。理论再好，也得看实际表现。我在一台 4 核 8G 的云服务器上，对封装好的 FastAPI 服务（背后是 Rasa）进行了简单的压力测试，使用 locust 工具模拟用户请求。
结论：这个配置足以支撑中小型应用。如果流量更大，可以考虑：1) 将 NLU 模型服务（BERT）单独部署，并做负载均衡；2) 使用性能更好的 Transformer 模型（如蒸馏后的模型）；3) 对话管理部分，Rasa 本身也可以水平扩展多个实例。
在项目后期，我们面临一个经典问题：有些对话流程非常固定（例如，'重置密码' -> 跳转到重置页面），用规则（rules.yml）写死又快又准；有些场景则变化多端（例如，用户描述产品故障），需要深度学习模型去理解。
我的经验是采用 '混合策略'：
nlu_fallback 动作，可以设计成让用户澄清，或转入基于规则的澄清流程。平衡之道：没有绝对的好坏。初期可以多用规则快速上线，保证核心流程畅通。随着数据积累，逐步将一些模式清晰的规则可以转化为训练数据，让模型学习，从而减少规则维护成本。最终目标是让规则负责'确定性'，模型应对'不确定性'，两者协同工作。
搭建一个完整的 AI 智能客服系统，就像搭积木，选对组件（Rasa, BERT, FastAPI）很重要，但更重要的是理解它们如何连接，以及如何在真实的流量和业务需求面前保持稳定和进化。这个过程充满挑战，但看到机器人能准确理解用户并解决问题时，成就感也是巨大的。希望这篇笔记能为你提供一个清晰的起点和实用的避坑地图。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 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