AgentScope 实时语音与多模态:Realtime Voice Agent 全栈实现导读:实时语音交互是 Agent 的重要能力。AgentScope 提供了完整的 Realtime Voice Agent 全栈实现,支持端到端语音交互、打断、VAD 检测,以及 TTS/ASR 集成和多模态消息处理。本文详解语音交互架构、流式处理机制和实战场景。一、Realtime Voice Agent 概述1.1 核心特性AgentScope 的 Realtime Voice Agent 支持:端到端语音交互:无需手动切换,流畅对话打断处理:支持用户随时打断 Agent 说话VAD 检测:语音活动检测,自动识别说话开始/结束TTS/ASR 集成:支持多种语音引擎低延迟响应:实时流式传输,优化延迟多模态消息:支持图片、视频、音频的混合处理┌─────────────────────────────────────────────────────────┐ │ Realtime Voice Agent 架构 │ ├─────────────────────────────────────────────────────────┤ │ │ │ ┌─────────────┐ ┌─────────────┐ │ │ │ 用户 │ │ Agent │ │ │ │ (语音输入) │ │ (语音输出) │ │ │ └──────┬──────┘ └──────┬──────┘ │ │ │ │ │ │ │ 音频流 │ 音频流 │ │ ▼ ▼ │ │ ┌─────────────────────────────────────────────────────┐│ │ │ 实时语音处理引擎 ││ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ││ │ │ │ VAD │→│ ASR │→│ LLM │ ││ │ │ │(活动检测)│ │(语音识别)│ │(推理) │ ││ │ │ └─────────┘ └─────────┘ └────┬────┘ ││ │ │ │ ││ │ │ ┌───────▼───────┐ ││ │ │ │ TTS │ ││ │ │ │ (语音合成) │ ││ │ │ └───────┬───────┘ ││ │ └──────────────────────────┬───────┴───────┘ │ │ │ │ │ │ 文本/音频流 │ │ ▼ │ │ ┌─────────────────────────────────────────────────────┐│ │ │ 多模态消息层 ││ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ││ │ │ │ 图片URL │ │ 视频帧 │ │音频Base64│ ││ │ │ └─────────┘ └─────────┘ └─────────┘ ││ │ └─────────────────────────────────────────────────────┘│ │ │ └─────────────────────────────────────────────────────────┘二、VAD(语音活动检测)2.1 VAD 基础配置fromagentscope.voiceimportVAD# 创建 VAD 检测器vadVAD(modelsilero,# silero/webrtcthreshold0.5,# 检测阈值min_speech_duration_ms250,# 最小语音时长max_speech_duration_ms5000,# 最大语音时长min_silence_duration_ms100,# 最小静默时长window_size_samples512# 窗口大小)# 实时检测defprocess_audio_stream(audio_stream):foraudio_chunkinaudio_stream:# 检测语音活动is_speechvad.detect(audio_chunk)ifis_speech:print(用户正在说话...)# 累积语音数据speech_buffer.append(audio_chunk)else:print(静音)# 检测到完整语音iflen(speech_buffer)0:complete_speechcombine_chunks(speech_buffer)process_speech(complete_speech)speech_buffer.clear()2.2 打断处理fromagentscope.voiceimportVoiceAgent# 配置打断处理agentVoiceAgent(modelgpt4_model,vadvad,interrupt_config{enabled:True,# 启用打断min_speech_duration:300,# 最小打断语音时长(ms)max_interrupt_latency:500,# 最大打断延迟(ms)stop_immediately:False# 等待当前句子完成})# 实时交互asyncdefvoice_interaction():whileTrue:# 1. 等待用户输入user_audioawaitlisten_for_speech()# 2. 检测是否打断 Agent 说话ifagent.is_speakingandvad.detect_interruption(user_audio):# 停止 Agentagent.stop_speaking()# 处理打断awaithandle_interruption(user_audio)else:# 正常处理用户输入awaitagent.process_input(user_audio)三、TTS/ASR 集成3.1 ASR(语音识别)阿里云 ASRfromagentscope.voiceimportDashScopeASR# 创建 ASRasrDashScopeASR(app_keyyour-app-key,languagezh-CN,formatwav,sample_rate16000)# 实时识别asyncdefreal_time_asr():asyncforaudio_chunkinaudio_stream:# 流式识别textawaitasr.recognize(audio_chunk)print(f识别:{text})# 批量识别asyncdefbatch_asr(audio_file):resultawaitasr.recognize_file(audio_file)print(f完整文本:{result[text]})print(f时间戳:{result[timestamps]})OpenAI Whisperfromagentscope.voiceimportWhisperASR# 本地 WhisperasrWhisperASR(model_sizebase,# tiny/base/small/medium/largedevicecuda,languagezh)# 识别resultasr.recognize(audio_file)print(result[text])3.2 TTS(语音合成)阿里云 TTSfromagentscope.voiceimportDashScopeTTS# 创建 TTSttsDashScopeTTS(app_keyyour-app-key,voicezhixiaoxia,# 发音人volume50,# 音量 0-100speech_rate0,# 语速 -500 到 500pitch_rate0# 音调 -500 到 500)# 文本转语音asyncdeftext_to_speech(text):audioawaittts.synthesize(text)returnaudio# 流式合成asyncdefstreaming_tts(text):asyncforaudio_chunkintts.synthesize_stream(text):# 实时播放awaitplay_audio_chunk(audio_chunk)OpenAI TTSfromagentscope.voiceimportOpenAITTS# OpenAI TTSttsOpenAITTS(modeltts-1,# tts-1/tts-1-hdvoicealloy,# alloy/echo/fable/onyx/nova/shimmerapi_keysk-xxx)# 合成audiotts.synthesize(你好,世界!)四、实时流式处理4.1 音频流式传输fromagentscope.voiceimportStreamingVoiceAgent# 流式语音 AgentagentStreamingVoiceAgent(modelgpt4_model,asrasr,ttstts,vadvad)# 实时对话循环asyncdefvoice_chat():whileTrue:# 1. 监听用户语音user_audio_streamawaitlisten_audio_stream()# 2. 实时 ASRuser_text_streamasr.recognize_stream(user_audio_stream)# 3. 流式推理agent_response_streamagent.stream_run(user_text_stream)# 4. 流式 TTSaudio_streamtts.synthesize_stream(agent_response_stream)# 5. 实时播放awaitplay_audio_stream(audio_stream)4.2 低延迟优化# 优化配置optimization_config{prefetch_enabled:True,# 预取推理结果chunk_overlap:0.1,# 音频块重叠parallel_pipeline:True,# 并行处理buffer_size:512# 缓冲大小}agentStreamingVoiceAgent(modelgpt4_model,optimizationoptimization_config)五、多模态消息5.1 消息对象封装fromagentscope.modelsimportMsg# 文本消息text_msgMsg(roleuser,content用户问题)# 图片 URL 消息image_msgMsg(roleuser,content描述这张图片,image_urlhttps://example.com/image.jpg)# 视频帧消息video_msgMsg(roleuser,content分析这个视频帧,video_framebinary_frame_data)# 音频 Base64 消息audio_msgMsg(roleuser,content这段音频说了什么?,audio_base64base64_encoded_audio)5.2 多模态 AgentfromagentscopeimportMultiModalAgent# 创建多模态 AgentagentMultiModalAgent(modelgpt4_vision_model,asrasr,ttstts)# 处理多模态输入asyncdefprocess_multimodal():# 语音输入 图片user_audioawaitlisten_audio()image_urlcapture_image()# 封装消息msgMsg(roleuser,contentawaitasr.recognize(user_audio),image_urlimage_url)# 推理responseagent.run(msg)# 语音输出awaittts.synthesize_and_play(response.content)六、实战场景6.1 电话客服机器人fromagentscope.voiceimportPhoneBot# 电话机器人phone_botPhoneBot(agentcustomer_service_agent,asrDashScopeASR(),ttsDashScopeTTS(voicezhixiaoxia),vadVAD(),phone_config{sip_server:sip.example.com,username:botagent.com,password:password})# 接听电话asyncdefhandle_incoming_call(call_id):# 1. 建立语音连接awaitphone_bot.connect(call_id)# 2. 问候awaitphone_bot.speak(您好,这里是智能客服,请问有什么可以帮您?)# 3. 对话循环whilephone_bot.is_connected():# 听用户说user_inputawaitphone_bot.listen()# 处理responsephone_bot.agent.run(user_input)# 回复awaitphone_bot.speak(response.content)6.2 智能硬件语音助手fromagentscope.voiceimportSmartDeviceAgent# 智能音箱speaker_agentSmartDeviceAgent(agenthome_assistant_agent,asrWhisperASR(model_sizebase),ttsDashScopeTTS(voicezhicheng),device_config{wakeup_word:小爱同学,led_control:gpio_pin,audio_output:pcm_device_0})# 唤醒对话asyncdefsmart_speaker_interaction():whileTrue:# 等待唤醒awaitspeaker_agent.wait_for_wakeup()# 亮灯speaker_agent.led_on()# 听指令user_inputawaitspeaker_agent.listen()# 执行responsespeaker_agent.agent.run(user_input)# 回复awaitspeaker_agent.speak(response.content)# 关灯speaker_agent.led_off()6.3 实时会议记录fromagentscope.voiceimportMeetingRecorder# 会议记录器recorderMeetingRecorder(asrWhisperASR(model_sizemedium),summarization_modelgpt4_model)# 实时记录asyncdefrecord_meeting():participants{Alice:speaker_1,Bob:speaker_2}asyncforaudio_chunkinmeeting_audio_stream:# 识别说话人speakeridentify_speaker(audio_chunk,participants)# 识别内容textawaitrecorder.asr.recognize(audio_chunk)# 记录recorder.add_entry(speakerspeaker,texttext,timestampdatetime.now())# 实时字幕display_subtitle(f{speaker}:{text})# 生成会议纪要summaryawaitrecorder.generate_summary()print(会议纪要:,summary)七、性能优化7.1 延迟优化fromagentscope.voiceimportLatencyOptimizer# 延迟优化器optimizerLatencyOptimizer(strategies[prefetch,# 预取推理结果parallel,# 并行处理cache,# 缓存常用短语early_tts# 提前开始 TTS])# 配置 AgentagentStreamingVoiceAgent(modelgpt4_model,optimizeroptimizer)# 监控延迟monitorLatencyMonitor()asyncdefmonitor_latency():whileTrue:metricsawaitagent.get_latency_metrics()print(fASR:{metrics[asr_latency]}ms)print(fLLM:{metrics[llm_latency]}ms)print(fTTS:{metrics[tts_latency]}ms)print(f总延迟:{metrics[total_latency]}ms)awaitasyncio.sleep(1)7.2 音频质量优化# 音频处理配置audio_config{noise_reduction:True,# 降噪echo_cancellation:True,# 回声消除gain_control:True,# 自动增益控制sample_rate:16000,# 采样率channels:1# 单声道}agentStreamingVoiceAgent(audio_configaudio_config)八、总结AgentScope 的 Realtime Voice Agent 提供了完整的语音交互解决方案:端到端交互:从语音输入到语音输出的完整链路VAD 检测:准确的语音活动检测,支持打断TTS/ASR 集成:支持阿里云、OpenAI、本地 Whisper流式处理:实时流式传输,低延迟优化多模态支持:图片、视频、音频的混合处理实战场景:电话客服、智能硬件、会议记录Realtime Voice Agent 让 Agent 从文本交互升级为语音交互,提供更自然、更便捷的用户体验。延伸阅读:语音交互文档VAD 算法详解Whisper 模型