简介这是一份基于WebSocket实现语音实时收发功能的Python代码资源面向需要构建长连接双向通信、语音识别或语音对讲场景的开发者。资源包体积紧凑仅9KB共3个文件包括2个Python脚本与1个备份文件服务器端采用Flask-SocketIO框架处理语音事件客户端基于websocket-client库连接服务并发送音频数据备份文件便于对照修改。已有3108人学习适合网络编程初学者参考。代码虽然精简但覆盖了服务器事件绑定、客户端消息收发、JSON格式语音数据封装等关键环节可在实际项目中快速替换逻辑模块或加入Kaldi等语音识别服务帮助读者理解WebSocket全双工通信在语音场景下的应用方式。 前一阵子做语音对讲原型老板只给了一句话“服务器和客户端之间实时语音互传你先用Python跑通一版。”做语音这块的人都知道最麻烦的不是怎么采集麦克风而是怎么把音频流稳定送到对端。我先后试了HTTP轮询、原始Socket、WebSocket三条路最终选定了WebSocket方案。Python生态里websockets库已经很成熟服务器端和客户端全用Python写代码量能压得非常小还能在Linux服务器上直接跑。这篇把服务器、客户端两端实现代码、音频参数选择、以及联调过程中踩过的各种坑都整理出来给想用WebSocket做语音收发的同行一个能直接抄作业的底子。这套方案完整跑通之后适合的场景很清晰语音对讲设备的私有后端、语音助手的双向音频通道、以及想用浏览器或Python客户端把实时音频送到云端的原型验证。全程不需要碰SIP、RTP那套重型协议栈只要有一台机器、一个Python环境就能把一条语音链路搭起来。1. 为什么是WebSocket而不是RTP或普通HTTP1.1 语音传输的几个硬指标语音实时传输和发文件完全是两码事。文件传输关心完整性延迟高一点没关系语音链路关心的是持续性和低延迟甚至在网络抖动时丢几个包也能接受但连接绝对不能断。具体拆开看有三个硬指标。第一是双向同时传输说话的同时要能听到对方这在协议上叫全双工。第二是低延迟端到端延迟最好控制在几百毫秒以内超过1秒基本就属于“电台式对讲”没法正常对话。第三是流式持续连接要长期保持不能像HTTP那样每次请求都重新建立连接。拿这三个指标去卡方案HTTP就率先被淘汰了。HTTP是请求-响应模型客户端问一次、服务器答一次服务端想主动说话还得等客户端来轮询这个轮询间隔就是天然的延迟上限。就算现在有Server-Sent Events能做单向推送它也只是服务器到客户端单向客户端到服务器还得靠单独的请求做不到真正的全双工。1.2 WebSocket恰好把协议栈简化到够用相比之下WebSocket的优势在于它把一个双工TCP连接包装成了浏览器和Python都能直接消费的消息通道。连接建立时走一次HTTP Upgrade握手之后就是一个长期存在的双向通道客户端可以向服务器发消息服务器也能主动向客户端推消息。语音在WebSocket里基本以二进制帧传输每个帧头部只有几个字节比HTTP那动辄几百字节的头部低得多。14字节的二进制帧头相比于HTTP头的开销优势非常明显虽然TCP本身还有开销但在应用层协议里已经算非常省的了。还有一个实际的好处WebSocket走TCP消息天然有序且不丢包。音频流最怕乱序乱序会导致声音变形WebSocket直接绕过了这个问题。如果自己拿UDP写一套还得自己搞序列号、缓冲队列、乱序重排工作量瞬间上去。1.3 什么时候不应该选WebSocket当然WebSocket不是银弹。如果你的场景是大规模双向实时通话、要求极低延迟并且能容忍丢包那WebSocket的TCP底层机制会让它在弱网下表现不好。TCP丢包重传会导致后续所有音频数据都排队等重传这在实时语音里是致命的。专业VoIP系统通常还是走UDPRTP配合前向纠错和抖动缓冲。但当场景限定在“私有客户端之间”或者“浏览器与服务器之间”的应用层实时语音链路WebSocket是我的首选。它不需要处理NAT穿透不需要处理RTP的时序同步不需要信令服务器一条连接解决所有问题。协议栈简化到够用这才是原型阶段最该看重的事。2. 音频参数与编码选择先别急着上Opus2.1 从PCM裸流起步的三个理由音频编码方案是开发初期最大的一个分叉口。一查资料会发现大家都在说Opus压缩率高、音质好是VoIP事实标准。但我实际测试后发现原型阶段真没必要一上来就上Opus。我用的是PCM裸流也就是模拟信号直接数字化不做任何压缩。三个理由零依赖PCM不需要安装opuslib、libopus这些编码库Python标准库的audioop加上numpy就能处理。不用考虑Windows下编译失败、Linux下装OpenSSL依赖这些破事。零压缩延迟Opus编码自身有算法延迟大约2.5~60msPCM直接就是原始采样发出去什么就是什么排查问题少一层变量。调试直观PCM的二进制流可以直接用十六进制编辑工具打开看波形哪里不对一目了然。如果是压缩流出一个音频模糊的问题你根本分不清是传输问题还是编码参数不对。当链路完全跑通、延迟表现确认之后再考虑加Opus压缩也不迟。优化编码应该放在链路稳定之后而不是链路没通之前。2.2 采样率、位深、声道数怎么定语音场景下我推荐一套保守组合16kHz采样率、16bit位深、单声道。为什么是16kHz语音的主要能量集中在300Hz到3400Hz根据奈奎斯特定理采样率至少要是最高频率的两倍8kHz理论够用但会明显损失音质细节。16kHz留出足够冗余也是语音识别和语音唤醒领域非常通用的采样率后续如果要接ASR引擎基本无缝。位深选16bit因为这是大多数麦克风硬件和音频库的默认位深转换成本最低。声道数选单声道节省一半带宽。算一笔带宽账16000Hz × 16bit × 1声道 256000bit/s也就是256kbps。在Wi-Fi和4G网络环境下这个带宽完全能接受。通话质量比电话好又不会像48kHz立体声那样要烧接近1.5Mbps带宽。如果是在弱网环境可以考虑把采样率降到8kHz带宽直接减半到128kbps但音质会像老式电话一样。具体取舍看网络条件。2.3 帧长100ms的取舍音频从麦克风出来是一串连续采样点但网络传输必须按块切分切分的大小就是帧长。我最终用的帧长是100ms也就是1600个采样点。帧长和延迟是直接相关的。帧长越小端到端延迟越低但单位时间的网络包数量变多。比如20ms帧长每秒要发50个UDP或WebSocket包100ms帧长每秒只有10个包。在TCP/WebSocket场景下包越多协议栈处理开销越大尤其在服务器并发连接数高的时候小包洪流会让CPU负载明显上升。20ms帧长在专业VoIP里是主流因为它适合实时通话。但在我这个WebSocket原型场景里我优先保证稳定性和低CPU开销100ms是综合测试下来最舒服的值。如果你的实际场景对延迟特别敏感可以从60ms或80ms往下调代码里把FRAME_MS常量改掉就行其余逻辑不用动。这里的取舍说白了就是原型阶段先要稳定延迟后续再优化。3. 服务器端实现异步接收、转发与连接管理3.1 用websockets库搭出来的服务器骨架服务器端我用的是Python的websockets库它基于asyncio代码风格干净没有回调地狱。安装只需要一行pip install websockets sounddevice numpy服务器端整个骨架就一个异步函数注册到websockets.serve里然后进入一个无限循环。核心代码我贴出来import asyncio import json import websockets class AudioServer: def __init__(self): self.clients {} # websocket对象 - 客户端注册信息 async def handle(self, websocket): # 客户端连上来之后先发一条注册消息 try: reg await asyncio.wait_for(websocket.recv(), timeout5) info json.loads(reg) self.clients[websocket] info print(f新客户端接入: {info}) except Exception: await websocket.close(code4000, reasonregister timeout) return try: async for message in websocket: if isinstance(message, bytes): await self.forward(websocket, message) else: # 处理控制消息比如心跳、模式切换 print(收到控制消息:, message) except websockets.ConnectionClosed: pass finally: self.clients.pop(websocket, None) print(客户端离开当前连接数:, len(self.clients)) async def forward(self, source, audio_bytes): for ws in list(self.clients.keys()): if ws is source: continue try: await ws.send(audio_bytes) except websockets.ConnectionClosed: continue async def main(): server AudioServer() async with websockets.serve(server.handle, 0.0.0.0, 8765): print(WebSocket语音服务器已启动端口 8765) await asyncio.Future() if __name__ __main__: asyncio.run(main())3.2 注册控制与二进制转发这里有一个很重要的设计控制消息和语音消息混用同一条连接但用类型区分。客户端连上来之后第一件事不是发语音而是发一条JSON文本消息完成注册里面携带客户端名称、角色或房间信息。服务器把它存进self.clients字典。之后所有WebSocket消息里bytes类型统一当作音频帧处理str类型统一当作控制消息处理。这样避免了再开一个控制端口逻辑也清晰。async for message in websocket这个写法底层会自动处理消息的拆包和粘包每个message都是一条完整的WebSocket消息。对于二进制语音帧来说由于WebSocket本身有消息边界收到的一定是完整的一帧PCM数据不需要自己解决流式TCP的断包问题。这是我选WebSocket而不是裸TCP的重要原因省掉了自己设计帧格式和缓冲区的活。转发逻辑里有一个细节我在self.clients.keys()外面包了一层list()。这是因为在转发过程中如果某个客户端恰好断连并触发finally清理字典会在遍历过程中被修改直接遍历会抛RuntimeError。包成list()就是做了一次快照等下次循环再检测连接是否还在。3.3 服务器心跳与异常清理websockets库默认启用了ping_interval20秒也就是每20秒向客户端发一个ping帧。这不算缺点但要注意两个情况。一是如果客户端是浏览器浏览器WebSocket协议对ping/pong的支持是透明的客户端会自动回pong不用额外处理。二是如果客户端代码里长时间阻塞在某个同步操作上没有及时让出事件循环导致没法回pong服务器会认为连接死了直接关闭。这就是后面要讲的“客户端被莫名断开”的坑之一。实际部署时我建议主动加一层业务心跳。客户端每5秒发一条{type: heartbeat}文本消息服务器收到就更新心跳时间。这样能区分“TCP还在但业务死了”和“真的断网了”后续做断线重连时判断更准确。4. 客户端实现麦克风采集、发送与播放4.1 sounddevice的采集/播放回调怎么和asyncio协作客户端我用了sounddevice库它依赖PortAudio能统一处理Windows、Linux、macOS的音频设备。核心机制是回调函数音频硬件在每个音频块准备好时会调用你注册的回调函数。麻烦在于sounddevice的回调运行在PortAudio自己的后台线程里而websockets的全双工通信跑在asyncio事件循环里。线程之间不能直接共享变量否则会有竞态条件。我的解法是用queue.Queue作为中转站采集回调往队列里放数据发送协程从队列里取数据。queue本身就是线程安全的不需要额外加锁。audio_in queue.Queue(maxsize100) def mic_callback(indata, frames, time_info, status): try: audio_in.put(indata.copy()) except queue.Full: pass注意到我用到了indata.copy()。sounddevice回调里的indata是缓冲区复用对象下次回调会覆盖这块内存。如果直接把indata放进队列发送线程读到的可能是被覆盖后的新数据。这个bug很难查因为表现为“声音偶尔正常、偶尔断续”跟网络卡顿一模一样。4.2 发送和接收两条并行的协程客户端整体结构是两条协程并行跑一条负责从采集队列取数据并发送一条负责接收音频并写入播放队列。import asyncio import json import queue import numpy as np import sounddevice as sd import websockets SAMPLE_RATE 16000 CHANNELS 1 DTYPE int16 FRAME_MS 100 FRAME_SIZE int(SAMPLE_RATE * FRAME_MS / 1000) # 1600 URL ws://127.0.0.1:8765 audio_in queue.Queue(maxsize100) audio_out queue.Queue(maxsize200) def mic_callback(indata, frames, time_info, status): try: audio_in.put(indata.copy()) except queue.Full: pass def play_callback(outdata, frames, time_info, status): try: block audio_out.get_nowait() outdata[:] block except queue.Empty: outdata.fill(0) async def send_loop(ws): while True: block await asyncio.get_event_loop().run_in_executor(None, audio_in.get) await ws.send(block.tobytes()) async def recv_loop(ws): while True: message await ws.recv() if isinstance(message, bytes): block np.frombuffer(message, dtypeDTYPE).reshape(-1, CHANNELS) try: audio_out.put_nowait(block) except queue.Full: pass async def main(): with sd.InputStream(samplerateSAMPLE_RATE, channelsCHANNELS, dtypeDTYPE, callbackmic_callback): with sd.OutputStream(samplerateSAMPLE_RATE, channelsCHANNELS, dtypeDTYPE, callbackplay_callback): async with websockets.connect(URL) as ws: await ws.send(json.dumps({type: register, name: client-1})) await asyncio.gather(send_loop(ws), recv_loop(ws)) if __name__ __main__: asyncio.run(main())发送协程里有一个关键细节audio_in.get()是阻塞调用会卡住当前线程。我把它包进了await asyncio.get_event_loop().run_in_executor(None, audio_in.get)让它在线程池里等待而不是阻塞事件循环。这样recv_loop能继续处理到达的音频消息两个方向互不干扰。4.3 为什么用numpy转码而不是struct接收端收到二进制消息后我用np.frombuffer一次性把字节数组转成numpy数组而不是用struct.unpack逐样本解包原因是性能差异巨大。每帧1600个int16采样点struct.unpack要循环1600次解包在树莓派这种低性能设备上会产生明显CPU占用。np.frombuffer底层是零拷贝视图不复制数据直接把字节缓冲区解释为int16数组然后reshape(-1, CHANNELS)变成单声道二维数组可以直接喂给sounddevice播放。发送端反向用block.tobytes()native memory order下也是零拷贝转换几乎没有额外开销。整个链路里音频数据从麦克风到网络几乎不做复制这一点在持续发送语音时至关重要。同样要注意的还有队列容量。采集队列maxsize100是按100ms一帧算的相当于10秒的缓冲。播放队列maxsize200可以容纳20秒。设置上限是为了防止网络不通时内存无限膨胀。如果发现音频延迟越来越大十有八九是网络发送慢于采集速度队列里堆积了太多待发送数据。5. 联调实录从断断续续到稳定跑通5.1 客户端被服务器断开的真凶第一次联调时客户端运行几秒钟后突然抛异常stream disconnected before completion: websocket closed by server before res连接被服务器主动关闭。排查过程很典型。第一步在服务器端加打印日志发现异常是websockets库默认的ping_interval20秒机制导致的。服务器每20秒发一个ping帧如果客户端在3个间隔内没有回pong服务器判定连接不可用主动断开。但问题来了我的客户端代码明明一直在跑事件循环为什么没回pong后面才发现问题出在声音设备初始化上。sd.OutputStream在极端情况下会把事件循环阻塞一小段时间导致pong响应被延迟超过服务器容忍度就被判定失联了。知道了根因解法就顺了服务器端把ping_interval调大到60秒客户端同时把sounddevice的音频块大小调小避免长时间占用主线程导致事件循环饿死。改完之后这个异常彻底消失。5.2 语音卡顿根因排查第二个坑是语音断断续续有点像电台信号差。我排查了很久才发现问题不在网络而在帧长不匹配。我的采集线程按100ms分帧但sounddevice底层设备的默认音频块大小在大多数平台上也是100ms上下看起来没问题。可一旦系统负载升高PortAudio回调频率会波动回调函数拿到的frames数量偶尔会变成两倍或一半。解决方法是回调函数不直接依赖frames参数而是每次都把indata整个复制并入队。队列里的数据块大小可能不完全一致这没关系因为每个块都是一个完整的WebSocket消息帧网络传输和播放端都以消息为单位消费。牺牲一点点字节对齐换来的是对采集波动的完全免疫。5.3 如果浏览器是客户端还有一个坑如果你的服务器将来要接浏览器前端注意浏览器WebSocket的默认二进制类型是Blob不是ArrayBuffer。如果直接拿Blob丢给Web Audio API解码会发现数据格式不对必须设置一行socket.binaryType arraybuffer;设置之后服务端发来的二进制消息才能在浏览器里通过Int16Array直接解析。这一点做Python客户端时不用关心因为Python端天然处理字节流但前后端混合调试时一定要记得。5.4 实测延时与调优记录局域网环境下一整套收发链路端到端延迟大约是130ms到180ms其中包括100ms的帧积累时间、10ms左右的网络传输、以及20ms左右的播放缓冲。这个延迟水平对于语音对讲完全够用正常对话不会有明显“慢半拍”的感觉。如果你需要更低延迟可以把FRAME_MS从100调到60或40帧积累时间降下来端到端延迟可以压到100ms内。但代价是每秒的WebSocket消息数量从10条增加到25条左右服务器并发能力会下降。实测在普通台式机上100ms帧长可以支撑约300个并发连接60ms帧长只剩200个不到。没有绝对好坏看你更在意延迟还是容量。6. 这套代码还能怎么扩展6.1 加入VAD判断省流量现在的代码是麦克风一有数据就发不管环境有没有人说话。这在真实场景里很浪费意味着背景噪音也会占据带宽和服务器开销。要做的话最佳切入点是发送协程里。在audio_in.get()拿到数据之后先做一个简单的短时能量检测计算这帧数据的均方根值低于阈值就丢弃不发送。阈值可以先根据环境噪音标定后续再接一个更完善的VAD模型。更细的做法是加一个“说话后保持”机制检测到超过阈值后继续发送后续300ms音频避免语音尾部被切掉。这个300ms参数来自实际经验VAD的灵敏度调太高会把尾音切断调太低会漏掉轻音开头的词。6.2 接入实时语音识别这套WebSocket链路天然适合接云端语音识别。服务器收到音频帧后不需要转发给其他客户端而是直接拼接后推给ASR引擎。识别文本再通过同一个WebSocket连接推回到客户端。我实测过把16kHz的PCM直接喂给常见ASR引擎识别率比8kHz高很多。这也印证了之前选16kHz采样率的正确性。服务器端只要把forward函数改成调用ASR接口然后往来源客户端发回文本消息就行协议和控制框架都不用动。6.3 多人语音房间的路由改造现在服务器是“一对多广播”一个客户端说话其他所有客户端都能听到。如果要实现多人房间比如房间A的人只发给房间A的人需要在注册消息里增加room_id字段服务器在forward时按房间过滤。房间过滤的实现不复杂在self.clients里给每个客户端存一个room_id转发时只挑相同room_id的客户端发送。但要注意多人同时说话时每个客户端都会收到多路音频叠加播放端要做混音处理把两路int16数组相加并做防溢出截断。这一步能在服务器端做也能在客户端做取决于你的服务器CPU余量。我这套原型目前还在用单房间广播模式后续如果真上多人会议混音算法会是下一个要啃的硬骨头。最后再分享一点实际体会整套方案从写代码到跑通差不多用了一个下午但把所有坑都摸清、把参数都调合适花了将近两天。音频这东西就是这样链路本身不难难的是音频参数之间互相耦合——采样率影响带宽帧长影响延迟队列深度影响稳定性调任何一个都要重新测一遍整体表现。建议你拿到代码之后先在局域网里跑通再逐步调参不要一上来就追求极限延迟。等你的局域网版本稳定了再考虑上Opus压缩、加VAD、处理弱网每一步都基于上一版的稳定输出排错效率会高很多。本文还有配套的精品资源点击获取