让语音智能体跳过 TTS，直接播放自定义 PCM

标准链路是 LLM 出文字、TTS 变声音。但有些场景绕不开自己的音频：品牌代言人的录音、游戏里的角色音效、自研的语音模型、已经合成好的固定话术。

把 llm.output_modalities 设成 ["audio"] 就能走这条路。Custom LLM 通过 SSE 直接返回 Base64 PCM，智能体收到后发布到 RTC 频道，中间不经过 TTS。好处是音色和内容完全由你掌控：品牌音一致、音效精确、不受 TTS 合成质量波动影响，也省掉了一次合成的耗时。

做完你会听到：频道里播放的是你自己的音频，而页面上的字幕和听到的内容完全对得上。

> 开始之前　先完成《给语音智能体接入自定义 LLM：OpenAI 兼容网关》，音频就是从那个服务里直接吐出来的。

一. 架构与准备工作

用户语音 → 凤鸣 → 自定义音频服务
                     │
                     ├── audio.transcript：字幕与短期记忆
                     └── audio.data：Base64 PCM16 / 16 kHz / mono
                                      ↓
                              RTC 直接播放，不再合成文字

需要 App ID、App Certificate、MiniMax Key，以及一个云端能访问的 HTTPS 音频服务。交付示例读取 16 kHz、16-bit、单声道的裸 PCM 文件，按 40 ms 分块发送。ALLOW_DEMO_TONE 只用于协议联调，别带到生产。

二. 环境变量

AGORA_APP_ID=REPLACE_WITH_SHENGWANG_APP_ID
AGORA_APP_CERTIFICATE=REPLACE_WITH_PRIMARY_CERTIFICATE
MINIMAX_API_KEY=REPLACE_WITH_MINIMAX_KEY
CUSTOM_LLM_URL=https://REPLACE_WITH_PUBLIC_HOST/chat/completions
CUSTOM_LLM_SHARED_SECRET=REPLACE_WITH_RANDOM_SHARED_SECRET
CUSTOM_AUDIO_PCM_PATH=/absolute/path/to/16k-mono-s16le.pcm
CUSTOM_AUDIO_TRANSCRIPT=你好，这是由自定义音频模态服务直接返回的语音。
AUDIO_REALTIME_PACING=true
ALLOW_DEMO_TONE=false

三. 读取并分块 PCM

from pathlib import Path

SAMPLE_RATE = 16_000
BYTES_PER_SAMPLE = 2
CHUNK_DURATION_MS = 40
CHUNK_SIZE = (
    SAMPLE_RATE * BYTES_PER_SAMPLE * CHUNK_DURATION_MS // 1000
)

def read_pcm(path: str) -> bytes:
    data = Path(path).read_bytes()
    if not data or len(data) > 5 * 1024 * 1024 or len(data) % 2:
        raise ValueError("PCM 必须非空、不超过 5 MB 且为 16-bit 对齐")
    return data

def split_pcm(audio: bytes) -> list[bytes]:
    return [
        audio[index:index + CHUNK_SIZE]
        for index in range(0, len(audio), CHUNK_SIZE)
    ]

裸 PCM 没有文件头，服务端没法从文件本身推断采样率和声道数，传错了不会报错，只会播出来不对劲。所以发布前一定要用音频工具确认：signed 16-bit little-endian、16 kHz、mono，三项都对。

四. 返回音频模态 SSE

import asyncio
import base64
import json
import uuid
from collections.abc import AsyncIterator

def sse(data: dict) -> str:
    return f"data: {json.dumps(data, ensure_ascii=False)}\n\n"

async def stream_audio(
    transcript: str,
    pcm: bytes,
) -> AsyncIterator[str]:
    message_id = f"chatcmpl-{uuid.uuid4().hex}"
    audio_id = uuid.uuid4().hex
    yield sse({
        "id": message_id,
        "choices": [{
            "index": 0,
            "delta": {"role": "assistant", "audio": {
                "id": audio_id,
                "transcript": transcript,
            }},
            "finish_reason": None,
        }],
    })
    for chunk in split_pcm(pcm):
        yield sse({
            "id": message_id,
            "choices": [{
                "index": 0,
                "delta": {"audio": {
                    "id": audio_id,
                    "data": base64.b64encode(chunk).decode(),
                }},
                "finish_reason": None,
            }],
        })
        await asyncio.sleep(CHUNK_DURATION_MS / 1000)
    yield sse({
        "id": message_id,
        "choices": [{
            "index": 0,
            "delta": {},
            "finish_reason": "stop",
        }],
    })
    yield "data: [DONE]\n\n"

