用 Amazon Nova Sonic 构建可扩展语音 Agent：多 Agent、工具调用与会话分段实战

用户语音 → ASR(语音转文本) → LLM推理 → TTS(文本转语音) → 播放

用户语音 → Nova Sonic(ASR+TTS) → 单个LLM Agent → 工具调用 → 回复

用户语音 → Nova Sonic → Orchestrator Agent → 分派给专项Agent → 结果汇总 → TTS回复

长对话 → [段1: 开场+意图识别] → [段2: 执行任务] → [段3: 确认+结束]
         每段独立上下文，段间传递摘要

# voice_agent_service.py
# 运行前：pip install strands-agents boto3
# 需要 AWS 凭证配置好，且 Bedrock 区域支持 Nova Sonic

from strands.agent import Agent
from strands.handlers.audio_handler import AudioHandler
from strands.multiagent import Orchestrator, SpecialistAgent
from strands.session import SessionManager
import boto3
import json

# --- 1. 定义专项 Agent ---
order_agent = SpecialistAgent(
    name="order_lookup",
    model_id="us.amazon.nova-micro-v1:0",  # 用快模型做简单查询
    description="查询用户订单状态、物流信息",
    tools=[{
        "name": "lookup_order",
        "description": "根据订单号查询订单状态",
        "parameters": {
            "order_id": {"type": "string", "description": "订单号"}
        },
        # 实际项目中这里接你的订单 API
        "handler": lambda params: {
            "order_id": params["order_id"],
            "status": "已发货",
            "estimated_delivery": "2025-01-20"
        }
    }]
)

refund_agent = SpecialistAgent(
    name="refund_processor",
    model_id="us.amazon.nova-micro-v1:0",
    description="处理退款申请，查询退款政策",
    tools=[{
        "name": "check_refund_policy",
        "description": "查询商品退款政策",
        "parameters": {
            "product_category": {"type": "string", "description": "商品类别"}
        },
        "handler": lambda params: {
            "category": params["product_category"],
            "policy": "7天无理由退货",
            "refund_time": "3-5个工作日"
        }
    }]
)

# --- 2. Orchestrator 路由逻辑 ---
orchestrator = Orchestrator(
    name="voice_router",
    model_id="us.amazon.nova-pro-v1:0",  # 路由需要更强的理解能力
    specialists=[order_agent, refund_agent],
    # 路由 prompt：快速判断意图，不要做多余推理
    system_prompt="""你是语音客服路由器。用户说的话会被实时转文本。
你的唯一任务是判断用户意图并分派给专项Agent。
可选Agent：
- order_lookup: 查订单、物流
- refund_processor: 退款、退货政策

只输出JSON: {"agent": "agent_name", "params": {...}}
不要解释，不要闲聊，直接输出JSON。"""
)

# --- 3. 会话分段管理 ---
session_mgr = SessionManager(
    max_segment_turns=5,          # 每5轮对话切一段
    summary_model="us.amazon.nova-micro-v1:0",  # 段间摘要用快模型
    context_window_tokens=4000    # 每段上下文限制
)

# --- 4. 启动双向音频流服务 ---
def handle_voice_call(audio_stream_in, audio_stream_out):
    """处理一次语音通话的双向流"""

    audio_handler = AudioHandler(
        model_id="us.amazon.nova-sonic-v1:0",  # Nova Sonic 处理ASR+TTS
        input_stream=audio_stream_in,
        output_stream=audio_stream_out
    )

    session = session_mgr.create_session()

    for transcript_chunk in audio_handler.transcribe_stream():
        # 流式拿到用户文本片段
        session.add_user_input(transcript_chunk.text)

        # 检查是否需要切段
        if session.should_segment():
            summary = session.summarize_current_segment()
            session.start_new_segment(summary)

        # 等用户说完一句话再路由（避免半句触发Agent）
        if transcript_chunk.is_complete:
            # 路由到专项Agent
            route_result = orchestrator.route(transcript_chunk.text)

            if route_result["agent"] == "none":
                # 不需要工具，直接让 Nova Sonic 生成回复
                response_text = audio_handler.generate_response(
                    transcript_chunk.text,
                    session.get_context()
                )
            else:
                # 调用专项Agent获取结果
                specialist = orchestrator.get_specialist(route_result["agent"])
                result = specialist.execute(route_result["params"])

                # 把结果喂给 Nova Sonic 生成语音回复
                response_text = audio_handler.generate_response(
                    f"根据查询结果：{json.dumps(result)}，请用自然语言回复用户",
                    session.get_context()
                )

            session.add_agent_response(response_text)

# --- 5. 本地测试入口 ---
if __name__ == "__main__":
    # 生产环境用 WebSocket 或 gRPC 接真实音频流
    # 这里用文件模拟测试
    print("语音Agent服务已配置完成")
    print(f"专项Agents: {order_agent.name}, {refund_agent.name}")
    print(f"Orchestrator: {orchestrator.name}")
    print(f"会话分段: 每{session_mgr.max_segment_turns}轮切一段")

    # 模拟一次路由
    test_route = orchestrator.route("我的订单123456发货了吗")
    print(f"路由结果: {json.dumps(test_route, ensure_ascii=False)}")

import asyncio

async def call_tool_with_timeout(tool_fn, params, timeout_sec=2.0):
    try:
        result = await asyncio.wait_for(tool_fn(params), timeout=timeout_sec)
        return result
    except asyncio.TimeoutError:
        return {"error": "timeout", "message": "查询超时，请稍后再试"}