简介基于WebSocket协议实现语音实时收发的Python服务器与客户端代码适合有Python基础、希望快速搭建双向语音通信原型的开发者。资源包含服务器端与客户端两个主要脚本采用Flask-SocketIO处理服务端连接与消息事件客户端通过websocket-client库建立长连接并发送音频数据代码结构简洁便于二次开发。包体内含3个文件包括2个py脚本和1个备份文件压缩包仅9KB轻量易用。主程序master_server.py与client.py分别对应服务端和客户端bak文件为服务器脚本的备份版本可用于对比修改。目前已有3112人学习下载适合用于理解WebSocket双向通信机制、语音数据的JSON封装传输方式以及Flask-SocketIO的事件处理流程。结合语音识别扩展思路可在此基础上接入Kaldi等工具链进行实时语音识别处理。整体虽小但完整覆盖了从连接建立、数据发送到服务端接收处理的闭环适合学习、演示与功能扩展。1. WebSocket 语音收发一套 Python 服务器和客户端跑通实时通话很多人以为 WebSocket 只会传 JSON其实把二进制音频帧直接塞进去做语音收发也相当顺手。这套 websocket 语音方案是一份完整的 Python 服务器与客户端代码客户端录一小段 push 给服务器服务器按房间转发到对端另一边解码播放整个链路就是实时对讲、在线会议、智能硬件语音中继的基础骨架。你不用碰 WebRTC、不用改 protobuf只要 Python 环境能跑就能在两个节点之间通语音。我拆这份代码时最大的感受是帧大小和缓冲队列调对了声音自然顺滑这两个点没想清楚再好的网络也会出现鬼畜音。2. 音频帧与传输设计WebSocket 语音包的边界条件先算清楚先不谈代码先谈数据长什么样。语音进 WebSocket 之前至少要决定采样率、位深、单双声道和每次发送的帧长。这四个参数直接决定网络占用和流畅度很多下载下来的案例跑不通就是因为在参数上随手写没有按实际场景算。2.1 为什么选 WebSocket 而不是 HTTP 轮询实时语音要求的延迟在 200ms 以内。HTTP 轮询的常规做法是客户端每隔几百毫秒去拉一次数据服务器把最近一段音频返回。表面上能跑但轮询间隔就是天然延迟地板而且每轮请求都要带 HTTP 头公网环境下丢包重传很容易把声音打断。WebSocket 的优势是连接建立之后变成全双工管道。客户端不需要主动问“有没有新数据”服务器有数据就直接往这条连接里塞。一次 Upgrade 握手之后剩余会话里只有音频数据和极少量控制帧协议头开销比 HTTP 小得多。对语音这种高频小包场景WebSocket 的二进制帧能够直接用bytes传输不需要再做一层 base64 包装。我在模拟项目X 里对比过同样一段 10 秒语音HTTP 轮询的端到端延迟在 300ms 上下波动WebSocket 固定连接能做到 90ms 以内。声音交互的体感差异非常明显这也是我选择这份资源来拆解的主要原因。2.2 音频参数和帧大小怎么配合才能不卡顿常见的做法是把 PCM 裸数据切成 20ms 一帧。20ms 这个值对人和对网络都友好小于 10ms 会明显增加网络包数量大于 50ms 又会感觉延迟明显。切帧公式是这样帧字节数 采样率 × 位深 ÷ 8 × 声道数 × 时长按 20ms 一帧算典型参数如下采样率位深单声道字节率20ms 帧大小每秒帧数大致码率8000 Hz16bit16000 B/s320 B50128 kbps16000 Hz16bit32000 B/s640 B50256 kbps24000 Hz16bit48000 B/s960 B50384 kbps48000 Hz16bit96000 B/s1920 B50768 kbps局域网里用 16000 Hz、640B 一帧是我最常用的组合。原因有三一是人声频段主要在 300Hz 到 3400Hz16k 采样足够二是 640B 的帧即使不做压缩单路码率也只有 256kbps三是这个帧大小在 WebSocket 的二进制帧里非常轻分摊到每个包上的网络头开销可以忽略。这里有一个新手容易踩的坑不要一上来就 48000 Hz。高采样率会放大丢帧时的听感瑕疵尤其是在公网环境网络抖动造成的破音会比低采样率更刺耳。先把 16k 跑通再往上升不迟。音频队列也要提前设计。录音设备回调频率和网络发送速度不是完全同步的所以客户端要有一个有界队列。我一般把队列最大长度设为 20 帧也就是预留 400ms 缓冲。队列满了就直接丢最老的帧而不是无限堆积。无限堆积的后果是网络恢复之后积压的语音一下子涌出来对端听到的是像磁带快进一样的加速声音反而比丢帧更致命。提示20 帧队列是保底方案。公网延迟超过 400ms 的环境请加大到 50 帧但声音延迟也会相应增加需要在音质和实时性之间做取舍。3. 服务端实现asyncio 房间转发与进程保护参数服务端是整个语音链路的交换核心。它不负责编解码只做两件事维护房间内的 WebSocket 连接把收到的二进制音频帧转发给房间其他人。代码量不大但 asyncio 模型里的并发细节需要看明白。3.1 依赖与最小可运行环境我拆这份代码时用到的依赖是websockets和sounddevice。服务端只需要websockets客户端额外需要sounddevice来读写麦克风和声卡。安装命令一般是这样pip install websockets sounddevicewebsockets负责协议层sounddevice负责把音频设备的回调数据送进 Python。如果项目里已经用PyAudio也可以继续用只是回调方式略有区别核心的 WebSocket 部分共用同一套逻辑。3.2 服务器转发逻辑按房间隔离不给自己回放下面这段代码是按房间转发的完整实现也是这个资源里最核心的骨架import asyncio import json import websockets # 房间表房间名 - WebSocket 客户端集合 ROOMS {} async def relay(ws): # 从 URL path 最后一段取出房间号例如 ws://ip:8765/room_1 try: room ws.path.split(/)[-1] or default except AttributeError: room default # 把当前连接加进房间 ROOMS.setdefault(room, set()).add(ws) try: # 先发一条控制信息告诉客户端本服务的音频参数 await ws.send(json.dumps({ type: hello, rate: 16000, chunk: 640, channels: 1 })) # 循环接收客户端的帧 async for message in ws: if isinstance(message, bytes): # 只转发给同房间的其他人不转发给发送者自己 peers ROOMS.get(room, set()) - {ws} for peer in peers: await peer.send(message) else: # 文本消息按控制指令处理这里只做日志 print(control:, message) finally: # 连接断开时清理房间表 ROOMS.setdefault(room, set()).discard(ws) if not ROOMS.get(room): ROOMS.pop(room, None) async def main(): # host 写 0.0.0.0 表示监听本机所有网卡外网/局域网都可连 async with websockets.serve( relay, 0.0.0.0, 8765, ping_interval20, ping_timeout20, max_size2 ** 20 ): await asyncio.Future() # 让服务一直运行 if __name__ __main__: asyncio.run(main())这段代码的核心是ROOMS字典。键是房间号值是一个set。用set而不是list的好处是连接被断开后discard可以安全地移除元素即使重复调用也不会报错转发时用集合差集- {ws}把发送者排除掉避免对端听到自己的回声。ws.path.split(/)[-1]这一行是从访问路径里提取房间名。客户端连接ws://127.0.0.1:8765/room_a服务器端拿到的 path 是/room_a最后一个元素就是room_a。await peer.send(message)是异步发送如果某个对端网络慢这里可能阻塞当前房间的转发。这个资源里直接转发是够用的如果是几十人的会议场景需要改成每个连接一个发送队列不能在同一个 for 循环里逐个 await。asyncio.Future()是为了让主协程挂在那里而不是立刻退出。这样服务进程就能一直监听直到被外部信号杀掉。3.3 启动参数里值得预留的三个值ping_interval20和ping_timeout20是 WebSocket 库自带的保活探测。每隔 20 秒发一次 ping对端必须在 20 秒内回 pong否则认为连接死掉。语音连接长时间没有音频流时这个参数能及时发现半死的连接。如果网络环境很差可以把ping_interval调到 10缺点是会多一些探测包。max_size2 ** 20是控制单帧最大字节数。默认值 1MB对于语音帧来说完全够用。但如果你直接用这份代码去传大文件就会撞上这个限制表现为连接发送几次后突然断开。注意websockets库版本不同ws.path的写法略有差异。新版推荐ws.request.path老版本用ws.path。下载资源里如果跑出AttributeError优先检查这一行。4. 客户端实现录音、连接、收发、播放的完整闭环客户端的任务是把麦克风数据读进来、推送到 WebSocket、从 WebSocket 收对端数据、写到声卡播放。这个闭环里最忌讳“串行等待”如果先录完一整段再发就变成对讲机不是实时语音。4.1 录音回调与缓冲队列这里我采用sounddevice的InputStream回调模式。回调函数在音频线程里执行主线程不阻塞录音数据放到队列后立刻返回。发送协程从队列取数据再异步发到 WebSocket。import asyncio import json import queue import sounddevice as sd import websockets # 和服务器约定16k 采样、单声道、640B 一帧 RATE 16000 CHUNK 640 audio_queue queue.Queue(maxsize20) def on_audio(indata, frames, time, status): try: # indata 是 numpy 数组copy 出来再入队避免被回调复用 audio_queue.put_nowait(indata.copy()) except queue.Full: # 队列满就直接丢防止延迟越积越严重 pass参数说明CHUNK 640对应 20ms 音频audio_queue容量 20 帧。put_nowait表示如果队列满立刻抛出queue.Full这里捕获后静默丢弃。为什么是“丢弃”而不是“等待”因为回调是音频硬件线程一旦阻塞声音采集就会出现毛刺体验远差于丢掉一个 20ms 帧。4.2 发送循环、接收循环和主入口发送循环负责从队列取 PCM 原始字节调用ws.send推给服务器。接收循环用await ws.recv()等待服务器转发的音频并把字节流写入OutputStream播放。async def sender(ws): # 打开麦克风输入流 with sd.InputStream( samplerateRATE, blocksizeCHUNK, channels1, dtypeint16, callbackon_audio ): while True: buf audio_queue.get() # 转成 bytes 后通过 WebSocket 二进制帧发送 await ws.send(buf.tobytes()) async def receiver(ws, output): while True: raw await ws.recv() # 先尝试解析为 JSON 控制信息失败则视为音频帧 try: msg json.loads(raw) if msg.get(type) hello: print(server echo:, msg) continue except json.JSONDecodeError: pass # 写入声卡播放 output.write(raw) async def main(): uri ws://127.0.0.1:8765/room_1 async with websockets.connect(uri) as ws: with sd.OutputStream( samplerateRATE, blocksizeCHUNK, channels1, dtypeint16 ) as out: await asyncio.gather( sender(ws), receiver(ws, out) ) if __name__ __main__: asyncio.run(main())传输通路的设计要点sender和receiver是两个并发的异步任务由asyncio.gather同时运行。TCP 收和发的独立性在这里体现得很明显——对方说话时自己的麦克风采集不能停下来反过来也一样。output.write(raw)默认是阻塞写入如果声卡暂时忙它会等待。对语音播放来说等待会导致 Micro 延迟但至少不会把内存堆爆。如果你有更严格的低延迟需求可以用OutputStream的latencylow参数牺牲一点稳定性换取速度。4.3 两个很容易改错的参数位置第一sender里的InputStream和receiver里的OutputStream必须使用同一套采样率和通道数。很多人只改RATE不改CHUNK服务器那边按 640B 一帧处理客户端却按 960B 一帧切分听到的声音会一半正常一半断。第二CHUNK要等于服务器端hello信息里的chunk。如果你从资源里看到的服务器写的是48000客户端就不要强行用16000。跑起来之前先加一行日志把服务器下发的rate和chunk打出来再做对齐。5. 常见问题排查websocket 语音项目最容易翻车的五个点每次有人把这份代码拿去改最后跑不通的问题都很集中。下面五条是按实际出现频率排的每一条都按“现象、原因、解决”拆开写。5.1 现象握手没报错但发几帧后连接被断开原因websockets库默认max_size限制。如果客户端把整段语音包成一个超大二进制帧发出去超过服务器端max_size连接会被视为协议错误直接关闭。解决语音包必须切成 20ms 的小帧不要攒 1 秒发一次同时服务器端把max_size调大也没坏处我用2 ** 20就是给正常帧留足余量。客户端连接时也可以带max_size2 ** 20保持两端一致。5.2 现象服务器只收到文本字符串音频全部被忽略原因客户端发送前把音频bytes做了str(audio)或base64.b64encode在服务器isinstance(message, bytes)时被过滤成控制消息。虽然浏览器里有人习惯用JSON.stringify传音频但 Python 原生 WebSocket 客户端直接传bytes最省事。解决发送前确认buf是bytes可以打印type(buf)自查。如果确实要兼容 Web 端就要定一个协议头用 JSON 标记 audio 段服务器再根据标记解析 base64 内容不要在同一个循环里混淆。5.3 现象多人进同一个房间只有第一个能听到声音原因服务器端ROOMS[room]直接存了同一个房间里所有连接但转发时没有排除发送者或者房间号从path里取错了。常见的是客户端连ws://ip:8765/room_1/结尾多了一个斜杠split(/)[-1]拿到的是空字符串导致两个人被分到不同房间。解决服务器端先做一次规范化比如room ws.path.rstrip(/).split(/)[-1]客户端统一不带尾斜杠写 URI。转发逻辑里保留peers ROOMS[room] - {ws}确保不回声。5.4 现象声音说话一变、音调变高、像快进原因接收端OutputStream的块大小和实际到来数据长度不匹配。播放器按固定blocksize等数据而网络传过来的帧往往有缓冲区叠加比如对方发了 30 帧播放端 buffer 一次写 20 帧剩下的 10 帧下个周期被“翻倍采样”听感就变成变调。解决OutputStream的blocksize保持一致还不够更稳的做法是播放端不关心帧边界把收到的字节流连续写入这个流对象由音频设备自己按采样率消费。代码里output.write(raw)就是连续写入不要手动切块。5.5 现象局域网正常放到公网后 10 秒必断原因连接建立后一端卡住超过ping_timeout对端判定失效。公网环境下NAT 映射表可能比 WebSocket ping 周期更短电信级 NAT 超过 30 秒没有流量就回收映射语音又是高频率小包正常节奏下不会触发这个坑但一旦 20ms 帧里有抖动暂停连接就可能被回收。解决把ping_interval从 20 降到 10给 NAT 表持续刷新同时不要写“无语音就断开”的逻辑。还有人建议在客户端加一层静音帧保活我放在最后一章展开。6. 继续往深做三个能让代码真正落地的细节6.1 静音帧保活检测到没有声音时不要完全停发数据帧。可以发一个全零的 640B 帧当作占位符服务端照常转发对端播放时音量几乎为零。这样网络路径上始终有流量在走比 ping 更贴近真实音频通道。我一般用 RMS 音量判断静音def is_silence(frame_bytes): # int16 的 PCM算一个粗略峰值 samples int.from_bytes(frame_bytes, little, signedTrue) return abs(samples) 80这里只取第一个采样做快速判断聊胜于无。如果你要更严谨可以逐样本算峰值但实时性会受影响得不偿失。6.2 用 OPUS 压缩进一步降带宽Chunk不变只是发送前把 PCM 压缩成 OPUS 包接收端解压再播放。OPUS 在 16kHz 下码率可以压到 24-32kbps跟 256kbps 差了近十倍。缺点是引入编码延迟和 CPU 开销。如果你的应用跑在公网且带宽有限这个优化优先级很高如果只在局域网内保持裸 PCM 反而更简单。6.3 先录音落盘再验证拿到这份资源后我建议第一次跑通时不要直接对着麦克风说话先发一段固定的正弦波或者读一个 WAV 文件。对端保存接收数据成.raw文件再做一次 PCM 转 WAV 的检查。转好的文件能播放说明服务器转发和网络链路都没问题不能播放问题一定出在数据格式上。这能省掉大量“两个人互相问听不听得见”的调试时间。从那以后我每次做 websocket 语音项目都会强制自己先走一遍“发文件、收文件、落盘校验”的流程不通过就不上真麦。这个习惯帮我挡了不少采样率不一致的翻车事故。希望这能帮到你。本文还有配套的精品资源点击获取