Gradio+Colab模型推理时周期性更新文本框报错的问题排查

问题场景 #

在Colab中使用Gradio实现模型推理，核心逻辑：

• 输入文本变化时触发test函数，该函数通过线程更新全局变量label
• 使用demo.load()设置every=1，让updatePredLabel每秒执行一次，同步label到输出文本框

复现代码 #

import gradio as gr
import time
import threading
from threading import Lock

lock = threading.Lock()

label = ""
def test(x):
  global label
  global lock
  print("Text change {}".format(x))
  time.sleep(5)
  lock.acquire()
  label = x
  lock.release()
  return

def updatePredLabel():
  global label
  time.sleep(0.5)

  lock.acquire()
  out_text = label
  lock.release()

  return out_text


with gr.Blocks() as demo:
  with gr.Row():
    input = gr.Textbox()
    out = gr.Textbox()
  input.change(test, input, None) 
  demo.load(updatePredLabel, None, out, every=1)   

demo.queue().launch(debug=True)

报错信息 #

ERROR:asyncio:Task exception was never retrieved
future: <Task finished coro=<Queue.process_events() done, defined at /usr/local/lib/python3.7/dist-packages/gradio/queue.py:265> exception=ValueError('[<gradio.queue.Event object at 0x7f7de35f0350>] is not in list')>
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/dist-packages/gradio/queue.py", line 341, in process_events
    self.active_jobs[self.active_jobs.index(events)] = None
ValueError: [<gradio.queue.Event object at 0x7f7de35f0350>] is not in list

问题原因 #

这个错误源于Gradio队列机制与周期性任务的兼容性冲突：

• demo.load(..., every=1)创建的周期性后台任务会持续向队列提交事件，在Colab的线程调度环境下，容易出现事件重复移除或找不到对应事件的情况
• 原代码中test函数未启动独立线程，直接在Gradio事件线程中执行time.sleep(5)，阻塞了事件处理流程，进一步干扰队列正常运行

调试方法 #

• 关闭队列验证：移除demo.queue()，直接调用demo.launch(debug=True)，确认错误是否由队列与周期性任务的冲突导致
• 打印队列状态：在updatePredLabel中添加队列状态打印（如print(demo.queue._active_jobs)，注意不同Gradio版本属性可能有差异），观察事件的添加与移除情况
• 逐步简化逻辑：先移除线程锁和test函数的time.sleep，再逐步恢复代码，定位触发错误的具体环节

替代实现方案 #

方案1：使用Gradio官方gr.Timer组件（Gradio 3.10+支持） #

gr.Timer是官方推荐的周期性任务组件，比demo.load(every=...)更稳定：

import gradio as gr
import time
import threading
from threading import Lock

lock = threading.Lock()
label = ""

def test(x):
    global label
    print("Text change {}".format(x))
    # 启动独立线程执行耗时操作，避免阻塞UI线程
    def update_label_task():
        time.sleep(5)
        with lock:
            label = x
    threading.Thread(target=update_label_task).start()

def updatePredLabel():
    with lock:
        return label

with gr.Blocks() as demo:
    with gr.Row():
        input_box = gr.Textbox(label="输入")
        out_box = gr.Textbox(label="输出")
    input_box.change(test, input_box, None)
    # 每秒触发一次更新
    gr.Timer(1).tick(updatePredLabel, None, out_box)

demo.queue().launch(debug=True)

方案2：后台线程主动更新组件 #

通过后台线程直接调用组件的update方法，不依赖队列周期性任务：

import gradio as gr
import time
import threading
from threading import Lock

lock = threading.Lock()
label = ""
stop_flag = threading.Event()

def test(x):
    global label
    print("Text change {}".format(x))
    def update_label_task():
        time.sleep(5)
        with lock:
            label = x
    threading.Thread(target=update_label_task).start()

def background_update_loop(out_box):
    while not stop_flag.is_set():
        time.sleep(1)
        with lock:
            current_content = label
        out_box.update(value=current_content)

with gr.Blocks() as demo:
    with gr.Row():
        input_box = gr.Textbox(label="输入")
        out_box = gr.Textbox(label="输出")
    input_box.change(test, input_box, None)
    
    # 启动后台更新线程（守护线程随程序退出自动结束）
    demo.load(lambda: threading.Thread(target=background_update_loop, args=(out_box,), daemon=True).start())

demo.launch(debug=True)

关键优化点 #

• 原test函数未启动独立线程，导致阻塞UI，现在改为在新线程中执行耗时操作
• 使用with lock:上下文管理器替代手动acquire()/release()，避免死锁风险
• 优先采用官方推荐组件，减少自定义逻辑与Gradio内部机制的冲突

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