14e - 延迟优化与多语言

┌─────────────────────────────────────────────────────────────────────┐
│                    语音 Agent 延迟分解                                │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  用户说话 ──→ [VAD] ──→ [STT] ──→ [LLM] ──→ [TTS] ──→ 用户听到    │
│              10-50ms   100-500ms  350ms-1s+  75-200ms               │
│                                                                     │
│  网络传输延迟：每个环节 +20-100ms（取决于地理位置）                    │
│  音频编解码：+10-30ms                                                │
│                                                                     │
│  典型总延迟：800ms - 2s+                                             │
│  目标延迟：< 300ms（端到端）                                         │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

传统模式（顺序）：
  STT 完成 ──→ LLM 完成 ──→ TTS 完成 ──→ 播放
  [====300ms====][====500ms====][====200ms====]
  总延迟：1000ms

流式模式（并行）：
  STT 流式输出 ──→ LLM 流式生成 ──→ TTS 流式合成 ──→ 边合成边播放
  [==100ms==]
       [==首token 150ms==]
            [==首音频块 75ms==]
            [播放▶]
  首音频延迟：~325ms

import asyncio
import websockets
import json
from typing import AsyncGenerator
 
class StreamingVoicePipeline:
    """流式语音 Agent 管线 - 实现 STT → LLM → TTS 全链路流式处理"""
 
    def __init__(self, stt_client, llm_client, tts_client):
        self.stt = stt_client
        self.llm = llm_client
        self.tts = tts_client
        self.audio_buffer = asyncio.Queue()
 
    async def process_audio_stream(
        self, audio_chunks: AsyncGenerator[bytes, None]
    ) -> AsyncGenerator[bytes, None]:
        """
        全流式处理：音频输入 → 文字流 → LLM 流 → 音频流输出
        关键：每个环节都是流式的，不等待完整结果
        """
        # 1. 流式 STT：音频块 → 文字片段
        transcript_stream = self.stt.stream_transcribe(audio_chunks)
 
        # 2. 收集完整用户输入（等待 VAD 端点检测）
        full_transcript = ""
        async for partial in transcript_stream:
            full_transcript += partial.text
            if partial.is_final:
                break
 
        # 3. 流式 LLM：文字 → token 流
        llm_stream = self.llm.stream_chat(
            messages=[{"role": "user", "content": full_transcript}],
            max_tokens=150,  # 限制输出长度以控制延迟
        )
 
        # 4. 流式 TTS：token 流 → 音频块流（关键优化点）
        text_buffer = ""
        async for token in llm_stream:
            text_buffer += token.content
            # 按句子边界分割，积累到自然断句再合成
            if self._is_sentence_boundary(text_buffer):
                async for audio_chunk in self.tts.stream_synthesize(text_buffer):
                    yield audio_chunk
                text_buffer = ""
 
        # 处理剩余文本
        if text_buffer.strip():
            async for audio_chunk in self.tts.stream_synthesize(text_buffer):
                yield audio_chunk
 
    def _is_sentence_boundary(self, text: str) -> bool:
        """检测句子边界 - 支持中英文标点"""
        boundaries = ["。", "！", "？", "；", ".", "!", "?", ";", "\n"]
        return any(text.rstrip().endswith(b) for b in boundaries)

import WebSocket from "ws";
 
interface TTSStreamOptions {
  apiKey: string;
  voiceId: string;
  model?: string; // "sonic-3" | "eleven_flash_v2" 等
  sampleRate?: number;
}
 
class StreamingTTSClient {
  private ws: WebSocket | null = null;
  private audioQueue: Buffer[] = [];
  private options: TTSStreamOptions;
 
  constructor(options: TTSStreamOptions) {
    this.options = { sampleRate: 24000, ...options };
  }
 
  /** 建立 WebSocket 持久连接 - 避免每次请求的连接开销 */
  async connect(): Promise<void> {
    return new Promise((resolve, reject) => {
      this.ws = new WebSocket(
        `wss://api.cartesia.ai/tts/websocket?api_key=${this.options.apiKey}`
      );
      this.ws.on("open", () => {
        console.log("[TTS] WebSocket 连接已建立");
        resolve();
      });
      this.ws.on("error", reject);
    });
  }
 
  /**
   * 流式合成：发送文本，逐块接收音频
   * 关键优化：使用 WebSocket 保持连接，省去 TLS 握手开销（~50-100ms）
   */
  async *synthesizeStream(text: string): AsyncGenerator<Buffer> {
    if (!this.ws || this.ws.readyState !== WebSocket.OPEN) {
      await this.connect();
    }
 
    const requestId = crypto.randomUUID();
 
    // 发送合成请求
    this.ws!.send(
      JSON.stringify({
        context_id: requestId,
        model_id: this.options.model ?? "sonic-3",
        transcript: text,
        voice: { mode: "id", id: this.options.voiceId },
        output_format: {
          container: "raw",
          encoding: "pcm_s16le",
          sample_rate: this.options.sampleRate,
        },
        // 启用流式输出 - 不等完整音频生成
        language: "zh",
      })
    );
 
    // 逐块接收音频数据
    for await (const message of this.receiveMessages(requestId)) {
      if (message.type === "chunk" && message.data) {
        yield Buffer.from(message.data, "base64");
      }
      if (message.type === "done") break;
    }
  }
 
  private async *receiveMessages(
    requestId: string
  ): AsyncGenerator<{ type: string; data?: string }> {
    const messageQueue: { type: string; data?: string }[] = [];
    let resolve: (() => void) | null = null;
 
    const handler = (raw: WebSocket.Data) => {
      const msg = JSON.parse(raw.toString());
      if (msg.context_id === requestId) {
        messageQueue.push(msg);
        resolve?.();
      }
    };
 
    this.ws!.on("message", handler);
 
    try {
      while (true) {
        if (messageQueue.length > 0) {
          const msg = messageQueue.shift()!;
          yield msg;
          if (msg.type === "done") return;
        } else {
          await new Promise<void>((r) => (resolve = r));
        }
      }
    } finally {
      this.ws!.off("message", handler);
    }
  }
}
 
