AI 客服系统构建指南：从知识库到多轮对话

from enum import Enum
from dataclasses import dataclass

class QueryType(Enum):
    """客服查询类型"""
    FAQ = "faq"                # 常见问题（退货政策、运费说明）
    TROUBLESHOOT = "trouble"   # 故障排查（设备不工作、登录失败）
    TRANSACTION = "transaction" # 交易查询（订单状态、退款进度）
    COMPLAINT = "complaint"    # 投诉建议
    CHITCHAT = "chitchat"      # 闲聊（超出业务范围）

@dataclass
class QueryStrategy:
    """不同查询类型的处理策略"""
    query_type: QueryType
    needs_rag: bool
    needs_tools: bool
    max_turns: int
    can_self_serve: bool
    escalation_threshold: float  # 置信度低于此值则转人工

STRATEGIES = {
    QueryType.FAQ: QueryStrategy(
        query_type=QueryType.FAQ,
        needs_rag=True, needs_tools=False,
        max_turns=3, can_self_serve=True,
        escalation_threshold=0.6
    ),
    QueryType.TROUBLESHOOT: QueryStrategy(
        query_type=QueryType.TROUBLESHOOT,
        needs_rag=True, needs_tools=True,
        max_turns=8, can_self_serve=True,
        escalation_threshold=0.5
    ),
    QueryType.TRANSACTION: QueryStrategy(
        query_type=QueryType.TRANSACTION,
        needs_rag=False, needs_tools=True,
        max_turns=5, can_self_serve=True,
        escalation_threshold=0.7
    ),
    QueryType.COMPLAINT: QueryStrategy(
        query_type=QueryType.COMPLAINT,
        needs_rag=False, needs_tools=False,
        max_turns=2, can_self_serve=False,
        escalation_threshold=0.8
    ),
    QueryType.CHITCHAT: QueryStrategy(
        query_type=QueryType.CHITCHAT,
        needs_rag=False, needs_tools=False,
        max_turns=2, can_self_serve=True,
        escalation_threshold=0.9
    ),
}

from dataclasses import dataclass, field
from typing import Optional
import hashlib

@dataclass
class KnowledgeChunk:
    """知识库文档块"""
    chunk_id: str
    content: str
    source: str           # 来源文档
    source_type: str      # manual, faq, product_doc, ticket
    category: str         # 业务分类
    metadata: dict = field(default_factory=dict)
    embedding: list[float] = field(default_factory=list)
    
    def __post_init__(self):
        if not self.chunk_id:
            self.chunk_id = hashlib.md5(self.content.encode()).hexdigest()[:16]


class KnowledgeBaseBuilder:
    """知识库构建器"""
    
    def __init__(self, embedder, vector_store):
        self.embedder = embedder
        self.vector_store = vector_store
    
    async def ingest_document(
        self,
        content: str,
        source: str,
        source_type: str,
        category: str,
        chunk_size: int = 500,
        chunk_overlap: int = 50
    ) -> list[KnowledgeChunk]:
        """文档分块 + 向量化 + 入库"""
        # 1. 智能分块（基于语义边界）
        chunks_text = self._semantic_chunk(content, chunk_size, chunk_overlap)
        
        # 2. 批量生成嵌入
        embeddings = await self.embedder.embed_batch(chunks_text)
        
        # 3. 构建知识块
        chunks = [
            KnowledgeChunk(
                chunk_id="",
                content=text,
                source=source,
                source_type=source_type,
                category=category,
                metadata={"chunk_index": i, "total_chunks": len(chunks_text)},
                embedding=emb
            )
            for i, (text, emb) in enumerate(zip(chunks_text, embeddings))
        ]
        
        # 4. 存入向量数据库
        await self.vector_store.upsert(chunks)
        return chunks
    
    def _semantic_chunk(self, text: str, size: int, overlap: int) -> list[str]:
        """基于语义边界的分块"""
        # 优先在段落边界分块，其次在句号处
        import re
        
        # 按段落分割
        paragraphs = text.split('\n\n')
        chunks = []
        current = ""
        
        for para in paragraphs:
            if len(current) + len(para) <= size:
                current += para + "\n\n"
            else:
                if current:
                    chunks.append(current.strip())
                # 如果单段落超过 size，按句子分割
                if len(para) > size:
                    sentences = re.split(r'(?<=[。！？.!?])\s*', para)
                    current = ""
                    for sent in sentences:
                        if len(current) + len(sent) <= size:
                            current += sent
                        else:
                            if current:
                                chunks.append(current.strip())
                            # 保留 overlap
                            current = chunks[-1][-overlap:] + sent if chunks else sent
                else:
                    current = para + "\n\n"
        
        if current.strip():
            chunks.append(current.strip())
        
        return chunks

