AWS · ML 博客

Stream Vision Agents 与 Amazon Nova 2 Sonic 实时语音代理

Real-time voice agents with Stream Vision Agents and Amazon Nova 2 Sonic

二〇二六年五月十四日 · 英文原文

摘要

本文介绍如何使用 Stream 的 Vision Agents 开源框架与 Amazon Bedrock 及 Amazon Nova 2 Sonic 构建实时语音代理。架构中 Stream 边缘网络处理媒体传输（加入时间低于 500 毫秒、音频延迟低于 30 毫秒），Amazon Nova 2 Sonic 提供语音到语音模型、原生话轮检测与函数调用。Vision Agents 通过基于装饰器的接口抽象连接管理、自动重连、VAD 感知及工具执行。代码示例展示在 30 行内实现可投入生产的语音代理，支持多语言、函数调用及生产部署。

本文与 Stream 技术营销负责人 Neevash Ramdial 共同撰写。构建感觉自然且响应迅速的生产级语音代理是一项复杂的工程挑战。你必须编排语音到语音模型、管理低延迟音频流，并处理连接生命周期。你还需要在 Web、移动和桌面应用程序中提供一致的体验。在本文中，你将学习如何将 Stream 的 Vision Agents 开源框架与 Amazon Bedrock 和 Amazon Nova 2 Sonic 结合，构建可在几分钟内投入生产的实时语音代理。你将了解集成在底层的工作原理，浏览代码示例，并探索函数调用、自动重连和多语言语音支持等高级功能。

挑战

构建支持语音的 AI 应用程序需要编排多个必须可靠协同工作的复杂系统。你面临着管理实时音频流基础设施的挑战，同时还要集成语音识别、语言模型和文本转语音服务。每个服务都有其自身的延迟特性和故障模式。一个典型的语音交互包括：从用户麦克风捕获音频，将其流式传输到语音转文本服务，通过语言模型处理转录文本，生成响应，将该响应转换回语音，并传递给用户。所有这些都必须在几百毫秒内完成，才能感觉自然。此流程中的任何延迟都可能打断对话的流畅性，让用户感到沮丧。

除了核心 AI 流程之外，生产级语音应用还必须处理现实部署中的各种棘手问题：不可靠的网络连接、浏览器兼容性问题、会话超时，以及服务不可用时的优雅降级。你花在构建重连逻辑、管理 WebRTC 连接和处理边缘情况上的时间，往往比花在实际 AI 能力上的时间还要多。这种基础设施负担意味着团队要么花费数月时间构建定制解决方案，要么只能选择无法满足其特定需求的有限现成产品。Vision Agents 抽象了基础设施的复杂性，同时提供了定制 AI 体验的灵活性。

解决方案概述

该解决方案整合了三个关键组件：

Amazon Nova 2 Sonic：一个通过 Amazon Bedrock 提供的语音到语音基础模型，提供实时双向音频流、原生话轮检测和函数调用能力。Nova 2 Sonic 处理完整的语音到语音流程，接收音频输入并产生音频输出。这避免了需要单独的 STT 和 TTS 服务。
Stream 的 Vision Agents：一个用于构建实时语音和视频 AI 代理的开源 Python 框架。它提供基于插件的架构，拥有 25 个以上的集成、生产部署工具以及用于 React、iOS、Android、Flutter 和 React Native 的客户端 SDK。该系统以灵活性为核心设计。你可以使用 Stream 的全球边缘网络实现高效性能，或集成你偏好的实时通信（RTC）提供商。Vision Agents 通过一个简洁的基于装饰器的接口处理特定于提供商的规范，支持客户支持代理、工作流自动化和 API 驱动的操作等用例，且样板代码极少。使用 Vision Agents，你可以使用开源框架、第三方模型提供商和电话服务构建 AI 应用程序。
Stream 的边缘网络：一个全球分布式的边缘网络，通常可实现低于 500 毫秒的加入时间和低于 30 毫秒的音频延迟，在客户端和你的代理后端之间提供实时传输层。

