AI 智能客服系统实战：从架构设计到日均处理10万咨询

# 第一阶段：规则匹配（毫秒级，覆盖约 40% 请求）
# 维护一个正则表达式库，匹配常见句式

INTENT_RULES = {
    "I001": [  # 订单查询
        r"订单[到状].?[哪哪]了",
        r"查.*物流",
        r"快递[到走]哪",
        r"订单号[是]?\d{10,}",
        r"看看.*订单",
    ],
    "I002": [  # 退款售后
        r"申请退款",
        r"想退[货钱]",
        r"退款[流程怎么]",
        r"不想要了",
        r"商品.*问题.*退款",
    ],
    "I003": [  # 优惠咨询
        r"有什么优惠",
        r"优惠券?[吗呢]",
        r"打[折几]折",
        r"满[减免].",
        r"能便宜[吗点]",
    ],
}

def rule_based_intent(text: str) -> Optional[str]:
    """
    第一阶段：正则规则匹配
    毫秒级响应，零模型调用成本
    """
    for intent_id, patterns in INTENT_RULES.items():
        for pattern in patterns:
            if re.search(pattern, text):
                return intent_id
    return None  # 未匹配，进入第二阶段


# 第二阶段：TF-IDF + SVM 分类器（10-30ms，覆盖约 45% 请求）
# 用历史标注数据训练，零大模型调用成本

import joblib
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.svm import LinearSVC

class IntentClassifier:
    """
    基于 TF-IDF + Linear SVM 的意图分类器
    训练数据：20,000 条人工标注的客服对话
    """
    
    def __init__(self):
        self.vectorizer = TfidfVectorizer(
            max_features=5000,       # 最多5000个特征词
            ngram_range=(1, 2),      # unigram + bigram
            min_df=3,               # 至少出现3次
            max_df=0.8,             # 最多出现在80%的文档中
            sublinear_tf=True       # 使用 1+log(tf) 而非原始tf
        )
        self.classifier = LinearSVC(
            C=1.0,                  # 正则化参数
            class_weight='balanced', # 类别权重（处理数据不平衡）
            max_iter=5000
        )
        self.is_trained = False
    
    def train(self, texts: List[str], labels: List[str]):
        """
        训练意图分类器
        
        Args:
            texts: 分词后的文本列表
            labels: 意图ID标签
        """
        X = self.vectorizer.fit_transform(texts)
        y = labels
        self.classifier.fit(X, y)
        self.is_trained = True
        
        # 输出训练报告
        from sklearn.metrics import classification_report
        y_pred = self.classifier.predict(X)
        print(classification_report(y, y_pred, digits=3))
    
    def predict(self, text: str) -> Tuple[str, float]:
        """
        预测意图类别和置信度
        
        Returns:
            (intent_id, confidence_score)
        """
        X = self.vectorizer.transform([text])
        pred = self.classifier.predict(X)[0]
        
        # 获取置信度（decision_function 的 Platt scaling）
        scores = self.classifier.decision_function(X)[0]
        confidence = float(max(scores) / sum(scores))  # 简化的置信度
        
        return pred, confidence
    
    def load(self, model_path: str):
        """加载已保存的模型"""
        data = joblib.load(model_path)
        self.vectorizer = data['vectorizer']
        self.classifier = data['classifier']
        self.is_trained = True
    
    def save(self, model_path: str):
        """保存模型"""
        joblib.dump({
            'vectorizer': self.vectorizer,
            'classifier': self.classifier
        }, model_path)


# 第三阶段：LLM 兜底（200-500ms，覆盖剩余 15% 复杂请求）
# 规则和 ML 分类器都无法高置信度判断时，调用大模型

