给语音智能体接入自定义 LLM：OpenAI 兼容网关

自定义 LLM 的价值不只是「换个模型」。真正的好处是，鉴权、审计、提示词、限流、降级、业务逻辑这些东西可以全部收进你自己的网关，而不是散落在各处。

要搭的东西不复杂：一个 /chat/completions 服务，对外保持 OpenAI Chat Completions 兼容，对内接 DeepSeek V4 Flash，中间那层你想加什么加什么。

> 开始之前　这篇复用《Python + FastAPI 搭建中文语音智能体》里的环境变量和凭证，先把那套最小后端跑起来。

一. 架构与准备工作

凤鸣 ASR → 声网云端 → https://llm.example.com/chat/completions → DeepSeek
                                             └─ 鉴权/过滤/审计
MiniMax TTS ←─────────────────────────────────────────────────────┘

除了 《Python + FastAPI 搭建中文语音智能体》用到的环境变量，还要加三个：

CUSTOM_LLM_URL=https://llm.example.com/chat/completions
CUSTOM_LLM_SHARED_SECRET=REPLACE_WITH_RANDOM_SECRET
DEEPSEEK_API_KEY=REPLACE_WITH_DEEPSEEK_KEY

CUSTOM_LLM_URL 必须是声网云端能访问到的公网 HTTPS 地址。localhost、私网 IP、开发机上的临时地址都不行——请求是从云端发出来的，不是从你本机。

二. Custom LLM 服务完整代码

import json, os, time, uuid
from typing import Any, AsyncIterator
import httpx
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse

app = FastAPI()

def clean_messages(messages: list[dict[str, Any]]):
    # 声网会附带 turn_id/timestamp 等字段，不要原样发给上游模型。
    allowed = {"role", "content", "name", "tool_calls", "tool_call_id"}
    return [{k: v for k, v in item.items() if k in allowed} for item in messages]

def check_auth(authorization: str | None):
    expected = os.environ["CUSTOM_LLM_SHARED_SECRET"]
    if authorization != f"Bearer {expected}":
        raise HTTPException(401, "invalid custom LLM credential")

async def deepseek_stream(payload: dict[str, Any]) -> AsyncIterator[bytes]:
    upstream = {
        "model": "deepseek-v4-flash",
        "messages": clean_messages(payload.get("messages", [])),
        "temperature": payload.get("temperature", 0.6),
        "max_tokens": payload.get("max_tokens", 512),
        "stream": True,
        "thinking": {"type": "disabled"},
    }
    async with httpx.AsyncClient(timeout=60) as client:
        async with client.stream(
            "POST", "https://api.deepseek.com/chat/completions",
            headers={"Authorization": f"Bearer {os.environ['DEEPSEEK_API_KEY']}"},
            json=upstream,
        ) as response:
            response.raise_for_status()
            async for chunk in response.aiter_raw():
                yield chunk

def fallback_sse(message: str) -> bytes:
    body = {
        "id": f"chatcmpl-{uuid.uuid4().hex}",
        "object": "chat.completion.chunk",
        "created": int(time.time()),
        "model": "deepseek-v4-flash",
        "choices": [{"index": 0, "delta": {"content": message},
                     "finish_reason": "stop"}],
    }
    return (f"data: {json.dumps(body, ensure_ascii=False)}\n\n"
            "data: [DONE]\n\n").encode()

async def safe_stream(payload: dict[str, Any]) -> AsyncIterator[bytes]:
    try:
        async for chunk in deepseek_stream(payload):
            yield chunk
    except (httpx.HTTPError, KeyError, RuntimeError):
        # 对用户返回可播报内容，不泄露上游响应、URL 或凭证。
        yield fallback_sse("模型服务暂时不可用，请稍后再试。")

@app.post("/chat/completions")
async def chat(payload: dict[str, Any], authorization: str | None = Header(None)):
    check_auth(authorization)
    return StreamingResponse(safe_stream(payload), media_type="text/event-stream")

@app.get("/health")
async def health():
    return {"status": "ok"}

上游的 SSE 必须原样流式转发，不能先攒成完整回答再一次性返回。这个细节直接决定体验好坏。攒完再发的话，用户要等模型把整段话生成完才能听到第一个字，首包延迟会非常明显。

三. 在智能体里换掉 LLM

凤鸣 ASR 和 MiniMax TTS 都不用动，把快速开始里的 DeepSeekLLM 换成 CustomLLM 就行：

from agora_agent.agentkit import CustomLLM

llm = CustomLLM(
    api_key=os.environ["CUSTOM_LLM_SHARED_SECRET"],
    base_url=os.environ["CUSTOM_LLM_URL"],
    model="shengwang-deepseek-gateway",
    system_messages=[{
        "role": "system",
        "content": "你是一名中文语音助手，回答应简短、口语化。",
    }],
    greeting_message="你好，有什么可以帮你？",
    failure_message="服务暂时繁忙，请稍后再试。",
    max_history=20,
)
agent = agent.with_llm(llm)

四. 运行与验证

uvicorn custom_llm:app --host 0.0.0.0 --port 9000
curl https://llm.example.com/health
curl -N https://llm.example.com/chat/completions \
  -H "Authorization: Bearer $CUSTOM_LLM_SHARED_SECRET" \
  -H "Content-Type: application/json" \
  -d '{"messages":[{"role":"user","content":"用一句话介绍声网"}],"stream":true}'

健康检查应该返回 {"status":"ok"}；第二个请求应该连续吐出 data: 开头的 SSE 块，最后以 [DONE] 收尾。

另外，上游模型挂掉时服务会转换成一段标准 SSE 降级回复——保证用户听到的是「我这边出了点问题」，而不是一片死寂。日志里注意别打印明文密钥。

五. 故障排查

• 智能体报 401：检查 CustomLLM 配置里传的 Key 和服务端的 CUSTOM_LLM_SHARED_SECRET 是不是同一个。
• 连接超时：确认公网 DNS 解析正常、TLS 证书链完整、443 端口通。别用自签证书。
• DeepSeek 返回参数错误：转发前要把附加的 turn_id、timestamp 这些字段删掉，上游不认识它们。

六. 下一步

网关搭好了，中间那层就能加东西。最常见的第一件事是接知识库，让它别再瞎编。

《给语音智能体接私有知识库：RAG 实战》