企业官网建设流程全解析

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 个可直接运行的示例脚本与一个公共音频工具模块：

公共工具模块 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 环境变量

运行示例前需要配置服务凭据：

• OpenAI（WebSocket 或 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，在设计用户界面与体验时需要提前考虑。

四、简单语音聊天：WebSocket（Azure OpenAI）与 WebRTC（OpenAI）

4.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. """, voice="shimmer", ) realtime_client = AzureRealtimeWebsocket(settings=settings) audio_player = AudioPlayerWebsocket() audio_recorder = AudioRecorderWebsocket(realtime_client=realtime_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="")

关键点：

1. 换用 OpenAI 只需替换类名：README 明确说明"把AzureRealtimeWebsocket换成OpenAIRealtimeWebsocket即可"，两者 API 形状一致。
2. instructions是会话级系统提示：Realtime API 不使用传统 system message，而是像 Agent 一样把指令作为会话参数传入。
3. voice选择音色：示例使用"shimmer"。源码注释指出可选音色列表会随服务端更新而变化，SDK 不做预校验，以服务端文档为准。
4. AudioRecorderWebsocket需要传入realtime_client：录音器采集到音频帧后直接通过realtime_client.send(...)发送RealtimeAudioEvent（见 utils.py 中的_start_stream：从InputStream读取 PCM 数据、base64 编码后封装为RealtimeAudioEvent发送）。
5. 上下文管理器是核心约定：async with内部会调用create_session创建会话并启动音频流监听（见 RealtimeClientBase 的create_session抽象方法）。

4.2 WebRTC 版：需要 audio_track 与输出回调

simple_realtime_chat_webrtc.py 展示了 WebRTC 路径的差异：

settings = AzureRealtimeExecutionSettings( instructions="""...Mosscap...""", voice="alloy", output_modalities=["text", "audio"], # 同时输出文本转写与音频 ) realtime_client = AzureRealtimeWebRTC( audio_track=AudioRecorderWebRTC(), settings=settings, ) audio_player = AudioPlayerWebRTC() async with audio_player, realtime_client: async for event in realtime_client.receive(audio_output_callback=audio_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="", flush=True) 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_callback=audio_player.client_callback)直接把音频输出回调交给客户端，音频不经过async for循环，README 与源码注释都强调"回调方式更快更平滑"。
• 文本事件要区分 delta 与 done：WebRTC 示例只打印delta增量文本，遇到done事件补一个换行，避免重复输出。
• WebRTC 采样率不同：AudioRecorderWebRTC默认 48000 Hz 单声道、AudioPlayerWebRTC默认 48000 Hz 双声道、帧时长 20ms；WebSocket 版则为 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_name="helpers", functions=[goodbye, get_weather, get_date_time]) realtime_agent = AzureRealtimeWebsocket(credential=AzureCliCredential()) settings = AzureRealtimeExecutionSettings( instructions="""...Mosscap...""", voice="alloy", turn_detection=TurnDetection( type="server_vad", create_response=True, silence_duration_ms=800, threshold=0.8, ), function_choice_behavior=FunctionChoiceBehavior.Auto(), ) chat_history = ChatHistory() chat_history.add_user_message("Hi there, I'm 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(settings=settings, chat_history=chat_history, kernel=kernel, create_response=True), audio_player, ): async for event in realtime_agent.receive(audio_output_callback=audio_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)

值得注意的工程细节：

1. 函数声明：三个函数均以@kernel_function装饰，通过Kernel.add_functions(plugin_name="helpers", ...)注册进 Kernel，再随kernel=kernel参数传给 Agent——这正是 Semantic Kernel 的插件机制在 Realtime 场景的落点。
2. FunctionChoiceBehavior.Auto()：让模型自动决定何时调用函数。该配置最终由服务端转换为tools与tool_choice（OpenAIRealtimeExecutionSettings.tools字段的注释明确写着"不要手动设置，由服务根据 function choice 配置自动生成"）。
3. chat_history播种对话：用ChatHistory预置一轮用户/助手消息，让 Agent 一开场就知道自己身处"阿姆斯特丹"，从而更自然地触发天气查询。
4. create_response=True：Agent 启动后立即开口说话（开场白），可移除此参数关闭。
5. 错误处理：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_track=AudioRecorderWebRTC(), plugins=[Helpers()], ) settings = OpenAIRealtimeExecutionSettings( instructions="""...Mosscap...""", voice="alloy", output_modalities=["text", "audio"], turn_detection=TurnDetection(type="server_vad", create_response=True, silence_duration_ms=800, threshold=0.8), function_choice_behavior=FunctionChoiceBehavior.Auto(), )

对应源码中，OpenAIRealtimeBase.model_post_init会从 model_extra 中读取kernel、plugins、settings、chat_history四个可选参数并完成注入（见 _open_ai_realtime.py），这也解释了为什么plugins=[Helpers()]与kernel=kernel是等价的插件注入方式。

5.3 TurnDetection：服务端 VAD 人声检测

TurnDetection（定义于 open_ai_realtime_execution_settings.py）是让对话"自然衔接"的关键配置，字段如下：

示例采用type="server_vad", create_response=True, silence_duration_ms=800, threshold=0.8：静音 800ms 即视为说完一轮并自动生成响应。源码注释明确警告：如果把turn_detection设为None（关闭服务端 VAD），你就必须自己向 API 发送input_audio_buffer.commit与response.create事件来标记用户说完并触发响应——即手动 VAD，本示例不涉及。

六、会话执行设置详解：输出模态与音频参数

OpenAIRealtimeExecutionSettings（继承自PromptExecutionSettings）与仅作别名子类的AzureRealtimeExecutionSettings定义了会话的全部可调参数，见 open_ai_realtime_execution_settings.py：

• output_modalities：["audio"]、["text"]或["text", "audio"]。WebRTC 示例启用["text", "audio"]以同时获得语音与转写；如果只要语音，可不设。
• voice：会话音色字符串。
• instructions：会话级系统指令。
• input_audio_format/output_audio_format："pcm16"/"g711_ulaw"/"g711_alaw"。
• input_audio_transcription：InputAudioTranscription对象，可选model（whisper-1/gpt-4o-transcribe/gpt-4o-mini-transcribe）、language（ISO-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.voice，turn_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 检测到你停顿后自动开始回复，按Ctrl+C停止程序。函数调用示例则默认Agent 一启动就主动开口（create_response=True）。

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）与threshold（0~1）改善"抢话/迟钝"体验。
• 音量与回声：说话离麦克风过近或扬声器音量过大可能触发回声误判，可结合input_audio_noise_reduction与设备间距调整。
• 用量记录：在 receive 循环中加入ListenEvents.RESPONSE_DONE分支，从event.service_event中提取 usage 信息落日志。

八、深入源码：音频工具类如何工作

utils.py 虽然标注为"演示级、非生产用"，但它完整揭示了 Realtime 音频链路的原理：

• 录音侧（WebSocket）：AudioRecorderWebsocket在__aenter__中启动后台任务，以 24000Hz、单声道、100ms 帧长读取InputStream，将每个frame_size（2400 个采样点）的 PCM 数据 base64 编码后包装成RealtimeAudioEvent通过realtime_client.send()推送（utils.py）。这就是"发送事件回服务端"的典型实现。
• 录音侧（WebRTC）：AudioRecorderWebRTC实现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