这些组件共同构成了一个完整的堆栈：Stream 处理实时媒体传输和客户端体验，Amazon Nova 2 Sonic 提供 AI 智能，而 Vision Agents 则提供将它们连接在一起的粘合代码。

架构概述

该系统围绕清晰的关注点分离进行设计：Stream 的基础设施处理实时媒体传输和客户端连接，而 Amazon Nova Sonic 在客户自己的 AWS 账户中运行并提供 AI 智能。这种分离有助于确保敏感数据和业务逻辑保持在客户的控制范围内，同时 Stream 的全球分布式边缘网络提供用户期望的低延迟媒体体验。

Stream 的边缘网络充当最终用户设备和 Vision Agent 工作进程之间的媒体代理。当用户说话时，音频被捕获、加密，并通过 RTP over UDP 传输到最近的 Stream SFU（选择性转发单元）。SFU 终止 WebRTC 连接，处理 NAT 穿越和带宽估计，并将音频轨道转发给 Vision Agent 工作进程，就像它是另一个通话参与者一样。这意味着代理自然地集成到通话模型中。代理是另一个对等方，通过人类参与者使用的相同基础设施接收和发送音频。

音频数据在系统中双向流动：来自用户的传入语音由 Vision Agent 工作进程解码为原始 PCM，通过 Bedrock 实时 API 流式传输到 Amazon Nova Sonic，来自 Nova Sonic 的响应音频帧被重新编码、打包为 RTP，并通过 SFU 传回客户端设备。端到端延迟通常低于 500 毫秒。语音活动检测（VAD）在工作进程中运行，以检测语音边界和插入事件，而浏览器中的回声消除有助于防止代理自身的输出重新触发 VAD 循环。

账户边界

客户 AWS 账户：业务逻辑和编排（代理策略、工具、数据访问）。通过 Amazon Bedrock 集成访问 Amazon Nova 模型。
Stream AWS 账户：全局 WebRTC/SFU 媒体平面、TURN/STUN 和信令。Vision Agent 运行时（工作进程）将 WebRTC 作为机器人对等方终止，并桥接客户的 Amazon Bedrock 集成。

端到端媒体流

用户加入：用户从 Web 或移动端加入。应用程序嵌入 Stream 的音频客户端 SDK，请求麦克风（以及可选的摄像头），并加入一个配置为允许 AI 参与的通话类型。媒体通过 RTP over UDP 发送，以实现可预测的低延迟和无队头阻塞传输。
区域 SFU 终止：最近区域的 Stream SFU 节点终止用户的 WebRTC 连接，处理带宽估计、Simulcast 和 NAT 穿越。SFU 将相关的音频轨道转发给 Vision Agent 工作进程，就像它是另一个参与者一样。
Vision Agent 工作进程：一个专用的 Vision Agent 工作进程持有该会话的 PeerConnection 状态。它将音频解码为原始 PCM，然后工作进程将 PCM 帧流式传输到 Amazon Bedrock 服务，再转发到客户 AWS 账户中的 Amazon Nova 2 Sonic 实时会话。
通过 Amazon Bedrock 与 Vision Agents 集成的 Amazon Nova 2 Sonic：Amazon Nova 2 Sonic 检测语音边界并执行语音到语音建模（理解、推理和 TTS），并可选择调用客户系统（RDS、API、知识库）中的工具。它优雅地处理插入，并维护完整的对话上下文，以确保对话自然且连贯。
将响应流式传输回用户：当 Amazon Nova Sonic 产生响应音频帧时，Vision Agent 工作进程会：将它们切片并包装到 RTP 中，使用单调递增的时间戳以避免间隙/漂移；通过相同的 WebRTC 会话，经由 SFU 发送 RTP 数据包。浏览器的 WebRTC 堆栈以低于 500 毫秒的延迟解码并播放音频。
插入、转录和辅助数据：浏览器中的回声消除有助于防止代理自身的输出重新触发 VAD。当用户打断时，新的语音会通过 RTCDataChannel 触发一个中断信号，导致工作进程停止转发 Amazon Nova Sonic 的输出并重置其本地缓冲区。

