Introducing Experimental Workflows and Orchestrators in TanStack AI

by Alem Tuzlak on May 28, 2026.

TanStack AI workflows and orchestrators

Most AI apps start with one chat() call.大多数 AI 应用都是从一个 chat() 调用开始的。

But as soon as you need something more complex, this all breaks apart. You either fall back to using sub-agents as tools, or you have to write your own glue and abstractions on top to make a semi-decent workflow or orchestration mechanism to power your app. This just detracts from your time to work on the features you really care about.但一旦你需要更复杂的功能，这一切就会分崩离析。你要么退而求其次，将子代理（sub-agents）作为工具使用，要么就得自己编写胶水代码和抽象层，来构建一个勉强够用的工作流或编排机制来驱动你的应用。这只会分散你专注于真正核心功能开发的时间。

The model needs to draft, critique, revise, ask for approval, call another model, update state, and show the user what is happening while the run is still in progress. At that point, a one-shot chat endpoint turns into a hand-rolled workflow engine made of fetch calls, temporary state, custom SSE events, and a lot of code nobody wanted to own.模型需要进行起草、评审、修订、请求批准、调用另一个模型、更新状态，并在运行过程中向用户展示当前进度。到那时，一个一次性的聊天接口就会变成一个由 fetch 调用、临时状态、自定义 SSE 事件以及大量没人愿意维护的代码组成的“手搓”工作流引擎。

Today, TanStack introduces an experimental answer: TanStack AI Workflows & Orchestrators.今天，TanStack 带来了一个实验性的解决方案：TanStack AI Workflows & Orchestrators（工作流与编排器）。

Before we go further, a fair warning!在进一步深入之前，先给出一个善意的提醒！

This is not merged to main. It is not shipped, stable, or available in normal npm versions. It is a PR build you can try today through pkg.pr.new while the API is still being shaped. The goal is to get it in front of real use cases, demos, and feedback before we commit to the public API shape.此功能尚未合并到主分支。它尚未发布、不稳定，也未在正常的 npm 版本中提供。这是一个 PR 构建版本，你可以在 API 仍在调整期间通过 pkg.pr.new 进行试用。我们的目标是在确定公共 API 形式之前，将其应用于实际用例、演示并收集反馈。

This is where you come in. We need your help. We need people to test out our workflows and our orchestration mechanisms, give us their thoughts and opinions, and help us shape the final APIs.这就是你需要参与的地方。我们需要你的帮助。我们需要大家测试我们的工作流和编排机制，提供想法和意见，并帮助我们塑造最终的 API。

PR: TanStack/ai#542PR: TanStack/ai#542
PR build packages: pkg.pr.new commentPR 构建包: pkg.pr.new 评论
Package: @tanstack/ai-orchestration包: @tanstack/ai-orchestration

The goal is simple: compose multiple typed LLM and agent steps as normal TypeScript async generators, stream each step to the UI, pause for human approval, and resume through the same SSE flow.目标很简单：将多个带类型的 LLM 和代理步骤组合为标准的 TypeScript 异步生成器，将每个步骤流式传输到 UI，暂停以等待人工批准，并通过相同的 SSE 流恢复运行。

Define your agents定义你的代理#

Agents are typed wrappers around a chat() call or any async function. defineAgent gives each step an input schema, output schema, and implementation.代理是对 chat() 调用或任何异步函数的带类型包装器。defineAgent 为每个步骤提供了输入模式、输出模式和具体实现。

import { chat } from '@tanstack/ai'
import { defineAgent } from '@tanstack/ai-orchestration'
import { openaiText } from '@tanstack/ai-openai'
import { z } from 'zod'

export const ArticleSchema = z.object({
  title: z.string(),
  body: z.string(),
})

export const writer = defineAgent({
  name: 'writer',
  input: z.object({
    topic: z.string(),
  }),
  output: ArticleSchema,
  run: ({ input }) =>
    chat({
      adapter: openaiText('gpt-4o'),
      outputSchema: ArticleSchema,
      stream: true,
      systemPrompts: [
        'Write a concise developer article with a clear title and practical body.',
      ],
      messages: [{ role: 'user', content: input.topic }],
    }),
})

export const EditorReviewSchema = z.object({
  approved: z.boolean(),
  article: ArticleSchema,
  notes: z.string(),
})

