AI智能客服系统开发完全教程

一、AI客服系统架构概述

一套完整的AI智能客服系统由四大核心模块构成：

用户消息 → [意图识别] → [实体抽取] → [对话管理] → [回复生成] → 用户
                ↑                        ↓
            NLU模块                   知识库/API

意图识别（Intent Classification）：判断用户想要做什么，例如"查询订单"、"申请退款"、"咨询产品"。

实体抽取（Entity Extraction）：从用户输入中提取关键信息，如订单号、商品名称、日期等。

对话管理（Dialogue Management）：维护对话状态，决定下一步动作——是继续追问、查询数据库还是转人工。

回复生成（Response Generation）：基于对话状态和模板/LLM生成最终回复。

各模块职责明确、松耦合设计，便于独立迭代和替换。

二、主流方案对比

选型建议：初创团队推荐Rasa快速验证；中型企业可用百度UNIT降低冷启动成本；大型企业建议自研LLM方案以获得最大控制权。

三、意图识别与NLU模块开发

3.1 基于Rasa的NLU Pipeline配置

Rasa使用YAML格式定义训练数据和Pipeline：

# nlu.yml - 训练数据
version: "3.1"
nlu:
- intent: query_order
  examples: |
    - 查一下我的订单
    - 我的快递到哪了
    - 订单号 [12345678](order_id) 物流信息
    - 帮我看看 [A20240101001](order_id) 的状态

- intent: request_refund
  examples: |
    - 我要退货
    - 申请退款
    - 这个商品有问题，我要退
    - 帮我退 [订单号98765432](order_id)

- intent: product_inquiry
  examples: |
    - 这款手机防水吗
    - [iPhone 15](product) 多少钱
    - 有没有 [红色](color) 的 [卫衣](product)

3.2 Pipeline配置

# config.yml
pipeline:
  - name: WhitespaceTokenizer
  - name: RegexFeaturizer
  - name: LexicalSyntacticFeaturizer
  - name: CountVectorsFeaturizer
  - name: CountVectorsFeaturizer
    analyzer: char_wb
    min_ngram: 1
    max_ngram: 4
  - name: DIETClassifier
    epochs: 100
    constrain_similarities: true
  - name: EntitySynonymMapper
  - name: ResponseSelector
    epochs: 100

3.3 自定义意图分类器（PyTorch实现）

当预训练模型不满足需求时，可以自定义分类器：

import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from transformers import BertTokenizer, BertModel

class IntentDataset(Dataset):
    def __init__(self, texts, labels, tokenizer, max_len=128):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_len = max_len

    def __len__(self):
        return len(self.texts)

    def __getitem__(self, idx):
        encoding = self.tokenizer(
            self.texts[idx],
            max_length=self.max_len,
            padding='max_length',
            truncation=True,
            return_tensors='pt'
        )
        return {
            'input_ids': encoding['input_ids'].squeeze(),
            'attention_mask': encoding['attention_mask'].squeeze(),
            'label': torch.tensor(self.labels[idx], dtype=torch.long)
        }

class IntentClassifier(nn.Module):
    def __init__(self, num_classes, bert_model='bert-base-chinese'):
        super().__init__()
        self.bert = BertModel.from_pretrained(bert_model)
        self.dropout = nn.Dropout(0.3)
        self.classifier = nn.Linear(self.bert.config.hidden_size, num_classes)

    def forward(self, input_ids, attention_mask):
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        pooled = outputs.pooler_output  # [CLS] token
        dropped = self.dropout(pooled)
        return self.classifier(dropped)

# 训练循环
def train_intent_model(model, train_loader, epochs=10, lr=2e-5):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model.to(device)
    optimizer = torch.optim.AdamW(model.parameters(), lr=lr)
    criterion = nn.CrossEntropyLoss()

    for epoch in range(epochs):
        model.train()
        total_loss = 0
        for batch in train_loader:
            input_ids = batch['input_ids'].to(device)
            mask = batch['attention_mask'].to(device)
            labels = batch['label'].to(device)

            logits = model(input_ids, mask)
            loss = criterion(logits, labels)

            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
            total_loss += loss.item()

        print(f"Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(train_loader):.4f}")

四、对话流程设计与状态管理

4.1 状态机设计