这种架构可能看起来很复杂，但 Vision Agents 抽象了其中的大部分复杂性。让我们看看实际的代码是什么样的：

前提条件

在开始之前，请确保你具备以下条件：

通过环境变量、IAM 角色或 AWS 命令行界面（AWS CLI）配置文件配置的 AWS 凭证。对于生产环境，请使用附加到计算资源的 IAM 角色，而不是长期凭证。对于本地开发，请使用 AWS CLI 配置文件（aws configure）或 AWS SSO。不要将包含凭证的 .env 文件提交到版本控制。
具有音频 API 密钥和密钥的 Stream 账户（你预计每月可免费获得 333,000 分钟的参与者时长）。
已安装 Python 3.12 或更高版本。
已安装 uv 包管理器（pip install uv）。
已安装 Vision Agents（uv add vision-agents）。

入门指南

步骤 1：创建新的项目目录并使用 AWS 插件安装 Vision Agents

mkdir voice-agent
cd voice-agent
uv init
uv add "vision-agents[getstream,aws]" python-dotenv

vision-agents[aws] 额外安装项会安装 Amazon Bedrock 插件及其依赖项，包括 boto3、aws-sdk-bedrock-runtime 和用于语音活动检测的 Silero VAD。

步骤 2：配置环境变量

在项目根目录创建一个 .env 文件来管理你的配置。对于 AWS 凭证，我们建议在此文件中指向你的 AWS_PROFILE，以便应用程序在与 AWS 资源交互时可以访问你的凭证。我们不建议直接在此文件中存储你的 AWS 访问密钥。对于 Stream API 凭证，你可以使用第三方库，如 HashiCorp Vault 或 AWS Secrets Manager，但安全考虑不在本文讨论范围内。

# Stream API credentials
STREAM_API_KEY=test/geststream/api_key
STREAM_API_SECRET=test/getstream/api_secret

# AWS credentials
AWS_PROFILE=your_aws_profile_name
AWS_REGION=us-east-1

Vision Agents 会在启动时自动发现这些环境变量，因此你无需将它们显式传递给每个客户端。

步骤 3：构建你的第一个语音代理

创建一个包含以下代码的 main.py 文件：

import asyncio
from dotenv import load_dotenv
from vision_agents.core import Agent, User, Runner
from vision_agents.core.agents import AgentLauncher
from vision_agents.plugins import aws, getstream

load_dotenv()

async def create_agent(**kwargs) -> Agent:
    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(name="Helpful Assistant", id="agent"),
        instructions="You are a helpful voice assistant. Be concise and friendly.",
        llm=aws.Realtime(
            model="amazon.nova-2-sonic-v1:0",
            region_name="us-east-1",
            voice_id="matthew",
        ),
    )
    return agent

async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
    call = await agent.create_call(call_type, call_id)
    async with agent.join(call):
        await asyncio.sleep(2)
        await agent.llm.simple_response(
            text="Greet the user warmly and ask how you can help."
        )
        await agent.finish()

# Run until the call ends
if __name__ == "__main__":
    Runner(AgentLauncher(create_agent=create_agent, join_call=join_call)).cli()

步骤 4：运行语音代理

运行代理：

uv run main.py run

在不到 30 行代码中，你就拥有了一个由 Amazon Nova Sonic 驱动的、功能齐全的实时语音代理，可通过 Stream 客户端 SDK 访问。

理解 Amazon Bedrock 集成

让我们更仔细地看看 aws.Realtime 插件在底层是如何工作的。

