Nova 2.5 Sonic 与 Strands Bidi Agents GA：实时语音 Agent 的异步工具、连接续期与打断处理

10 月 5 日，AWS 宣布 Amazon Nova 2.5 Sonic 正式可用（GA）。这是一个 speech-to-speech 模型，官方说法是推理、指令遵循和工具调用准确率都有提升，延迟也更低。同一天，开源 Agent 框架 Strands 宣布 Bidi Agents GA，Python SDK python/v1.58.0 于 UTC 10 月 5 日 23:59（北京时间 10 月 6 日 07:59）发布。

只看新闻，这像是“又一个语音模型”。但对做 Agent 的工程师来说，更有用的是这次 GA 一起定下来的工程约束：连接有时长上限、工具在后台异步跑、用户随时打断、状态要靠重放来续。这几条决定了语音 Agent 的架构不能照搬文本 Agent。

本文不讲模型排行，只回答一个问题：如果你要把文本 Agent 改造成可上线的实时语音 Agent，架构、代码和测试要怎么改。

一、先把事实和建议分开

类别	内容	依据
公开事实	Nova 2.5 Sonic 于 2026-10-05 GA，speech-to-speech，支持同一会话中混合语音与文本、异步工具调用、可控轮次切换（turn-taking），256K 上下文，7 种语言的表现力声音	AWS What's New
公开事实	发布时可用区域：US East (N. Virginia)、US West (Oregon)、Europe (Stockholm)、Asia Pacific (Tokyo)	AWS What's New
公开事实	Nova Sonic 单连接上限 8 分钟；Strands 文档列出 Gemini Live 约 10 分钟、OpenAI Realtime 60 分钟	Nova 用户指南、Strands 模型文档
公开事实	Strands `BidiAgent` 对 Nova Sonic 默认在连接建立 7 分钟后重启，并把文本历史重放进新连接；单条消息截断到 50 KiB，最多保留最近 200 KiB	Strands Nova Sonic 文档、v1.58.0 源码
公开事实	多个工具会作为一组并发执行，整组完成后结果才回给模型；工具可跨越 barge-in 继续运行；`BidiAgent` 暂不支持工具 interrupt（人审中断）	Strands Tools / Interrupts 文档
公开事实	v1.57.2 把 bidi API 从 experimental 毕业；v1.58.0 把 `auto_reconnect` 改名为 `auto_restart`，`strands.experimental.bidi` 成为弃用别名，默认模型从 `amazon.nova-2-sonic-v1:0` 变为 `amazon.nova-2-5-sonic`	GitHub Release、PyPI 包内容对比
作者建议	三层架构（实时层 / 动作层 / 后台层）、两段式提交、状态外置、故障注入测试矩阵、上线清单	本文
作者建议	文中所有代码、YAML、阈值、工具名都是示例，需结合自己的业务和 SDK 版本核对	本文

二、语音 Agent 和文本 Agent 差在哪

文本 Agent 的循环很清楚：用户发一条，模型想一会儿，调几个工具，回一条。语音 Agent 则是一条一直开着的双向流，至少有四处不同。

输入输出解耦。 BidiAgent 里 send() 发出输入后不等响应，receive() 持续产出事件。用户说话、模型说话、工具结果回来，全部挤在同一条事件流里。
工具异步，结果可能落在下一轮。 官方文档明确：工具组可能在请求它的那次响应结束之后才完成，结果会在之后的某次响应里被“讲出来”。按“相邻事件”配对会出错，必须按 toolUseId 关联。
连接一定会断。 8 分钟的上限意味着一通稍长的客服电话就要换好几次连接。框架会在后台换，但换的时候会发生什么，业务得知道。
打断是常态。 用户插话（barge-in）时，当前输出要立刻停，已排队的音频要清空；可后台工具不会因此停下。

图1：实时语音 Agent 三层架构与状态外置

三、架构：把“说话”和“办事”拆开

建议按三层来设计（作者建议）：

实时层：BidiAgent + 语音模型，只负责听、说、决定调哪个工具。系统提示词要短，规则要硬。
动作层：业务工具。所有有副作用的工具都过一道 hook 守卫，并且做成幂等的。
后台层：耗时的检索和推理交给另一个文本 Agent。Strands 官方示例用 create_harness() 把研究型 Agent 包成工具（as_tool），语音模型和研究模型可以分别选型。

