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

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 实例）

各关键文件职责如下：

文件	职责
api.tsx	Axios 实例、ApiInterceptor 鉴权拦截器组件、performStreamingRequest() SSE 流式请求
request-processor.ts	UseRequestProcessor 钩子，为 useQuery/useMutation 注入默认重试与失效逻辑
constants.ts	URL 常量 URLs 与 getURL() 路径构造助手
queries/ 目录	按领域（flows、folders、variables、auth、messages 等）组织的 query/mutation 钩子
types/api/index.ts	useQueryFunctionType、useMutationFunctionType 等类型助手

这条调用链决定了后文所有规则的核心思想：重试与失效逻辑收敛在钩子定义处，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()。它基于浏览器原生 fetch API（而非 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 的选型表

操作	方式
构建 Flow（Build flow）	performStreamingRequest() + SSE
聊天交互（Chat interaction）	performStreamingRequest() + SSE
CRUD 操作（flows、folders、variables）	Axios api 实例，经由 query/mutation 钩子
文件上传/下载	Axios api 实例
鉴权操作	Axios api 实例

用 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，不同会话的轮询缓存互不干扰，切换会话不会命中旧缓存。

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

主题	规则
条件查询	组合条件时始终检查 options?.enabled ?? true；缺失参数用 enabled 而非非空断言
缓存失效	失效逻辑写在 mutation 钩子的 onSettled，组件回调只做导航/toast
Query key	首元素永远是钩子名；同一查询只用一个规范 key
mutate	默认 mutate + 回调；mutateAsync 仅限顺序依赖场景且必须 try/catch
鉴权错误	401/403 由 ApiInterceptor 统一刷新/登出，钩子与组件不要处理
4xx	客户端错误不重试（isClientError），瞬时 5xx/网络故障才重试
流式请求	build/chat 用 performStreamingRequest() + SSE；中止用 AbortController 并重建实例
轮询	refetchInterval + refetchIntervalInBackground: false
重试覆写	非幂等创建、全局更新、删除类 mutation 使用 retry: false

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