// 使用示例
async function main() {
  const tts = new StreamingTTSClient({
    apiKey: process.env.CARTESIA_API_KEY!,
    voiceId: "your-voice-id",
    model: "sonic-3",
  });
 
  await tts.connect();
 
  // 流式合成中文语音
  for await (const audioChunk of tts.synthesizeStream(
    "您好，我是您的智能客服助手，请问有什么可以帮您？"
  )) {
    // 实时播放或通过 WebRTC/WebSocket 发送给客户端
    process.stdout.write(`[音频块: ${audioChunk.length} bytes] `);
  }
}

┌─────────────────────────────────────────────────────────────────┐
│                  SLM + LLM 并行架构                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  用户输入 ──┬──→ [SLM: Llama 3.1 8B on Groq]                   │
│             │     ~50ms 生成快速回复                              │
│             │     → 立即开始 TTS 合成                             │
│             │                                                    │
│             └──→ [LLM: GPT-4o / Claude 3.5]                    │
│                   ~300ms 生成高质量回复                           │
│                   → 如果 SLM 回复不够好，用 LLM 结果替换          │
│                                                                 │
│  决策逻辑：                                                      │
│  - 简单问候/确认 → 直接用 SLM 结果                               │
│  - 复杂问题 → 等 LLM 结果（SLM 先说"让我想想..."）              │
│  - 质量检查 → 对比两个结果，选择更好的                            │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

import asyncio
from dataclasses import dataclass
from enum import Enum
 
class ResponseQuality(Enum):
    FILLER = "filler"       # 填充语（"好的，让我查一下"）
    QUICK = "quick"         # SLM 快速回复
    FULL = "full"           # LLM 完整回复
 
@dataclass
class DualModelResponse:
    text: str
    quality: ResponseQuality
    latency_ms: float
 
class ParallelLLMRouter:
    """SLM + LLM 并行路由器 - 用快速模型填充等待时间"""
 
    def __init__(self, slm_client, llm_client):
        self.slm = slm_client  # Groq Llama 3.1 8B（~50ms）
        self.llm = llm_client  # GPT-4o / Claude 3.5（~300ms）
        # 预定义的填充语，无需模型生成
        self.fillers = {
            "zh": ["好的，让我查一下", "稍等，我来帮您处理", "明白了，请稍候"],
            "en": ["Let me check that for you", "One moment please", "Sure, looking into it"],
        }
 
    async def get_response(
        self, user_input: str, language: str = "zh"
    ) -> AsyncGenerator[DualModelResponse, None]:
        """
        并行调用 SLM 和 LLM，按延迟优先级返回结果：
        1. 立即返回填充语（0ms）
        2. SLM 结果就绪时返回（~50ms）
        3. LLM 结果就绪时返回（~300ms）
        """
        import time
        start = time.monotonic()
 
        # 判断是否需要 LLM（简单问题直接用 SLM）
        needs_llm = self._needs_complex_reasoning(user_input)
 
        # 并行启动两个模型
        slm_task = asyncio.create_task(self._call_slm(user_input))
 
        if needs_llm:
            llm_task = asyncio.create_task(self._call_llm(user_input))
            # 先返回填充语
            filler = self._get_filler(language)
            yield DualModelResponse(
                text=filler,
                quality=ResponseQuality.FILLER,
                latency_ms=0,
            )
 
        # 等待 SLM 结果
        slm_result = await slm_task
        slm_latency = (time.monotonic() - start) * 1000
 
        if not needs_llm:
            # 简单问题，直接用 SLM 结果
            yield DualModelResponse(
                text=slm_result,
                quality=ResponseQuality.QUICK,
                latency_ms=slm_latency,
            )
            return
 
        # 复杂问题，等待 LLM 结果
        llm_result = await llm_task
        llm_latency = (time.monotonic() - start) * 1000
 
        yield DualModelResponse(
            text=llm_result,
            quality=ResponseQuality.FULL,
            latency_ms=llm_latency,
        )
 
    def _needs_complex_reasoning(self, text: str) -> bool:
        """简单启发式判断是否需要 LLM"""
        simple_patterns = ["你好", "谢谢", "再见", "好的", "是的", "不是"]
        return not any(p in text for p in simple_patterns)
 
    def _get_filler(self, lang: str) -> str:
        import random
        return random.choice(self.fillers.get(lang, self.fillers["en"]))
 
    async def _call_slm(self, text: str) -> str:
        return await self.slm.chat(text, max_tokens=50)
 
    async def _call_llm(self, text: str) -> str:
        return await self.llm.chat(text, max_tokens=200)

# 连接池最佳实践
import aiohttp
from contextlib import asynccontextmanager
 
class VoiceAgentConnectionPool:
    """语音 Agent 连接池 - 预建立所有服务连接"""
 
    def __init__(self):
        self.stt_session: aiohttp.ClientSession | None = None
        self.llm_session: aiohttp.ClientSession | None = None
        self.tts_ws: WebSocket | None = None
 
    async def warmup(self):
        """预热所有连接 - 在 Agent 启动时调用，而非首次请求时"""
        # HTTP 连接池（STT、LLM）
        connector = aiohttp.TCPConnector(
            limit=20,              # 最大连接数
            keepalive_timeout=300,  # 5 分钟保活
            enable_cleanup_closed=True,
            ssl=False,             # 内网通信可关闭 SSL
        )
        self.stt_session = aiohttp.ClientSession(connector=connector)
        self.llm_session = aiohttp.ClientSession(connector=connector)
 
        # WebSocket 长连接（TTS）- 避免每次请求重新握手
        self.tts_ws = await websockets.connect(
            "wss://api.cartesia.ai/tts/websocket",
            ping_interval=30,      # 保活心跳
            ping_timeout=10,
            max_size=10 * 1024 * 1024,  # 10MB
        )
 
        # 发送预热请求（触发模型加载）
        await self._warmup_stt()
        await self._warmup_tts()
        print("[连接池] 所有服务连接已预热")
 
    async def _warmup_stt(self):
        """发送静音音频预热 STT 模型"""
        silence = b"\x00" * 3200  # 100ms 静音
        async with self.stt_session.post(
            "https://api.deepgram.com/v1/listen",
            data=silence,
            headers={"Authorization": f"Token {STT_API_KEY}"},
        ) as resp:
            await resp.read()
 
    async def _warmup_tts(self):
        """发送短文本预热 TTS 模型"""
        await self.tts_ws.send(json.dumps({
            "transcript": "。",  # 最短文本
            "voice": {"mode": "id", "id": VOICE_ID},
        }))

