5.2 智能客服系统构建

5.2 智能客服系统构建

本节导读

本节深入讲解如何基于大模型API构建智能客服系统，涵盖多轮对话管理、知识库集成、意图识别、工单流转和人机协作等核心功能，帮助读者掌握智能客服系统的设计与实现。

学习目标

• 掌握智能客服系统的整体架构设计原则
• 学会多轮对话管理机制的设计与实现
• 理解知识库集成与意图识别的关键技术
• 掌握工单流转系统的设计与优化
• 学会人机协作机制的设计与实现

核心概念

智能客服系统架构

智能客服系统是一个复杂的对话系统，需要处理自然语言理解、多轮对话管理、知识检索、工单管理等多个模块。其核心架构包括：

关键技术组件

1. 对话管理器：管理对话流程，维护对话状态
2. 意图识别：识别用户真实意图，支持多轮对话
3. 知识库检索：从知识库中检索相关答案
4. 实体识别：提取关键信息，如用户ID、订单号等
5. 回复生成：基于对话状态生成回复内容
6. 工单系统：处理复杂问题的工单流转
7. 人机协作：实现AI与人工客服的无缝切换

分步实战

步骤1：对话管理系统实现

首先构建对话管理系统，处理多轮对话：

# dialogue_manager.py
import asyncio
import time
import json
from typing import Dict, List, Optional, Any, Tuple
from dataclasses import dataclass, asdict
from enum import Enum
import uuid
import aiohttp
import logging

class DialogueState(Enum):
    """对话状态枚举"""
    INIT = "init"
    GREETING = "greeting"
    COLLECTING_INFO = "collecting_info"
    PROCESSING = "processing"
    RESOLVED = "resolved"
    ESCALATED = "escalated"
    TERMINATED = "terminated"

class Intent(Enum):
    """意图枚举"""
    QUERY = "query"
    COMPLAINT = "complaint"
    SUGGESTION = "suggestion"
    REFUND = "refund"
    TECH_SUPPORT = "tech_support"
    ORDER_STATUS = "order_status"
    PRODUCT_INFO = "product_info"
    GENERAL = "general"

@dataclass
class DialogueContext:
    """对话上下文"""
    session_id: str
    user_id: str
    state: DialogueState
    intent: Optional[Intent] = None
    entities: Dict[str, Any] = None
    messages: List[Dict[str, Any]] = None
    created_at: float = None
    updated_at: float = None
    
    def __post_init__(self):
        if self.entities is None:
            self.entities = {}
        if self.messages is None:
            self.messages = []
        if self.created_at is None:
            self.created_at = time.time()
        self.updated_at = time.time()
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return asdict(self)
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'DialogueContext':
        """从字典创建"""
        return cls(**data)

@dataclass
class Message:
    """消息类"""
    role: str  # "user" or "assistant"
    content: str
    timestamp: float
    metadata: Dict[str, Any] = None
    
    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}
        if self.timestamp is None:
            self.timestamp = time.time()

