简介基于Flet框架的智能聊天机器人自定义模板旨在提供开箱即用的桌面端对话界面适合需要接入大模型API并快速落地的开发者和产品人员。模板覆盖客户服务、教育辅助、个人助理、开发调试等多个应用场景通过DeepSeek API实现自然语言响应并重点针对流式输出与实时滚动页面做了优化用户能在响应生成过程中看到逐字输出避免长时间等待。资源包共3个文件包含Python主程序、txt说明文档和gif效果动图压缩后仅3.35MB结构轻量清晰便于阅读和二次修改。主程序中通过设置streamTrue调用分段接口并逐步更新UI聊天记录增多时页面自动滚动到最新消息交互体验流畅说明文档可帮助快速配置API密钥与运行环境动图则直观演示了实际聊天效果。目前已有83人学习使用基于该模板从零搭建一套带流式输出的智能对话工具只需替换API配置和提示词能够大幅节省界面开发时间。1. Flet做聊天机器人的第一个坎流式输出和滚动页面不是同一个问题写一个能聊天的Flet页面只需要几十行代码但把聊天做成真正能用的产品难点几乎都在界面层。大模型是流式吐字的而Flet默认的交互模型是改控件、再update两者的节奏对不上聊天记录会越来越长页面必须实时滚动但用户翻看历史时又不能让系统强行抢回滚动条每条消息的形态还不一样代码、表格、推理过程需要完全不同的控件排版。所以“Flet框架流式输出和实时滚动页面的智能聊天机器人自定义模板”这串词其实是三个独立问题加一个封装层流式刷新的节奏、滚动区的交互策略以及把消息类型映射成控件树的模板机制。下面按这条链拆开适合正在用Python搭LLM应用、但不满足于调完接口就交差的开发者。2. 从控件树到流式刷新Flet的更新模型与聊天页面选型2.1 为什么流式输出在Flet里不是print而是update循环先说底层模型。Flet的页面不是HTML加DOM它把Python侧的控件树同步到Flutter渲染端中间走WebSocket。你写page.add(...)只是改了Python侧的树真正让用户看到变化需要调用page.update()或page.update_async()。换句话说流式输出在你的代码里体现为一连串的update每收到一小段token就往文本控件里追加字符然后触发一次页面同步。常见做法是直接在事件循环里做async def stream_llm(text_ctrl, gen): for chunk in gen: text_ctrl.value chunk await text_ctrl.update_async()逻辑说明gen是LLM SDK返回的流式生成器每次滚出一个chunk通常是一到几十个token。text_ctrl是Flet的ft.Text控件value改了之后必须显式同步。这个过程本身不复杂难点在于同步频率的控制如果每个token都update_async一次WebSocket会被消息淹没页面明显变卡如果攒得太久又失去流式的“打字机”体验。如何取舍放到最后一章处理这里先用最小循环把链路跑通。参数说明update_async是异步版本必须放在async函数里如果回调本身不是异步的用update()也很等价但流式生成通常是异步IO统一走async更顺。chunk的类型取决于SDK有些返回对象而不是字符串记得先取delta.content或对应字段统一转成str再拼接。这个模型同时决定了“智能聊天机器人”的页面结构所有消息都必须挂在同一个可滚动容器里而不是东一个Text西一个Text。原因很直接每次流式刷新只更新当前消息对应的那个控件容器本身不能重复重建否则滚动位置会丢。2.2 消息模型用dict还是dataclass自定义模板的前置条件聊天页面必然要有消息列表。粗略有两种存法字典列表和对象列表。早期原型用dict最方便但当你开始做自定义模板“消息是什么”会越来越复杂它要有类型、状态、元数据还可能带渲染参数。用字典存这些很容易变成一大坨散键值改模板时要到处判断if code in msg。我一般会用dataclass把消息定义为显式模型。from dataclasses import dataclass, field import time dataclass class Message: role: str # user / assistant / system content: str # 当前展示的完整文本 template: str # 模板名决定渲染方式 status: str done # streaming / done / error created_at: float field(default_factorytime.time) meta: dict field(default_factorydict)参数说明role决定气泡靠左还是靠右template是自定义模板的唯一标识默认给一个text模板status在流式输出期间置为streaming结束后改回done这是后面流结束后重渲染的关键meta用来带模板需要的附加信息比如代码语言、表格行列、某条回复的耗时。有了这个模型页面上存List[Message]而不是零散的dict渲染函数就能做到“拿到Message返回一组控件”。这里有一个容易被忽略的点meta是dict所以它在dataclass里需要用field(default_factorydict)而不是{}。否则所有Message实例共享同一个字典一条消息写坏meta其余的全被污染。这种错误在长对话里很难排查因为症状是“某条历史消息突然多出了别的键”。2.3 实时滚动页面用ListView还是Column先看清这点再动手聊天记录是动态增长的必须用ft.ListView而不是ft.Column。Column会把所有子控件一次性布置完消息超过几百条时不光是内存问题Flutter端的布局计算也会明显变慢ListView做虚拟化滚动只构建可视区域内的项这才是“实时滚动页面”的可靠底座。维度ft.Columnft.ListView子项构建全部构建可视区懒加载长对话性能快速劣化基本平稳自动滚到底部手动算位置自带auto_scroll滚动事件不支持有on_scroll适合场景消息量小、不滚动的面板聊天记录、日志流第二个问题是auto_scroll的边界auto_scrollTrue确实会自动滚到底部但它不懂“用户正在翻历史”。如果用户往上翻再回来一条流式回复页面会被强行拽到最底下体验很糟。常见做法是监听on_scroll当检测到用户方向上翻时把auto_scroll关掉等用户手动滚回底部再重新打开。具体实现放在第三章。有一个值得提前说明的细节即便auto_scrollTrue也要在ListView上设置expandTrue。否则ListView会收缩成内容高度内容多了之后外层被撑出滚动条控件内部的滚动机制根本不触发auto_scroll也就失效了。3. 流式输出与实时滚动页面的最小实现3.1 搭一个异步聊天循环把LLM的token逐步落到页面上先给出一个能跑起来的最小骨架。下面的代码不绑定具体云厂商只要你的LLM服务提供OpenAI兼容的流式接口就能替换到chat_stream函数里。import flet as ft import asyncio from openai import OpenAI # 可换成任意OpenAI兼容SDK client OpenAI(base_urlhttp://127.0.0.1:8000/v1, api_keynot-needed) def chat_stream(messages): # 以OpenAI兼容接口为例streamTrue resp client.chat.completions.create( modelqwen2.5-14b-instruct, messagesmessages, streamTrue, ) for chunk in resp: delta chunk.choices[0].delta.content if delta: yield delta async def main(page: ft.Page): page.title Flet流式聊天 list_view ft.ListView(expandTrue, spacing10, auto_scrollTrue) input_field ft.TextField(hint_text输入问题, expandTrue) async def send(e): text_ctrl ft.Text(, selectableTrue) list_view.controls.append(ft.Container( contenttext_ctrl, bgcolorft.Colors.BLUE_GREY_200, border_radius10, padding10, )) await list_view.update_async() llm_messages [{role: user, content: input_field.value}] input_field.value await input_field.update_async() for token in chat_stream(llm_messages): text_ctrl.value token await text_ctrl.update_async() input_field.on_submit send await page.add_async(list_view, ft.Row([input_field, ft.IconButton(ft.Icons.SEND, on_clicksend)])) ft.app(main)代码说明每次发送消息都同步追加一个文本控件auto_scrollTrue让ListView在新控件加入时自动滚到底部。expandTrue是Flet布局里非常关键的一个参数它告诉页面“这个控件占用剩余全部空间”否则ListView会缩成内容高度聊天区看不到滚动效果。参数说明spacing10消息之间的垂直间距单位是逻辑像素按你的UI密度调常用区间是4到12。selectableTrue让文本可选取。成品聊天界面至少要对assistant消息开这个否则用户没法复制长代码。on_submitsend在输入框按回车就发送比点按钮顺手。border_radius和padding可以统一到消息模板里后面做自定义模板时会把这段抽出去。这个骨架已经能流式输出但有两个明显问题一是每个token都刷新消息长了卡顿二是用户手动上翻后新token仍把滚动条拽走。如果要在此基础上再接工具调用循环把chat_stream换成“调用工具→拿到结果→拼接上下文→继续生成”的循环这套流式渲染层同样适用界面部分完全不用改。3.2 用on_scroll状态机解决“抢滚动条”问题auto_scroll只能表达“要不要自动滚到底部”而用户意图是一个持续变化的状态上翻查看、回到底部、开始流式、继续跟踪。简单做法是在on_scroll回调里记录方向把它当作用户意图scroll_lock False def on_scroll(e): global scroll_lock if e.direction up and not scroll_lock: scroll_lock True list_view.auto_scroll False elif e.direction down: # 用户往下滚不一定滚到底部先不恢复 pass list_view.on_scroll on_scroll这里有个细节需要注意用户往下滚不一定到达底部所以不能一检测到direction down就立即把auto_scroll恢复。更稳的策略是配合滚动偏移判断但这部分在Flet里没有一个像浏览器scrollHeight那样直观的属性不同小版本API行为也有差异。我一般退一步用方向事件加一个小兜底只有当用户连续向下滚动并且下一次流式刷新前没有再次上翻时才恢复auto_scroll。具体到流式刷新把send里的循环改成检查锁async def append_token(text_ctrl, token): text_ctrl.value token if not scroll_lock: await list_view.update_async()这样做的好处是用户上翻后新token仍然持续写入控件的value只是不触发刷屏等他回到底部再手动调用一次list_view.update_async()把积压的内容一次性同步过去。这比直接禁止输出更合理因为用户上翻只是想看历史并不想中断生成。3.3 流式刷新不动的三类原因实际排障时“流式输出完全不显示”比“卡顿”更常见。先检查三个点现象常见原因检查动作界面一直空白无报错token拼接后没赋值回控件打印chunk内容确认不是空对象事件循环阻塞UI冻结同步阻塞操作卡住了Flet事件循环确认生成器里没有耗时CPU操作改用async SDK流结束后才一次性出现update_async没在循环内调用数一下代码里await ..._async()出现的次数特别注意第三类很多人会在for token in ...循环里忘记await text_ctrl.update_async()Flet的异步模式不会自动同步页面所有更新都积压在结束后的那次页面刷新里。这类问题不看报错因为逻辑没错纯粹是更新时机没对上。检查方法很直接在循环体里加一行print(text_ctrl.value[-20:])如果控制台在流式过程中不断有输出但界面不变那就是update没被调用。4. 把消息渲染成自定义模板从字符串拼接到组件工厂4.1 模板的本质消息类型到控件树的映射灰色文本块能聊天但远远不够。代码消息要等宽字体和高亮背景推理链要收进可折叠区域普通回答要更像气泡而不是一条长文本。所谓“自定义模板引擎”核心就一句话给定一条Message返回一个ft.Control或控件列表。它不需要你为此写一套DSL一个字典映射加几个渲染函数就足够。如果你做过短信模板那种“占位符加字符串替换”这里的思路完全不同。短信模板的渲染目标是纯文本把变量替换进去就结束Flet自定义模板的渲染目标是控件树占位符替换只能生成字符串到了控件层面还是要重新解析。所以不要试图把{{content}}这类模板语法直接搬过来正确的抽象是“组件工厂”消息类型作为键渲染函数作为值。4.2 三个高频模板文本气泡、代码块、思维链折叠区先做一个最小注册表用装饰器把模板名和渲染函数绑起来TEMPLATES {} def template(name): def deco(fn): TEMPLATES[name] fn return fn return deco def render_message(msg: Message) - ft.Control: fn TEMPLATES.get(msg.template, TEMPLATES[text]) return fn(msg)基座是文本气泡模板。以assistant消息为例template(text) def render_text(msg: Message): align ft.MainAxisAlignment.END if msg.role user else ft.MainAxisAlignment.START color ft.Colors.BLUE_100 if msg.role user else ft.Colors.GREY_100 return ft.Row( [ ft.Container( contentft.Text(msg.content, selectableTrue), bgcolorcolor, border_radius12, paddingft.padding.all(12), width600 if msg.role assistant else None, ) ], alignmentalign, )逻辑说明所有模板最终都返回控件。对外展示时数据模型和控件树完全解耦以后要加“消息时间戳”“复制按钮”只改这个函数即可不影响聊天循环里其他代码。代码块模板会比文本多读一个字段template(code) def render_code(msg: Message): lang msg.meta.get(lang, text) header ft.Text(lang, size12, colorft.Colors.GREY_400) body ft.Container( contentft.Text( msg.content, font_familymonospace, selectableTrue, colorft.Colors.GREY_50, ), bgcolorft.Colors.GREY_900, border_radius8, padding12, width720, ) return ft.Column([header, body], tightTrue)这里meta里放lang而不是塞进content里拼成python ...原因是为了避免模板函数再去解析文本。如果LLM已经习惯输出markdown代码块也可以在拿到原始content后先用正则提取语言标记再回填到meta但那是另一层解析逻辑建议放在流式结束后的重渲染阶段。思维链折叠区用ft.ExpansionTiletemplate(thinking) def render_thinking(msg: Message): return ft.ExpansionTile( titleft.Text(思考过程), controls[ft.Text(msg.content, size13, colorft.Colors.GREY_700)], initially_expandedFalse, tile_paddingft.padding.symmetric(vertical4), )思维链是用户既想看到、又不该占主要面积的推理过程折叠区刚好合适。initially_expanded按产品习惯设我一般默认折叠因为流式对话里用户最终要的是答案不是推理过程。4.3 新模板不靠if-else堆积注册机制和ChatFeed封装上面三个模板已经覆盖大多数场景。新增模板时不需要动render_message本身写一个新函数加template(xxx)注册即可。在列表循环里统一调用render_messageclass ChatFeed(ft.Column): def __init__(self): super().__init__(expandTrue, spacing8) self.messages: list[Message] [] def append(self, msg: Message): self.messages.append(msg) self.controls.append(render_message(msg)) self.update()有了这个ChatFeed聊天循环就只管生成Message不再关心控件构造。模板增删变成纯函数的注册行为调试时可以单独为某个模板写测试用例构造一条Message调render_message断言返回的控件类型。这比在几十个if-else里找渲染分支要清晰得多。需要注意ChatFeed继承了ft.Column所以self.update()是同步刷新。如果你的聊天循环跑在async环境里把update()换成await self.update_async()否则页面不会随消息追加而刷新。5. 流式模板的最后一公里缓冲刷新和流结束后的完整重渲染5.1 用TokenBuffer控制刷新频率token级刷新虽然实时但开销高。更稳的方案是“帧缓冲”把流式token先攒进一个Buffer每隔固定时间或攒够一定数量再刷新一次页面。有了缓冲自定义模板的流式展示阶段也能从“每token重排一次列表”降到“每帧重排一次”。class TokenBuffer: def __init__(self, text_ctrl, interval0.2, max_chunks20): self.text_ctrl text_ctrl self.interval interval self.max_chunks max_chunks self.buf [] self._task None async def push(self, token): self.buf.append(token) if len(self.buf) self.max_chunks and self._task is None: self._task asyncio.create_task(self._drain()) async def _drain(self): while self.buf: await asyncio.sleep(self.interval) self.text_ctrl.value .join(self.buf) self.buf.clear() await self.text_ctrl.update_async() self._task None async def flush(self): if self.buf: self.text_ctrl.value .join(self.buf) self.buf.clear() await self.text_ctrl.update_async()参数说明interval是每次刷新间隔单位秒0.2在桌面端和Web端都比较顺滑max_chunks是触发提前刷新的阈值避免单轮token生成太快时缓冲区积压过多。_task只保留一个drain任务因为drain循环本身会持续消费缓冲区多个任务并发会造成重复写入。5.2 流结束后重渲染完整模板流式过程中使用的通常是轻量模板只显示Text这样刷新成本最低。流结束后应该把该条Message从streaming状态切回done再从模板注册表重新取渲染函数替换ListView里那一个控件。这样最终页面才能带上时间戳、复制按钮、原始响应区这类附加信息。async def finalize_message(feed, msg, text_ctrl, list_view): msg.status done idx feed.controls.index(text_ctrl) feed.controls[idx] render_message(msg) await list_view.update_async()逻辑说明msg.status可以在render_message里被每个模板读取例如status done时才渲染时间戳和复制按钮streaming时只渲染文本。idx查找开销可以忽略单页聊天消息量远没到性能瓶颈。关键在于流式过程用一种轻模板结束后才切到重模板整个页面既流畅又不丢失最终信息。验证方法也简单把interval分别设为0.1、0.2、0.5在_drain里加一行print(f{time.time():.3f} flush)看日志里每次flush时间戳的间隔同时观察页面流畅度。经验上0.2秒一刷、一次攒20到50个token本地Web端和桌面端的体感差别最小。如果LLM端本身出字速度很慢interval没必要小于出字间隔否则每次drain只刷两三个字等于没缓冲如果LLM端暴快就把max_chunks调大避免drain里拼接长字符串拖慢重构。最后记得在流式生成器结束处手动调用await buffer.flush()把尾部残留的token清干净否则最后几个字会一直停留在缓冲区里。本文还有配套的精品资源点击获取
Flet智能聊天机器人：流式输出、实时滚动与自定义模板实战
简介基于Flet框架的智能聊天机器人自定义模板旨在提供开箱即用的桌面端对话界面适合需要接入大模型API并快速落地的开发者和产品人员。模板覆盖客户服务、教育辅助、个人助理、开发调试等多个应用场景通过DeepSeek API实现自然语言响应并重点针对流式输出与实时滚动页面做了优化用户能在响应生成过程中看到逐字输出避免长时间等待。资源包共3个文件包含Python主程序、txt说明文档和gif效果动图压缩后仅3.35MB结构轻量清晰便于阅读和二次修改。主程序中通过设置streamTrue调用分段接口并逐步更新UI聊天记录增多时页面自动滚动到最新消息交互体验流畅说明文档可帮助快速配置API密钥与运行环境动图则直观演示了实际聊天效果。目前已有83人学习使用基于该模板从零搭建一套带流式输出的智能对话工具只需替换API配置和提示词能够大幅节省界面开发时间。1. Flet做聊天机器人的第一个坎流式输出和滚动页面不是同一个问题写一个能聊天的Flet页面只需要几十行代码但把聊天做成真正能用的产品难点几乎都在界面层。大模型是流式吐字的而Flet默认的交互模型是改控件、再update两者的节奏对不上聊天记录会越来越长页面必须实时滚动但用户翻看历史时又不能让系统强行抢回滚动条每条消息的形态还不一样代码、表格、推理过程需要完全不同的控件排版。所以“Flet框架流式输出和实时滚动页面的智能聊天机器人自定义模板”这串词其实是三个独立问题加一个封装层流式刷新的节奏、滚动区的交互策略以及把消息类型映射成控件树的模板机制。下面按这条链拆开适合正在用Python搭LLM应用、但不满足于调完接口就交差的开发者。2. 从控件树到流式刷新Flet的更新模型与聊天页面选型2.1 为什么流式输出在Flet里不是print而是update循环先说底层模型。Flet的页面不是HTML加DOM它把Python侧的控件树同步到Flutter渲染端中间走WebSocket。你写page.add(...)只是改了Python侧的树真正让用户看到变化需要调用page.update()或page.update_async()。换句话说流式输出在你的代码里体现为一连串的update每收到一小段token就往文本控件里追加字符然后触发一次页面同步。常见做法是直接在事件循环里做async def stream_llm(text_ctrl, gen): for chunk in gen: text_ctrl.value chunk await text_ctrl.update_async()逻辑说明gen是LLM SDK返回的流式生成器每次滚出一个chunk通常是一到几十个token。text_ctrl是Flet的ft.Text控件value改了之后必须显式同步。这个过程本身不复杂难点在于同步频率的控制如果每个token都update_async一次WebSocket会被消息淹没页面明显变卡如果攒得太久又失去流式的“打字机”体验。如何取舍放到最后一章处理这里先用最小循环把链路跑通。参数说明update_async是异步版本必须放在async函数里如果回调本身不是异步的用update()也很等价但流式生成通常是异步IO统一走async更顺。chunk的类型取决于SDK有些返回对象而不是字符串记得先取delta.content或对应字段统一转成str再拼接。这个模型同时决定了“智能聊天机器人”的页面结构所有消息都必须挂在同一个可滚动容器里而不是东一个Text西一个Text。原因很直接每次流式刷新只更新当前消息对应的那个控件容器本身不能重复重建否则滚动位置会丢。2.2 消息模型用dict还是dataclass自定义模板的前置条件聊天页面必然要有消息列表。粗略有两种存法字典列表和对象列表。早期原型用dict最方便但当你开始做自定义模板“消息是什么”会越来越复杂它要有类型、状态、元数据还可能带渲染参数。用字典存这些很容易变成一大坨散键值改模板时要到处判断if code in msg。我一般会用dataclass把消息定义为显式模型。from dataclasses import dataclass, field import time dataclass class Message: role: str # user / assistant / system content: str # 当前展示的完整文本 template: str # 模板名决定渲染方式 status: str done # streaming / done / error created_at: float field(default_factorytime.time) meta: dict field(default_factorydict)参数说明role决定气泡靠左还是靠右template是自定义模板的唯一标识默认给一个text模板status在流式输出期间置为streaming结束后改回done这是后面流结束后重渲染的关键meta用来带模板需要的附加信息比如代码语言、表格行列、某条回复的耗时。有了这个模型页面上存List[Message]而不是零散的dict渲染函数就能做到“拿到Message返回一组控件”。这里有一个容易被忽略的点meta是dict所以它在dataclass里需要用field(default_factorydict)而不是{}。否则所有Message实例共享同一个字典一条消息写坏meta其余的全被污染。这种错误在长对话里很难排查因为症状是“某条历史消息突然多出了别的键”。2.3 实时滚动页面用ListView还是Column先看清这点再动手聊天记录是动态增长的必须用ft.ListView而不是ft.Column。Column会把所有子控件一次性布置完消息超过几百条时不光是内存问题Flutter端的布局计算也会明显变慢ListView做虚拟化滚动只构建可视区域内的项这才是“实时滚动页面”的可靠底座。维度ft.Columnft.ListView子项构建全部构建可视区懒加载长对话性能快速劣化基本平稳自动滚到底部手动算位置自带auto_scroll滚动事件不支持有on_scroll适合场景消息量小、不滚动的面板聊天记录、日志流第二个问题是auto_scroll的边界auto_scrollTrue确实会自动滚到底部但它不懂“用户正在翻历史”。如果用户往上翻再回来一条流式回复页面会被强行拽到最底下体验很糟。常见做法是监听on_scroll当检测到用户方向上翻时把auto_scroll关掉等用户手动滚回底部再重新打开。具体实现放在第三章。有一个值得提前说明的细节即便auto_scrollTrue也要在ListView上设置expandTrue。否则ListView会收缩成内容高度内容多了之后外层被撑出滚动条控件内部的滚动机制根本不触发auto_scroll也就失效了。3. 流式输出与实时滚动页面的最小实现3.1 搭一个异步聊天循环把LLM的token逐步落到页面上先给出一个能跑起来的最小骨架。下面的代码不绑定具体云厂商只要你的LLM服务提供OpenAI兼容的流式接口就能替换到chat_stream函数里。import flet as ft import asyncio from openai import OpenAI # 可换成任意OpenAI兼容SDK client OpenAI(base_urlhttp://127.0.0.1:8000/v1, api_keynot-needed) def chat_stream(messages): # 以OpenAI兼容接口为例streamTrue resp client.chat.completions.create( modelqwen2.5-14b-instruct, messagesmessages, streamTrue, ) for chunk in resp: delta chunk.choices[0].delta.content if delta: yield delta async def main(page: ft.Page): page.title Flet流式聊天 list_view ft.ListView(expandTrue, spacing10, auto_scrollTrue) input_field ft.TextField(hint_text输入问题, expandTrue) async def send(e): text_ctrl ft.Text(, selectableTrue) list_view.controls.append(ft.Container( contenttext_ctrl, bgcolorft.Colors.BLUE_GREY_200, border_radius10, padding10, )) await list_view.update_async() llm_messages [{role: user, content: input_field.value}] input_field.value await input_field.update_async() for token in chat_stream(llm_messages): text_ctrl.value token await text_ctrl.update_async() input_field.on_submit send await page.add_async(list_view, ft.Row([input_field, ft.IconButton(ft.Icons.SEND, on_clicksend)])) ft.app(main)代码说明每次发送消息都同步追加一个文本控件auto_scrollTrue让ListView在新控件加入时自动滚到底部。expandTrue是Flet布局里非常关键的一个参数它告诉页面“这个控件占用剩余全部空间”否则ListView会缩成内容高度聊天区看不到滚动效果。参数说明spacing10消息之间的垂直间距单位是逻辑像素按你的UI密度调常用区间是4到12。selectableTrue让文本可选取。成品聊天界面至少要对assistant消息开这个否则用户没法复制长代码。on_submitsend在输入框按回车就发送比点按钮顺手。border_radius和padding可以统一到消息模板里后面做自定义模板时会把这段抽出去。这个骨架已经能流式输出但有两个明显问题一是每个token都刷新消息长了卡顿二是用户手动上翻后新token仍把滚动条拽走。如果要在此基础上再接工具调用循环把chat_stream换成“调用工具→拿到结果→拼接上下文→继续生成”的循环这套流式渲染层同样适用界面部分完全不用改。3.2 用on_scroll状态机解决“抢滚动条”问题auto_scroll只能表达“要不要自动滚到底部”而用户意图是一个持续变化的状态上翻查看、回到底部、开始流式、继续跟踪。简单做法是在on_scroll回调里记录方向把它当作用户意图scroll_lock False def on_scroll(e): global scroll_lock if e.direction up and not scroll_lock: scroll_lock True list_view.auto_scroll False elif e.direction down: # 用户往下滚不一定滚到底部先不恢复 pass list_view.on_scroll on_scroll这里有个细节需要注意用户往下滚不一定到达底部所以不能一检测到direction down就立即把auto_scroll恢复。更稳的策略是配合滚动偏移判断但这部分在Flet里没有一个像浏览器scrollHeight那样直观的属性不同小版本API行为也有差异。我一般退一步用方向事件加一个小兜底只有当用户连续向下滚动并且下一次流式刷新前没有再次上翻时才恢复auto_scroll。具体到流式刷新把send里的循环改成检查锁async def append_token(text_ctrl, token): text_ctrl.value token if not scroll_lock: await list_view.update_async()这样做的好处是用户上翻后新token仍然持续写入控件的value只是不触发刷屏等他回到底部再手动调用一次list_view.update_async()把积压的内容一次性同步过去。这比直接禁止输出更合理因为用户上翻只是想看历史并不想中断生成。3.3 流式刷新不动的三类原因实际排障时“流式输出完全不显示”比“卡顿”更常见。先检查三个点现象常见原因检查动作界面一直空白无报错token拼接后没赋值回控件打印chunk内容确认不是空对象事件循环阻塞UI冻结同步阻塞操作卡住了Flet事件循环确认生成器里没有耗时CPU操作改用async SDK流结束后才一次性出现update_async没在循环内调用数一下代码里await ..._async()出现的次数特别注意第三类很多人会在for token in ...循环里忘记await text_ctrl.update_async()Flet的异步模式不会自动同步页面所有更新都积压在结束后的那次页面刷新里。这类问题不看报错因为逻辑没错纯粹是更新时机没对上。检查方法很直接在循环体里加一行print(text_ctrl.value[-20:])如果控制台在流式过程中不断有输出但界面不变那就是update没被调用。4. 把消息渲染成自定义模板从字符串拼接到组件工厂4.1 模板的本质消息类型到控件树的映射灰色文本块能聊天但远远不够。代码消息要等宽字体和高亮背景推理链要收进可折叠区域普通回答要更像气泡而不是一条长文本。所谓“自定义模板引擎”核心就一句话给定一条Message返回一个ft.Control或控件列表。它不需要你为此写一套DSL一个字典映射加几个渲染函数就足够。如果你做过短信模板那种“占位符加字符串替换”这里的思路完全不同。短信模板的渲染目标是纯文本把变量替换进去就结束Flet自定义模板的渲染目标是控件树占位符替换只能生成字符串到了控件层面还是要重新解析。所以不要试图把{{content}}这类模板语法直接搬过来正确的抽象是“组件工厂”消息类型作为键渲染函数作为值。4.2 三个高频模板文本气泡、代码块、思维链折叠区先做一个最小注册表用装饰器把模板名和渲染函数绑起来TEMPLATES {} def template(name): def deco(fn): TEMPLATES[name] fn return fn return deco def render_message(msg: Message) - ft.Control: fn TEMPLATES.get(msg.template, TEMPLATES[text]) return fn(msg)基座是文本气泡模板。以assistant消息为例template(text) def render_text(msg: Message): align ft.MainAxisAlignment.END if msg.role user else ft.MainAxisAlignment.START color ft.Colors.BLUE_100 if msg.role user else ft.Colors.GREY_100 return ft.Row( [ ft.Container( contentft.Text(msg.content, selectableTrue), bgcolorcolor, border_radius12, paddingft.padding.all(12), width600 if msg.role assistant else None, ) ], alignmentalign, )逻辑说明所有模板最终都返回控件。对外展示时数据模型和控件树完全解耦以后要加“消息时间戳”“复制按钮”只改这个函数即可不影响聊天循环里其他代码。代码块模板会比文本多读一个字段template(code) def render_code(msg: Message): lang msg.meta.get(lang, text) header ft.Text(lang, size12, colorft.Colors.GREY_400) body ft.Container( contentft.Text( msg.content, font_familymonospace, selectableTrue, colorft.Colors.GREY_50, ), bgcolorft.Colors.GREY_900, border_radius8, padding12, width720, ) return ft.Column([header, body], tightTrue)这里meta里放lang而不是塞进content里拼成python ...原因是为了避免模板函数再去解析文本。如果LLM已经习惯输出markdown代码块也可以在拿到原始content后先用正则提取语言标记再回填到meta但那是另一层解析逻辑建议放在流式结束后的重渲染阶段。思维链折叠区用ft.ExpansionTiletemplate(thinking) def render_thinking(msg: Message): return ft.ExpansionTile( titleft.Text(思考过程), controls[ft.Text(msg.content, size13, colorft.Colors.GREY_700)], initially_expandedFalse, tile_paddingft.padding.symmetric(vertical4), )思维链是用户既想看到、又不该占主要面积的推理过程折叠区刚好合适。initially_expanded按产品习惯设我一般默认折叠因为流式对话里用户最终要的是答案不是推理过程。4.3 新模板不靠if-else堆积注册机制和ChatFeed封装上面三个模板已经覆盖大多数场景。新增模板时不需要动render_message本身写一个新函数加template(xxx)注册即可。在列表循环里统一调用render_messageclass ChatFeed(ft.Column): def __init__(self): super().__init__(expandTrue, spacing8) self.messages: list[Message] [] def append(self, msg: Message): self.messages.append(msg) self.controls.append(render_message(msg)) self.update()有了这个ChatFeed聊天循环就只管生成Message不再关心控件构造。模板增删变成纯函数的注册行为调试时可以单独为某个模板写测试用例构造一条Message调render_message断言返回的控件类型。这比在几十个if-else里找渲染分支要清晰得多。需要注意ChatFeed继承了ft.Column所以self.update()是同步刷新。如果你的聊天循环跑在async环境里把update()换成await self.update_async()否则页面不会随消息追加而刷新。5. 流式模板的最后一公里缓冲刷新和流结束后的完整重渲染5.1 用TokenBuffer控制刷新频率token级刷新虽然实时但开销高。更稳的方案是“帧缓冲”把流式token先攒进一个Buffer每隔固定时间或攒够一定数量再刷新一次页面。有了缓冲自定义模板的流式展示阶段也能从“每token重排一次列表”降到“每帧重排一次”。class TokenBuffer: def __init__(self, text_ctrl, interval0.2, max_chunks20): self.text_ctrl text_ctrl self.interval interval self.max_chunks max_chunks self.buf [] self._task None async def push(self, token): self.buf.append(token) if len(self.buf) self.max_chunks and self._task is None: self._task asyncio.create_task(self._drain()) async def _drain(self): while self.buf: await asyncio.sleep(self.interval) self.text_ctrl.value .join(self.buf) self.buf.clear() await self.text_ctrl.update_async() self._task None async def flush(self): if self.buf: self.text_ctrl.value .join(self.buf) self.buf.clear() await self.text_ctrl.update_async()参数说明interval是每次刷新间隔单位秒0.2在桌面端和Web端都比较顺滑max_chunks是触发提前刷新的阈值避免单轮token生成太快时缓冲区积压过多。_task只保留一个drain任务因为drain循环本身会持续消费缓冲区多个任务并发会造成重复写入。5.2 流结束后重渲染完整模板流式过程中使用的通常是轻量模板只显示Text这样刷新成本最低。流结束后应该把该条Message从streaming状态切回done再从模板注册表重新取渲染函数替换ListView里那一个控件。这样最终页面才能带上时间戳、复制按钮、原始响应区这类附加信息。async def finalize_message(feed, msg, text_ctrl, list_view): msg.status done idx feed.controls.index(text_ctrl) feed.controls[idx] render_message(msg) await list_view.update_async()逻辑说明msg.status可以在render_message里被每个模板读取例如status done时才渲染时间戳和复制按钮streaming时只渲染文本。idx查找开销可以忽略单页聊天消息量远没到性能瓶颈。关键在于流式过程用一种轻模板结束后才切到重模板整个页面既流畅又不丢失最终信息。验证方法也简单把interval分别设为0.1、0.2、0.5在_drain里加一行print(f{time.time():.3f} flush)看日志里每次flush时间戳的间隔同时观察页面流畅度。经验上0.2秒一刷、一次攒20到50个token本地Web端和桌面端的体感差别最小。如果LLM端本身出字速度很慢interval没必要小于出字间隔否则每次drain只刷两三个字等于没缓冲如果LLM端暴快就把max_chunks调大避免drain里拼接长字符串拖慢重构。最后记得在流式生成器结束处手动调用await buffer.flush()把尾部残留的token清干净否则最后几个字会一直停留在缓冲区里。本文还有配套的精品资源点击获取