Hugging Face 发布 Python 实时通信库 FastRTC

在过去的几个月里，许多新的实时语音模型被发布，围绕开源和闭源模型成立了不少公司。列举几个里程碑：

OpenAI 和 Google 分别发布了 ChatGPT 和 Gemini 的实时多模态 API。OpenAI 甚至更进一步，推出了 1-800-ChatGPT 电话号码！
Kyutai 发布了 Moshi，一个完全开源的音频到音频 LLM。阿里巴巴发布了 Qwen2-Audio，Fixie.ai 发布了 Ultravox——两个原生理解音频的开源 LLM。
ElevenLabs 在 C 轮融资中筹集了 1.8 亿美元。

尽管在模型和资金方面出现了爆发式增长，构建能够流式传输音频和视频的实时 AI 应用仍然很困难，尤其是在 Python 中。

机器学习工程师可能没有构建实时应用所需技术的经验，比如 WebRTC。
即使是像 Cursor 和 Copilot 这样的代码助手工具，也难以编写支持实时音频/视频应用的 Python 代码。我有过亲身体验！

这就是为什么我们兴奋地宣布 FastRTC，这是一个 Python 的实时通信库。该库旨在让完全用 Python 构建实时音频和视频 AI 应用变得超级简单！

在这篇博客文章中，我们将通过构建实时音频应用来介绍 FastRTC 的基础知识。最后，你将了解 FastRTC 的核心功能：

🗣️ 内置自动语音检测和轮流发言，所以你只需要关心响应用户的逻辑。
💻 自动 UI - 内置支持 WebRTC 的 Gradio UI，用于测试（或部署到生产环境！）。
📞 通过电话呼叫 - 使用 fastphone() 获取一个免费电话号码，拨打进入你的音频流（需要 HF Token。PRO 账户有更高的限制）。
⚡️ 支持 WebRTC 和 Websocket。
💪 可定制 - 你可以将流挂载到任何 FastAPI 应用，以便提供自定义 UI 或部署到 Gradio 之外。
🧰 大量实用工具，包括文本到语音、语音到文本、停止词检测等，让你快速上手。

让我们开始吧。

入门

我们将从构建实时音频的“hello world”开始：回显用户说的话。在 FastRTC 中，这就像这样简单：

from fastrtc import Stream, ReplyOnPause
import numpy as np

def echo(audio: tuple[int, np.ndarray]) -> tuple[int, np.ndarray]:
    yield audio

stream = Stream(ReplyOnPause(echo), modality="audio", mode="send-receive")
stream.ui.launch()

让我们分解一下：

ReplyOnPause 会为你处理语音检测和轮流发言。你只需要关心响应用户的逻辑。任何返回音频元组（表示为 (sample_rate, audio_data)）的生成器都可以工作。
Stream 类会为你构建一个 Gradio UI，以便快速测试你的流。完成原型设计后，你可以用一行代码将流部署为生产就绪的 FastAPI 应用——stream.mount(app)。其中 app 是一个 FastAPI 应用。

实际操作如下：

升级：LLM 语音聊天

下一个层次是使用 LLM 来响应用户。FastRTC 内置了语音到文本和文本到语音功能，因此与 LLM 一起工作非常容易。让我们相应地修改我们的 echo 函数：

import os

from fastrtc import (ReplyOnPause, Stream, get_stt_model, get_tts_model)
from openai import OpenAI

sambanova_client = OpenAI(
    api_key=os.getenv("SAMBANOVA_API_KEY"), base_url="https://api.sambanova.ai/v1"
)
stt_model = get_stt_model()
tts_model = get_tts_model()

def echo(audio):
    prompt = stt_model.stt(audio)
    response = sambanova_client.chat.completions.create(
        model="Meta-Llama-3.2-3B-Instruct",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=200,
    )
    prompt = response.choices[0].message.content
    for audio_chunk in tts_model.stream_tts_sync(prompt):
        yield audio_chunk

stream = Stream(ReplyOnPause(echo), modality="audio", mode="send-receive")
stream.ui.launch()

我们使用 SambaNova API，因为它速度快。get_stt_model() 会从 Hub 获取 Moonshine Base，get_tts_model() 会获取 Kokoro，两者都进一步优化了设备端 CPU 推理。但你可以使用任何 LLM/文本到语音/语音到文本 API，甚至语音到语音模型。带上你喜欢的工具——FastRTC 只负责实时通信层。

额外福利：通过电话呼叫

如果你调用 stream.fastphone() 而不是 stream.ui.launch()，你会得到一个免费电话号码，可以拨打进入你的流。注意，需要 HF Mirror token。PRO 账户的限制更高。

你会在终端中看到类似这样的内容：

INFO:	  Your FastPhone is now live! Call +1 877-713-4471 and use code 530574 to connect to your stream.
INFO:	  You have 30:00 minutes remaining in your quota (Resetting on 2025-03-23)

然后你可以拨打这个号码，它就会把你连接到你的流！

下一步

阅读文档以了解 FastRTC 的基础知识。
开始构建的最佳方式是查看cookbook。了解如何与流行的 LLM 提供商（包括 OpenAI 和 Gemini 的实时 API）集成、将你的流与 FastAPI 应用集成并进行自定义部署、从处理器返回额外数据、进行视频处理等！
⭐️ 给仓库加星，并提交 bug 和 issue 请求！
在 HuggingFace 上关注FastRTC Org以获取更新，并查看已部署的示例！

感谢你查看 FastRTC！