对话管理的核心是一个有限状态机（FSM）：

from enum import Enum
from typing import Dict, Optional, Callable

class DialogueState(Enum):
    GREETING = "greeting"
    COLLECTING_INFO = "collecting_info"
    QUERYING = "querying"
    CONFIRMING = "confirming"
    RESOLVING = "resolving"
    CLOSING = "closing"
    TRANSFERRING = "transferring"

class DialogueStateMachine:
    def __init__(self):
        self.state = DialogueState.GREETING
        self.context: Dict = {}
        self.transitions: Dict[tuple, Callable] = {}
        self._setup_transitions()

    def _setup_transitions(self):
        """定义状态转移规则"""
        self.transitions = {
            (DialogueState.GREETING, "query_order"): self._start_order_query,
            (DialogueState.COLLECTING_INFO, "provide_info"): self._collect_info,
            (DialogueState.QUERYING, "query_complete"): self._present_result,
            (DialogueState.CONFIRMING, "confirm"): self._resolve,
            (DialogueState.CONFIRMING, "deny"): self._restart,
        }

    def _start_order_query(self, entities: Dict):
        self.context['intent'] = 'query_order'
        self.context['order_id'] = entities.get('order_id')
        if self.context['order_id']:
            self.state = DialogueState.QUERYING
        else:
            self.state = DialogueState.COLLECTING_INFO
            return "请提供您的订单号，我来帮您查询。"

    def _collect_info(self, entities: Dict):
        self.context.update(entities)
        missing = self._check_missing_slots()
        if not missing:
            self.state = DialogueState.QUERYING
            return None  # 信息齐全，进入查询
        return f"还需要您提供：{', '.join(missing)}"

    def _present_result(self, result: str):
        self.state = DialogueState.CONFIRMING
        return f"{result}\n请问还有其他问题吗？"

    def _resolve(self, _entities: Dict):
        self.state = DialogueState.CLOSING
        return "感谢您的咨询，祝您生活愉快！"

    def _restart(self, _entities: Dict):
        self.context.clear()
        self.state = DialogueState.GREETING
        return "好的，请问还有什么可以帮您的？"

    def _check_missing_slots(self) -> list:
        required = {'order_id'}
        return [s for s in required if not self.context.get(s)]

    def process(self, intent: str, entities: Dict) -> str:
        key = (self.state, intent)
        handler = self.transitions.get(key)
        if handler:
            result = handler(entities)
            if result:
                return result
            # 递归处理状态转移后的自动动作
            return self.process(intent, entities)
        return "抱歉，我没有理解您的意思，能换个说法吗？"

4.2 Rasa Stories与Rules

在Rasa中，通过Stories和Rules定义对话流程：

# stories.yml
stories:
- story: 查询订单流程
  steps:
  - intent: query_order
  - action: action_query_order
  - slot_was_set:
    - order_id: null
  - action: utter_ask_order_id
  - intent: provide_order_id
  - action: action_query_order
  - action: utter_order_result

- story: 退款流程
  steps:
  - intent: request_refund
  - action: action_validate_refund
  - action: utter_confirm_refund
  - intent: confirm
  - action: action_process_refund
  - action: utter_refund_success

五、知识库构建与FAQ检索

5.1 向量检索方案

使用Sentence-Transformers将FAQ转为向量，通过余弦相似度检索：

import numpy as np
from sentence_transformers import SentenceTransformer
from typing import List, Tuple

class FAQRetriever:
    def __init__(self, model_name='paraphrase-multilingual-MiniLM-L12-v2'):
        self.model = SentenceTransformer(model_name)
        self.questions: List[str] = []
        self.answers: List[str] = []
        self.embeddings: np.ndarray = None

    def build_index(self, faq_data: List[dict]):
        """构建FAQ索引"""
        self.questions = [item['question'] for item in faq_data]
        self.answers = [item['answer'] for item in faq_data]
        self.embeddings = self.model.encode(
            self.questions,
            normalize_embeddings=True,
            show_progress_bar=True
        )

    def search(self, query: str, top_k: int = 3,
               threshold: float = 0.6) -> List[Tuple[str, str, float]]:
        """检索最相关的FAQ"""
        query_vec = self.model.encode([query], normalize_embeddings=True)
        scores = np.dot(self.embeddings, query_vec.T).squeeze()

        top_indices = np.argsort(scores)[::-1][:top_k]
        results = []
        for idx in top_indices:
            if scores[idx] >= threshold:
                results.append((
                    self.questions[idx],
                    self.answers[idx],
                    float(scores[idx])
                ))
        return results

