在线咨询

专属客服在线解答，提供专业解决方案

工单支持

专业技术支持团队，随时响应服务需求

首页博客正文

语音智能体打断实战：VAD、关键词与不可打断

2026-07-27

开发者教程

打断做得好不好，是语音助手「像不像人」的分水岭，也是最难调的一环。

它不是「检测到声音就停止播放」那么简单：调得太敏感，键盘声、咳嗽、一句「嗯」都能把智能体打断；调得太迟钝，用户已经开口了它还在自顾自地念，变成抢话。

下面把三种模式都实现一遍：用户开口即可打断、只有特定关键词才能打断、当前回答不可打断，并给出一组可以直接用的 VAD 参数基线。

配好之后你会看到：智能体正在长篇回复时，你说一句「打断你，别说了」，播报立刻停下，新问题被完整识别并继续处理。

> 开始之前　先跑通《Next.js 搭建浏览器端语音智能体》，这篇只换智能体侧的打断配置，客户端一行不用改。

一. 实现效果与架构

智能体正在播报
  → turn_detection.start_of_speech 连续检测用户语音 160 ms
  → interruption 判断是否停止当前播报
  → turn_detection.end_of_speech 等待用户安静 480 ms
  → 完整新问句进入 DeepSeek

这里有两个配置项，得一起看：turn_detection 判断用户什么时候开始说、什么时候说完；interruption 决定用户开口之后，智能体当前这段播报该怎么办。只配一个是不够的。

二. 准备工作

已经跑通 Python 或《Next.js 搭建浏览器端语音智能体》。
继续用同一套 App ID、App Certificate、DeepSeek Key 和 MiniMax Key。
Python 3.10 或更高版本，推荐 3.11。
带麦克风的电脑和允许麦克风权限的浏览器。

三. 三种模式怎么选

`INTERRUPTION_MODE`	行为	适用场景
`start_of_speech`	用户正常开口就停止当前播报	默认对话体验，推荐
`keywords`	只有「停一下」等关键词才触发打断	嘈杂环境，或不希望被随便打断的业务播报
`disabled`	当前会话不可中途打断	合规条款、关键风险提示

选了不可打断，还要再定一个策略：

append：先把当前内容播完，再处理用户刚才说的话。
ignore：播报期间用户说的话直接丢弃。

一般选 append。ignore 会让用户觉得「它根本没听见我说话」，体验上很挫败。毕竟用户看不到界面反馈，只能靠听。

四. 项目结构

interruption-agent/
├─ .env.local
├─ requirements.txt
├─ agent_service.py
└─ app.py

浏览器端继续用 Next.js 快速开始那份客户端，把 NEXT_PUBLIC_AGENT_API_BASE 指过来就行。

五. 安装依赖

创建 requirements.txt：

agora-agents==2.4.1
fastapi==0.116.1
httpx==0.28.1
python-dotenv==1.1.1
uvicorn==0.35.0

安装：

python3.11 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

六. 配置环境变量

创建 .env.local：

AGORA_APP_ID=REPLACE_WITH_APP_ID
AGORA_APP_CERTIFICATE=REPLACE_WITH_APP_CERTIFICATE
DEEPSEEK_API_KEY=REPLACE_WITH_DEEPSEEK_API_KEY
DEEPSEEK_URL=https://api.deepseek.com/chat/completions
DEEPSEEK_MODEL=deepseek-v4-flash
MINIMAX_API_KEY=REPLACE_WITH_MINIMAX_API_KEY
MINIMAX_TTS_MODEL=speech-01-turbo
MINIMAX_VOICE_ID=female-shaonv
AGENT_GREETING=你好，我是可以自然打断的声网 AI 助手。想聊什么？
WEB_ORIGINS=http://localhost:3000,http://127.0.0.1:3000

# start_of_speech、keywords 或 disabled
INTERRUPTION_MODE=start_of_speech
INTERRUPTION_KEYWORDS=停一下,等一下,先别说了
INTERRUPTION_DISABLED_STRATEGY=append

注意：改完模式要重新创建智能体会话才生效，光重启服务不够。

七. 智能体服务完整代码

创建 agent_service.py：

"""声网语音打断示例。"""

from __future__ import annotations

import os
from typing import Any

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


def interruption_config() -> dict[str, Any]:
    """把环境变量转换为 SDK 2.4.1 支持的统一打断配置。"""
    mode = os.getenv("INTERRUPTION_MODE", "start_of_speech")
    if mode == "keywords":
        keywords = [
            item.strip()
            for item in os.getenv(
                "INTERRUPTION_KEYWORDS", "停一下,等一下,先别说了"
            ).split(",")
            if item.strip()
        ]
        return {
            "enable": True,
            "mode": "keywords",
            "keywords_config": {"trigger_keywords": keywords},
        }
    if mode == "disabled":
        strategy = os.getenv("INTERRUPTION_DISABLED_STRATEGY", "append")
        if strategy not in {"append", "ignore"}:
            raise ValueError("INTERRUPTION_DISABLED_STRATEGY 只能是 append 或 ignore")
        return {
            "enable": False,
            "disabled_config": {"strategy": strategy},
        }
    if mode != "start_of_speech":
        raise ValueError(
            "INTERRUPTION_MODE 只能是 start_of_speech、keywords 或 disabled"
        )
    return {"enable": True, "mode": "start_of_speech"}


