企业官网建设流程全解析

Langflow 前端数据层运行时规则：React Query 条件查询、缓存失效与 Mutation 重试的源码级实践

【免费下载链接】langflowLangflow is a powerful tool for building and deploying AI-powered agents and workflows.项目地址: https://gitcode.com/GitHub_Trending/la/langflow

Langflow 前端基于 Axios + TanStack React Query v5 构建数据请求层，所有查询与变更（mutation）钩子都经由统一的UseRequestProcessor封装，以获得一致的默认重试、缓存失效与鉴权错误处理。本文基于仓库中的运行时规则文档（runtime-rules.md），逐条解析条件查询、缓存失效、query key 约定、错误处理、SSE 流式请求与轮询等实战模式，并结合 request-processor.ts 和 api.tsx 的源码实现，说明每个规则的底层依据，帮助你在 Langflow 中创建或修改 API 钩子时严格遵循既有约定。

整体架构：一次 API 请求的完整调用链

Langflow 前端的请求架构没有 oRPC 或额外的契约层，其调用链是单向且固定的（见 SKILL.md）：

Component -> API 钩子（controllers/API/queries/{domain}/use-{verb}-{resource}.ts） -> UseRequestProcessor（controllers/API/services/request-processor.ts） -> useQuery / useMutation（TanStack React Query） -> queryFn / mutationFn -> api.get/post/patch/delete（controllers/API/api.tsx 中的共享 Axios 实例）

各关键文件职责如下：

这条调用链决定了后文所有规则的核心思想：重试与失效逻辑收敛在钩子定义处，UI 反馈收敛在调用处，两者不越界。

条件查询：用enabled控制请求是否发出

Langflow 钩子会把options透传给UseRequestProcessor，后者再透传给useQuery，因此消费方可以通过enabled选项条件性地启用或停用查询。

模式一：未认证时禁用查询

// Pattern: Disable query when not authenticated export const useGetGlobalVariables: useQueryFunctionType< undefined, GlobalVariable[] > = (options?) => { const { query } = UseRequestProcessor() const isAuthenticated = useAuthStore((state) => state.isAuthenticated) const getGlobalVariablesFn = async (): Promise<GlobalVariable[]> => { if (!isAuthenticated) return [] const res = await api.get(`${getURL("VARIABLES")}/`) return res.data } return query(["useGetGlobalVariables"], getGlobalVariablesFn, { refetchOnWindowFocus: false, enabled: isAuthenticated && (options?.enabled ?? true), ...options, }) }

注意enabled: isAuthenticated && (options?.enabled ?? true)这一行的写法：先组合自身条件，再让消费方的options.enabled可以覆写或叠加禁用。

模式二：必需参数缺失时禁用查询

// Pattern: Disable query when required param is missing export const useGetFlow: useQueryFunctionType<{ id: string }, FlowResponse> = ( params, options?, ) => { const { query } = UseRequestProcessor() const getFlowFn = async (): Promise<FlowResponse> => { const res = await api.get(`${getURL("FLOWS")}/${params.id}`) return res.data } return query(["useGetFlow", params.id], getFlowFn, { enabled: !!params.id && (options?.enabled ?? true), ...options, }) }

模式三：消费方通过 options 禁用查询

const { data: flow } = useGetFlow( { id: flowId }, { enabled: showFlowDetails }, )

消费方在showFlowDetails为假时即可停发请求，而钩子内部无需感知该场景。

规则总结

• 组合其他条件时始终检查options?.enabled ?? true，让消费方也能禁用查询；
• 当认证状态或必需数据可能缺失时，在查询函数开头提前守卫（如上面if (!isAuthenticated) return []）；
• 不要用非空断言（!）绕过缺失的 params，改用enabled让查询根本不执行。

缓存失效：在 mutation 钩子定义处绑定失效逻辑

规则非常明确：缓存失效必须绑定在 mutation 钩子的定义处；组件只允许添加 UI 反馈（toast、导航），不允许决定失效哪些查询。

通过 onSettled 扩展失效

UseRequestProcessor.mutate()包装器默认会在onSettled中调用queryClient.invalidateQueries({ queryKey: mutationKey })（见 request-processor.ts 中的onSettled包装）。需要额外失效其他查询时，在钩子内扩展onSettled：