与 Amazon Nova 2 Sonic 的双向流

Amazon Nova 2 Sonic 使用事件驱动的双向流 API。这种方法不是使用请求-响应模式，而是允许近乎连续的音频同时在两个方向上流动。Vision Agents AWS 插件通过结构化的事件序列来管理这种复杂性：

会话初始化 – 发送一个 sessionStart 事件，包含推理配置（temperature、maxTokens、topP）。
提示设置 – 一个 promptStart 事件配置音频输出格式（24kHz PCM）、语音选择和工具定义。
系统指令 – 系统指令作为具有 SYSTEM 角色的文本内容块发送。
音频流 – 麦克风音频帧（约 32ms 每帧）作为 audioInput 事件流式传输。
响应流 – Nova Sonic 流式传输回包含生成语音的 audioOutput 事件。
会话拆除 – promptEnd 和 sessionEnd 事件干净地关闭连接。

每个内容块遵循三部分模式：contentStart → 内容负载 → contentEnd。这种分层结构允许模型在整个交互过程中维护正确的上下文。

以下是插件中会话启动事件的样子：

def _create_session_start_event(self) -> Dict[str, Any]:
    return {
        "event": {
            "sessionStart": {
                "inferenceConfiguration": {
                    "maxTokens": 1024,
                    "topP": 0.9,
                    "temperature": 0.7,
                },
                "turnDetectionConfiguration": {
                    "endpointingSensitivity": "MEDIUM"
                },
            }
        }
    }

添加函数调用

Amazon Nova 2 Sonic 的关键能力之一是在实时对话期间进行原生函数调用。这允许你的语音代理执行诸如查询数据库、调用 API 和触发工作流等操作，同时保持自然的语音对话。

使用 @llm.register_function 装饰器来定义模型可以调用的函数：

import asyncio
from dotenv import load_dotenv
from typing import Dict, Any
from vision_agents.core import Agent, User, Runner
from vision_agents.core.agents import AgentLauncher
from vision_agents.plugins import aws, getstream

load_dotenv()

async def create_agent(**kwargs) -> Agent:
    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(name="Weather Assistant", id="agent"),
        instructions="""You are a helpful weather assistant. When users ask about weather, use the get_weather function to fetch current conditions. You can also help with simple calculations.""",
        llm=aws.Realtime(
            model="amazon.nova-2-sonic-v1:0",
            region_name="us-east-1",
        ),
    )

    @agent.llm.register_function(
        name="get_weather",
        description="Get the current weather for a given city"
    )
    async def get_weather(location: str) -> Dict[str, Any]:
        # In production, call a real weather API
        return {
            "city": location,
            "temperature": 72,
            "condition": "Sunny",
            "humidity": "45%"
        }

    @agent.llm.register_function(
        name="calculate",
        description="Perform a mathematical calculation"
    )
    def calculate(operation: str, a: float, b: float) -> dict:
        operations = {
            "add": lambda x, y: x + y,
            "subtract": lambda x, y: x - y,
            "multiply": lambda x, y: x * y,
            "divide": lambda x, y: x / y if y != 0 else None,
        }
        result = operations.get(operation, lambda x, y: None)(a, b)
        return {"operation": operation, "a": a, "b": b, "result": result}

    return agent

async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
    await agent.create_user()
    call = await agent.create_call(call_type, call_id)
    async with agent.join(call):
        await asyncio.sleep(2)
        await agent.llm.simple_response(
            text="Greet the user and let them know you can check the weather."
        )
        await agent.finish()

if __name__ == "__main__":
    Runner(AgentLauncher(create_agent=create_agent, join_call=join_call)).cli()

函数调用如何与 Amazon Nova 2 Sonic 协同工作

当模型决定调用一个函数时，会发生以下序列：