class AdaptiveVAD:
    """自适应语音活动检测 - 根据对话上下文动态调整灵敏度"""
 
    def __init__(self):
        self.silence_threshold_ms = 300  # 默认静音阈值
        self.min_threshold_ms = 150      # 最小阈值（快速对话）
        self.max_threshold_ms = 800      # 最大阈值（思考型问题）
 
    def adjust_threshold(self, context: dict):
        """
        根据对话上下文动态调整端点检测阈值：
        - 是/否问题 → 短阈值（用户回答快）
        - 开放性问题 → 长阈值（用户需要思考）
        - 连续对话 → 逐渐缩短阈值
        """
        last_agent_utterance = context.get("last_agent_text", "")
        turn_count = context.get("turn_count", 0)
 
        # 是/否问题 → 缩短等待
        if self._is_yes_no_question(last_agent_utterance):
            self.silence_threshold_ms = self.min_threshold_ms
        # 开放性问题 → 延长等待
        elif self._is_open_question(last_agent_utterance):
            self.silence_threshold_ms = self.max_threshold_ms
        # 对话进行中 → 逐渐适应用户节奏
        else:
            # 根据用户历史响应时间调整
            avg_response_time = context.get("avg_user_response_ms", 300)
            self.silence_threshold_ms = min(
                max(avg_response_time * 0.8, self.min_threshold_ms),
                self.max_threshold_ms,
            )
 
    def _is_yes_no_question(self, text: str) -> bool:
        patterns = ["是否", "对吗", "好吗", "可以吗", "需要吗", "确认"]
        return any(p in text for p in patterns)
 
    def _is_open_question(self, text: str) -> bool:
        patterns = ["请描述", "请详细", "您觉得", "怎么看", "什么原因"]
        return any(p in text for p in patterns)

import time
from dataclasses import dataclass, field
 
@dataclass
class LatencyBreakdown:
    """延迟分解记录"""
    vad_ms: float = 0
    stt_ms: float = 0
    llm_ttft_ms: float = 0    # LLM 首 token 时间
    llm_total_ms: float = 0   # LLM 完整生成时间
    tts_ttfa_ms: float = 0    # TTS 首音频块时间
    tts_total_ms: float = 0   # TTS 完整合成时间
    network_ms: float = 0
    total_ms: float = 0
 
    def report(self) -> str:
        return (
            f"延迟分解:\n"
            f"  VAD:       {self.vad_ms:6.1f}ms\n"
            f"  STT:       {self.stt_ms:6.1f}ms\n"
            f"  LLM TTFT:  {self.llm_ttft_ms:6.1f}ms\n"
            f"  TTS TTFA:  {self.tts_ttfa_ms:6.1f}ms\n"
            f"  网络:      {self.network_ms:6.1f}ms\n"
            f"  ─────────────────\n"
            f"  端到端:    {self.total_ms:6.1f}ms"
        )
 
class LatencyTracker:
    """延迟追踪器 - 记录每次交互的详细延迟"""
 
    def __init__(self):
        self.history: list[LatencyBreakdown] = []
 
    def track(self) -> "LatencyContext":
        return LatencyContext(self)
 
    def get_p50(self) -> float:
        """获取 P50 延迟"""
        totals = sorted(b.total_ms for b in self.history)
        idx = len(totals) // 2
        return totals[idx] if totals else 0
 
    def get_p95(self) -> float:
        """获取 P95 延迟"""
        totals = sorted(b.total_ms for b in self.history)
        idx = int(len(totals) * 0.95)
        return totals[min(idx, len(totals) - 1)] if totals else 0
 
    def get_bottleneck(self) -> str:
        """识别最大瓶颈环节"""
        if not self.history:
            return "无数据"
        latest = self.history[-1]
        components = {
            "VAD": latest.vad_ms,
            "STT": latest.stt_ms,
            "LLM": latest.llm_ttft_ms,
            "TTS": latest.tts_ttfa_ms,
            "网络": latest.network_ms,
        }
        return max(components, key=components.get)

# 使用 Prometheus 指标监控延迟
from prometheus_client import Histogram, Counter
 
voice_latency = Histogram(
    "voice_agent_latency_seconds",
    "语音 Agent 端到端延迟",
    ["component"],  # vad, stt, llm, tts, total
    buckets=[0.05, 0.1, 0.15, 0.2, 0.3, 0.5, 0.8, 1.0, 2.0],
)
 
voice_timeout = Counter(
    "voice_agent_timeout_total",
    "超时次数（>1s）",
    ["component"],
)

你是一个电话客服 Agent。请遵循以下规则以确保低延迟响应：

1. 回复必须简短精炼，每次回复不超过 [2-3] 句话
2. 如果需要查询信息，先说"好的，我帮您查一下"，然后再查询
3. 使用口语化表达，避免书面语
4. 不要重复用户说过的话
5. 如果问题复杂，分步骤回答，每步不超过 [1] 句话
6. 优先使用短句，避免从句嵌套

当前对话上下文：
- 客户姓名：[客户姓名]
- 业务类型：[业务类型]
- 已知信息：[已收集的信息]