class KnowledgeQualityChecker:
    """知识库质量检查"""
    
    @staticmethod
    def check_coverage(kb_stats: dict) -> dict:
        """检查知识覆盖度"""
        issues = []
        
        # 检查每个分类的文档数
        for category, count in kb_stats["categories"].items():
            if count < 10:
                issues.append(f"分类 '{category}' 文档数过少: {count}")
        
        # 检查是否有孤立知识（没有引用的文档）
        if kb_stats["orphan_docs"] > 0:
            issues.append(f"{kb_stats['orphan_docs']} 个文档从未被检索命中")
        
        # 检查过期知识
        if kb_stats["stale_docs"] > 0:
            issues.append(f"{kb_stats['stale_docs']} 个文档超过 6 个月未更新")
        
        return {
            "healthy": len(issues) == 0,
            "issues": issues,
            "total_docs": kb_stats["total"],
            "avg_chunk_size": kb_stats["avg_chunk_size"],
        }

from dataclasses import dataclass
from typing import Optional
import json

@dataclass
class Intent:
    """识别出的意图"""
    type: QueryType
    confidence: float
    entities: dict  # 提取的实体
    raw_text: str

class IntentClassifier:
    """意图识别器：LLM + 规则混合"""
    
    RULE_PATTERNS = {
        QueryType.TRANSACTION: ["订单", "退款", "物流", "发货", "进度", "查询"],
        QueryType.COMPLAINT: ["投诉", "差评", "态度", "不满", "经理", "曝光"],
        QueryType.TROUBLESHOOT: ["无法", "失败", "错误", "故障", "不工作", "崩溃"],
    }
    
    async def classify(self, user_message: str, context: list[dict] = None) -> Intent:
        """混合策略：先规则快速匹配，再 LLM 精确分类"""
        
        # 1. 规则快速匹配
        rule_intent = self._rule_based_classify(user_message)
        if rule_intent and rule_intent.confidence > 0.85:
            return rule_intent
        
        # 2. LLM 精确分类
        llm_intent = await self._llm_classify(user_message, context)
        return llm_intent
    
    def _rule_based_classify(self, message: str) -> Optional[Intent]:
        """基于关键词的快速分类"""
        for intent_type, keywords in self.RULE_PATTERNS.items():
            matches = [kw for kw in keywords if kw in message]
            if matches:
                confidence = min(0.5 + 0.15 * len(matches), 0.95)
                return Intent(
                    type=intent_type,
                    confidence=confidence,
                    entities={"matched_keywords": matches},
                    raw_text=message
                )
        return None
    
    async def _llm_classify(self, message: str, context: list[dict]) -> Intent:
        """使用 LLM 进行意图分类"""
        prompt = f"""分析用户消息的意图，返回 JSON。

意图类型：
- faq: 常见问题咨询
- trouble: 故障排查请求
- transaction: 交易/订单查询
- complaint: 投诉建议
- chitchat: 闲聊

用户消息: {message}

返回格式:
{{"intent": "类型", "confidence": 0.0-1.0, "entities": {{}}}}
"""
        # 调用 LLM
        response = await self._call_llm(prompt)
        result = json.loads(response)
        
        return Intent(
            type=QueryType(result["intent"]),
            confidence=result["confidence"],
            entities=result.get("entities", {}),
            raw_text=message
        )
    
    async def _call_llm(self, prompt: str) -> str:
        """调用 LLM API"""
        pass

