在做 LLM 工具类项目时我经常遇到一个很现实的场景业务同学在同一套 GUI 工作台里反复提交相似的文本每次都触发一次大模型调用。虽然没有报错但成本、响应时长、接口抖动都在成倍累积。后来我调整了设计思路——让用户第一次通过界面完整体验工作流然后把这次执行结果缓存下来以后再遇到相同输入直接从缓存重放让 LLM 调用次数变成 0。这个思路用 Reflex 来实现非常简单而且非常适合沉淀成团队内部工具。本文会用 Reflex 完整实现一个“GUI 工作流一次演示、零调用重放”的示例。文章覆盖核心概念、环境准备、代码实现、运行验证、常见问题与工程化建议。新手可以从中理解 Reflex 应用的基本写法有经验的开发者可以直接把缓存重放模块抽出来复用到自己的项目中。1. 背景与核心概念1.1 一个很常见的重复调用场景假设你给运营团队做了一个文本处理工具里面有一个“翻译”功能。运营同学每天都会把相同或相似的标题、评论、商品描述粘贴进来点击“翻译”按钮。如果每一次点击都直接请求大模型接口会发生三件事每一笔请求都在消耗 token 费用哪怕上次用相同文本跑过一模一样的翻译。用户的等待时间取决于大模型接口的响应速度而不是本地处理速度。如果你的 GUI 工具承担着多人同时使用的流量后端和大模型 API 之间的并发压力会明显增大。这种问题在 LLM 应用里非常常见。大模型很适合处理“不确定性问题”但如果你只是要复现一个已经做过的固定结果完全没必要再让模型跑一次。1.2 Reflex 能做什么Reflex 是一个用纯 Python 开发全栈 Web 应用的框架。你不需要写 HTML、CSS、JavaScript而是用 Python 代码描述页面组件、样式和事件逻辑。前端界面由 Reflex 自动编译生成后端逻辑运行在 Python 进程里。它非常适合做以下事情内部数据管理后台。算法实验平台。低代码 workflow 配置界面。接入了大模型 API 的团队工具。相比传统的前后端分离开发Reflex 最大的优势是只要你会 Python就能快速做出一个可交互的 GUI 应用。对于“写一个工作流界面再完善一套执行逻辑”这种需求效率会高很多。1.3 工作流重放与零 LLM 调用的定义先明确一下本文讨论的“工作流”是什么。用户通过 GUI 输入参数点击按钮后系统按照固定步骤执行最后返回结果。比如输入一段商品文案。选择任务类型摘要、翻译或关键词提取。系统调用 LLM 接口获得结果。系统把结果展示在界面上。这一整串过程就是一条 GUI 工作流。“工作流重放”的意思是当用户第二次提交一模一样的输入时系统不必重新执行这条工作流中的昂贵步骤而是直接读取上一次记录下来的结果并展示。“零 LLM 调用”指的就是针对重复输入不会再去请求大模型接口。这不是绕过模型而是通过缓存和重放机制避免重复计算相同内容。2. Reflex 工作流重放核心原理解析2.1 Reflex 的状态、事件与组件在写代码之前需要先理解 Reflex 应用的三要素。组件Component用来描述界面。组件之间可以嵌套类似 HTML 标签但用 Python 函数来表示。比如rx.text()表示文本rx.button()表示按钮rx.text_area()表示多行输入框。状态State用来保存应用数据。你可以在 State 类中定义变量这些变量会被自动同步到前端。比如用户输入的文本、选择的任务类型、执行结果、日志列表都可以作为 State 变量保存。事件Event是用户操作触发的逻辑。比如用户点击按钮会调用 State 中定义的run_workflow()方法。事件方法内部可以修改 State 变量修改之后界面会自动刷新。理解这三者的关系是后续代码的基础。2.2 工作流缓存与重放机制我们可以在 LLM 服务前面加一层缓存模块。整体流程如下用户输入参数 ↓ 计算输入指纹 ↓ 查询缓存 ↓ 命中缓存——是——→ 读取历史结果 → 界面展示零 LLM 调用 ↓ 否 调用 LLM 服务 ↓ 写入缓存 ↓ 界面展示这里缓存的不只是 LLM 返回的裸响应而是整条工作流的结果。也就是说只要输入参数相同无论工作流中间有多少个步骤都直接复用最终结果。这样做的好处是响应时间更快。API 成本更低。工作流执行过程更容易审计。在更复杂的工作流场景中你还可以把每一步的中间结果都记录下来形成“执行轨迹”。重放时不只展示最终结果还能还原每一步的细节。2.3 为什么用 MD5 作为输入指纹为了判断“用户输入是否相同”最简单的方式是把参数串联成一个字符串然后计算哈希值。本文示例使用 MD5 作为指纹计算方式。import hashlib def build_key(task: str, text: str) - str: raw f{task}::{text} return hashlib.md5(raw.encode(utf-8)).hexdigest()这里需要注意几个点MD5 不是加密算法它只是为了生成一个固定长度的索引值。我们不需要把它用于安全场景。task和text之前的连接符号要保留避免task摘要 textA和task摘 text要A这类组合产生相同指纹。如果输入文本包含换行或空格差异建议先做规范化处理否则缓存命中率会降低。3. 环境准备与项目初始化3.1 环境要求在开始编写代码之前需要准备好以下环境Python 3.10 及以上版本。pip 包管理工具。Node.js 18 及以上版本。Reflex 在编译前端资源时需要 Node.js如果本机没有安装启动时会报类似node.js not found的错误。现代浏览器推荐 Chrome 或 Edge。不同 Reflex 版本文档可能略有差异本文示例以常见版本为准重点演示完整实现思路。如果你的 Reflex 版本较新个别组件参数可能有所调整按官方文档微调即可。3.2 安装 Reflex使用 pip 安装pip install reflex安装之后可以执行以下命令验证是否安装成功reflex --version如果你本机同时有多个 Python 环境建议先创建虚拟环境再安装python -m venv .venv source .venv/bin/activate pip install reflex3.3 初始化项目结构创建一个项目目录mkdir reflex_workflow_demo cd reflex_workflow_demo reflex initreflex init会生成基础项目结构包括rxconfig.py、assets目录和默认示例代码。我们可以在此基础上替换成自己的实现。最终示例项目结构如下reflex_workflow_demo/ ├── rxconfig.py ├── workflow_cache.py ├── llm_service.py └── my_llm_workflow.py说明rxconfig.pyReflex 项目的配置文件。workflow_cache.py缓存与重放模块。llm_service.pyLLM 服务适配层。my_llm_workflow.pyReflex 主应用包含状态、事件和页面组件。4. 完整实战一次演示、零调用重放下面我们按照模块拆分逐步实现整个功能。4.1 编写缓存模块 workflow_cache.py缓存模块负责计算输入指纹、读取缓存、写入缓存。这里使用 JSON 文件作为存储方便读者直接查看缓存内容。生产环境中可以替换为 Redis 或数据库。文件路径workflow_cache.pyimport hashlib import json import time from pathlib import Path from typing import Optional CACHE_FILE Path(workflow_cache.json) def build_key(task: str, text: str) - str: raw f{task}::{text} return hashlib.md5(raw.encode(utf-8)).hexdigest() def load_cache() - dict: if CACHE_FILE.exists(): try: return json.loads(CACHE_FILE.read_text(encodingutf-8)) except json.JSONDecodeError: return {} return {} def get_cache(task: str, text: str) - Optional[dict]: cache load_cache() return cache.get(build_key(task, text)) def save_cache(task: str, text: str, result: str) - None: cache load_cache() key build_key(task, text) cache[key] { task: task, text: text, result: result, created_at: time.time(), } CACHE_FILE.write_text( json.dumps(cache, ensure_asciiFalse, indent2), encodingutf-8, )这段代码做了三件事build_key()根据任务类型和输入文本生成唯一的缓存键。get_cache()读取 JSON 文件并尝试获取缓存。如果文件不存在或文件格式损坏返回空字典。save_cache()把工作流执行结果写入 JSON 文件。这里使用ensure_asciiFalse确保中文内容在 JSON 文件中可读方便排查问题。4.2 编写 LLM 服务 llm_service.py为了让示例可以直接运行我写了一个模拟 LLM 调用函数。你可以把这个函数替换成 OpenAI、DeepSeek 或其他大模型 SDK 的真实调用。文件路径llm_service.pydef call_llm(task: str, text: str) - str: 模拟一次 LLM 调用。生产环境可替换为真实大模型接口。 # 真实接入示例 # from openai import OpenAI # client OpenAI() # response client.chat.completions.create( # modelgpt-4o-mini, # messages[ # {role: system, content: 你是工作流助手。}, # {role: user, content: f任务{task}\n文本{text}}, # ], # ) # return response.choices[0].message.content if task 翻译: return f[模拟翻译] {text[:50]} if task 关键词提取: return GUI, Workflow, LLM, Cache, Replay return f[模拟摘要] {text[:80]}...这个模拟函数的核心作用是让没有真实 API Key 的读者也能跑通整个流程。让“第一次调用 LLM、第二次零调用重放”这个过程变得可观察。在真实项目中你只需要改call_llm()函数内部实现不需要改动 GUI 和缓存逻辑。4.3 编写 Reflex 主应用 my_llm_workflow.py主应用是整个页面的核心包含 State 状态类、事件处理方法和页面组件。文件路径my_llm_workflow.pyimport time import reflex as rx from workflow_cache import build_key, get_cache, save_cache from llm_service import call_llm class State(rx.State): text_input: str task: str 摘要 result: str is_hit: bool False run_count: int 0 hit_count: int 0 llm_call_count: int 0 log: list[str] [] def set_text_input(self, value: str): self.text_input value def set_task(self, value: str): self.task value def run_workflow(self): if not self.text_input.strip(): self.log [*self.log, 输入为空请先填写文本。] return self.run_count 1 self.is_hit False start time.time() self.log [*self.log, f 第 {self.run_count} 次执行 ] key build_key(self.task, self.text_input) cached get_cache(self.task, self.text_input) if cached: self.hit_count 1 self.result cached[result] self.is_hit True elapsed time.time() - start self.log [ *self.log, 步骤1校验输入 - 通过, f步骤2缓存指纹 - {key}, 步骤3命中缓存 - 直接重放上一次工作流结果, f步骤4输出结果 - 耗时 {elapsed:.3f}s, 提示本次没有调用 LLMLLM 调用次数保持为 0。, ] else: self.llm_call_count 1 self.log [ *self.log, 步骤1校验输入 - 通过, f步骤2缓存指纹 - {key}, 步骤3缓存未命中 - 调用 LLM 服务, 步骤4LLM 返回结果写入缓存, ] self.result call_llm(self.task, self.text_input) save_cache(self.task, self.text_input, self.result) elapsed time.time() - start self.log [ *self.log, f步骤5输出结果 - 耗时 {elapsed:.3f}s, 提示结果已缓存相同输入下次将零调用重放。, ] def clear_log(self): self.log [] self.result self.is_hit False def index(): return rx.container( rx.heading(Reflex一次演示工作流重放零 LLM 调用, sizelg), rx.text( 输入文本并选择任务点击运行按钮。 第一次会执行 LLM 调用并缓存结果第二次开始遇到相同输入会直接重放缓存结果。 ), rx.hstack( rx.vstack( rx.text(输入文本), rx.text_area( valueState.text_input, on_changeState.set_text_input, placeholder请输入要处理的文本..., height150px, width100%, ), rx.text(任务类型), rx.select( [摘要, 翻译, 关键词提取], valueState.task, on_changeState.set_task, ), rx.hstack( rx.button(运行工作流, on_clickState.run_workflow), rx.button(清空日志, on_clickState.clear_log), ), rx.divider(), rx.text(执行结果), rx.text(State.result), rx.cond( State.is_hit, rx.text(本次状态命中缓存零 LLM 调用, colorgreen), rx.text(本次状态已调用 LLM, colororange), ), width50%, ), rx.vstack( rx.text(执行日志), rx.foreach( State.log, lambda item: rx.text(item), ), width50%, ), width100%, ), rx.hstack( rx.text(f运行次数{State.run_count}), rx.text(f缓存命中{State.hit_count}), rx.text(fLLM 调用{State.llm_call_count}), spacing6, ), padding4, max_width1200px, ) app rx.App() app.add_page(index, route/)这段代码中需要重点理解几个部分。State 变量text_input用户输入的文本。task当前选择的任务类型。result工作流执行结果。is_hit本次执行是否命中缓存。run_count、hit_count、llm_call_count统计信息。log执行日志列表。在run_workflow()方法中最重要的判断逻辑是if cached:。如果缓存存在直接读取cached[result]不调用call_llm()。如果缓存不存在调用 LLM并把结果写入缓存。这里使用self.log [*self.log, 新日志]而不是self.log.append()是为了确保 Reflex 状态变量能够可靠地触发界面更新。虽然append()在某些版本中也可以工作但显式创建新列表更稳妥。4.4 编写 rxconfig.py 配置文件路径rxconfig.pyimport reflex as rx config rx.Config( app_namemy_llm_workflow, )这里指定了 Reflex 应用名称为my_llm_workflow因此主应用文件应该命名为my_llm_workflow.py。如果你的文件名不同需要同步修改app_name配置。4.5 运行项目在项目根目录执行reflex run启动成功后浏览器访问http://localhost:3000。第一次执行在文本框中输入“这是一段用于测试工作流缓存的商品文案。”任务类型选择“摘要”。点击“运行工作流”按钮。界面上会出现类似下面的日志 第 1 次执行 步骤1校验输入 - 通过 步骤2缓存指纹 - 8a3f6c... 步骤3缓存未命中 - 调用 LLM 服务 步骤4LLM 返回结果写入缓存 步骤5输出结果 - 耗时 0.021s 提示结果已缓存相同输入下次将零调用重放。第二次执行保持完全相同的文本和任务类型再次点击“运行工作流” 第 2 次执行 步骤1