企业官网建设流程全解析

Mastra × Inngest 集成实战：用 @mastra/inngest 为工作流与 Agent 提供持久化执行

【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra

@mastra/inngest是 Mastra 官方提供的 Inngest 集成包，它把 Mastra Workflow 与 Agent 接入 Inngest 的持久化执行引擎，从而获得跨进程重启的耐用性（durable execution）、自动重试（retries）与步骤记忆化（step memoization）。读完本文你将掌握：如何用init()创建 Inngest 后端的工作流原语并配置并发/限流/定时调度，如何用serve()/createServe()将工作流暴露为 Inngest 函数，以及如何用createInngestAgent()让 Agent 在进程崩溃或网络抖动后仍能从断点继续运行。

一、模块定位：Mastra 与 Inngest 的桥接层

Inngest 是一套以事件驱动、步骤记忆化为核心的持久化执行平台：函数内部每个"步骤"的执行结果都会被缓存，重放（replay）时不会重复执行已完成步骤。@mastra/inngest将这套能力以两种形式提供给 Mastra：

• 工作流路线：通过init()创建createWorkflow/createStep等原语，构造出由 Inngest 驱动执行的InngestWorkflow；
• Agent 路线：通过createInngestAgent()将普通 Mastra Agent 包装为"耐用 Agent"，使其单次运行能在进程重启、瞬时故障后存活，并支持挂起（suspend）后恢复（resume）。

两条路线共享同一套底层实现：InngestExecutionEngine（见 execution-engine.ts）负责把工作流步骤翻译成 Inngest 的step.run()等持久化原语；InngestWorkflow（见 workflow.ts）负责把整个工作流编译为一个（或多个）Inngest Function。

二、安装与前置条件

npm install @mastra/inngest

根据 package.json，运行时依赖为inngest（^4.5.0）与@opentelemetry/api，并通过 peerDependencies 约束宿主环境：

本地开发调试时，仓库提供了 docker-compose.yaml 一键拉起 Inngest 开发服务：

services: inngest-test: image: inngest/inngest:v1.34.0 command: inngest dev -p 4000 -u http://host.docker.internal:4001/inngest/api --poll-interval=1 ports: - '4000:4000'

它启动一个监听4000端口的 Inngest Dev Server，并自动轮询4001端口上的 Handler 来发现函数（--poll-interval=1即每秒发现一次）。

三、快速上手：init() 与工作流原语

README 给出的最小接入方式如下：

import { Inngest } from 'inngest'; import { init } from '@mastra/inngest'; const inngest = new Inngest({ id: 'my-app' }); const { createWorkflow, createStep } = init(inngest);

init()返回一组以inngest客户端为后端的原语（实现见 index.ts）：

• createWorkflow(config)：创建一个InngestWorkflow实例（引擎类型为inngest）；
• createStep(...)：创建一个可由 Inngest 引擎执行的步骤；
• createTool(...)：透传核心包的工具工厂；
• cloneStep(step, { id })/cloneWorkflow(workflow, { id })：以新 ID 复制已有步骤/工作流，用于复用定义。

createStep 的多种形态

createStep通过重载支持五种输入（index.ts 中的类型守卫依次判定）：

1. StepParams 显式参数——id、inputSchema、outputSchema、execute，最常用；
2. Agent——直接传入 Mastra Agent，并可通过structuredOutput声明结构化输出，默认输出为{ text: string }；
3. Tool——传入 Mastra 工具对象，可附加retries、scorers、metadata；
4. Processor——包装核心包的处理器（processInput/processInputStep/processOutputStream/processOutputResult/processOutputStep等阶段方法），步骤 ID 自动命名为processor:<id>；
5. InngestWorkflow——原样透传，使嵌套工作流在foreach等场景中被正确识别。

一个完整的步骤 + 工作流示例（参考 index.test.ts 与适配器测试 express.integration.test.ts 的写法）：

import { z } from 'zod'; import { Mastra } from '@mastra/core'; const { createWorkflow, createStep } = init(inngest); const step1 = createStep({ id: 'step1', inputSchema: z.object({ input: z.string() }), outputSchema: z.object({ value: z.string() }), execute: async ({ inputData }) => ({ value: `${inputData.input}-step1` }), }); const step2 = createStep({ id: 'step2', inputSchema: z.object({ value: z.string() }), outputSchema: z.object({ result: z.string() }), execute: async ({ inputData }) => ({ result: `${inputData.value}-step2` }), }); const workflow = createWorkflow({ id: 'my-workflow', inputSchema: z.object({ input: z.string() }), outputSchema: z.object({ result: z.string() }), steps: [step1, step2], }); workflow.then(step1).then(step2).commit(); const mastra = new Mastra({ workflows: { myWorkflow: workflow }, });

Mastra实例注册工作流后，即可通过workflow.createRun()+run.start({ inputData })触发执行，也可以把执行入口交给 Inngest 的函数发现机制。

