简介:基于Flet框架的智能聊天机器人自定义模板,旨在提供开箱即用的桌面端对话界面,适合需要接入大模型API并快速落地的开发者和产品人员。模板覆盖客户服务、教育辅助、个人助理、开发调试等多个应用场景,通过DeepSeek API实现自然语言响应,并重点针对流式输出与实时滚动页面做了优化,用户能在响应生成过程中看到逐字输出,避免长时间等待。资源包共3个文件,包含Python主程序、txt说明文档和gif效果动图,压缩后仅3.35MB,结构轻量清晰,便于阅读和二次修改。主程序中通过设置stream=True调用分段接口并逐步更新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_factory=time.time) meta: dict = field(default_factory=dict)参数说明:role决定气泡靠左还是靠右;template是自定义模板的唯一标识,默认给一个text模板;status在流式输出期间置为streaming,结束后改回done,这是后面流结束后重渲染的关键;meta用来带模板需要的附加信息,比如代码语言、表格行列、某条回复的耗时。有了这个模型,页面上存List[Message]而不是零散的dict,渲染函数就能做到“拿到Message,返回一组控件”。
这里有一个容易被忽略的点:meta是dict,所以它在dataclass里需要用field(default_factory=dict)而不是={}。否则所有Message实例共享同一个字典,一条消息写坏meta,其余的全被污染。这种错误在长对话里很难排查,因为症状是“某条历史消息突然多出了别的键”。
2.3 实时滚动页面用ListView还是Column:先看清这点再动手
聊天记录是动态增长的,必须用ft.ListView而不是ft.Column。Column会把所有子控件一次性布置完,消息超过几百条时不光是内存问题,Flutter端的布局计算也会明显变慢;ListView做虚拟化滚动,只构建可视区域内的项,这才是“实时滚动页面”的可靠底座。
| 维度 | ft.Column | ft.ListView |
|---|---|---|
| 子项构建 | 全部构建 | 可视区懒加载 |
| 长对话性能 | 快速劣化 | 基本平稳 |
| 自动滚到底部 | 手动算位置 | 自带auto_scroll |
| 滚动事件 | 不支持 | 有on_scroll |
| 适合场景 | 消息量小、不滚动的面板 | 聊天记录、日志流 |
第二个问题是auto_scroll的边界:auto_scroll=True确实会自动滚到底部,但它不懂“用户正在翻历史”。如果用户往上翻,再回来一条流式回复,页面会被强行拽到最底下,体验很糟。常见做法是监听on_scroll,当检测到用户方向上翻时,把auto_scroll关掉;等用户手动滚回底部,再重新打开。具体实现放在第三章。
有一个值得提前说明的细节:即便auto_scroll=True,也要在ListView上设置expand=True。否则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_url="http://127.0.0.1:8000/v1", api_key="not-needed") def chat_stream(messages): # 以OpenAI兼容接口为例,stream=True resp = client.chat.completions.create( model="qwen2.5-14b-instruct", messages=messages, stream=True, ) 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(expand=True, spacing=10, auto_scroll=True) input_field = ft.TextField(hint_text="输入问题", expand=True) async def send(e): text_ctrl = ft.Text("", selectable=True) list_view.controls.append(ft.Container( content=text_ctrl, bgcolor=ft.Colors.BLUE_GREY_200, border_radius=10, padding=10, )) 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_click=send)])) ft.app(main)代码说明:每次发送消息都同步追加一个文本控件,auto_scroll=True让ListView在新控件加入时自动滚到底部。expand=True是Flet布局里非常关键的一个参数,它告诉页面“这个控件占用剩余全部空间”,否则ListView会缩成内容高度,聊天区看不到滚动效果。
参数说明:
spacing=10:消息之间的垂直间距,单位是逻辑像素,按你的UI密度调,常用区间是4到12。selectable=True:让文本可选取。成品聊天界面至少要对assistant消息开这个,否则用户没法复制长代码。on_submit=send:在输入框按回车就发送,比点按钮顺手。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( content=ft.Text(msg.content, selectable=True), bgcolor=color, border_radius=12, padding=ft.padding.all(12), width=600 if msg.role == "assistant" else None, ) ], alignment=align, )逻辑说明:所有模板最终都返回控件。对外展示时,数据模型和控件树完全解耦,以后要加“消息时间戳”“复制按钮”,只改这个函数即可,不影响聊天循环里其他代码。
代码块模板会比文本多读一个字段:
@template("code") def render_code(msg: Message): lang = msg.meta.get("lang", "text") header = ft.Text(lang, size=12, color=ft.Colors.GREY_400) body = ft.Container( content=ft.Text( msg.content, font_family="monospace", selectable=True, color=ft.Colors.GREY_50, ), bgcolor=ft.Colors.GREY_900, border_radius=8, padding=12, width=720, ) return ft.Column([header, body], tight=True)这里meta里放lang,而不是塞进content里拼成```python ...,原因是为了避免模板函数再去解析文本。如果LLM已经习惯输出markdown代码块,也可以在拿到原始content后先用正则提取语言标记,再回填到meta,但那是另一层解析逻辑,建议放在流式结束后的重渲染阶段。
思维链折叠区用ft.ExpansionTile:
@template("thinking") def render_thinking(msg: Message): return ft.ExpansionTile( title=ft.Text("思考过程"), controls=[ft.Text(msg.content, size=13, color=ft.Colors.GREY_700)], initially_expanded=False, tile_padding=ft.padding.symmetric(vertical=4), )思维链是用户既想看到、又不该占主要面积的推理过程,折叠区刚好合适。initially_expanded按产品习惯设,我一般默认折叠,因为流式对话里用户最终要的是答案,不是推理过程。
4.3 新模板不靠if-else堆积:注册机制和ChatFeed封装
上面三个模板已经覆盖大多数场景。新增模板时,不需要动render_message本身,写一个新函数加@template("xxx")注册即可。在列表循环里统一调用render_message:
class ChatFeed(ft.Column): def __init__(self): super().__init__(expand=True, spacing=8) 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, interval=0.2, max_chunks=20): 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清干净,否则最后几个字会一直停留在缓冲区里。
本文还有配套的精品资源,点击获取