当前位置：首页 > news >正文

一个可本地运行的实时字幕翻译 Demo（Whisper + GPT + Streamlit），可以边说边出中英文字幕

news 2025/10/31 9:25:11

这个 demo 在本机运行（开发/会议桌面场景）能做到“边说边出字幕（英文 → 中文）”，并包含增量显示、简单断句与缓存策略，方便后续替换为低延迟 streaming ASR 或本地 LLM。

注意：示例使用 faster-whisper 做本地 ASR（也可换 OpenAI Whisper API / SenseVoice / FunASR），翻译用 OpenAI ChatCompletion（可替换为本地 LLM）。如果要把它部署到远端服务器并通过浏览器远程录音，需要改用浏览器端采集（WebRTC / MediaRecorder）并通过 WebSocket 上传音频；还可以后续优化。

依赖与准备
Python 版本建议：3.9+（支持 torch/cuda 环境更佳）
安装依赖：
pip install streamlit sounddevice numpy faster-whisper openai python-dotenv
说明：
faster-whisper：高性能 Whisper 实现（CPU/GPU均可）
sounddevice：本地麦克风抓取（适合在本机运行）
streamlit：可视化与实时显示
openai：调用 GPT 翻译（可替换为其他 LLM）
python-dotenv：从 .env 读取 API Key（更安全）
在项目根目录新建 .env：
OPENAI_API_KEY=sk-…
完整 demo 源码（保存为 app.py）

app.py

import os
import threading
import queue
import time
from datetime import datetime, timedelta
import numpy as np
import sounddevice as sd
import streamlit as st
from dotenv import load_dotenv
from faster_whisper import WhisperModel
import openai

load_dotenv()
OPENAI_API_KEY = os.getenv(“OPENAI_API_KEY”)
if not OPENAI_API_KEY:
st.error(“请在 .env 设置 OPENAI_API_KEY 后重启”)
openai.api_key = OPENAI_API_KEY

========== 配置 ==========

SAMPLE_RATE = 16000
CHUNK_SECONDS = 1.0 # 每块音频长度（秒）
CHUNK_SIZE = int(SAMPLE_RATE * CHUNK_SECONDS)
DEVICE = None # None 默认设备
WHISPER_SIZE = “small” # 根据硬件可改为 “tiny”,“small”,“medium”,“large”
TRANSCRIBE_BEAM = 3
LANGUAGE = None # None：自动检测；或者 “en” / “zh” 等

Whisper 模型加载（会消耗时间）

@st.cache_resource(show_spinner=False)
def load_asr_model(size):
device = “cuda” if (sd.default.device and hasattr(sd, “RawInputStream”) and False) else “cpu”
# faster-whisper 默认会选择 cpu/gpu，根据环境自动
model = WhisperModel(size, device=“cpu”) # 如果有 GPU 可改为 “cuda”
return model

model = load_asr_model(WHISPER_SIZE)

========== 音频采集线程 ==========

audio_queue = queue.Queue()
stop_event = threading.Event()

def audio_callback(indata, frames, time_info, status):
# indata 是 numpy array int16/float32 取决 dtype
if status:
# 打印一些设备警告
print(“音频状态：”, status)
# 将原始帧 push 到队列（转为 int16 bytes）
audio_queue.put(indata.copy())

def recording_thread():
with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype=“int16”,
blocksize=CHUNK_SIZE, callback=audio_callback, device=DEVICE):
while not stop_event.is_set():
time.sleep(0.1)

========== ASR + 翻译逻辑 ==========

我们会把多块 chunk 合并为短片段（例如 1~3 秒）做增量识别，识别后送 GPT 翻译并显示

transcribe_lock = threading.Lock()
display_queue = queue.Queue()

def asr_translate_worker():
“”"
不断从 audio_queue 读取 chunk（numpy arrays），合并后写入文件并用 whisper transcribe，
然后调用 OpenAI 翻译并把结果放 display_queue。
“”"
buffer = np.zeros((0,), dtype=np.int16)
max_buffer_seconds = 6.0
max_buffer_len = int(SAMPLE_RATE * max_buffer_seconds)

