LobeHub Upstash Workflow 三层模式实战：以 welcome-placeholder 与 agent-welcome 两个真实工作流为例

【免费下载链接】lobehub 🤯 LobeHub is your Chief Agent Operator, organizing your agents into 7×24 operations by hiring, scheduling, and reporting on your entire AI team. 【免费下载链接】lobehub 项目地址: https://gitcode.com/GitHub_Trending/lo/lobehub

本文基于 LobeHub 仓库中的技能参考文档 examples.md，完整拆解两个已经落地代码库的 Upstash Workflow + QStash 异步工作流——welcome-placeholder（AI 欢迎占位符生成）与 agent-welcome（Agent 欢迎语与开放问题生成）。读完本文，你将掌握 LobeHub 统一的「入口 → 分页 → 单任务」三层工作流架构，理解 dry-run 预演、fan-out 扇出、单任务幂等执行三大核心模式如何在真实业务中落地，并学会如何以「实体替换」的方式低成本新增自己的工作流。

一、模式背景：为什么 LobeHub 的工作流都是同一个形状

在 LobeHub 中，基于 Upstash Workflow 的异步批量任务（例如为全量用户生成内容、为全量 Agent 预热数据）必须同时应对三个平台层面的约束（见 SKILL.md）：

  1. 速率限制让无脑全量扇出（blind fan-out）非常危险；
  2. 步骤上限限制单个 workflow 的执行规模；
  3. 幂等性要求决定了重试时不能重复处理同一个对象。

因此每个工作流都由三个核心模式组合而成：

模式作用出现层级
Dry-Run 模式只统计不执行，预览「将会处理多少条」Layer 1 入口层
Fan-Out 扇出模式把超大批次切成小分片并行递归处理Layer 2 分页层
单任务执行每次 workflow 执行恰好处理一个对象Layer 3 执行层

所有工作流遵循同一套三层架构：

Layer 1: 入口层 (process-*)
  ├─ 校验前置条件
  ├─ 计算待处理对象总数
  ├─ 过滤已有结果的对象
  ├─ 支持 dry-run 模式（仅统计）
  └─ 有待处理工作时触发 Layer 2

Layer 2: 分页层 (paginate-*)
  ├─ 游标（cursor）分页
  ├─ 大批次 fan-out
  ├─ 递归处理所有页
  └─ 为每个对象触发 Layer 3

Layer 3: 单任务执行层 (execute-* / generate-*)
  └─ 针对单个对象执行真正的业务逻辑

文件布局约定上，API 路由放在 src/app/(backend)/api/workflows/{workflow-name}/ 下（三个 route 文件分别对应三层），而工作流类放在 apps/server/src/workflows/{workflowName}/index.ts。当前仓库的 apps/server/src/workflows 目录中可以看到多个按该约定组织的实现（如 onboardingUnderstanding、topicAutoSummary 等），印证了这是代码库的长期约定。

需要说明的是：下文引用的 welcome-placeholder、agent-welcome 两个示例的路径（/api/workflows/welcome-placeholder/... 等）来自 examples.md 文档的历史快照；从当前仓库源码结构看，代码库经过重构后部分具体文件已迁移，但「三层切分 + 三大模式」这一模式本身在文档与技能体系中仍被作为标准范式保留。

二、示例一：welcome-placeholder（用户欢迎占位符）

使用场景：为平台用户生成 AI 驱动的欢迎占位符（welcome placeholders）。

三层结构：

  • Layer 1：process-users — 入口层，检查哪些用户符合条件；
  • Layer 2：paginate-users — 分页遍历所有活跃用户；
  • Layer 3：generate-user — 为一个用户生成占位符。

关键特性（这一节是 examples.md 的核心内容，完整列出）：

  • 过滤掉 Redis 中已缓存占位符的用户——这是幂等性的第一道闸门；
  • paidOnly 标志位，将处理范围限定为订阅（付费）用户；
  • dryRun 模式，只输出统计信息，不产生任何副作用；
  • 大批量用户时的 fan-out，分片大小 CHUNK_SIZE=20。

Layer 3 的代码形态：

export const { POST } = serve<GenerateUserPlaceholderPayload>(async (context) => {
  const { userId } = context.requestPayload ?? {};

  const workflow = new WelcomePlaceholderWorkflow(db, userId);
  const placeholders = await context.run('generate', () => workflow.generate());

  return { success: true, userId, placeholdersCount: placeholders.length };
});

注意这里的几个模式要点：

  1. 路由处理器通过 @upstash/workflow/nextjs 的 serve<Payload> 包裹，payload 类型化（GenerateUserPlaceholderPayload）；
  2. 真正的业务被封装进 WelcomePlaceholderWorkflow 类（构造函数接收 db 与单个 userId），路由层只做「取 payload → 调类方法 → 返回统一响应」的薄壳；
  3. context.run('generate', ...) 为这一步赋予唯一步骤名——步骤名是 workflow 持久化与断点恢复的依据，SKILL.md 的清单中明确要求「Unique context.run() step names」。

文档中给出的对应文件（历史路径，供溯源）：