INTENT_PROMPT = """你是一个电商客服意图分类器，请将用户问题分类到以下12个意图之一：

I001=订单查询（如：查物流、订单状态）
I002=退款售后（如：申请退款、商品问题）
I003=优惠咨询（如：优惠券、折扣活动）
I004=商品咨询（如：产品怎么样、有没有货）
I005=账号问题（如：忘记密码、账户异常）
I006=投诉建议（如：投诉、反馈问题）
I007=物流异常（如：快递丢失、一直不派送）
I008=发票相关（如：开发票、电子发票）
I009=支付问题（如：支付失败、无法付款）
I010=评价管理（如：追评、删除评价）
I011=活动咨询（如：活动规则、参与方式）
I012=其他/寒暄（如：你好、谢谢、再见）

用户问题：{user_input}

请输出JSON格式：{{"intent":"IXXX","confidence":0.XX,"reason":"分类理由"}}
只输出JSON，不要其他内容。"""

def llm_intent_classify(text: str, model: str = "qwen2-72b") -> dict:
    """
    LLM 意图分类兜底方案
    当规则和 ML 都无法高置信度判断时调用
    """
    # 使用本地部署的 Qwen2-72B
    response = ollama.chat(
        model=model,
        messages=[
            {"role": "system", "content": INTENT_PROMPT},
            {"role": "user", "content": text}
        ],
        options={
            "temperature": 0.1,  # 低温度保证稳定分类
            "num_predict": 200  # 限制输出长度
        }
    )
    result = json.loads(response['message']['content'])
    return result


# 完整的三层级联意图识别流程
def classify_intent(text: str) -> dict:
    """
    三层级联意图识别
    
    性能指标：
    - 第一阶段（规则）: < 1ms，覆盖 ~40% 请求
    - 第二阶段（SVM）: 10-30ms，覆盖 ~45% 请求
    - 第三阶段（LLM）: 200-500ms，覆盖 ~15% 请求
    - 平均响应时间: ~15ms（整体）
    """
    # 层1：规则匹配
    intent = rule_based_intent(text)
    if intent:
        return {"intent": intent, "confidence": 0.99, "stage": "rule"}
    
    # 层2：SVM 分类
    intent, confidence = svm_classifier.predict(text)
    if confidence > 0.85:
        return {"intent": intent, "confidence": confidence, "stage": "svm"}
    
    # 层3：LLM 兜底
    result = llm_intent_classify(text)
    return {
        "intent": result["intent"],
        "confidence": result["confidence"],
        "stage": "llm",
        "reason": result["reason"]
    }

# 文档处理流水线：从原始文档到可检索向量

import pymupdf          # PDF 解析
import pypandoc         # Word 转换
from bs4 import BeautifulSoup  # HTML 解析
import jieba            # 中文分词
import hashlib          # 内容去重

