用 Stream Vision Agents 和 Amazon Nova 2 Sonic 搭建实时语音 Agent

pip install stream-vision-agents boto3

import asyncio
from vision_agents import VoiceAgent, AudioStreamConfig
from vision_agents.models import BedrockNova2Sonic

# 1. 配置 Nova 2 Sonic 模型接入
model = BedrockNova2Sonic(
    model_id="amazon.nova-2-sonic",   # Bedrock 上的模型标识
    region="us-east-1",               # Bedrock 可用区域
    voice="matthew",                  # 合成语音的角色名
    language="en",                    # 默认语言
)

# 2. 配置音频流参数
audio_config = AudioStreamConfig(
    sample_rate=16000,      # 输入音频采样率
    channels=1,             # 单声道
    format="pcm_s16le",    # 16-bit little-endian PCM
    chunk_duration_ms=100,  # 每 100ms 发一个音频帧
)

# 3. 定义 Agent 行为（system prompt）
system_prompt = """
You are a helpful voice assistant for a coffee shop.
Answer questions about menu items, prices, and hours.
Keep responses short — under 2 sentences — since this is a voice conversation.
"""

# 4. 组装并启动 Agent
agent = VoiceAgent(
    model=model,
    audio_config=audio_config,
    system_prompt=system_prompt,
)

async def main():
    # 启动 Agent 服务，监听 WebSocket 连接
    await agent.serve(host="0.0.0.0", port=8765)

asyncio.run(main())

vision-agents chat \
  --ws-url ws://localhost:8765 \
  --mic-device 0 \
  --speaker-device 0

from vision_agents import VoiceAgent, tool

# 定义函数工具
@tool(
    name="get_menu_item_price",
    description="Get the price of a menu item by name",
    parameters={
        "item_name": {
            "type": "string",
            "description": "The name of the menu item, e.g. 'Americano'",
        }
    },
)
def get_menu_item_price(item_name: str) -> str:
    # 实际项目中这里查数据库或调内部 API
    menu = {
        "Americano": "$3.50",
        "Latte": "$4.75",
        "Cappuccino": "$4.50",
        "Mocha": "$5.25",
    }
    price = menu.get(item_name, "not found")
    return f"{item_name} is {price}"

# 把工具注册到 Agent
agent = VoiceAgent(
    model=model,
    audio_config=audio_config,
    system_prompt=system_prompt,
    tools=[get_menu_item_price],  # 注册函数
)

from vision_agents import VoiceAgent, ReconnectConfig

reconnect_config = ReconnectConfig(
    max_retries=5,             # 最大重连次数
    retry_interval_ms=2000,    # 每次重连间隔
    backoff_multiplier=1.5,    # 间隔指数退避系数
    replay_buffer_ms=3000,     # 重连后回放最近3秒音频，避免上下文丢失
)

agent = VoiceAgent(
    model=model,
    audio_config=audio_config,
    system_prompt=system_prompt,
    reconnect_config=reconnect_config,
)

model = BedrockNova2Sonic(
    model_id="amazon.nova-2-sonic",
    region="us-east-1",
    voice="zhen",             # 中文语音角色
    language="zh",            # 切换为中文
)

system_prompt = """
你是咖啡店的语音助手。用中文回答关于菜单、价格和营业时间的问题。
语音对话请保持简短，每次回答不超过两句话。
"""

@tool(
    name="get_menu_item_price",
    description="Get the price of a menu item by name",
    timeout_ms=3000,  # 3秒超时
    parameters={...},
)
def get_menu_item_price(item_name: str) -> str:
    ...