audio.transcript 别当成可选项。它不只是界面上的字幕，还会进入智能体的短期记忆。如果音频内容和 transcript 对不上，用户听到的和模型记住的就是两回事，下一轮上下文直接错乱。

五. 保护音频接口

import os

from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat/completions")
async def chat(
    payload: dict,
    authorization: str | None = Header(None),
):
    secret = os.getenv("CUSTOM_LLM_SHARED_SECRET")
    if not secret or authorization != f"Bearer {secret}":
        raise HTTPException(401, "invalid credential")
    if "audio" not in (payload.get("modalities") or ["audio"]):
        raise HTTPException(400, "audio modality is required")
    pcm = read_pcm(os.environ["CUSTOM_AUDIO_PCM_PATH"])
    transcript = os.environ["CUSTOM_AUDIO_TRANSCRIPT"]
    return StreamingResponse(
        stream_audio(transcript, pcm),
        media_type="text/event-stream",
    )

六. 配置智能体

import os

from agora_agent import Area, AsyncAgora
from agora_agent.agentkit import Agent, CustomLLM
from agora_agent.cn import FengmingSTT, MiniMaxTTS

client = AsyncAgora(
    area=Area.CN,
    app_id=os.environ["AGORA_APP_ID"],
    app_certificate=os.environ["AGORA_APP_CERTIFICATE"],
)
llm = CustomLLM(
    api_key=os.environ["CUSTOM_LLM_SHARED_SECRET"],
    base_url=os.environ["CUSTOM_LLM_URL"],
    model="shengwang-custom-audio",
    output_modalities=["audio"],
    params={"modalities": ["audio"]},
    greeting_message="你好，自定义音频模态已经就绪。",
    failure_message="音频服务暂时不可用，请稍后再试。",
    max_history=10,
)
inert_tts = MiniMaxTTS(
    key=os.environ["MINIMAX_API_KEY"],
    model="speech-01-turbo",
    voice_id="female-shaonv",
    sample_rate=16000,
    language_boost="Chinese",
)
agent = (
    Agent(
        client=client,
        turn_detection={"language": "zh-CN"},
        advanced_features={"enable_rtm": True},
        parameters={
            "audio_scenario": "chorus",
            "data_channel": "rtm",
            "enable_metrics": True,
            "enable_error_message": True,
        },
    )
    .with_stt(FengmingSTT())
    .with_llm(llm)
    .with_tts(inert_tts)
)

REST 接口本身允许纯音频模态不配 TTS，但 agora-agents==2.4.1 的级联 builder 仍然要求传一个 TTS 对象，所以示例里保留了 MiniMax 作为构建器的兼容配置。实际有 audio.data 返回时，服务不会把 LLM 文本交给它合成——留着它只是为了让构建器过关。

七. 运行与验证

cd code/custom-modalities
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env.local
uvicorn audio_service:app --host 0.0.0.0 --port 9400

先在本地验证 SSE：

curl -N http://127.0.0.1:9400/chat/completions \
  -H "Authorization: Bearer $CUSTOM_LLM_SHARED_SECRET" \
  -H 'Content-Type: application/json' \
  -d '{"stream":true,"modalities":["audio"],"messages":[{"role":"user","content":"播放欢迎语"}]}'

预期顺序是：先出现 audio.transcript，然后连续的 audio.data，最后 finish_reason:"stop" 和 [DONE]。

公网语音验收再确认这几项：播放速度正常、声道正确、没有爆音、字幕和音频对得上、能被打断、第二轮上下文正常、异常时有降级。

八. 故障排查

• 播放速度不对（听起来像快进或慢放）：几乎都是输入格式问题：不是 16 kHz 单声道 PCM，或者把带文件头的 WAV/MP3 当成裸 PCM 发了。
• 有声音但没有上下文：确保至少发送了一次完整的 audio.transcript。
• 字幕和音频不同步：需要额外返回带 start_ts 和 end_ts/duration 的 audio.words。
• 创建成功但一点声音都没有：检查 Custom LLM 的公网 HTTPS 是否可达、output_modalities 是否为 ["audio"]，以及每段 Base64 解码后拼起来是不是连续的 PCM。

九. 下一步

音频自己掌控之后，填充语也可以换成你自己的录音，而不是合成音。

《语音智能体的填充语与优雅退出：消除对话空白》