export const editor = defineAgent({
  name: 'editor',
  input: z.object({
    article: ArticleSchema,
    feedback: z.string().optional(),
  }),
  output: EditorReviewSchema,
  run: ({ input }) =>
    chat({
      adapter: openaiText('gpt-4o'),
      outputSchema: EditorReviewSchema,
      stream: true,
      systemPrompts: [
        'Edit the article for accuracy, clarity, and practical developer tone. Apply any optional reviewer feedback.',
      ],
      messages: [{ role: 'user', content: JSON.stringify(input) }],
    }),
})

defineAgent wraps either a chat() call or a normal async function with input and output schemas. The workflow runtime uses those schemas to validate what enters and leaves the step, and TypeScript uses them to infer the callable shape inside a workflow. From the workflow's perspective, writer and editor are normal typed async steps.defineAgent 使用输入和输出模式包装 chat() 调用或普通的异步函数。工作流运行时使用这些模式来验证进入和离开步骤的数据，而 TypeScript 则利用它们在工作流内部推断可调用的形式。从工作流的角度来看，writer 和 editor 都是普通的带类型异步步骤。

Define a workflow定义一个工作流#

Once agents exist, defineWorkflow composes them with yield* inside an async function*.一旦代理定义完成，defineWorkflow 就会在 async function* 中使用 yield* 将它们组合起来。

import {
  approve,
  defineWorkflow,
  fail,
  succeed,
} from '@tanstack/ai-orchestration'
import { z } from 'zod'
import { ArticleSchema, editor, writer } from './agents'

export const ArticleInputSchema = z.object({
  topic: z.string(),
})

export const ArticleWorkflowOutputSchema = z.discriminatedUnion('ok', [
  z.object({
    ok: z.literal(true),
    article: ArticleSchema,
  }),
  z.object({
    ok: z.literal(false),
    reason: z.string(),
  }),
])

export const articleWorkflow = defineWorkflow({
  name: 'article-workflow',
  input: ArticleInputSchema,
  output: ArticleWorkflowOutputSchema,
  agents: {
    writer,
    editor,
  },
  run: async function* ({ input, agents }) {
    const draft = yield* agents.writer({ topic: input.topic })
    const edited = yield* agents.editor({ article: draft })

    if (!edited.approved) {
      return fail(edited.notes)
    }

    const decision = yield* approve({
      title: `Publish "${edited.article.title}"?`,
      description: edited.notes,
    })

    if (!decision.approved) {
      return fail(decision.feedback ?? 'Publication declined')
    }

    return succeed({ article: edited.article })
  },
})

The interesting part is the lack of framework ceremony. TypeScript knows the input expected by agents.writer, and it knows the shape returned after the yield*.有趣之处在于，它几乎没有框架的繁文缛节。TypeScript 知道 agents.writer 所需的输入，也知道 yield* 之后返回的数据结构。

Each yield* agents.someAgent(...) becomes a typed step. The runtime can emit lifecycle events around it, stream text while it runs, validate the result, snapshot state, and resume the generator with the typed output.每个 yield* agents.someAgent(...) 都会成为一个带类型的步骤。运行时可以在其周围触发生命周期事件、在运行时流式传输文本、验证结果、快照状态，并使用带类型的输出恢复生成器。

The workflow body is just TypeScript. Use if, for, while, try, await, helper functions, and whatever domain code you already have. The orchestration runtime only cares about the things you yield*.工作流主体就是标准的 TypeScript 代码。你可以使用 if、for、while、try、await、辅助函数以及你现有的任何业务逻辑代码。编排运行时只关心你 yield* 的内容。

Why async generator workflows? There are a lot of ways to model agent workflows. You can build a graph DSL. You can define nodes in JSON. You can describe a DAG and ask the runtime to interpret it. The reason we went with generator workflows is that whenever you yield the agent's step, it's streamed straight down to the client. The user sees everything in real time — tool calls, reasoning, whatever happens along the way — and by the end you just get the final output back.为什么要使用异步生成器工作流？建模代理工作流的方法有很多。你可以构建一个图 DSL，可以在 JSON 中定义节点，也可以描述一个 DAG 并让运行时去解释它。我们选择生成器工作流的原因是，每当你 yield 代理步骤时，它都会直接流式传输到客户端。用户可以实时看到一切——工具调用、推理过程以及沿途发生的任何事情——而在结束时，你只需获取最终输出即可。

Expose it over SSE通过 SSE 暴露#

Workflows run on the server. The browser consumes an event stream.工作流在服务器上运行。浏览器消费一个事件流。

The PR adds parseWorkflowRequest, runWorkflow, and inMemoryRunStore from @tanstack/ai-orchestration. You can pipe the returned stream through the existing toServerSentEventsResponse helper from @tanstack/ai.该 PR 从 @tanstack/ai-orchestration 中添加了 parseWorkflowRequest、runWorkflow 和 inMemoryRunStore。你可以通过 @tanstack/ai 中现有的 toServerSentEventsResponse 辅助函数来管道化返回的流。