from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Optional, Any

class DialogState(Enum):
    """对话状态"""
    GREETING = auto()        # 初始问候
    INTENT_COLLECTING = auto()  # 收集意图信息
    INFO_GATHERING = auto()  # 收集必要信息（订单号等）
    PROCESSING = auto()      # 处理中
    RESPONDING = auto()      # 回复中
    CONFIRMING = auto()      # 确认解决
    ESCALATING = auto()      # 转人工
    CLOSING = auto()         # 结束

@dataclass
class ConversationContext:
    """对话上下文"""
    session_id: str
    state: DialogState = DialogState.GREETING
    intent: Optional[Intent] = None
    collected_info: dict = field(default_factory=dict)
    required_fields: list[str] = field(default_factory=list)
    turn_count: int = 0
    max_turns: int = 10
    history: list[dict] = field(default_factory=dict)
    satisfaction_score: Optional[float] = None

class DialogManager:
    """多轮对话管理器"""
    
    # 意图对应需要收集的信息
    REQUIRED_FIELDS = {
        QueryType.TRANSACTION: ["order_id"],
        QueryType.TROUBLESHOOT: ["device_model", "issue_description"],
        QueryType.FAQ: [],
        QueryType.COMPLAINT: ["issue_description"],
    }
    
    async def handle_message(
        self,
        ctx: ConversationContext,
        user_message: str
    ) -> tuple[str, ConversationContext]:
        """处理用户消息，返回回复和更新后的上下文"""
        ctx.turn_count += 1
        
        # 状态机驱动对话流程
        if ctx.state == DialogState.GREETING:
            return await self._handle_greeting(ctx, user_message)
        elif ctx.state == DialogState.INTENT_COLLECTING:
            return await self._handle_intent(ctx, user_message)
        elif ctx.state == DialogState.INFO_GATHERING:
            return await self._handle_info(ctx, user_message)
        elif ctx.state == DialogState.PROCESSING:
            return await self._handle_processing(ctx, user_message)
        elif ctx.state == DialogState.CONFIRMING:
            return await self._handle_confirm(ctx, user_message)
        else:
            return await self._handle_fallback(ctx, user_message)
    
    async def _handle_greeting(self, ctx: ConversationContext, msg: str) -> tuple[str, ConversationContext]:
        """初始状态：识别意图"""
        intent = await self.classifier.classify(msg, ctx.history)
        ctx.intent = intent
        ctx.history.append({"role": "user", "content": msg})
        
        # 检查是否需要转人工
        strategy = STRATEGIES.get(intent.type)
        if strategy and intent.confidence < strategy.escalation_threshold:
            ctx.state = DialogState.ESCALATING
            reply = "您好，这个问题我需要转接给人工客服为您处理，请稍等。"
            return reply, ctx
        
        # 设置需要收集的信息
        ctx.required_fields = self.REQUIRED_FIELDS.get(intent.type, [])
        
        if ctx.required_fields:
            ctx.state = DialogState.INFO_GATHERING
            missing = [f for f in ctx.required_fields if f not in ctx.collected_info]
            reply = self._ask_for_field(missing[0])
        else:
            ctx.state = DialogState.PROCESSING
            reply = await self._process_request(ctx)
        
        return reply, ctx
    
    async def _handle_info(self, ctx: ConversationContext, msg: str) -> tuple[str, ConversationContext]:
        """信息收集状态：提取用户提供的信息"""
        # 用 LLM 提取实体
        extracted = await self._extract_entities(msg, ctx.required_fields)
        ctx.collected_info.update(extracted)
        
        # 检查是否所有必需信息都已收集
        missing = [f for f in ctx.required_fields if f not in ctx.collected_info]
        if missing:
            reply = self._ask_for_field(missing[0])
        else:
            ctx.state = DialogState.PROCESSING
            reply = await self._process_request(ctx)
        
        return reply, ctx
    
    async def _process_request(self, ctx: ConversationContext) -> str:
        """处理请求：调用 RAG / API / 工具"""
        ctx.state = DialogState.PROCESSING
        
        if ctx.intent.type == QueryType.FAQ:
            # RAG 检索 + 生成
            answer = await self._rag_answer(ctx)
        elif ctx.intent.type == QueryType.TRANSACTION:
            # 调用业务 API
            answer = await self._query_transaction(ctx)
        elif ctx.intent.type == QueryType.TROUBLESHOOT:
            # RAG + 诊断步骤
            answer = await self._troubleshoot(ctx)
        else:
            answer = "我理解您的问题，让我为您查询相关信息。"
        
        ctx.state = DialogState.CONFIRMING
        return answer + "\n\n请问这个回答是否解决了您的问题？"
    
    async def _handle_confirm(self, ctx: ConversationContext, msg: str) -> tuple[str, ConversationContext]:
        """确认状态：判断是否解决"""
        msg_lower = msg.lower()
        if any(w in msg_lower for w in ["是", "解决了", "好的", "谢谢", "yes"]):
            ctx.state = DialogState.CLOSING
            ctx.satisfaction_score = 5.0
            return "很高兴能帮到您！还有其他问题随时联系我。再见！", ctx
        elif any(w in msg_lower for w in ["不是", "没有", "还不行", "no"]):
            if ctx.turn_count < ctx.max_turns:
                ctx.state = DialogState.INFO_GATHERING
                return "抱歉没能完全解决。能否提供更多细节，我再帮您看看？", ctx
            else:
                ctx.state = DialogState.ESCALATING
                return "这个问题比较复杂，我帮您转接人工客服。", ctx
        else:
            # 新问题
            ctx.state = DialogState.GREETING
            return await self._handle_greeting(ctx, msg)
    
    def _ask_for_field(self, field: str) -> str:
        """请求用户提供某个信息字段"""
        prompts = {
            "order_id": "请问您的订单号是多少？",
            "device_model": "请问您的设备型号是什么？",
            "issue_description": "能否详细描述一下遇到的问题？",
        }
        return prompts.get(field, f"请提供{field}。")
    
    async def _rag_answer(self, ctx: ConversationContext) -> str:
        """RAG 检索 + 生成回答"""
        query = ctx.history[-1]["content"] if ctx.history else ""
        # 检索知识库
        docs = await self.vector_store.search(query, top_k=5)
        # 生成回答
        context = "\n".join([d.content for d in docs])
        return f"基于知识库的回答（上下文: {context[:100]}...）"
    
    async def _query_transaction(self, ctx: ConversationContext) -> str:
        """查询交易信息"""
        order_id = ctx.collected_info.get("order_id")
        # 调用业务 API
        return f"订单 {order_id} 的状态查询结果：已发货，预计明天送达。"
    
    async def _troubleshoot(self, ctx: ConversationContext) -> str:
        """故障排查"""
        issue = ctx.collected_info.get("issue_description", "")
        return f"针对「{issue}」，建议尝试以下步骤：\n1. 重启设备\n2. 检查网络连接\n3. 更新到最新版本"
    
    async def _extract_entities(self, text: str, fields: list[str]) -> dict:
        """从用户消息中提取实体"""
        # 简化版：实际中用 NER 或 LLM 提取
        import re
        entities = {}
        if "order_id" in fields:
            match = re.search(r'\b[A-Z]{2}\d{8,12}\b', text)
            if match:
                entities["order_id"] = match.group()
        if "issue_description" in fields:
            entities["issue_description"] = text
        return entities