另外有一条贯穿三层：业务状态外置。重放历史有 200 KiB 上限，而且只重放文本，音频不进历史。订单号、已确认的事项、当前流程节点，这些不要指望模型“记得”，应该放进 agent.state，用 SnapshotSessionManager 持久化（v1.58.0 起支持 BidiAgent）。

一个最小骨架如下（示例代码，API 以 Strands 官方文档为准，参数值为示意）：

# 示例：Strands Bidi Agents + Nova 2.5 Sonic 最小骨架（参数为示意值）
import asyncio
import os

from strands.bidi.agent import BidiAgent
from strands.bidi.io import AudioIO
from strands.bidi.models import BedrockNovaSonicModel
from strands.session import SnapshotSessionManager
from strands.storage import LocalFileStorage
from strands.telemetry import StrandsTelemetry

StrandsTelemetry().setup_otlp_exporter()          # 在 agent 启动前配置

model = BedrockNovaSonicModel(
    # 显式钉住模型 ID，不依赖 SDK 默认值；官方文档示例为 "amazon.nova-2-5-sonic"
    model_id=os.environ["NOVA_SONIC_MODEL_ID"],
    region="us-east-1",
    params={"turnDetectionConfiguration": {"endpointingSensitivity": "LOW"}},
    connection={"restart_after_s": 360},          # 示例：比默认 420 秒更早，给等待轮次边界留余量
)

agent = BidiAgent(
    model=model,
    agent_id="returns-voice-agent",               # 示例名
    system_prompt="你是售后语音助手。回答简短；涉及退货等操作必须先复述并取得用户口头确认。",
    tools=[],                                      # 见下一节
    hooks=[],                                      # 见下一节
    session_manager=SnapshotSessionManager(
        session_id="call-0001",                    # 示例：通话 ID
        storage=LocalFileStorage("./sessions/"),   # 生产可换 S3 等存储
    ),
)

async def main() -> None:
    audio = AudioIO(audio_processor=True)          # 开启 WebRTC 回声消除 / 降噪 / 自动增益
    await agent.run(inputs=[audio.input()], outputs=[audio.output()])

asyncio.run(main())

这里有两个细节值得单独说：

钉住 model_id。 对比 PyPI 上 1.57.1 和 1.58.0 两个包的内容可以看到，不传模型时的默认值从 amazon.nova-2-sonic-v1:0 变成了 amazon.nova-2-5-sonic。依赖默认值的话，升一次 SDK 底层模型就悄悄换了。
restart_after_s 留余量。 官方文档提醒：计划内重启会先短暂等待轮次边界，必须给“等边界 + 建新连接”留出时间。

四、动作层：异步工具的三条规矩

规矩 1：有副作用的动作做成两段式。 BidiAgent 暂不支持工具 interrupt，没法像文本 Agent 那样“暂停等人审批”。语音场景下的替代做法是拆成 prepare_*（只生成摘要和确认码，不执行）和 commit_*（凭确认码执行）。模型必须先把摘要念给用户，听到明确确认后再调用 commit。

规矩 2：守卫写在 hook 里，不写在提示词里。 Strands 的 BeforeToolCallEvent 在 BidiAgent 上同样可用，并支持 tool interception。设置 event.cancel_tool 就能拦下这次调用，并把原因回给模型。

# 示例：两段式提交 + BeforeToolCallEvent 守卫（工具名、字段、TTL 均为示例）
import secrets
import time
from typing import Any

from strands import tool
from strands.hooks import BeforeToolCallEvent, HookRegistry

PENDING: dict[str, dict] = {}   # 示例用内存；生产应放带 TTL 的共享存储

@tool
async def prepare_return(order_id: str, reason: str) -> dict:
    """准备退货申请但不执行。返回需要向用户复述的摘要和确认码。"""
    token = secrets.token_hex(4)
    PENDING[token] = {"order_id": order_id, "reason": reason, "exp": time.time() + 120}
    return {"say_to_user": f"将为订单 {order_id} 发起退货，原因：{reason}。确认吗？",
            "confirm_token": token}