┌──────────────────────────────────────────────────────────────────┐
│                WebSocket 流式 TTS 架构                            │
├──────────────────────────────────────────────────────────────────┤
│                                                                  │
│  LLM Token 流                                                    │
│  ┌─────┐                                                         │
│  │ "您" │──┐                                                     │
│  │ "好" │  │   文本缓冲区          WebSocket           客户端     │
│  │ "，" │  ├──→ [您好，] ──────→ TTS 服务 ──→ 音频块1 ──→ 🔊    │
│  │ "请" │  │                                                     │
│  │ "问" │  │   [请问有什么] ──→ TTS 服务 ──→ 音频块2 ──→ 🔊     │
│  │ "有" │  │                                                     │
│  │ "什" │  │   [可以帮您？] ──→ TTS 服务 ──→ 音频块3 ──→ 🔊     │
│  │ "么" │──┘                                                     │
│  └─────┘                                                         │
│                                                                  │
│  关键：按标点符号分割文本，每个片段独立合成                          │
│  好处：用户在 LLM 还在生成时就能听到前半句                          │
│                                                                  │
└──────────────────────────────────────────────────────────────────┘

class TextChunker:
    """
    智能文本分块器 - 在延迟和语音自然度之间取得平衡
 
    分块太小：语音不自然，TTS 请求过多
    分块太大：延迟增加，用户等待时间长
    最佳策略：按自然语句边界分割
    """
 
    def __init__(self, min_chars: int = 5, max_chars: int = 100):
        self.buffer = ""
        self.min_chars = min_chars
        self.max_chars = max_chars
        # 中文句子边界标点
        self.zh_boundaries = set("。！？；，、：")
        # 英文句子边界标点
        self.en_boundaries = set(".!?;,:")
 
    def add_token(self, token: str) -> str | None:
        """
        添加 token，返回可发送的文本块（如果有的话）
 
        策略：
        1. 遇到句子结束标点（。！？）→ 立即发送
        2. 遇到逗号等停顿标点 + 已积累足够文本 → 发送
        3. 超过最大长度 → 强制发送
        """
        self.buffer += token
 
        # 检查是否遇到强边界（句号、问号、感叹号）
        if self.buffer and self.buffer[-1] in "。！？.!?":
            return self._flush()
 
        # 检查弱边界（逗号等）+ 最小长度
        if (
            len(self.buffer) >= self.min_chars
            and self.buffer[-1] in self.zh_boundaries | self.en_boundaries
        ):
            return self._flush()
 
        # 超过最大长度，强制分割
        if len(self.buffer) >= self.max_chars:
            return self._flush()
 
        return None
 
    def flush(self) -> str | None:
        """强制刷新剩余内容"""
        return self._flush() if self.buffer.strip() else None
 
    def _flush(self) -> str:
        text = self.buffer.strip()
        self.buffer = ""
        return text

// 音频编码选择建议
interface AudioConfig {
  // 实时对话场景：优先选择 Opus 或 PCM
  realtime: {
    encoding: "opus" | "pcm_s16le";
    sampleRate: 24000; // 24kHz 足够语音质量
    channels: 1; // 单声道即可
    // Opus 配置
    opusBitrate: 32000; // 32kbps 语音足够
    opusFrameSize: 20; // 20ms 帧大小（平衡延迟和效率）
  };
  // 电话场景：使用 μ-law
  telephony: {
    encoding: "mulaw";
    sampleRate: 8000; // 电话标准
    channels: 1;
  };
}

场景：中国用户 → 美国西海岸服务器

单次往返延迟：~150ms
语音管线需要 3-4 次往返（STT + LLM + TTS + 音频传输）
网络延迟总计：450-600ms（已超过 300ms 目标！）

解决方案：边缘部署
中国用户 → 亚太边缘节点（~20ms 往返）
网络延迟总计：60-80ms ✅

// Cloudflare Workers - 语音 Agent 边缘网关
// 功能：在边缘节点路由语音请求到最近的 AI 服务
 
export default {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
 
    // 根据请求类型路由
    switch (url.pathname) {
      case "/voice/stt":
        return handleSTT(request, env);
      case "/voice/tts":
        return handleTTS(request, env);
      case "/voice/ws":
        return handleWebSocket(request, env);
      default:
        return new Response("Not Found", { status: 404 });
    }
  },
};
 
async function handleSTT(request: Request, env: Env): Promise<Response> {
  const audioData = await request.arrayBuffer();
 
  // 使用 Cloudflare Workers AI 内置的 Whisper 模型
  // 在边缘节点直接推理，无需外部 API 调用
  const result = await env.AI.run("@cf/openai/whisper-tiny-en", {
    audio: [...new Uint8Array(audioData)],
  });
 
  return Response.json({
    text: result.text,
    // 边缘推理延迟通常 < 100ms
    processing_location: request.cf?.colo, // 显示处理节点
  });
}
 
async function handleTTS(request: Request, env: Env): Promise<Response> {
  const { text, voice_id, language } = await request.json();
 
  // 选择最近的 TTS 服务端点
  const ttsEndpoint = selectNearestTTSEndpoint(
    request.cf?.colo ?? "SJC",
    language
  );
 
  // 流式转发 TTS 请求
  const ttsResponse = await fetch(ttsEndpoint, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${env.TTS_API_KEY}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ text, voice_id, stream: true }),
  });
 
  // 直接流式转发音频数据
  return new Response(ttsResponse.body, {
    headers: {
      "Content-Type": "audio/mpeg",
      "Transfer-Encoding": "chunked",
      // 缓存控制：相同文本可缓存
      "Cache-Control": "public, max-age=3600",
    },
  });
}
 
function selectNearestTTSEndpoint(colo: string, language: string): string {
  // 根据边缘节点位置和语言选择最近的 TTS 服务
  const regionMap: Record<string, string> = {
    // 亚太节点 → 亚太 TTS 服务
    HKG: "https://api-ap.cartesia.ai/tts/bytes",
    NRT: "https://api-ap.cartesia.ai/tts/bytes",
    SIN: "https://api-ap.cartesia.ai/tts/bytes",
    // 欧洲节点 → 欧洲 TTS 服务
    FRA: "https://api-eu.cartesia.ai/tts/bytes",
    LHR: "https://api-eu.cartesia.ai/tts/bytes",
    // 默认 → 美国
    default: "https://api.cartesia.ai/tts/bytes",
  };
 
  return regionMap[colo] ?? regionMap["default"];
}
 
