六大语音Agent引擎架构与延迟实测

graph LR
    subgraph "传统三级管道 (2024)"
        A1[ASR<br/>Whisper/Deepgram] --> B1[LLM<br/>GPT-4o/Claude] --> C1[TTS<br/>ElevenLabs/Cartesia]
        style A1 fill:#ff9999
        style B1 fill:#ffcc99
        style C1 fill:#99ccff
    end

    subgraph "融合管道 (2025)"
        A2[Streaming ASR] -.-> B2[LLM + Streaming TTS<br/>流式交织]
        style A2 fill:#ffcc99
        style B2 fill:#99ff99
    end

    subgraph "端到端 (2026)"
        A3[Speech-to-Speech<br/>GPT-Realtime-2 / Ultravox]
        style A3 fill:#99ccff
    end

    A1 -.->|延迟 1.5-3s| A2
    A2 -.->|延迟 500-800ms| A3
    A3 -.->|延迟 100-500ms| X[目标：<200ms]

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def voice_agent():
    async with client.realtime.connect(
        model="gpt-realtime-2",
        voice="coral",
    ) as conn:
        await conn.session.update(session={
            "turn_detection": {"type": "server_vad"},
            "tools": [{
                "type": "function",
                "name": "lookup_order",
                "description": "查询订单状态",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "order_id": {"type": "string"}
                    }
                }
            }]
        })

        async for event in conn.listen():
            if event.type == "response.audio_transcript.done":
                print(f"Agent: {event.transcript}")
            elif event.type == "response.function_call_arguments.done":
                result = await lookup_order(event.arguments)
                await conn.conversation.item.create(
                    function_call_output={
                        "call_id": event.call_id,
                        "output": result
                    }
                )

asyncio.run(voice_agent())

// Vapi Voice Agent 配置示例
const assistant = await vapi.assistants.create({
  name: "客服助手",
  model: {
    provider: "openai",
    model: "gpt-4o",
    messages: [{
      role: "system",
      content: "你是客服代表，用中文回答客户问题。"
    }],
    tools: [{
      type: "function",
      name: "check_balance",
      description: "查询用户余额"
    }]
  },
  voice: {
    provider: "11labs",
    voiceId: "pNInz6obpgDQGcFmaJgB",
    stability: 0.7,
    similarity_boost: 0.8
  },
  transcriber: {
    provider: "deepgram",
    model: "nova-2",
    language: "zh"
  },
  firstMessage: "您好，请问有什么可以帮您？"
});

// 发起电话呼叫
const call = await vapi.calls.phoneCall({
  assistantId: assistant.id,
  phoneNumberId: "phone_123",
  customer: {
    number: "+8613800138000"
  }
});

from retell_sdk import Retell

retell = Retell(api_key="your_key")

# 创建语音Agent
agent = retell.agents.create(
    name="销售助手",
    llm={
        "provider": "openai",
        "model": "gpt-4o",
        "system_prompt": "你是电话销售，用自然语气引导客户。",
    },
    voice={
        "provider": "elevenlabs",
        "voice_id": "Xb7hH8MSUJpSbSDYk0k2",
    },
    turn_detection={
        "type": "semantic",
        "params": {
            "wait_time": 0.6,       # 等待确认语义结束
            "max_wait": 2.0,        # 最长等待
        }
    },
    respond_options={
        "interruption": "agent_stops",  # 打断策略：Agent立即停止
    }
)

# 发起SIP电话
call = retell.calls.create_phone_call(
    agent_id=agent.id,
    metadata={"customer_id": "cust_789"},
    retell_llm_dynamic_variables={
        "customer_name": "张先生",
        "recent_order": "iPhone 16 Pro"
    }
)

import { ElevenLabsClient } from "elevenlabs";

const client = new ElevenLabsClient({ apiKey: "your_key" });

// 创建Conversational Agent
const agent = await client.conversationalAi.agents.create({
  name: "语音助手",
  language: "zh",
  llm: {
    provider: "openai",
    model: "gpt-4o",
    systemPrompt: "用自然语气回答用户问题，语速适中。",
  },
  tts: {
    provider: "elevenlabs",
    voiceId: "custom_voice_id",
    model: "eleven_turbo_v2_5",
    stability: 0.6,
    similarityBoost: 0.8,
    style: 0.3,
  },
  asr: {
    provider: "elevenlabs",
  }
});

// 生成WebSocket签名用于前端连接
const signedUrl = await client.conversationalAi.getSignedUrl({
  agentId: agent.id,
});

# Pipecat 语音Agent示例
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.services.deepgram import DeepgramSTTService
from pipecat.services.openai import OpenAILLMService
from pipecat.services.elevenlabs import ElevenLabsTTSService
from pipecat.transports.services.daily import DailyParams

# 构建管道
pipeline = Pipeline([
    DeepgramSTTService(api_key="dg_key", model="nova-2"),
    OpenAILLMService(
        api_key="openai_key",
        model="gpt-4o",
        system_message="你是语音助手，简洁回答。"
    ),
    ElevenLabsTTSService(
        api_key="11labs_key",
        voice_id="Xb7hH8MSUJpSbSDYk0k2",
        model="eleven_turbo_v2_5"
    )
])

# 运行
runner = PipelineRunner()
await runner.run(pipeline)

graph TD
    subgraph "三大核心技术挑战"
        T1[Turn-Taking<br/>话轮交替]
        T2[Barge-In<br/>打断处理]
        T3[Latency Budget<br/>延迟预算分配]
    end

    T1 --> S1[语义vs声学检测]
    T1 --> S2[多说话人场景]
    T1 --> S3[文化差异<br/>中文vs英文停顿习惯]

    T2 --> S4[即停即说<br/>Agent中断生成]
    T2 --> S5[上下文保持<br/>打断后如何续接]
    T2 --> S6[误触发过滤<br/>噪音/背景声vs真实打断]

    T3 --> S7[网络传输: 50-100ms]
    T3 --> S8[ASR流式: 150-300ms]
    T3 --> S9[LLM首token: 200-500ms]
    T3 --> S10[TTS首字节: 150-250ms]
    T3 --> S11[目标: 总计 &lt;800ms]

graph LR
    subgraph "选型决策树"
        Q1{需要电话集成？}
        Q1 -->|是| Q2{外呼为主？}
        Q1 -->|否| Q3{语音品质优先？}

        Q2 -->|是| Q4{结构化对话？}
        Q2 -->|否| Q5{合规要求高？}

        Q4 -->|是| R1[Bland AI]
        Q4 -->|否| R2[Vapi]

        Q5 -->|是 HIPAA| R3[Retell AI]
        Q5 -->|否| R2

        Q3 -->|是| R4[ElevenLabs]
        Q3 -->|否| Q6{自主可控？}

        Q6 -->|是| R5[LiveKit + Pipecat]
        Q6 -->|否| R6[OpenAI Realtime API]
    end

    style R1 fill:#ff9999
    style R2 fill:#ffcc99
    style R3 fill:#99ff99
    style R4 fill:#99ccff
    style R5 fill:#cc99ff
    style R6 fill:#ff99ff