import { toServerSentEventsResponse } from '@tanstack/ai'
import {
  inMemoryRunStore,
  parseWorkflowRequest,
  runWorkflow,
} from '@tanstack/ai-orchestration'
import { articleWorkflow } from './article-workflow'

const runStore = inMemoryRunStore({ ttl: 60 * 60 * 1000 })

export async function POST(request: Request) {
  const params = await parseWorkflowRequest(request)
  const stream = runWorkflow({
    workflow: articleWorkflow,
    runStore,
    ...params,
  })

  return toServerSentEventsResponse(stream)
}

runWorkflow emits AG-UI-style lifecycle events for the run. That includes run and step events, state snapshots, JSON Patch state deltas, output, and errors. The UI does not need to invent its own event protocol for "writer started", "editor streamed text", "approval requested", or "run finished".runWorkflow 会为运行过程发出 AG-UI 风格的生命周期事件。这包括运行和步骤事件、状态快照、JSON Patch 状态增量、输出和错误。UI 无需为“writer 已启动”、“editor 流式传输文本”、“请求批准”或“运行完成”发明自己的事件协议。

The current built-in persistence is inMemoryRunStore. That is useful for local demos and single-process evaluation. Production durability is still future-facing and experimental run-store-interface territory, especially for long pauses, deploys, restarts, and multi-node environments. But the API is there to implement your own durable run store and swap it in when you're ready.目前内置的持久化方案是 inMemoryRunStore。这对于本地演示和单进程评估很有用。生产环境的持久性仍处于面向未来和实验性的 run-store-interface 阶段，特别是对于长时间暂停、部署、重启和多节点环境而言。但 API 已经存在，你可以实现自己的持久化运行存储，并在准备好时进行替换。

Run it without a UI在没有 UI 的情况下运行#

A UI is optional. The SSE endpoint above exists so a browser can watch a run unfold, but the workflow itself is just a server-side async generator. If nothing is watching — a cron job, a queue worker, a batch script, or a plain JSON endpoint that only returns the final answer — you can run the workflow headless and never stream a single event to a client.UI 是可选的。上述 SSE 端点存在是为了让浏览器可以观察运行过程，但工作流本身只是一个服务器端的异步生成器。如果没有人观察——比如定时任务、队列工作进程、批处理脚本或仅返回最终答案的普通 JSON 端点——你可以在无头（headless）模式下运行工作流，而无需向客户端流式传输任何事件。

runWorkflow returns an async iterable, and iterating it is what drives the workflow forward. So for a headless run you drain the stream to completion, then read the final result back from the run store:runWorkflow 返回一个异步可迭代对象，迭代它就是驱动工作流前进的动力。因此，对于无头运行，你可以将流消耗至完成，然后从运行存储中读取最终结果：

import { inMemoryRunStore, runWorkflow } from '@tanstack/ai-orchestration'
import { articleWorkflow } from './article-workflow'

const runStore = inMemoryRunStore({ ttl: 60 * 60 * 1000 })

export async function generateArticle(topic: string) {
  const runId = crypto.randomUUID()

  const stream = runWorkflow({
    workflow: articleWorkflow,
    runStore,
    runId,
    input: { topic },
  })

  // No client is listening, so just drive the run to completion.
  for await (const _event of stream) {
    // Optionally log progress or persist events here.
  }

  const run = await runStore.getRunState(runId)

  if (run?.status === 'finished') {
    return run.output
  }

  throw new Error(run?.error?.message ?? `Workflow ended as "${run?.status}"`)
}

The same typed agents, schemas, state snapshots, and validation apply. You just skip toServerSentEventsResponse, the React hooks, and the browser entirely.同样的带类型代理、模式、状态快照和验证逻辑依然适用。你只需跳过 toServerSentEventsResponse、React hooks 以及浏览器部分即可。

One caveat: a headless run has nobody to answer a yield* approve(...). If the workflow hits an approval step, it pauses and getRunState reports status: 'paused' instead of 'finished'. Server-only workflows shine when the pipeline runs end to end on its own. If you need a human in the loop, either resume the paused run programmatically with an approval result or keep a UI connected.需要注意的一点：无头运行没有用户来响应 yield* approve(...)。如果工作流到达批准步骤，它会暂停，且 getRunState 会报告 status: 'paused' 而不是 'finished'。仅服务器端的工作流在流水线能够自动端到端运行时表现最佳。如果你需要人工介入，要么通过程序方式使用批准结果恢复暂停的运行，要么保持 UI 连接。