// WebSocket 处理 - 全双工实时语音通信
async function handleWebSocket(
  request: Request,
  env: Env
): Promise<Response> {
  const upgradeHeader = request.headers.get("Upgrade");
  if (upgradeHeader !== "websocket") {
    return new Response("Expected WebSocket", { status: 426 });
  }
 
  const [client, server] = Object.values(new WebSocketPair());
 
  server.accept();
  server.addEventListener("message", async (event) => {
    // 处理音频数据流
    const data = JSON.parse(event.data as string);
 
    if (data.type === "audio") {
      // 边缘 STT 处理
      const transcript = await env.AI.run("@cf/openai/whisper-tiny-en", {
        audio: data.audio,
      });
      server.send(JSON.stringify({ type: "transcript", text: transcript.text }));
    }
  });
 
  return new Response(null, { status: 101, webSocket: client });
}

┌─────────────────────────────────────────────────────────────────────┐
│                    多区域语音 Agent 部署架构                          │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│                    ┌─────────────────┐                              │
│                    │  Cloudflare DNS  │                              │
│                    │  (GeoDNS 路由)   │                              │
│                    └────────┬────────┘                              │
│                             │                                       │
│              ┌──────────────┼──────────────┐                        │
│              ▼              ▼              ▼                        │
│     ┌────────────┐  ┌────────────┐  ┌────────────┐                 │
│     │ 亚太边缘    │  │ 欧洲边缘    │  │ 北美边缘    │                 │
│     │ (HKG/NRT)  │  │ (FRA/LHR)  │  │ (SJC/IAD)  │                 │
│     └─────┬──────┘  └─────┬──────┘  └─────┬──────┘                 │
│           │               │               │                        │
│     ┌─────▼──────┐  ┌─────▼──────┐  ┌─────▼──────┐                 │
│     │ STT: 深鉴   │  │ STT: Deepgram│  │ STT: Deepgram│             │
│     │ TTS: 讯飞   │  │ TTS: Cartesia│  │ TTS: Cartesia│             │
│     │ LLM: 通义   │  │ LLM: Claude  │  │ LLM: GPT-4o │             │
│     └────────────┘  └────────────┘  └────────────┘                 │
│                                                                     │
│  策略：                                                              │
│  1. GeoDNS 将用户路由到最近的边缘节点                                 │
│  2. 每个区域使用本地化的 AI 服务（降低网络延迟）                       │
│  3. 中国区域使用国内服务商（避免跨境延迟）                             │
│  4. 对话状态通过 Redis Cluster 跨区域同步                             │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

# 分析现有流量的地理分布
# 使用 Cloudflare Analytics 或 AWS CloudFront 日志
# 确定需要部署边缘节点的区域

# Cloudflare DNS 配置示例
# 使用 Load Balancing 实现地理路由
load_balancer:
  name: voice-agent.example.com
  steering_policy: geo
  pools:
    - name: asia-pacific
      origins:
        - address: voice-ap.example.com
      regions: ["ENAM", "OC", "EAS", "SEAS"]
    - name: europe
      origins:
        - address: voice-eu.example.com
      regions: ["WEU", "EEU"]
    - name: north-america
      origins:
        - address: voice-us.example.com
      regions: ["NAM", "SAM"]

class RegionAwareServiceRouter:
    """区域感知服务路由器 - 为每个区域选择最优的 AI 服务组合"""
 
    REGION_CONFIGS = {
        "asia-pacific": {
            "stt": {"provider": "deepgram", "endpoint": "https://api-ap.deepgram.com"},
            "llm": {"provider": "openai", "endpoint": "https://api.openai.com"},
            "tts": {"provider": "cartesia", "endpoint": "https://api-ap.cartesia.ai"},
        },
        "china": {
            # 中国区域使用国内服务商，避免跨境延迟
            "stt": {"provider": "xunfei", "endpoint": "https://iat-api.xfyun.cn"},
            "llm": {"provider": "qwen", "endpoint": "https://dashscope.aliyuncs.com"},
            "tts": {"provider": "xunfei", "endpoint": "https://tts-api.xfyun.cn"},
        },
        "europe": {
            "stt": {"provider": "deepgram", "endpoint": "https://api-eu.deepgram.com"},
            "llm": {"provider": "anthropic", "endpoint": "https://api.anthropic.com"},
            "tts": {"provider": "cartesia", "endpoint": "https://api-eu.cartesia.ai"},
        },
        "north-america": {
            "stt": {"provider": "deepgram", "endpoint": "https://api.deepgram.com"},
            "llm": {"provider": "openai", "endpoint": "https://api.openai.com"},
            "tts": {"provider": "cartesia", "endpoint": "https://api.cartesia.ai"},
        },
    }
 
    def get_config(self, user_region: str) -> dict:
        return self.REGION_CONFIGS.get(user_region, self.REGION_CONFIGS["north-america"])

┌─────────────────────────────────────────────────────────────────┐
│                  多语言语音 Agent 挑战                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  1. 语言检测                                                     │
│     · 用户可能随时切换语言                                        │
│     · 短语音片段难以准确识别语言                                   │
│     · 方言 vs 标准语的区分                                        │
│                                                                 │
│  2. 代码切换（Code-Switching）                                   │
│     · "我想 check 一下 my order"（中英混合）                     │
│     · "Quiero pagar my bill"（西英混合）                         │
│     · 大多数 ASR 模型在混合语言时准确率骤降                       │
│                                                                 │
│  3. 口音处理                                                     │
│     · 非母语口音识别率低                                          │
│     · 地区口音差异大（如印度英语 vs 英式英语）                     │
│     · 同一语言不同地区的表达习惯差异                               │
│                                                                 │
│  4. TTS 语音匹配                                                 │
│     · 切换语言时需要匹配对应语言的语音                             │
│     · 保持语音风格一致性（语速、语调、情感）                       │
│     · 多语言语音克隆的质量差异                                    │
│                                                                 │
│  5. LLM 多语言能力                                               │
│     · 不同语言的推理能力差异                                      │
│     · 文化敏感性和本地化表达                                      │
│     · 多语言 prompt 的设计                                        │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

用户语音 → [多语言 STT] → [多语言 LLM] → [多语言 TTS] → 语音输出
                ↓
         自动检测语言
         同一模型处理所有语言

用户语音 → [语言检测] → 路由到对应语言管线
                │
                ├→ 中文管线：讯飞 STT → 通义 LLM → 讯飞 TTS
                ├→ 英文管线：Deepgram STT → GPT-4o → ElevenLabs TTS
                ├→ 日文管线：Google STT → Claude → Google TTS
                └→ 其他语言：通用管线

import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Optional
 
class Language(Enum):
    ZH = "zh"       # 中文
    EN = "en"       # 英文
    JA = "ja"       # 日文
    KO = "ko"       # 韩文
    ES = "es"       # 西班牙文
    UNKNOWN = "unknown"
 
@dataclass
class LanguageDetectionResult:
    language: Language
    confidence: float
    is_code_switching: bool  # 是否检测到语言混合
 
@dataclass
class VoiceConfig:
    """每种语言对应的语音配置"""
    voice_id: str
    model: str
    speaking_rate: float = 1.0
    pitch: float = 0.0
 
class MultilingualVoiceAgent:
    """
    多语言语音 Agent - 支持实时语言检测和无缝切换
 
    核心设计：
    1. 每轮对话检测用户语言
    2. 动态切换 STT/TTS 配置
    3. 保持对话上下文跨语言连续
    """
 
    def __init__(self):
        self.current_language = Language.ZH  # 默认中文
        self.language_history: list[Language] = []
 
        # 每种语言的 TTS 语音配置
        self.voice_configs: dict[Language, VoiceConfig] = {
            Language.ZH: VoiceConfig(
                voice_id="zh-female-professional",
                model="sonic-3",
                speaking_rate=1.0,
            ),
            Language.EN: VoiceConfig(
                voice_id="en-female-professional",
                model="sonic-3",
                speaking_rate=1.0,
            ),
            Language.JA: VoiceConfig(
                voice_id="ja-female-professional",
                model="sonic-3",
                speaking_rate=0.95,  # 日语稍慢更自然
            ),
        }
 
        # 每种语言的系统提示词
        self.system_prompts: dict[Language, str] = {
            Language.ZH: "你是一个专业的中文客服助手。请用简洁的中文回复。",
            Language.EN: "You are a professional customer service assistant. Reply concisely in English.",
            Language.JA: "あなたはプロのカスタマーサービスアシスタントです。簡潔な日本語で返答してください。",
        }
 
    async def detect_language(self, transcript: str) -> LanguageDetectionResult:
        """
        检测用户语言 - 结合多种信号：
        1. STT 返回的语言标签
        2. 文本内容的字符分析
        3. 历史语言偏好
        """
        # 字符级语言检测
        zh_chars = sum(1 for c in transcript if "\u4e00" <= c <= "\u9fff")
        en_chars = sum(1 for c in transcript if c.isascii() and c.isalpha())
        ja_chars = sum(
            1 for c in transcript
            if "\u3040" <= c <= "\u309f" or "\u30a0" <= c <= "\u30ff"
        )
        total = max(zh_chars + en_chars + ja_chars, 1)
 
        # 检测代码切换（多语言混合）
        lang_counts = sum(1 for count in [zh_chars, en_chars, ja_chars] if count > 0)
        is_code_switching = lang_counts >= 2 and min(
            c for c in [zh_chars, en_chars, ja_chars] if c > 0
        ) / total > 0.15
 
        # 确定主要语言
        if zh_chars / total > 0.3:
            detected = Language.ZH
            confidence = zh_chars / total
        elif ja_chars / total > 0.2:
            detected = Language.JA
            confidence = ja_chars / total
        elif en_chars / total > 0.5:
            detected = Language.EN
            confidence = en_chars / total
        else:
            detected = self.current_language  # 保持当前语言
            confidence = 0.5
 
        return LanguageDetectionResult(
            language=detected,
            confidence=confidence,
            is_code_switching=is_code_switching,
        )
 
    async def handle_language_switch(
        self, detection: LanguageDetectionResult
    ) -> bool:
        """
        处理语言切换 - 防止误切换的策略：
        1. 置信度 > 0.7 才切换
        2. 连续 2 次检测到同一语言才切换
        3. 代码切换时保持主要语言不变
        """
        if detection.is_code_switching:
            # 代码切换时不改变主语言，但 STT 需要支持混合识别
            return False
 
        if detection.confidence < 0.7:
            return False
 
        if detection.language == self.current_language:
            return False
 
        # 检查是否连续检测到新语言
        self.language_history.append(detection.language)
        recent = self.language_history[-2:]
        if len(recent) >= 2 and all(l == detection.language for l in recent):
            old_lang = self.current_language
            self.current_language = detection.language
            print(f"[语言切换] {old_lang.value} → {detection.language.value}")
            return True
 
        return False
 
    def get_tts_config(self) -> VoiceConfig:
        """获取当前语言的 TTS 配置"""
        return self.voice_configs.get(
            self.current_language,
            self.voice_configs[Language.EN],  # 默认英文
        )
 
    def get_system_prompt(self) -> str:
        """获取当前语言的系统提示词"""
        return self.system_prompts.get(
            self.current_language,
            self.system_prompts[Language.EN],
        )

import { Agent, AgentSession } from "@livekit/agents";
import { DeepgramSTT } from "@livekit/agents-plugin-deepgram";
import { OpenAILLM } from "@livekit/agents-plugin-openai";
 
interface LanguageConfig {
  sttLanguage: string;
  ttsVoiceId: string;
  ttsModel: string;
  systemPrompt: string;
}
 
