企业官网建设流程全解析

AgentScope Realtime Voice Agent 实战指南：用麦克风与 DashScope 语音模型实时对话

【免费下载链接】agentscopeBuild and run agents you can see, understand and trust.项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope

AgentScope 的realtime模块提供了一条完整的语音 Agent 通路：把 DashScope 的语音到语音（speech-to-speech）实时模型、本地声卡采集/播放与一个轮次状态机组装起来，实现"说话——听回复——打断插话——让模型调用工具"的完整闭环。本文将围绕官方示例 examples/realtime/README.md 展开，结合examples/realtime/local_mic.py及src/agentscope/realtime与src/agentscope/agent/_realtime的源码实现，带你从安装、运行、换模型、调试设备，一路深入到 barge-in、播放淡出、权限确认、延迟指标等底层机制，最终掌握用 AgentScope 搭建可打断、可调用工具的实时语音助手的完整方法。

一、整体架构：三个组件各司其职

示例 local_mic.py 的核心思路是把实时语音系统拆成三个互相独立的部分，由一段 30 行左右的胶水代码接起来：

三者之间存在明确的所有权约定，这一点在 RealtimeAgent 类文档 中有详细说明：

• Agent 拥有模型会话：agent.connect()打开会话，会话的生命周期与 agent 一样长，而不是与某一次传输绑定；
• 调用方拥有 transport：传输（声卡、浏览器连接、或你自己驱动的队列）由创建它的人持有，agent 只是"借用"；
• 一次agent.reply_stream(transport)同时借用两者，直到传输结束。

正因为会话不随传输消亡，客户端断开重连不会丢失模型会话；反过来，长时间静默导致模型会话超时后，下一次开口会自动重建会话，完全不用动传输层。

三者的类定义分别位于 realtime/_base.py（RealtimeModelBase）、realtime/_transport/_base.py（TransportBase）与 agent/_realtime/_agent.py（RealtimeAgent）。其中TransportBase的接口刻意保持最小：只有start/close/incoming/send_audio/clear_audio/playout六个方法，agent 无法分辨音频来自本地声卡、远程浏览器还是一个测试队列，这为后续接入 Web 端等更多传输形态留好了扩展点。

二、环境准备

Python 版本与安装

• 需要Python 3.11 或更高版本；
• 通过 extra 安装 realtime 支持：

pip install "agentscope[realtime]"

realtimeextra 在 pyproject.toml 中定义为两个依赖：

realtime = [ "websockets>=13", "sounddevice", ]

• websockets用于与 DashScope 建立实时 WebSocket 会话；
• sounddevice负责本地声卡的采集与播放，它底层依赖 PortAudio。

PortAudio 安装提示：macOS 和 Windows 上sounddevice会自带捆绑的 PortAudio；在 Debian/Ubuntu 上需要手动安装：

apt install libportaudio2

API Key

准备一个具备 realtime 模型访问权限的 DashScope API Key。示例运行时通过环境变量DASHSCOPE_API_KEY读取，缺失时会直接退出并提示：

api_key = os.environ.get("DASHSCOPE_API_KEY") if not api_key: raise SystemExit("Set DASHSCOPE_API_KEY first.")

三、运行第一个语音会话

最小启动命令

export DASHSCOPE_API_KEY=sk-... python examples/realtime/local_mic.py

启动后终端会打印[qwen-audio-3.0-realtime-plus] listening... (Ctrl-C to quit)，此时直接对着麦克风说话，就能听到模型回复；在回复过程中继续说话即可打断（barge-in）；按Ctrl-C退出。

示例创建的 agent 名为Friday，系统提示词是"你是一个中文语音助手，回答尽量简短"，你可以在 local_mic.py 中随意调整。

默认模型与模型卡解析机制

默认模型是qwen-audio-3.0-realtime-plus。任何该凭证下卡片列出的模型都能工作，因为模型类不是硬编码的，而是从模型卡（model card）反向解析出来的，与服务层（service layer）采用同一套机制：

REALTIME_MODEL=qwen3.5-omni-flash-realtime python examples/realtime/local_mic.py

解析过程（见 local_mic.py）分三步：

1. 通过credential.list_realtime_models()拿到该凭证支持的全部实时模型卡片，按名字建索引；
2. 通过credential.get_realtime_model_classes()拿到所有实时模型适配器类，按model_type建索引；
3. 取name对应卡片的model_type，实例化对应的模型类，并把卡片对象一并传入。

如果名字未知，程序会直接退出并打印可用模型列表：

Unknown model 'xxx'!; try: ['qwen-audio-3.0-realtime-flash', 'qwen-audio-3.0-realtime-plus', ...]

list_realtime_models与get_realtime_model_classes的默认实现在 credential/_base.py：前者遍历该凭证支持的每个实时模型类，把每张卡片打上适配器的model_type标记后汇总返回；DashScopeCredential 则覆盖了get_realtime_model_classes，返回 DashScope 的两个实时模型类。模型卡本身由 RealtimeModelCard 定义，从_models目录下的 YAML 文件加载（list_from_directory），加载失败的卡片会被跳过并打警告日志。

模型卡都记录了哪些能力

以默认模型 qwen-audio-3.0-realtime-plus.yaml 为例：

• 采样率：input_sample_rate: 16000（采集 PCM），output_sample_rate: 24000（播放 PCM）；
• 工具：supports_tools: true，可接受工具 schema；
• 上下文限制：max_audio_turns: 50（最多记住 50 轮音频对话），max_audio_duration_s: 300（累计音频时长上限），两者同时生效，任一超限都会丢弃最旧的轮次；
• 输入类型：audio/pcm与text/plain（支持文字输入）；
• 音色：通过parameter_overrides.voice提供默认值longanqian及一组可选枚举。

而 qwen3-omni-flash-realtime.yaml 则不同：supports_tools: false（函数调用只在 qwen3.5 系列提供）、max_audio_turns: 8（只记得 8 轮）、max_session_duration_s: 7200（会话最长 2 小时），且仅接受音频输入。这些差异会在运行时直接影响 agent 的行为——详见"Tips"一节。

四、理解reply_stream：一次调用把两端接起来

示例的核心循环非常简洁（local_mic.py）：

async with agent, transport: async for event in agent.reply_stream(transport): match event: ...

从源码看（RealtimeAgent.reply_stream），这个流做了几件关键的事：

1. 校验采样率：transport.input_sample_rate必须等于model.input_sample_rate，输出端同理，不匹配直接抛ValueError——因为音频是两端原样转发的，重采样属于传输层职责，不在这里做；
2. 双泵并行：_pump_uplink（transport → model）与_pump_downlink（model → transport）各自独立运行，reply_stream只是消费 agent 事件队列；
3. 传输结束时善后：如果传输先结束（比如客户端断开），agent 会先_barge_in()切断仍在播放的回复，把产生的ReplyEnd事件在流结束前吐给调用方，避免事件泄漏到下一次运行；传输抛出的异常不会被包装成"正常断开"，会原样上抛。

事件流里值得注意的几点：

• 用户的语音也被报告为一次回复：ReplyStartEvent(role="user")出现时示例把它记为user_turns集合，随后的TextBlockDeltaEvent就按"用户转写"处理，在终端回显[you] ...；
• 工具调用、权限请求、工具结果分别对应RequireUserConfirmEvent、ToolResultStartEvent、ToolResultEndEvent；
• 每次回复结束时（ReplyEndEvent，且不是用户轮次），示例打印一行诊断信息（详见下一节）。

五、诊断输出：每轮延迟指标的解读

每次助手回复结束后，终端都会出现一行诊断：

(completed | ttfb=540ms | e2e=1120ms) context[-1] = assistant: '好的，马上为您查询。'

各字段含义：

• finished_reason：本回合如何结束，如completed（正常完成）、interrupted（被打断）、error（出错）；
• ttfb（time to first byte）：模型首字节音频到达的时间，即backend_ttfb，衡量"模型侧响应速度"；
• e2e（end-to-end）：用户的端到端等待时间，即e2e_latency，从用户语音结束到听到第一个音频字节；
• context[-1]：当前上下文最后一条助手消息的内容——如果是 barge-in 打断的，这里只显示你实际听到的那部分文本。

这个"只保留实际听到的部分"是 barge-in 的核心语义，靠_Reply对象维护"文本增量与音频时间轴的对应关系"来实现：每收到一段转写增量就记录(当前音频时长, 当前文本长度)到marks列表（见 agent/_realtime/_agent.py），打断时用spoken_prefix(played_ms)反查"播了这么多毫秒对应的文本前缀"。由于转写通常比音频稍微超前，算法刻意保守——宁可多保留一个词，也不丢词。指标本身由 TurnMetrics 定义，通过agent.last_turn_metrics访问。

六、工具调用与终端权限确认

示例为 agent 注册了四个内置工具：

toolkit=Toolkit(tools=[Bash(), Edit(), Write(), Read()]),

（对应 local_mic.py，工具实现位于 tool/_builtin。）当模型请求调用某个需要权限的工具时，终端会出现交互提示：

[permission] Bash({"command": "ls"}) allow? [y/N]

输入y或yes放行，回车或输入其他内容则拒绝。拒绝后模型会收到Tool "Bash" denied by user.的结果，状态标记为DENIED。

权限提示在独立线程中运行（local_mic.py）：

answer = await loop.run_in_executor(None, input, " allow? [y/N] ")

这样等待用户输入期间，音频泵照常工作——你可以边考虑边听，也可以直接开口打断。而 agent 一侧给每次权限确认设置了五分钟超时：_ask_user里asyncio.wait_for(future, timeout=300)（agent/_realtime/_agent.py），超时或取消都按"未确认"处理。