@tool
async def commit_return(confirm_token: str) -> dict:
    """仅在用户明确口头确认后调用，提交退货。"""
    req = PENDING.pop(confirm_token)
    # 以 confirm_token 作为幂等键调用下游，重复提交不会产生第二笔
    return {"status": "submitted", "order_id": req["order_id"]}

class SideEffectGuard:
    COMMIT_TOOLS = {"commit_return"}

    def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
        registry.add_callback(BeforeToolCallEvent, self.check)

    def check(self, event: BeforeToolCallEvent) -> None:
        if event.tool_use["name"] not in self.COMMIT_TOOLS:
            return
        token = event.tool_use["input"].get("confirm_token", "")
        req = PENDING.get(token)
        if req is None or req["exp"] < time.time():
            event.cancel_tool = "确认码无效或已过期：请重新调用 prepare_return 并取得用户口头确认。"

规矩 3：长任务改成“提交 + 查询”。 官方文档说明，同一组工具要全部完成，结果才会一起回给模型；Agent 的 cancel() 也只在工具组结束后生效。一个 40 秒的同步查询会把整组结果都拖住。更稳的做法是工具立即返回 job_id，再提供一个 get_job_status 工具，并且一律用 async def 和异步 I/O，避免阻塞会话。

还有一个容易漏掉的细节：在 v1.58.0 的 Nova 适配器里，如果工具结果在连接重启之后才回来，因为新连接里没有对应的 tool use，结果会被改写成一段用户文本补发给模型，其中非文本内容会被省略。所以工具返回值应该是紧凑、自描述的文本或 JSON，不要依赖二进制或富格式。

五、连接续期：把“换线”当成一等事件

图2：8 分钟连接上限下的续期时间线（以 Nova Sonic 默认配置为例）

框架的续期流程（公开事实）：计划内重启前发 BidiConnectionWarningEvent（带 time_left_s）；重启时发 BidiConnectionRestartEvent，其中 reason 为 scheduled 或 timeout；新连接就绪后发新的 BidiConnectionStartEvent。换线期间 send() 会等待，后台工具照常运行。

最关键的是 turn_interrupted 字段。文档原意是：如果重启打断了进行中的回答或尚未回答的一轮，重放的历史只提供上下文，不会自动要求模型回答那一轮。业务侧要自己补一句。

# 示例：自定义事件消费——处理 barge-in 与换线（播放器、指标对象为示例）
from strands.bidi.types import (
    BidiBargeInEvent,
    BidiConnectionRestartEvent,
    BidiConnectionStartEvent,
)

async def consume(agent, player, metrics) -> None:
    need_reprompt = False
    async for event in agent.receive():
        if isinstance(event, BidiBargeInEvent):
            player.clear()                      # 自定义输出流必须自己清空播放队列
            metrics.inc("barge_in")
        elif isinstance(event, BidiConnectionRestartEvent):
            metrics.inc("restart", reason=event.reason)
            need_reprompt = need_reprompt or event.turn_interrupted
        elif isinstance(event, BidiConnectionStartEvent) and need_reprompt:
            need_reprompt = False
            await agent.send("连接刚刚切换，请用一句话接着回答用户上一个问题。")
        else:
            player.handle(event)                # 音频 / 转写 / 文本按 content_id 归并

配套还有三条建议（作者建议）：

流式内容按 content_id 归并，工具按 toolUseId 关联，不要按事件相邻关系推断；
连接中断后可能出现没有 stop 事件的半截内容（历史里标为 incomplete，未完成的转写显示为 [Transcript unavailable.]），清理状态不能依赖收到 stop；
关键业务事实写进 agent.state，并在系统提示词或工具结果里显式回读，不依赖 200 KiB 的重放窗口。

六、Barge-in：停的是“嘴”，不是“手”

官方行为：barge-in 只停当前输出，会话和后台工具继续。内置 AudioIO.output() 会自动清空缓冲音频；自定义输出流则要自己清。

这对业务的含义是：用户说“算了，不退了”，正在跑的 prepare_return 仍会返回结果。好在两段式设计下 prepare 没有副作用，确认码 2 分钟后自动失效；真正要防的是 commit。所以撤销语义要落在动作层：commit 前查一次最新状态，或者提供一个 cancel_pending 工具清掉确认码，而不是指望打断能“取消”什么。

