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

语音助手一旦要回答公司产品、客户订单、内部流程这类问题，就不能再指望大模型的参数记忆了——它不知道你们的东西，而且会一本正经地编。

解法是在自定义 LLM 网关里加一层 RAG：取出用户当前的问题，检索中文语料，把命中的证据作为 system message 一起发给 DeepSeek。做完之后，助手回答的是你的知识库里的内容，而不是它想象出来的内容。

> 开始之前　先完成《给语音智能体接入自定义 LLM：OpenAI 兼容网关》，本文的检索层就加在那个网关里。

一. 架构与准备工作

用户语音 → 凤鸣 ASR → /chat/completions
                              ├─ 提取最后一条 user message
                              ├─ 检索知识库 Top K
                              └─ 证据 + 对话 → DeepSeek SSE
                         MiniMax TTS ←────────────┘

准备好 DEEPSEEK_API_KEY 和 CUSTOM_LLM_SHARED_SECRET。

本文用内存语料跑通最小闭环，重点是让你看清 RAG 在这条链路里到底插在哪一步。真接业务时把 retrieve() 换成向量库、Elasticsearch 或你们已有的搜索服务即可，其余代码不用动。

二. 环境变量

DEEPSEEK_API_KEY=REPLACE_WITH_DEEPSEEK_KEY
CUSTOM_LLM_SHARED_SECRET=REPLACE_WITH_RANDOM_SECRET

三. 完整代码

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

app = FastAPI()
CORPUS = [
    {"title": "服务地址", "text": "声网对话式 AI 的 REST 路径包含 /cn/api/conversational-ai-agent。"},
    {"title": "客户端组件", "text": "实时字幕和智能体事件需要开启 RTM、data_channel、metrics 与 error message。"},
    {"title": "密钥安全", "text": "App Certificate、LLM API Key 和 TTS Key 只能保存在业务服务端。"},
]

def tokens(text: str) -> set[str]:
    return set(re.findall(r"[a-z0-9_]+|[\u4e00-\u9fff]", text.lower()))

def retrieve(query: str, top_k: int = 2):
    query_tokens = tokens(query)
    scored = [(len(query_tokens & tokens(d["title"] + d["text"])), d) for d in CORPUS]
    return [d for score, d in sorted(scored, key=lambda x: x[0], reverse=True)[:top_k] if score]

def last_user_text(messages: list[dict[str, Any]]) -> str:
    for message in reversed(messages):
        if message.get("role") == "user" and isinstance(message.get("content"), str):
            return message["content"]
    return ""

def clean_messages(messages):
    allowed = {"role", "content", "name", "tool_calls", "tool_call_id"}
    return [{k: v for k, v in m.items() if k in allowed} for m in messages]

async def answer(payload: dict[str, Any]) -> AsyncIterator[bytes]:
    hits = retrieve(last_user_text(payload.get("messages", [])))
    context = "\n".join(f"[{d['title']}] {d['text']}" for d in hits)
    grounding = {"role": "system", "content":
        "只能根据下面资料回答；资料不足时明确说不知道。\n资料：\n" + (context or "无")}
    upstream = {
        "model": "deepseek-v4-flash",
        "messages": [grounding, *clean_messages(payload.get("messages", []))],
        "temperature": 0.3, "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_answer(payload: dict[str, Any]) -> AsyncIterator[bytes]:
    try:
        async for chunk in answer(payload):
            yield chunk
    except (httpx.HTTPError, KeyError, RuntimeError):
        yield fallback_sse("知识库服务暂时不可用，请稍后再试。")

@app.post("/chat/completions")
async def chat(payload: dict[str, Any], authorization: str | None = Header(None)):
    if authorization != f"Bearer {os.environ['CUSTOM_LLM_SHARED_SECRET']}":
        raise HTTPException(401, "invalid credential")
    return StreamingResponse(safe_answer(payload), media_type="text/event-stream")

智能体那边的配置和 Custom LLM 教程完全一样，只要把 base_url 指向这个 RAG 服务的公网 /chat/completions。

四. 运行与验证

uvicorn rag_service:app --host 0.0.0.0 --port 9000
curl -N http://localhost:9000/chat/completions \
  -H "Authorization: Bearer $CUSTOM_LLM_SHARED_SECRET" \
  -H "Content-Type: application/json" \
  -d '{"messages":[{"role":"user","content":"密钥应该放在哪里？"}],"stream":true}'

预期回答应该命中「密钥安全」那篇文档，明确说出密钥只保存在业务服务端。检索或模型出问题时，会输出一段能正常播报的降级提示，不会让用户对着沉默的助手发呆。

上线后建议记录 SSE 首包时间和命中的文档 ID，方便定位「答得慢」和「答错了」分别出在哪一环。但别把用户的完整敏感文本写进日志。

五. 故障排查

• 总是检索不到：本文的字符重合算法只是教学用的最小实现，中文效果有限。生产环境换成合适的中文 embedding 和召回策略。
• 答案和证据对不上：把 grounding 放在对话消息的前面，并把 temperature 调低。
• 多轮对话命中错误：检索 query 要以当前这条 user message 为主。遇到「那它多少钱」这种指代句，先用模型把问题改写完整再检索。

六. 下一步

知识库解决了「它知不知道」，下一步解决「它能不能动手」——调用你的业务函数。

《函数调用 + SQLite：让语音智能体记住上次说的话》