四、工作流配置详解：流控与定时调度

InngestWorkflowConfig（见 types.ts）在核心包WorkflowConfig之上叠加了两类 Inngest 专属配置。

流控配置（Flow Control）

InngestFlowControlConfig直接提取自 InngestcreateFunction的参数类型，包括五个可选项：

在 workflow.ts 的构造函数中，这些字段会被从参数中剥离，与cron一起单独保存，其余参数才交给父类Workflow；随后在生成 Inngest 函数时通过展开运算符（...this.flowControlConfig）注入createFunction（workflow.ts）。

const workflow = createWorkflow({ id: 'rate-limited-workflow', inputSchema: z.object({}), outputSchema: z.object({}), steps: [step1], concurrency: 10, // 最多 10 个并发 rateLimit: { limit: 100, period: '1h' }, // 每小时 100 次 priority: 1, // 提高调度优先级 });

定时调度（Cron）

InngestFlowCronConfig提供三个字段：

• cron：标准 cron 表达式，例如'0 9 * * *'；
• inputData：每次定时触发时注入工作流的输入数据；
• initialState：每次定时触发时注入的初始状态。

仅当cron存在时，createCronFunction()（workflow.ts）才会生成一个独立的 Inngest 函数workflow.<id>.cron（retries: 0，cancelOn: cancel.workflow.<id>，触发器为 cron 表达式），其内部先createRun()再run.start()。getFunctions()（workflow.ts）最终返回主函数、可选的 cron 函数，以及图中所有嵌套InngestWorkflow各自对应的函数——嵌套工作流在 Inngest 中是以独立 Function 形式运行的。

五、服务暴露：serve()、createServe() 与 connect()

默认 Hono 路由

serve（见 serve.ts）是封装了inngest/hono适配器的默认入口：

import { serve } from '@mastra/inngest'; app.use('/inngest/api', async (c) => { return serve({ mastra, inngest })(c); });

createServe()：适配任意 Web 框架

createServe(adapter)是一个高阶工厂，接收 Inngest 官方任意框架的serve适配器（Express / Fastify / Next.js / Koa / Hono 等），并自动完成函数收集。三个官方示例：

// Express —— 需要先挂载 JSON 中间件（测试见 express.integration.test.ts） import { serve } from 'inngest/express'; const serveExpress = createServe(serve); app.use(express.json()); app.use('/inngest/api', serveExpress({ mastra, inngest })); // Fastify import { serve } from 'inngest/fastify'; const serveFastify = createServe(serve); fastify.route({ method: ['GET', 'POST', 'PUT'], handler: serveFastify({ mastra, inngest }), url: '/inngest/api', }); // Next.js（App Router：app/inngest/api/route.ts） import { serve } from 'inngest/next'; const serveNext = createServe(serve); export const { GET, POST, PUT } = serveNext({ mastra, inngest });

prepareServeOptions()内部调用collectInngestFunctions()（见 functions.ts）：遍历mastra.listWorkflows()，凡是InngestWorkflow实例都执行__registerMastra(mastra)并收集getFunctions()的完整函数列表，再与用户自定义functions合并。

connect()：出站 Worker 模式

当进程不便暴露入站 HTTP 端点时，可改用 connect.ts 提供的connect()，它以inngest/connect建立出站长连接：

import { connect } from '@mastra/inngest/connect'; await connect({ mastra, inngest });

如果既没有InngestWorkflow也没有额外functions，connect()会发出警告（否则 Worker 将空转无事可做）。registerOptions中的字段（如signingKey）优先级高于顶层选项，与serve()的行为保持一致。

六、底层执行引擎：记忆化、重试与持久化痕迹

所有 Inngest 工作流的执行都经由InngestExecutionEngine（execution-engine.ts），它继承核心包的DefaultExecutionEngine并针对 Inngest 覆盖关键行为。

步骤记忆化（Memoization）

wrapDurableOperation()（execution-engine.ts）把每个步骤包进this.inngestStep.run(operationId, ...)。Inngest 以operationId为键缓存步骤结果：进程重启后重放时，已完成步骤直接返回缓存，不重复执行。值得注意的是，它刻意把序列化错误放进cause字段——因为 Inngest 的错误序列化只保留标准 Error 属性，AI SDK 等来源的自定义属性（如statusCode）通过cause+ 自定义toJSON()得以保留。

重试策略

函数级retries被固定为0（workflow.ts），因为重试在步骤级由executeStepWithRetry()（execution-engine.ts）手动处理：循环retries + 1次，每次通过AsyncLocalStorage记录重试计数，非可重试错误（MastraNonRetryableError/ Inngest 的NonRetriableError）立即短路。此外：

• executeSleepDuration()/executeSleepUntilDate()分别映射到 Inngest 的step.sleep()/step.sleepUntil()，实现持久化的等待；
• 工作流快照（persistWorkflowSnapshot/loadWorkflowSnapshot）由InngestRun与 workflows storage 协作维护，挂起/恢复、resume事件都会先读取快照再继续（见 workflow.ts 与 run.ts）。

可观测性：耐用 Span

Span 的创建与结束同样被记忆化（createStepSpan/endStepSpan/errorStepSpan及对应的 child 系列，见 execution-engine.ts）：首次执行创建并exportSpan()，重放时通过rebuildSpan()恢复、不重复创建，从而让一条 trace 完整贯穿多次重放。

取消的精确作用域

主函数通过cancelOn: [{ event: 'cancel.workflow.<id>', match: 'data.runId' }]（workflow.ts）把取消事件精确限定到单个运行——源码注释特别说明，若不加match，取消一次运行会波及同一部署下所有共享函数的运行。直接向触发器事件发送而未携带runId的调用会被警告"无法按 ID 取消"（workflow.ts）。

七、createInngestAgent()：Agent 的耐用执行

当 Agent 运行需要"扛过进程重启与瞬时故障"时，使用createInngestAgent()。官方示例（见 create-inngest-agent.ts 的文档注释）：

import { Agent } from '@mastra/core/agent'; import { createInngestAgent } from '@mastra/inngest'; import { Inngest } from 'inngest'; const inngest = new Inngest({ id: 'my-app' }); const agent = new Agent({ id: 'my-agent', name: 'My Agent', instructions: 'You are a helpful assistant', model: openai('gpt-4'), }); const durableAgent = createInngestAgent({ agent, inngest }); const mastra = new Mastra({ agents: { myAgent: durableAgent }, }); // 使用该 Agent const { output, cleanup } = await durableAgent.stream('Hello!'); const text = await output.text; cleanup();

返回的InngestAgent可以像普通 Agent 一样注册进 Mastra，其必需的耐用工作流会被自动注册；运行时通过 Proxy 把generate、listTools、getMemory等未显式声明的 Agent 方法转发给底层 Agent。

工厂参数（CreateInngestAgentOptions）

完整的 API 面

• stream(messages, options?)→{ output, runId, threadId?, resourceId?, cleanup, abort, fullStream }：启动一次耐用流式运行；
• resume(runId, resumeData, options?)：恢复被挂起的运行，options.toolCallId可精确定位某个挂起的工具调用叶子；
• prepare(...)：只做耐用执行准备（生成runId、快照请求上下文），不真正启动；
• observe(runId, { offset? })：断线重连已有流，offset指定从第几个事件开始回放，缺省重放全部；
• generate(...)/resumeGenerate(...)：把流式运行收束为单个FullOutput；若运行因工具审批等挂起，finishReason为'suspended'；
• abort(reason?)：同时翻转本进程的AbortController，并通过 pubsub 发布中止请求让步骤 Worker 优雅收尾（见requestRemoteAbort，create-inngest-agent.ts）。

底层：耐用的 Agentic Loop 工作流

createInngestDurableAgenticWorkflow（create-inngest-agentic-workflow.ts）构建的工作流包含四个阶段：

1. LLM 执行步骤——调用模型获取响应/工具调用；
2. 工具调用步骤（foreach）——每个工具调用作为独立步骤执行，支持 suspend；
3. LLM 映射步骤——把工具结果合并回状态；
4. 循环——仍有工具调用需要处理则继续（dowhile）。

所有状态都经由工作流输入/输出流转，因此跨进程重启与引擎重放都是安全的。该工作流的 ID 以inngest:为前缀（InngestDurableStepIds，见 create-inngest-agentic-workflow.ts），避免与其他引擎的工作流 ID 冲突。

实时事件：InngestPubSub

InngestPubSub（pubsub.ts）把 Mastra 的PubSub抽象桥接到 Inngest Realtime：

publish()使用inngest.realtime.publish()（非持久化、立即执行，函数内自动附带当前 runId）；subscribe()使用inngest/realtime建立 WebSocket 订阅，并且会先等连接就绪再触发工作流，避免"事件先于订阅到达"的竞态。Agent 路线还会把 pubsub 包一层CachingPubSub（缓存解析顺序：用户传入 >mastra.serverCache> 内存缓存），使observe()能回放历史事件。

八、测试与本地联调

仓库自带完整的测试矩阵（见 package.json 的 scripts）：

集成测试（如 express.integration.test.ts）展示了完整的端到端姿势：启动 Web 服务挂载/inngest/api，createRun()后run.start()，再断言各步骤输出——这正是第六节所述InngestWorkflow执行链路的可运行验证。

九、版本与演进

版本历史与发布说明见包内 CHANGELOG.md。从源码结构可以推断，该集成对 Inngest SDK v4 的realtime.publish()、inngest/connect等新 API 有深度依赖，升级inngest依赖时建议同步回归上述测试套件，尤其关注 pubsub 通道命名与取消事件match行为的变化。

【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra