尧图网页设计 
FastRTC 接入指南通过 WebRTC 与 WebSocket API 对接实时音视频流【免费下载链接】fastrtcThe python library for real-time communication项目地址: https://gitcode.com/GitHub_Trending/fa/fastrtcFastRTC 是一个面向实时通信的 Python 库其核心Stream对象既可以生成开箱即用的 Gradio 界面也可以通过 mount() 挂载到 FastAPI 应用上对外暴露标准的 WebRTC 与 WebSocket 信令接口。本文是 docs/userguide/api.md 的完整展开面向需要自建前端、绕过 Gradio UI 直接对接服务的开发者读完你可以掌握如何根据modality音频/视频/音视频与mode发送/接收/双向组合选择连接方式、如何在浏览器端用原生 JavaScript 完成 WebRTC 建连与 WebSocket 音频收发、如何理解并处理服务端下发的各类控制消息以及如何通过set_input/output_stream打通自定义前后端的附加输入输出通道。连接前先确定 Connection、Modality 与 Mode在动手写客户端代码之前必须先明确三个维度它们共同决定了可用的协议与代码形态维度可选值说明ConnectionWebRTC/WebSocket传输协议。WebRTC 支持全部 modalityWebSocket 目前仅支持音频、且仅支持send-receive模式从 websocket.py 的WebSocketHandler实现可以看出其消息协议只处理音频media帧Modalityaudio/video/audio-video媒体类型对应Stream(modality...)参数见 stream.pyModesend-receive/send/receive数据方向。send-receive为默认的双向流send仅客户端上行receive仅服务端下行三个维度组合后在服务端对应不同的 handler 形态详见 streams.md 中的 Handler 表格视频各模式使用普通函数音频双向流使用StreamHandler/AsyncStreamHandler子类音视频双向流使用AudioVideoStreamHandler系列。下文分别给出 WebRTC 与 WebSocket 两套客户端接入代码。通过 WebRTC 接入RTCPeerConnection 建连全流程WebRTC 是 FastRTC 最完整的接入方式。核心思路是客户端创建RTCPeerConnection通过POST /webrtc/offer与服务器交换 SDP并通过同一个端点交换 ICE candidate。该路由由 Stream.mount() 注册请求体结构由 stream.py 中的 Body 模型 定义sdp、candidate、type与webrtc_id四个字段。以最常见的音频send-receive双向模式为例完整的浏览器端代码如下其他模式的差异点见后文// 可按需传入 rtc_configuration 参数 const pc new RTCPeerConnection(); // send-receive / receive 模式需要输出组件 // audio 模式对应 audio 元素video 模式对应 video 元素 const audio_output_component document.getElementById(audio_output_component_id); async function setupWebRTC(peerConnection) { // 1. 采集本机音频send-receive / send 模式 const stream await navigator.mediaDevices.getUserMedia({ audio: true, }); // 2. 将本地音轨发送到服务端send-receive 模式 stream.getTracks().forEach(async (track) { const sender pc.addTrack(track, stream); }); // 3. 接收服务端回传的音频 peerConnection.addEventListener(track, (evt) { if (audio_output_component audio_output_component.srcObject ! evt.streams[0]) { audio_output_component.srcObject evt.streams[0]; } }); // 4. 创建 DataChannel必需控制消息都走这里 const dataChannel peerConnection.createDataChannel(text); // 5. 创建并发送 SDP offer const offer await peerConnection.createOffer(); await peerConnection.setLocalDescription(offer); let webrtc_id Math.random().toString(36).substring(7); // 6. 将 ICE candidate 发往服务端 // 服务器在防火墙之后时尤其需要 peerConnection.onicecandidate ({ candidate }) { if (candidate) { console.debug(Sending ICE candidate, candidate); fetch(/webrtc/offer, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ candidate: candidate.toJSON(), webrtc_id: webrtc_id, type: ice-candidate, }) }) } }; // 7. 发送 offer 并设置远端描述 const response await fetch(/webrtc/offer, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ sdp: offer.sdp, type: offer.type, webrtc_id: webrtc_id }) }); // 8. 处理服务端返回的 SDP answer const serverResponse await response.json(); await peerConnection.setRemoteDescription(serverResponse); }各模式的关键差异点send-receive双向既addTrack上行音轨又监听track事件渲染下行媒体HTML 中需要一个audio音频或video视频输出元素其 id 形如audio_output_component_id/video_output_component_id。send仅发送只addTrack不需要输出元素。receive仅接收不需要getUserMedia改为pc.addTransceiver(audio, { direction: recvonly })声明只收不发并监听track事件。webrtc_id由客户端自行生成示例用随机字符串后续set_input、output_stream等 API 都以它作为会话标识。服务端侧的原理印证/webrtc/offer最终进入 WebRTCConnectionMixin.handle_offer()它依次处理 ICE candidate、检查重复连接与并发上限、用 aiortc 创建RTCPeerConnection、注册track/datachannel/ 连接状态事件最后setRemoteDescription → createAnswer → setLocalDescription返回 SDP answer。注意其中webrtc_id同时被用作handlers、pcs、connections等字典的 key客户端必须保证其唯一性。通过 WebSocket 接入mu-law 音频流收发WebSocket 通道面向音频send-receive场景也是 Twilio 电话接入的底层协议。接入点是ws://host/websocket/offer消息采用 JSON 事件协议客户端发送start携带唯一 id、media携带 base64 编码的 mu-law 音频、stop、ping事件服务端回发media音频事件与pong。这些事件分支都可以在 WebSocketHandler.handle_websocket() 中找到对应实现。关键约束原文档明确说明且与 websocket.py 的 convert_to_mulaw 一致输入音频必须是 mu-lawµ-law编码采样率等于所连接 handler 的input_sample_rate输出音频同样是 mu-law 编码采样率等于 handler 的output_sample_rate关于默认采样率原文档指出默认按 48k Hz 处理从源码看StreamHandlerBase 中input_sample_rate默认确实为 48000而output_sample_rate默认为 24000——两者都可以在自定义 handler 的构造函数中覆盖因此客户端应以目标 handler 的实际配置为准。浏览器端完整示例// 1. 建立音频上下文并采集麦克风 const audioContext new AudioContext(); const stream await navigator.mediaDevices.getUserMedia({ audio: true }); // 2. 创建 WebSocket 连接自动适配 https/wss const ws new WebSocket(${window.location.protocol https: ? wss: : ws:}//${window.location.host}/websocket/offer); ws.onopen () { // 发送 start 消息携带唯一 websocket_id ws.send(JSON.stringify({ event: start, websocket_id: generateId() // 自行实现 ID 生成器 })); // 3. 采集 → mu-law 编码 → base64 → media 事件发送 const source audioContext.createMediaStreamSource(stream); const processor audioContext.createScriptProcessor(2048, 1, 1); source.connect(processor); processor.connect(audioContext.destination); processor.onaudioprocess (e) { const inputData e.inputBuffer.getChannelData(0); const mulawData convertToMulaw(inputData, audioContext.sampleRate); const base64Audio btoa(String.fromCharCode.apply(null, mulawData)); if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ event: media, media: { payload: base64Audio } })); } }; }; ws.onmessage (event) { const data JSON.parse(event.data); // 4. 服务端请求附加输入时可触发自定义输入钩子 if (data?.type send_input) { fetch(/input_hook, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ webrtc_id: wsId }) }); } // 5. 接收并解码服务端音频 if (data.event media) { const audioData atob(data.media.payload); const mulawData new Uint8Array(audioData.length); for (let i 0; i audioData.length; i) { mulawData[i] audioData.charCodeAt(i); } const linearData alawmulaw.mulaw.decode(mulawData); // mu-law → PCM const audioBuffer outputContext.createBuffer(1, linearData.length, sampleRate); const channelData audioBuffer.getChannelData(0); for (let i 0; i linearData.length; i) { channelData[i] linearData[i] / 32768.0; // int16 → float } // 将 audioBuffer 交给播放器或后续处理 } };原理补充服务端收到media事件后在 websocket.py 中做base64 解码 → ulaw2lin → numpy int16还原再以(sample_rate, audio_array)元组交给 handler 的receive发送方向则通过_emit_loop将 handler 产出的音频帧用convert_to_mulaw重新编码为 mu-law 后回传电话模式下还会以 0.75 倍实时时长节奏略快于实时发送见 websocket.py。理解服务端消息格式type 与 data无论走 WebRTC 还是 WebSocket服务端都会下发如下统一格式的控制消息{ type: send_input | fetch_output | stopword | error | warning | log, data: string | object }各类型语义如下type含义与 data 内容触发场景send_input要求客户端向服务器发送任意输入数据供 handler 使用见下文「Additional Inputs」handler 调用fetch_args时通过 DataChannel 下发fetch_output提示客户端有新的AdditionalOutputs可拉取每当 handler 产出AdditionalOutputs时下发见 utils.py 的 player_worker_decodestopword检测到停止词ReplyOnStopWords触发见 reply_on_stopwords.pyerror发生错误data为错误消息字符串服务端异常WebRTCError会通过通道发送warning警告data为警告字符串代码中调用Warning()时触发log日志消息data为日志字符串通用日志/运行状态上述 type 枚举与序列化逻辑集中定义在 utils.py 的 create_message()内部还包含end_stream、update_connection等额外类型。此外ReplyOnPause停顿检测handler 会额外下发以下三类log消息可用于驱动暂停→开始回复→用户开口的对话节奏 UI{ type: log, data: pause_detected | response_starting | started_talking }这三条日志分别对应 reply_on_pause.py、reply_on_pause.py#L403-L411 与 reply_on_pause.py#L234 处的create_message(log, ...)调用且触发条件与started_talking_threshold默认 0.2 秒、speech_threshold等 VAD 参数相关见 reply_on_pause.py 的 AlgoOptions。重要提示WebRTCWebRTC 场景下这些消息经 DataChannel 传输到达客户端时是字符串必须先用JSON.parse解析后再使用WebSocket 场景下消息本身就是 JSON 对象可直接访问data.type/data.event。Additional Inputs向 handler 注入动态参数当客户端收到send_input消息时需要把自定义参数回传给服务端。服务端侧通过Stream.set_input()方法更新 handler 的输入第一个参数是webrtc_id后面跟随自定义参数实现见 WebRTCConnectionMixin.set_input()它会把参数写入该会话所有连接的args。典型的 POST 输入钩子写法from pydantic import BaseModel, Field class InputData(BaseModel): webrtc_id: str conf_threshold: float Field(ge0, le1) # 0~1 范围内校验 app.post(/input_hook) async def _(data: InputData): stream.set_input(data.webrtc_id, data.conf_threshold)在 handler 启动时读取输入如果 handler 的初始化逻辑依赖这些动态参数可在start_up()中先调用wait_for_args()阻塞等待输入就绪再通过latest_args[1:]取参——下标 0 是内部元数据WebRTC 场景下为WebRTCData业务参数从下标 1 开始from fastrtc import AsyncStreamHandler class CustomHandler(AsyncStreamHandler): async def start_up(self) - None: await self.wait_for_args() conf_threshold self.latest_args[1] # 用 conf_threshold 完成初始化原理细节tracks.py 的 StreamHandlerBase 中wait_for_args()会先通过通道发送send_input消息fetch_args然后等待args_set事件set_args()写入latest_args并置位事件非 WebRTC 数据会自动在列表头补__webrtc_value__占位。因此本次更新会在 handler 的「下一次」调用中生效而不是立即打断当前正在处理的帧。Additional Outputs拉取 handler 的附加输出当 handler 返回AdditionalOutputs定义于 utils.py本质是args元组的轻量封装时服务端会下发fetch_output消息。此时客户端或你自己的后端代码有两种取数方式方式一单次拉取最新输出output await stream.fetch_latest_output(webrtc_id)该方法内部从该会话的异步队列取一条数据超时上限为 10 秒见 WebRTCConnectionMixin.fetch_latest_output()。方式二持续订阅输出流推荐与其逐条手动拉取更常见的模式是用output_stream()一次性拿到该会话的全部输出流配合 FastAPI 的StreamingResponse以 SSEServer-Sent Events对外暴露from fastapi.responses import StreamingResponse app.get(/updates) async def stream_updates(webrtc_id: str): async def output_stream(): async for output in stream.output_stream(webrtc_id): # output 是 AdditionalOutputs 实例 # 按需自行序列化 yield fdata: {output.args[0]}\n\n return StreamingResponse( output_stream(), media_typetext/event-stream )原理细节output_stream() 以 0.1 秒超时轮询asyncio.Queue连接断开时由clean_up()置位quit事件结束迭代handler 每次产出附加输出都会通过set_additional_outputs回调写入队列见 WebRTCConnectionMixin.set_additional_outputs()。由于 WebRTC 延迟极低、帧率很高不建议每一帧都返回附加输出应只在有意义的状态变化时返回以减轻队列与网络压力。处理错误并发上限与 200 状态码的约定当连接数达到上限时服务端会拒绝新连接。WebRTC 场景下POST /webrtc/offer会返回如下 JSONHTTP 状态码为200{ status: failed, meta: { error: concurrency_limit_reached, limit: 10 } }WebSocket 场景下服务端会在关闭连接前发送同样的 JSON 消息。两条必须注意的约定为什么用 200 而不是 4xx/5xx原文档明确指出服务器故意返回 200 状态码是因为否则 gradio client 将无法解析 JSON 响应并展示错误信息。因此客户端不能只依赖 HTTP 状态码判断成败必须检查响应体内的status字段。并发上限来自哪里limit的数值即Stream(concurrency_limit...)的配置。从 stream.py 的 Stream.init可以看到concurrency_limit传入default或None时实际映射为 1并发检查发生在 handle_offer()len(self.pcs) concurrency_limit以及 websocket_offer()。此外该错误枚举还出现在handle_offer的其他分支中例如connection_already_exists同一webrtc_id重复连接与unknown_connection未知 ICE candidate。错误响应中的error字符串可以直接用于前端分支处理例如在fetch(/webrtc/offer)之后const res await response.json(); if (res.status failed) { if (res.meta.error concurrency_limit_reached) { // 提示用户当前并发已满稍后重试 } } else { await peerConnection.setRemoteDescription(res); }常见接入问题速查现象排查方向WebSocket 连不上确认目标 Stream 是audiosend-receive模式WebSocket 不支持视频与单向模式原文档明确限制收到音频但播放异常/噪声确认上行音频是 mu-law 编码且采样率与 handler 的input_sample_rate一致默认 48k见 tracks.py确认输出按output_sample_rate解码send_input后参数不生效参数在 handler 的下一次调用才生效确认start_up中调用了wait_for_args()且读取latest_args[1:]WebRTC 建连超时检查webrtc_id是否唯一确认 ICE candidate 与 offer 都 POST 到了/webrtc/offer服务器在 NAT/防火墙后时必须转发 candidate收到fetch_output但取不到数据使用output_stream()持续消费而非只调一次fetch_latest_output()后者单次等待超时为 10 秒连接被拒查看返回 JSON 的meta.errorconcurrency_limit_reached表示并发满默认 1可用Stream(concurrency_limitN)调大见 streams.mdconnection_already_exists表示webrtc_id重复参考实现与进一步阅读本文对应的原始文档docs/userguide/api.mdStream 核心对象与mount注册的路由/webrtc/offer、/websocket/offer、/telephone/*backend/fastrtc/stream.pyWebRTC 信令与连接生命周期、set_input/output_stream/fetch_latest_output实现backend/fastrtc/webrtc_connection_mixin.pyWebSocket 音频帧编解码与事件循环backend/fastrtc/websocket.pycreate_message、AdditionalOutputs、CloseStream、错误包装等消息基础件backend/fastrtc/utils.pyStreamHandler 家族与wait_for_args/set_args/ 采样率默认值backend/fastrtc/tracks.py停止词消息触发逻辑backend/fastrtc/reply_on_stopwords.py停顿检测三类 log 消息backend/fastrtc/reply_on_pause.pyStream 核心概念、Additional Inputs/Outputs 与并发控制总览docs/userguide/streams.md一个可直接运行的端到端示例Stream.mount(app) FastAPIdemo/echo_audio/app.py【免费下载链接】fastrtcThe python library for real-time communication项目地址: https://gitcode.com/GitHub_Trending/fa/fastrtc创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 FastRTC 接入指南：通过 WebRTC 与 WebSocket API 对接实时音视频流
FastRTC 接入指南通过 WebRTC 与 WebSocket API 对接实时音视频流【免费下载链接】fastrtcThe python library for real-time communication项目地址: https://gitcode.com/GitHub_Trending/fa/fastrtcFastRTC 是一个面向实时通信的 Python 库其核心Stream对象既可以生成开箱即用的 Gradio 界面也可以通过 mount() 挂载到 FastAPI 应用上对外暴露标准的 WebRTC 与 WebSocket 信令接口。本文是 docs/userguide/api.md 的完整展开面向需要自建前端、绕过 Gradio UI 直接对接服务的开发者读完你可以掌握如何根据modality音频/视频/音视频与mode发送/接收/双向组合选择连接方式、如何在浏览器端用原生 JavaScript 完成 WebRTC 建连与 WebSocket 音频收发、如何理解并处理服务端下发的各类控制消息以及如何通过set_input/output_stream打通自定义前后端的附加输入输出通道。连接前先确定 Connection、Modality 与 Mode在动手写客户端代码之前必须先明确三个维度它们共同决定了可用的协议与代码形态维度可选值说明ConnectionWebRTC/WebSocket传输协议。WebRTC 支持全部 modalityWebSocket 目前仅支持音频、且仅支持send-receive模式从 websocket.py 的WebSocketHandler实现可以看出其消息协议只处理音频media帧Modalityaudio/video/audio-video媒体类型对应Stream(modality...)参数见 stream.pyModesend-receive/send/receive数据方向。send-receive为默认的双向流send仅客户端上行receive仅服务端下行三个维度组合后在服务端对应不同的 handler 形态详见 streams.md 中的 Handler 表格视频各模式使用普通函数音频双向流使用StreamHandler/AsyncStreamHandler子类音视频双向流使用AudioVideoStreamHandler系列。下文分别给出 WebRTC 与 WebSocket 两套客户端接入代码。通过 WebRTC 接入RTCPeerConnection 建连全流程WebRTC 是 FastRTC 最完整的接入方式。核心思路是客户端创建RTCPeerConnection通过POST /webrtc/offer与服务器交换 SDP并通过同一个端点交换 ICE candidate。该路由由 Stream.mount() 注册请求体结构由 stream.py 中的 Body 模型 定义sdp、candidate、type与webrtc_id四个字段。以最常见的音频send-receive双向模式为例完整的浏览器端代码如下其他模式的差异点见后文// 可按需传入 rtc_configuration 参数 const pc new RTCPeerConnection(); // send-receive / receive 模式需要输出组件 // audio 模式对应 audio 元素video 模式对应 video 元素 const audio_output_component document.getElementById(audio_output_component_id); async function setupWebRTC(peerConnection) { // 1. 采集本机音频send-receive / send 模式 const stream await navigator.mediaDevices.getUserMedia({ audio: true, }); // 2. 将本地音轨发送到服务端send-receive 模式 stream.getTracks().forEach(async (track) { const sender pc.addTrack(track, stream); }); // 3. 接收服务端回传的音频 peerConnection.addEventListener(track, (evt) { if (audio_output_component audio_output_component.srcObject ! evt.streams[0]) { audio_output_component.srcObject evt.streams[0]; } }); // 4. 创建 DataChannel必需控制消息都走这里 const dataChannel peerConnection.createDataChannel(text); // 5. 创建并发送 SDP offer const offer await peerConnection.createOffer(); await peerConnection.setLocalDescription(offer); let webrtc_id Math.random().toString(36).substring(7); // 6. 将 ICE candidate 发往服务端 // 服务器在防火墙之后时尤其需要 peerConnection.onicecandidate ({ candidate }) { if (candidate) { console.debug(Sending ICE candidate, candidate); fetch(/webrtc/offer, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ candidate: candidate.toJSON(), webrtc_id: webrtc_id, type: ice-candidate, }) }) } }; // 7. 发送 offer 并设置远端描述 const response await fetch(/webrtc/offer, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ sdp: offer.sdp, type: offer.type, webrtc_id: webrtc_id }) }); // 8. 处理服务端返回的 SDP answer const serverResponse await response.json(); await peerConnection.setRemoteDescription(serverResponse); }各模式的关键差异点send-receive双向既addTrack上行音轨又监听track事件渲染下行媒体HTML 中需要一个audio音频或video视频输出元素其 id 形如audio_output_component_id/video_output_component_id。send仅发送只addTrack不需要输出元素。receive仅接收不需要getUserMedia改为pc.addTransceiver(audio, { direction: recvonly })声明只收不发并监听track事件。webrtc_id由客户端自行生成示例用随机字符串后续set_input、output_stream等 API 都以它作为会话标识。服务端侧的原理印证/webrtc/offer最终进入 WebRTCConnectionMixin.handle_offer()它依次处理 ICE candidate、检查重复连接与并发上限、用 aiortc 创建RTCPeerConnection、注册track/datachannel/ 连接状态事件最后setRemoteDescription → createAnswer → setLocalDescription返回 SDP answer。注意其中webrtc_id同时被用作handlers、pcs、connections等字典的 key客户端必须保证其唯一性。通过 WebSocket 接入mu-law 音频流收发WebSocket 通道面向音频send-receive场景也是 Twilio 电话接入的底层协议。接入点是ws://host/websocket/offer消息采用 JSON 事件协议客户端发送start携带唯一 id、media携带 base64 编码的 mu-law 音频、stop、ping事件服务端回发media音频事件与pong。这些事件分支都可以在 WebSocketHandler.handle_websocket() 中找到对应实现。关键约束原文档明确说明且与 websocket.py 的 convert_to_mulaw 一致输入音频必须是 mu-lawµ-law编码采样率等于所连接 handler 的input_sample_rate输出音频同样是 mu-law 编码采样率等于 handler 的output_sample_rate关于默认采样率原文档指出默认按 48k Hz 处理从源码看StreamHandlerBase 中input_sample_rate默认确实为 48000而output_sample_rate默认为 24000——两者都可以在自定义 handler 的构造函数中覆盖因此客户端应以目标 handler 的实际配置为准。浏览器端完整示例// 1. 建立音频上下文并采集麦克风 const audioContext new AudioContext(); const stream await navigator.mediaDevices.getUserMedia({ audio: true }); // 2. 创建 WebSocket 连接自动适配 https/wss const ws new WebSocket(${window.location.protocol https: ? wss: : ws:}//${window.location.host}/websocket/offer); ws.onopen () { // 发送 start 消息携带唯一 websocket_id ws.send(JSON.stringify({ event: start, websocket_id: generateId() // 自行实现 ID 生成器 })); // 3. 采集 → mu-law 编码 → base64 → media 事件发送 const source audioContext.createMediaStreamSource(stream); const processor audioContext.createScriptProcessor(2048, 1, 1); source.connect(processor); processor.connect(audioContext.destination); processor.onaudioprocess (e) { const inputData e.inputBuffer.getChannelData(0); const mulawData convertToMulaw(inputData, audioContext.sampleRate); const base64Audio btoa(String.fromCharCode.apply(null, mulawData)); if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ event: media, media: { payload: base64Audio } })); } }; }; ws.onmessage (event) { const data JSON.parse(event.data); // 4. 服务端请求附加输入时可触发自定义输入钩子 if (data?.type send_input) { fetch(/input_hook, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ webrtc_id: wsId }) }); } // 5. 接收并解码服务端音频 if (data.event media) { const audioData atob(data.media.payload); const mulawData new Uint8Array(audioData.length); for (let i 0; i audioData.length; i) { mulawData[i] audioData.charCodeAt(i); } const linearData alawmulaw.mulaw.decode(mulawData); // mu-law → PCM const audioBuffer outputContext.createBuffer(1, linearData.length, sampleRate); const channelData audioBuffer.getChannelData(0); for (let i 0; i linearData.length; i) { channelData[i] linearData[i] / 32768.0; // int16 → float } // 将 audioBuffer 交给播放器或后续处理 } };原理补充服务端收到media事件后在 websocket.py 中做base64 解码 → ulaw2lin → numpy int16还原再以(sample_rate, audio_array)元组交给 handler 的receive发送方向则通过_emit_loop将 handler 产出的音频帧用convert_to_mulaw重新编码为 mu-law 后回传电话模式下还会以 0.75 倍实时时长节奏略快于实时发送见 websocket.py。理解服务端消息格式type 与 data无论走 WebRTC 还是 WebSocket服务端都会下发如下统一格式的控制消息{ type: send_input | fetch_output | stopword | error | warning | log, data: string | object }各类型语义如下type含义与 data 内容触发场景send_input要求客户端向服务器发送任意输入数据供 handler 使用见下文「Additional Inputs」handler 调用fetch_args时通过 DataChannel 下发fetch_output提示客户端有新的AdditionalOutputs可拉取每当 handler 产出AdditionalOutputs时下发见 utils.py 的 player_worker_decodestopword检测到停止词ReplyOnStopWords触发见 reply_on_stopwords.pyerror发生错误data为错误消息字符串服务端异常WebRTCError会通过通道发送warning警告data为警告字符串代码中调用Warning()时触发log日志消息data为日志字符串通用日志/运行状态上述 type 枚举与序列化逻辑集中定义在 utils.py 的 create_message()内部还包含end_stream、update_connection等额外类型。此外ReplyOnPause停顿检测handler 会额外下发以下三类log消息可用于驱动暂停→开始回复→用户开口的对话节奏 UI{ type: log, data: pause_detected | response_starting | started_talking }这三条日志分别对应 reply_on_pause.py、reply_on_pause.py#L403-L411 与 reply_on_pause.py#L234 处的create_message(log, ...)调用且触发条件与started_talking_threshold默认 0.2 秒、speech_threshold等 VAD 参数相关见 reply_on_pause.py 的 AlgoOptions。重要提示WebRTCWebRTC 场景下这些消息经 DataChannel 传输到达客户端时是字符串必须先用JSON.parse解析后再使用WebSocket 场景下消息本身就是 JSON 对象可直接访问data.type/data.event。Additional Inputs向 handler 注入动态参数当客户端收到send_input消息时需要把自定义参数回传给服务端。服务端侧通过Stream.set_input()方法更新 handler 的输入第一个参数是webrtc_id后面跟随自定义参数实现见 WebRTCConnectionMixin.set_input()它会把参数写入该会话所有连接的args。典型的 POST 输入钩子写法from pydantic import BaseModel, Field class InputData(BaseModel): webrtc_id: str conf_threshold: float Field(ge0, le1) # 0~1 范围内校验 app.post(/input_hook) async def _(data: InputData): stream.set_input(data.webrtc_id, data.conf_threshold)在 handler 启动时读取输入如果 handler 的初始化逻辑依赖这些动态参数可在start_up()中先调用wait_for_args()阻塞等待输入就绪再通过latest_args[1:]取参——下标 0 是内部元数据WebRTC 场景下为WebRTCData业务参数从下标 1 开始from fastrtc import AsyncStreamHandler class CustomHandler(AsyncStreamHandler): async def start_up(self) - None: await self.wait_for_args() conf_threshold self.latest_args[1] # 用 conf_threshold 完成初始化原理细节tracks.py 的 StreamHandlerBase 中wait_for_args()会先通过通道发送send_input消息fetch_args然后等待args_set事件set_args()写入latest_args并置位事件非 WebRTC 数据会自动在列表头补__webrtc_value__占位。因此本次更新会在 handler 的「下一次」调用中生效而不是立即打断当前正在处理的帧。Additional Outputs拉取 handler 的附加输出当 handler 返回AdditionalOutputs定义于 utils.py本质是args元组的轻量封装时服务端会下发fetch_output消息。此时客户端或你自己的后端代码有两种取数方式方式一单次拉取最新输出output await stream.fetch_latest_output(webrtc_id)该方法内部从该会话的异步队列取一条数据超时上限为 10 秒见 WebRTCConnectionMixin.fetch_latest_output()。方式二持续订阅输出流推荐与其逐条手动拉取更常见的模式是用output_stream()一次性拿到该会话的全部输出流配合 FastAPI 的StreamingResponse以 SSEServer-Sent Events对外暴露from fastapi.responses import StreamingResponse app.get(/updates) async def stream_updates(webrtc_id: str): async def output_stream(): async for output in stream.output_stream(webrtc_id): # output 是 AdditionalOutputs 实例 # 按需自行序列化 yield fdata: {output.args[0]}\n\n return StreamingResponse( output_stream(), media_typetext/event-stream )原理细节output_stream() 以 0.1 秒超时轮询asyncio.Queue连接断开时由clean_up()置位quit事件结束迭代handler 每次产出附加输出都会通过set_additional_outputs回调写入队列见 WebRTCConnectionMixin.set_additional_outputs()。由于 WebRTC 延迟极低、帧率很高不建议每一帧都返回附加输出应只在有意义的状态变化时返回以减轻队列与网络压力。处理错误并发上限与 200 状态码的约定当连接数达到上限时服务端会拒绝新连接。WebRTC 场景下POST /webrtc/offer会返回如下 JSONHTTP 状态码为200{ status: failed, meta: { error: concurrency_limit_reached, limit: 10 } }WebSocket 场景下服务端会在关闭连接前发送同样的 JSON 消息。两条必须注意的约定为什么用 200 而不是 4xx/5xx原文档明确指出服务器故意返回 200 状态码是因为否则 gradio client 将无法解析 JSON 响应并展示错误信息。因此客户端不能只依赖 HTTP 状态码判断成败必须检查响应体内的status字段。并发上限来自哪里limit的数值即Stream(concurrency_limit...)的配置。从 stream.py 的 Stream.init可以看到concurrency_limit传入default或None时实际映射为 1并发检查发生在 handle_offer()len(self.pcs) concurrency_limit以及 websocket_offer()。此外该错误枚举还出现在handle_offer的其他分支中例如connection_already_exists同一webrtc_id重复连接与unknown_connection未知 ICE candidate。错误响应中的error字符串可以直接用于前端分支处理例如在fetch(/webrtc/offer)之后const res await response.json(); if (res.status failed) { if (res.meta.error concurrency_limit_reached) { // 提示用户当前并发已满稍后重试 } } else { await peerConnection.setRemoteDescription(res); }常见接入问题速查现象排查方向WebSocket 连不上确认目标 Stream 是audiosend-receive模式WebSocket 不支持视频与单向模式原文档明确限制收到音频但播放异常/噪声确认上行音频是 mu-law 编码且采样率与 handler 的input_sample_rate一致默认 48k见 tracks.py确认输出按output_sample_rate解码send_input后参数不生效参数在 handler 的下一次调用才生效确认start_up中调用了wait_for_args()且读取latest_args[1:]WebRTC 建连超时检查webrtc_id是否唯一确认 ICE candidate 与 offer 都 POST 到了/webrtc/offer服务器在 NAT/防火墙后时必须转发 candidate收到fetch_output但取不到数据使用output_stream()持续消费而非只调一次fetch_latest_output()后者单次等待超时为 10 秒连接被拒查看返回 JSON 的meta.errorconcurrency_limit_reached表示并发满默认 1可用Stream(concurrency_limitN)调大见 streams.mdconnection_already_exists表示webrtc_id重复参考实现与进一步阅读本文对应的原始文档docs/userguide/api.mdStream 核心对象与mount注册的路由/webrtc/offer、/websocket/offer、/telephone/*backend/fastrtc/stream.pyWebRTC 信令与连接生命周期、set_input/output_stream/fetch_latest_output实现backend/fastrtc/webrtc_connection_mixin.pyWebSocket 音频帧编解码与事件循环backend/fastrtc/websocket.pycreate_message、AdditionalOutputs、CloseStream、错误包装等消息基础件backend/fastrtc/utils.pyStreamHandler 家族与wait_for_args/set_args/ 采样率默认值backend/fastrtc/tracks.py停止词消息触发逻辑backend/fastrtc/reply_on_stopwords.py停顿检测三类 log 消息backend/fastrtc/reply_on_pause.pyStream 核心概念、Additional Inputs/Outputs 与并发控制总览docs/userguide/streams.md一个可直接运行的端到端示例Stream.mount(app) FastAPIdemo/echo_audio/app.py【免费下载链接】fastrtcThe python library for real-time communication项目地址: https://gitcode.com/GitHub_Trending/fa/fastrtc创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考