第 78 章
案例一：企业级代码审查系统（Plugin + Hook + LSP + CI/CD 完整架构）
第七十八章：构建 AI 客服系统：意图识别、知识库问答与人工兜底
78.1 AI 客服系统的架构全景
AI 客服系统是企业部署 Claude 最成熟、ROI 最清晰的场景之一。一个完整的 AI 客服系统不是单一的"问答机器人"，而是一个由多个协同模块组成的智能管道：
用户消息输入
↓
[预处理层] 语言检测 → 消息清洗 → 会话上下文拼接
↓
[意图识别层] 意图分类 → 实体提取 → 置信度评估
↓
[路由层] 规则路由 / 模型路由 → 决定处理路径
↙ ↓ ↘
[FAQ回答] [知识库问答] [人工升级]
↓ ↓ ↓
[后处理层] 响应格式化 → 品牌声音检查 → 发送
↓
[反馈循环] 用户评价 → 日志记录 → 持续优化
本章将逐层拆解这个架构，提供可直接用于生产的代码实现。
78.2 意图识别（Intent Recognition）
78.2.1 意图分类体系设计
意图分类是整个系统的路由核心。设计意图体系时需要考虑：
- 粒度：意图不能太粗（一个"询价"意图无法区分不同产品），也不能太细（导致覆盖率不足）
- 层次结构：使用树状结构，先分大类，再细分
- 置信度阈值：每个意图需要设定触发自动回复的置信度门槛
from anthropic import Anthropic
import json
client = Anthropic()
# 意图分类体系定义
INTENT_TAXONOMY = {
"account_issues": {
"description": "账户相关问题",
"subtypes": {
"login_problem": "登录/密码问题",
"account_verification": "账户验证/实名认证",
"account_suspension": "账户封禁/限制",
"account_deletion": "注销账户"
},
"auto_handle_threshold": 0.85
},
"billing": {
"description": "账单/支付问题",
"subtypes": {
"payment_failed": "支付失败",
"refund_request": "退款申请",
"invoice_request": "发票申请",
"subscription_inquiry": "订阅咨询"
},
"auto_handle_threshold": 0.80 # 账单问题需更高置信度
},
"product_inquiry": {
"description": "产品功能咨询",
"subtypes": {
"feature_question": "功能使用疑问",
"pricing_inquiry": "定价咨询",
"compatibility": "兼容性问题",
"feature_request": "功能建议"
},
"auto_handle_threshold": 0.75
},
"technical_support": {
"description": "技术故障报告",
"subtypes": {
"bug_report": "Bug 报告",
"performance_issue": "性能问题",
"integration_help": "集成帮助",
"api_issue": "API 问题"
},
"auto_handle_threshold": 0.70 # 技术问题复杂度高，降低自动处理阈值
},
"complaint": {
"description": "投诉/不满",
"subtypes": {
"service_complaint": "服务投诉",
"product_complaint": "产品投诉",
"escalation_request": "要求升级人工"
},
"auto_handle_threshold": 0.0 # 投诉始终升级人工
}
}
INTENT_CLASSIFIER_SYSTEM = f"""你是一个精确的客服意图分类器。
意图分类体系：
{json.dumps(INTENT_TAXONOMY, ensure_ascii=False, indent=2)}
分类规则：
1. 识别用户消息的主要意图和子意图
2. 如果消息包含多个意图，按重要性排序，最多识别 2 个
3. 提取关键实体（产品名称、订单号、金额等）
4. 评估用户的情绪状态（neutral/frustrated/urgent/angry）
输出 JSON 格式：
{{
"primary_intent": "账单/refund_request",
"secondary_intent": null,
"entities": {{"order_id": "ORD-12345", "amount": "299元"}},
"emotion": "frustrated",
"confidence": 0.92,
"requires_context": false,
"raw_issue": "用户想要退款"
}}"""
def classify_intent(user_message: str, conversation_history: list = None) -> dict:
"""
分类用户意图
conversation_history: 最近 N 轮对话历史（用于上下文理解）
"""
messages = []
# 添加对话历史作为上下文
if conversation_history:
history_text = "\n".join([
f"用户：{turn['user']}\n客服：{turn['assistant']}"
for turn in conversation_history[-3:] # 最近3轮
])
messages.append({
"role": "user",
"content": f"对话历史：\n{history_text}\n\n当前消息：{user_message}"
})
else:
messages.append({"role": "user", "content": user_message})
response = client.messages.create(
model="claude-haiku-4-5", # 使用轻量模型，意图分类不需要最强模型
max_tokens=300,
system=INTENT_CLASSIFIER_SYSTEM,
messages=messages
)
try:
return json.loads(response.content[0].text)
except json.JSONDecodeError:
return {
"primary_intent": "unknown",
"confidence": 0.0,
"error": "分类失败"
}
78.2.2 情绪检测与升级触发
ESCALATION_TRIGGERS = {
"explicit_request": ["要人工", "转人工", "找真人", "speak to human", "human agent"],
"emotion_threshold": "angry", # 检测到愤怒情绪时升级
"intent_always_escalate": ["complaint", "escalation_request"],
"failed_attempts": 3, # 连续 3 次未解决自动升级
"high_value_keywords": ["法律", "律师", "投诉到工商", "媒体曝光", "退款不受理"]
}
def should_escalate(
intent_result: dict,
conversation_history: list,
failed_attempts: int
) -> dict:
"""
判断是否需要升级到人工客服
"""
reasons = []
# 检查意图是否总是需要升级
primary = intent_result.get("primary_intent", "")
for always_escalate in ESCALATION_TRIGGERS["intent_always_escalate"]:
if always_escalate in primary:
reasons.append(f"意图类型需要人工处理：{always_escalate}")
# 检查情绪
if intent_result.get("emotion") == "angry":
reasons.append("用户情绪激动")
# 检查置信度
confidence = intent_result.get("confidence", 0)
intent_category = primary.split("/")[0] if "/" in primary else primary
threshold = INTENT_TAXONOMY.get(intent_category, {}).get("auto_handle_threshold", 0.8)
if confidence < threshold:
reasons.append(f"置信度 {confidence:.2f} 低于阈值 {threshold}")
# 检查失败次数
if failed_attempts >= ESCALATION_TRIGGERS["failed_attempts"]:
reasons.append(f"已尝试 {failed_attempts} 次未解决")
# 检查高风险关键词
latest_message = ""
if conversation_history:
latest_message = conversation_history[-1].get("user", "")
for keyword in ESCALATION_TRIGGERS["high_value_keywords"]:
if keyword in latest_message:
reasons.append(f"检测到高风险关键词：{keyword}")
return {
"should_escalate": len(reasons) > 0,
"reasons": reasons,
"priority": "high" if "情绪激动" in reasons or "高风险" in str(reasons) else "normal"
}
78.3 知识库问答（Knowledge Base Q&A）
78.3.1 知识库架构
AI 客服的知识库通常由三类内容组成：
KNOWLEDGE_BASE_STRUCTURE = {
"faq": {
"description": "常见问题与标准答案",
"format": "问题-答案对",
"update_frequency": "weekly",
"retrieval_method": "semantic_search"
},
"product_docs": {
"description": "产品文档、操作手册",
"format": "结构化文档",
"update_frequency": "per_release",
"retrieval_method": "hybrid_search"
},
"policy_docs": {
"description": "退款政策、服务条款、隐私政策",
"format": "条款文档",
"update_frequency": "monthly",
"retrieval_method": "keyword_search"
},
"case_history": {
"description": "历史成功解决的客服案例",
"format": "案例库",
"update_frequency": "continuous",
"retrieval_method": "semantic_search"
}
}
78.3.2 基于检索增强的问答
class CustomerServiceKnowledgeBase:
"""客服知识库（简化实现，生产环境应接入向量数据库）"""
def __init__(self):
self.faqs = []
self.policies = {}
self.product_docs = []
def search(self, query: str, top_k: int = 5) -> list:
"""
检索相关知识库内容
生产环境应替换为向量数据库检索
"""
# 这里展示接口设计，实际应用应调用 Pinecone/Qdrant 等
results = []
# FAQ 语义搜索（简化版）
for faq in self.faqs:
relevance = self._simple_relevance(query, faq["question"])
if relevance > 0.5:
results.append({
"type": "faq",
"content": f"Q: {faq['question']}\nA: {faq['answer']}",
"relevance": relevance,
"metadata": faq.get("metadata", {})
})
# 政策文档关键词匹配
query_lower = query.lower()
for policy_name, policy_content in self.policies.items():
keywords = policy_content.get("keywords", [])
if any(kw in query_lower for kw in keywords):
results.append({
"type": "policy",
"content": policy_content["content"],
"relevance": 0.8,
"metadata": {"policy": policy_name}
})
# 按相关度排序
results.sort(key=lambda x: x["relevance"], reverse=True)
return results[:top_k]
def _simple_relevance(self, query: str, text: str) -> float:
"""简化的相关度计算（生产环境应使用嵌入向量）"""
query_words = set(query.lower().split())
text_words = set(text.lower().split())
overlap = query_words & text_words
return len(overlap) / (len(query_words) + 1)
def answer_with_knowledge_base(
user_query: str,
knowledge_base: CustomerServiceKnowledgeBase,
company_name: str,
intent_result: dict
) -> dict:
"""
基于知识库的客服问答
"""
# 检索相关知识
relevant_docs = knowledge_base.search(user_query, top_k=5)
if not relevant_docs:
return {
"answer": None,
"confidence": 0.0,
"source": "no_knowledge",
"should_escalate": True
}
# 构建上下文
context_text = "\n\n".join([
f"[{doc['type'].upper()}] {doc['content']}"
for doc in relevant_docs
])
# 调用 Claude 生成答案
qa_system = f"""你是 {company_name} 的专业客服代表。
回答规则：
1. 只基于提供的知识库内容回答
2. 如果知识库中没有相关信息，明确说明无法回答并建议联系人工
3. 不要编造信息或做出知识库外的承诺
4. 语气要专业、温暖、有帮助
5. 如果涉及具体政策，引用具体条款
当前用户情绪：{intent_result.get('emotion', 'neutral')}
（如果用户情绪消极，在回答中先表达理解和歉意）"""
qa_prompt = f"""用户问题：{user_query}
相关知识库内容：
{context_text}
请根据知识库内容回答用户问题。如果知识库内容不足以回答，请直接说明。"""
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=500,
system=qa_system,
messages=[{"role": "user", "content": qa_prompt}]
)
answer = response.content[0].text
# 评估答案置信度
has_answer = "无法" not in answer and "无相关" not in answer and "请联系人工" not in answer
return {
"answer": answer,
"confidence": 0.85 if has_answer else 0.2,
"source": "knowledge_base",
"referenced_docs": [doc["type"] for doc in relevant_docs[:3]],
"should_escalate": not has_answer
}
78.4 对话流程管理
78.4.1 多轮对话状态机
from enum import Enum
class ConversationState(Enum):
GREETING = "greeting"
INTENT_COLLECTION = "intent_collection"
INFO_GATHERING = "info_gathering"
RESOLVING = "resolving"
ESCALATING = "escalating"
RESOLVED = "resolved"
CLOSED = "closed"
class CustomerServiceConversation:
"""AI 客服会话管理器"""
def __init__(self, session_id: str, company_config: dict):
self.session_id = session_id
self.config = company_config
self.state = ConversationState.GREETING
self.history = []
self.intent_history = []
self.failed_attempts = 0
self.gathered_info = {}
self.escalation_info = None
def process_message(self, user_message: str) -> dict:
"""
处理用户消息，返回响应
"""
# Step 1: 意图识别
intent = classify_intent(user_message, self.history[-5:] if self.history else None)
self.intent_history.append(intent)
# Step 2: 判断是否需要升级
escalation_check = should_escalate(intent, self.history, self.failed_attempts)
if escalation_check["should_escalate"]:
return self._handle_escalation(user_message, escalation_check)
# Step 3: 根据意图路由
if intent.get("confidence", 0) < 0.6:
return self._clarify_intent(user_message)
# Step 4: 知识库查询和回答
kb = self.config.get("knowledge_base")
if kb:
qa_result = answer_with_knowledge_base(
user_message, kb,
self.config["company_name"],
intent
)
if qa_result["confidence"] > 0.7:
response = qa_result["answer"]
self._record_turn(user_message, response, intent, "knowledge_base")
return {
"response": response,
"state": "resolved",
"requires_followup": True
}
# Step 5: 通用 LLM 回答
self.failed_attempts += 1
return self._general_response(user_message, intent)
def _handle_escalation(self, user_message: str, escalation_check: dict) -> dict:
"""处理升级到人工客服"""
self.state = ConversationState.ESCALATING
# 生成升级摘要
summary = self._generate_handoff_summary()
escalation_message = self._generate_escalation_message(
escalation_check["reasons"],
escalation_check["priority"]
)
self.escalation_info = {
"trigger_reasons": escalation_check["reasons"],
"priority": escalation_check["priority"],
"conversation_summary": summary,
"user_info": self.gathered_info,
"intent_history": [i.get("primary_intent") for i in self.intent_history]
}
return {
"response": escalation_message,
"state": "escalating",
"escalation_data": self.escalation_info
}
def _generate_handoff_summary(self) -> str:
"""生成交接给人工客服的会话摘要"""
if not self.history:
return "新会话，无历史记录"
history_text = "\n".join([
f"用户：{turn['user']}\nAI：{turn['assistant']}"
for turn in self.history[-5:]
])
summary_prompt = f"""请为以下客服对话生成简洁的接手摘要（不超过200字）。
对话记录：
{history_text}
摘要应包含：
1. 用户的主要问题
2. 已尝试的解决方法
3. 为什么需要升级人工
4. 用户情绪状态"""
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=300,
messages=[{"role": "user", "content": summary_prompt}]
)
return response.content[0].text
def _generate_escalation_message(self, reasons: list, priority: str) -> str:
"""生成通知用户升级到人工的消息"""
wait_time = "2-3 分钟" if priority == "high" else "5-10 分钟"
return f"""我理解您的问题需要专业人工客服的协助。我已将您的对话记录和问题详情传递给人工客服团队。
预计等待时间：{wait_time}
在等待期间，您也可以：
- 查看我们的帮助中心（help.example.com）
- 通过邮件联系我们（[email protected]）
感谢您的耐心等待！"""
def _record_turn(self, user_msg: str, assistant_msg: str, intent: dict, source: str):
self.history.append({
"user": user_msg,
"assistant": assistant_msg,
"intent": intent.get("primary_intent"),
"source": source
})
78.5 人工兜底的交接协议（Handoff Protocol）
78.5.1 交接数据包
class HandoffPackage:
"""
人工客服接手时的完整数据包
"""
def __init__(self, conversation: CustomerServiceConversation):
self.conversation = conversation
def generate(self) -> dict:
return {
"session_id": self.conversation.session_id,
"timestamp": get_current_timestamp(),
"priority": self._assess_priority(),
"user_profile": {
"identified_info": self.conversation.gathered_info,
"emotion_trajectory": self._get_emotion_trajectory(),
"interaction_count": len(self.conversation.history)
},
"issue_summary": {
"primary_issue": self._get_primary_issue(),
"intent_history": [i.get("primary_intent") for i in self.conversation.intent_history],
"attempted_solutions": self._get_attempted_solutions(),
"unresolved_aspects": self._get_unresolved_aspects()
},
"ai_assessment": {
"escalation_reasons": self.conversation.escalation_info.get("trigger_reasons", []),
"recommended_action": self._recommend_action(),
"relevant_policies": self._identify_relevant_policies()
},
"full_conversation": self.conversation.history
}
def _assess_priority(self) -> str:
if self.conversation.escalation_info:
return self.conversation.escalation_info.get("priority", "normal")
return "normal"
def _get_emotion_trajectory(self) -> list:
return [
intent.get("emotion", "neutral")
for intent in self.conversation.intent_history
]
def _get_primary_issue(self) -> str:
if self.conversation.intent_history:
return self.conversation.intent_history[0].get("raw_issue", "未识别")
return "未知问题"
def _get_attempted_solutions(self) -> list:
return [
turn.get("source", "unknown")
for turn in self.conversation.history
if turn.get("source") != "unknown"
]
def _get_unresolved_aspects(self) -> str:
if not self.conversation.history:
return "尚未开始对话"
# 最后一轮 AI 回复中未解决的内容
return self.conversation.history[-1].get("assistant", "")[:200]
def _recommend_action(self) -> str:
intents = [i.get("primary_intent", "") for i in self.conversation.intent_history]
if any("refund" in i for i in intents):
return "验证订单信息，处理退款申请"
if any("complaint" in i for i in intents):
return "先表达关切，了解具体诉求，按投诉流程处理"
return "继续了解用户问题并提供解决方案"
def _identify_relevant_policies(self) -> list:
intents = [i.get("primary_intent", "") for i in self.conversation.intent_history]
policies = []
if any("refund" in i for i in intents):
policies.append("退款政策")
if any("billing" in i for i in intents):
policies.append("账单政策")
return policies
def get_current_timestamp(self):
from datetime import datetime
return datetime.now().isoformat()
78.6 监控与持续优化
78.6.1 关键指标
CUSTOMER_SERVICE_METRICS = {
"deflection_rate": {
"description": "AI 成功自主解决（无需人工）的比例",
"target": ">= 0.70",
"formula": "ai_resolved / total_conversations"
},
"escalation_accuracy": {
"description": "升级决策的准确性（升级后人工确认确实需要升级的比例）",
"target": ">= 0.85",
"formula": "justified_escalations / total_escalations"
},
"first_contact_resolution": {
"description": "单次对话解决率",
"target": ">= 0.65",
"formula": "single_session_resolved / total_conversations"
},
"intent_recognition_accuracy": {
"description": "意图识别准确率",
"target": ">= 0.90",
"formula": "correct_intents / total_classified"
},
"user_satisfaction_score": {
"description": "用户满意度（1-5）",
"target": ">= 4.2",
"formula": "user_rating_average"
}
}
小结
AI 客服系统的核心价值在于：用精准的意图识别把问题导向正确的处理路径，用知识库问答高效解决标准化问题，用可信赖的升级机制在 AI 能力边界处将用户安全交接给人工。
设计这类系统的核心原则是保守优于冒进：宁可多升级几次人工，也不能让用户的合理诉求在自动化流程中被卡死。升级触发条件要宽松，升级的体验要流畅，交接数据包要完整——这三点是 AI 客服系统可信赖性的基础。