class AgentService:
    def __init__(self) -> None:
        required = (
            "AGORA_APP_ID",
            "AGORA_APP_CERTIFICATE",
            "DEEPSEEK_API_KEY",
            "MINIMAX_API_KEY",
        )
        missing = [name for name in required if not os.getenv(name)]
        if missing:
            raise RuntimeError(f"缺少服务端环境变量：{', '.join(missing)}")

        self.client = AsyncAgora(
            area=Area.CN,
            app_id=os.environ["AGORA_APP_ID"],
            app_certificate=os.environ["AGORA_APP_CERTIFICATE"],
        )
        self.sessions: dict[str, Any] = {}

    def build_agent(self) -> Agent:
        llm = DeepSeekLLM(
            api_key=os.environ["DEEPSEEK_API_KEY"],
            base_url=os.getenv(
                "DEEPSEEK_URL", "https://api.deepseek.com/chat/completions"
            ),
            model=os.getenv("DEEPSEEK_MODEL", "deepseek-v4-flash"),
            system_messages=[
                {
                    "role": "system",
                    "content": "你是简洁、自然的中文语音助手。",
                }
            ],
            greeting_message=os.getenv(
                "AGENT_GREETING",
                "你好，我是可以自然打断的声网 AI 助手。想聊什么？",
            ),
            failure_message="抱歉，我刚才没有处理成功，请再说一次。",
            max_history=20,
            temperature=0.6,
            max_tokens=512,
            params={"thinking": {"type": "disabled"}},
        )
        tts = MiniMaxTTS(
            key=os.environ["MINIMAX_API_KEY"],
            model=os.getenv("MINIMAX_TTS_MODEL", "speech-01-turbo"),
            voice_id=os.getenv("MINIMAX_VOICE_ID", "female-shaonv"),
            sample_rate=16000,
            language_boost="Chinese",
        )

        return (
            Agent(
                client=self.client,
                turn_detection={
                    "language": "zh-CN",
                    "config": {
                        "speech_threshold": 0.5,
                        "start_of_speech": {
                            "mode": "vad",
                            "vad_config": {
                                "interrupt_duration_ms": 160,
                                "prefix_padding_ms": 300,
                            },
                        },
                        "end_of_speech": {
                            "mode": "vad",
                            "vad_config": {"silence_duration_ms": 480},
                        },
                    },
                },
                interruption=interruption_config(),
                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(tts)
        )

    async def start(self, channel: str, agent_uid: int, user_uid: int) -> str:
        session = self.build_agent().create_async_session(
            channel=channel,
            agent_uid=str(agent_uid),
            remote_uids=[str(user_uid)],
            enable_string_uid=False,
            idle_timeout=120,
            expires_in=3600,
        )
        agent_id = await session.start()
        self.sessions[agent_id] = session
        return agent_id

    async def stop(self, agent_id: str) -> None:
        session = self.sessions.pop(agent_id, None)
        if session is not None:
            await session.stop()
            return
        await self.client.stop_agent(agent_id)

SDK 2.4.1 里的关键词字段叫 trigger_keywords，不是 keywords。写错了不报错，只是关键词永远不生效，而这种问题排查起来最费时间。三种模式照抄即可。

八. FastAPI 接口完整代码

创建 app.py：

"""FastAPI：生成 Token、启动打断示例、停止智能体。"""

from __future__ import annotations

import os
import random
import time
from functools import lru_cache

from agora_agent.agentkit.token import generate_convo_ai_token
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field

from agent_service import AgentService

load_dotenv(".env.local")
load_dotenv(".env")

app = FastAPI(title="声网语音打断示例", version="1.0.0")
web_origins = [
    origin.strip()
    for origin in (
        os.getenv("WEB_ORIGINS")
        or os.getenv("WEB_ORIGIN")
        or "http://localhost:3000,http://127.0.0.1:3000"
    ).split(",")
    if origin.strip()
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=web_origins,
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["Content-Type", "Authorization"],
)


@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok"}


@lru_cache
def service() -> AgentService:
    return AgentService()


class StartRequest(BaseModel):
    channelName: str = Field(min_length=1)
    rtcUid: int = Field(gt=0)
    userUid: int = Field(gt=0)


class StopRequest(BaseModel):
    agentId: str = Field(min_length=1)


@app.get("/get_config")
async def get_config() -> dict:
    app_id = os.getenv("AGORA_APP_ID")
    certificate = os.getenv("AGORA_APP_CERTIFICATE")
    if not app_id or not certificate:
        raise HTTPException(500, "服务端缺少 AGORA_APP_ID 或 AGORA_APP_CERTIFICATE")

    user_uid = random.randint(1000, 9_999_999)
    agent_uid = random.randint(10_000_000, 99_999_999)
    channel = f"shengwang-interruption-{int(time.time())}-{random.randint(100, 999)}"
    token = generate_convo_ai_token(
        app_id=app_id,
        app_certificate=certificate,
        channel_name=channel,
        uid=user_uid,
        token_expire=3600,
    )
    return {
        "code": 0,
        "data": {
            "app_id": app_id,
            "token": token,
            "uid": str(user_uid),
            "agent_uid": str(agent_uid),
            "channel_name": channel,
        },
    }


@app.post("/startAgent")
async def start_agent(body: StartRequest) -> dict:
    try:
        agent_id = await service().start(body.channelName, body.rtcUid, body.userUid)
        return {"code": 0, "data": {"agent_id": agent_id}}
    except Exception as exc:
        raise HTTPException(502, f"启动智能体失败：{type(exc).__name__}") from exc


@app.post("/stopAgent")
async def stop_agent(body: StopRequest) -> dict:
    try:
        await service().stop(body.agentId)
        return {"code": 0, "message": "stopped"}
    except Exception as exc:
        raise HTTPException(502, f"停止智能体失败：{type(exc).__name__}") from exc

九. 接上浏览器客户端

用 Next.js 快速开始的客户端，在它的 .env.local 里设置：

NEXT_PUBLIC_AGENT_API_BASE=http://localhost:8000

这个工程和《Python + FastAPI 搭建中文语音智能体》用的是同一套 /get_config、/startAgent、/stopAgent 契约，所以客户端一行代码都不用改。

十. 跑起来

启动打断服务：

source .venv/bin/activate
uvicorn app:app --host 127.0.0.1 --port 8000

再启动 Next.js 客户端：

npm install
npm run dev

打开 http://localhost:3000，允许麦克风权限，点「开始对话」。

十一. 怎么测

这三种模式必须分开测，光看代码跑通了没意义，打断的手感只能靠耳朵判断。

自然开口打断

设置 INTERRUPTION_MODE=start_of_speech。
问一个智能体要答至少 20 秒的问题。
它开始播报后，用正常音量说「停一下，换个问题」。
确认旧播报立刻停下，新问句被完整识别并回答。

关键词打断

设置 INTERRUPTION_MODE=keywords。
先在播报期间说一句不在列表里的话，确认不会被打断。
再说「停一下」或「先别说了」，确认打断生效。

不可打断

设置 INTERRUPTION_MODE=disabled，INTERRUPTION_DISABLED_STRATEGY=append。
播报期间说出新问题。
确认当前播报完整结束后，智能体才处理新问题。

三个场景都要在安静办公室、有键盘声、有扬声器回音三种环境下各测一遍，记录「用户开口到停止播报」的延迟和误触发次数。安静环境下调好的参数，到开放工位可能完全不能用。

十二. 参数调优

speech_threshold：越低越容易把声音判定为人声。环境嘈杂就往上调。
interrupt_duration_ms：用户持续说话多久算「开始说话」。越小打断越快，也越容易误触发。
prefix_padding_ms：保留用户开口前多少毫秒的音频。开头丢字就往上加。
silence_duration_ms：用户安静多久算「说完了」。越小响应越快，但也越容易把中间的停顿当成结束。

本文给的 0.5 / 160 / 300 / 480 是中文近场耳机的起始基线，不是万能值。换设备、换环境都得重新调。

十三. 故障排查

环境声一响就停播：增大 interrupt_duration_ms，或调高 speech_threshold。
用户说了好几个字还不停：减小 interrupt_duration_ms，同时检查麦克风音量和回声消除。
新问句开头丢字：增大 prefix_padding_ms。
用户中间停顿一下就被抢答：增大 silence_duration_ms。
关键词不生效：确认字段写的是 keywords_config.trigger_keywords，并重新创建智能体会话。
不可打断模式下用户的问题丢了：用 append，别用 ignore。
浏览器报 DEVICE_NOT_FOUND：接上真实麦克风重新授权。

十四. 上线前还要做什么

生产环境建议按设备类型保存各自的参数（耳机、笔记本内置麦、车载麦的表现差异很大），并通过日志统计误打断率和打断延迟，用数据而不是感觉来调。

合规播报要用明确的会话状态来控制不可打断，而不是把整个智能体的打断能力长期关掉——那会毁掉其他所有场景的体验。

十五. 下一步

打断调顺之后，剩下的体验缺口是等待时的沉默和挂断时的截断。

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

在声网，连接无限可能

想进一步了解「对话式 AI 与实时互动」？欢迎注册，开启探索之旅。

注册体验

本博客为技术交流与平台行业信息分享平台，内容仅供交流参考，文章内容不代表本公司立场和观点，亦不构成任何出版或销售行为。