基于Vosk与Streamlit的实时离线语音识别问题求助

核心问题分析 #

• Streamlit线程安全违规：Streamlit的UI渲染由主线程独占，子线程直接调用st.write会触发线程冲突，导致无输出或报错。
• 音频数据损坏：b' '.join(recordings.get())会在音频字节流中插入无效空格字节，Vosk无法识别损坏的音频格式，导致识别中断或无结果。
• 队列设计冗余：每次将单帧音频包装成列表存入队列，既浪费资源又增加解析成本，应直接存入原始音频字节。
• 识别逻辑不实时：仅等待AcceptWaveform返回完整语句结果，未利用Vosk的PartialResult输出实时中间转录文本，导致"需等待转录完成才能继续"的问题。
• 资源泄漏风险：未处理音频流、PyAudio实例的关闭逻辑，主线程退出时子线程可能残留，引发错误。

修改方案与代码实现 #

关键调整点 #

1. 用Streamlit的st.session_state在主线程更新UI，子线程仅负责将识别结果存入共享状态
2. 直接将原始音频字节存入队列，避免拼接破坏数据
3. 启用Vosk的PartialResult实现实时增量识别，提升连续识别体验
4. 设置子线程为守护线程，确保主线程退出时子线程自动终止
5. 添加资源清理逻辑，避免音频设备占用

修改后的完整代码 #

import pyaudio
from vosk import Model, KaldiRecognizer
from threading import Thread
from queue import Queue
import streamlit as st
import time

# 初始化全局配置
CHANNELS = 1
FRAME_RATE = 32000
AUDIO_FORMAT = pyaudio.paInt16
CHUNK_SIZE = 4096  # 平衡延迟与性能的合理帧大小

# 初始化Vosk模型和识别器
model = Model(model_name="vosk-model-small-en-in-0.4")
rec = KaldiRecognizer(model, FRAME_RATE)
rec.SetWords(True)  # 可选：启用词级识别

# 初始化音频队列和Streamlit会话状态
audio_queue = Queue()
if "transcript" not in st.session_state:
    st.session_state.transcript = ""
if "is_running" not in st.session_state:
    st.session_state.is_running = False

def audio_capture():
    """音频捕获线程：持续读取麦克风数据存入队列"""
    mic = pyaudio.PyAudio()
    stream = mic.open(
        rate=FRAME_RATE,
        channels=CHANNELS,
        format=AUDIO_FORMAT,
        input=True,
        frames_per_buffer=CHUNK_SIZE
    )
    stream.start_stream()
    
    while st.session_state.is_running:
        data = stream.read(CHUNK_SIZE, exception_on_overflow=False)
        audio_queue.put(data)
    
    # 资源清理
    stream.stop_stream()
    stream.close()
    mic.terminate()

def speech_recognition():
    """语音识别线程：处理队列中的音频数据，更新会话状态"""
    while st.session_state.is_running:
        if not audio_queue.empty():
            data = audio_queue.get()
            if rec.AcceptWaveform(data):
                # 获取完整语句结果
                full_result = rec.Result()
                st.session_state.transcript += full_result[14:-3] + " "
            else:
                # 获取实时部分结果
                partial_result = rec.PartialResult()
                st.session_state["partial_transcript"] = partial_result[17:-3]

# Streamlit UI布局
st.title("实时离线语音识别系统")
st.subheader("转录结果")
transcript_area = st.empty()
partial_area = st.empty()

col1, col2 = st.columns(2)
with col1:
    if st.button("开始识别") and not st.session_state.is_running:
        st.session_state.is_running = True
        # 启动捕获和识别线程
        Thread(target=audio_capture, daemon=True).start()
        Thread(target=speech_recognition, daemon=True).start()
with col2:
    if st.button("停止识别") and st.session_state.is_running:
        st.session_state.is_running = False

# 主线程更新UI
while st.session_state.is_running:
    transcript_area.markdown(f"**完整转录:**\n{st.session_state.transcript}")
    if "partial_transcript" in st.session_state:
        partial_area.markdown(f"**实时识别:**\n{st.session_state.partial_transcript}")
    time.sleep(0.1)  # 控制UI更新频率，避免过度刷新

关键优化说明 #

• 线程安全的UI更新：所有st相关操作都在主线程循环中执行，子线程仅更新st.session_state数据，避免线程冲突
• 无损音频处理：直接存入原始音频字节，Vosk可连续处理增量数据，实现连续识别
• 实时增量识别：通过PartialResult输出中间结果，用户无需等待完整语句即可看到实时转录内容
• 资源管理：添加音频流和PyAudio实例的关闭逻辑，守护线程确保主线程退出时子线程自动终止
• 用户体验优化：添加开始/停止按钮，区分完整转录和实时识别结果，UI更新频率可控

额外建议 #

• 调整CHUNK_SIZE参数：更小的块降低延迟但增加CPU占用，根据设备性能调整（推荐2048-8192）
• 模型选择：如需更高识别精度，可切换到更大的Vosk模型（如vosk-model-en-us-0.22），但会增加内存占用
• 错误处理：可添加队列超时、模型加载失败的捕获逻辑，提升系统稳定性

内容的提问来源于stack exchange，提问作者Voleti Nagendra kumar