面向 Python 的实时通信库。

将任何 Python 函数转换为通过 WebRTC 或 WebSocket 实现的实时音视频流。

安装

pip install fastrtc

若需使用内置的静音检测功能（参见 ReplyOnPause）以及文本转语音功能（参见 Text To Speech），请安装 vad 和 tts 附加组件：

pip install "fastrtc[vad, tts]"

核心特性

• 🗣️ 内置自动语音检测与发言权管理，您只需关注如何响应用户逻辑。
• 💻 自动 UI — 使用 .ui.launch() 方法即可启动支持 WebRTC 的内置 Gradio 界面。
• 🔌 自动 WebRTC 支持 — 使用 .mount(app) 方法可将流挂载到 FastAPI 应用上，从而为您的前端提供 WebRTC 端点！
• ⚡️ WebSocket 支持 — 同样使用 .mount(app) 方法，您可以将流挂载到 FastAPI 应用上，获取适用于自定义前端的 WebSocket 端点！
• 📞 自动电话支持 — 调用流的 fastphone() 方法即可启动应用，并获得一个免费的临时电话号码！
• 🤖 完全可定制的后端 — Stream 对象可轻松挂载到 FastAPI 应用上，方便您根据生产需求进行扩展。以 Talk To Claude 演示为例，展示了如何部署自定义 JS 前端。

文档

https://fastrtc.org

示例

更多使用示例，请参阅 Cookbook。

🗣️👀 Gemini 音视频聊天 将您的摄像头视频和音频同时传输至 Google Gemini。您还可以上传图片来丰富对话内容！ 演示 | 代码	🗣️ Google Gemini 实时语音 API 利用 Google 的语音 API 与 Gemini 进行实时对话。 演示 | 代码
🗣️ OpenAI 实时语音 API 使用 OpenAI 的语音 API 与 ChatGPT 进行实时对话。 演示 | 代码	🤖 Hello Computer 提问前先说“computer”！ 演示 | 代码
🤖 Llama 代码编辑器 仅凭语音即可创建和编辑 HTML 页面！由 SambaNova 系统提供支持。 演示 | 代码	🗣️ 与 Claude 对话 借助 Anthropic 和 Play.Ht 的 API，与 Claude 进行语音对话。 演示 | 代码
🎵 Whisper 转录 让 Whisper 实时转录您的语音！ 演示 | 代码	📷 Yolov10 目标检测 在用户的实时摄像头流上运行 Yolov10 模型。 演示 | 代码
🗣️ Kyutai Moshi Kyutai 的 Moshi 是一种用于模拟人类对话的新型语音到语音模型。 演示 | 代码	🗣️ Hello Llama：停用词检测 基于 Llama 3.3 70b 构建的代码编辑器，可通过“Hello Llama”这一短语触发。仅需 100 行代码，即可打造类似 Siri 的编程助手。 演示 | 代码

使用方法

这是官方使用指南的精简版。

• .ui.launch(): 启动一个内置的 UI，方便测试和分享你的流。基于 Gradio 构建。
• .fastphone(): 获取一个免费的临时电话号码，用于拨打到你的流中。需要 Hugging Face 的 Token。
• .mount(app): 将流挂载到一个 FastAPI 应用上。非常适合与你现有的生产系统集成。

快速入门

回声音频

from fastrtc import Stream, ReplyOnPause
import numpy as np

def echo(audio: tuple[int, np.ndarray]):
    # 该函数会持续接收音频，直到用户暂停
    # 可以实现任何生成音频的迭代器
    # 更完整的示例请参见“LLM 语音聊天”
    yield audio

stream = Stream(
    handler=ReplyOnPause(echo),
    modality="audio", 
    mode="send-receive",
)

LLM 语音聊天

from fastrtc import (
    ReplyOnPause, AdditionalOutputs, Stream,
    audio_to_bytes, aggregate_bytes_to_16bit
)
import gradio as gr
from groq import Groq
import anthropic
from elevenlabs import ElevenLabs

groq_client = Groq()
claude_client = anthropic.Anthropic()
tts_client = ElevenLabs()


# 请参阅 Cookbook 中的“与 Claude 对话”部分，了解如何维护聊天历史。
def response(
    audio: tuple[int, np.ndarray],
):
    prompt = groq_client.audio.transcriptions.create(
        file=("audio-file.mp3", audio_to_bytes(audio)),
        model="whisper-large-v3-turbo",
        response_format="verbose_json",
    ).text
    response = claude_client.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}],
    )
    response_text = " ".join(
        block.text
        for block in response.content
        if getattr(block, "type", None) == "text"
    )
    iterator = tts_client.text_to_speech.convert_as_stream(
        text=response_text,
        voice_id="JBFqnCBsd6RMkjVDRZzb",
        model_id="eleven_multilingual_v2",
        output_format="pcm_24000"
        
    )
    for chunk in aggregate_bytes_to_16bit(iterator):
        audio_array = np.frombuffer(chunk, dtype=np.int16).reshape(1, -1)
        yield (24000, audio_array)

stream = Stream(
    modality="audio",
    mode="send-receive",
    handler=ReplyOnPause(response),
)

网络摄像头流

from fastrtc import Stream
import numpy as np


def flip_vertically(image):
    return np.flip(image, axis=0)


stream = Stream(
    handler=flip_vertically,
    modality="video",
    mode="send-receive",
)

物体检测

from fastrtc import Stream
import gradio as gr
import cv2
from huggingface_hub import hf_hub_download
from .inference import YOLOv10

model_file = hf_hub_download(
    repo_id="onnx-community/yolov10n", filename="onnx/model.onnx"
)

# 可克隆 https://huggingface.co/spaces/fastrtc/object-detection 获取 YOLOv10 的实现
model = YOLOv10(model_file)

