大模型让智能客服从"关键词匹配"时代跨入了"真正理解语义"的时代。但"能聊天"和"能当客服"之间还有很大的差距。一个可用的客服系统需要处理：意图识别、知识库检索、多轮上下文管理、工单创建、人机协作转接。
这篇文章从工程角度，完整拆解一个基于大模型的多轮对话客服系统。
一、系统架构
用户消息
│
┌──────▼──────┐
│ 对话管理器 │ ← 维护会话状态和上下文
└──────┬──────┘
│
┌────────▼────────┐
│ 意图识别 + 路由 │ ← Function Calling驱动
└────────┬────────┘
│
┌──────────────┼──────────────┐
│ │ │
┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐
│ 知识库问答 │ │ 业务操作 │ │ 转人工 │
│ (RAG检索) │ │ (查订单等)│ │ (工单系统)│
└───────────┘ └───────────┘ └───────────┘二、对话管理器
多轮对话的核心是上下文管理。每次用户发消息，系统需要把历史对话一起发给大模型。
import time
import json
class ConversationManager:
def __init__(self, max_history=20, session_timeout=1800):
self.sessions = {} # {session_id: {messages, last_active, metadata}}
self.max_history = max_history
self.session_timeout = session_timeout # 30分钟超时
def get_session(self, session_id):
"""获取或创建会话"""
now = time.time()
# 检查超时
if session_id in self.sessions:
if now - self.sessions[session_id]["last_active"] > self.session_timeout:
del self.sessions[session_id] # 超时，清除旧会话
if session_id not in self.sessions:
self.sessions[session_id] = {
"messages": [],
"last_active": now,
"metadata": {"turn_count": 0, "intent_history": []}
}
session = self.sessions[session_id]
session["last_active"] = now
return session
def add_message(self, session_id, role, content):
"""添加消息到会话"""
session = self.get_session(session_id)
session["messages"].append({"role": role, "content": content})
session["metadata"]["turn_count"] += 1
# 保持历史长度在限制内
if len(session["messages"]) > self.max_history:
# 保留system消息 + 最近的N条
system_msgs = [m for m in session["messages"] if m["role"] == "system"]
recent_msgs = session["messages"][-self.max_history:]
session["messages"] = system_msgs + recent_msgs
def get_messages(self, session_id):
"""获取完整的消息历史（供大模型使用）"""
session = self.get_session(session_id)
return session["messages"]会话超时非常重要。如果用户30分钟没说话再回来，大概率是新问题了。继续用旧上下文反而会干扰模型理解。
三、基于Function Calling的意图路由
传统客服用规则或分类模型做意图识别，准确率有限且维护成本高。大模型的Function Calling天然适合这个场景——让模型自己判断应该调用哪个"工具"。
CUSTOMER_SERVICE_TOOLS = [
{
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "从知识库中搜索产品信息、使用教程、常见问题的答案。当用户提问关于产品功能、使用方法、故障排查等问题时使用。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "用户问题的核心关键词，用于知识库检索"
},
"category": {
"type": "string",
"enum": ["产品功能", "使用教程", "故障排查", "退换货政策", "账户问题"],
"description": "问题所属分类"
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "query_order",
"description": "查询用户的订单信息，包括订单状态、物流信息、支付信息等。",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "订单编号"
},
"phone": {
"type": "string",
"description": "下单时的手机号（当没有订单号时，用手机号查询）"
}
}
}
}
},
{
"type": "function",
"function": {
"name": "create_ticket",
"description": "创建工单，将问题转交给人工客服处理。当问题超出AI能力范围、用户明确要求转人工、或涉及退款/投诉等敏感操作时使用。",
"parameters": {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "工单标题，简述用户问题"
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "urgent"],
"description": "优先级。投诉和退款设为high，一般问题设为medium"
},
"summary": {
"type": "string",
"description": "问题摘要，包含关键信息供人工客服参考"
}
},
"required": ["title", "priority", "summary"]
}
}
},
{
"type": "function",
"function": {
"name": "direct_reply",
"description": "直接回复用户。用于闲聊、打招呼、简单问题等不需要查询外部系统的场景。",
"parameters": {
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "回复给用户的消息"
}
},
"required": ["message"]
}
}
}
]四、知识库检索集成
客服系统的知识库检索就是RAG（详见我的RAG实战文章）。这里重点讲客服场景下的特殊处理：
def search_knowledge_base(query, category=None):
"""客服知识库检索"""
# 1. 向量检索
results = vector_db.search(
query_embedding=embed(query),
top_k=10,
filter={"category": category} if category else None
)
# 2. 重排序
reranked = reranker.rank(query, [r["content"] for r in results])
# 3. 客服场景特殊处理：优先返回精确匹配的FAQ
faq_results = [r for r in reranked if r["type"] == "faq" and r["score"] > 0.85]
if faq_results:
return faq_results[0] # FAQ精确匹配，直接返回标准答案
# 4. 无精确FAQ，返回Top-3文档供大模型综合回答
return reranked[:3]FAQ优先策略
客服场景下，很多问题有标准答案（如退换货政策、营业时间）。这类问题不应该让大模型自由发挥，而是直接返回预设的标准回答。
# FAQ数据结构
{
"question": "退货流程是什么？",
"answer": "退货流程如下：\n1. 在订单页面点击「申请退货」\n2. 选择退货原因并提交\n3. 等待审核（1-2个工作日）\n4. 审核通过后，按提示寄回商品\n5. 收到退货后3个工作日内退款\n\n注意：商品需保持完好，签收后7天内可申请退货。",
"type": "faq",
"category": "退换货政策"
}五、完整的对话处理流程
from openai import OpenAI
class CustomerServiceBot:
def __init__(self):
self.client = OpenAI(api_key="xxx", base_url="https://api.deepseek.com")
self.conv_manager = ConversationManager()
self.system_prompt = """你是一个专业、友好的客服助手。
规则：
1. 回答要简洁准确，不要啰嗦
2. 不确定的信息不要编造，请调用知识库搜索
3. 涉及退款、投诉、账号安全等敏感问题，主动创建工单转人工
4. 用户情绪激动时，先安抚再解决问题
5. 不要透露你是AI，以客服身份回答"""
def handle_message(self, session_id, user_message):
"""处理用户消息的主函数"""
# 1. 添加system prompt（首次）
session = self.conv_manager.get_session(session_id)
if not session["messages"]:
self.conv_manager.add_message(session_id, "system", self.system_prompt)
# 2. 添加用户消息
self.conv_manager.add_message(session_id, "user", user_message)
# 3. 调用大模型（带Function Calling）
messages = self.conv_manager.get_messages(session_id)
response = self.client.chat.completions.create(
model="deepseek-chat",
messages=messages,
tools=CUSTOMER_SERVICE_TOOLS,
temperature=0.3 # 客服场景用低温度，保证稳定性
)
choice = response.choices[0]
# 4. 处理工具调用
if choice.message.tool_calls:
tool_call = choice.message.tool_calls[0]
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
# 执行对应的函数
tool_result = self.execute_tool(func_name, func_args)
# 把工具结果返回给模型，让模型组织最终回复
self.conv_manager.add_message(session_id, "assistant",
json.dumps({"tool_call": func_name, "args": func_args}))
self.conv_manager.add_message(session_id, "tool", json.dumps(tool_result))
# 再次调用模型生成最终回复
final_response = self.client.chat.completions.create(
model="deepseek-chat",
messages=self.conv_manager.get_messages(session_id),
temperature=0.3
)
reply = final_response.choices[0].message.content
else:
reply = choice.message.content
# 5. 记录助手回复
self.conv_manager.add_message(session_id, "assistant", reply)
return reply
def execute_tool(self, func_name, args):
"""执行工具调用"""
if func_name == "search_knowledge_base":
return search_knowledge_base(args["query"], args.get("category"))
elif func_name == "query_order":
return query_order_from_db(args.get("order_id"), args.get("phone"))
elif func_name == "create_ticket":
return create_support_ticket(args["title"], args["priority"], args["summary"])
elif func_name == "direct_reply":
return {"message": args["message"]}
return {"error": "未知工具"}六、人机协作：转人工机制
AI客服不可能解决所有问题。关键是知道什么时候该转人工。
自动转人工的触发条件
- 用户主动要求："我要找人工客服" / "转人工"
- 敏感操作：退款、投诉、账号安全（通过Function Calling的create_ticket触发）
- 多轮未解决：连续3轮以上用户表达不满或问题未解决
- 置信度低：知识库检索结果相关度低于阈值
def should_transfer_to_human(session):
"""判断是否应该转人工"""
messages = session["messages"]
metadata = session["metadata"]
# 规则1: 用户主动要求
last_msg = messages[-1]["content"] if messages else ""
transfer_keywords = ["转人工", "找人工", "真人客服", "人工服务", "找你们经理"]
if any(kw in last_msg for kw in transfer_keywords):
return True, "用户主动要求转人工"
# 规则2: 多轮未解决（连续3轮包含负面表达）
negative_keywords = ["没用", "解决不了", "还是不行", "不对", "你听不懂"]
recent = [m["content"] for m in messages[-6:] if m["role"] == "user"]
negative_count = sum(1 for m in recent if any(kw in m for kw in negative_keywords))
if negative_count >= 2:
return True, "用户多轮表达不满，建议转人工"
return False, ""转接时的信息传递
转人工时，把AI的对话摘要一起传给人工客服，避免用户重复描述问题：
def create_handoff_summary(session):
"""生成转人工摘要"""
messages = session["messages"]
summary_prompt = f"""请根据以下客服对话记录，生成一份简洁的摘要供人工客服参考。
包含：用户的核心问题、已尝试的解决方案、用户情绪状态。
对话记录：
{json.dumps(messages[-10:], ensure_ascii=False)}"""
response = llm.chat(summary_prompt)
return response七、监控与优化
关键指标
- 自动解决率：AI独立解决的对话占比（目标>70%）
- 平均对话轮数：越少说明解决效率越高
- 转人工率：过高说明AI能力不足，过低可能是该转没转
- 用户满意度：对话结束后的评分
持续优化循环
- 收集AI无法回答的问题 → 补充知识库
- 收集用户负面反馈的对话 → 优化System Prompt
- 分析高频问题 → 制作标准FAQ
- 定期评估模型效果 → 考虑是否需要微调
不要追求100%的自动解决率。客服系统的核心是用户满意度，不是AI的自主率。该转人工的时候果断转，比让AI强行回答错误答案要好得多。