Consume it from React从 React 中消费#

On the client, WorkflowClient, useWorkflow, and useOrchestration consume the streamed events and keep local run state updated.在客户端，WorkflowClient、useWorkflow 和 useOrchestration 会消费流式事件并保持本地运行状态的更新。

tsxtsx

import { fetchWorkflowEvents, useWorkflow } from '@tanstack/ai-react'
import {
  ArticleInputSchema,
  ArticleWorkflowOutputSchema,
} from './article-workflow'

export function ArticleWorkflowDemo() {
  const workflow = useWorkflow({
    input: ArticleInputSchema,
    output: ArticleWorkflowOutputSchema,
    connection: fetchWorkflowEvents('/api/article-workflow'),
  })

  return (
    <section>
      <button
        type="button"
        disabled={workflow.status === 'running'}
        onClick={() =>
          workflow.start({
            topic: 'How typed agent workflows improve AI product UX',
          })
        }
      >
        Start workflow
      </button>

      {workflow.steps.map((step) => (
        <div key={step.stepId}>
          <strong>{step.stepName}</strong>
          <span>{step.status}</span>
        </div>
      ))}

      {workflow.currentText ? <pre>{workflow.currentText}</pre> : null}

      {workflow.pendingApproval ? (
        <div>
          <h2>{workflow.pendingApproval.title}</h2>
          <p>{workflow.pendingApproval.description}</p>
          <button type="button" onClick={() => workflow.approve(true)}>
            Approve
          </button>
          <button
            type="button"
            onClick={() => workflow.approve(false, 'Tighten the conclusion.')}
          >
            Request changes
          </button>
        </div>
      ) : null}

      {workflow.output?.ok ? (
        <article>
          <h1>{workflow.output.article.title}</h1>
          <p>{workflow.output.article.body}</p>
        </article>
      ) : null}

      {workflow.output && !workflow.output.ok ? (
        <p>{workflow.output.reason}</p>
      ) : null}
    </section>
  )
}

There is also useOrchestration for the same runtime with orchestration vocabulary. If your mental model is "run a workflow", use useWorkflow. If your mental model is "let a router pick the next agent", use useOrchestration.还有一个用于相同运行时但带有编排词汇的 useOrchestration。如果你的心智模型是“运行一个工作流”，请使用 useWorkflow。如果你的心智模型是“让路由器选择下一个代理”，请使用 useOrchestration。

tsx

import { fetchWorkflowEvents, useOrchestration } from '@tanstack/ai-react'

export function ArticleOrchestrationDemo() {
  const orchestration = useOrchestration({
    input: ArticleInputSchema,
    output: ArticleWorkflowOutputSchema,
    connection: fetchWorkflowEvents('/api/article-orchestrator'),
  })

  return (
    <button
      type="button"
      disabled={orchestration.status === 'running'}
      onClick={() =>
        orchestration.start({
          topic: 'How typed orchestration improves AI product UX',
        })
      }
    >
      Start orchestration
    </button>
  )
}

Human approval is part of the stream人工批准是流的一部分#

Human-in-the-loop control is not a side channel in this PR.在此 PR 中，人机交互控制不是一个旁路通道。

When workflow code calls yield* approve(...), the runtime emits an approval-requested event, persists the paused run in the run store, and closes the SSE response. The HTTP request is done. The server does not keep a socket open while a user thinks, checks a diff, or gets approval from someone else.当工作流代码调用 yield* approve(...) 时，运行时会触发一个 approval-requested 事件，将暂停的运行持久化到运行存储中，并关闭 SSE 响应。HTTP 请求随之结束。当用户在思考、检查差异或等待他人批准时，服务器不会保持 socket 连接。

When the client calls approve(), it POSTs back to the same endpoint with the run id and approval result. runWorkflow resumes the generator and the next SSE response continues from the paused point.当客户端调用 approve() 时，它会带着运行 ID 和批准结果 POST 回同一个端点。runWorkflow 会恢复生成器，下一个 SSE 响应将从暂停点继续。

That means approvals, revisions, and denial feedback can be modeled in the workflow itself:这意味着批准、修订和拒绝反馈都可以直接在工作流中建模：

const decision =
  yield *
  approve({
    title: 'Publish article?',
    description: edited.notes,
  })

if (!decision.approved) {
  const revised =
    yield *
    agents.editor({
      article: edited.article,
      feedback: decision.feedback ?? 'Revise before publishing.',
    })

  return succeed({ article: revised.article })
}