class DocumentProcessor:
    """
    文档处理流水线：
    1. 格式检测与解析（PDF/Word/HTML/TXT）
    2. 内容清洗（去除噪音、提取正文）
    3. 分块处理（基于语义/长度双维度）
    4. 向量化（embedding）
    5. 写入向量数据库
    """
    
    def __init__(self, embedder, vector_db):
        self.embedder = embedder
        self.vector_db = vector_db
    
    def process(self, file_path: str, category: str, metadata: dict = None):
        """
        处理单个文档
        
        Args:
            file_path: 文件路径
            category: 知识库分类（I001-I012）
            metadata: 额外元数据（作者、日期、标签等）
        """
        # 1. 检测文件格式并解析
        content = self._parse_file(file_path)
        
        # 2. 内容清洗
        cleaned = self._clean_content(content)
        
        # 3. 分块处理（递归字符分割 + 语义边界检测）
        chunks = self._chunk_text(cleaned, max_tokens=500)
        
        # 4. 生成向量并写入数据库
        for i, chunk in enumerate(chunks):
            chunk_hash = hashlib.md5(chunk.encode()).hexdigest()
            
            # 检查是否已存在（去重）
            if self._exists_in_db(chunk_hash):
                continue
            
            # 生成向量
            embedding = self.embedder.encode(chunk)
            
            # 写入向量数据库
            self.vector_db.insert(
                id=chunk_hash,
                text=chunk,
                vector=embedding,
                metadata={
                    "category": category,
                    "source": file_path,
                    "chunk_index": i,
                    **metadata
                }
            )
    
    def _parse_file(self, file_path: str) -> str:
        """根据文件格式选择解析器"""
        ext = os.path.splitext(file_path)[1].lower()
        
        parsers = {
            '.pdf': self._parse_pdf,
            '.docx': self._parse_docx,
            '.doc': self._parse_doc,
            '.html': self._parse_html,
            '.txt': self._parse_txt,
        }
        
        parser = parsers.get(ext, self._parse_txt)
        return parser(file_path)
    
    def _parse_pdf(self, file_path: str) -> str:
        """解析 PDF 文件"""
        text_parts = []
        with pymupdf.open(file_path) as doc:
            for page in doc:
                text_parts.append(page.get_text())
        return "\n".join(text_parts)
    
    def _parse_docx(self, file_path: str) -> str:
        """解析 Word 文档"""
        from docx import Document
        doc = Document(file_path)
        return "\n".join([p.text for p in doc.paragraphs])
    
    def _parse_html(self, file_path: str) -> str:
        """解析 HTML 文件"""
        with open(file_path, 'r', encoding='utf-8') as f:
            soup = BeautifulSoup(f.read(), 'html.parser')
        # 移除 script 和 style 标签
        for tag in soup(['script', 'style', 'nav', 'footer']):
            tag.decompose()
        return soup.get_text(separator="\n", strip=True)
    
    def _clean_content(self, text: str) -> str:
        """内容清洗"""
        # 移除多余空白
        text = re.sub(r'\n{3,}', '\n\n', text)
        text = re.sub(r' {2,}', ' ', text)
        # 移除特殊控制字符
        text = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', text)
        return text.strip()
    
    def _chunk_text(self, text: str, max_tokens: int = 500) -> List[str]:
        """
        智能分块：基于 token 数和语义边界双重限制
        
        语义边界优先级：
        1. 段落分隔（\n\n）—— 优先保证段落完整
        2. 句子分隔（。）—— 次优先
        3. 硬截断 —— 最后的兜底手段
        """
        chunks = []
        
        # 先按段落分割
        paragraphs = text.split('\n\n')
        current_chunk = []
        current_token_count = 0
        
        for para in paragraphs:
            para_tokens = self._estimate_tokens(para)
            
            # 如果单个段落就超过 max_tokens，进一步分割
            if para_tokens > max_tokens:
                if current_chunk:
                    chunks.append('\n\n'.join(current_chunk))
                    current_chunk = []
                    current_token_count = 0
                
                # 递归分割长段落
                sub_chunks = self._split_long_paragraph(para, max_tokens)
                chunks.extend(sub_chunks)
            elif current_token_count + para_tokens > max_tokens:
                # 当前块满了，保存并新建
                chunks.append('\n\n'.join(current_chunk))
                current_chunk = [para]
                current_token_count = para_tokens
            else:
                current_chunk.append(para)
                current_token_count += para_tokens
        
        if current_chunk:
            chunks.append('\n\n'.join(current_chunk))
        
        return [c.strip() for c in chunks if c.strip()]
    
    def _estimate_tokens(self, text: str) -> int:
        """估算中文字符数（粗略：1中文≈1.5 token）"""
        chinese = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        other = len(text) - chinese
        return int(chinese * 1.5 + other * 0.25)
    
    def _exists_in_db(self, chunk_hash: str) -> bool:
        """检查内容是否已存在（去重）"""
        return self.vector_db.exists(chunk_hash)

# Dify 工作流 JSON 配置（简化版）
# 实际通过 Dify 可视化界面配置，此处展示结构

