实现实时情感路由：客服团队技术指南

情感路由真正解决的问题

大多数客服团队将情感分析用作报告工具：了解客户感受、追踪 CSAT 趋势、标记需要审查的异常对话。这很有价值，但属于事后分析。

实时情感路由利用相同的信号，在对话进行中做出路由决策——在客户有机会流失、发布负面评价或向管理层升级之前介入。

本指南涵盖完整实施流程：模型选择、路由逻辑、集成模式及调优注意事项。

架构概览

Incoming Message
      ↓
Sentiment Model (inference < 50ms)
      ↓
Sentiment Score + Intensity + Trend
      ↓
Routing Decision Engine
      ↓
  ├── Negative + High Intensity + Escalating → Senior Agent Queue
  ├── Negative + Moderate + Stable → Standard Queue (priority bump)
  ├── Neutral/Positive → AI Handling or Standard Queue
  └── Repeated Negative across sessions → Retention Queue
      ↓
Assignment + Context Package

核心架构原则：情感是路由决策的输入，而非决策本身。在路由时，需将情感信号与对话历史、客户层级、问题类别及当前队列深度相结合。

第一步：模型选择

选项

基于词典的模型（VADER、SentiWordNet）：

速度快（< 1ms）
无需训练数据
对特定领域语言处理效果差（“this is broken” → 评分模糊不清）
适用场景：简单的正面/负面/中性分类

经过微调的 Transformer 模型（DistilBERT、RoBERTa，在客服数据上微调）：

CPU 推理耗时 10-50ms，GPU 下低于 5ms
需要 5,000+ 条标注样本
能良好处理特定领域语言
支持多分类输出（愤怒、沮丧、困惑、满意、中性）

LLM 即分类器（GPT-4o-mini、Claude Haiku，通过 API 调用）：

延迟 100-300ms
无需训练数据
对细微文本的识别准确率高
成本：每次对话约 $0.002
适用场景：低流量或高价值路由

我们的建议：实时路由使用经过微调的 DistilBERT（延迟 < 10ms，对客服语言识别准确），在低置信度输出时以 LLM 即分类器作为兜底方案。

需要训练的标签

标准的正面/负面/中性分类远远不够。需训练以下标签：

沮丧（frustrated） — 重复投诉、表达时间浪费
愤怒（angry） — 粗口、明确表示要取消或升级投诉
困惑（confused） — 反复提问、出现"我不理解"等表达模式
满意（satisfied） — 表示感谢、确认问题已解决
紧急（urgent） — 时间压力语言，如"今天必须解决"、“这阻塞了我的工作”

多标签对大多数消息是正确的做法（客户可以同时处于沮丧和紧急状态）。

第二步：实时推理集成

集成模式

对于基于 webhook 的客服平台：

import httpx
from fastapi import FastAPI, Request

app = FastAPI()
sentiment_client = httpx.AsyncClient(base_url="http://sentiment-service:8000")

@app.post("/webhook/message")
async def handle_message(request: Request):
    payload = await request.json()
    message_text = payload["message"]["body"]
    conversation_id = payload["conversation"]["id"]

    # Async sentiment inference — don't block the message receipt
    sentiment_task = asyncio.create_task(
        sentiment_client.post("/infer", json={"text": message_text})
    )

    # Enqueue message immediately
    await enqueue_message(payload)

    # Await sentiment result (target: < 50ms total)
    sentiment_response = await sentiment_task
    sentiment = sentiment_response.json()

    # Update routing with sentiment
    await update_routing_context(conversation_id, sentiment)
    await maybe_reroute(conversation_id, sentiment)

    return {"status": "accepted"}

关键设计决策：立即将消息入队，不要阻塞在情感推理上。推理完成后再更新路由上下文。若情感推理失败，则回退至标准路由。

推理服务（DistilBERT）

from transformers import pipeline
from fastapi import FastAPI

app = FastAPI()
classifier = pipeline(
    "text-classification",
    model="./models/support-sentiment-distilbert",
    return_all_scores=True,
    device=0  # GPU; use -1 for CPU
)

@app.post("/infer")
async def infer(body: dict):
    scores = classifier(body["text"], truncation=True, max_length=512)[0]
    labels = {s["label"]: round(s["score"], 3) for s in scores}
    primary = max(scores, key=lambda x: x["score"])
    return {
        "primary": primary["label"],
        "confidence": primary["score"],
        "scores": labels,
        "high_intensity": labels.get("angry", 0) > 0.6 or labels.get("frustrated", 0) > 0.7
    }

第三步：路由决策逻辑

async def maybe_reroute(conversation_id: str, sentiment: dict):
    conv = await get_conversation_context(conversation_id)
    current_queue = conv["queue"]
    customer_tier = conv["customer"]["tier"]
    sentiment_history = conv["sentiment_history"]  # list of last N sentiments

    # Escalation trigger: current message angry/frustrated AND intensity rising
    trend = calculate_trend(sentiment_history + [sentiment])
    if (
        sentiment["primary"] in ("angry", "frustrated")
        and sentiment["high_intensity"]
        and trend == "escalating"
        and current_queue != "senior"
    ):
        await reroute(conversation_id, "senior", reason="sentiment_escalation")
        await notify_agent(conv["assigned_agent_id"], "Conversation rerouted to senior queue — customer sentiment escalating")
        return

    # Priority bump: frustrated but not at escalation threshold
    if (
        sentiment["primary"] == "frustrated"
        and not sentiment["high_intensity"]
        and current_queue == "standard"
    ):
        await bump_priority(conversation_id, delta=2)
        return

    # Retention queue: Enterprise customer with repeated negative sessions
    if customer_tier == "enterprise":
        recent_sessions = await get_recent_session_sentiments(conv["customer"]["id"], days=30)
        negative_sessions = [s for s in recent_sessions if s["primary"] in ("angry", "frustrated")]
        if len(negative_sessions) >= 2 and current_queue != "retention":
            await reroute(conversation_id, "retention", reason="repeat_negative_sessions")
            await flag_for_csm_review(conv["customer"]["id"])

第四步：为接收坐席准备上下文包

当对话被重新路由时，接收坐席需要快速获取上下文。发送一个预先计算好的摘要：

{
  "routing_reason": "sentiment_escalation",
  "sentiment_summary": {
    "current": "angry (0.82 confidence)",
    "trend": "escalating over last 4 messages",
    "trigger_phrases": ["this is completely broken", "I've been waiting three days"]
  },
  "suggested_opener": "I can see this has been a frustrating experience. I'm taking over and I have full context — let's get this resolved right now.",
  "resolution_authority": ["refund_up_to_500", "extend_trial_7_days", "escalate_to_engineering"]
}

解决权限列表至关重要。若一位沮丧的客户被路由至无法实际解决其问题的高级坐席，效果还不如不路由。