本文目录导读：
我来提供一个完整的自动回复客服脚本设计思路,涵盖从基础到进阶的完整框架。
整体架构设计
系统流程图
用户消息 → 消息预处理 → 意图识别 → 回复生成 → 回复发送
↓ ↓ ↓
清洗/分词 分类/匹配 人工/自动技术栈选择
# 主流方案 - 规则引擎: Python + re + json - 机器学习: sklearn + jieba + TF-IDF - 深度学习: BERT + PyTorch/TensorFlow - 混合方案: 规则 + ML + 人工兜底
核心模块实现
消息预处理模块
import re
import jieba
from typing import Dict, List
class MessagePreprocessor:
"""消息预处理模块"""
def __init__(self):
self.stop_words = self.load_stop_words()
def clean_text(self, text: str) -> str:
"""文本清洗"""
# 去除特殊字符
text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9]', ' ', text)
# 去除URL
text = re.sub(r'http\S+|www\.\S+', '', text)
# 去除表情符
text = re.sub(r'[\U0001F300-\U0001F9FF]', '', text)
return text.strip()
def tokenize(self, text: str) -> List[str]:
"""中文分词"""
words = jieba.cut(text)
# 过滤停用词和单字
return [w for w in words if w not in self.stop_words and len(w) > 1]
意图识别模块
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.svm import SVC
import numpy as np
class IntentClassifier:
"""意图识别模块"""
def __init__(self):
self.vectorizer = TfidfVectorizer(max_features=10000)
self.classifier = SVC(kernel='linear', probability=True)
self.intents_map = {
'greeting': '寒暄问候',
'price': '价格咨询',
'shipping': '物流问题',
'return': '退换货',
'complaint': '投诉建议',
'product': '产品咨询',
'other': '其他'
}
def train(self, texts: List[str], labels: List[str]):
"""训练模型"""
X = self.vectorizer.fit_transform(texts)
self.classifier.fit(X, labels)
def predict(self, text: str) -> tuple:
"""预测意图"""
X = self.vectorizer.transform([text])
intent = self.classifier.predict(X)[0]
confidence = self.classifier.predict_proba(X).max()
return intent, confidence
回复生成引擎
import json
import random
from typing import Dict, Optional
class ReplyEngine:
"""回复生成引擎"""
def __init__(self, config_path: str):
self.load_reply_templates(config_path)
self.context = {} # 用户上下文
def load_reply_templates(self, config_path: str):
"""加载回复模板"""
with open(config_path, 'r', encoding='utf-8') as f:
self.templates = json.load(f)
def get_reply(self, intent: str, confidence: float,
user_message: str, user_id: str) -> str:
"""获取回复"""
# 1. 高置信度 - 直接回复
if confidence > 0.8:
return self.get_direct_reply(intent, user_message)
# 2. 中等置信度 - 带确认
elif confidence > 0.5:
return self.get_confirmation_reply(intent)
# 3. 低置信度 - 转人工
else:
return self.get_human_transfer_reply()
def get_direct_reply(self, intent: str, user_message: str) -> str:
"""获取直接回复"""
replies = self.templates.get(intent, [])
if not replies:
return "抱歉，我还在学习中，请稍后再试"
# 模板匹配回复
reply = random.choice(replies)
# 如果有商品名称等信息，可以个性化
product = self.extract_product(user_message)
if product:
reply = reply.replace('{product}', product)
return reply
上下文管理
import time
from collections import defaultdict, deque
class ContextManager:
"""对话上下文管理"""
def __init__(self, max_history: int = 10):
self.sessions = defaultdict(lambda: deque(maxlen=max_history))
self.session_timeout = 1800 # 30分钟
def get_context(self, user_id: str) -> dict:
"""获取用户上下文"""
session = self.sessions[user_id]
# 获取当前上下文
context = {
'history': list(session[-5:]), # 最近5条消息
'intent': self.get_current_intent(user_id),
'last_activity': self.get_last_activity(user_id)
}
return context
def update_context(self, user_id: str, message: str, intent: str, reply: str):
"""更新上下文"""
self.sessions[user_id].append({
'message': message,
'intent': intent,
'reply': reply,
'time': time.time()
})
知识库设计
from elasticsearch import Elasticsearch
import sqlite3
class KnowledgeBase:
"""知识库模块"""
def __init__(self):
self.es = Elasticsearch()
self.db = sqlite3.connect('knowledge.db')
self.init_db()
def search_answers(self, query: str, top_k: int = 3) -> List[dict]:
"""语义搜索"""
# ES搜索
body = {
"query": {
"match": {
"question": {
"query": query,
"fuzziness": "AUTO"
}
}
},
"size": top_k
}
result = self.es.search(index="faq", body=body)
return [hit['_source'] for hit in result['hits']['hits']]
def add_qa(self, question: str, answer: str, category: str = 'general'):
"""新增问答对"""
doc = {
'question': question,
'answer': answer,
'category': category,
'create_time': int(time.time())
}
self.es.index(index='faq', body=doc)
人工客服转接
from flask import Flask, request
import threading
class HumanHandoff:
"""人工转接模块"""
def __init__(self):
self.online_agents = {} # 在线客服列表
self.queues = defaultdict(list) # 等待队列
self.routing_rules = {
'complaint': 'priority', # 投诉优先
'vip': 'priority',
}
def route_to_human(self, user_id: str, intent: str,
messages: List[str]) -> bool:
"""转接人工客服"""
# 检查在线客服
if not self.online_agents:
return False
# 优先级处理
priority = self.routing_rules.get(intent, 'normal')
# 队列分配
if priority == 'priority':
agent = self.find_best_agent(user_id, priority=True)
self.assign_agent(user_id, agent)
return True
return False
def find_best_agent(self, user_id: str, priority: bool = False) -> str:
"""查找最合适的客服"""
# 简单策略：空闲时间最长的客服
available = [agent for agent, status in self.online_agents.items()
if status == 'free']
if priority and len(available) > 1:
# 优先分配给高级客服
return self.sort_agents_by_level(available)[0]
return available[0] if available else None
完整集成示例
class AutoReplySystem:
"""自动回复系统主类"""
def __init__(self, config: dict):
self.preprocessor = MessagePreprocessor()
self.intent_classifier = IntentClassifier()
self.reply_engine = ReplyEngine(config['templates_path'])
self.context_manager = ContextManager()
self.knowledge_base = KnowledgeBase()
self.human_handoff = HumanHandoff()
def handle_message(self, user_id: str, message: str) -> dict:
"""处理用户消息"""
# 1. 预处理
cleaned_text = self.preprocessor.clean_text(message)
tokens = self.preprocessor.tokenize(cleaned_text)
# 2. 获取上下文
context = self.context_manager.get_context(user_id)
# 3. 意图识别
intent, confidence = self.intent_classifier.predict(cleaned_text)
# 4. 知识库匹配
kb_answers = self.knowledge_base.search_answers(cleaned_text)
# 5. 生成回复
if confidence > 0.8 and kb_answers:
reply = kb_answers[0][:200] # 最匹配的答案
elif confidence > 0.5:
reply = self.reply_engine.get_reply(
intent, confidence, message, user_id
)
else:
# 转人工或提供兜底回复
if self.human_handoff.route_to_human(user_id, intent, [message]):
reply = "正在为您转接人工客服，请稍候..."
else:
reply = "抱歉，我暂时无法理解您的问题，请拨打客服热线或稍后重试。"
# 6. 更新上下文
self.context_manager.update_context(user_id, message, intent, reply)
return {
'user_id': user_id,
'reply': reply,
'intent': intent,
'confidence': confidence,
'timestamp': time.time()
}
评估与优化
# 1. 效果评估指标
metrics = {
'accuracy': '意图识别准确率',
'response_time': '响应时间',
'resolution_rate': '问题解决率',
'customer_satisfaction': '客户满意度',
'handoff_rate': '转人工率'
}
# 2. A/B测试
class ABTesting:
def test_reply_strategies(self, variant_a, variant_b):
"""测试不同回复策略"""
# 分别统计转化率、满意度
pass
# 3. 持续优化
- 定期收集失败案例
- 扩充知识库
- 优化模型定期重训练
- 调整意图分类阈值
部署建议
架构选择
- 轻量级：Flask + SQLite + 模板匹配
- 中等规模：FastAPI + Redis + Elasticsearch
- 大规模：微服务架构 + Kubernetes
性能优化
# 缓存机制
from functools import lru_cache
@lru_cache(maxsize=1000)
def get_cached_reply(intent, message):
"""缓存热门回复"""
return reply_engine.generate(intent, message)
# 异步处理
import asyncio
async def handle_async(user_id, message):
# 异步处理消息
pass
监控告警
# 关键指标监控 - 平均响应时间 - 意图识别置信度分布 - 转人工率变化 - 知识库命中率
这个思路框架涵盖了自动回复客服系统的核心功能，你可以根据实际需求选择所需模块，从简单的规则系统开始逐步升级到智能系统,关键是做好数据沉淀和持续优化。
