如何在函数调用中实现Textual TUI的流式实时更新？

问题根源 #

你的代码里，on_key方法中的循环直接调用了带有time.sleep的生成器，这会阻塞Textual的事件循环——Textual依赖事件循环处理界面更新、用户输入等操作，一旦被阻塞，所有界面渲染都会暂停，直到阻塞操作完成，所以最后只会显示最终的数字9。

解决方案：使用Textual的异步API #

Textual基于异步编程模型，需用异步方法处理耗时/流式操作，避免阻塞事件循环。

修改后的代码 #

import asyncio
from textual.app import App
from textual.widgets import Header, Label
from textual.reactive import reactive


async def async_gen():
    for i in range(10):
        await asyncio.sleep(0.1)
        yield str(i)


class Counter(Label):
    countervalue = reactive("Press Enter to start")

    def watch_countervalue(self, countervalue):
        self.update(countervalue)


class Minimal(App):
    def compose(self):
        yield Header()
        yield Counter(id="counter")

    async def on_key(self, event):
        if event.key == "enter":
            async for i in async_gen():
                self.query_one("#counter").countervalue = i  # pyright: ignore


if __name__ == "__main__":
    app = Minimal()
    app.run()

关键修改点 #

• 将生成器改为异步生成器：把time.sleep替换为asyncio.sleep，并给生成器函数加上async关键字
• 将on_key方法改为异步方法：添加async关键字，使用async for遍历异步生成器

这样每一次等待都不会阻塞事件循环，Textual可以在每次生成值后及时更新界面，实现流式显示的效果。

针对ChatGPT流式返回的适配 #

如果对接ChatGPT的流式接口，只需把异步生成器里的逻辑替换为调用OpenAI的异步API即可，示例框架如下：

async def chatgpt_stream(prompt):
    # 替换为OpenAI异步流式调用逻辑
    async for chunk in openai.ChatCompletion.acreate(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    ):
        content = chunk["choices"][0]["delta"].get("content", "")
        if content:
            yield content

然后在触发方法中用async for遍历这个异步生成器，实时更新界面内容。

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