class HumanHandoff:
    """人工转接管理"""
    
    def __init__(self):
        self.queue: list[dict] = []  # 等待队列
        self.agents: dict[str, dict] = {}  # 在线人工客服
        self.active_sessions: dict[str, dict] = {}  # 活跃转接会话
    
    async def request_handoff(
        self,
        session_id: str,
        ctx: ConversationContext,
        reason: str = "user_request"
    ) -> dict:
        """请求转接人工"""
        # 收集转接上下文
        handoff_context = {
            "session_id": session_id,
            "user_intent": ctx.intent.type.value if ctx.intent else "unknown",
            "conversation_summary": await self._summarize(ctx),
            "collected_info": ctx.collected_info,
            "reason": reason,
            "priority": self._calculate_priority(ctx),
            "timestamp": asyncio.get_event_loop().time(),
        }
        
        # 检查是否有可用的人工客服
        available_agent = self._find_available_agent()
        if available_agent:
            return await self._connect(session_id, handoff_context, available_agent)
        else:
            # 加入等待队列
            self.queue.append(handoff_context)
            queue_position = len(self.queue)
            estimated_wait = queue_position * 2  # 每个等待预计 2 分钟
            
            return {
                "status": "queued",
                "position": queue_position,
                "estimated_wait_minutes": estimated_wait,
                "message": f"已加入排队，当前第 {queue_position} 位，预计等待 {estimated_wait} 分钟。"
            }
    
    async def _summarize(self, ctx: ConversationContext) -> str:
        """总结对话上下文，帮助人工客服快速了解情况"""
        summary_parts = []
        if ctx.intent:
            summary_parts.append(f"用户意图: {ctx.intent.type.value}")
        if ctx.collected_info:
            info_str = ", ".join(f"{k}: {v}" for k, v in ctx.collected_info.items())
            summary_parts.append(f"已知信息: {info_str}")
        summary_parts.append(f"已对话 {ctx.turn_count} 轮")
        return " | ".join(summary_parts)
    
    def _calculate_priority(self, ctx: ConversationContext) -> int:
        """计算优先级（1=最高）"""
        if ctx.intent and ctx.intent.type == QueryType.COMPLAINT:
            return 1  # 投诉最高优先级
        if ctx.turn_count > 5:
            return 2  # 多轮未解决，提高优先级
        return 3  # 普通
    
    def _find_available_agent(self) -> Optional[str]:
        """找到可用的客服"""
        for agent_id, info in self.agents.items():
            if info["active_sessions"] < info["max_sessions"]:
                return agent_id
        return None
    
    async def _connect(self, session_id: str, context: dict, agent_id: str) -> dict:
        """连接到人工客服"""
        self.agents[agent_id]["active_sessions"] += 1
        self.active_sessions[session_id] = {
            "agent_id": agent_id,
            "context": context,
        }
        return {
            "status": "connected",
            "agent_id": agent_id,
            "context_summary": context["conversation_summary"],
            "message": "已为您接通人工客服，客服已了解您的诉求。"
        }

