在 Python 中通过 WebSocket 发送实时音频流，可以通过逐块读取 PCM 文件并使用 `websockets` 库进行传输，实现低延迟的双向通信。WebSocket 协议支持二进制数据传输，适合用于音频流等实时性要求较高的场景[^1]。 ### 实时音频流发送实现 以下是一个使用 `websockets` 库发送实时音频流的示例。该方法通过逐块读取 PCM 文件并逐帧发送，模拟实时音频流的传输过程： ```python import asyncio import websockets async def send_audio_stream(): async with websockets.connect("ws://example.com/audio") as websocket: file_path = "path/to/your/audio.pcm" chunk_size = 1024 # 每次发送的数据块大小 with open(file_path, "rb") as pcm_file: while True: chunk = pcm_file.read(chunk_size) if not chunk: break await websocket.send(chunk) print(f"已发送 {len(chunk)} 字节音频数据") print("音频流发送完成") # 运行异步任务 asyncio.get_event_loop().run_until_complete(send_audio_stream()) ``` ### 接收端处理实时音频流 在接收端，可以使用异步 WebSocket 服务器接收音频流并实时处理或保存为 PCM 文件： ```python import asyncio import websockets async def handle_audio(websocket, path): output_file = "received_audio.pcm" with open(output_file, "wb") as pcm_file: try: while True: chunk = await websocket.recv() pcm_file.write(chunk) print(f"已接收 {len(chunk)} 字节音频数据") except websockets.exceptions.ConnectionClosed: print("连接已关闭") # 启动 WebSocket 服务器 start_server = websockets.serve(handle_audio, "0.0.0.0", 8765) asyncio.get_event_loop().run_until_complete(start_server) asyncio.get_event_loop().run_forever() ``` ### 使用 Base64 编码发送音频流 如果需要将音频流编码为 Base64 字符串进行传输，可以在发送端进行编码并在接收端解码： ```python import asyncio import websockets import base64 async def send_base64_audio_stream(): async with websockets.connect("ws://example.com/audio") as websocket: file_path = "path/to/your/audio.pcm" chunk_size = 1024 with open(file_path, "rb") as pcm_file: while True: chunk = pcm_file.read(chunk_size) if not chunk: break encoded_chunk = base64.b64encode(chunk).decode("utf-8") await websocket.send(encoded_chunk) print(f"已发送 {len(encoded_chunk)} 字节 Base64 编码数据") print("Base64 编码音频流发送完成") asyncio.get_event_loop().run_until_complete(send_base64_audio_stream()) ``` 接收端需进行 Base64 解码以还原原始音频数据： ```python import asyncio import websockets import base64 async def handle_base64_audio(websocket, path): output_file = "received_base64_audio.pcm" with open(output_file, "wb") as pcm_file: try: while True: encoded_chunk = await websocket.recv() chunk = base64.b64decode(encoded_chunk) pcm_file.write(chunk) print(f"已接收 {len(chunk)} 字节原始音频数据") except websockets.exceptions.ConnectionClosed: print("连接已关闭") start_server = websockets.serve(handle_base64_audio, "0.0.0.0", 8765) asyncio.get_event_loop().run_until_complete(start_server) asyncio.get_event_loop().run_forever() ``` 上述方法利用 WebSocket 的双向通信机制，实现低延迟的音频流传输。通过逐块发送 PCM 数据或 Base64 编码字符串，可以在不依赖传统 HTTP 请求-响应模型的情况下实现高效的实时音频传输[^2]。 ---