/api/workflows/welcome-placeholder/process-users/route.ts
/api/workflows/welcome-placeholder/paginate-users/route.ts
/api/workflows/welcome-placeholder/generate-user/route.ts
/server/workflows/welcomePlaceholder/index.ts

三、示例二：agent-welcome（Agent 欢迎数据）

使用场景：为 AI Agent 生成欢迎消息与开放问题（welcome messages and open questions）。

三层结构：

  • Layer 1：process-agents — 入口层，检查哪些 Agent 符合条件；
  • Layer 2：paginate-agents — 分页遍历活跃 Agent；
  • Layer 3：generate-agent — 为一个 Agent 生成欢迎数据。

关键特性：与示例一完全对应的四件套——

  • 过滤 Redis 中已有缓存数据的 Agent；
  • paidOnly 标志位，只处理订阅用户的 Agent；
  • dryRun 统计模式;
  • 大批量 Agent 时 fan-out（CHUNK_SIZE=20）。

Layer 3 的代码形态：

export const { POST } = serve<GenerateAgentWelcomePayload>(async (context) => {
  const { agentId } = context.requestPayload ?? {};

  const workflow = new AgentWelcomeWorkflow(db, agentId);
  const data = await context.run('generate', () => workflow.generate());

  return { success: true, agentId, data };
});

文档中给出的对应文件（历史路径，供溯源）：

/api/workflows/agent-welcome/process-agents/route.ts
/api/workflows/agent-welcome/paginate-agents/route.ts
/api/workflows/agent-welcome/generate-agent/route.ts
/server/workflows/agentWelcome/index.ts

四、两者的相同与差异：「实体替换」才是这套模式的真正卖点

examples.md 用一句话点破了两个示例的本质关系：它们是同一个模式，只在三点上不同：

  1. 实体类型：users vs agents；
  2. 业务逻辑：placeholder 生成 vs welcome 数据生成；
  3. 数据源：不同的数据库查询。

除此之外的一切——三层切分、dry-run 处理、fan-out、filter-existing 过滤、flowControl 调参——完全一致。文档原话的结论是：

一旦内化了这个模式，新增一个工作流主要就是实体替换（entity-substitution）。

这正是「示例文档」的价值所在：读这两个端到端实现，等于同时读到了 LobeHub 工作流范式的模板实例。

五、纵深：三大模式在完整模板中的落点

为了把示例中「只展示了 Layer 3」的部分补全，references/implementation.md 给出了与上述两个示例逐行同构的完整代码模板，其中与示例一、示例二直接对应的实现细节如下。

5.1 工作流类：静态 trigger 方法 + 幂等过滤器

export class {WorkflowName}Workflow {
  private static client: Client;

  static triggerProcessItems(payload: ProcessItemsPayload) {
    const url = getWorkflowUrl(WORKFLOW_PATHS.processItems);
    return this.getClient().trigger({ body: payload, url });
  }

  static triggerPaginateItems(payload: PaginateItemsPayload) {
    const url = getWorkflowUrl(WORKFLOW_PATHS.paginateItems);
    return this.getClient().trigger({ body: payload, url });
  }

  static triggerExecuteItem(payload: ExecuteItemPayload) {
    const url = getWorkflowUrl(WORKFLOW_PATHS.executeItem);
    return this.getClient().trigger({ body: payload, url });
  }

  /**
   * Filter items that need processing (e.g. check Redis cache, database state).
   * Return only the ones that actually need work — keeps the pipeline idempotent.
   */
  static async filterItemsNeedingProcessing(itemIds: string[]): Promise<string[]> {
    // ...
  }
}

对应到两个示例：WelcomePlaceholderWorkflow / AgentWelcomeWorkflow 的实例方法 generate() 就是 Layer 3 调用的业务入口，而静态 trigger* 方法承担了「workflow 之间通过 QStash HTTP 触发」的桥接——每个静态方法都对应一个 HTTP 路由 URL。

5.2 Layer 1：dry-run 短路与并行度锁定

入口层模板中，dry-run 在一切副作用之前短路返回统计信息：

if (dryRun) {
  console.log('[{workflow}:process] Dry run mode, returning statistics only');
  return {
    ...result,
    dryRun: true,
    message: `[DryRun] Would process ${itemsNeedingProcessing.length} items`,
  };
}

这解释了示例中「dryRun mode for statistics」的实际机制：先 get-all-items 查询全量符合条件的实体 id，再经 filter-existing 剔除已处理项，然后先判断 dryRun 再决定是否触发分页。入口层的 flowControl 被刻意压到 parallelism: 1, ratePerSecond: 1，从结构上看是为了保证「同一时刻只有一个入口实例在派发」，避免重复触发。

5.3 Layer 2：fan-out 的递归自触发

示例中标注的 CHUNK_SIZE=20 在模板中对应这样的分支逻辑：

const PAGE_SIZE = 50;   // 每页拉取的对象数
const CHUNK_SIZE = 20;   // 每个 fan-out 分片的对象数