last_sent_time = datetime.utcnow()
segment_count = 0while not stop_event.is_set():try:chunk = audio_queue.get(timeout=0.5)   # block 等待except queue.Empty:# 超时，若有残余 buffer 且时间超过阈值，则强制识别if buffer.size > 0 and (datetime.utcnow() - last_sent_time) > timedelta(seconds=1.0):with transcribe_lock:segment_count += 1do_transcribe_and_translate(buffer.tobytes(), segment_count)buffer = np.zeros((0,), dtype=np.int16)last_sent_time = datetime.utcnow()continue# chunk shape -> (N,1)flat = np.squeeze(chunk)buffer = np.concatenate([buffer, flat])# 当 buffer 超过阈值或者已经有较长静音后，触发识别if buffer.size >= int(SAMPLE_RATE * 1.0):  # 最少 1 秒触发# 控制最大长度，过长先切段处理if buffer.size > max_buffer_len:take = buffer[:max_buffer_len]remainder = buffer[max_buffer_len:]with transcribe_lock:segment_count += 1do_transcribe_and_translate(take.tobytes(), segment_count)buffer = remainderlast_sent_time = datetime.utcnow()else:# 如果近期没有新输入（短时间静音），或者累计到 2s，则触发if (datetime.utcnow() - last_sent_time) > timedelta(seconds=0.7) or buffer.size >= int(SAMPLE_RATE * 2.0):with transcribe_lock:segment_count += 1do_transcribe_and_translate(buffer.tobytes(), segment_count)buffer = np.zeros((0,), dtype=np.int16)last_sent_time = datetime.utcnow()

def do_transcribe_and_translate(raw_bytes, seg_idx):
“”"
raw_bytes: int16 bytes, 16000 Hz mono
“”"
# 保存为临时 wav（faster-whisper 可接收 numpy 或文件路径）
import tempfile, soundfile as sf, io
arr = np.frombuffer(raw_bytes, dtype=np.int16).astype(np.float32) / 32768.0
# 写入临时 wav
tmpf = tempfile.NamedTemporaryFile(suffix=“.wav”, delete=False)
sf.write(tmpf.name, arr, SAMPLE_RATE, subtype=‘FLOAT’)
tmpf.flush()