Nova 2 Sonic 发出一个包含函数名称和参数的 toolUse 事件。
Vision Agents 插件拦截此事件，反序列化参数，并运行注册的 Python 函数。
结果通过 toolResult 事件发送回 Nova，包装在标准的 contentStart → toolResult → contentEnd 模式中。
Nova 2 Sonic 将函数结果整合到其响应中，并自然地继续语音对话。

你可以使用这种方法构建复杂的多步骤工作流。例如，一个语音代理可以在一次自然对话中查找客户记录、检查库存并下订单。

使用带有 Amazon Bedrock 的标准 LLM

除了实时语音到语音之外，AWS 插件还通过 aws.LLM 提供标准的 LLM 集成。这对于自定义管道架构非常有用，你希望将 Amazon Bedrock 模型与单独的 STT 和 TTS 提供商配对：

from vision_agents.core import Agent, User
from vision_agents.plugins import aws, getstream, cartesia, deepgram, smart_turn

agent = Agent(
    edge=getstream.Edge(),
    agent_user=User(name="Custom Pipeline Agent"),
    instructions="Be helpful and concise.",
    llm=aws.LLM(
        model="anthropic.claude-3-haiku-20240307-v1:0",
        region_name="us-east-1"
    ),
    tts=cartesia.TTS(),
    stt=deepgram.STT(),
    turn_detection=smart_turn.TurnDetection(
        buffer_duration=2.0,
        confidence_threshold=0.5
    ),
)

标准 LLM 支持通过 converse_stream() 进行流式响应、完整的对话历史管理、针对 Claude 等模型的视觉输入，以及每次请求最多 3 轮函数执行的多轮工具调用。

使用 Amazon Polly 进行文本转语音

对于自定义管道架构，AWS 插件还包含一个 Amazon Polly TTS 集成。当你使用非实时 LLM（如 Amazon Bedrock 上的 Claude 或其他提供商）并需要高质量语音合成时，这非常有用：

from vision_agents.plugins import aws

tts = aws.TTS(
    region_name="us-east-1",
    voice_id="Joanna",
    engine="neural",  # 'standard' or 'neural'
    language_code="en-US"
)

Amazon Polly TTS 支持标准和神经引擎、用于精细语音控制的 SSML 输入，以及多种语言和语音。神经引擎产生更自然的语音，使其成为你在 AWS 基础设施上构建自定义 STT → LLM → TTS 管道时的有力选择。

清理资源

要删除 Stream 通话并终止正在运行的 Vision Agent 进程：

uv run main.py stop

重要提示：所有对 Amazon Nova 2 Sonic 的 API 调用都会产生 Amazon Bedrock 费用。你可以运行清理命令来终止会话并避免持续产生费用。活跃会话可能会持续产生费用，直到被显式终止。

用例

有了技术基础，值得探讨这些能力在哪些方面可以转化为有意义的实际影响。低延迟语音、对话管理和工具集成的结合，为各行各业开辟了广泛的应用场景，在这些场景中，自然的语音交互可以取代或增强传统界面。

用例 1：无屏幕和低注意力环境的语音界面

Vision Agents 与 Amazon Nova 2 Sonic 结合，非常适合用户无法可靠地与屏幕交互的环境，例如驾驶、现场服务、物流、医疗保健或现场操作。

在这些环境中，语音成为主要界面，而不是便利功能。借助 Amazon Nova 2 Sonic，你可以获得低延迟、自然的语音到语音交互，允许用户自由说话、打断响应和自我纠正，而不会打断流程。Vision Agents 管理跨话轮的对话状态和任务逻辑，将语音输入转换为结构化操作，例如检索下一个工作任务、更新任务状态、记录笔记或请求人工协助。由于代理在整个交互过程中维护上下文，用户可以发出后续命令或澄清，而无需重复信息。

例如，送货司机可以问“我的下一站是哪里？”，听到语音指示，说“将上次送货标记为完成”，然后接着说“呼叫调度中心”，所有这些都无需触摸屏幕，同时代理实时更新后端系统。