const LANGUAGE_CONFIGS: Record<string, LanguageConfig> = {
  zh: {
    sttLanguage: "zh",
    ttsVoiceId: "zh-CN-XiaoxiaoNeural",
    ttsModel: "sonic-3",
    systemPrompt: "你是专业客服助手。用简洁中文回复，每次不超过2句话。",
  },
  en: {
    sttLanguage: "en",
    ttsVoiceId: "en-US-AriaNeural",
    ttsModel: "sonic-3",
    systemPrompt:
      "You are a professional assistant. Reply concisely in 1-2 sentences.",
  },
  ja: {
    sttLanguage: "ja",
    ttsVoiceId: "ja-JP-NanamiNeural",
    ttsModel: "sonic-3",
    systemPrompt:
      "あなたはプロのアシスタントです。1-2文で簡潔に返答してください。",
  },
};
 
class MultilingualVoiceAgent {
  private currentLanguage = "zh";
  private languageHistory: string[] = [];
 
  /**
   * 检测语言并动态更新 Agent 配置
   * 使用 LiveKit Agents 框架实现无缝语言切换
   */
  async processTranscript(
    transcript: string,
    session: AgentSession
  ): Promise<void> {
    const detected = this.detectLanguage(transcript);
 
    if (detected !== this.currentLanguage) {
      this.languageHistory.push(detected);
 
      // 连续 2 次检测到新语言才切换（防止误判）
      const recent = this.languageHistory.slice(-2);
      if (recent.length >= 2 && recent.every((l) => l === detected)) {
        console.log(`[语言切换] ${this.currentLanguage} → ${detected}`);
        this.currentLanguage = detected;
 
        // 动态更新 TTS 配置
        const config = LANGUAGE_CONFIGS[detected] ?? LANGUAGE_CONFIGS["en"];
        await session.updateTTSConfig({
          voiceId: config.ttsVoiceId,
          model: config.ttsModel,
        });
      }
    }
  }
 
  private detectLanguage(text: string): string {
    // 基于 Unicode 范围的快速语言检测
    const zhCount = (text.match(/[\u4e00-\u9fff]/g) || []).length;
    const jaCount = (text.match(/[\u3040-\u30ff]/g) || []).length;
    const total = text.length;
 
    if (zhCount / total > 0.3) return "zh";
    if (jaCount / total > 0.2) return "ja";
    return "en";
  }
 
  getConfig(): LanguageConfig {
    return LANGUAGE_CONFIGS[this.currentLanguage] ?? LANGUAGE_CONFIGS["en"];
  }
}

你是一个多语言客服 Agent，支持以下语言：中文、英文、日文。

语言处理规则：
1. 始终使用用户最后使用的语言回复
2. 如果用户混合使用多种语言（如中英混合），使用其中占比更大的语言回复
3. 当检测到语言切换时，自然过渡，不要提及语言变化
4. 保持跨语言的对话上下文连续性

文化适配规则：
- 中文：使用"您"作为尊称，语气温和专业
- 英文：使用 professional but friendly 的语气
- 日文：使用敬语（です/ます体）

当前检测到的用户语言：[detected_language]
对话历史：[conversation_history]

┌─────────────────────────────────────────────────────────────────┐
│                  中文方言处理架构                                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  用户语音输入                                                     │
│       │                                                         │
│       ▼                                                         │
│  ┌──────────────┐                                               │
│  │ 方言检测模块  │ ← 识别：普通话 / 粤语 / 闽南语 / 四川话 / 上海话│
│  └──────┬───────┘                                               │
│         │                                                       │
│    ┌────┴────┐                                                  │
│    ▼         ▼                                                  │
│  普通话    方言                                                   │
│    │         │                                                  │
│    ▼         ▼                                                  │
│  标准 STT  方言 STT                                              │
│  (Deepgram) (讯飞方言识别)                                       │
│    │         │                                                  │
│    └────┬────┘                                                  │
│         ▼                                                       │
│  统一文本（普通话）                                                │
│         │                                                       │
│         ▼                                                       │
│  LLM 处理（普通话）                                               │
│         │                                                       │
│    ┌────┴────┐                                                  │
│    ▼         ▼                                                  │
│  普通话 TTS  方言 TTS（可选）                                     │
│  (默认回复)  (匹配用户方言)                                       │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

class ChineseTextNormalizer:
    """
    中文文本规范化器 - 处理语音 AI 中的中文特殊问题
 
    核心功能：
    1. 数字表达规范化（"一百二" → "120"）
    2. 同音字消歧（基于上下文）
    3. 语气词处理
    4. 地址/人名优化
    """
 
    def __init__(self):
        # 中文数字映射
        self.cn_digits = {
            "零": 0, "一": 1, "二": 2, "两": 2, "三": 3,
            "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9,
            "十": 10, "百": 100, "千": 1000, "万": 10000, "亿": 100000000,
        }
        # 常见语气词及其语义
        self.mood_particles = {
            "嗯": "确认/思考",
            "啊": "惊讶/强调",
            "哦": "理解/恍然",
            "呢": "疑问/反问",
            "吧": "建议/不确定",
            "嘛": "显而易见",
            "呀": "感叹",
        }
        # 自定义词典（提高 STT 准确率）
        self.custom_vocabulary: list[str] = []
 
    def normalize_chinese_number(self, text: str) -> str:
        """
        规范化中文数字表达
        "一百二" → "120"（口语省略"十"）
        "三千五" → "3500"
        "两万三千" → "23000"
        """
        import re
 
        # 处理口语化数字（"一百二" = 120，不是 102）
        patterns = [
            (r"(\d+)百(\d)(?!\d|十|百|千)", lambda m: str(int(m.group(1)) * 100 + int(m.group(2)) * 10)),
            (r"(\d+)千(\d)(?!\d|十|百|千)", lambda m: str(int(m.group(1)) * 1000 + int(m.group(2)) * 100)),
            (r"(\d+)万(\d)(?!\d|十|百|千)", lambda m: str(int(m.group(1)) * 10000 + int(m.group(2)) * 1000)),
        ]
        for pattern, replacement in patterns:
            text = re.sub(pattern, replacement, text)
        return text
 
    def extract_mood_intent(self, text: str) -> dict:
        """
        分析语气词，辅助意图理解
        "嗯...好吧" → {"mood": "hesitant", "intent": "reluctant_agreement"}
        "哦！原来如此" → {"mood": "surprised", "intent": "understanding"}
        """
        mood_signals = []
        for particle, meaning in self.mood_particles.items():
            if particle in text:
                mood_signals.append(meaning)
 
        # 综合判断语气
        if "确认" in mood_signals and "不确定" in mood_signals:
            return {"mood": "hesitant", "confidence": 0.5}
        elif "确认" in mood_signals:
            return {"mood": "affirmative", "confidence": 0.8}
        elif "疑问" in mood_signals:
            return {"mood": "questioning", "confidence": 0.7}
        return {"mood": "neutral", "confidence": 0.5}
 
    def build_custom_vocabulary(self, domain: str) -> list[str]:
        """
        构建领域自定义词典 - 提高 STT 对专业术语的识别率
 
        用法：将词典传给 STT API 的 keywords/vocabulary 参数
        """
        domain_vocabs = {
            "banking": [
                "转账", "汇款", "理财", "定期", "活期", "信用卡",
                "还款日", "账单", "额度", "分期", "年化收益率",
            ],
            "medical": [
                "挂号", "门诊", "住院", "处方", "医保",
                "报销", "化验", "CT", "核磁", "B超",
            ],
            "ecommerce": [
                "下单", "退款", "退货", "物流", "快递",
                "优惠券", "满减", "包邮", "预售", "尾款",
            ],
        }
        self.custom_vocabulary = domain_vocabs.get(domain, [])
        return self.custom_vocabulary

