Azure Speech Services连续识别时Streamlit的st.write()无法显示内容求助

问题描述 #

使用Azure Speech Services构建Streamlit应用时，recognize_once()逻辑能正常在UI展示识别结果，但使用start_continuous_recognition()时，回调函数中的st.write()无法更新UI内容，仅能在控制台打印识别结果。

单次识别可正常运行的代码 #

def speech_recognize_once_from_mic():
    # Set up the speech config and audio config
    speech_config = speechsdk.SpeechConfig(subscription=subscription_key, region=service_region)

    audio_config = speechsdk.AudioConfig(use_default_microphone=True)

    # Create a speech recognizer with the given settings
    speech_recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, audio_config=audio_config)

    st.write("Speak into your microphone.")
    result = speech_recognizer.recognize_once_async().get()

    # Check the result
    if result.reason == speechsdk.ResultReason.RecognizedSpeech:
        return f"Recognized: {result.text}"
    elif result.reason == speechsdk.ResultReason.NoMatch:
        return "No speech could be recognized"
    elif result.reason == speechsdk.ResultReason.Canceled:
        cancellation_details = result.cancellation_details
        return f"Speech Recognition canceled: {cancellation_details.reason}"
    else:
        return "Unknown error"

# Simple UI for processing the audio input
st.title("Azure Speech Service with Streamlit")

if st.button('Start speech recognition'):
    recognition_result = speech_recognize_once_from_mic()
    st.write(recognition_result)

连续识别无法更新UI的代码 #

def recognized_callback(evt):
    # Callback function that appends recognized speech to chunks
    global chunks
    if evt.result.reason == speechsdk.ResultReason.RecognizedSpeech:
        chunks.append(evt.result.text)
        print(f"Recognized: {evt.result.text}")
        st.write(chunks)
        print('Done')


def process_audio():
    # Set up the recognizer
    recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, audio_config=audio_config)
    recognizer.recognized.connect(recognized_callback)
    recognizer.session_stopped.connect(session_stopped_callback)

    # Store the recognizer in the session state to access it later
    st.session_state.recognizer = recognizer

    # Start continuous recognition
    recognizer.start_continuous_recognition()

原因分析 #

Streamlit的UI更新依赖于脚本的重新运行，而Azure Speech的连续识别回调函数是在后台独立线程中执行的，直接在回调里调用st.write()无法触发Streamlit的UI渲染流程，因此UI不会更新。

解决方案 #

通过以下步骤调整代码，适配Streamlit的运行机制：

1. 使用st.session_state存储识别结果，替代全局变量，确保状态在脚本重跑时不丢失；
2. 用st.empty()创建可动态更新的UI容器，用于展示实时识别内容；
3. 通过线程安全的标志位触发Streamlit的UI重跑，确保回调中的状态变更能同步到UI。

修改后的完整代码 #

import streamlit as st
import azure.cognitiveservices.speech as speechsdk
import threading

# 初始化Azure Speech配置（替换为你的密钥和区域）
subscription_key = "YOUR_SUBSCRIPTION_KEY"
service_region = "YOUR_SERVICE_REGION"

# 初始化Streamlit会话状态
if "chunks" not in st.session_state:
    st.session_state.chunks = []
if "recognizer" not in st.session_state:
    st.session_state.recognizer = None
if "needs_rerun" not in st.session_state:
    st.session_state.needs_rerun = False
if "is_running" not in st.session_state:
    st.session_state.is_running = False

# 线程锁，确保状态修改的线程安全
lock = threading.Lock()

def recognized_callback(evt):
    if evt.result.reason == speechsdk.ResultReason.RecognizedSpeech:
        with lock:
            st.session_state.chunks.append(evt.result.text)
            st.session_state.needs_rerun = True
        print(f"Recognized: {evt.result.text}")

def session_stopped_callback(evt):
    with lock:
        st.session_state.is_running = False
        st.session_state.needs_rerun = True
    print("Recognition session stopped")

def start_recognition():
    speech_config = speechsdk.SpeechConfig(subscription=subscription_key, region=service_region)
    audio_config = speechsdk.AudioConfig(use_default_microphone=True)
    
    recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, audio_config=audio_config)
    recognizer.recognized.connect(recognized_callback)
    recognizer.session_stopped.connect(session_stopped_callback)
    
    st.session_state.recognizer = recognizer
    st.session_state.is_running = True
    recognizer.start_continuous_recognition()

def stop_recognition():
    if st.session_state.recognizer:
        st.session_state.recognizer.stop_continuous_recognition()
        st.session_state.is_running = False

# UI部分
st.title("Azure Speech Service 实时连续识别")

# 创建可更新的内容容器
result_container = st.empty()

# 控制按钮
col1, col2 = st.columns(2)
with col1:
    if st.button("开始识别", disabled=st.session_state.is_running):
        start_recognition()
with col2:
    if st.button("停止识别", disabled=not st.session_state.is_running):
        stop_recognition()

# 检查是否需要重跑UI
with lock:
    if st.session_state.needs_rerun:
        st.session_state.needs_rerun = False
        st.rerun()

# 在容器中展示实时结果
result_container.write("### 实时识别内容")
result_container.write("\n".join(st.session_state.chunks))

关键说明 #

• 线程安全：使用threading.Lock()确保会话状态的修改不会因多线程操作出现异常；
• UI重跑触发：通过st.session_state.needs_rerun标志位，在主线程中触发st.rerun()，让Streamlit重新渲染UI；
• 动态容器：st.empty()创建的容器可以在脚本重跑时更新内容，实现实时展示效果；
• 状态管理：所有关键状态（识别器实例、识别结果、运行状态）都存储在st.session_state中，确保页面刷新或重跑时状态不丢失。

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