第 22 章

实战:智能客服系统——意图识别、多轮对话与人工接管

第22章:实战——智能客服系统——意图识别、多轮对话与人工接管

一套真正能用的智能客服,不是用 AI 堵住用户,而是让用户在合适的时机遇到合适的帮助——无论是 AI 还是人工。

本章导读

智能客服是 Dify 落地最多的场景,也是踩坑最多的场景。很多团队以为"接入 LLM + FAQ 知识库 = 智能客服",结果上线后被用户骂得狗血淋头。

真正的生产级智能客服需要解决三个核心挑战:

  1. 精准意图识别:用户说"我的东西坏了",系统要判断出是"售后维修申请"而不是"产品差评"
  2. 多轮对话管理:跨越 10 轮以上对话,记住用户说过的所有信息,不让用户重复自己
  3. 无缝人工接管:AI 处理不了时,能把完整的对话上下文交给人工客服,而不是让用户从头再说

案例背景:某家居电商(年 GMV 3 亿)的客服中心,每日客服会话量约 8,000 次,客服团队 30 人,高峰期(促销节)日会话量可达 25,000 次。上线 AI 客服目标:


Level 1:基础认知(1-3 年经验)

客服场景的特殊性

与内部知识助理不同,对外客服有几个特殊挑战:

挑战1:用户不会好好说话

内部员工提问:"请假需要提前多少天申请?"(结构清晰)

外部用户提问:"我明天想休息,行吗" / "你们这破系统" / "给我说说那个规定" / "???"

AI 必须处理歧义、口语、情绪化表达、不完整信息。

挑战2:意图多且复杂

电商客服的常见意图(不完全列表):

售前类:
- 产品咨询(材质、尺寸、颜色、库存)
- 价格咨询(是否有优惠、比价)
- 配送咨询(支持哪些地区、多久到)
- 安装咨询(是否包安装、费用)

售后类:
- 订单查询(物流状态、发货情况)
- 退换货(流程、条件、进度)
- 维修申请(如何申请、费用)
- 投诉(产品质量、服务态度)
- 好评追加(催评)

账户类:
- 密码/登录问题
- 积分查询/兑换
- 地址修改

挑战3:系统集成复杂

客服系统需要对接:订单系统、物流系统、CRM、工单系统、人工座席平台。每个系统都是一个"工具",需要 Dify Agent 来调用。

整体架构设计

用户端(Web/APP/微信小程序)
         │
         ▼
  消息网关层(WebSocket/HTTP)
         │
         ▼
   Dify AI 客服应用
   ┌─────────────────────────────────────┐
   │  意图识别模块(Workflow)            │
   │       ↓                            │
   │  多轮对话引擎(Chatbot Agent)       │
   │    ├── 售前 FAQ 知识库              │
   │    ├── 售后 FAQ 知识库              │
   │    └── 工具调用:                  │
   │         ├── 查询订单                │
   │         ├── 查询物流                │
   │         ├── 提交退款申请            │
   │         └── 创建工单               │
   │  人工接管触发器                     │
   └─────────────────────────────────────┘
         │                    │
         ▼                    ▼
   AI 自动回复           转人工客服平台
                        (含完整对话记录)

快速搭建最简版本

步骤一:创建聊天助手应用

在 Dify Console 中:

  1. 新建应用 → 聊天助手
  2. 选择模型:GPT-4 Turbo(客服需要强理解能力)
  3. 开启"对话历史":保留最近 10 轮
  4. 关联知识库:售前 FAQ、售后 FAQ

步骤二:配置基础 System Prompt

你是[品牌名]的客服助理小美。

## 你的能力:
- 解答产品相关问题(材质、尺寸、使用方法)
- 查询订单状态(需要用户提供订单号)
- 处理退换货申请
- 解决常见售后问题

## 你的原则:
1. 始终友好、耐心,即使用户情绪激动
2. 不确定时主动询问,而不是猜测
3. 不能解决的问题,及时告知将转接人工客服
4. 不承诺具体的赔偿金额(需人工审核)
5. 对投诉类问题,先表示歉意,再解决问题

## 你的限制:
- 只处理与[品牌名]产品和服务相关的问题
- 不提供竞品推荐
- 不发表产品以外的政治、社会评论

步骤三:发布并集成

在 Dify 的"发布"页面:

  1. 启用 API 访问
  2. 复制 API 密钥和端点
  3. 在您的客服系统中集成(见后续代码示例)

Level 2:机制深解(3-5 年经验)

意图识别的精准实现