# 使用示例
faq_data = [
    {"question": "如何退货", "answer": "在订单详情页点击'申请退货'，填写退货原因后提交。"},
    {"question": "退货流程是什么", "answer": "1. 提交退货申请 2. 等待审核 3. 寄回商品 4. 收到退款"},
    {"question": "多久能收到退款", "answer": "审核通过后，退款将在3-5个工作日内原路返回。"},
    {"question": "可以修改收货地址吗", "answer": "未发货的订单可以在订单详情页修改地址。"},
]

retriever = FAQRetriever()
retriever.build_index(faq_data)
results = retriever.search("我想退货怎么办")
for q, a, score in results:
    print(f"[{score:.3f}] Q: {q}\n   A: {a}\n")

5.2 混合检索策略

结合向量检索和关键词匹配，提升召回率：

import jieba
from collections import Counter

class HybridRetriever:
    def __init__(self, faq_data: List[dict]):
        self.vector_retriever = FAQRetriever()
        self.vector_retriever.build_index(faq_data)
        self.faq_data = faq_data
        # 构建倒排索引
        self.inverted_index: Dict[str, List[int]] = {}
        for i, item in enumerate(faq_data):
            words = set(jieba.cut(item['question']))
            for word in words:
                self.inverted_index.setdefault(word, []).append(i)

    def keyword_search(self, query: str, top_k: int = 3) -> List[int]:
        words = set(jieba.cut(query))
        candidate_counts = Counter()
        for word in words:
            for idx in self.inverted_index.get(word, []):
                candidate_counts[idx] += 1
        return [idx for idx, _ in candidate_counts.most_common(top_k)]

    def search(self, query: str, top_k: int = 3,
               vector_weight: float = 0.7) -> List[Tuple[str, str, float]]:
        # 向量检索结果
        vector_results = self.vector_retriever.search(query, top_k=top_k)
        # 关键词检索结果
        keyword_indices = self.keyword_search(query, top_k=top_k)

        # 融合评分
        score_map = {}
        for q, a, score in vector_results:
            score_map[q] = score_map.get(q, 0) + score * vector_weight

        for idx in keyword_indices:
            q = self.faq_data[idx]['question']
            a = self.faq_data[idx]['answer']
            keyword_score = 0.5  # 基础关键词分数
            score_map[q] = score_map.get(q, 0) + keyword_score * (1 - vector_weight)

        sorted_items = sorted(score_map.items(), key=lambda x: x[1], reverse=True)
        results = []
        for q, score in sorted_items[:top_k]:
            a = next(item['answer'] for item in self.faq_data if item['question'] == q)
            results.append((q, a, score))
        return results

六、多轮对话管理与上下文维护

6.1 上下文管理器

import time
from dataclasses import dataclass, field
from typing import List, Dict, Optional

@dataclass
class Message:
    role: str          # 'user' 或 'assistant'
    content: str
    timestamp: float = field(default_factory=time.time)
    intent: Optional[str] = None
    entities: Dict = field(default_factory=dict)

class ConversationContext:
    def __init__(self, session_id: str, max_history: int = 20):
        self.session_id = session_id
        self.max_history = max_history
        self.history: List[Message] = []
        self.slots: Dict = {}           # 已收集的槽位
        self.pending_slots: List[str] = []  # 待收集的槽位
        self.state: str = "active"
        self.created_at = time.time()

    def add_message(self, role: str, content: str, **kwargs):
        msg = Message(role=role, content=content, **kwargs)
        self.history.append(msg)
        # 超出长度时截断
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]

    def get_recent_context(self, n: int = 5) -> str:
        """获取最近n轮对话作为上下文"""
        recent = self.history[-n*2:]  # n轮 = n*2条消息
        lines = []
        for msg in recent:
            role = "用户" if msg.role == "user" else "客服"
            lines.append(f"{role}: {msg.content}")
        return "\n".join(lines)

    def update_slots(self, entities: Dict):
        self.slots.update(entities)
        self.pending_slots = [
            s for s in self.pending_slots if s not in self.slots
        ]

    def is_complete(self) -> bool:
        return len(self.pending_slots) == 0

