Semantic Kernel Python 实时语音多模态实战基于 OpenAI/Azure OpenAI Realtime API 的 WebSocket 与 WebRTC 语音 Agent【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel导读本文以 python/samples/concepts/realtime 目录下的一组官方示例为骨架系统讲解如何在 Semantic Kernel Python 中接入 OpenAI / Azure OpenAI 的Realtime 多模态 API构建边说边听、实时响应的语音对话机器人与具备函数调用能力的语音 Agent。读完本文你将掌握 Realtime API 的事件驱动编程模型、WebSocket 与 WebRTC 两种传输协议的差异与选择、会话Session关键配置项语音、VAD 人声检测、输出模态、函数选择并能直接运行仓库中的四个示例脚本。示例代码本身是演示级实现适合作为学习与二次开发起点生产环境需自行完善音频设备管理与错误处理。一、Realtime 示例概览四份脚本两种协议仓库中 python/samples/concepts/realtime 目录共包含 4 个可直接运行的示例脚本与一个公共音频工具模块示例脚本传输协议服务提供商能力simple_realtime_chat_websocket.pyWebSocketAzure OpenAI可切换到 OpenAI简单语音聊天simple_realtime_chat_webrtc.pyWebRTCOpenAI简单语音聊天realtime_agent_with_function_calling_websocket.pyWebSocketAzure OpenAI语音 Agent 函数调用realtime_agent_with_function_calling_webrtc.pyWebRTCOpenAI语音 Agent 函数调用公共工具模块 utils.py 提供了AudioRecorderWebsocket、AudioPlayerWebsocket、AudioRecorderWebRTC、AudioPlayerWebRTC四个类分别封装麦克风采集与扬声器播放以及check_audio_devices()设备自检函数。从源码注释看这 4 个脚本均由async def main()驱动、以asyncio.run(main())启动既可以通过命令行直接执行也可以在 IDE 中运行。它们均依赖本机麦克风与扬声器属于实时运行类示例。二、环境准备依赖安装与环境变量2.1 安装依赖README 明确要求安装以下 Python 包pip install pyaudio sounddevice pydub semantic-kernel[realtime]其中semantic-kernel[realtime]是带 realtime 扩展的 Semantic Kernel 主包pyaudio、sounddevice、pydub用于本机音频采集、播放与处理。需要注意的是函数调用 WebRTC 示例realtime_agent_with_function_calling_webrtc.py在 docstring 中写的是pip install pyaudio sounddevice pydub semantic-kernel即不依赖 realtime extra 也能运行实际以 README 的统一安装命令为准。2.2 环境变量运行示例前需要配置服务凭据OpenAIWebSocket 或 WebRTC设置 OpenAI API Key以及OPENAI_REALTIME_MODEL_ID指定 Realtime 模型。Azure OpenAI仅 WebSocket设置 endpoint可选设置 Key并设置AZURE_OPENAI_REALTIME_DEPLOYMENT_NAME指定部署名。API 版本至少为2025-08-28——这一点在多个示例源码的注释中被反复强调如 simple_realtime_chat_websocket.py是 Azure Realtime 部署的硬性前提。注意Azure Realtime 目前仅提供 WebSocket 客户端OpenAI 则同时支持 WebSocket 与 WebRTC 两种传输。README 明确说明Environment variables for Azure (websocket only)且函数调用 WebSocket 示例还通过AzureCliCredential()来自azure.identity使用 Azure CLI 登录态进行无密钥认证见 realtime_agent_with_function_calling_websocket.py。2.3 音频设备检查示例在main()外直接调用check_audio_devices()它会打印本机所有音频设备的索引与属性便于你定位正确的麦克风/扬声器设备号check_audio_devices()AudioRecorder*与AudioPlayer*类均接受device设备索引参数默认None表示使用系统默认设备。由于每个人的声卡与麦克风特性不同README 与源码注释均建议演讲者与麦克风的设备特性是流畅对话的最大影响因素可能需要针对不同设备反复试听必要时手动调整设备索引。三、事件驱动模型理解 Realtime 的收发循环Realtime API 的工作方式是双向异步事件流服务端不断向你推送事件你也持续向服务端回发事件。Semantic Kernel 将其封装为统一的异步生成器receive()示例中的核心模式是async for event in realtime_client.receive(): match event: case RealtimeTextEvent(): print(event.text.text, end) case _: ...3.1 事件类型体系SDK 在 realtime_events.py 中定义了事件类型并以event_type作为 Pydantic 判别字段RealtimeEvent所有服务事件的基类携带service_event原始事件内容与service_type事件字符串标识。RealtimeAudioEvent音频事件包含audio: AudioContent用于流式播放模型返回的语音。RealtimeTextEvent文本事件包含text: TextContent对应语音转写文本transcript。RealtimeFunctionCallEvent/RealtimeFunctionResultEvent函数调用与结果事件是函数调用 Agent 的底层支撑。RealtimeImageEvent图像事件。3.2 服务事件枚举ListenEvents底层服务事件的字符串标识集中在 ListenEvents 枚举 中示例中常用的有SESSION_UPDATED session.updated会话创建/更新成功此时可以开始说话。RESPONSE_CREATED response.created模型开始生成响应示例在此打印新的转写前缀。RESPONSE_DONE response.done响应完成该事件携带 usage 用量信息。README 特别提示你可以在 receive 循环的 match case 中追加一条分支把response.done里的用量记录下来用于计费与监控。ERROR error错误事件函数调用示例会将其打印并记入日志。此外还有INPUT_AUDIO_BUFFER_SPEECH_STARTED/STOPPED、CONVERSATION_ITEM_CREATED、RATE_LIMITS_UPDATED等可用于扩展交互逻辑的事件。3.3 转写先于音频中断导致的对不上README 强调了一个重要事实这些 API 的特性决定了转写文本transcript总是先于语音到达。因此如果你在模型说话过程中打断它屏幕上打印的转写将与实际听到的音频不匹配音频已被截断转写却保留了完整内容。这是 Realtime 交互的固有行为并非 bug在设计用户界面与体验时需要提前考虑。四、简单语音聊天WebSocketAzure OpenAI与 WebRTCOpenAI4.1 WebSocket 版Azure 默认OpenAI 一键切换simple_realtime_chat_websocket.py 的核心代码settings AzureRealtimeExecutionSettings( instructions You are a chat bot. Your name is Mosscap and you have one goal: figure out what people need. Your full name, should you need to know it, is Splendid Speckled Mosscap. You communicate effectively, but you tend to answer with long flowery prose. , voiceshimmer, ) realtime_client AzureRealtimeWebsocket(settingssettings) audio_player AudioPlayerWebsocket() audio_recorder AudioRecorderWebsocket(realtime_clientrealtime_client) async with audio_player, audio_recorder, realtime_client: async for event in realtime_client.receive(): match event: case RealtimeAudioEvent(): await audio_player.add_audio(event.audio) case RealtimeTextEvent(): print(event.text.text, end) case _: if event.service_type ListenEvents.SESSION_UPDATED: print(Session updated) if event.service_type ListenEvents.RESPONSE_CREATED: print(\nMosscap (transcript): , end)关键点换用 OpenAI 只需替换类名README 明确说明把AzureRealtimeWebsocket换成OpenAIRealtimeWebsocket即可两者 API 形状一致。instructions是会话级系统提示Realtime API 不使用传统 system message而是像 Agent 一样把指令作为会话参数传入。voice选择音色示例使用shimmer。源码注释指出可选音色列表会随服务端更新而变化SDK 不做预校验以服务端文档为准。AudioRecorderWebsocket需要传入realtime_client录音器采集到音频帧后直接通过realtime_client.send(...)发送RealtimeAudioEvent见 utils.py 中的_start_stream从InputStream读取 PCM 数据、base64 编码后封装为RealtimeAudioEvent发送。上下文管理器是核心约定async with内部会调用create_session创建会话并启动音频流监听见 RealtimeClientBase 的create_session抽象方法。4.2 WebRTC 版需要 audio_track 与输出回调simple_realtime_chat_webrtc.py 展示了 WebRTC 路径的差异settings AzureRealtimeExecutionSettings( instructions...Mosscap..., voicealloy, output_modalities[text, audio], # 同时输出文本转写与音频 ) realtime_client AzureRealtimeWebRTC( audio_trackAudioRecorderWebRTC(), settingssettings, ) audio_player AudioPlayerWebRTC() async with audio_player, realtime_client: async for event in realtime_client.receive(audio_output_callbackaudio_player.client_callback): match event: case RealtimeTextEvent(): if event.service_type and delta in event.service_type and event.text.text: print(event.text.text, end, flushTrue) elif event.service_type and done in event.service_type: print() ...与 WebSocket 版的区别audio_track是必需参数WebRTC 通过RTCPeerConnection传输媒体流AudioRecorderWebRTC实现了MediaStreamTrack接口recv()方法不断产出AudioFrame作为发送给服务端的音频轨道见 utils.py。音频播放走回调而非事件循环receive(audio_output_callbackaudio_player.client_callback)直接把音频输出回调交给客户端音频不经过async for循环README 与源码注释都强调回调方式更快更平滑。文本事件要区分 delta 与 doneWebRTC 示例只打印delta增量文本遇到done事件补一个换行避免重复输出。WebRTC 采样率不同AudioRecorderWebRTC默认 48000 Hz 单声道、AudioPlayerWebRTC默认 48000 Hz 双声道、帧时长 20msWebSocket 版则为 24000 Hz、帧时长 100ms见 utils.py 顶部常量这是因为两种协议下的音频格式约定不同混用会导致杂音或无声。注意该脚本虽然导入了AzureRealtimeExecutionSettings与AzureRealtimeWebRTC但 README 将其归类为 OpenAI WebRTC 示例——从源码看AzureRealtimeWebRTC与OpenAIRealtimeWebRTC共享同一套执行设置结构AzureRealtimeExecutionSettings直接继承OpenAIRealtimeExecutionSettings切换提供商时同样只需替换客户端类名。五、语音 Agent 函数调用让 Agent 替你做事两个函数调用示例演示了如何让语音 Agent 在对话中调用 Semantic Kernel 插件函数。README 列出了三个内置函数get_weather(location)返回指定城市的天气数据是随机生成的不代表真实天气get_date_time()返回当前日期时间goodbye()结束对话raise KeyboardInterrupt。每次函数被调用都会输出一行日志。5.1 WebSocket 版Kernel 插件注册realtime_agent_with_function_calling_websocket.py 的关键流程kernel_function def get_weather(location: str) - str: Get the weather for a location. ... kernel Kernel() kernel.add_functions(plugin_namehelpers, functions[goodbye, get_weather, get_date_time]) realtime_agent AzureRealtimeWebsocket(credentialAzureCliCredential()) settings AzureRealtimeExecutionSettings( instructions...Mosscap..., voicealloy, turn_detectionTurnDetection( typeserver_vad, create_responseTrue, silence_duration_ms800, threshold0.8, ), function_choice_behaviorFunctionChoiceBehavior.Auto(), ) chat_history ChatHistory() chat_history.add_user_message(Hi there, Im based in Amsterdam.) chat_history.add_assistant_message(I am Mosscap, ... I can tell you what the weather is or the time.) async with ( audio_recorder, realtime_agent(settingssettings, chat_historychat_history, kernelkernel, create_responseTrue), audio_player, ): async for event in realtime_agent.receive(audio_output_callbackaudio_player.client_callback): match event: case RealtimeTextEvent(): if print_transcript: print(event.text.text, end) case _: match event.service_type: case ListenEvents.RESPONSE_CREATED: if print_transcript: print(\nMosscap (transcript): , end) case ListenEvents.ERROR: print(event.service_event) logger.error(event.service_event)值得注意的工程细节函数声明三个函数均以kernel_function装饰通过Kernel.add_functions(plugin_namehelpers, ...)注册进 Kernel再随kernelkernel参数传给 Agent——这正是 Semantic Kernel 的插件机制在 Realtime 场景的落点。FunctionChoiceBehavior.Auto()让模型自动决定何时调用函数。该配置最终由服务端转换为tools与tool_choiceOpenAIRealtimeExecutionSettings.tools字段的注释明确写着不要手动设置由服务根据 function choice 配置自动生成。chat_history播种对话用ChatHistory预置一轮用户/助手消息让 Agent 一开场就知道自己身处阿姆斯特丹从而更自然地触发天气查询。create_responseTrueAgent 启动后立即开口说话开场白可移除此参数关闭。错误处理ListenEvents.ERROR分支会打印event.service_event并记入日志便于排查。5.2 WebRTC 版以 plugins 参数注入插件realtime_agent_with_function_calling_webrtc.py 展示了两种等价写法——函数被封装进Helpers类并通过plugins[Helpers()]参数注入class Helpers: kernel_function def get_weather(self, location: str) - str: ... kernel_function def get_date_time(self) - str: ... kernel_function def goodbye(self): ... realtime_agent AzureRealtimeWebRTC( audio_trackAudioRecorderWebRTC(), plugins[Helpers()], ) settings OpenAIRealtimeExecutionSettings( instructions...Mosscap..., voicealloy, output_modalities[text, audio], turn_detectionTurnDetection(typeserver_vad, create_responseTrue, silence_duration_ms800, threshold0.8), function_choice_behaviorFunctionChoiceBehavior.Auto(), )对应源码中OpenAIRealtimeBase.model_post_init会从 model_extra 中读取kernel、plugins、settings、chat_history四个可选参数并完成注入见 _open_ai_realtime.py这也解释了为什么plugins[Helpers()]与kernelkernel是等价的插件注入方式。5.3 TurnDetection服务端 VAD 人声检测TurnDetection定义于 open_ai_realtime_execution_settings.py是让对话自然衔接的关键配置字段如下字段取值说明typeserver_vad默认/semantic_vad人声检测模式create_responsebool检测到一轮说完后是否自动创建响应interrupt_responsebool是否允许打断当前响应eagernesslow/medium/high/auto仅semantic_vad使用检测积极度prefix_padding_msint 0检测到语音前的填充时间毫秒silence_duration_msint 0判定说完了所需的静音时长毫秒thresholdfloat0~1仅server_vad使用语音检测阈值示例采用typeserver_vad, create_responseTrue, silence_duration_ms800, threshold0.8静音 800ms 即视为说完一轮并自动生成响应。源码注释明确警告如果把turn_detection设为None关闭服务端 VAD你就必须自己向 API 发送input_audio_buffer.commit与response.create事件来标记用户说完并触发响应——即手动 VAD本示例不涉及。六、会话执行设置详解输出模态与音频参数OpenAIRealtimeExecutionSettings继承自PromptExecutionSettings与仅作别名子类的AzureRealtimeExecutionSettings定义了会话的全部可调参数见 open_ai_realtime_execution_settings.pyoutput_modalities[audio]、[text]或[text, audio]。WebRTC 示例启用[text, audio]以同时获得语音与转写如果只要语音可不设。voice会话音色字符串。instructions会话级系统指令。input_audio_format/output_audio_formatpcm16/g711_ulaw/g711_alaw。input_audio_transcriptionInputAudioTranscription对象可选modelwhisper-1/gpt-4o-transcribe/gpt-4o-mini-transcribe、languageISO-639-1 格式与prompt。input_audio_noise_reduction{type: near_field | far_field}降噪配置。max_output_tokens正整数或inf。tools/tool_choice不要手动设置由函数选择配置自动生成。底层序列化值得关注prepare_settings_dict()方法会把voice、turn_detection、input_audio_format、output_audio_format、input_audio_transcription、input_audio_noise_reduction从平铺字段重组为 OpenAI API 要求的嵌套结构——voice归入audio.output.voiceturn_detection归入audio.input.turn_detection格式类字段归入audio.input/output.format。也就是说你在 SDK 里写的是平铺的 Pydantic 字段发往服务端时自动变成{ instructions: ..., audio: { input: { turn_detection: {...}, format: pcm16, transcription: {...} }, output: { voice: alloy, format: pcm16 } } }理解这一点对排查为什么设置没生效非常有帮助。七、运行方式与预期输出7.1 启动与操作以 WebSocket 简单聊天为例python python/samples/concepts/realtime/simple_realtime_chat_websocket.py启动后终端会打印操作提示看到 Session updated. 后开始说话模型通过服务端 VAD 检测到你停顿后自动开始回复按CtrlC停止程序。函数调用示例则默认Agent 一启动就主动开口create_responseTrue。7.2 输出格式约定所有示例的输出格式一致每次新的response item到达时打印一行新的Mosscap (transcript):前缀RESPONSE_CREATED事件触发转写文本紧随其后逐个增量打印函数调用发生时日志输出 Getting weather for ...、 Getting current datetime、 Goodbye has been called!之类的标记行。7.3 常见调优点设备选择为麦克风与扬声器分别挑选合适的设备索引传入AudioRecorder*/AudioPlayer*的device参数。VAD 灵敏度调节silence_duration_ms如 500~1000ms与threshold0~1改善抢话/迟钝体验。音量与回声说话离麦克风过近或扬声器音量过大可能触发回声误判可结合input_audio_noise_reduction与设备间距调整。用量记录在 receive 循环中加入ListenEvents.RESPONSE_DONE分支从event.service_event中提取 usage 信息落日志。八、深入源码音频工具类如何工作utils.py 虽然标注为演示级、非生产用但它完整揭示了 Realtime 音频链路的原理录音侧WebSocketAudioRecorderWebsocket在__aenter__中启动后台任务以 24000Hz、单声道、100ms 帧长读取InputStream将每个frame_size2400 个采样点的 PCM 数据 base64 编码后包装成RealtimeAudioEvent通过realtime_client.send()推送utils.py。这就是发送事件回服务端的典型实现。录音侧WebRTCAudioRecorderWebRTC实现MediaStreamTrack.recv()通过队列把sounddevice回调里的np.ndarray转成 48000Hz 的AudioFrame交给 WebRTC 连接utils.py。播放侧两个AudioPlayer*类内部都维护一个音频队列client_callback(content)把ndarray推入队列OutputStream的 sounddevice 回调在数据不足时用零填充避免爆音或卡顿utils.py。WebSocket 播放器还额外提供add_audio()方法供在 receive 循环里手动播放RealtimeAudioEvent这一更简单的替代路径使用。理解这层封装后你可以替换为自己的生产级实现例如使用更稳健的音频缓冲区策略、加入回声消除或设备热插拔处理。九、小结在 Semantic Kernel Python 中使用 Realtime API 的核心是四个客户端类AzureRealtimeWebsocket、OpenAIRealtimeWebsocket、AzureRealtimeWebRTC、OpenAIRealtimeWebRTC以及配套的执行设置类AzureRealtimeExecutionSettings/OpenAIRealtimeExecutionSettings。WebSocket 与 WebRTC 行为一致、协议迥异WebRTC 必须提供audio_track、音频播放推荐走audio_output_callback且两者音频采样率/声道约定不同。事件驱动是唯一交互方式用async for event in receive()配合RealtimeTextEvent/RealtimeAudioEvent与ListenEvents枚举处理转写、音频与服务端事件。会话配置决定体验instructions、voice、output_modalities、TurnDetection、FunctionChoiceBehavior.Auto()与ChatHistory播种共同构成一个可对话、可调函数的语音 Agent。Azure 硬性前提API 版本需 ≥2025-08-28并设置AZURE_OPENAI_REALTIME_DEPLOYMENT_NAME。建议下一步依次运行 simple_realtime_chat_websocket.py 与 realtime_agent_with_function_calling_webrtc.py先建立能听会说的基线再逐步调整 VAD 参数与插件函数体会两种协议在实时语音场景下的工程差异。【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考