外放场景还要注意回声。Strands 博客提到，开启 audio_processor=True 后用 WebRTC 回声消除，可以减少扬声器声音被麦克风收进去、导致模型“打断自己”的情况。浏览器或手机端接入时，采集和回声处理在客户端，这一点需要单独验证。

七、可观测与测试：把故障注入做成回归

Strands Bidi 会为会话、模型响应、工具调用、建连和换线产出 OpenTelemetry span。几个值得盯的点（公开事实）：

响应 span 上的 gen_ai.server.time_to_first_audio：从响应开始到第一个音频块的毫秒数，这是用户能直接感受到的延迟；
bidi_connect span：把握手耗时和模型响应耗时分开看；
barge-in 记录为会话 span 上的 bidi_barge_in 事件；
系统提示词会原样进入会话 span。生产环境可以通过 OTEL_SEMCONV_STABILITY_OPT_IN="gen_ai_unredacted_attributes=" 开启脱敏。

图3：语音 Agent 故障注入测试矩阵与观测指标（作者示意）

测试建议（作者建议）：语音 Agent 不容易做成单元测试，但完全可以做成录音回放 + 故障注入的回归集。下面的矩阵是示例：

# 示例：语音 Agent 回归矩阵（场景、阈值均为示例值）
fixtures: ./audio-fixtures/          # 预录 16kHz 单声道 PCM 用户语音
cases:
  - id: restart-mid-answer
    inject: { connection.restart_after_s: 45 }   # 测试环境把重启间隔调短
    expect: [ "turn_interrupted 时补发一次提示", "不重复调用工具" ]
  - id: tool-result-after-restart
    inject: { tool_latency_s: 60, connection.restart_after_s: 30 }
    expect: [ "结果以文本形式送达", "模型不再次调用同一工具" ]
  - id: barge-in-during-prepare
    audio: interrupt_cancel.pcm
    expect: [ "播放队列 < 200ms 内清空", "commit 未被调用" ]
  - id: commit-without-token
    expect: [ "cancel_tool 生效", "下游接口零调用" ]
  - id: long-call-30min
    expect: [ "state 中订单号保持一致", "time_to_first_audio P95 不随时长恶化" ]

八、上线前检查清单

model_id 显式钉住，SDK 升级前后各跑一次回归集；
从 strands.experimental.bidi 迁到 strands.bidi，auto_reconnect 改为 auto_restart；
restart_after_s 小于提供商上限，并为等待轮次边界留出余量；
处理 turn_interrupted，换线后能接上话；
所有副作用工具做成两段式，并有幂等键和 BeforeToolCallEvent 守卫；
长任务改为“提交 + 查询”，工具全部用 async def；
工具返回紧凑的文本或 JSON，经得起换线后被改写成文本；
业务状态放进 agent.state + SnapshotSessionManager，不依赖重放窗口；
自定义输出流在 barge-in 时清空播放队列；外放场景开启回声处理；
开 OTel，盯 time_to_first_audio 和 bidi_connect，生产环境对系统提示词脱敏；
确认部署区域在 Nova 2.5 Sonic 的可用区域内，数据合规按自身要求评估。

结语

Nova 2.5 Sonic 解决的是“模型听得懂、会调工具、反应快”，Strands Bidi Agents GA 解决的是“连接能续、事件能看、换模型不用重写”。剩下的部分，比如副作用怎么确认、状态放哪、断线后怎么接话、打断后怎么撤销，仍然是业务工程的责任。把这几条做成架构约束和回归测试，语音 Agent 才算真正可以上线。

参考资料

AWS What's New：Announcing Amazon Nova 2.5 Sonic with improved reasoning for voice agents（2026-10-05）
Strands Agents Blog：Bidi Agents now GA（2026-10-05）
Strands Agents 文档：Bidirectional Streaming（Agent / Tools / Stream Events / Hooks / Session Management / Models / Observability / Bedrock Nova Sonic）
GitHub strands-agents：python/v1.58.0、python/v1.57.2 Release Notes
Amazon Nova 用户指南：Using conversational speech

本文同步发布于 CSDN：https://blog.csdn.net/qq_29737119/article/details/167258764