单纯依靠 LLM 做意图识别存在问题:每次对话都要做意图识别,成本高、延迟高。

推荐方案:分层意图识别

第一层(规则):正则 + 关键词匹配
  - 覆盖 40% 的高频明确意图(如含"退款"→退款意图)
  - 响应时间 < 1ms,成本为零

第二层(小模型):BERT/ALBERT 分类器
  - 覆盖 85% 的意图
  - 响应时间 < 50ms,成本极低

第三层(大模型):GPT-3.5/Claude Haiku
  - 处理剩余 15% 的复杂/歧义意图
  - 响应时间 500ms-1s,成本较高

Dify Workflow 实现意图识别

# intent_recognition_workflow.yaml(Dify Workflow 配置伪代码)

nodes:
  # 节点1:规则层意图识别
  - id: rule_intent
    type: code
    language: python
    code: |
      import re
      
      INTENT_PATTERNS = {
          'refund': [r'退款', r'退货', r'不想要了', r'申请退'],
          'order_query': [r'订单', r'物流', r'快递', r'发货', r'到货'],
          'repair': [r'坏了', r'损坏', r'维修', r'故障', r'修'],
          'complaint': [r'投诉', r'差评', r'太差了', r'骗人'],
          'product_inquiry': [r'什么材质', r'尺寸', r'颜色', r'型号'],
      }
      
      user_input = inputs['user_input']
      detected_intent = None
      confidence = 0.0
      
      for intent, patterns in INTENT_PATTERNS.items():
          for pattern in patterns:
              if re.search(pattern, user_input):
                  detected_intent = intent
                  confidence = 0.95  # 规则匹配高置信度
                  break
          if detected_intent:
              break
      
      return {
          'intent': detected_intent,
          'confidence': confidence,
          'method': 'rule'
      }
    outputs:
      intent: string
      confidence: number
      method: string

  # 节点2:判断是否需要 LLM 识别
  - id: need_llm_intent
    type: if_else
    condition: "{{rule_intent.intent}} == null or {{rule_intent.confidence}} < 0.8"
    
  # 节点3:LLM 意图识别
  - id: llm_intent
    type: llm
    model: gpt-3.5-turbo
    prompt: |
      从以下列表中选择最匹配用户意图的类别,只输出类别名称:
      
      类别列表:
      - refund(退款/退货申请)
      - order_query(订单/物流查询)
      - repair(维修申请)
      - complaint(投诉)
      - product_inquiry(产品咨询)
      - account(账户问题)
      - price(价格/优惠咨询)
      - delivery(配送咨询)
      - other(其他)
      
      用户说:"{{user_input}}"
      
      意图:
    output_variable: llm_detected_intent
  
  # 节点4:汇总意图结果
  - id: merge_intent
    type: code
    code: |
      intent = inputs['rule_intent'] or inputs['llm_detected_intent']
      return {'final_intent': intent}

多轮对话状态管理

多轮对话的核心挑战是:如何在多轮中收集完成任务所需的所有信息?

退款流程的信息收集示例

退款需要收集:订单号、退款原因、退款方式偏好(原路退回/退到余额)。

这 3 个信息可能分散在 3-5 轮对话中。Dify 通过变量(Variables)来追踪状态:

# Dify 应用中的变量定义
variables = {
    # 对话状态变量
    "refund_order_id": "",        # 退款订单号
    "refund_reason": "",          # 退款原因
    "refund_method": "",          # 退款方式
    "refund_step": "initial",     # 当前步骤
    "user_emotion": "neutral",    # 用户情绪状态
    "escalation_triggered": False # 是否触发了人工接管
}

槽位填充(Slot Filling)System Prompt

你是退款处理助理。请按以下步骤引导用户完成退款申请:

## 需要收集的信息(槽位):
1. 订单号 [已获取: {{refund_order_id}}]
2. 退款原因 [已获取: {{refund_reason}}]  
3. 退款方式 [已获取: {{refund_method}}]

## 当前步骤:{{refund_step}}

## 指令:
- 如果某个槽位为空,礼貌地询问该信息
- 不要一次询问所有信息,一次只问一个
- 用户提供信息后,确认并更新对应槽位
- 所有槽位填满后,输出结构化的退款申请并调用"提交退款"工具

## 重要:
- 用户情绪激动时({{user_emotion}} == "angry"),先安抚后收集信息
- 退款金额超过500元时,标记需要人工审核

情绪识别与智能升级策略

# emotion_detector.py — 情绪识别模块

from openai import OpenAI

client = OpenAI()

EMOTION_LABELS = ['positive', 'neutral', 'frustrated', 'angry', 'very_angry']

def detect_emotion(user_message: str, conversation_history: list) -> dict:
    """检测用户情绪,决定是否需要人工介入"""
    
    # 快速规则检测(避免每次都调用 LLM)
    angry_keywords = ['气死了', '无语', '太差了', '骗人', '投诉', '12315', '曝光', '律师']
    urgent_keywords = ['急', '马上', '今天必须', '赶紧']
    
    for keyword in angry_keywords:
        if keyword in user_message:
            return {
                'emotion': 'very_angry',
                'score': 0.95,
                'escalation_recommended': True,
                'reason': f'触发关键词: {keyword}'
            }
    
    # 分析连续负面情绪
    if len(conversation_history) >= 4:
        recent_messages = [
            msg['content'] for msg in conversation_history[-4:]
            if msg['role'] == 'user'
        ]
        # 如果最近2条用户消息都很简短且负面,可能是挫败感
        if all(len(msg) < 20 for msg in recent_messages[-2:]):
            return {
                'emotion': 'frustrated',
                'score': 0.7,
                'escalation_recommended': False,
                'reason': '用户回复越来越简短,可能不耐烦'
            }
    
    return {'emotion': 'neutral', 'score': 0.5, 'escalation_recommended': False}


def should_escalate(context: dict) -> tuple[bool, str]:
    """判断是否应该转人工客服"""
    
    reasons = []
    
    # 规则1:用户明确要求人工
    if any(kw in context['last_user_message'] for kw in ['人工', '真人', '客服', '转接']):
        reasons.append('用户主动要求转人工')
    
    # 规则2:情绪严重
    if context['user_emotion'] in ['angry', 'very_angry']:
        reasons.append('用户情绪激动')
    
    # 规则3:AI 连续 3 次无法解决
    if context['ai_failure_count'] >= 3:
        reasons.append('AI 连续 3 次未能解决问题')
    
    # 规则4:金额超过阈值
    if context.get('refund_amount', 0) > 500:
        reasons.append('退款金额超过 500 元,需人工审核')
    
    # 规则5:敏感类型
    if context.get('issue_type') in ['legal', 'media', 'vip_complaint']:
        reasons.append('高优先级投诉类型')
    
    # 规则6:对话轮次过多
    if context.get('turn_count', 0) > 15:
        reasons.append('对话轮次超过 15 轮,问题可能复杂')
    
    if reasons:
        return True, ' + '.join(reasons)
    return False, ''

人工接管:无缝交接协议

人工接管是智能客服的关键体验节点。差的实现:AI 说"请稍等,为您转接人工",然后用户要重新说一遍所有问题。

好的实现:人工接管时,座席看到的是完整的对话摘要和结构化的问题描述。

# handover.py — 人工接管交接协议

def generate_handover_context(conversation: dict) -> dict:
    """生成交给人工客服的结构化交接信息"""
    
    # 用 LLM 从对话历史中提取关键信息
    extraction_prompt = f"""
    从以下客服对话中提取关键信息,输出 JSON 格式:
    
    对话历史:
    {format_conversation(conversation['history'])}
    
    请提取:
    {{
        "user_name": "用户称谓(如有)",
        "contact_info": "联系方式(如有)",
        "order_id": "订单号(如有)",
        "product_name": "产品名称",
        "issue_summary": "问题简述(2句话以内)",
        "user_requests": ["用户的具体诉求列表"],
        "steps_taken": ["AI 已尝试的解决步骤"],
        "blockers": ["为什么 AI 无法解决"],
        "urgency": "高/中/低",
        "user_emotion": "情绪状态",
        "recommended_action": "建议人工客服的处理方向"
    }}
    """
    
    handover_info = llm.extract(extraction_prompt)
    
    return {
        "handover_time": datetime.now().isoformat(),
        "conversation_id": conversation['id'],
        "ai_session_duration": conversation['duration_seconds'],
        "turn_count": len(conversation['history']) // 2,
        "escalation_reason": conversation['escalation_reason'],
        "structured_info": handover_info,
        "full_history_url": f"/conversations/{conversation['id']}"
    }


def push_to_agent_platform(handover_context: dict, platform: str = 'zendesk') -> str:
    """将交接信息推送到人工客服平台"""
    
    if platform == 'zendesk':
        # Zendesk API
        ticket = {
            "ticket": {
                "subject": f"AI转人工 - {handover_context['structured_info']['issue_summary']}",
                "comment": {
                    "body": format_handover_for_agent(handover_context)
                },
                "priority": map_urgency_to_priority(
                    handover_context['structured_info']['urgency']
                ),
                "tags": ["ai_escalation", f"emotion_{handover_context['structured_info']['user_emotion']}"]
            }
        }
        # 调用 Zendesk API 创建工单
        response = zendesk_client.create_ticket(ticket)
        return response['ticket']['id']
    
    elif platform == 'custom_crm':
        # 自定义 CRM 系统
        return crm_client.create_session(handover_context)

def format_handover_for_agent(context: dict) -> str:
    """将交接信息格式化为人工客服易读的格式"""
    info = context['structured_info']
    
    return f"""
═══════════════════════════════════════
【AI 智能客服交接记录】
时间:{context['handover_time']}
对话 ID:{context['conversation_id']}
对话轮数:{context['turn_count']} 轮
交接原因:{context['escalation_reason']}
═══════════════════════════════════════

【用户基本信息】
• 称谓:{info.get('user_name', '未获取')}
• 联系方式:{info.get('contact_info', '未获取')}

【订单信息】
• 订单号:{info.get('order_id', '未获取')}
• 产品:{info.get('product_name', '未知')}

【问题描述】
{info.get('issue_summary')}

【用户诉求】
{chr(10).join('• ' + r for r in info.get('user_requests', []))}

【已尝试的解决方案】
{chr(10).join('• ' + s for s in info.get('steps_taken', []))}

【建议处理方向】
{info.get('recommended_action')}

【用户情绪】
{info.get('user_emotion')} | 紧迫度:{info.get('urgency')}

【完整对话记录】
{context['full_history_url']}
═══════════════════════════════════════
"""

Level 3:源码与原理(5 年以上)

高并发架构设计

促销期间(双11、618),日会话量从 8,000 激增到 25,000,并发会话数峰值可达 5,000。

瓶颈分析

用户请求 → WebSocket 服务 → Dify API → OpenAI API → 返回

瓶颈1:OpenAI API 并发限制(Tier 3:10,000 RPM)
瓶颈2:Dify API 单节点能力(约 500 并发连接)
瓶颈3:会话状态存储(Redis 内存)
瓶颈4:WebSocket 长连接数(单节点约 10,000 连接)

高并发解决方案

# async_customer_service.py — 异步高并发架构

import asyncio
import aiohttp
import redis.asyncio as aioredis
from typing import AsyncGenerator

# 使用异步 Redis 存储会话状态
redis_client = aioredis.from_url("redis://redis:6379")

# 创建连接池(限制并发 API 调用数)
DIFY_CONNECTION_POOL = aiohttp.TCPConnector(
    limit=200,        # 最多 200 个并发连接
    limit_per_host=200,
    keepalive_timeout=60
)

async def handle_chat_message(
    session_id: str,
    user_message: str,
    user_id: str
) -> AsyncGenerator[str, None]:
    """
    异步流式处理客服消息
    使用 AsyncGenerator 支持 SSE 流式输出
    """
    
    # 从 Redis 获取会话状态
    session_data = await redis_client.get(f"session:{session_id}")
    if session_data:
        session = json.loads(session_data)
    else:
        session = {
            "conversation_id": None,
            "turn_count": 0,
            "user_emotion": "neutral",
            "ai_failure_count": 0
        }
    
    # 情绪检测(异步并行执行,不阻塞主流程)
    emotion_task = asyncio.create_task(
        detect_emotion_async(user_message, session.get('history', []))
    )
    
    # 调用 Dify 流式 API
    async with aiohttp.ClientSession(connector=DIFY_CONNECTION_POOL) as http_session:
        async with http_session.post(
            f"{DIFY_API_URL}/chat-messages",
            headers={"Authorization": f"Bearer {DIFY_APP_TOKEN}"},
            json={
                "query": user_message,
                "response_mode": "streaming",
                "conversation_id": session.get("conversation_id") or "",
                "inputs": {
                    "user_emotion": session.get("user_emotion"),
                    "turn_count": str(session.get("turn_count", 0))
                },
                "user": user_id
            }
        ) as response:
            # 流式读取响应
            full_response = ""
            async for line in response.content:
                line_str = line.decode('utf-8').strip()
                if line_str.startswith('data: '):
                    data = json.loads(line_str[6:])
                    if data.get('event') == 'message':
                        chunk = data.get('answer', '')
                        full_response += chunk
                        yield chunk  # 流式输出给用户
                    elif data.get('event') == 'message_end':
                        # 更新会话 ID
                        session['conversation_id'] = data.get('conversation_id')
    
    # 等待情绪检测结果
    emotion_result = await emotion_task
    session['user_emotion'] = emotion_result['emotion']
    session['turn_count'] = session.get('turn_count', 0) + 1
    
    # 判断是否需要转人工
    should_transfer, reason = should_escalate({
        'last_user_message': user_message,
        'user_emotion': session['user_emotion'],
        'ai_failure_count': session.get('ai_failure_count', 0),
        'turn_count': session['turn_count']
    })
    
    if should_transfer:
        # 触发人工接管流程
        yield "\n\n[SYSTEM:ESCALATE]"  # 特殊标记,前端识别后显示转人工 UI
        await trigger_escalation(session_id, session, reason)
    
    # 保存会话状态(TTL 24 小时)
    await redis_client.setex(
        f"session:{session_id}",
        86400,
        json.dumps(session)
    )

