智能客服 Agent

┌─────────────────────────────────────────────────────────────┐
│                    智能客服 Agent 能力矩阵                   │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                 对话理解能力                         │  │
│   │  ┌──────────┐ ┌──────────┐ ┌──────────┐            │  │
│   │  │ 意图识别 │ │ 槽位填充 │ │ 情感分析 │            │  │
│   │  │          │ │          │ │          │            │  │
│   │  └──────────┘ └──────────┘ └──────────┘            │  │
│   │  ┌──────────┐ ┌──────────┐ ┌──────────┐            │  │
│   │  │ 多轮对话 │ │ 上下文   │ │ 用户画像 │            │  │
│   │  │ 管理     │ │ 理解     │ │ 构建     │            │  │
│   │  └──────────┘ └──────────┘ └──────────┘            │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                 业务处理能力                         │  │
│   │  ┌──────────┐ ┌──────────┐ ┌──────────┐            │  │
│   │  │ 知识问答 │ │ 工单创建 │ │ 订单查询 │            │  │
│   │  │          │ │          │ │          │            │  │
│   │  └──────────┘ └──────────┘ └──────────┘            │  │
│   │  ┌──────────┐ ┌──────────┐ ┌──────────┐            │  │
│   │  │ 退款处理 │ │ 投诉处理 │ │ 建议收集 │            │  │
│   │  │          │ │          │ │          │            │  │
│   │  └──────────┘ └──────────┘ └──────────┘            │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                 协作能力                             │  │
│   │  ┌──────────┐ ┌──────────┐ ┌──────────┐            │  │
│   │  │ 人工转接 │ │ 系统对接 │ │ 知识更新 │            │  │
│   │  │          │ │          │ │          │            │  │
│   │  └──────────┘ └──────────┘ └──────────┘            │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                    智能客服 Agent 架构                       │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                    接入层                           │  │
│   │  ┌──────────┐ ┌──────────┐ ┌──────────┐            │  │
│   │  │ Web Chat │ │ 微信     │ │ APP SDK  │            │  │
│   │  │          │ │ 公众号   │ │          │            │  │
│   │  └──────────┘ └──────────┘ └──────────┘            │  │
│   └─────────────────────────────────────────────────────┘  │
│                          │                                  │
│                          ↓                                  │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                    对话管理层                        │  │
│   │  ┌───────────────────────────────────────────────┐  │  │
│   │  │              Session Manager                   │  │  │
│   │  │  会话创建、状态管理、历史记录                 │  │  │
│   │  └───────────────────────────────────────────────┘  │  │
│   │                         │                            │  │
│   │  ┌───────────────────────────────────────────────┐  │  │
│   │  │              Dialogue Manager                  │  │  │
│   │  │  意图识别、槽位填充、策略决策                 │  │  │
│   │  └───────────────────────────────────────────────┘  │  │
│   └─────────────────────────────────────────────────────┘  │
│                          │                                  │
│                          ↓                                  │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                    业务能力层                        │  │
│   │  ┌──────────┐ ┌──────────┐ ┌──────────┐            │  │
│   │  │ 知识库   │ │ 工单系统 │ │ CRM 系统 │            │  │
│   │  │ 问答     │ │ 对接     │ │ 对接     │            │  │
│   │  └──────────┘ └──────────┘ └──────────┘            │  │
│   │  ┌──────────┐ ┌──────────┐ ┌──────────┐            │  │
│   │  │ 订单系统 │ │ 支付系统 │ │ 人工坐席 │            │  │
│   │  │ 对接     │ │ 对接     │ │ 转接     │            │  │
│   │  └──────────┘ └──────────┘ └──────────┘            │  │
│   └─────────────────────────────────────────────────────┘  │
│                          │                                  │
│                          ↓                                  │
│   ┌─────────────────────────────────────────────────────┐  │
│   │                    知识层                           │  │
│   │  ┌──────────┐ ┌──────────┐ ┌──────────┐            │  │
│   │  │ FAQ 库   │ │ 产品文档 │ │ 对话样本 │            │  │
│   │  │          │ │          │ │ 库       │            │  │
│   │  └──────────┘ └──────────┘ └──────────┘            │  │
│   │  ┌──────────┐ ┌──────────┐                         │  │
│   │  │ 向量索引 │ │ 用户画像 │                         │  │
│   │  │          │ │ 数据     │                         │  │
│   │  └──────────┘ └──────────┘                         │  │
│   └─────────────────────────────────────────────────────┘  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

"""
智能客服 Agent 实现
"""
 
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
 
 
class IntentType(Enum):
    """意图类型"""
    FAQ = "faq"                    # 常见问题
    ORDER_QUERY = "order_query"    # 订单查询
    REFUND = "refund"              # 退款
    COMPLAINT = "complaint"        # 投诉
    TRANSFER = "transfer"          # 转人工
    CHITCHAT = "chitchat"          # 闲聊
    UNKNOWN = "unknown"            # 未知
 
 
@dataclass
class Slot:
    """槽位"""
    name: str
    value: Any = None
    required: bool = False
    prompt: str = ""
 
 
@dataclass
class DialogueState:
    """对话状态"""
    session_id: str
    intent: Optional[IntentType] = None
    slots: Dict[str, Slot] = field(default_factory=dict)
    history: List[Dict] = field(default_factory=list)
    turn_count: int = 0
    status: str = "ongoing"
 
 
class CustomerServiceAgent:
    """智能客服 Agent"""
    
    def __init__(
        self,
        llm=None,
        knowledge_base=None,
        crm_client=None,
        order_client=None
    ):
        self.llm = llm
        self.knowledge_base = knowledge_base
        self.crm_client = crm_client
        self.order_client = order_client
        self.sessions: Dict[str, DialogueState] = {}
    
    async def chat(
        self,
        session_id: str,
        user_message: str,
        user_id: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        处理用户消息
        
        Args:
            session_id: 会话 ID
            user_message: 用户消息
            user_id: 用户 ID
            
        Returns:
            回复结果
        """
        # 1. 获取或创建会话状态
        state = self._get_or_create_session(session_id)
        
        # 2. 更新历史
        state.history.append({
            "role": "user",
            "content": user_message,
            "timestamp": datetime.now().isoformat()
        })
        state.turn_count += 1
        
        # 3. 意图识别
        if state.intent is None:
            state.intent = await self._classify_intent(user_message, state.history)
        
        # 4. 槽位填充
        state = await self._fill_slots(state, user_message)
        
        # 5. 检查是否需要更多信息
        missing_slot = self._check_missing_slots(state)
        if missing_slot:
            response = self._generate_slot_prompt(missing_slot)
            state.history.append({
                "role": "assistant",
                "content": response,
                "timestamp": datetime.now().isoformat()
            })
            return {
                "response": response,
                "intent": state.intent.value if state.intent else None,
                "status": "need_info",
                "required_slot": missing_slot.name
            }
        
        # 6. 执行业务逻辑
        response = await self._execute_intent(state, user_id)
        
        # 7. 更新历史
        state.history.append({
            "role": "assistant",
            "content": response,
            "timestamp": datetime.now().isoformat()
        })
        
        return {
            "response": response,
            "intent": state.intent.value if state.intent else None,
            "status": "completed"
        }
    
    def _get_or_create_session(self, session_id: str) -> DialogueState:
        """获取或创建会话"""
        if session_id not in self.sessions:
            self.sessions[session_id] = DialogueState(
                session_id=session_id,
                slots=self._init_slots()
            )
        return self.sessions[session_id]
    
    def _init_slots(self) -> Dict[str, Slot]:
        """初始化槽位"""
        return {
            "order_id": Slot(
                name="order_id",
                required=False,
                prompt="请问您的订单号是多少？"
            ),
            "product_name": Slot(
                name="product_name",
                required=False,
                prompt="请问是哪个产品？"
            ),
            "refund_reason": Slot(
                name="refund_reason",
                required=False,
                prompt="请问退款的原因是什么？"
            )
        }
    
    async def _classify_intent(
        self,
        message: str,
        history: List[Dict]
    ) -> IntentType:
        """意图识别"""
        prompt = f"""
请分析用户意图，从以下选项中选择一个：
 
意图类别：
- faq: 常见问题咨询
- order_query: 订单查询
- refund: 退款请求
- complaint: 投诉
- transfer: 要求转人工
- chitchat: 闲聊
 
用户消息：{message}
 
对话历史：
{self._format_history(history)}
 
请只输出意图类别，不要输出其他内容。
"""
        
        response = await self.llm.ainvoke(prompt)
        intent_str = response.content.strip().lower()
        
        try:
            return IntentType(intent_str)
        except ValueError:
            return IntentType.UNKNOWN
    
    async def _fill_slots(
        self,
        state: DialogueState,
        message: str
    ) -> DialogueState:
        """槽位填充"""
        # 根据意图确定需要填充的槽位
        required_slots = self._get_required_slots(state.intent)
        
        for slot_name in required_slots:
            if slot_name in state.slots and state.slots[slot_name].value is None:
                # 尝试从消息中提取槽位值
                value = await self._extract_slot_value(
                    message, 
                    slot_name, 
                    state.intent
                )
                if value:
                    state.slots[slot_name].value = value
        
        return state
    
    def _get_required_slots(self, intent: Optional[IntentType]) -> List[str]:
        """获取意图所需的槽位"""
        slot_mapping = {
            IntentType.ORDER_QUERY: ["order_id"],
            IntentType.REFUND: ["order_id", "refund_reason"],
            IntentType.COMPLAINT: ["order_id"],
        }
        return slot_mapping.get(intent, [])
    
    async def _extract_slot_value(
        self,
        message: str,
        slot_name: str,
        intent: IntentType
    ) -> Optional[str]:
        """提取槽位值"""
        prompt = f"""
从用户消息中提取 {slot_name} 的值。
 
用户消息：{message}
意图：{intent.value}
槽位名：{slot_name}
 
如果无法提取，请输出 "None"。
只输出槽位值，不要输出其他内容。
"""
        
        response = await self.llm.ainvoke(prompt)
        value = response.content.strip()
        
        if value.lower() == "none":
            return None
        return value
    
    def _check_missing_slots(self, state: DialogueState) -> Optional[Slot]:
        """检查缺失的必要槽位"""
        required_slots = self._get_required_slots(state.intent)
        
        for slot_name in required_slots:
            slot = state.slots.get(slot_name)
            if slot and slot.value is None:
                return slot
        
        return None
    
    def _generate_slot_prompt(self, slot: Slot) -> str:
        """生成槽位提示"""
        return slot.prompt
    
    async def _execute_intent(
        self,
        state: DialogueState,
        user_id: Optional[str] = None
    ) -> str:
        """执行意图"""
        if state.intent == IntentType.FAQ:
            return await self._handle_faq(state)
        elif state.intent == IntentType.ORDER_QUERY:
            return await self._handle_order_query(state)
        elif state.intent == IntentType.REFUND:
            return await self._handle_refund(state)
        elif state.intent == IntentType.COMPLAINT:
            return await self._handle_complaint(state)
        elif state.intent == IntentType.TRANSFER:
            return self._handle_transfer(state)
        elif state.intent == IntentType.CHITCHAT:
            return await self._handle_chitchat(state)
        else:
            return "抱歉，我不太理解您的问题。请问您想咨询什么？"
    
    async def _handle_faq(self, state: DialogueState) -> str:
        """处理 FAQ"""
        last_message = state.history[-1]["content"] if state.history else ""
        
        if self.knowledge_base:
            answer = await self.knowledge_base.query(last_message)
            return answer
        
        return "让我为您查询相关信息..."
    
    async def _handle_order_query(self, state: DialogueState) -> str:
        """处理订单查询"""
        order_id = state.slots.get("order_id", {}).get("value")
        
        if self.order_client and order_id:
            order_info = await self.order_client.get_order(order_id)
            return f"您的订单 {order_id} 状态：{order_info.get('status', '未知')}"
        
        return "请提供您的订单号，我来为您查询。"
    
    async def _handle_refund(self, state: DialogueState) -> str:
        """处理退款"""
        order_id = state.slots.get("order_id", {}).get("value")
        reason = state.slots.get("refund_reason", {}).get("value")
        
        # 创建退款工单
        return f"已为您创建退款申请（订单号：{order_id}），原因：{reason}。我们会在1-3个工作日内处理。"
    
    async def _handle_complaint(self, state: DialogueState) -> str:
        """处理投诉"""
        return "非常抱歉给您带来不好的体验。我已记录您的反馈，客服专员会在24小时内联系您处理。"
    
    def _handle_transfer(self, state: DialogueState) -> str:
        """处理转人工"""
        return "正在为您转接人工客服，请稍候..."
    
    async def _handle_chitchat(self, state: DialogueState) -> str:
        """处理闲聊"""
        last_message = state.history[-1]["content"] if state.history else ""
        response = await self.llm.ainvoke(f"用户说：{last_message}\n请友好回复：")
        return response.content
    
    def _format_history(self, history: List[Dict]) -> str:
        """格式化历史"""
        return "\n".join([
            f"{h['role']}: {h['content']}" 
            for h in history[-5:]  # 最近5轮
        ])
 
 
# ========== 知识库查询 ==========
 
class KnowledgeBase:
    """知识库"""
    
    def __init__(self, vector_store=None, llm=None):
        self.vector_store = vector_store
        self.llm = llm
    
    async def query(self, question: str) -> str:
        """查询知识库"""
        if not self.vector_store:
            return "知识库暂未配置。"
        
        # 检索相关文档
        docs = self.vector_store.similarity_search(question, k=3)
        
        if not docs:
            return "抱歉，没有找到相关信息。"
        
        # 使用 LLM 生成回答
        context = "\n\n".join([doc.page_content for doc in docs])
        
        prompt = f"""
根据以下知识回答用户问题：
 
知识内容：
{context}
 
用户问题：{question}
 
请给出准确、有帮助的回答。
"""
        
        response = await self.llm.ainvoke(prompt)
        return response.content

"""
意图识别模块
"""
 
from typing import List, Dict
import re
 
 
class IntentClassifier:
    """意图分类器"""
    
    def __init__(self, llm=None):
        self.llm = llm
        
        # 关键词规则（快速匹配）
        self.keyword_rules = {
            IntentType.ORDER_QUERY: [
                r"订单", r"查询订单", r"我的订单", r"物流", r"快递"
            ],
            IntentType.REFUND: [
                r"退款", r"退货", r"不想要了", r"退换"
            ],
            IntentType.COMPLAINT: [
                r"投诉", r"差评", r"不满意", r"质量差", r"投诉"
            ],
            IntentType.TRANSFER: [
                r"人工", r"转人工", r"真人", r"客服人员"
            ]
        }
    
    async def classify(
        self, 
        message: str,
        use_llm: bool = True
    ) -> IntentType:
        """
        分类意图
        
        Args:
            message: 用户消息
            use_llm: 是否使用 LLM
            
        Returns:
            意图类型
        """
        # 1. 先用关键词规则快速匹配
        for intent, patterns in self.keyword_rules.items():
            for pattern in patterns:
                if re.search(pattern, message):
                    return intent
        
        # 2. 使用 LLM 分类
        if use_llm and self.llm:
            return await self._llm_classify(message)
        
        return IntentType.UNKNOWN
    
    async def _llm_classify(self, message: str) -> IntentType:
        """使用 LLM 分类"""
        prompt = f"""
请分析用户意图。
 
用户消息：{message}
 
意图选项：
1. faq - 咨询问题
2. order_query - 查订单
3. refund - 退款
4. complaint - 投诉
5. transfer - 转人工
6. chitchat - 闲聊
 
只输出意图标签。
"""
        
        response = await self.llm.ainvoke(prompt)
        intent_str = response.content.strip().lower()
        
        try:
            return IntentType(intent_str)
        except ValueError:
            return IntentType.UNKNOWN

"""
多轮对话管理
"""
 
class DialogueManager:
    """对话管理器"""
    
    def __init__(self, max_turns: int = 20):
        self.max_turns = max_turns
    
    def should_transfer(self, state: DialogueState) -> bool:
        """判断是否应该转人工"""
        # 1. 用户明确要求转人工
        if state.intent == IntentType.TRANSFER:
            return True
        
        # 2. 轮数过多
        if state.turn_count >= self.max_turns:
            return True
        
        # 3. 意图识别多次失败
        unknown_count = sum(
            1 for h in state.history 
            if h.get("intent") == IntentType.UNKNOWN
        )
        if unknown_count >= 3:
            return True
        
        return False
    
    def get_context_summary(self, state: DialogueState) -> str:
        """获取上下文摘要"""
        # 压缩历史对话
        if len(state.history) <= 5:
            return self._format_history(state.history)
        
        # 只保留最近的对话
        recent = state.history[-5:]
        
        # 对早期对话生成摘要
        early = state.history[:-5]
        early_summary = self._summarize_early_dialogue(early)
        
        return f"[历史摘要] {early_summary}\n\n[最近对话]\n{self._format_history(recent)}"
    
    def _format_history(self, history: List[Dict]) -> str:
        """格式化历史"""
        return "\n".join([
            f"{h['role']}: {h['content']}" 
            for h in history
        ])
    
    def _summarize_early_dialogue(self, history: List[Dict]) -> str:
        """摘要早期对话"""
        # 简化：只提取关键信息
        intents = [h.get("intent") for h in history if h.get("intent")]
        if intents:
            return f"用户之前咨询了：{', '.join(set(intents))}"
        return "用户进行了多轮对话"