底层流程在 RealtimeAgent._run_tool 中：工具调用先经过PermissionEngine的check_permission，按行为分派——ASK/PASSTHROUGH需要人工确认，DENY直接按策略拒绝；确认通过后经toolkit.call_tool流式执行，把结果以ToolResultBlock推回模型，并触发后续回复（request_response）。工具执行期间用户仍可打断，若打断发生在工具运行中，后续的续写会被跳过（通过比较reply_id判断）。

七、Tips：实战中的关键注意事项

1. 务必使用耳机

用外放时，麦克风会采集到 agent 自己的声音，提供商的 VAD（语音活动检测）会误以为"用户开始说话"，从而切掉正在播放的回复——agent 打断了自己。LocalAudioTransport不做回声消除（浏览器端通常会做），所以耳机是标准解法。

2. 不要用同一副蓝牙耳机同时走双向音频

macOS 在设备同时用于输入和输出时，会把 AirPods 之类的设备切换到免提（hands-free）配置文件，此时 PortAudio 在该设备上创建独立的输入/输出流往往无声。解决办法是先枚举设备再按索引指定：

python -m sounddevice

然后用环境变量分别指定输入、输出设备，例如"蓝牙耳机麦克风 + 内置扬声器"的组合：

REALTIME_INPUT_DEVICE=3 REALTIME_OUTPUT_DEVICE=2 python examples/realtime/local_mic.py

设备参数既支持数字索引也支持设备名字符串（见 local_mic.py 的_device函数：纯数字按索引解析，否则按名称匹配）。对应到LocalAudioTransport构造参数即input_device/output_device（默认None表示使用系统默认设备）。

3. 静默没关系：会话超时会自动重连

DashScope 在约三分钟没有响应时会关闭会话。示例会记录"会话已结束"并保持麦克风打开；你说出的下一句话会触发模型重连，且之前对话的转写会以追加到系统提示词的方式带入新会话，让模型保持对话上下文。

这个机制在源码里有两处体现：

• 上行侧：_on_audio捕获ModelDisconnectedError后把当前帧放入_backlog，下一帧到来时_try_connect()按退避策略重连（agent/_realtime/_agent.py，退避从 1 秒起、指数翻倍、封顶 30 秒），重连成功后把 backlog 里最多 10 秒（100 帧 × 100ms）的缓冲音频一次性补推给模型；
• 下行侧：_pump_downlink在会话结束后不退出，而是标记断开等待connect()再次被调用（agent/_realtime/_agent.py）；
• 上下文衔接：connect()会把state.context中的历史消息以## Conversation so far段落拼进 instructions 一起发送（agent/_realtime/_agent.py），这就是"转写跟着走"的实现。

4. 模型限制按轮次计，不按 token 计

qwen3-omni-flash-realtime只记得8 轮对话，qwen-audio-3.0-realtime-*系列记得50 轮；更旧的轮次由提供商静默丢弃。具体限制就写在 agentscope/realtime/_dashscope/ 下的各模型卡片里（max_audio_turns、max_audio_duration_s、max_session_duration_s等字段，见 RealtimeModelCard），需要精确记忆深度时先查卡片。

5. 文字输入（typed input）仅限特定模型

只有支持文本输入的模型才能中途注入文字轮次（qwen-audio-3.0-realtime-*系列可以，qwen3-omni-flash-realtime等 Omni 模型仅支持音频输入）。对应源码中的supports_text_input标记（realtime/_base.py），agent 的send()收到文本时会先检查该标记，不支持则抛NotImplementedError（agent/_realtime/_agent.py）。这也在模型卡input_types字段（audio/pcmvstext/plain）中得到印证。

八、从示例到自定义：改造方向

local_mic.py虽然只有 150 余行，却演示了几乎所有agentscope.realtime的扩展点：

• 换传输：实现 TransportBase 就能把音频来源换成浏览器、WebSocket 客户端或你自己的队列——RealtimeAgent对传输的差异完全无感；
• 换模型：只要模型卡片存在，REALTIME_MODEL一个环境变量即可切换；自定义模型卡片可在任意 YAML 中声明，经RealtimeModelBase.list_models(custom_yaml_dir=...)加载；
• 自定义 VAD：RealtimeAgent支持传入自实现的VADBase，此时会以turn_detection_disabled=True请求提供商关闭自带端点检测，由你的 VAD 独占轮次边界判定（agent/_realtime/_agent.py）；
• 控制帧协议：ControlFrame（TEXT/USER_CONFIRM/INTERRUPT）把离散输入统一成线上帧，浏览器端与 Python 端走同一条send()代码路径（realtime/_transport/_base.py）。

对应模块的单元测试可参考 tests/realtime_agent_test.py 与 tests/realtime_local_transport_test.py，它们覆盖了 barge-in、播放记账、传输生命周期等核心路径，是深入理解本模块行为的最佳读物。

【免费下载链接】agentscopeBuild and run agents you can see, understand and trust.项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope