Streamlit应用中AWS Transcribe实时转录无法停止的问题修复

问题根源 #

你的代码存在三个核心问题导致停止按钮无效：

1. 状态管理错误：用全局变量保存转录状态，但Streamlit每次交互都会重新执行整个脚本，全局变量会被重置，无法维持状态。
2. 阻塞式事件循环：loop.run_until_complete()会卡住整个Streamlit进程，停止按钮的点击事件要等到转录结束才能被处理。
3. 未主动终止会话：仅靠全局变量标记停止，无法主动关闭音频流和通知AWS Transcribe结束转录会话。

修复后的完整代码 #

import os
import asyncio
import streamlit as st
import sounddevice
from amazon_transcribe.auth import AwsCrtCredentialResolver
from amazon_transcribe.client import TranscribeStreamingClient
from amazon_transcribe.handlers import TranscriptResultStreamHandler
from amazon_transcribe.model import TranscriptEvent
from dotenv import load_dotenv

load_dotenv()

# 用Streamlit Session State替代全局变量，确保状态在交互中保留
if "transcription_started" not in st.session_state:
    st.session_state.transcription_started = False
if "latest_transcript" not in st.session_state:
    st.session_state.latest_transcript = ""
if "audio_stream" not in st.session_state:
    st.session_state.audio_stream = None
if "transcribe_task" not in st.session_state:
    st.session_state.transcribe_task = None
if "input_queue" not in st.session_state:
    st.session_state.input_queue = asyncio.Queue()

class MyEventHandler(TranscriptResultStreamHandler):
    async def handle_transcript_event(self, transcript_event: TranscriptEvent):
        results = transcript_event.transcript.results
        for result in results:
            if not result.is_partial:
                for alt in result.alternatives:
                    st.session_state.latest_transcript += alt.transcript + " "

async def mic_stream():
    loop = asyncio.get_event_loop()

    def callback(indata, frame_count, time_info, status):
        if st.session_state.transcription_started:
            loop.call_soon_threadsafe(st.session_state.input_queue.put_nowait, (bytes(indata), status))

    # 保存音频流到Session State，方便后续关闭
    st.session_state.audio_stream = sounddevice.RawInputStream(
        channels=1,
        samplerate=16000,
        callback=callback,
        blocksize=1024 * 2,
        dtype="int16",
    )

    with st.session_state.audio_stream:
        while st.session_state.transcription_started:
            try:
                # 加入超时检测，确保能及时响应停止信号
                indata, status = await asyncio.wait_for(st.session_state.input_queue.get(), timeout=0.5)
                yield indata, status
            except asyncio.TimeoutError:
                continue

async def write_chunks(stream):
    async for chunk, status in mic_stream():
        if not st.session_state.transcription_started:
            break
        await stream.input_stream.send_audio_event(audio_chunk=chunk)
    # 主动发送结束信号给AWS Transcribe
    await stream.input_stream.end_stream()

async def basic_transcribe():
    client = TranscribeStreamingClient(region="us-east-1")
    stream = await client.start_stream_transcription(
        language_code="en-US",
        media_sample_rate_hz=16000,
        media_encoding="pcm",
    )

    handler = MyEventHandler(stream.output_stream)
    try:
        await asyncio.gather(write_chunks(stream), handler.handle_events())
    except asyncio.CancelledError:
        # 处理任务取消事件，主动结束流
        await stream.input_stream.end_stream()
    finally:
        st.session_state.transcription_started = False
        if st.session_state.audio_stream:
            st.session_state.audio_stream.stop()
            st.session_state.audio_stream.close()

def main():
    st.title("实时语音转录")
    
    # 布局按钮，避免重复点击
    col1, col2 = st.columns(2)
    with col1:
        start_btn = st.button("开始转录", disabled=st.session_state.transcription_started)
    with col2:
        stop_btn = st.button("停止转录", disabled=not st.session_state.transcription_started)

    if start_btn:
        st.session_state.transcription_started = True
        st.session_state.latest_transcript = ""  # 清空历史转录内容
        # 创建非阻塞异步任务
        loop = asyncio.get_event_loop()
        st.session_state.transcribe_task = loop.create_task(basic_transcribe())

    if stop_btn:
        st.session_state.transcription_started = False
        # 取消正在运行的转录任务
        if st.session_state.transcribe_task and not st.session_state.transcribe_task.done():
            st.session_state.transcribe_task.cancel()
        st.success("转录已停止")

    # 实时展示转录结果
    st.subheader("最新转录结果")
    st.write(st.session_state.latest_transcript)

if __name__ == "__main__":
    main()

关键修复说明 #

• Session State管理状态：用st.session_state存储转录状态、音频流、异步任务等，确保页面交互时状态不丢失。
• 非阻塞异步任务：用asyncio.create_task()启动转录，避免卡住Streamlit UI，让停止按钮能及时响应。
• 主动终止会话：点击停止时取消异步任务，调用end_stream()通知AWS Transcribe结束会话，同时关闭音频输入流。
• 超时检测：在音频流读取中加入超时，确保能及时检测到停止信号，避免无限等待队列数据。

内容的提问来源于stack exchange，提问作者Nirodya Pussadeniya