语音智能体内容过滤：在进 TTS 之前拦住它

文本产品出问题，你还能在前端把内容藏起来。语音不行——模型的输出会直接进 TTS，用户先听到了，你再隐藏字幕已经没有意义。

所以过滤必须往前放，放到 Custom LLM 网关这一层：进模型之前先脱敏，模型返回之后再过一遍，只有处理干净的文本才允许进入 TTS。

装上之后，命中规则的内容既不会出现在字幕里，也不会被念出来。

> 开始之前　先完成《给语音智能体接入自定义 LLM：OpenAI 兼容网关》，过滤器就插在那一层。

一. 架构与边界

凤鸣 ASR → 声网云端 → Custom LLM 网关
                         ├─ Bearer 鉴权
                         ├─ 输入关键词/PII 过滤
                         ├─ DeepSeek 生成
                         └─ 输出二次过滤 → SSE
MiniMax TTS ← 声网云端 ←────────────────┘

先把话说清楚：本文的关键词和正则规则是用来展示工程闭环的，不等于一套完整的内容安全方案。真做合规业务，还要接业务分类器、模型审核、人工处置、审计和申诉流程。这篇解决的是「过滤器应该插在哪、怎么插不破坏体验」的问题。

二. 环境变量

DEEPSEEK_API_KEY=REPLACE_WITH_DEEPSEEK_KEY
CUSTOM_LLM_SHARED_SECRET=REPLACE_WITH_RANDOM_SECRET
FILTER_BANNED_TERMS=内部代号|测试禁词

智能体那边继续用 《给语音智能体接入自定义 LLM：OpenAI 兼容网关》的四项凭证，把 CUSTOM_LLM_URL 指向本文服务的公网 HTTPS /chat/completions。

三. 中文关键词与隐私过滤

import os, re

REDACTION = "相关内容已过滤。"
PHONE = re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)")
ID_CARD = re.compile(r"(?<!\d)\d{17}[\dXx](?!\d)")

def banned_terms() -> list[str]:
    return [item.strip().lower() for item in os.getenv(
        "FILTER_BANNED_TERMS", "内部代号|测试禁词"
    ).split("|") if item.strip()]

def is_allowed(text: str) -> bool:
    lowered = text.lower()
    return (not any(term in lowered for term in banned_terms())
            and PHONE.search(text) is None
            and ID_CARD.search(text) is None)

def filter_text(text: str) -> str:
    sentences = [item for item in re.split(r"(?<=[。！？!?])", text) if item]
    return "".join(part if is_allowed(part) else REDACTION for part in sentences)

def clean_messages(messages: list[dict]) -> list[dict]:
    cleaned = []
    for message in messages:
        role, content = message.get("role"), message.get("content")
        if role in {"system", "user", "assistant", "tool"} and isinstance(content, str):
            cleaned.append({"role": role, "content": filter_text(content)})
    return cleaned

命中之后是整句替换，而不是把敏感词单独抠掉。抠词看着干净，但剩下的上下文往往还能推断出原文，「我的手机号是 ███」这种反而更显眼。整句换掉更安全。

手机号和身份证号的正则只覆盖常见格式，替代不了成熟的 PII 检测器，真上量的时候该换还得换。

四. 调用 DeepSeek 并二次过滤

from typing import Any
import httpx

async def generate_answer(payload: dict[str, Any]) -> str:
    request_body = {
        "model": "deepseek-v4-flash",
        "messages": [
            {"role": "system", "content":
                "你是中文语音助手。不要生成违法、有害或泄露隐私的内容；"
                "不能安全回答时请给出简短说明。"},
            *clean_messages(payload.get("messages", [])),
        ],
        "temperature": 0.3,
        "max_tokens": min(int(payload.get("max_tokens", 512)), 512),
        "stream": False,
        "thinking": {"type": "disabled"},
    }
    async with httpx.AsyncClient(timeout=60) as client:
        response = await client.post(
            "https://api.deepseek.com/chat/completions",
            headers={"Authorization": f"Bearer {os.environ['DEEPSEEK_API_KEY']}"},
            json=request_body,
        )
        response.raise_for_status()
        answer = response.json()["choices"][0]["message"]["content"]
    return filter_text(str(answer))