if (itemIds.length > CHUNK_SIZE) {
  const chunks = chunk(itemIds, CHUNK_SIZE);
  await Promise.all(
    chunks.map((ids, idx) =>
      context.run(`{workflow}:fanout:${idx + 1}/${chunks.length}`, () =>
        WorkflowClass.triggerPaginateItems({ itemIds: ids }),
      ),
    ),
  );
} else {
  // 直接为每个对象触发 Layer 3
  await Promise.all(
    itemIds.map((itemId) =>
      context.run(`{workflow}:execute:${itemId}`, () =>
        WorkflowClass.triggerExecuteItem({ itemId }),
      ),
    ),
  );
}

两个值得注意的机制：

  • fan-out 是递归自触发：分片不是在本 workflow 内继续循环，而是带着 itemIds 重新 trigger 一次 paginate 路由，天然绕开了单个 workflow 的步骤数上限；
  • 双模式 payload：PaginateItemsPayload 同时支持 cursor（常规分页）与 itemIds（fan-out 分片直入）两种入参，收到 itemIds 时直接走执行分支，不再翻页。

分页层的 flowControl 默认 parallelism: 20, ratePerSecond: 5，与执行层的 parallelism: 10, ratePerSecond: 5 相配合，形成「入口串行 → 分页半并行 → 执行受控并行」的速率漏斗，对应 SKILL.md 中「速率限制让盲目 fan-out 危险」的约束。

5.4 环境变量与触发前提

三个 trigger* 方法在运行时依赖两个必需环境变量（缺失时直接抛错）：

# Required for all workflows
APP_URL=https://your-app.com # Base URL for workflow endpoints
QSTASH_TOKEN=qstash_xxx      # QStash authentication token

# Optional (for custom QStash URL)
QSTASH_URL=https://custom-qstash.com

即：APP_URL 用于拼接 /api/workflows/... 的完整触发地址，QSTASH_TOKEN 用于构造 Upstash Client 的鉴权；QSTASH_URL 可选，用于指向自建/自定义的 QStash 端点。

六、复用到你自己的实体：从示例到新工作流的最小路径

把示例一/二抽象后，在 LobeHub 代码库中新增一个「对全量实体做 X」的工作流，路径基本是填空：

  1. 确定实体与业务逻辑：明确处理对象（users / agents / items…）与单对象业务函数；
  2. 写工作流类：在 apps/server/src/workflows/{workflowName}/index.ts 定义 WORKFLOW_PATHS、三种 payload 接口、三个静态 trigger 方法与 filterItemsNeedingProcessing（Redis 缓存或数据库状态判断）；
  3. 写三层路由：
    • process-*：查询全量候选 → filter-existing → dry-run 短路或触发分页，flowControl parallelism: 1；
    • paginate-*：PAGE_SIZE=50 游标分页，CHUNK_SIZE=20 超阈 fan-out，flowControl parallelism: 20, ratePerSecond: 5；
    • generate-*/execute-*：单对象执行，context.run 步骤名唯一，flowControl parallelism: 10, ratePerSecond: 5；
  4. 日志与响应：统一 [workflow:layer] 前缀日志、统一 { success, ... } 响应形状;
  5. 验证顺序：先 dryRun 冒烟，再用小批量试跑，最后全量。

SKILL.md 中还维护了一份覆盖「规划 / 实现 / 质量与部署」三段的新工作流清单，可作为发布前的逐项核对；flowControl 调参、错误处理、日志与测试的进一步细节见 best-practices.md，lobehub-cloud 环境下的 re-export 与云上运维见 cloud.md。

七、小结

  • examples.md 用 welcome-placeholder 与 agent-welcome 两个真实工作流证明：LobeHub 的 Upstash 异步工作流是一个模式、多次实例化——三层切分、dry-run、fan-out（CHUNK_SIZE=20）、filter-existing、flowControl 调参全部同构，差异仅在于实体、业务逻辑与数据源；
  • Layer 3 的 serve<Payload> + context.run + 工作流类薄壳调用，是可直接套用的代码形态；
  • 从示例到落地，implementation.md 提供逐行同构的完整模板，配合 APP_URL / QSTASH_TOKEN 环境变量即可在本地或云端触发；
  • 掌握「实体替换」这一视角后，新增一个面向任意实体的批量 AI 工作流，主要就是替换查询、过滤与生成函数三处代码。

【免费下载链接】lobehub 🤯 LobeHub is your Chief Agent Operator, organizing your agents into 7×24 operations by hiring, scheduling, and reporting on your entire AI team. 【免费下载链接】lobehub 项目地址: https://gitcode.com/GitHub_Trending/lo/lobehub

Logo
乐奇 Rokid 开放社区

更多推荐

cover

AI眼镜的赢家不在硅谷，在一个县城

avatar 乐奇 Rokid 开放社区
cover

极刻觅镜 | AI眼镜日报｜乐奇二代光学升级、Rokid Prism打通桌面智能体

avatar 乐奇 Rokid 开放社区
cover

ChatGPT流量不到Google的1%，机会在智能体搜索

avatar 乐奇 Rokid 开放社区