class ChineseTTSOptimizer:
    """中文 TTS 优化器 - 处理多音字、数字朗读、混合语言等"""
 
    def __init__(self):
        # 多音字上下文规则
        self.polyphone_rules = {
            "行": {
                "银行": "háng", "行业": "háng", "行情": "háng",
                "行走": "xíng", "执行": "xíng", "行为": "xíng",
            },
            "了": {
                "了解": "liǎo", "了不起": "liǎo",
                # 默认读 "le"（语气助词）
            },
            "还": {
                "还款": "huán", "归还": "huán",
                "还是": "hái", "还有": "hái",
            },
        }
 
    def preprocess_for_tts(self, text: str, context: str = "") -> str:
        """
        TTS 预处理：优化文本以获得更自然的语音输出
 
        处理流程：
        1. 数字朗读规范化
        2. 添加 SSML 停顿标记
        3. 英文单词发音标注
        """
        text = self._normalize_numbers_for_speech(text)
        text = self._add_natural_pauses(text)
        text = self._handle_mixed_language(text)
        return text
 
    def _normalize_numbers_for_speech(self, text: str) -> str:
        """数字朗读规范化"""
        import re
 
        # 电话号码：逐位朗读
        text = re.sub(
            r"(\d{3})[-\s]?(\d{4})[-\s]?(\d{4})",
            lambda m: " ".join(m.group(0).replace("-", "").replace(" ", "")),
            text,
        )
 
        # 金额：标准朗读
        text = re.sub(
            r"(\d+)元",
            lambda m: f"{m.group(1)}元",
            text,
        )
 
        return text
 
    def _add_natural_pauses(self, text: str) -> str:
        """在适当位置添加停顿，提升自然度"""
        # 在逗号后添加短停顿
        text = text.replace("，", "，ˌ")  # 使用特殊标记
        # 在句号后添加长停顿
        text = text.replace("。", "。ˈ")
        return text
 
    def _handle_mixed_language(self, text: str) -> str:
        """处理中英混合文本"""
        import re
        # 在英文单词前后添加空格，帮助 TTS 正确切换语言
        text = re.sub(r"([\u4e00-\u9fff])([a-zA-Z])", r"\1 \2", text)
        text = re.sub(r"([a-zA-Z])([\u4e00-\u9fff])", r"\1 \2", text)
        return text

┌─────────────────────────────────────────────────────────────────────┐
│              全球化客服语音 Agent 架构                                │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  ┌──────────┐     ┌──────────────┐     ┌──────────────┐            │
│  │ 用户电话  │────→│ Twilio/Vonage │────→│ LiveKit 网关  │            │
│  └──────────┘     │ (电话接入)    │     │ (WebRTC 桥接) │            │
│                   └──────────────┘     └──────┬───────┘            │
│                                               │                    │
│                                        ┌──────▼───────┐            │
│                                        │  语言检测模块  │            │
│                                        │  (首 3 秒音频) │            │
│                                        └──────┬───────┘            │
│                                               │                    │
│                    ┌──────────────────────────┼──────────┐         │
│                    ▼                          ▼          ▼         │
│             ┌────────────┐          ┌────────────┐ ┌──────────┐   │
│             │  中文管线    │          │  英文管线    │ │ 日文管线  │   │
│             ├────────────┤          ├────────────┤ ├──────────┤   │
│             │STT: Deepgram│          │STT: Deepgram│ │STT: Google│  │
│             │    Nova-3   │          │    Nova-3   │ │  Cloud   │   │
│             │LLM: GPT-4o  │          │LLM: GPT-4o  │ │LLM: Claude│  │
│             │    mini     │          │    mini     │ │  Haiku   │   │
│             │TTS: Cartesia│          │TTS: Cartesia│ │TTS: Google│  │
│             │   Sonic-3   │          │   Sonic-3   │ │  Cloud   │   │
│             └──────┬─────┘          └──────┬─────┘ └────┬─────┘   │
│                    │                       │            │          │
│                    └───────────┬───────────┘            │          │
│                                ▼                        │          │
│                    ┌──────────────────┐                 │          │
│                    │  对话状态管理     │←────────────────┘          │
│                    │  (Redis Cluster)  │                            │
│                    └────────┬─────────┘                            │
│                             │                                      │
│                    ┌────────▼─────────┐                            │
│                    │  业务系统集成     │                            │
│                    │  · 订单查询 API   │                            │
│                    │  · 物流追踪 API   │                            │
│                    │  · CRM 系统       │                            │
│                    └──────────────────┘                            │
│                                                                    │
└─────────────────────────────────────────────────────────────────────┘