OpenAI Realtime 语音 API 的 Python 客户端：ell 仓库 openai_realtime 子项目实战解析

人工智能大模型提示工程AI 应用【免费下载链接】ellA language model programming library.项目地址https://gitcode.com/gh_mirrors/ell/ell点击查看免费下载x/openai_realtime是 ell 语言模型编程仓库中的一个独立子项目它把 OpenAI 官方的 Realtime API 客户端原为 JavaScript 实现移植为 Python 版本让你可以通过 WebSocket 与gpt-4o-realtime-preview系列模型进行实时文本/音频对话、注册自定义函数工具并管理会话状态。本文将以该子项目文档为主线结合仓库源码、测试与示例带你掌握从安装、连接、发送消息、处理流式增量、工具调用到语音输入输出的完整实战能力并厘清它作为 ell 未来实时绑定参考实现的设计思路。项目定位ell 的实时语音实验分支在动手之前先明确这个子项目在整个仓库中的位置。ell 主仓库在 x/README.md 中说明x目录存放的是「面向各类 LLM 提供方的实验性客户端未来可能被整合进 ell」。openai_realtime正是其中之一。关联文档 明确给出了两条重要信息它是 William Guss 对 OpenAI Realtime 官方客户端JS 版的Python 移植遵循 MIT 协议并保留原作者的版权声明见 x/openai_realtime/LICENSE它被设计为ell 未来实现 realtime Python 绑定的参考实现。也就是说读这个子项目不仅能直接用于开发实时语音对话应用还能提前理解 ell 团队对未来实时交互能力的设计取向。另外ell 主仓库的 OpenAI 模型适配层 中已经收录了gpt-4o-realtime-preview-2024-10-01、gpt-4o-realtime等实时模型名称可见实时能力已被纳入 ell 的整体模型生态规划。功能特性概览文档给出的核心特性如下它们也是下文将要逐一展开的实战主线Realtime 通信通过 WebSocket 与 OpenAI Realtime API 建立持久双向连接多模态支持同时支持text文本与audio音频两种模态工具集成注册自定义 function 工具让模型在对话中实时调用会话管理与事件处理维护会话状态、订阅并分发各类服务端事件异步操作基于asyncio实现高性能并发收发。安装与运行环境安装步骤文档给出的安装方式是从仓库克隆后以可编辑模式安装git clone https://gitcode.com/gh_mirrors/ell/ell cd x/openai_realtime pip install -e .依赖与版本约束从 x/openai_realtime/pyproject.toml 可以看到该项目的依赖与约束安装前建议先对照你的环境依赖版本要求用途Python^3.12运行时版本要求websockets^10.4WebSocket 底层通信aiohttp^3.8.4异步 HTTP 支持asyncio^3.4.3异步编程基础库pydub^0.25.1音频文件加载与转换开发与测试依赖包括pytest、pytest-asyncio、black与isort。pyproject.toml中的asyncio_mode auto配置意味着测试中的async测试函数会被pytest-asyncio自动识别执行。注意pydub在 Python 3.13 有弃用警告建议按工程声明使用 Python 3.12。快速上手一次文本对话文档给出的 Quick Start 是最小可用示例完整继承了它并稍作整理如下from openai_realtime import RealtimeClient async def main(): client RealtimeClient(api_keyyour-api-key) await client.connect() # 发送一条文本消息 client.send_user_message_content([{type: text, text: Hello, AI!}]) # 等待 AI 回复完成 response await client.wait_for_next_completed_item() print(response[item][formatted][text]) client.disconnect() if __name__ __main__: import asyncio asyncio.run(main())这段代码背后实际发生了什么结合 client.py 源码看RealtimeClient(api_key...)会创建RealtimeAPI负责 WebSocket、RealtimeConversation负责会话状态并预先注册一批内部事件处理器await client.connect()建立 WebSocket 连接并发送session.update推送默认会话配置send_user_message_content()先发送conversation.item.create构造一条role: user的 message 条目随后立即发送response.create触发模型响应await client.wait_for_next_completed_item()内部监听conversation.item.completed事件详见下文事件机制返回包含formatted.text的完整条目client.disconnect()关闭连接并清空会话。这一「发送消息 → 触发响应 → 等待完成条目」的三角循环是使用该客户端的最基本节奏。五大核心组件文档列出了五个核心组件下面结合源码逐一说明其职责这五者在 x/openai_realtime/src/openai_realtime/init.py 中全部作为公开 API 导出组件源码文件职责RealtimeClientclient.py面向应用层的主客户端组合其余组件提供send_user_message_content、add_tool、update_session等高层方法RealtimeAPIapi.py负责 WebSocket 连接的建立、消息收发与低层事件分发RealtimeConversationconversation.py维护会话中的条目items、响应responses并将服务端事件增量更新到本地状态RealtimeEventHandlerevent_handler.py通用事件系统支持on/on_next/off注册、注销与wait_for_next异步等待RealtimeUtilsutils.py工具函数float32↔PCM16 转换、base64 编解码、数组合并、事件 ID 生成它们之间的依赖关系是RealtimeClient继承RealtimeEventHandler内部组合RealtimeAPI同样继承RealtimeEventHandler与RealtimeConversation并静态使用RealtimeUtils。默认会话配置参数全解RealtimeClient在初始化时构造了一份default_session_config见 client.py这些参数会随session.update事件发送给服务端。参数及其默认值、含义如下参数默认值说明modalities[text, audio]本次会话启用的响应模态可仅保留[text]instructions传入值默认为空字符串系统级指令类似系统提示词voicealloy语音合成音色input_audio_formatpcm16输入音频编码格式output_audio_formatpcm16输出音频编码格式input_audio_transcriptionNone输入音频转写配置置为{enabled: True, model: whisper-1}可启用语音转写turn_detectionNone轮次检测None表示手动控制字典则用于配置服务端 VADtools[]工具列表运行时由已注册工具动态生成tool_choiceauto工具选择策略temperature0.8采样温度max_response_output_tokens4096单次响应的最大输出 token 数此外还有两个关联配置transcription_models默认为[{model: whisper-1}]用于input_audio_transcription的转写模型default_server_vad_config提供了一套开箱即用的服务端语音活动检测VAD参数threshold0.5语音门限、prefix_padding_ms300前置填充、silence_duration_ms200静音判定时长。这些配置可以通过client.update_session(**kwargs)在运行期覆盖例如client.update_session( instructionsYou are a helpful assistant., modalities[text], temperature0.2, turn_detection{ type: server_vad, threshold: 0.5, prefix_padding_ms: 300, silence_duration_ms: 300, }, input_audio_transcription{enabled: True, model: whisper-1}, )update_session的实现要点见 client.py它会将当前已注册工具合并进tools每个工具自动附带type: function如果连接已建立则立即发送session.update事件。也就是说先注册工具再调用update_session工具会随会话更新一并生效。底层原理WebSocket 连接与事件分发建立连接RealtimeAPI.connect()见 api.py的默认连接地址是wss://api.openai.com/v1/realtime通过url参数可覆盖模型默认值为gpt-4o-realtime-preview-2024-10-01通过?model查询参数指定。握手时携带两个请求头headers { Authorization: fBearer {self.api_key}, OpenAI-Beta: realtimev1, }连接成功后RealtimeAPI启动一个后台任务_message_handler()用async for message in self.ws持续接收服务端 JSON 消息解析出type字段后调用receive()分发。双向事件分发RealtimeAPI.send()在发送前会为每条事件生成event_id前缀evt_构造{event_id, type, **data}结构并同时触发两个本地事件client.{event_name}精确事件名client.*通配事件名。receive()同理对每条服务端消息触发server.{event_name}与server.*。这一「精确 通配」的双通道设计让你既能订阅单一事件也能用realtime.on(server.*, ...)之类的方式观察全部流量非常便于调试。RealtimeEventHandler见 event_handler.py提供了完整的事件原语on(event_name, callback)注册常驻处理器同时支持装饰器与直接传回调两种写法on_next(event_name, callback)注册一次性处理器触发后自动移除off(event_name, callbackNone)注销处理器wait_for_next(event_name, timeoutNone)异步阻塞等待下一次指定事件的触发这是wait_for_next_item()/wait_for_next_completed_item()的底层实现dispatch(event_name, event)同步调用常驻处理器并消费一次性处理器。RealtimeClient 对服务端事件的内部接线RealtimeClient._add_api_event_handlers()见 client.py将这些底层事件转换为高层业务信号关键映射包括服务端事件处理动作server.session.created置session_created True作为「会话就绪」标志server.response.created/output_item.added/content_part.added转发给conversation.process_event()更新状态并派发conversation.updatedserver.input_audio_buffer.speech_started记录语音起始同时派发conversation.interrupted用于打断播放server.input_audio_buffer.speech_stopped结合本地输入音频缓冲切片出该轮语音server.conversation.item.created派发conversation.item.appendedwait_for_next_item()依赖它server.response.output_item.done若条目状态为completed派发conversation.item.completedwait_for_next_completed_item()依赖它若该条目是 function 调用则自动触发工具执行response.audio.delta/text.delta/audio_transcript.delta/function_call_arguments.delta流式增量逐段更新条目内容is_connected()的实现是「WebSocket 已连接且会话已创建」双重判定而wait_for_session_created()则是在循环中轮询session_created标志这是快速上手示例中「连接后应先等待会话创建」的原因。会话状态管理增量如何累积成完整条目RealtimeConversation见 conversation.py以 24kHzdefault_frequency 24000为音频时间基准内部维护item_lookup条目索引、items条目列表、response_lookup/responses以及若干队列。它的核心机制是按事件类型动态分发process_event()读取事件的type将其中的.替换为_后查找对应的_process_*方法。例如_process_conversation_item_created为新条目建立formatted结构含audio/text/transcript/tool等格式化字段并依据角色与类型初始化状态_process_response_text_delta把文本增量累加到内容与formatted[text]_process_response_audio_delta把 base64 的音频增量解码为 int16 数组并np.concatenate到formatted[audio]_process_response_function_call_arguments_delta把工具参数增量追加到arguments_process_conversation_item_truncated按audio_end_ms截断音频数组并清空转写文本。这也解释了为什么响应条目同时带有「内容数组」和「格式化字段」两套表示前者是服务端事件的原始积累后者是面向应用的便捷视图formatted.text、formatted.transcript、formatted.audio。实战进阶注册自定义工具文档给出了注册工具的最小示例。结合 client.py 与 tests/test_mock.py 的校验逻辑add_tool(definition, handler)会执行三项校验definition必须包含name否则抛ValueError(Missing tool name in definition)工具名不能重复注册否则需要先remove_tool(name)handler必须可调用。一个更完整、可直接运行的示例async def my_tool_handler(args): # 在这里实现你的工具逻辑 return {result: Tool output} client.add_tool( {name: my_tool, description: A custom tool, parameters: {}}, my_tool_handler )需要注意一个从源码中能明确看到的细节_call_tool()中处理器是以await tool_confighandler方式调用的见 client.py因此实际运行时处理器应定义为async def文档中的同步def写法只用于示意工具定义结构。测试 test_mock.py 也正是用AsyncMock验证这一调用约定。工具调用闭环如下模型在响应中产出function_call条目其arguments以增量事件逐段送达server.response.output_item.done触发_call_tool()解析argumentsJSON处理器执行后客户端发送conversation.item.create类型的function_call_output携带call_id与output回传结果若处理器抛异常则把{error: str(e)}作为输出回传参见test_call_tool_error的验证最后调用create_response()让模型基于工具结果继续生成。test_mock.py的test_add_tool还验证了注册工具后session.update的载荷中会包含{name: ..., description: ..., type: function}的工具定义印证了update_session中自动合并tools的实现。音频处理从麦克风到 PCM 流Realtime API 的音频以16-bit PCM、24kHz 单声道为统一交换格式。文档给出的音频处理片段对应的是手动非 VAD模式import numpy as np # 追加输入音频数据 audio_data np.array([...], dtypenp.int16) client.append_input_audio(audio_data) # 触发一次响应如可用则包含音频输出 client.create_response()append_input_audio()见 client.py做了两件事把音频数组 base64 编码后发送input_audio_buffer.append事件同时把 int16 数组合并进本地input_audio_buffer。而create_response()在未启用轮次检测时会先发送input_audio_buffer.commit提交缓冲并排队本地音频再发送response.create。RealtimeUtils 提供了完整的数据转换工具链float_to_16bit_pcm(float32_array)将[-1, 1]的 float32 数组截断并放大为 int16 字节流array_buffer_to_base64(array_buffer)智能转换——float32 数组先转 PCM16int16 数组直接tobytes()再统一 base64 编码send_user_message_content处理input_audio类型内容时即调用它base64_to_array_buffer(base64_string)解码服务端音频增量merge_int16_arrays(left, right)合并两段 int16 数据兼容 bytes 与 ndarray。需要特别说明的是文档中的音频示例只是手动模式。在实际语音助手场景中更推荐启用服务端 VAD即前文turn_detection的server_vad配置由模型按语音起止自动分段客户端只需持续把麦克风数据送进append_input_audio()。RealtimeConversation会在speech_stopped时用本地缓冲与audio_start_ms/audio_end_ms切出本轮语音见 conversation.py。仓库示例实战语音助手、音频文件与 Discord 机器人x/openai_realtime/examples目录提供了三个由浅入深的完整示例可作为实战模板1. audio_example.py音频文件对话用pydub加载toronto.mp3示例音频并转 base64构造[{type: input_audio, audio: audio_sample}]内容后调用send_user_message_content()同时通过client.realtime.on(server.response.audio.delta)订阅输出音频增量并用sounddevice实时播放。该脚本结构简洁是理解「输入音频文件 → 输出语音」的最小闭环。2. chat_assistant_clone.py双工语音助手这是最完整的双工示例用sounddevice的InputStream采集麦克风、OutputStream播放音频两个异步 worker 分别负责音频输入append_input_audio与输出播放初始化时通过update_session启用server_vad并借助speech_started事件在用户开口时清空播放队列实现打断barge-in。它演示了RealtimeClient与真实音频设备、事件循环集成的全部要点。3. discord_gpt4o.pyDiscord 语音机器人将客户端接入 Discord 的语音频道MySink把 48kHz 立体声 PCM 降采样为 24kHz 单声道送入输入队列PyAudioSource把 24kHz 输出重采样回 48kHz 立体声播放。配套的 run_bot.sh 提供了带日志与崩溃自动重启的守护式运行脚本。该示例展示了把 realtime 能力嵌入第三方语音平台的完整思路。测试与验证如何确认客户端行为子项目自带两套测试可帮助你理解并验证客户端行为tests/test_mock.py纯 Mock 单元测试不访问网络。它系统验证了初始化、reset、connect连接后发送session.update、add_tool/remove_tool、delete_item、update_session、send_user_message_content先conversation.item.create后response.create、append_input_audio、create_response、cancel_response、两个wait_for_next_*方法以及工具调用的成功/异常路径。tests/test_audio.py真实端到端集成测试需要设置OPENAI_API_KEY环境变量并可访问 Realtime API。它使用 tests/samples/toronto.mp3 作为输入音频通过client.on(realtime.event, ...)观察事件流断言事件顺序为client.session.update→server.session.created→client.conversation.item.create→client.response.create并最终校验助手回复的formatted.transcript中包含 toronto。在x/openai_realtime目录下可执行pytest运行测试。需要注意的是Mock 测试无需任何密钥即可快速验证逻辑而音频集成测试会产生真实 API 调用与费用建议按需谨慎运行。与 ell 的关系与适用边界最后回到定位问题。本客户端在 ell 仓库中处于「实验性参考」状态见 x/README.md文档也明确说明它是 ell 未来 realtime 绑定的实现参考。因此使用时请把握以下边界这是独立子项目与 ell 主库的ell.simple等装饰器体系暂不打通需在x/openai_realtime目录内单独安装、单独使用连接依赖 OpenAI Realtime API 的可用性与 API Key模型默认指向gpt-4o-realtime-preview-2024-10-01文档中的同步工具处理器示例与源码的await调用方式存在出入实际开发应以async def处理器为准参见前文依赖声明面向 Python 3.12使用其他 Python 版本时请自行验证依赖兼容性。总体而言x/openai_realtime是一个结构清晰、注释到位、测试完善的实时语音客户端参考实现从RealtimeAPI的 WebSocket 细节、RealtimeConversation的增量状态机到RealtimeClient的高层业务 API每一层都值得细读。无论是想快速搭建实时语音对话应用还是想为 ell 的未来实时能力做技术储备它都是一个理想的起点。参考资源仓库内子项目说明文档x/openai_realtime/README.md高层客户端src/openai_realtime/client.pyWebSocket 层src/openai_realtime/api.py会话状态机src/openai_realtime/conversation.py事件系统src/openai_realtime/event_handler.py音频/编解码工具src/openai_realtime/utils.py工程配置与依赖pyproject.toml示例程序chat_assistant_clone.py、audio_example.py、discord_gpt4o.py测试用例test_mock.py、test_audio.py赞分享人工智能大模型提示工程AI 应用【免费下载链接】ellA language model programming library.项目地址https://gitcode.com/gh_mirrors/ell/ell点击查看免费下载相关推荐3行代码实现实时语音交互OpenAI Python库Realtime API实战指南3行代码实现实时语音交互OpenAI Python库Realtime API实战指南 你是否还在为实时语音交互的复杂实现而困扰是否想让你的应用拥有像智能音箱人工智能大模型Semantic Kernel Python 实时语音多模态实战基于 OpenAI/Azure OpenAI Realtime API 的 WebSocket 与 WebRTC 语音 AgentSemantic Kernel Python 实时语音多模态实战基于 OpenAI/Azure OpenAI Realtime API 的 WebSocket人工智能大模型AI AgentAgent 框架多智能体RAGopenai-agents-python Realtime Tracing 架构解析双轨追踪体系与客户端/服务端 Trace 关联实践openai agents python Realtime Tracing 架构解析双轨追踪体系与客户端/服务端 Trace 关联实践 导读 本文基于 ope人工智能AI AgentAgent 框架多智能体工具调用MCP Clients上一篇沉浸式翻译扩展使用教程下一篇Git Magic分支魔法如何快速掌握Git分支与合并的终极秘籍创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考