def detection(image, conf_threshold=0.3):
    image = cv2.resize(image, (model.input_width, model.input_height))
    new_image = model.detect_objects(image, conf_threshold)
    return cv2.resize(new_image, (500, 500))

stream = Stream(
    handler=detection,
    modality="video", 
    mode="send-receive",
    additional_inputs=[
        gr.Slider(minimum=0, maximum=1, step=0.01, value=0.3)
    ]
)

运行流

运行方式如下：

Gradio

stream.ui.launch()

电话（仅音频）

stream.fastphone()

FastAPI

app = FastAPI()
stream.mount(app)

# 可选：添加路由
@app.get("/")
async def _():
    return HTMLResponse(content=open("index.html").read())

# 使用 uvicorn app:app --host 0.0.0.0 --port 8000 启动应用

FastRTC 快速上手指南

FastRTC 是一个用于 Python 的实时通信库，可将任意 Python 函数快速转换为基于 WebRTC 或 WebSocket 的实时音视频流。它内置了语音检测、自动 UI 生成及电话接入功能，非常适合构建实时 AI 交互应用。

环境准备

• 系统要求：支持 Windows、macOS 和 Linux。
• Python 版本：建议 Python 3.9 及以上版本。
• 前置依赖：
  • 若需使用内置的暂停检测（Voice Activity Detection）和文本转语音（TTS）功能，需安装额外依赖包。
  • 国内开发者如遇网络问题，建议使用国内镜像源加速安装。

安装步骤

1. 基础安装

仅安装核心功能：

pip install fastrtc -i https://pypi.tuna.tsinghua.edu.cn/simple

2. 完整功能安装

推荐安装包含 vad (语音活动检测) 和 tts (文本转语音) 的完整版本，以启用“暂停回复”等高级功能：

pip install "fastrtc[vad, tts]" -i https://pypi.tuna.tsinghua.edu.cn/simple

基本使用

FastRTC 的核心是通过 Stream 类包装你的处理函数。以下是三个最典型的使用场景。

场景一：音频回声测试 (最简单示例)

此示例将用户输入的音频直接返回，用于测试连通性。它使用了 ReplyOnPause 处理器，即检测到用户停止说话后才开始处理。

from fastrtc import Stream, ReplyOnPause
import numpy as np

def echo(audio: tuple[int, np.ndarray]):
    # 函数接收音频数据直到用户暂停
    # 直接 yield 返回相同的音频
    yield audio

stream = Stream(
    handler=ReplyOnPause(echo),
    modality="audio", 
    mode="send-receive",
)

# 启动内置的 Gradio UI 进行测试
if __name__ == "__main__":
    stream.ui.launch()

场景二：实时视频处理

此示例读取摄像头画面，将其垂直翻转后实时推流。

from fastrtc import Stream
import numpy as np

def flip_vertically(image):
    # image 是 numpy 数组
    return np.flip(image, axis=0)

stream = Stream(
    handler=flip_vertically,
    modality="video",
    mode="send-receive",
)

if __name__ == "__main__":
    stream.ui.launch()

场景三：集成 LLM 语音对话

结合 Groq (Whisper)、Anthropic (Claude) 和 ElevenLabs (TTS) 实现完整的语音对话流程。

from fastrtc import (
    ReplyOnPause, Stream,
    audio_to_bytes, aggregate_bytes_to_16bit
)
import numpy as np
# 请确保已安装相关 SDK: pip install groq anthropic elevenlabs
from groq import Groq
import anthropic
from elevenlabs import ElevenLabs

# 初始化客户端 (请设置环境变量 API KEY)
groq_client = Groq()
claude_client = anthropic.Anthropic()
tts_client = ElevenLabs()

def response(audio: tuple[int, np.ndarray]):
    # 1. 语音转文字 (Whisper)
    prompt = groq_client.audio.transcriptions.create(
        file=("audio-file.mp3", audio_to_bytes(audio)),
        model="whisper-large-v3-turbo",
        response_format="verbose_json",
    ).text
    
    # 2. 获取 LLM 回复 (Claude)
    response = claude_client.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}],
    )
    response_text = " ".join(
        block.text
        for block in response.content
        if getattr(block, "type", None) == "text"
    )
    
    # 3. 文字转语音 (ElevenLabs) 并流式返回
    iterator = tts_client.text_to_speech.convert_as_stream(
        text=response_text,
        voice_id="JBFqnCBsd6RMkjVDRZzb",
        model_id="eleven_multilingual_v2",
        output_format="pcm_24000"
    )
    
    for chunk in aggregate_bytes_to_16bit(iterator):
        audio_array = np.frombuffer(chunk, dtype=np.int16).reshape(1, -1)
        yield (24000, audio_array)

stream = Stream(
    modality="audio",
    mode="send-receive",
    handler=ReplyOnPause(response),
)

if __name__ == "__main__":
    stream.ui.launch()

运行方式

创建 stream 对象后，可通过以下三种方式运行：

1. Web UI 模式 (推荐开发调试) 启动内置的 Gradio 界面，直接在浏览器中测试。

   stream.ui.launch()
   

2. 电话接入模式 (仅限音频) 获取一个免费的临时电话号码，拨打该号码即可与你的程序交互（需要 Hugging Face Token）。

   stream.fastphone()
   

3. 生产部署模式 (FastAPI) 将流挂载到 FastAPI 应用，以便集成到现有的前端或系统中。

   from fastapi import FastAPI
   from fastapi.responses import HTMLResponse
   
   app = FastAPI()
   stream.mount(app)
   
   # 可选：添加自定义前端路由
   @app.get("/")
   async def _():
       return HTMLResponse(content="<h1>FastRTC Service Running</h1>")
   
   # 启动命令: uvicorn app:app --host 0.0.0.0 --port 8000