如何用Textualize和Rich库构建支持异步流式响应的终端AI助手？

当AI模型以流式方式返回token时，直观做法是每收到一个片段就调用print函数输出到终端。这种做法在简单测试中似乎可用，但实际存在三个问题。终端光标位置会随着输出不断变化，如果输出内容包含换行或超出窗口宽度，画面会出现跳跃；Rich等库已经绘制的表格或边框会被新行打乱；异步任务与打印顺序之间难以保证一致，多个并发输出会互相穿插。

Textualize将终端视为一个可更新的画布，所有组件都基于布局树渲染。流式响应可以触发组件内部状态变化，由框架统一计算差异并只更新变化区域，这样既保证界面稳定，又避免了整屏重绘。对于AI助手场景，我们可以把模型输出放入一个可变的Text或静态区域，通过修改组件内容触发刷新。

Rich的Live类提供了类似动态更新的能力，适合在非Textualize应用中使用。它通过ANSI控制序列控制光标位置，在刷新时恢复上一帧并重新绘制。Textualize内部同样依赖Rich处理颜色与样式，但提供了完整的事件循环和消息系统，更适合构建多区域交互界面。

Textualize与Rich的核心组件和异步机制

Textualize应用继承自App类，通过compose方法声明UI结构。常用的组件包括Header、Footer、Input、Static以及RichLog。其中RichLog适合展示不断追加的日志内容，但它默认是滚动缓冲，不一定适合需要整体替换的AI回复区域。这里使用Static配合自定义文本属性，可以自由控制显示内容。

异步流式接口通常返回AsyncIterator[str]。在Textualize中，可以用@work装饰器或者call_after_refresh启动后台任务。更推荐使用asyncio.create_task配合消息回调：当收到模型片段时，通过self.query_one找到目标组件并调用update方法。由于UI操作必须发生在主线程，需要借助self.call_from_thread或事件循环的线程安全机制，如果全部是异步代码则可以直接在事件循环中更新。

Rich提供的Panel、Text和Style允许构建美观的对话气泡。例如，将用户输入和模型回复分别用不同颜色的面板包裹，设置圆角边框，并使用Text的justify和overflow属性控制排版。Textualize组件内部可以接收Rich渲染对象，因此可以直接把Panel(Text(...))赋给Static的renderable属性。

实现异步流式响应显示的完整代码

下面给出一个精简但可运行的终端AI问答助手。程序使用asyncio模拟流式接口，每0.05秒返回一个单词。实际项目中可以将该函数替换为调用OpenAI或本地模型的流式API。

import asyncio
from textual.app import App, ComposeResult
from textual.containers import VerticalScroll
from textual.widgets import Header, Footer, Input, Static
from rich.panel import Panel
from rich.text import Text

async def fake_stream_response(prompt: str):
    """模拟AI流式输出，实际可替换为模型流式接口"""
    words = ["这是", "一段", "由", "异步", "流式", "接口", "逐步", "返回", "的", "回答", "内容。"]
    for word in words:
        await asyncio.sleep(0.05)
        yield word

class AIAssistantApp(App):
    CSS = """
    #response-box {
        height: auto;
        min-height: 8;
        border: none;
    }
    """
    def compose(self) -> ComposeResult:
        yield Header()
        yield VerticalScroll(Static("等待输入...", id="response-box"))
        yield Input(placeholder="输入问题后按回车")
        yield Footer()

    async def on_input_submitted(self, event: Input.Submitted) -> None:
        prompt = event.value.strip()
        if not prompt:
            return
        event.input.value = ""
        response_area = self.query_one("#response-box", Static)
        panel = Panel(
            Text("思考中...", style="yellow"),
            title="AI回复",
            border_style="cyan",
            padding=(1, 2),
        )
        response_area.update(panel)

        chunks = []
        stream = fake_stream_response(prompt)
        async for chunk in stream:
            chunks.append(chunk)
            current_text = "".join(chunks)
            text = Text(current_text, style="white")
            text.no_wrap = False
            panel = Panel(
                text,
                title="AI回复",
                border_style="cyan",
                padding=(1, 2),
            )
            response_area.update(panel)

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

代码中on_input_submitted是Textualize内置的输入提交事件处理函数，事件循环会自动调度。流式循环每收到一个片段就更新面板，框架会局部刷新响应区域，而不是整屏清空。Rich的Text对象设置no_wrap为False后，长文本会在面板宽度内自动换行，避免超出终端边界。

实际接入模型API时，需要将fake_stream_response改为真正的异步生成器。例如很多大模型SDK提供async for chunk in client.chat.stream(...)形式的接口，直接在循环中提取delta内容即可。注意部分接口返回的是完整消息而不是增量，需要根据具体SDK做去重或拼接处理。

处理性能与交互细节的常见问题

流式输出速度通常很快，频繁更新组件可能带来性能压力。Textualize会合并同一帧内的多次更新，但仍建议在更新前做节流，例如每50毫秒或积累一定字数后再刷新。可以使用time.monotonic()记录上次刷新时间，或者使用asyncio.sleep让出事件循环。此外，如果模型返回速度远快于渲染速度，缓冲区会持续增长，可以考虑在后台任务中暂存内容，通过定时器批量写入组件。

另一个问题是光标闪烁。终端应用在动态刷新时，光标可能停留在输入框或响应区域。可以使用Input的focus事件管理，在流式输出期间将焦点保持在输入框，避免用户按键被吞掉。Textualize的Footer会显示可用快捷键，默认布局对演示已经足够，正式工具可能需要自定义CSS隐藏不需要的元素。

颜色方面，Rich的Style支持RGB和终端主题色。在黑色背景和白色背景的终端上都应测试可读性。如果目标用户可能使用Windows命令提示符，需要对旧版终端做降级处理，例如关闭圆角边框和复杂样式，只保留基本颜色。Textualize默认会检测终端能力，必要时可以显式设置color_system。

还有错误处理：流式中途断开会留下半截回答，应该在try/finally中恢复界面状态，并在except分支中给出明确提示。对于长回复，可以在滚动容器中限制最大高度，将旧内容上移而不是无限增长。通过组合VerticalScroll与RichLog可以分别处理多轮对话和单条长回复。

TextualizeRich库异步流式响应修改时间：2026-08-27 11:13:56