export const usePostAddFlow: useMutationFunctionType< undefined, PostAddFlowPayload > = (options?) => { const { mutate, queryClient } = UseRequestProcessor() const myCollectionId = useFolderStore((state) => state.myCollectionId) const postAddFlowFn = async (payload: PostAddFlowPayload): Promise<any> => { const response = await api.post(`${getURL("FLOWS")}/`, payload) return response.data } return mutate(["usePostAddFlow"], postAddFlowFn, { onSettled: (response) => { if (response) { queryClient.refetchQueries({ queryKey: ["useGetRefreshFlowsQuery", { get_all: true, header_flows: true }], }) queryClient.refetchQueries({ queryKey: ["useGetFolder", response.folder_id ?? myCollectionId], }) } }, ...options, // Consumer options come LAST }) }

这里创建 Flow 后不仅刷新了全局的 Flow 列表（useGetRefreshFlowsQuery），还精准刷新了新建 Flow 所在文件夹的缓存（useGetFolder），response.folder_id ?? myCollectionId处理了未指定文件夹时落入“我的集合”的场景。

三种失效/刷新写法

// Broad invalidation: 失效某个领域下的所有查询 queryClient.invalidateQueries({ queryKey: ["useGetFlows"] }) // Specific invalidation: 只失效单个缓存条目 queryClient.invalidateQueries({ queryKey: ["useGetFlow", flowId] }) // Refetch instead of invalidate: 需要立即拿到新数据时直接重取 queryClient.refetchQueries({ queryKey: ["useGetFolder", folderId] })

从语义上看，invalidateQueries只是标记缓存过期、由激活的查询在适当时机重取；refetchQueries则立即触发重新请求。列表页、文件夹详情这类“用户能立刻看到”的数据，Langflow 倾向于用refetchQueries。

组件侧回调只负责 UI

// Component only adds UI behavior const { mutate: addFlow } = usePostAddFlow() const handleCreate = () => { addFlow(flowData, { onSuccess: (response) => { // UI-only: navigate, show toast navigate(`/flow/${response.id}`) setSuccessData({ title: "Flow created successfully" }) }, onError: (error) => { setErrorData({ title: "Failed to create flow", list: [error.message], }) }, }) }

组件完全不需要知道“创建 Flow 之后应该刷新哪些查询”——那已经是usePostAddFlow内部的事。这种职责划分使得钩子可以在任意组件复用而不产生失效逻辑的重复或遗漏。

Query Key 约定

Query key 是标识缓存条目的数组，Langflow 的约定是第一个元素永远是钩子名字符串，后续元素是区分缓存的参数：

// Base key: hook name ["useGetGlobalVariables"] // Parameterized key: hook name + params ["useGetFlow", flowId] ["useGetFolder", folderId] // Complex key: hook name + param object ["useGetRefreshFlowsQuery", { get_all: true, header_flows: true }] ["useGetMessages", { flowId, sessionId }] ["useGetBuilds", { flowId }] // Mutation key: hook name (used for automatic invalidation by UseRequestProcessor) ["usePostAddFlow"] ["useDeleteMessages"]

规则：

• 第一个元素始终是钩子名字符串；
• mutation key 保持与钩子名一致，这样UseRequestProcessor.mutate()才能在 settled 时自动失效它；
• 同一查询不允许出现多个不同的 key 字符串（如["getFlows"]、["useGetFlows"]、["flows-list"]三处各写一个），否则失效逻辑会漏掉其中一份缓存；
• 额外需要失效的目标必须在onSettled中显式添加。

key 的稳定性直接决定失效是否命中——invalidateQueries是按前缀匹配的，所以“钩子名 + 参数”这种结构既是缓存隔离手段，也是失效寻址手段。

mutate与mutateAsync的取舍

默认使用mutate，只有确需 Promise 语义时才用mutateAsync。

规则：

• 事件处理器应调用mutate(...)并配合onSuccess/onError回调；
• 每个await mutateAsync(...)必须包裹在try/catch中；
• 当回调已能清晰表达流程时，不要改用mutateAsync。

// Default: use mutate with callbacks const { mutate: deleteFlow } = useDeleteFlow() const handleDelete = () => { deleteFlow(flowId, { onSuccess: () => { navigate("/flows") setSuccessData({ title: "Flow deleted" }) }, onError: (error) => { setErrorData({ title: "Delete failed", list: [error.message] }) }, }) }

例外场景是顺序依赖操作，例如“复制 Flow 并重命名后打开新副本”，必须等第一个请求返回才能拿到newFlow.id：

// Exception: Promise semantics needed for sequential operations const handleDuplicateAndOpen = async () => { try { const newFlow = await duplicateFlow.mutateAsync(flowData) await renameFlow.mutateAsync({ id: newFlow.id, name: `${flowData.name} (copy)` }) navigate(`/flow/${newFlow.id}`) } catch (error) { setErrorData({ title: "Failed to duplicate flow", list: [error instanceof Error ? error.message : "Unknown error"], }) } }

错误处理：拦截器、mutation 与 query 三层分工

API 拦截器统一处理鉴权错误

api.tsx中的ApiInterceptor组件自动处理以下情况，各个钩子和组件不需要再处理 401/403：

• 401 Unauthorized：通过useRefreshAccessToken尝试刷新 token，然后重试原请求；
• 403 Forbidden：走与 401 相同的鉴权错误流程；
• 累计 3 次以上鉴权错误：自动将用户登出；
• 500 错误：清空 flow store 中的 build vertex 状态。

源码可以逐条印证这些行为。在 api.tsx 中：

• checkErrorCount()（约 L250-L262）维护authenticationErrorCount，当计数超过 3 时调用mutationLogout()登出；
• tryToRenewAccessToken(error)（约 L264-L283）调用mutationRenewAccessToken刷新 token，刷新成功后计数归零，刷新失败且非网络错误时直接登出；
• clearBuildVerticesState(error)（约 L285-L293）专门处理 500：把useFlowStore中处于构建中的顶点标记为BUILT并setIsBuilding(false)，避免一次服务端 5xx 让构建状态机永久卡在“构建中”；
• remakeRequest(error)（约 L295-L303）返回完整的AxiosResponse重放请求，注释特别说明如果只返回response.data会导致调用点双重解包得到undefined。

Mutation 错误处理

用户可见的失败反馈放在调用处的onError中，优先展示后端返回的detail：

const { mutate: saveFlow } = useSaveFlow() const handleSave = () => { saveFlow(flowData, { onError: (error) => { setErrorData({ title: "Failed to save flow", list: [error.response?.data?.detail ?? error.message], }) }, }) }

Query 错误处理

对查询，UseRequestProcessor的默认重试逻辑（5 次重试 + 指数退避）会消化掉瞬时故障。对于永久性错误，用 options 中的retry: false加onError或错误边界处理：

const { data, error, isError } = useGetFlow( { id: flowId }, { retry: false, // Override default retry for known-missing resources onError: (error) => { if (error.response?.status === 404) { navigate("/flows") } }, }, )

“已知资源不存在”（404）时覆盖默认重试并直接导航回列表页，是典型的处理模式。

流式请求：SSE 与 AbortController

Langflow 的构建（build）和聊天（chat）流式交互使用 api.tsx 中的performStreamingRequest()。它基于浏览器原生fetchAPI（而非 Axios），手工解析 Server-Sent Events：

import { performStreamingRequest } from "@/controllers/API/api" const buildController = new AbortController() await performStreamingRequest({ method: "POST", url: `${baseURL}/api/v1/build/${flowId}/flow`, body: { inputs, files }, buildController, onData: async (event) => { // Process individual SSE events // Return true to continue, false to abort return true }, onDataBatch: async (events) => { // Process batch of events from a single chunk (more efficient) // Return true to continue, false to abort return true }, onError: (statusCode) => { // Handle HTTP error status }, onNetworkError: (error) => { // Handle network-level errors }, })

源码层面（api.tsx L330-L400 附近）有几个实现细节值得了解：

• 请求头固定带Connection: close，注释说明这个 flag 用于“确保客户端断开时服务端停止任务”；
• SSE 事件以\n\n分块，单个事件可能被切在两个网络 chunk 中间，函数用current数组拼接半截 JSON，只有以}结尾时才尝试解析；
• 解析前会经过sanitizeJsonString()，把后端 JSON 里非法的裸NaN替换为null，避免JSON.parse抛错；
• 派发策略是：优先把同一个 chunk 内解析出的全部事件交给onDataBatch批量处理（更高效），否则回退到逐事件的onData；
• 请求体通过JSON.stringify(body)序列化，凭证模式由getFetchCredentials()决定，buildController.signal贯穿整个读取循环以实现中断。

流式与 REST 的选型表

用 AbortController 中断流

const buildController = useRef(new AbortController()) const handleStopBuild = () => { buildController.current.abort() buildController.current = new AbortController() }

abort()之后要立刻替换成新的AbortController，否则下一次构建会拿到已 abort 的旧 signal，直接失败。

UseRequestProcessor 默认值与onSettled的微妙之处

UseRequestProcessor（request-processor.ts）是全部默认行为的注入点。

Query 默认值

{ retry: 5, retryDelay: (attemptIndex) => Math.min(1000 * 2 ** attemptIndex, 30000), // 1s, 2s, 4s, 8s, 16s（上限 30s） }

源码中这是makeRetry(5)工厂生成的函数（L50-L57），且带有一个文档未强调但重要的过滤条件：isClientError（L20-L24）判定 4xx 为“客户端有意拒绝”（鉴权、校验、部署守卫等），4xx 绝不重试；isRetryableServerError（L29-L34）只对 5xx 响应和“发出了请求但没有收到响应”的网络类失败返回可重试。也就是说“5 次重试”只覆盖真正的瞬时故障，不会把校验错误反复打到后端。

Mutation 默认值与 in-band 重试

运行时文档给出的mutate()概念实现如下：

function mutate(mutationKey, mutationFn, options = {}) { return useMutation({ mutationKey, mutationFn, onSettled: (data, error, variables, context) => { queryClient.invalidateQueries({ queryKey: mutationKey }); options.onSettled && options.onSettled(data, error, variables, context); }, ...options, // Spreads AFTER the wrapper onSettled retry: options.retry ?? 3, // Comes AFTER the spread retryDelay: (attemptIndex) => Math.min(1000 * 2 ** attemptIndex, 30000), }); }

实际源码中（request-processor.ts L127-L153）默认重试预算同样是 3 次（MAX_MUTATION_RETRIES = 3），但实现上有一个值得注意的演进：重试不在 react-query 的retry选项上做，而是包在 mutation 函数内部。源码注释（L79-L86）解释了原因——react-query 的 retryer 在文档隐藏（document hidden）或浏览器报告离线时会暂停计时，导致失败的 mutation 可能永远停在暂停态：onSettled不触发、调用方的onError永远不响、用户面对 5xx 没有任何反馈。而withTransientErrorRetry（L87-L104）用普通的setTimeout在隐藏标签页中继续计时，保证重试预算耗尽后 mutation 必然以 resolve 或 throw 结算，onError/onSettled一定执行。此外getRetryAfterMs（L37-L48）会解析服务端Retry-After响应头（delta-seconds 或 HTTP-date，RFC 9110 §10.2.3），让 mutation 重试尊重服务端的限流提示。

onSettled 机制的关键微妙之处（务必理解）：

1. UseRequestProcessor.mutate()定义了一个包装onSettled：先自动失效mutationKey，再调用钩子的options.onSettled；
2. 但...options展开位于包装器之后，因此只要钩子的options自带onSettled，就会覆盖包装器——自动失效实际上被跳过；
3. 实践中大多数钩子确实自带onSettled（里面是针对具体领域的 refetch/失效逻辑），所以 mutation key 的自动失效很少真正执行；
4. 这没有问题，因为失效一个 mutation key 通常本就无意义——mutation 不像 query 那样有缓存条目。

由此得到代码库约定：自定义onSettled放在...options之前，让消费方仍能覆写：

mutate(["usePostAddFlow"], fn, { onSettled: () => { queryClient.refetchQueries(...) }, // Hook-specific invalidation retry: false, // Override default retry if needed ...options, // Consumer options come LAST (can override onSettled, retry, etc.) })

何时用retry: false

对于重试会带来副作用的 mutation，把默认重试覆盖为retry: false：

• 非幂等的创建操作（重复创建风险）；
• 全局变量或设置的更新操作（旧数据覆盖风险）；
• 删除操作（资源可能已不存在）。

// Example: POST that creates a resource (not safe to retry) const mutation = mutate(["usePostGlobalVariables"], postFn, { onSettled: () => { queryClient.refetchQueries({ queryKey: ["useGetGlobalVariables"] }) }, retry: false, ...options, })

轮询模式：构建状态与消息的周期刷新

对需要周期性刷新的数据（构建状态、聊天消息），用refetchInterval实现轮询：

export const useGetMessagesPolling: useQueryFunctionType< { flowId: string; sessionId: string }, Message[] > = (params, options?) => { const { query } = UseRequestProcessor() const getMessagesFn = async (): Promise<Message[]> => { const res = await api.get( `${getURL("MESSAGES")}/?flow_id=${params.flowId}&session_id=${params.sessionId}`, ) return res.data } return query( ["useGetMessagesPolling", params.flowId, params.sessionId], getMessagesFn, { refetchInterval: 3000, // Poll every 3 seconds refetchIntervalInBackground: false, ...options, }, ) }

两个参数的配合要点：

• refetchInterval: 3000让查询每 3 秒重取一次，适用于 SSE 流之外的“慢变”数据；
• refetchIntervalInBackground: false确保标签页切到后台时停止轮询，避免后台无谓请求；
• query key 带上flowId和sessionId，不同会话的轮询缓存互不干扰，切换会话不会命中旧缓存。

小结：一张可核对的规则清单

掌握以上规则后，你在 queries/ 目录下新增或修改任何use-get-*/use-post-*/use-patch-*/use-delete-*钩子时，都能与 Langflow 前端既有的重试预算、缓存失效寻址和鉴权恢复机制无缝协作。完整的钩子结构、命名约定与反模式清单，可进一步参考 query-patterns.md。

【免费下载链接】langflowLangflow is a powerful tool for building and deploying AI-powered agents and workflows.项目地址: https://gitcode.com/GitHub_Trending/la/langflow