AI智能客服系统开发完全教程
本教程全面讲解AI智能客服系统的核心架构与开发技术，帮助开发者构建生产级AI客服系统。
目录
- AI智能客服概述与架构
- 意图识别与槽位填充
- 多轮对话管理
- 知识库构建与RAG检索
- 情感分析与情绪安抚
- 对话流编排引擎
- FAQ自动匹配系统
- 多渠道接入方案
- 人工客服无缝转接
- 语音客服集成
- 客服数据分析与报表
- 企业级部署方案
- 实战：电商智能客服系统
- 最佳实践与常见问题
1. AI智能客服概述与架构
1.1 什么是AI智能客服
AI智能客服是利用自然语言处理（NLP）、大语言模型（LLM）、知识图谱等技术构建的自动化客户服务系统。与传统的基于规则的客服机器人不同，AI智能客服能够理解用户意图、进行多轮对话、检索知识库并生成自然流畅的回复。
1.2 系统架构总览
一个完整的AI智能客服系统通常包含以下核心模块：
┌─────────────────────────────────────────────────┐
│ 用户接入层 │
│ (Web/APP/微信/钉钉/电话/邮件) │
└──────────────────────┬──────────────────────────┘
│
┌──────────────────────▼──────────────────────────┐
│ 消息路由层 │
│ (统一消息收发/多渠道适配/消息队列) │
└──────────────────────┬──────────────────────────┘
│
┌──────────────────────▼──────────────────────────┐
│ AI理解层 │
│ (意图识别/槽位填充/情感分析/实体抽取) │
└──────────────────────┬──────────────────────────┘
│
┌──────────────────────▼──────────────────────────┐
│ 对话管理层 │
│ (对话状态跟踪/对话流编排/多轮管理) │
└──────────────────────┬──────────────────────────┘
│
┌──────────────────────▼──────────────────────────┐
│ 知识检索层 │
│ (FAQ匹配/RAG检索/知识图谱/数据库查询) │
└──────────────────────┬──────────────────────────┘
│
┌──────────────────────▼──────────────────────────┐
│ 回复生成层 │
│ (模板回复/LLM生成/多模态回复) │
└──────────────────────┬──────────────────────────┘
│
┌──────────────────────▼──────────────────────────┐
│ 人工协作层 │
│ (转人工判断/人工坐席/质检分析) │
└─────────────────────────────────────────────────┘
1.3 技术栈选型
| 组件 | 推荐方案 | 适用场景 |
|---|---|---|
| LLM引擎 | DeepSeek/Qwen/GPT-4o | 对话理解与生成 |
| 向量数据库 | Milvus/Qdrant/Chroma | 知识库检索 |
| 对话框架 | Rasa/Dify/自研 | 对话管理 |
| 消息队列 | RabbitMQ/Kafka | 高并发消息处理 |
| 缓存 | Redis | 会话状态存储 |
| 数据库 | PostgreSQL/MySQL | 业务数据存储 |
2. 意图识别与槽位填充
2.1 意图识别原理
意图识别是将用户输入分类到预定义类别的过程。常见方法包括：
传统方法： 基于规则和关键词匹配 机器学习方法： 使用BERT等预训练模型进行分类 大模型方法： 使用LLM进行零样本或少样本意图分类
2.2 基于BERT的意图识别
import torch
from transformers import BertTokenizer, BertForSequenceClassification
from torch.utils.data import Dataset, DataLoader
class IntentDataset(Dataset):
def __init__(self, texts, labels, tokenizer, max_len=128):
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):
encoding = self.tokenizer(
self.texts[idx],
max_length=self.max_len,
padding='max_length',
truncation=True,
return_tensors='pt'
)
return {
'input_ids': encoding['input_ids'].squeeze(),
'attention_mask': encoding['attention_mask'].squeeze(),
'label': torch.tensor(self.labels[idx])
}
# 意图类别定义
INTENTS = {
0: "查询订单",
1: "申请退款",
2: "商品咨询",
3: "物流查询",
4: "投诉建议",
5: "账户问题",
6: "其他"
}
# 训练数据示例
train_data = [
("我的订单到哪了", 3),
("想退货怎么操作", 1),
("这个商品有什么颜色", 2),
("快递怎么还没到", 3),
("我要投诉", 4),
("密码忘了怎么办", 5),
]
# 模型训练
def train_intent_model():
tokenizer = BertTokenizer.from_pretrained('bert-base-chinese')
model = BertForSequenceClassification.from_pretrained(
'bert-base-chinese',
num_labels=len(INTENTS)
)
texts = [t for t, _ in train_data]
labels = [l for _, l in train_data]
dataset = IntentDataset(texts, labels, tokenizer)
dataloader = DataLoader(dataset, batch_size=16, shuffle=True)
optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5)
model.train()
for epoch in range(5):
for batch in dataloader:
outputs = model(
input_ids=batch['input_ids'],
attention_mask=batch['attention_mask'],
labels=batch['label']
)
loss = outputs.loss
loss.backward()
optimizer.step()
optimizer.zero_grad()
return model, tokenizer
2.3 基于LLM的零样本意图识别
import json
from openai import OpenAI
client = OpenAI(api_key="your-api-key", base_url="https://api.deepseek.com")
INTENT_DEFINITIONS = """
可用意图类别：
1. 查询订单 - 用户想查询订单状态、订单详情
2. 申请退款 - 用户想退货、退款、换货
3. 商品咨询 - 用户询问商品信息、规格、价格
4. 物流查询 - 用户查询物流状态、配送进度
5. 投诉建议 - 用户投诉问题或提出建议
6. 账户问题 - 用户的账户相关问题（密码、登录等）
7. 其他 - 不属于以上任何类别
"""
def classify_intent_llm(user_message: str) -> dict:
"""使用LLM进行意图识别"""
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": f"""你是一个意图分类助手。
{INTENT_DEFINITIONS}
请以JSON格式返回分类结果：
{{"intent": "意图名称", "confidence": 0.95, "entities": {{"关键实体": "值"}}}}
"""},
{"role": "user", "content": user_message}
],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
# 测试
result = classify_intent_llm("我上周买的那个蓝色连衣裙还没收到")
# 输出: {"intent": "物流查询", "confidence": 0.92, "entities": {"商品": "蓝色连衣裙", "时间": "上周"}}
2.4 槽位填充
槽位填充是从用户输入中提取结构化信息的过程：
class SlotFiller:
"""基于规则+LLM的槽位填充"""
# 定义各意图需要的槽位
SLOT_SCHEMA = {
"查询订单": ["order_id", "time_range"],
"申请退款": ["order_id", "reason", "refund_type"],
"商品咨询": ["product_name", "attribute"],
"物流查询": ["order_id", "tracking_number"],
}
def fill_slots(self, intent: str, user_message: str,
existing_slots: dict = None) -> dict:
"""填充槽位"""
slots = existing_slots or {}
required = self.SLOT_SCHEMA.get(intent, [])
missing = [s for s in required if s not in slots]
if not missing:
return slots
# 使用LLM提取缺失槽位
prompt = f"""从用户消息中提取以下信息：
需要提取：{', '.join(missing)}
用户消息：{user_message}
已知信息：{json.dumps(slots, ensure_ascii=False)}
以JSON格式返回提取结果，未找到的字段返回null。"""
response = client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
new_slots = json.loads(response.choices[0].message.content)
slots.update({k: v for k, v in new_slots.items() if v is not None})
return slots
3. 多轮对话管理
3.1 对话状态跟踪
对话状态跟踪（DST）是维护对话上下文的核心机制：
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import json
@dataclass
class DialogueState:
"""对话状态"""
session_id: str
user_id: str
current_intent: Optional[str] = None
slots: dict = field(default_factory=dict)
history: list = field(default_factory=list)
turn_count: int = 0
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
last_active: str = field(default_factory=lambda: datetime.now().isoformat())
context: dict = field(default_factory=dict) # 额外上下文
class DialogueManager:
"""对话管理器"""
def __init__(self, redis_client=None):
self.redis = redis_client
self.states = {} # 内存存储（生产环境用Redis）
def get_state(self, session_id: str) -> DialogueState:
"""获取对话状态"""
if session_id in self.states:
return self.states[session_id]
# 尝试从Redis加载
if self.redis:
data = self.redis.get(f"dialogue:{session_id}")
if data:
state = DialogueState(**json.loads(data))
self.states[session_id] = state
return state
# 创建新状态
state = DialogueState(session_id=session_id, user_id="")
self.states[session_id] = state
return state
def update_state(self, state: DialogueState):
"""更新对话状态"""
state.last_active = datetime.now().isoformat()
state.turn_count += 1
self.states[state.session_id] = state
# 持久化到Redis
if self.redis:
self.redis.setex(
f"dialogue:{state.session_id}",
1800, # 30分钟过期
json.dumps(state.__dict__, ensure_ascii=False)
)
def add_turn(self, state: DialogueState, role: str, content: str):
"""添加对话轮次"""
state.history.append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
# 保留最近20轮
if len(state.history) > 40:
state.history = state.history[-40:]
3.2 对话流程控制
class DialogueController:
"""对话流程控制器"""
def __init__(self, intent_classifier, slot_filler,
knowledge_retriever, response_generator):
self.intent_classifier = intent_classifier
self.slot_filler = slot_filler
self.knowledge_retriever = knowledge_retriever
self.response_generator = response_generator
self.dialogue_manager = DialogueManager()
async def process_message(self, session_id: str,
user_message: str) -> str:
"""处理用户消息"""
# 1. 获取对话状态
state = self.dialogue_manager.get_state(session_id)
# 2. 意图识别
intent_result = self.intent_classifier.classify(user_message)
intent = intent_result['intent']
confidence = intent_result['confidence']
# 3. 低置信度处理
if confidence < 0.6:
return self._handle_low_confidence(state, user_message)
# 4. 槽位填充
state.current_intent = intent
state.slots = self.slot_filler.fill_slots(
intent, user_message, state.slots
)
# 5. 检查是否需要追问
missing_slots = self.slot_filler.get_missing_slots(
intent, state.slots
)
if missing_slots:
return self._ask_for_slot(state, missing_slots[0])
# 6. 检查是否需要转人工
if self._should_transfer_to_human(state, user_message):
return self._transfer_to_human(state)
# 7. 知识检索与回复生成
context = self._build_context(state)
knowledge = self.knowledge_retriever.retrieve(
user_message, intent, context
)
response = self.response_generator.generate(
user_message, intent, state.slots, knowledge, context
)
# 8. 更新状态
self.dialogue_manager.add_turn(state, "user", user_message)
self.dialogue_manager.add_turn(state, "assistant", response)
self.dialogue_manager.update_state(state)
return response
def _handle_low_confidence(self, state, message):
"""低置信度处理"""
if state.turn_count > 2:
return "抱歉，我没有完全理解您的意思。您是想咨询以下哪个问题呢？\n" \
"1. 查询订单\n2. 申请退款\n3. 商品咨询\n4. 转人工客服"
return "抱歉，我没有理解您的意思，能换个方式描述一下吗？"
def _ask_for_slot(self, state, slot_name):
"""追问缺失槽位"""
prompts = {
"order_id": "请问您的订单号是多少？",
"reason": "请问您退货的原因是什么？",
"product_name": "请问您想咨询哪个商品？",
"tracking_number": "请提供您的快递单号。",
}
return prompts.get(slot_name, f"请提供{slot_name}信息")
def _should_transfer_to_human(self, state, message):
"""判断是否需要转人工"""
# 关键词触发
transfer_keywords = ["转人工", "人工客服", "投诉", "经理"]
if any(kw in message for kw in transfer_keywords):
return True
# 多轮未解决
if state.turn_count > 5 and state.context.get('unresolved', 0) > 2:
return True
return False
4. 知识库构建与RAG检索
4.1 知识库设计
from dataclasses import dataclass
from typing import List
import hashlib
@dataclass
class KnowledgeItem:
"""知识条目"""
id: str
title: str
content: str
category: str
tags: List[str]
source: str = ""
def __post_init__(self):
if not self.id:
self.id = hashlib.md5(
(self.title + self.content).encode()
).hexdigest()[:12]
class KnowledgeBase:
"""知识库管理"""
def __init__(self, vector_store, llm_client):
self.vector_store = vector_store
self.llm = llm_client
self.items = {}
def add_item(self, item: KnowledgeItem):
"""添加知识条目"""
self.items[item.id] = item
# 向量化存储
embedding = self._get_embedding(item.title + "\n" + item.content)
self.vector_store.upsert(
id=item.id,
vector=embedding,
metadata={
"title": item.title,
"category": item.category,
"tags": ",".join(item.tags)
}
)
def add_from_faq(self, faq_list: List[dict]):
"""从FAQ列表导入"""
for faq in faq_list:
item = KnowledgeItem(
id="",
title=faq['question'],
content=faq['answer'],
category=faq.get('category', '通用'),
tags=faq.get('tags', [])
)
self.add_item(item)
def retrieve(self, query: str, top_k: int = 3) -> List[dict]:
"""检索相关知识"""
query_embedding = self._get_embedding(query)
results = self.vector_store.search(
vector=query_embedding,
top_k=top_k
)
retrieved = []
for result in results:
item = self.items.get(result.id)
if item:
retrieved.append({
"title": item.title,
"content": item.content,
"score": result.score,
"category": item.category
})
return retrieved
def _get_embedding(self, text: str) -> List[float]:
"""获取文本向量"""
# 使用embedding模型
response = self.llm.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
4.2 RAG检索增强
class RAGRetriever:
"""RAG检索器"""
def __init__(self, knowledge_base, llm_client):
self.kb = knowledge_base
self.llm = llm_client
def retrieve_and_generate(self, query: str,
intent: str = None) -> str:
"""检索并生成回复"""
# 1. 查询改写 - 提升检索质量
rewritten = self._rewrite_query(query, intent)
# 2. 检索相关知识
results = self.kb.retrieve(rewritten, top_k=3)
if not results:
return None
# 3. 构建上下文
context = "\n\n".join([
f"【{r['title']}】\n{r['content']}"
for r in results
])
# 4. LLM生成回复
response = self.llm.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": """你是一个专业的客服助手。
请根据提供的知识库内容回答用户问题。
- 只使用知识库中的信息回答
- 如果知识库中没有相关信息，说"抱歉，我需要为您转接人工客服"
- 回复要简洁、专业、友好"""},
{"role": "user", "content": f"""知识库内容：
{context}
用户问题：{query}"""}
]
)
return response.choices[0].message.content
def _rewrite_query(self, query: str, intent: str = None) -> str:
"""查询改写"""
prompt = f"将以下客服对话改写为更适合搜索的形式：\n{query}"
if intent:
prompt += f"\n意图：{intent}"
response = self.llm.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": prompt}],
max_tokens=100
)
return response.choices[0].message.content
5. 情感分析与情绪安抚
5.1 情感检测
class SentimentAnalyzer:
"""情感分析器"""
SENTIMENT_PROMPT = """分析以下客服对话中用户的情感状态。
返回JSON格式：
{
"sentiment": "positive/neutral/negative/angry",
"intensity": 0.0-1.0,
"emotion_tags": ["焦虑", "愤怒", "满意", ...],
"need_comfort": true/false
}"""
def analyze(self, message: str, history: list = None) -> dict:
"""分析用户情感"""
context = ""
if history:
recent = history[-6:] # 最近3轮
context = "\n".join([f"{h['role']}: {h['content']}" for h in recent])
response = client.chat.completions.create(
model="deepseek-chat",
messages=[
{"role": "system", "content": self.SENTIMENT_PROMPT},
{"role": "user", "content": f"对话历史：\n{context}\n\n当前消息：{message}"}
],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
class ComfortResponse:
"""情绪安抚策略"""
COMFORT_TEMPLATES = {
"angry": [
"非常理解您的心情，遇到这样的情况确实让人着急。我会尽快帮您解决。",
"真的很抱歉给您带来了不好的体验，我马上为您处理。"
],
"anxious": [
"请您放心，我来帮您查看一下，很快就会有结果。",
"我理解您的担心，让我马上为您核实。"
],
"disappointed": [
"非常抱歉没有达到您的期望，我来看看能怎么帮您改善。",
"感谢您的反馈，我们会认真对待，让我为您妥善处理。"
]
}
def get_comfort(self, sentiment: dict) -> str:
"""获取安抚话术"""
emotion = sentiment.get('sentiment', 'neutral')
if emotion in self.COMFORT_TEMPLATES:
import random
return random.choice(self.COMFORT_TEMPLATES[emotion])
return ""
6. 对话流编排引擎
6.1 状态机对话流
from enum import Enum
from typing import Callable, Dict, List, Optional
class NodeType(Enum):
START = "start"
MESSAGE = "message"
INPUT = "input"
CONDITION = "condition"
ACTION = "action"
END = "end"
class FlowNode:
"""对话流节点"""
def __init__(self, node_id: str, node_type: NodeType,
content: str = "", **kwargs):
self.id = node_id
self.type = node_type
self.content = content
self.next_nodes: Dict[str, str] = {} # condition -> node_id
self.actions: List[Callable] = []
self.metadata = kwargs
class DialogueFlow:
"""对话流引擎"""
def __init__(self):
self.nodes: Dict[str, FlowNode] = {}
self.flows: Dict[str, str] = {} # flow_name -> start_node_id
def add_node(self, node: FlowNode):
self.nodes[node.id] = node
def connect(self, from_id: str, to_id: str, condition: str = "default"):
self.nodes[from_id].next_nodes[condition] = to_id
def execute(self, flow_name: str, session_state: dict) -> str:
"""执行对话流"""
current_id = self.flows.get(flow_name)
if not current_id:
return "对话流不存在"
while current_id:
node = self.nodes.get(current_id)
if not node:
break
if node.type == NodeType.END:
return node.content
if node.type == NodeType.MESSAGE:
return node.content
if node.type == NodeType.CONDITION:
# 根据条件选择下一个节点
condition = self._evaluate_condition(node, session_state)
current_id = node.next_nodes.get(condition)
continue
if node.type == NodeType.ACTION:
for action in node.actions:
action(session_state)
current_id = node.next_nodes.get("default")
continue
current_id = node.next_nodes.get("default")
return "对话流程结束"
def _evaluate_condition(self, node: FlowNode,
state: dict) -> str:
"""评估条件"""
condition_func = node.metadata.get('condition_func')
if condition_func:
return condition_func(state)
return "default"
# 构建退款对话流
def build_refund_flow():
flow = DialogueFlow()
flow.add_node(FlowNode("start", NodeType.START))
flow.add_node(FlowNode(
"ask_order", NodeType.MESSAGE,
"请提供您的订单号，我来为您查看退款资格。"
))
flow.add_node(FlowNode(
"check_eligibility", NodeType.ACTION,
actions=[lambda s: s.update({"eligible": True})]
))
flow.add_node(FlowNode(
"ask_reason", NodeType.INPUT,
"请选择退款原因：\n1. 商品质量问题\n2. 不想要了\n3. 收到商品与描述不符\n4. 其他"
))
flow.add_node(FlowNode(
"process_refund", NodeType.MESSAGE,
"已为您提交退款申请，预计1-3个工作日内到账。退款单号：{refund_id}"
))
flow.add_node(FlowNode("end", NodeType.END, "感谢您的咨询，祝您生活愉快！"))
flow.connect("start", "ask_order")
flow.connect("ask_order", "check_eligibility")
flow.connect("check_eligibility", "ask_reason")
flow.connect("ask_reason", "process_refund")
flow.connect("process_refund", "end")
flow.flows["退款流程"] = "start"
return flow
7. FAQ自动匹配系统
7.1 语义匹配
class FAQMatcher:
"""FAQ语义匹配"""
def __init__(self, embedding_model, faq_data: List[dict]):
self.embedding_model = embedding_model
self.faqs = faq_data
self._build_index()
def _build_index(self):
"""构建索引"""
self.questions = [faq['question'] for faq in self.faqs]
self.embeddings = self.embedding_model.encode(self.questions)
def match(self, query: str, top_k: int = 3,
threshold: float = 0.7) -> List[dict]:
"""匹配FAQ"""
query_embedding = self.embedding_model.encode([query])
# 计算余弦相似度
similarities = self._cosine_similarity(
query_embedding, self.embeddings
)[0]
# 排序取top_k
top_indices = similarities.argsort()[-top_k:][::-1]
results = []
for idx in top_indices:
score = float(similarities[idx])
if score >= threshold:
results.append({
"question": self.faqs[idx]['question'],
"answer": self.faqs[idx]['answer'],
"score": score,
"category": self.faqs[idx].get('category', '')
})
return results
def _cosine_similarity(self, a, b):
"""计算余弦相似度"""
import numpy as np
a_norm = a / np.linalg.norm(a, axis=1, keepdims=True)
b_norm = b / np.linalg.norm(b, axis=1, keepdims=True)
return np.dot(a_norm, b_norm.T)
8. 多渠道接入方案
8.1 统一消息接口
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
@dataclass
class UnifiedMessage:
"""统一消息格式"""
channel: str # 渠道：web/wechat/dingtalk/telegram
user_id: str # 用户ID
message_type: str # text/image/audio/file
content: str # 消息内容
media_url: Optional[str] = None
extra: dict = None
class ChannelAdapter(ABC):
"""渠道适配器基类"""
@abstractmethod
async def receive(self, raw_data: dict) -> UnifiedMessage:
"""接收消息并转换为统一格式"""
pass
@abstractmethod
async def send(self, user_id: str, response: str,
**kwargs) -> bool:
"""发送回复"""
pass
class WebChannelAdapter(ChannelAdapter):
"""Web渠道适配器"""
async def receive(self, raw_data: dict) -> UnifiedMessage:
return UnifiedMessage(
channel="web",
user_id=raw_data['user_id'],
message_type=raw_data.get('type', 'text'),
content=raw_data['message']
)
async def send(self, user_id: str, response: str,
**kwargs) -> bool:
# 通过WebSocket发送
ws = kwargs.get('ws_connection')
if ws:
await ws.send_json({
"type": "message",
"content": response
})
return True
return False
class WeChatChannelAdapter(ChannelAdapter):
"""微信公众号渠道适配器"""
def __init__(self, app_id: str, app_secret: str):
self.app_id = app_id
self.app_secret = app_secret
async def receive(self, raw_data: dict) -> UnifiedMessage:
return UnifiedMessage(
channel="wechat",
user_id=raw_data['FromUserName'],
message_type=raw_data.get('MsgType', 'text'),
content=raw_data.get('Content', '')
)
async def send(self, user_id: str, response: str,
**kwargs) -> bool:
# 调用微信API发送消息
import httpx
async with httpx.AsyncClient() as client:
token = await self._get_access_token(client)
url = f"https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token={token}"
data = {
"touser": user_id,
"msgtype": "text",
"text": {"content": response}
}
resp = await client.post(url, json=data)
return resp.status_code == 200
class ChannelRouter:
"""渠道路由器"""
def __init__(self):
self.adapters: Dict[str, ChannelAdapter] = {}
def register(self, channel: str, adapter: ChannelAdapter):
self.adapters[channel] = adapter
async def process_incoming(self, channel: str,
raw_data: dict) -> str:
adapter = self.adapters.get(channel)
if not adapter:
raise ValueError(f"未注册的渠道: {channel}")
message = await adapter.receive(raw_data)
return message # 返回统一消息供后续处理
async def send_response(self, channel: str, user_id: str,
response: str, **kwargs):
adapter = self.adapters.get(channel)
if adapter:
await adapter.send(user_id, response, **kwargs)
9. 人工客服无缝转接
9.1 转接策略
class TransferManager:
"""人工转接管理"""
def __init__(self):
self.agents = {} # 人工坐席
self.queue = [] # 等待队列
self.active_sessions = {} # 会话分配
def should_transfer(self, state: DialogueState,
sentiment: dict) -> tuple:
"""判断是否需要转人工"""
reasons = []
# 1. 用户主动要求
if state.context.get('user_requested_transfer'):
reasons.append("用户主动要求")
# 2. 情绪激动
if sentiment.get('sentiment') == 'angry' and \
sentiment.get('intensity', 0) > 0.7:
reasons.append("用户情绪激动")
# 3. 多轮未解决
if state.turn_count > 5:
reasons.append("多轮对话未解决问题")
# 4. 敏感场景
sensitive_intents = ["投诉建议", "账户安全"]
if state.current_intent in sensitive_intents:
reasons.append("敏感场景需要人工介入")
# 5. LLM判断无法回答
if state.context.get('llm_cannot_answer'):
reasons.append("AI无法解答")
return len(reasons) > 0, reasons
async def transfer(self, session_id: str, state: DialogueState,
reasons: list) -> str:
"""执行转接"""
# 获取可用坐席
agent = self._find_available_agent(state.current_intent)
if agent:
# 直接转接
self.active_sessions[session_id] = agent['id']
return f"正在为您转接人工客服（{agent['name']}），请稍候...\n" \
f"转接原因：{', '.join(reasons)}"
else:
# 加入等待队列
self.queue.append({
'session_id': session_id,
'state': state,
'reasons': reasons,
'position': len(self.queue) + 1
})
return f"当前排队人数：{len(self.queue)}人，" \
f"预计等待{len(self.queue) * 2}分钟。请耐心等待。"
def _find_available_agent(self, intent: str) -> dict:
"""查找可用坐席"""
for agent_id, agent in self.agents.items():
if agent['status'] == 'online' and \
intent in agent.get('skills', []):
return agent
return None
10. 语音客服集成
10.1 语音识别与合成
class VoiceCustomerService:
"""语音客服"""
def __init__(self, asr_model, tts_model, text_engine):
self.asr = asr_model # 语音识别
self.tts = tts_model # 语音合成
self.text_engine = text_engine # 文本对话引擎
async def process_audio(self, audio_data: bytes,
session_id: str) -> bytes:
"""处理语音输入"""
# 1. 语音转文字
text = await self.asr.recognize(audio_data)
# 2. 文本对话处理
response_text = await self.text_engine.process_message(
session_id, text
)
# 3. 文字转语音
audio_response = await self.tts.synthesize(response_text)
return audio_response
# 使用Whisper进行语音识别
import whisper
class WhisperASR:
def __init__(self, model_size="base"):
self.model = whisper.load_model(model_size)
async def recognize(self, audio_data: bytes) -> str:
import tempfile
with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f:
f.write(audio_data)
result = self.model.transcribe(f.name, language='zh')
return result['text']
11. 客服数据分析与报表
11.1 核心指标
from dataclasses import dataclass
from datetime import datetime, timedelta
@dataclass
class ServiceMetrics:
"""客服指标"""
total_conversations: int = 0
ai_resolved: int = 0
human_resolved: int = 0
avg_response_time: float = 0.0
avg_resolution_time: float = 0.0
satisfaction_score: float = 0.0
transfer_rate: float = 0.0
class AnalyticsEngine:
"""分析引擎"""
def __init__(self, db_connection):
self.db = db_connection
def generate_daily_report(self, date: str) -> dict:
"""生成日报"""
metrics = self._calculate_metrics(date)
top_intents = self._get_top_intents(date)
hourly_distribution = self._get_hourly_distribution(date)
return {
"date": date,
"metrics": metrics.__dict__,
"top_intents": top_intents,
"hourly_distribution": hourly_distribution,
"ai_resolution_rate": (
metrics.ai_resolved / metrics.total_conversations
if metrics.total_conversations > 0 else 0
)
}
def _calculate_metrics(self, date: str) -> ServiceMetrics:
"""计算核心指标"""
query = """
SELECT
COUNT(*) as total,
SUM(CASE WHEN resolved_by = 'ai' THEN 1 ELSE 0 END) as ai_resolved,
SUM(CASE WHEN resolved_by = 'human' THEN 1 ELSE 0 END) as human_resolved,
AVG(response_time_ms) as avg_response,
AVG(resolution_time_ms) as avg_resolution,
AVG(satisfaction) as avg_satisfaction
FROM conversations
WHERE DATE(created_at) = %s
"""
result = self.db.execute(query, (date,)).fetchone()
return ServiceMetrics(
total_conversations=result['total'],
ai_resolved=result['ai_resolved'],
human_resolved=result['human_resolved'],
avg_response_time=result['avg_response'],
avg_resolution_time=result['avg_resolution'],
satisfaction_score=result['avg_satisfaction'],
transfer_rate=result['human_resolved'] / result['total']
if result['total'] > 0 else 0
)
12. 企业级部署方案
12.1 高可用架构
# docker-compose.yml
version: '3.8'
services:
# API网关
nginx:
image: nginx:alpine
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
depends_on:
- customer-service
# 客服服务
customer-service:
build: .
replicas: 3
environment:
- REDIS_URL=redis://redis:6379
- DATABASE_URL=postgresql://user:pass@db:5432/cs
- LLM_API_KEY=${LLM_API_KEY}
depends_on:
- redis
- db
- milvus
# Redis缓存
redis:
image: redis:7-alpine
volumes:
- redis-data:/data
# PostgreSQL数据库
db:
image: postgres:15
environment:
POSTGRES_DB: cs
POSTGRES_USER: user
POSTGRES_PASSWORD: pass
volumes:
- pg-data:/var/lib/postgresql/data
# Milvus向量数据库
milvus:
image: milvusdb/milvus:latest
ports:
- "19530:19530"
volumes:
- milvus-data:/var/lib/milvus
volumes:
redis-data:
pg-data:
milvus-data:
12.2 性能优化
# 异步消息处理
import asyncio
from collections import defaultdict
import time
class RateLimiter:
"""速率限制器"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window = window_seconds
self.requests = defaultdict(list)
def is_allowed(self, user_id: str) -> bool:
now = time.time()
# 清理过期记录
self.requests[user_id] = [
t for t in self.requests[user_id]
if now - t < self.window
]
if len(self.requests[user_id]) >= self.max_requests:
return False
self.requests[user_id].append(now)
return True
class MessageQueue:
"""消息队列处理"""
def __init__(self, redis_client):
self.redis = redis_client
self.queue_name = "cs:message_queue"
async def enqueue(self, message: dict):
"""入队"""
await self.redis.rpush(
self.queue_name,
json.dumps(message)
)
async def dequeue(self, timeout: int = 0) -> dict:
"""出队"""
result = await self.redis.blpop(
self.queue_name, timeout=timeout
)
if result:
return json.loads(result[1])
return None
13. 实战：电商智能客服系统
13.1 系统完整实现
"""
电商智能客服系统 - 完整示例
"""
class ECommerceCustomerService:
"""电商智能客服"""
def __init__(self):
# 初始化各组件
self.intent_classifier = IntentClassifier()
self.slot_filler = SlotFiller()
self.knowledge_base = KnowledgeBase(vector_store, llm_client)
self.sentiment_analyzer = SentimentAnalyzer()
self.dialogue_manager = DialogueManager()
self.transfer_manager = TransferManager()
self.response_generator = ResponseGenerator()
async def handle_message(self, session_id: str,
user_message: str) -> dict:
"""处理用户消息"""
# 获取状态
state = self.dialogue_manager.get_state(session_id)
# 情感分析
sentiment = self.sentiment_analyzer.analyze(
user_message, state.history
)
# 判断是否需要转人工
need_transfer, reasons = self.transfer_manager.should_transfer(
state, sentiment
)
if need_transfer:
transfer_msg = await self.transfer_manager.transfer(
session_id, state, reasons
)
return {"response": transfer_msg, "type": "transfer"}
# 意图识别
intent_result = self.intent_classifier.classify(user_message)
# 槽位填充
state.slots = self.slot_filler.fill_slots(
intent_result['intent'], user_message, state.slots
)
# 根据意图路由处理
response = await self._route_by_intent(
intent_result, state, user_message
)
# 情绪安抚
if sentiment.get('need_comfort'):
comfort = ComfortResponse().get_comfort(sentiment)
response = comfort + "\n\n" + response
# 更新状态
self.dialogue_manager.add_turn(state, "user", user_message)
self.dialogue_manager.add_turn(state, "assistant", response)
self.dialogue_manager.update_state(state)
return {"response": response, "type": "ai"}
async def _route_by_intent(self, intent_result: dict,
state: DialogueState,
message: str) -> str:
"""根据意图路由"""
intent = intent_result['intent']
handlers = {
"查询订单": self._handle_order_query,
"申请退款": self._handle_refund,
"商品咨询": self._handle_product_query,
"物流查询": self._handle_logistics,
"投诉建议": self._handle_complaint,
"账户问题": self._handle_account,
}
handler = handlers.get(intent, self._handle_general)
return await handler(state, message)
async def _handle_order_query(self, state, message):
"""处理订单查询"""
order_id = state.slots.get('order_id')
if not order_id:
return "请提供您的订单号，我来为您查询。"
# 查询订单（示例）
order = self._query_order(order_id)
if order:
return f"订单 {order_id} 状态：{order['status']}\n" \
f"商品：{order['product']}\n" \
f"下单时间：{order['created_at']}"
return f"未找到订单 {order_id}，请确认订单号是否正确。"
async def _handle_refund(self, state, message):
"""处理退款"""
order_id = state.slots.get('order_id')
reason = state.slots.get('reason')
if not order_id:
return "请提供需要退款的订单号。"
if not reason:
return "请选择退款原因：\n1. 商品质量问题\n2. 不想要了\n3. 与描述不符\n4. 其他"
return f"已为订单 {order_id} 提交退款申请，原因：{reason}。" \
f"预计1-3个工作日内处理完成。"
async def _handle_product_query(self, state, message):
"""处理商品咨询"""
results = self.knowledge_base.retrieve(message, top_k=3)
if results:
return results[0]['content']
return "抱歉，暂未找到该商品的详细信息。需要我为您转接人工客服吗？"
async def _handle_logistics(self, state, message):
"""处理物流查询"""
order_id = state.slots.get('order_id')
if not order_id:
return "请提供订单号或快递单号，我来为您查询物流信息。"
return f"订单 {order_id} 的物流信息：\n" \
f"快递公司：顺丰速运\n" \
f"当前状态：运输中\n" \
f"预计明天送达"
async def _handle_complaint(self, state, message):
"""处理投诉"""
return "非常抱歉给您带来了不好的体验。" \
"您的投诉我们已经记录，工单号为：CP" + \
str(hash(state.session_id))[:8] + \
"\n我们会在24小时内给您回复。"
async def _handle_account(self, state, message):
"""处理账户问题"""
return "账户相关问题需要身份验证。" \
"请提供您的注册手机号或邮箱，我来为您核实。"
async def _handle_general(self, state, message):
"""通用处理"""
results = self.knowledge_base.retrieve(message, top_k=1)
if results and results[0]['score'] > 0.8:
return results[0]['content']
return "抱歉，我不太确定如何回答这个问题。需要我为您转接人工客服吗？"
def _query_order(self, order_id: str) -> dict:
"""查询订单（模拟）"""
# 实际应查询数据库
return {
"status": "已发货",
"product": "示例商品",
"created_at": "2024-01-15 10:30:00"
}
14. 最佳实践与常见问题
14.1 最佳实践
- 渐进式上线：先用FAQ匹配覆盖高频问题，再逐步引入LLM
- 兜底策略：设置多层兜底，确保用户问题总能得到响应
- 持续优化：定期分析对话日志，优化意图识别和知识库
- A/B测试：对比不同策略效果，数据驱动优化
- 监控告警：监控响应时间、解决率等关键指标
14.2 常见问题
| 问题 | 解决方案 |
|---|---|
| 回复不准确 | 优化知识库质量，增加FAQ覆盖 |
| 响应慢 | 引入缓存，优化LLM调用 |
| 转人工率高 | 提升AI能力，扩展知识库 |
| 用户体验差 | 增加情绪安抚，优化对话流程 |
总结
本教程详细讲解了AI智能客服系统的完整技术栈，从意图识别、多轮对话、知识检索到多渠道接入和人工转接。通过结合大语言模型和传统NLP技术，可以构建出真正实用的智能客服系统。
关键要点：
- 采用模块化架构，便于维护和扩展
- 结合规则和LLM，平衡准确性和灵活性
- 重视数据收集和持续优化
- 做好兜底和转人工机制
本教程内容原创，仅供参考学习使用。