class ContextManager:
    """管理所有会话的上下文"""

    def __init__(self, session_ttl: int = 1800):
        self.sessions: Dict[str, ConversationContext] = {}
        self.session_ttl = session_ttl

    def get_or_create(self, session_id: str) -> ConversationContext:
        self._cleanup_expired()
        if session_id not in self.sessions:
            self.sessions[session_id] = ConversationContext(session_id)
        return self.sessions[session_id]

    def _cleanup_expired(self):
        now = time.time()
        expired = [
            sid for sid, ctx in self.sessions.items()
            if now - ctx.created_at > self.session_ttl
        ]
        for sid in expired:
            del self.sessions[sid]

七、LLM增强的智能客服

7.1 接入大语言模型

利用LLM处理复杂问题和生成自然回复：

import openai
from typing import Optional

class LLMEnhanced客服:
    def __init__(self, api_key: str, base_url: Optional[str] = None):
        self.client = openai.OpenAI(
            api_key=api_key,
            base_url=base_url  # 兼容其他OpenAI兼容接口
        )
        self.system_prompt = """你是一个专业的电商客服助手。
规则：
1. 回答要简洁专业，语气友好
2. 不确定的信息不要编造，引导用户联系人工客服
3. 涉及退款、投诉等敏感操作需确认后再执行
4. 始终以用户利益为先"""

    def generate_response(self, user_query: str,
                          context: str,
                          kb_context: str = "") -> str:
        messages = [
            {"role": "system", "content": self.system_prompt},
        ]

        # 注入知识库上下文
        if kb_context:
            messages.append({
                "role": "system",
                "content": f"参考知识库信息：\n{kb_context}"
            })

        # 注入对话历史
        if context:
            messages.append({
                "role": "system",
                "content": f"对话历史：\n{context}"
            })

        messages.append({"role": "user", "content": user_query})

        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            temperature=0.3,
            max_tokens=500
        )
        return response.choices[0].message.content

    def classify_intent_with_llm(self, user_query: str) -> dict:
        """使用LLM进行意图识别（兜底方案）"""
        prompt = f"""分析以下用户消息，返回JSON格式：
- intent: 意图类别（query_order/refund/product_inquiry/complaint/greeting/other）
- entities: 提取的实体字典
- confidence: 置信度(0-1)

用户消息：{user_query}

只返回JSON，不要其他内容。"""

        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            response_format={"type": "json_object"}
        )
        import json
        return json.loads(response.choices[0].message.content)

7.2 RAG增强回复

将知识库检索结果注入LLM上下文，实现精准回答：

class RAG客服:
    def __init__(self, llm客服: LLMEnhanced客服, retriever: HybridRetriever):
        self.llm = llm客服
        self.retriever = retriever

    def answer(self, query: str, context: str) -> str:
        # 检索相关知识
        kb_results = self.retriever.search(query, top_k=3)
        kb_context = "\n".join([
            f"Q: {q}\nA: {a}" for q, a, _ in kb_results
        ]) if kb_results else ""

        # 生成回复
        return self.llm.generate_response(
            user_query=query,
            context=context,
            kb_context=kb_context
        )

八、情感分析与人工转接策略

8.1 情感分析模块

from transformers import pipeline

class SentimentAnalyzer:
    def __init__(self):
        self.classifier = pipeline(
            "sentiment-analysis",
            model="uer/roberta-base-finetuned-chinanews-chinese"
        )
        self.anger_keywords = {'投诉', '举报', '垃圾', '骗子', '差评',
                               '恶心', '愤怒', '举报', '消协', '315'}

    def analyze(self, text: str) -> dict:
        # 关键词快速检测
        has_anger = any(kw in text for kw in self.anger_keywords)

        # 模型分析
        result = self.classifier(text[:512])[0]

        return {
            'sentiment': result['label'],
            'score': result['score'],
            'is_angry': has_anger or (result['label'] == 'negative' and result['score'] > 0.8),
            'should_transfer': has_anger  # 遇到愤怒情绪建议转人工
        }