class DialogueManager:
    """对话管理器"""
    
    def __init__(self, 
                 llm_api_client,
                 knowledge_base,
                 intent_classifier,
                 entity_extractor,
                 max_dialogue_length: int = 20,
                 session_timeout: int = 1800):  # 30分钟
        
        self.llm_api_client = llm_api_client
        self.knowledge_base = knowledge_base
        self.intent_classifier = intent_classifier
        self.entity_extractor = entity_extractor
        
        self.max_dialogue_length = max_dialogue_length
        self.session_timeout = session_timeout
        
        # 对话上下文存储（实际项目中应使用Redis等存储）
        self.dialogue_sessions: Dict[str, DialogueContext] = {}
        
        # 配置日志
        self.logger = logging.getLogger(__name__)
    
    async def process_message(
        self, 
        user_id: str, 
        message_text: str,
        metadata: Dict[str, Any] = None
    ) -> Tuple[str, DialogueContext]:
        """
        处理用户消息
        
        Args:
            user_id: 用户ID
            message_text: 用户消息
            metadata: 消息元数据
        
        Returns:
            Tuple[str, DialogueContext]: 回复文本和对话上下文
        """
        # 获取或创建对话上下文
        context = await self._get_or_create_context(user_id)
        
        # 添加用户消息到对话历史
        user_message = Message(
            role="user",
            content=message_text,
            metadata=metadata or {}
        )
        context.messages.append(user_message.to_dict())
        
        try:
            # 意图识别
            intent = await self.intent_classifier.classify_intent(
                message_text, 
                context.to_dict()
            )
            context.intent = intent
            
            # 实体提取
            entities = await self.entity_extractor.extract_entities(
                message_text, 
                context.to_dict()
            )
            context.entities.update(entities)
            
            # 根据意图和状态处理对话
            response_text = await self._handle_dialogue(context)
            
            # 添加助手回复到对话历史
            assistant_message = Message(
                role="assistant",
                content=response_text,
                metadata={"handled_by": "ai", "intent": intent.value}
            )
            context.messages.append(assistant_message.to_dict())
            
            # 更新对话状态
            await self._update_context(context)
            
            return response_text, context
            
        except Exception as e:
            self.logger.error(f"处理消息失败: {e}")
            context.state = DialogueState.TERMINATED
            await self._update_context(context)
            raise
    
    async def _get_or_create_context(self, user_id: str) -> DialogueContext:
        """获取或创建对话上下文"""
        session_id = f"user_{user_id}"
        
        if session_id in self.dialogue_sessions:
            context = self.dialogue_sessions[session_id]
            # 检查是否超时
            if time.time() - context.updated_at > self.session_timeout:
                # 清除过期会话
                del self.dialogue_sessions[session_id]
                context = DialogueContext(
                    session_id=session_id,
                    user_id=user_id,
                    state=DialogueState.INIT
                )
            else:
                # 重置为活跃状态
                context.updated_at = time.time()
        else:
            context = DialogueContext(
                session_id=session_id,
                user_id=user_id,
                state=DialogueState.INIT
            )
        
        self.dialogue_sessions[session_id] = context
        return context
    
    async def _handle_dialogue(self, context: DialogueContext) -> str:
        """根据对话状态处理对话"""
        handlers = {
            DialogueState.INIT: self._handle_init,
            DialogueState.GREETING: self._handle_greeting,
            DialogueState.COLLECTING_INFO: self._handle_collecting_info,
            DialogueState.PROCESSING: self._handle_processing,
            DialogueState.RESOLVED: self._handle_resolved,
            DialogueState.ESCALATED: self._handle_escalated,
            DialogueState.TERMINATED: self._handle_terminated
        }
        
        handler = handlers.get(context.state, self._handle_terminated)
        return await handler(context)
    
    async def _handle_init(self, context: DialogueContext) -> str:
        """处理初始状态"""
        context.state = DialogueState.GREETING
        return "您好！我是智能客服助手，请问有什么可以帮助您的吗？"
    
    async def _handle_greeting(self, context: DialogueContext) -> str:
        """处理问候状态"""
        if context.intent == Intent.GENERAL:
            context.state = DialogueState.PROCESSING
            return "我理解您想咨询一般问题。请告诉我具体想了解什么？"
        elif context.intent == Intent.QUERY:
            context.state = DialogueState.COLLECTING_INFO
            return "请问您想查询什么信息呢？"
        else:
            context.state = DialogueState.COLLECTING_INFO
            return "请告诉我您的需求，我会尽力为您提供帮助。"
    
    async def _handle_collecting_info(self, context: DialogueContext) -> str:
        """收集信息状态处理"""
        # 根据意图决定如何收集信息
        if context.intent == Intent.REFUND:
            return "请问您的订单号是多少？这样我可以帮您查询退款信息。"
        elif context.intent == Intent.TECH_SUPPORT:
            if "order_number" not in context.entities:
                return "请提供您的订单号，以便我为您查询技术支持信息。"
            else:
                context.state = DialogueState.PROCESSING
                return "正在为您查询技术支持信息，请稍候..."
        elif context.intent == Intent.ORDER_STATUS:
            if "order_number" not in context.entities:
                return "请提供您的订单号，我来帮您查询订单状态。"
            else:
                context.state = DialogueState.PROCESSING
                return "正在查询订单状态..."
        else:
            context.state = DialogueState.PROCESSING
            return "正在处理您的问题..."
    
    async def _handle_processing(self, context: DialogueContext) -> str:
        """处理状态"""
        # 根据意图进行不同的处理
        if context.intent == Intent.QUERY:
            return await self._handle_query(context)
        elif context.intent == Intent.COMPLAINT:
            return await self._handle_complaint(context)
        elif context.intent == Intent.REFUND:
            return await self._handle_refund(context)
        elif context.intent == Intent.TECH_SUPPORT:
            return await self._handle_tech_support(context)
        elif context.intent == Intent.ORDER_STATUS:
            return await self._handle_order_status(context)
        else:
            return await self._handle_general_query(context)
    
    async def _handle_query(self, context: DialogueContext) -> str:
        """处理查询请求"""
        # 从知识库检索相关信息
        search_results = await self.knowledge_base.search(
            context.messages[-1]["content"],
            context.entities
        )
        
        if search_results:
            # 使用大模型生成回复
            prompt = f"""
            基于以下知识库信息，回答用户的问题：
            
            知识库信息：
            {search_results}
            
            用户问题：{context.messages[-1]["content"]}
            
            请提供准确、有用的回答。
            """
            
            response = await self.llm_api_client.chat_completion([
                {"role": "system", "content": "你是一个专业的客服助手，基于提供的知识库信息回答问题。"},
                {"role": "user", "content": prompt}
            ])
            
            return response.choices[0]["message"]["content"]
        else:
            return "抱歉，我没有找到相关的信息。您可以尝试重新描述问题，或者告诉我您的具体需求。"
    
    async def _handle_complaint(self, context: DialogueContext) -> str:
        """处理投诉"""
        if "order_number" not in context.entities:
            context.state = DialogueState.COLLECTING_INFO
            return "为了更好地处理您的投诉，请提供您的订单号。"
        
        # 检查是否有必要升级到人工客服
        if len(context.messages) > 3 and context.state == DialogueState.PROCESSING:
            context.state = DialogueState.ESCALATED
            return "我理解您的问题比较复杂，已为您转接人工客服，请稍候..."
        
        return "我已记录您的投诉，我们会尽快处理。请问还有其他需要帮助的吗？"
    
    async def _handle_refund(self, context: DialogueContext) -> str:
        """处理退款请求"""
        order_number = context.entities.get("order_number")
        
        if not order_number:
            return "请提供您的订单号，这样我可以帮您查询退款信息。"
        
        # 模拟查询订单信息
        order_info = await self._query_order_info(order_number)
        
        if not order_info:
            return "没有找到该订单号，请确认订单号是否正确。"
        
        if order_info.get("status") == "cancelled":
            refund_amount = order_info.get("total_amount", 0)
            return f"您的订单已取消，退款金额 {refund_amount} 元将在3-7个工作日内退回到您的支付账户。"
        elif order_info.get("status") == "refunded":
            return "您的订单已退款完成。"
        else:
            context.state = DialogueState.ESCALATED
            return "您的退款申请需要人工处理，已为您转接人工客服，请稍候..."
    
    async def _handle_tech_support(self, context: DialogueContext) -> str:
        """处理技术支持"""
        order_number = context.entities.get("order_number")
        
        if not order_number:
            return "请提供您的订单号，以便我为您查询技术支持信息。"
        
        # 模拟查询技术支持信息
        tech_info = await self._query_tech_support(order_number)
        
        if tech_info:
            return f"技术支持信息：{tech_info}"
        else:
            context.state = DialogueState.ESCALATED
            return "您的问题需要专业的技术人员处理，已为您转接人工客服，请稍候..."
    
    async def _handle_order_status(self, context: DialogueContext) -> str:
        """处理订单状态查询"""
        order_number = context.entities.get("order_number")
        
        if not order_number:
            return "请提供您的订单号，我来帮您查询订单状态。"
        
        # 模拟查询订单状态
        order_status = await self._query_order_status(order_number)
        
        if order_status:
            return f"订单 {order_number} 的状态：{order_status}"
        else:
            return "没有找到该订单号，请确认订单号是否正确。"
    
    async def _handle_general_query(self, context: DialogueContext) -> str:
        """处理一般查询"""
        # 从知识库检索
        search_results = await self.knowledge_base.search(
            context.messages[-1]["content"],
            context.entities
        )
        
        if search_results:
            prompt = f"""
            基于以下信息回答用户的问题：
            
            {search_results}
            
            用户问题：{context.messages[-1]["content"]}
            """
            
            response = await self.llm_api_client.chat_completion([
                {"role": "system", "content": "你是一个专业的客服助手。"},
                {"role": "user", "content": prompt}
            ])
            
            return response.choices[0]["message"]["content"]
        else:
            return "抱歉，我还没有学习到这方面的知识。您可以尝试重新描述问题，或者联系人工客服获取更专业的帮助。"
    
    async def _handle_resolved(self, context: DialogueContext) -> str:
        """处理已解决状态"""
        return "很高兴能帮助到您！如果您还有其他问题，随时可以继续咨询。"
    
    async def _handle_escalated(self, context: DialogueContext) -> str:
        """处理升级状态"""
        return "正在为您转接人工客服，请稍候片刻..."
    
    async def _handle_terminated(self, context: DialogueContext) -> str:
        """处理终止状态"""
        return "对话已结束，感谢您的使用。"
    
    async def _update_context(self, context: DialogueContext):
        """更新对话上下文"""
        context.updated_at = time.time()
        
        # 检查是否需要终止对话
        if len(context.messages) > self.max_dialogue_length:
            context.state = DialogueState.TERMINATED
        
        # 检查对话是否已解决
        if context.state in [DialogueState.RESOLVED, DialogueState.ESCALATED]:
            # 设置过期时间
            context.updated_at = time.time() + 300  # 5分钟后过期
        
        # 清理过期会话
        await self._cleanup_sessions()
    
    async def _cleanup_sessions(self):
        """清理过期会话"""
        current_time = time.time()
        expired_sessions = []
        
        for session_id, context in self.dialogue_sessions.items():
            if current_time - context.updated_at > self.session_timeout:
                expired_sessions.append(session_id)
        
        for session_id in expired_sessions:
            del self.dialogue_sessions[session_id]
            self.logger.info(f"清理过期会话: {session_id}")
    
    async def get_session_context(self, user_id: str) -> Optional[DialogueContext]:
        """获取用户会话上下文"""
        session_id = f"user_{user_id}"
        return self.dialogue_sessions.get(session_id)
    
    async def end_session(self, user_id: str):
        """结束用户会话"""
        session_id = f"user_{user_id}"
        if session_id in self.dialogue_sessions:
            del self.dialogue_sessions[session_id]
            self.logger.info(f"会话已结束: {session_id}")

# 使用示例
async def dialogue_manager_example():
    """对话管理器使用示例"""
    
    # 模拟LLM客户端
    class MockLLMClient:
        async def chat_completion(self, messages):
            return type('Response', (), {
​