这里做了一个明确的取舍：为了保证过滤器拿到的是完整句子，这个最小示例先收完整段回答，再由本地网关吐 SSE。代价是首包延迟比直接透传高——句子没收完就没法判断该不该拦。

生产环境的做法是折中：按中文句末标点缓冲上游流，每凑齐一句就审核一句、下发一句。既能拦住内容，首包延迟也不至于让用户等到怀疑人生。

五. 返回 OpenAI 兼容 SSE

import json, time, uuid
from collections.abc import AsyncIterator
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse

app = FastAPI()

def sse_chunk(chunk_id: str, content: str, finish_reason=None) -> str:
    body = {"id": chunk_id, "object": "chat.completion.chunk",
            "created": int(time.time()), "model": "deepseek-v4-flash",
            "choices": [{"index": 0,
                "delta": {"content": content} if content else {},
                "finish_reason": finish_reason}]}
    return f"data: {json.dumps(body, ensure_ascii=False)}\n\n"

async def stream_answer(payload: dict) -> AsyncIterator[str]:
    try:
        answer = await generate_answer(payload)
    except (httpx.HTTPError, KeyError, RuntimeError, ValueError):
        answer = "内容服务暂时不可用，请稍后再试。"
    chunk_id = f"chatcmpl-{uuid.uuid4().hex}"
    for character in answer:
        yield sse_chunk(chunk_id, character)
    yield sse_chunk(chunk_id, "", "stop")
    yield "data: [DONE]\n\n"

@app.post("/chat/completions")
async def chat(payload: dict, authorization: str | None = Header(None)):
    expected = os.getenv("CUSTOM_LLM_SHARED_SECRET")
    if not expected or authorization != f"Bearer {expected}":
        raise HTTPException(401, "invalid credential")
    return StreamingResponse(stream_answer(payload), media_type="text/event-stream")

注意 fail-closed 这个设计：服务端没配共享密钥时必须直接拒绝访问，不能「没配就当作不需要鉴权」。安全组件降级成无鉴权，比没有这个组件更危险。

六. 智能体配置

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-content-filter-gateway",
    system_messages=[{"role": "system", "content":
        "你是中文语音助手，回答简短、自然。"}],
    greeting_message="你好，有什么可以帮你？",
    failure_message="内容服务暂时不可用，请稍后再试。",
    max_history=20,
)
agent = agent.with_llm(llm)

七. 运行与验证

cd code/content-filtering
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env.local
uvicorn filter_service:app --host 0.0.0.0 --port 9000

先在本地验证鉴权和 SSE，再部署到公网 HTTPS：

curl -N http://localhost:9000/chat/completions \
  -H "Authorization: Bearer $CUSTOM_LLM_SHARED_SECRET" \
  -H 'Content-Type: application/json' \
  -d '{"messages":[{"role":"user","content":"请复述测试禁词和手机号 13800138000。"}],"stream":true}'

预期是 SSE 里看不到原关键词和手机号，只有过滤提示，最后以 [DONE] 收尾。

公网端到端还要再确认：云端 TLS 调用成功、被过滤的内容确实没进字幕和 TTS、上游超时有降级播报、日志里没有留下敏感原文。

八. 故障排查

• 智能体返回 401：CustomLLM.api_key 和网关的 CUSTOM_LLM_SHARED_SECRET 必须一致。
• 延迟明显变高：这是整段过滤的必然代价。换成逐句缓冲审核，并盯住 p95。
• 误杀太多：关键词规则要有版本、生效范围、例外名单和灰度机制。把所有业务禁词硬编码在代码里，第一次误杀就得发版，这在真实业务里撑不住。

九. 下一步

过滤是拦在前面，审计是记在后面。两件事一起做，合规链路才闭环。

《语音智能体的服务端 Webhook：验签、幂等与订阅》