8.2 转人工策略

class TransferPolicy:
    """转人工判断策略"""

    def __init__(self):
        self.max_auto_rounds = 10        # 最大自动对话轮数
        self.max_no_match = 3            # 连续未匹配次数
        self.transfer_keywords = {'转人工', '人工客服', '真人', '找人'}

    def should_transfer(self, context: ConversationContext,
                        sentiment: dict) -> tuple:
        """返回 (是否转人工, 原因)"""

        # 用户主动要求
        last_msg = context.history[-1].content if context.history else ""
        if any(kw in last_msg for kw in self.transfer_keywords):
            return True, "user_request"

        # 情绪激动
        if sentiment.get('is_angry'):
            return True, "anger_detected"

        # 多轮未解决
        if len(context.history) > self.max_auto_rounds:
            return True, "max_rounds_exceeded"

        # 连续意图未识别
        recent_intents = [
            m.intent for m in context.history[-5:]
            if m.role == 'user'
        ]
        if recent_intents.count(None) >= self.max_no_match:
            return True, "repeated_no_match"

        return False, None

九、多渠道接入

9.1 统一消息适配层

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional

@dataclass
class UnifiedMessage:
    channel: str           # 'wechat', 'web', 'app'
    user_id: str
    content: str
    msg_type: str = 'text' # text, image, voice
    extra: dict = None

class ChannelAdapter(ABC):
    @abstractmethod
    async def receive(self, raw_data: dict) -> UnifiedMessage:
        """将原始数据转为统一消息格式"""
        pass

    @abstractmethod
    async def send(self, user_id: str, content: str, **kwargs):
        """发送回复到对应渠道"""
        pass

class WebAdapter(ChannelAdapter):
    def __init__(self):
        self.connections: Dict[str, any] = {}

    async def receive(self, raw_data: dict) -> UnifiedMessage:
        return UnifiedMessage(
            channel='web',
            user_id=raw_data['session_id'],
            content=raw_data['message'],
            msg_type=raw_data.get('type', 'text')
        )

    async def send(self, user_id: str, content: str, **kwargs):
        ws = self.connections.get(user_id)
        if ws:
            await ws.send_json({
                'type': 'message',
                'content': content,
                'timestamp': time.time()
            })

class WeChatAdapter(ChannelAdapter):
    """微信公众号/企业微信适配器"""

    def __init__(self, token: str, encoding_aes_key: str):
        self.token = token
        self.encoding_aes_key = encoding_aes_key

    async def receive(self, raw_data: dict) -> UnifiedMessage:
        return UnifiedMessage(
            channel='wechat',
            user_id=raw_data['FromUserName'],
            content=raw_data.get('Content', ''),
            msg_type=raw_data.get('MsgType', 'text')
        )

    async def send(self, user_id: str, content: str, **kwargs):
        # 调用微信API发送消息
        pass

class ChannelRouter:
    """消息路由器"""

    def __init__(self):
        self.adapters: Dict[str, ChannelAdapter] = {}

    def register(self, channel: str, adapter: ChannelAdapter):
        self.adapters[channel] = adapter

    async def handle_incoming(self, channel: str, raw_data: dict):
        adapter = self.adapters[channel]
        message = await adapter.receive(raw_data)
        # 交给核心处理引擎
        response = await self.process_engine.handle(message)
        await adapter.send(message.user_id, response)

9.2 FastAPI服务

from fastapi import FastAPI, WebSocket, Request
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="AI客服系统")
app.add_middleware(CORSMiddleware, allow_origins=["*"])

router = ChannelRouter()
context_mgr = ContextManager()

@app.post("/api/webhook/wechat")
async def wechat_webhook(request: Request):
    data = await request.json()
    await router.handle_incoming('wechat', data)
    return {"status": "ok"}

@app.websocket("/ws/chat/{session_id}")
async def websocket_chat(websocket: WebSocket, session_id: str):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_json()
            data['session_id'] = session_id
            await router.handle_incoming('web', data)
    except Exception:
        pass

十、实战案例：电商智能客服系统

将上述模块整合为完整的电商客服系统：

"""
电商智能客服系统 - 完整实现
"""
import json
import time
import uuid
from typing import Dict, Optional