# ASR 识别（非流式，短片段）
segments, info = model.transcribe(tmpf.name, beam_size=TRANSCRIBE_BEAM, language=LANGUAGE)
text = " ".join([seg.text for seg in segments]).strip()if not text:# 无识别结果（纯静音）return# 发送给 LLM 翻译（这里示例翻英文到中文），并做简单润色
prompt = ("你是一个字幕润色与翻译助手。\n""把下面的英文短句翻译成自然、口语化、适合实时字幕的中文。""\n只返回翻译文本（不要添加注释）。\n\n"f"英文原句：\n{text}\n\n翻译："
)try:resp = openai.ChatCompletion.create(model="gpt-4o-mini",   # 可替换为 gpt-4o, gpt-4 或本地 LLM APImessages=[{"role":"user","content":prompt}],temperature=0.2,max_tokens=200,)translated = resp.choices[0].message.content.strip()
except Exception as e:translated = "(翻译失败：" + str(e) + ")"# 放到显示队列（包含原文、译文、时间戳）
display_queue.put({"seg": seg_idx,"time": datetime.utcnow().isoformat(),"source": text,"target": translated
})

========== Streamlit 前端 ==========

st.set_page_config(page_title=“实时翻译字幕 Demo”, layout=“wide”)
st.title(“🔴 实时翻译字幕 Demo（Whisper + GPT）”)
left, right = st.columns([3,2])

with right:
st.markdown(“### 控制”)
if ‘running’ not in st.session_state:
st.session_state.running = False
if st.session_state.running:
if st.button(“停止录音”):
stop_event.set()
st.session_state.running = False
else:
if st.button(“开始录音”):
# 清理并启动线程
stop_event.clear()
t_rec = threading.Thread(target=recording_thread, daemon=True)
t_asr = threading.Thread(target=asr_translate_worker, daemon=True)
t_rec.start(); t_asr.start()
st.session_state.running = True

st.markdown("---")
st.markdown("### 配置")
st.write(f"- Whisper 模型：{WHISPER_SIZE}")
st.write(f"- 采样率：{SAMPLE_RATE} Hz")
st.write(f"- 每块长度：{CHUNK_SECONDS}s")
st.info("Tip：本 Demo 在本机运行最佳。部署到远端请改用浏览器端采集并通过 WebSocket 发送音频。")

with left:
st.markdown(“### 实时字幕（最新显示在顶部）”)
subtitle_box = st.empty()
# 历史列表
history = []

主循环：从 display_queue 读取并更新界面

try:
while True:
try:
item = display_queue.get(timeout=0.3)
except queue.Empty:
# 若停止且队列空，退出
if stop_event.is_set() and display_queue.empty():
break
# 否则继续刷新
pass
else:
# 新条目到达
history.insert(0, item)
# 保持历史长度在 50 条以内
history = history[:50]
# 渲染
lines = []
for h in history[:20]:
ts = h[“time”].split(“T”)[-1][:12]
lines.append(f"[{ts}] {h[‘target’]} — 原文: {h[‘source’]}“)
subtitle_box.markdown(”\n\n".join(lines))
# small sleep for UI CPU
time.sleep(0.05)
except Exception as e:
st.error(f"前端循环出错：{e}")
finally:
# 停止线程
stop_event.set()

如何运行
把 app.py 保存好，并在同目录放 .env（含 OPENAI_API_KEY）。
运行：
streamlit run app.py
打开浏览器（Streamlit 会给你本地地址，通常是 http://localhost:8501），点 “开始录音”，对麦克风讲话，页面会实时显示识别并翻译后的字幕（最新在上方）。
设计说明与关键点
录音方式：使用 sounddevice 在本机直接获取 PCM 数据（int16），每 CHUNK_SECONDS 推入队列。
增量识别策略：

我把短 chunk 合并为 1–2 秒小片段做“短批次识别”，这样能在较低延迟下拿到结果。
如果长时间静音或缓冲到 2s 则强制触发识别，避免一直等待完整句子导致延迟。
ASR：faster-whisper 的 transcribe() 是批处理接口（对短片段效果好）。若需要更低延迟，应改用 streaming ASR（比如 OpenAI Realtime 或 SenseVoice streaming / FunASR streaming）。
翻译：使用 GPT 作上下文理解与润色，能把不完整或断裂的识别结果补全成更自然字幕。你也可以直接使用翻译模型（Marian, mBART, NLLB）做更便宜、离线翻译。
UI：Streamlit 仅作演示与调试用。用于生产可替换为 React + WebSocket + WebRTC。

延迟、成本与注意事项
延迟来源：

采集块长度（CHUNK_SECONDS）直接影响最小识别延迟。
ASR 推理时间（取决于模型大小与是否使用 GPU）。
LLM 翻译网络延迟 & token 生成时间（使用云端 GPT 会显著增加延迟）。
成本：

使用 GPT 云 API 会产生成本（按 token/模型计费）。本地 LLM 可以节省费用，但需要更强的硬件。
隐私：音频/文本被传到 OpenAI 会触发数据处理条款。若有隐私需求请使用本地 LLM 或自托管模型。

后续优化建议（从工程角度给你可直接落地的路线）
短期（易实现）
把 CHUNK_SECONDS 缩短到 0.5s，搭配更快的 streaming ASR（比如 SenseVoiceSmall）。
把 OpenAI 换成 cheaper 翻译（如 HuggingFace 的翻译模型）以降低成本。
在翻译 prompt 中加入“保留专有名词”或“使用字幕短句”，提高字幕质量。
中期（更复杂）
把 ASR 换为 streaming 接口（WebSocket），前端用 WebRTC 采集并实时发送 PCM，后端实现增量识别（partial hypotheses）。
使用增量翻译（incremental translation）与句子边界检测（VAD + punctuation restoration）减少回滚。
将 LLM 做“字幕润色”而非完整翻译：先用 MT 做基础翻译，再用 LLM 做上下文润色（减少 token 用量和延迟）。
长期（Production）
多路音源支持 + 说话人分离（diarization）+ 实时翻译回传（同传语音合成）。
同声传译（低延迟 TTS）与字幕同步（时间轴精确控制）。
单机 GPU 加速 + 本地 LLM（Qwen、Mistral、Llama2 等）做离线全链路。

查看全文

http://www.dtcms.com/a/549350.html