class SatisfactionEvaluator:
    """多维度满意度评估"""
    
    async def evaluate(self, ctx: ConversationContext) -> dict:
        """评估单次会话满意度"""
        
        # 1. 显式评分（用户主动打分）
        explicit = ctx.satisfaction_score
        
        # 2. 隐式评分（基于行为信号）
        implicit = await self._implicit_score(ctx)
        
        # 3. 情绪分析
        sentiment = await self._analyze_sentiment(ctx.history)
        
        # 4. 综合评分
        overall = self._compute_overall(explicit, implicit, sentiment)
        
        return {
            "session_id": ctx.session_id,
            "explicit_score": explicit,
            "implicit_score": implicit,
            "sentiment": sentiment,
            "overall": overall,
            "resolved": ctx.state == DialogState.CLOSING,
            "turn_count": ctx.turn_count,
            "escalated": ctx.state == DialogState.ESCALATING,
        }
    
    async def _implicit_score(self, ctx: ConversationContext) -> float:
        """基于行为信号的隐式评分"""
        score = 3.0  # 基准分
        
        # 解决了问题 +1
        if ctx.state == DialogState.CLOSING:
            score += 1.0
        
        # 轮次少且解决 +0.5
        if ctx.state == DialogState.CLOSING and ctx.turn_count <= 3:
            score += 0.5
        
        # 转人工 -1
        if ctx.state == DialogState.ESCALATING:
            score -= 1.0
        
        # 超过最大轮次 -0.5
        if ctx.turn_count >= ctx.max_turns:
            score -= 0.5
        
        return max(1.0, min(5.0, score))
    
    async def _analyze_sentiment(self, history: list[dict]) -> float:
        """分析用户情绪 [-1, 1]"""
        # 使用情绪分析模型
        user_messages = [m["content"] for m in history if m["role"] == "user"]
        
        negative_words = ["生气", "失望", "差", "投诉", "无语", "垃圾", "气死"]
        positive_words = ["谢谢", "好的", "解决了", "很棒", "感谢", "满意"]
        
        combined = "".join(user_messages)
        neg_count = sum(1 for w in negative_words if w in combined)
        pos_count = sum(1 for w in positive_words if w in combined)
        
        if neg_count + pos_count == 0:
            return 0.0
        return (pos_count - neg_count) / (pos_count + neg_count)
    
    def _compute_overall(self, explicit, implicit, sentiment) -> float:
        """综合评分"""
        weights = {"explicit": 0.5, "implicit": 0.3, "sentiment": 0.2}
        
        scores = []
        if explicit is not None:
            scores.append(("explicit", explicit, weights["explicit"]))
        scores.append(("implicit", implicit, weights["implicit"]))
        # sentiment 映射到 1-5
        sentiment_score = 3.0 + sentiment * 2
        scores.append(("sentiment", sentiment_score, weights["sentiment"]))
        
        # 重新归一化权重
        total_weight = sum(w for _, _, w in scores)
        overall = sum(s * w / total_weight for _, s, w in scores)
        return round(overall, 2)