class Ecommerce客服系统:
    def __init__(self, config: dict):
        # 初始化各模块
        self.context_mgr = ContextManager(session_ttl=1800)
        self.sentiment = SentimentAnalyzer()
        self.transfer_policy = TransferPolicy()
        self.faq_retriever = HybridRetriever(self._load_faq())
        self.llm = LLMEnhanced客服(
            api_key=config['llm_api_key'],
            base_url=config.get('llm_base_url')
        )
        self.rag = RAG客服(self.llm, self.faq_retriever)
        self.state_machine_factory = DialogueStateMachine

        # 订单查询模拟
        self.order_db = {
            "12345678": {
                "status": "已发货", "tracking": "SF1234567890",
                "items": ["iPhone 15 Pro Max x1"],
                "total": 9999.00
            }
        }

    def _load_faq(self) -> list:
        return [
            {"question": "如何退货", "answer": "在订单详情页点击'申请退货'，填写原因后提交。"},
            {"question": "退货流程", "answer": "1.提交申请 2.等待审核 3.寄回商品 4.3-5天退款到账"},
            {"question": "如何修改地址", "answer": "未发货订单可在订单详情页修改收货地址。"},
            {"question": "运费谁承担", "answer": "质量问题退货运费由商家承担；个人原因退货需自行承担。"},
            {"question": "会员权益", "answer": "普通会员享9.5折，黄金会员9折，钻石会员8.5折。"},
        ]

    async def handle_message(self, channel: str, user_id: str,
                              content: str) -> str:
        # 获取/创建会话上下文
        session_id = f"{channel}:{user_id}"
        ctx = self.context_mgr.get_or_create(session_id)

        # 记录用户消息
        ctx.add_message('user', content)

        # 情感分析
        sentiment = self.sentiment.analyze(content)

        # 检查是否需要转人工
        should_transfer, reason = self.transfer_policy.should_transfer(
            ctx, sentiment
        )
        if should_transfer:
            ctx.state = "transferring"
            return self._handle_transfer(reason)

        # 意图识别与实体抽取
        nlu_result = self.llm.classify_intent_with_llm(content)
        intent = nlu_result.get('intent', 'other')
        entities = nlu_result.get('entities', {})
        ctx.add_message('assistant', '', intent=intent, entities=entities)

        # 根据意图分发处理
        response = await self._dispatch(intent, entities, ctx)

        # 记录回复
        ctx.add_message('assistant', response)
        return response

    async def _dispatch(self, intent: str, entities: Dict,
                        ctx: ConversationContext) -> str:
        handlers = {
            'query_order': self._handle_order_query,
            'refund': self._handle_refund,
            'product_inquiry': self._handle_product_inquiry,
            'greeting': self._handle_greeting,
            'complaint': self._handle_complaint,
        }

        handler = handlers.get(intent, self._handle_unknown)
        return await handler(entities, ctx)

    async def _handle_order_query(self, entities: Dict,
                                   ctx: ConversationContext) -> str:
        order_id = entities.get('order_id')
        if not order_id:
            ctx.pending_slots = ['order_id']
            return "请提供您的订单号，我来帮您查询物流信息。"

        order = self.order_db.get(order_id)
        if not order:
            return f"未找到订单号 {order_id}，请核实后重试。"

        items = "、".join(order['items'])
        return (f"订单 {order_id} 状态：{order['status']}\n"
                f"商品：{items}\n"
                f"物流单号：{order['tracking']}\n"
                f"如需更多帮助，请随时告诉我。")

    async def _handle_refund(self, entities: Dict,
                              ctx: ConversationContext) -> str:
        order_id = entities.get('order_id')
        if not order_id:
            return "请提供需要退款的订单号，我来为您处理。"

        order = self.order_db.get(order_id)
        if not order:
            return f"未找到订单 {order_id}，请核实。"

        return (f"订单 {order_id} 退款申请已受理。\n"
                f"退款金额：¥{order['total']}\n"
                f"预计3-5个工作日原路返回，请注意查收。")

    async def _handle_product_inquiry(self, entities: Dict,
                                       ctx: ConversationContext) -> str:
        context = ctx.get_recent_context()
        return self.rag.answer(
            entities.get('query', ctx.history[-1].content),
            context
        )

    async def _handle_greeting(self, _entities: Dict,
                                _ctx: ConversationContext) -> str:
        return ("您好！我是智能客服小助手 🤖\n"
                "可以帮您：\n"
                "📦 查询订单物流\n"
                "💰 申请退款退货\n"
                "🛍️ 产品咨询\n"
                "请问有什么可以帮您的？")

    async def _handle_complaint(self, entities: Dict,
                                 ctx: ConversationContext) -> str:
        return ("非常抱歉给您带来了不好的体验 🙏\n"
                "已为您记录问题，我们的售后团队会在2小时内联系您。\n"
                "如需紧急处理，可拨打 400-XXX-XXXX。")

    async def _handle_unknown(self, _entities: Dict,
                               ctx: ConversationContext) -> str:
        context = ctx.get_recent_context()
        return self.rag.answer(ctx.history[-1].content, context)

    def _handle_transfer(self, reason: str) -> str:
        messages = {
            'user_request': "正在为您转接人工客服，请稍候...",
            'anger_detected': "非常理解您的心情，正在为您转接专属客服...",
            'max_rounds_exceeded': "您的问题比较复杂，正在转接人工客服为您处理...",
            'repeated_no_match': "正在为您转接人工客服，请稍候...",
        }
        return messages.get(reason, "正在转接人工客服...")