workflow = {
    "version": "1.0",
    "nodes": [
        {
            "id": "node_start",
            "type": "start",
            "config": {
                "inputs": {"user_message": "{{user.message}}"},
                "outputs": {"message": "user_message"}
            }
        },
        {
            "id": "node_intent",
            "type": "llm",
            "config": {
                "model": "qwen2-72b",
                "prompt": INTENT_PROMPT,
                "inputs": {"user_input": "{{node_start.message}}"},
                "outputs": {"intent": "result.intent", "confidence": "result.confidence"}
            }
        },
        {
            "id": "node_router",
            "type": "if-else",
            "config": {
                "condition": "{{node_intent.confidence}} > 0.75",
                "true_branch": "node_rag",    # 高置信 → RAG 检索
                "false_branch": "node_human" # 低置信 → 转人工
            }
        },
        {
            "id": "node_rag",
            "type": "rag-retrieval",
            "config": {
                "retrieval_strategy": "semantic",
                "top_k": 5,
                "score_threshold": 0.6,
                "category_filter": "{{node_intent.intent}}"
            }
        },
        {
            "id": "node_answer",
            "type": "llm",
            "config": {
                "model": "qwen2-72b",
                "prompt": """根据以下参考知识回答用户问题。
                如果知识不足以回答，请说"这个问题我需要转人工为您解答"。
                
                参考知识：
                {{node_rag.context}}
                
                用户问题：{{node_start.message}}
                
                要求：
                - 直接给出答案，不要说"根据知识库"
                - 如需引用，在答案末尾加"[来源：文件名]"
                - 保持礼貌、简洁、口语化""",
                "inputs": {
                    "context": "{{node_rag.context}}",
                    "message": "{{node_start.message}}"
                }
            }
        },
        {
            "id": "node_human",
            "type": "http-request",
            "config": {
                "url": "http://ticket-service:8084/api/transfer",
                "method": "POST",
                "body": {
                    "user_id": "{{user.id}}",
                    "message": "{{node_start.message}}",
                    "intent": "{{node_intent.intent}}",
                    "history": "{{conversation.history}}"
                }
            }
        },
        {
            "id": "node_end",
            "type": "end",
            "config": {
                "outputs": "{{node_answer.result}}"
            }
        }
    ],
    "edges": [
        {"from": "node_start", "to": "node_intent"},
        {"from": "node_intent", "to": "node_router"},
        {"from": "node_router", "to": "node_rag", "condition": "true"},
        {"from": "node_router", "to": "node_human", "condition": "false"},
        {"from": "node_rag", "to": "node_answer"},
        {"from": "node_answer", "to": "node_end"},
        {"from": "node_human", "to": "node_end"}
    ]
}

# Kubernetes 部署配置（关键部分）

---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: customer-service-intent
spec:
  replicas: 3  # 3副本保障高可用
  selector:
    matchLabels:
      app: intent-service
  template:
    spec:
      containers:
      - name: intent
        image: registry.example.com/customer-service:intent-v2.3
        resources:
          requests:
            memory: "4Gi"
            cpu: "2"
          limits:
            memory: "8Gi"
            gpu.intel.com/gpu: "1"  # 申请 1 个 GPU
        env:
        - name: OLLAMA_BASE_URL
          value: "http://ollama-service:11434"
        - name: REDIS_URL
          value: "redis://redis:6379/0"
        readinessProbe:
          httpGet:
            path: /health
            port: 8082
          initialDelaySeconds: 30
          periodSeconds: 5
        livenessProbe:
          httpGet:
            path: /health
            port: 8082
          initialDelaySeconds: 60
          periodSeconds: 10
      nodeSelector:
        gpu: "true"  # 仅调度到有 GPU 的节点

---
apiVersion: v1
kind: Service
metadata:
  name: intent-service
spec:
  type: ClusterIP
  selector:
    app: intent-service
  ports:
  - port: 8082
    targetPort: 8082

# HPA 自动扩缩容（基于请求量）
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: intent-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: customer-service-intent
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70