用例 2：大规模高容量呼入电话支持

Vision Agents 与 Amazon Nova 2 Sonic 结合，专为处理大量呼入支持电话而设计，在这些场景中，人工座席成为瓶颈。此用例从根本上关乎规模：减少排队时间、分流重复请求，并将人工座席保留给需要他们参与的情况。

借助 Amazon Nova 2 Sonic，呼叫者可以进行低延迟、实时的语音到语音对话，自然地解释问题，而无需导航脚本化的 IVR 树。Vision Agents 编排意图检测、对话状态和后端集成（如订单系统、账户记录或工单服务），因此常见请求可以在通话中自动解决。当问题超出预定义的置信度阈值或需要人工干预时，代理会附上结构化上下文升级给人工座席，从而减轻呼叫者重复自己的负担。

在高峰时段，可能有数百名客户打电话询问配送延迟。呼叫者无需排队等候，而是立即由语音代理接听，该代理检查订单状态、解释延迟、提供后续步骤，并且仅在检测到异常时才转接给人工座席。

这将电话系统从一个基于队列的成本中心转变为一个持续的、一线的问题解决层。

结论

本文介绍了如何使用 Stream 的 Vision Agents 框架和 Amazon Bedrock（与 Amazon Nova 2 Sonic 配合）构建实时语音代理。我们涵盖了架构、双向流协议、自动重连处理、函数调用、多语言支持和生产部署。

Stream 的低延迟边缘网络与 Amazon Nova Sonic 的原生语音到语音能力的结合，为构建语音 AI 应用程序提供了坚实的基础。Vision Agents 框架抽象了连接生命周期管理、音频编码、VAD 感知重连和工具执行等复杂的编排工作，因此你可以专注于代理的逻辑和用户体验。

如果你准备进一步探索，我们鼓励你尝试为你的特定用例扩展代理的自定义函数，或探索用于全球应用的多语言能力。Vision Agents 仓库（https://github.com/GetStream/Vision-Agents）是一个很好的起点。你可以在那里找到更多示例、插件文档和社区讨论。有关更深入的集成细节，AWS 插件文档可在 https://visionagents.ai/integrations/aws-bedrock 获取，而 AWS Nova 用户指南中的 Amazon Nova 2 Sonic 文档为双向流 API 提供了全面的参考。你可以在 https://getstream.io/ 注册一个 Stream 开发者账户，并立即开始免费构建。

关于作者

Manasi Bhutada 是 AWS 驻荷兰的 ISV 解决方案架构师，她与客户合作设计和实施架构良好的解决方案，以应对其独特的业务挑战。她热衷于帮助公司转型为成功的 SaaS 提供商，在指导组织完成云之旅以及在 AWS 上构建可扩展、有弹性的架构方面拥有深厚的专业知识。她的专长在于数据分析。

Jagdeep Singh Soni 是 AWS 驻荷兰的高级 AI/ML 解决方案架构师，专注于生成式 AI 和 Amazon Bedrock。他帮助客户和合作伙伴使用 Amazon Bedrock 和其他 AWS AI/ML 服务架构和实施智能代理解决方案。凭借 17 年的创新和云架构经验，Jagdeep 专注于帮助组织构建可用于生产的生成式 AI 应用程序，利用基础模型和代理框架实现实际的业务成果。

Neevash Ramdial 是一名工程师、Google 开发者专家（GDE），也是 Stream 驻科罗拉多州博尔德的工程和技术营销负责人，专攻实时视频、音频和 AI。他领导 Stream 的 Vision Agents 项目，帮助开发者和公司大规模构建快速、低延迟的视觉 AI 代理。凭借在 Stream 超过五年的经验以及涵盖开源项目和社区倡议的背景，Neevash 专注于帮助组织利用代理 AI 构建可用于生产的应用程序。

译自 AWS · ML 博客 · 录于二〇二六年五月十四日