# === 启动服务 ===
if __name__ == "__main__":
    import uvicorn

    config = {
        'llm_api_key': 'your-api-key-here',
        'llm_base_url': None
    }

    system = Ecommerce客服系统(config)

    @app.post("/api/chat")
    async def chat(request: Request):
        data = await request.json()
        response = await system.handle_message(
            channel=data.get('channel', 'web'),
            user_id=data['user_id'],
            content=data['message']
        )
        return {"response": response, "timestamp": time.time()}

    uvicorn.run(app, host="0.0.0.0", port=8000)

十一、评估指标与持续优化

11.1 核心评估指标

11.2 持续优化闭环

class QualityMonitor:
    def __init__(self):
        self.metrics = {
            'total_sessions': 0,
            'resolved_sessions': 0,
            'transferred_sessions': 0,
            'intent_accuracy': [],
            'response_times': [],
            'csat_scores': []
        }

    def log_session(self, session_data: dict):
        self.metrics['total_sessions'] += 1
        if session_data.get('resolved'):
            self.metrics['resolved_sessions'] += 1
        if session_data.get('transferred'):
            self.metrics['transferred_sessions'] += 1
        if session_data.get('response_time'):
            self.metrics['response_times'].append(
                session_data['response_time']
            )

    def get_report(self) -> dict:
        total = self.metrics['total_sessions'] or 1
        return {
            'resolution_rate': self.metrics['resolved_sessions'] / total,
            'transfer_rate': self.metrics['transferred_sessions'] / total,
            'avg_response_time': (
                sum(self.metrics['response_times']) /
                max(len(self.metrics['response_times']), 1)
            ),
            'total_sessions': total
        }

    def identify_improvements(self) -> list:
        """自动发现优化点"""
        report = self.get_report()
        improvements = []

        if report['transfer_rate'] > 0.3:
            improvements.append(
                "转人工率过高(>{:.0%})，建议扩充FAQ知识库或优化意图识别模型".format(
                    report['transfer_rate']
                )
            )
        if report['avg_response_time'] > 2.0:
            improvements.append(
                "平均响应时间过长(>{:.1f}s)，建议优化检索策略或增加缓存".format(
                    report['avg_response_time']
                )
            )
        return improvements

11.3 优化方向

1. 数据驱动迭代：定期分析bad case，补充训练数据，重训NLU模型
2. A/B测试：对比不同回复模板/LLM prompt的效果
3. 知识库维护：根据高频未解决问题，持续补充FAQ条目
4. 对话流优化：缩短对话轮次，提升一次性解决率
5. 个性化：基于用户历史偏好定制回复风格和推荐策略

以上就是AI智能客服系统的完整开发方案。从架构设计到各模块实现，再到生产级的评估优化，覆盖了构建智能客服的全链路。根据实际业务需求选择合适的模块组合，快速搭建并持续迭代，才能打造真正好用的AI客服系统。