@dataclass
class CostBreakdown:
    """单次会话成本分解"""
    llm_input_tokens: int = 0
    llm_output_tokens: int = 0
    rag_queries: int = 0
    api_calls: int = 0
    session_duration_minutes: float = 0
    
    def compute_cost(self, pricing: dict) -> dict:
        return {
            "llm_cost": (
                self.llm_input_tokens * pricing["llm_input_per_1k"] / 1000 +
                self.llm_output_tokens * pricing["llm_output_per_1k"] / 1000
            ),
            "rag_cost": self.rag_queries * pricing["rag_per_query"],
            "api_cost": self.api_calls * pricing["api_per_call"],
            "infra_cost": self.session_duration_minutes * pricing["infra_per_minute"],
        }

# 定价模型示例
PRICING = {
    "llm_input_per_1k": 0.01,   # ¥0.01/1k input tokens
    "llm_output_per_1k": 0.03,  # ¥0.03/1k output tokens
    "rag_per_query": 0.001,     # ¥0.001/次向量检索
    "api_per_call": 0.005,      # ¥0.005/次业务API调用
    "infra_per_minute": 0.02,   # ¥0.02/分钟基础设施
}

# 典型会话成本
# 5轮对话，平均每轮 500 input + 200 output tokens
# 3次RAG检索，2次API调用，持续5分钟
# LLM: 5*(500*0.01/1000 + 200*0.03/1000) = 0.055
# RAG: 3*0.001 = 0.003
# API: 2*0.005 = 0.01
# Infra: 5*0.02 = 0.1
# Total: ¥0.168/会话