Streamlit问题：点击Continue按钮AI响应不显示，仅超时后显示

问题描述 #

开发基于Streamlit的应用，使用pytubefix获取YouTube视频音频并转录。目前遇到的问题是：AI转录结果仅在计时器超时后正常显示，但点击custom_oauth_verifier函数中的“Continue”按钮后，应用中看不到响应。

代码使用threading.Event等待用户点击按钮或计时器到期，计时器超时后流程正常，但点击按钮后无结果展示。测试时将计时器设为5秒，并硬编码了YouTube URL。

代码示例 #

import queue
import tempfile
import threading
from io import BytesIO

import streamlit as st
from pytubefix import YouTube
from streamlit.runtime.scriptrunner import add_script_run_ctx


def custom_oauth_verifier(verification_url, user_code):
    continue_event = threading.Event()
    container = st.container()
    with container:
        message_area = st.empty()
        st.button("Continue...", on_click=continue_event.set)

    base_message = f"""
#### Please complete the verification process:

- Open the following URL: [{verification_url}]({verification_url})
- Enter the code: `{user_code}`

Once you have done that, click the **'Continue...'** button below.

_This action will expire in {{time_remaining}} seconds._
"""

    for time_remaining in range(5, 0, -1):
        message_area.markdown(base_message.format(time_remaining=time_remaining))
        if continue_event.wait(1):
            break
    container.empty()


def get_video_transcript(url, result_queue):
    try:
        buf = BytesIO()
        audio = YouTube(
            url,
            use_oauth=True,
            allow_oauth_cache=False,
            oauth_verifier=custom_oauth_verifier,
        ).streams.get_audio_only()
        audio.stream_to_buffer(buf)
        buf.seek(0)
        with tempfile.NamedTemporaryFile(delete=True, suffix=".mp4") as temp_file:
            temp_file.write(buf.read())
            # TODO: Transcribe the audio here
            result = "Lorem ipsum dolor..."
            result_queue.put(result)
    except Exception as e:
        result = f"Error: {e}"
        result_queue.put(result)


def transcribe_youtube_url(url):
    result_queue = queue.Queue()
    pytube_thread = threading.Thread(target=get_video_transcript, args=(url, result_queue))
    add_script_run_ctx(pytube_thread)
    pytube_thread.start()
    pytube_thread.join()
    return result_queue.get()


chat_history: list = st.session_state.setdefault(
    "chat_history",
    [{"type": "ai", "content": "Hello, how are you?"}],
)


for message in chat_history:
    with st.chat_message(message["type"]):
        st.write(message["content"])


if user_prompt := st.chat_input(placeholder="Type your message..."):
    message = {"type": "human", "content": user_prompt}
    chat_history.append(message)
    with st.chat_message(message["type"]):
        st.write(message["content"])

    with st.spinner("Loading..."):
        ai_response = transcribe_youtube_url("https://youtu.be/ahq6MF349Fw")
        message = {"type": "ai", "content": ai_response}
        chat_history.append(message)
        with st.chat_message(message["type"]):
            st.write(message["content"])
        st.rerun()

问题原因分析 #

1. Streamlit脚本重运行模型与线程事件冲突：
   Streamlit是基于脚本每次交互都重新运行的模型，点击“Continue”按钮会触发新的脚本运行。而原有的threading.Event属于上一次脚本运行的子线程实例，新脚本无法感知该事件状态，导致原线程的wait虽被触发，但主脚本的join等待流程已与新上下文脱节，无法触发UI更新。

2. 子线程操作UI的局限性：
   custom_oauth_verifier在子线程中创建按钮和UI元素，尽管使用了add_script_run_ctx，但Streamlit的UI渲染依赖主脚本执行流程，子线程中的UI交互无法正确触发主脚本状态更新，导致点击按钮后转录流程无法被主脚本感知。

3. 容器清空时机干扰：
   无论超时还是点击按钮，container.empty()都会执行。如果是按钮点击触发的脚本重运行，该清空操作会在原线程中执行，干扰新脚本的UI状态，导致转录结果无法正常渲染。

解决方案 #

核心思路：改用Session State管理验证状态 #

放弃threading.Event，改用Streamlit的session_state跟踪OAuth验证完成状态，适配其脚本重运行模型。

修改后的代码示例 #

import tempfile
from io import BytesIO

import streamlit as st
from pytubefix import YouTube


def handle_oauth_verification(verification_url, user_code):
    # 初始化session_state中的验证状态
    if "oauth_verified" not in st.session_state:
        st.session_state.oauth_verified = False

    container = st.container()
    with container:
        st.markdown(f"""
#### Please complete the verification process:

- Open the following URL: [{verification_url}]({verification_url})
- Enter the code: `{user_code}`

Once you have done that, click the **'Continue...'** button below.
        """)
        if st.button("Continue...", on_click=lambda: setattr(st.session_state, "oauth_verified", True)):
            pass

    # 如果已验证，清空容器并返回True
    if st.session_state.oauth_verified:
        container.empty()
        del st.session_state.oauth_verified
        return True
    return False


def get_video_transcript(url):
    try:
        buf = BytesIO()
        # 自定义验证逻辑，直到验证完成才继续
        def custom_oauth_flow(verification_url, user_code):
            while not handle_oauth_verification(verification_url, user_code):
                st.rerun()

        audio = YouTube(
            url,
            use_oauth=True,
            allow_oauth_cache=False,
            oauth_verifier=custom_oauth_flow,
        ).streams.get_audio_only()
        audio.stream_to_buffer(buf)
        buf.seek(0)
        with tempfile.NamedTemporaryFile(delete=True, suffix=".mp4") as temp_file:
            temp_file.write(buf.read())
            # TODO: 执行转录逻辑
            return "Lorem ipsum dolor..."
    except Exception as e:
        return f"Error: {e}"


# 初始化聊天历史
chat_history: list = st.session_state.setdefault(
    "chat_history",
    [{"type": "ai", "content": "Hello, how are you?"}],
)

# 渲染聊天历史
for message in chat_history:
    with st.chat_message(message["type"]):
        st.write(message["content"])

# 处理用户输入
if user_prompt := st.chat_input(placeholder="Type your message..."):
    # 添加用户消息到历史
    user_msg = {"type": "human", "content": user_prompt}
    chat_history.append(user_msg)
    with st.chat_message(user_msg["type"]):
        st.write(user_msg["content"])

    # 处理转录流程
    if "transcript_in_progress" not in st.session_state:
        st.session_state.transcript_in_progress = True

    if st.session_state.transcript_in_progress:
        with st.spinner("Loading..."):
            ai_response = get_video_transcript("https://youtu.be/ahq6MF349Fw")
            # 添加AI响应到历史
            ai_msg = {"type": "ai", "content": ai_response}
            chat_history.append(ai_msg)
            with st.chat_message(ai_msg["type"]):
                st.write(ai_msg["content"])
            # 重置流程状态
            del st.session_state.transcript_in_progress
            st.rerun()

关键修改点 #

1. 用Session State跟踪验证状态：通过st.session_state.oauth_verified记录用户是否完成OAuth验证，替代threading.Event。
2. 状态驱动的验证流程：在custom_oauth_flow中循环检查验证状态，未完成时调用st.rerun()触发脚本重运行，直到用户点击按钮完成验证。
3. 移除不必要的线程：直接在主脚本流程中处理转录，适配Streamlit运行模型。
4. 管理转录流程状态：用st.session_state.transcript_in_progress避免重复触发转录流程。

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