The same event stream also carries state snapshots and JSON Patch deltas, so a UI can show a live state inspector, a timeline, or a draft preview without waiting for the final result.相同的事件流还携带状态快照和 JSON Patch 增量，因此 UI 可以显示实时状态检查器、时间轴或草稿预览，而无需等待最终结果。

Workflows vs orchestrators工作流与编排器#

Use a workflow when you know the pipeline.当你明确流水线流程时，请使用工作流。

Writer, then editor, then approval is a workflow. Extract topics, draft outline, expand sections is a workflow. Run static checks, ask for approval, deploy to staging, ask again, deploy to production is a workflow.先写作者，再编辑，最后批准，这是一个工作流。提取主题、起草大纲、扩展章节，这是一个工作流。运行静态检查、请求批准、部署到预发布环境、再次请求、部署到生产环境，这是一个工作流。

Use an orchestrator when the next step should be selected turn by turn.当需要轮流选择下一步操作时，请使用编排器。

defineOrchestrator is a thin wrapper over workflows, so the workflow behavior already described applies the same way to orchestrators. It uses the same runtime pieces: typed agents, streaming steps, state snapshots, approvals, SSE transport, and React hooks.defineOrchestrator 是对工作流的轻量级包装，因此上述工作流行为同样适用于编排器。它使用相同的运行时组件：带类型代理、流式步骤、状态快照、批准、SSE 传输和 React hooks。

The body is a router. The router looks at input, mutable state, and the previous result, then returns the next agent to run. When the router is done, it returns the final output.其主体是一个路由器。路由器查看输入、可变状态和上一个结果，然后返回下一个要运行的代理。当路由器完成工作时，它会返回最终输出。

import {
  defineOrchestrator,
  defineRouter,
  succeed,
} from '@tanstack/ai-orchestration'
import { z } from 'zod'
import { ArticleSchema, editor, writer } from './agents'
import {
  ArticleInputSchema,
  ArticleWorkflowOutputSchema,
} from './article-workflow'

const ArticleStateSchema = z.object({
  draft: ArticleSchema.optional(),
  approved: z.boolean().default(false),
})

const articleRouter = defineRouter(
  {
    name: 'article-router',
    state: ArticleStateSchema,
  },
  async function* ({ input, state }) {
    if (!state.draft) {
      return {
        agent: 'writer',
        input: {
          topic: input.topic,
        },
      }
    }

    if (!state.approved) {
      return {
        agent: 'editor',
        input: {
          article: state.draft,
        },
      }
    }

    return {
      done: true,
      output: succeed({
        article: state.draft,
      }),
    }
  },
)

export const articleOrchestrator = defineOrchestrator({
  name: 'article-orchestrator',
  input: ArticleInputSchema,
  output: ArticleWorkflowOutputSchema,
  agents: {
    writer,
    editor,
  },
  router: articleRouter,
})

Routers may also yield if their decision needs async work, but their return value is the important part: either the next { agent, input } pair to run or { done, output } when orchestration is complete.如果路由器的决策需要异步工作，它也可以 yield，但其返回值才是重点：要么是下一个要运行的 { agent, input } 对，要么是编排完成时的 { done, output }。

That gives you the same behavior with a different control-flow style. A workflow puts the sequence directly in the generator body. An orchestrator puts the next-step decision in a router.这为你提供了相同行为但不同的控制流风格。工作流将序列直接放在生成器主体中。编排器则将下一步的决策放在路由器中。

Try it and send feedback试用并发送反馈#

If you are building multi-step AI product flows, this is the moment to test the shape. This is a PR build meant for evaluation, demos, and feedback — the public API can still change before stabilization.如果你正在构建多步 AI 产品流程，现在正是测试这种形式的好时机。这是一个用于评估、演示和反馈的 PR 构建版本——在稳定之前，公共 API 仍可能发生变化。

Install the PR packages directly from pkg.pr.new:直接从 pkg.pr.new 安装 PR 包：

shellshell

npm i https://pkg.pr.new/TanStack/ai/@tanstack/ai-orchestration@542
npm i https://pkg.pr.new/TanStack/ai/@tanstack/ai@542
npm i https://pkg.pr.new/TanStack/ai/@tanstack/ai-react@542
npm i https://pkg.pr.new/TanStack/ai/@tanstack/ai-client@542
npm i https://pkg.pr.new/TanStack/ai/@tanstack/ai-openai@542

Full documentation for this PR lives on the GitHub branch, not the released TanStack docs site:此 PR 的完整文档位于 GitHub 分支上，而不是已发布的 TanStack 文档网站上：