WebSocket 服务(使用 FastAPI + WebSocket)

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict

app = FastAPI()

# 活跃连接管理
active_connections: Dict[str, WebSocket] = {}

@app.websocket("/ws/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: str):
    await websocket.accept()
    active_connections[session_id] = websocket
    
    try:
        while True:
            # 接收用户消息
            data = await websocket.receive_json()
            user_message = data.get('message', '')
            user_id = data.get('user_id', session_id)
            
            # 流式处理并推送
            try:
                async for chunk in handle_chat_message(session_id, user_message, user_id):
                    if chunk == "[SYSTEM:ESCALATE]":
                        # 发送人工接管通知
                        await websocket.send_json({
                            "type": "escalation",
                            "message": "正在为您转接人工客服,请稍候..."
                        })
                    else:
                        await websocket.send_json({
                            "type": "message_chunk",
                            "content": chunk
                        })
                
                # 发送消息完成信号
                await websocket.send_json({"type": "message_end"})
                
            except Exception as e:
                await websocket.send_json({
                    "type": "error",
                    "message": "抱歉,处理您的消息时遇到问题,正在为您转接人工客服"
                })
                await trigger_escalation(session_id, {}, str(e))
    
    except WebSocketDisconnect:
        del active_connections[session_id]

工具调用:订单系统集成

在 Dify 中配置自定义工具,让 AI 能直接查询业务系统:

# Dify 自定义工具定义(OpenAPI 格式)
openapi: 3.0.0
info:
  title: 电商订单查询 API
  version: 1.0.0
paths:
  /orders/{order_id}:
    get:
      summary: 查询订单详情
      operationId: getOrderDetails
      parameters:
        - name: order_id
          in: path
          required: true
          schema:
            type: string
          description: 订单号,格式如 2024030812345678
      responses:
        '200':
          description: 订单详情
          content:
            application/json:
              schema:
                type: object
                properties:
                  order_id:
                    type: string
                  status:
                    type: string
                    enum: [pending, paid, shipped, delivered, refunding, refunded]
                  product_name:
                    type: string
                  amount:
                    type: number
                  shipping_info:
                    type: object
                    properties:
                      carrier:
                        type: string
                      tracking_number:
                        type: string
                      estimated_delivery:
                        type: string
        '404':
          description: 订单不存在

  /orders/{order_id}/refund:
    post:
      summary: 申请退款
      operationId: applyRefund
      parameters:
        - name: order_id
          in: path
          required: true
          schema:
            type: string
      requestBody:
        required: true
        content:
          application/json:
            schema:
              type: object
              required: [reason]
              properties:
                reason:
                  type: string
                  description: 退款原因
                method:
                  type: string
                  enum: [original_payment, balance]
                  description: 退款方式
      responses:
        '200':
          description: 申请成功
# order_api.py — 工具 API 实现(需要做权限隔离)

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer

app = FastAPI()
security = HTTPBearer()

@app.get("/orders/{order_id}")
async def get_order(order_id: str, token: str = Depends(security)):
    """查询订单 API — 供 Dify 工具调用"""
    
    # 验证 Dify 调用令牌
    if not validate_dify_token(token.credentials):
        raise HTTPException(status_code=401)
    
    # 从 ERP 系统查询订单(隐藏真实 API 细节)
    order = await erp_client.get_order(order_id)
    if not order:
        raise HTTPException(status_code=404, detail="订单不存在")
    
    # 返回脱敏的订单信息(不返回完整支付信息)
    return {
        "order_id": order.id,
        "status": order.status,
        "product_name": order.product_name,
        "amount": order.amount,
        "shipping_info": {
            "carrier": order.shipping_carrier,
            "tracking_number": order.tracking_number,
            "estimated_delivery": order.estimated_delivery
        }
    }

Level 4:生产陷阱与决策(专家视角)

陷阱1:AI 过度道歉导致用户更愤怒

症状:AI 对每个问题都说"非常抱歉给您带来不便",用户会说"你只会道歉,能不能解决问题!"

根因:System Prompt 过度强调"友好""道歉",AI 把道歉当成了万能公式。

解决:在 System Prompt 中限制道歉频率:

## 道歉原则:
- 同一次对话中,道歉词("抱歉"/"对不起"/"非常遗憾")最多使用 2 次
- 道歉之后必须立即给出解决方案或下一步行动
- 不要在问候语中包含道歉(如不要说"您好,很抱歉...")

陷阱2:工具调用失败时 AI 编造信息

症状:查订单 API 超时,AI 直接编造了一个物流信息告诉用户。

根因:没有严格限制 AI 在工具失败时的行为。

解决:

## 工具调用失败处理:
当查询工具返回错误或超时时:
1. 明确告知用户"系统查询超时,暂时无法获取信息"
2. 提供手动查询链接或电话
3. 绝对不要猜测或编造查询结果
4. 如果 2 次重试均失败,主动转人工

陷阱3:多轮对话中槽位污染

症状:用户先问了 A 订单的退款,然后聊了其他话题,再问退款时,AI 继续用的是 A 订单的信息。

根因:Dify 的变量(Variables)在整个会话生命周期内持久,没有清零机制。

解决:在意图切换时显式清零相关槽位:

# 检测到新的退款意图时,清零退款相关变量
def handle_intent_switch(new_intent: str, current_intent: str, variables: dict) -> dict:
    """意图切换时清理相关变量"""
    
    if new_intent != current_intent:
        # 退款意图的相关槽位
        if new_intent == 'refund':
            variables['refund_order_id'] = ''
            variables['refund_reason'] = ''
            variables['refund_method'] = ''
            variables['refund_step'] = 'initial'
        
        # 更新当前意图
        variables['current_intent'] = new_intent
    
    return variables

陷阱4:人工接管后 AI 重新接管

症状:用户转到人工后,还在输入消息,AI 自动又回复了,和人工客服同时回复,用户懵了。

解决:人工接管后,会话必须加锁:

async def check_human_takeover(session_id: str) -> bool:
    """检查会话是否已转人工"""
    status = await redis_client.get(f"session:{session_id}:status")
    return status == b"human_agent"

@app.websocket("/ws/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: str):
    await websocket.accept()
    
    while True:
        data = await websocket.receive_json()
        
        # 关键:每次收到消息前检查是否已转人工
        if await check_human_takeover(session_id):
            # 不调用 AI,只转发给人工客服平台
            await forward_to_human_agent(session_id, data['message'])
            continue
        
        # 正常 AI 处理流程
        async for chunk in handle_chat_message(session_id, data['message'], data['user_id']):
            await websocket.send_json({"type": "chunk", "content": chunk})

关键指标与成功标准

指标 目标值 行业平均 该案例实际
AI 自助解决率 ≥ 70% 45-55% 68%(第3月)
转人工后等待 < 2 分钟 5-8 分钟 1.5 分钟
用户满意度(CSAT) ≥ 4.2/5 3.8/5 4.1/5
平均对话轮次 < 6 轮 8-10 轮 5.3 轮
意图识别准确率 ≥ 90% 75-80% 93%
峰值并发支持 5000 - 4800(618期间实测)

本章小结

核心设计原则

  1. 分层意图识别:规则 → 小模型 → 大模型,兼顾速度和精准,70% 的意图应由规则层解决。

  2. 情绪驱动的升级策略:检测到愤怒情绪应立即触发转人工,不要让 AI 强行"安慰"愤怒的用户。

  3. 无缝交接是体验关键:人工接管时必须传递完整的结构化上下文,让座席"接着打"而不是"从头来"。

  4. 会话状态要有锁机制:人工接管后 AI 必须停止回复,否则会造成双重回复的混乱。

  5. 工具失败要有降级策略:API 超时时不能让 AI 猜测或编造,必须明确告知用户并提供备用方案。

  6. 持续监控意图准确率:每周抽样检查意图识别结果,发现偏差及时调整规则和 Prompt。

本章评分
4.5  / 5  (7 评分)

💬 留言讨论