尧图网站建设 
Mastra × Inngest 集成实战用 mastra/inngest 为工作流与 Agent 提供持久化执行【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastramastra/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 驱动执行的InngestWorkflowAgent 路线通过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 约束宿主环境依赖版本要求mastra/core1.58.0-0 2.0.0-0zod^3.25.0 \|\| ^4.0.0Node.js22.13.0本地开发调试时仓库提供了 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-interval1 ports: - 4000:4000它启动一个监听4000端口的 Inngest Dev Server并自动轮询4001端口上的 Handler 来发现函数--poll-interval1即每秒发现一次。三、快速上手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.tscreateWorkflow(config)创建一个InngestWorkflow实例引擎类型为inngestcreateStep(...)创建一个可由 Inngest 引擎执行的步骤createTool(...)透传核心包的工具工厂cloneStep(step, { id })/cloneWorkflow(workflow, { id })以新 ID 复制已有步骤/工作流用于复用定义。createStep 的多种形态createStep通过重载支持五种输入index.ts 中的类型守卫依次判定StepParams 显式参数——id、inputSchema、outputSchema、execute最常用Agent——直接传入 Mastra Agent并可通过structuredOutput声明结构化输出默认输出为{ text: string }Tool——传入 Mastra 工具对象可附加retries、scorers、metadataProcessor——包装核心包的处理器processInput/processInputStep/processOutputStream/processOutputResult/processOutputStep等阶段方法步骤 ID 自动命名为processor:idInngestWorkflow——原样透传使嵌套工作流在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 ControlInngestFlowControlConfig直接提取自 InngestcreateFunction的参数类型包括五个可选项配置项作用concurrency限制同一函数的最大并发执行数rateLimit按时间窗限制调用速率throttle节流时间窗内最多执行 N 次debounce防抖合并高频触发静默一段时间后才执行priority为执行任务设置优先级影响调度顺序在 workflow.ts 的构造函数中这些字段会被从参数中剥离与cron一起单独保存其余参数才交给父类Workflow随后在生成 Inngest 函数时通过展开运算符...this.flowControlConfig注入createFunctionworkflow.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, // 提高调度优先级 });定时调度CronInngestFlowCronConfig提供三个字段cron标准 cron 表达式例如0 9 * * *inputData每次定时触发时注入工作流的输入数据initialState每次定时触发时注入的初始状态。仅当cron存在时createCronFunction()workflow.ts才会生成一个独立的 Inngest 函数workflow.id.cronretries: 0cancelOn: 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.jsApp Routerapp/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也没有额外functionsconnect()会发出警告否则 Worker 将空转无事可做。registerOptions中的字段如signingKey优先级高于顶层选项与serve()的行为保持一致。六、底层执行引擎记忆化、重试与持久化痕迹所有 Inngest 工作流的执行都经由InngestExecutionEngineexecution-engine.ts它继承核心包的DefaultExecutionEngine并针对 Inngest 覆盖关键行为。步骤记忆化MemoizationwrapDurableOperation()execution-engine.ts把每个步骤包进this.inngestStep.run(operationId, ...)。Inngest 以operationId为键缓存步骤结果进程重启后重放时已完成步骤直接返回缓存不重复执行。值得注意的是它刻意把序列化错误放进cause字段——因为 Inngest 的错误序列化只保留标准 Error 属性AI SDK 等来源的自定义属性如statusCode通过cause 自定义toJSON()得以保留。重试策略函数级retries被固定为0workflow.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。可观测性耐用 SpanSpan 的创建与结束同样被记忆化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参数说明agent被包装的 Mastra Agent必填inngestInngest 客户端必填id/name覆盖 Agent 默认 ID / 名称pubsub覆盖默认的InngestPubSubcache提供缓存实例以启用可恢复流resumable streams配合CachingPubSub实现断线重连不丢事件mastra可观测性所需的 Mastra 实例注册时自动设置完整的 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为suspendedabort(reason?)同时翻转本进程的AbortController并通过 pubsub 发布中止请求让步骤 Worker 优雅收尾见requestRemoteAbortcreate-inngest-agent.ts。底层耐用的 Agentic Loop 工作流createInngestDurableAgenticWorkflowcreate-inngest-agentic-workflow.ts构建的工作流包含四个阶段LLM 执行步骤——调用模型获取响应/工具调用工具调用步骤foreach——每个工具调用作为独立步骤执行支持 suspendLLM 映射步骤——把工具结果合并回状态循环——仍有工具调用需要处理则继续dowhile。所有状态都经由工作流输入/输出流转因此跨进程重启与引擎重放都是安全的。该工作流的 ID 以inngest:为前缀InngestDurableStepIds见 create-inngest-agentic-workflow.ts避免与其他引擎的工作流 ID 冲突。实时事件InngestPubSubInngestPubSubpubsub.ts把 Mastra 的PubSub抽象桥接到 Inngest Realtime主题topicInngest channel / topicworkflow.events.v2.{runId}workflow:{workflowId}:{runId}/watchagent.stream.{runId}agent:{runId}/agent-streampublish()使用inngest.realtime.publish()非持久化、立即执行函数内自动附带当前 runIdsubscribe()使用inngest/realtime建立 WebSocket 订阅并且会先等连接就绪再触发工作流避免事件先于订阅到达的竞态。Agent 路线还会把 pubsub 包一层CachingPubSub缓存解析顺序用户传入 mastra.serverCache 内存缓存使observe()能回放历史事件。八、测试与本地联调仓库自带完整的测试矩阵见 package.json 的 scripts命令覆盖范围pnpm test全量单元/集成测试排除适配器目录pnpm test:unitcreateInngestAgent工厂单测pnpm test:suite耐用 Agent 套件挂起/恢复、副作用、上下文等pnpm test:workflowInngest Engine 工作流执行pnpm test:integrationExpress / Fastify / Hono / Koa 适配器集成测试pnpm test:docker先docker-compose up -d再跑测试并回收容器集成测试如 express.integration.test.ts展示了完整的端到端姿势启动 Web 服务挂载/inngest/apicreateRun()后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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Mastra × Inngest 集成实战：用 @mastra/inngest 为工作流与 Agent 提供持久化执行
Mastra × Inngest 集成实战用 mastra/inngest 为工作流与 Agent 提供持久化执行【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastramastra/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 驱动执行的InngestWorkflowAgent 路线通过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 约束宿主环境依赖版本要求mastra/core1.58.0-0 2.0.0-0zod^3.25.0 \|\| ^4.0.0Node.js22.13.0本地开发调试时仓库提供了 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-interval1 ports: - 4000:4000它启动一个监听4000端口的 Inngest Dev Server并自动轮询4001端口上的 Handler 来发现函数--poll-interval1即每秒发现一次。三、快速上手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.tscreateWorkflow(config)创建一个InngestWorkflow实例引擎类型为inngestcreateStep(...)创建一个可由 Inngest 引擎执行的步骤createTool(...)透传核心包的工具工厂cloneStep(step, { id })/cloneWorkflow(workflow, { id })以新 ID 复制已有步骤/工作流用于复用定义。createStep 的多种形态createStep通过重载支持五种输入index.ts 中的类型守卫依次判定StepParams 显式参数——id、inputSchema、outputSchema、execute最常用Agent——直接传入 Mastra Agent并可通过structuredOutput声明结构化输出默认输出为{ text: string }Tool——传入 Mastra 工具对象可附加retries、scorers、metadataProcessor——包装核心包的处理器processInput/processInputStep/processOutputStream/processOutputResult/processOutputStep等阶段方法步骤 ID 自动命名为processor:idInngestWorkflow——原样透传使嵌套工作流在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 ControlInngestFlowControlConfig直接提取自 InngestcreateFunction的参数类型包括五个可选项配置项作用concurrency限制同一函数的最大并发执行数rateLimit按时间窗限制调用速率throttle节流时间窗内最多执行 N 次debounce防抖合并高频触发静默一段时间后才执行priority为执行任务设置优先级影响调度顺序在 workflow.ts 的构造函数中这些字段会被从参数中剥离与cron一起单独保存其余参数才交给父类Workflow随后在生成 Inngest 函数时通过展开运算符...this.flowControlConfig注入createFunctionworkflow.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, // 提高调度优先级 });定时调度CronInngestFlowCronConfig提供三个字段cron标准 cron 表达式例如0 9 * * *inputData每次定时触发时注入工作流的输入数据initialState每次定时触发时注入的初始状态。仅当cron存在时createCronFunction()workflow.ts才会生成一个独立的 Inngest 函数workflow.id.cronretries: 0cancelOn: 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.jsApp Routerapp/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也没有额外functionsconnect()会发出警告否则 Worker 将空转无事可做。registerOptions中的字段如signingKey优先级高于顶层选项与serve()的行为保持一致。六、底层执行引擎记忆化、重试与持久化痕迹所有 Inngest 工作流的执行都经由InngestExecutionEngineexecution-engine.ts它继承核心包的DefaultExecutionEngine并针对 Inngest 覆盖关键行为。步骤记忆化MemoizationwrapDurableOperation()execution-engine.ts把每个步骤包进this.inngestStep.run(operationId, ...)。Inngest 以operationId为键缓存步骤结果进程重启后重放时已完成步骤直接返回缓存不重复执行。值得注意的是它刻意把序列化错误放进cause字段——因为 Inngest 的错误序列化只保留标准 Error 属性AI SDK 等来源的自定义属性如statusCode通过cause 自定义toJSON()得以保留。重试策略函数级retries被固定为0workflow.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。可观测性耐用 SpanSpan 的创建与结束同样被记忆化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参数说明agent被包装的 Mastra Agent必填inngestInngest 客户端必填id/name覆盖 Agent 默认 ID / 名称pubsub覆盖默认的InngestPubSubcache提供缓存实例以启用可恢复流resumable streams配合CachingPubSub实现断线重连不丢事件mastra可观测性所需的 Mastra 实例注册时自动设置完整的 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为suspendedabort(reason?)同时翻转本进程的AbortController并通过 pubsub 发布中止请求让步骤 Worker 优雅收尾见requestRemoteAbortcreate-inngest-agent.ts。底层耐用的 Agentic Loop 工作流createInngestDurableAgenticWorkflowcreate-inngest-agentic-workflow.ts构建的工作流包含四个阶段LLM 执行步骤——调用模型获取响应/工具调用工具调用步骤foreach——每个工具调用作为独立步骤执行支持 suspendLLM 映射步骤——把工具结果合并回状态循环——仍有工具调用需要处理则继续dowhile。所有状态都经由工作流输入/输出流转因此跨进程重启与引擎重放都是安全的。该工作流的 ID 以inngest:为前缀InngestDurableStepIds见 create-inngest-agentic-workflow.ts避免与其他引擎的工作流 ID 冲突。实时事件InngestPubSubInngestPubSubpubsub.ts把 Mastra 的PubSub抽象桥接到 Inngest Realtime主题topicInngest channel / topicworkflow.events.v2.{runId}workflow:{workflowId}:{runId}/watchagent.stream.{runId}agent:{runId}/agent-streampublish()使用inngest.realtime.publish()非持久化、立即执行函数内自动附带当前 runIdsubscribe()使用inngest/realtime建立 WebSocket 订阅并且会先等连接就绪再触发工作流避免事件先于订阅到达的竞态。Agent 路线还会把 pubsub 包一层CachingPubSub缓存解析顺序用户传入 mastra.serverCache 内存缓存使observe()能回放历史事件。八、测试与本地联调仓库自带完整的测试矩阵见 package.json 的 scripts命令覆盖范围pnpm test全量单元/集成测试排除适配器目录pnpm test:unitcreateInngestAgent工厂单测pnpm test:suite耐用 Agent 套件挂起/恢复、副作用、上下文等pnpm test:workflowInngest Engine 工作流执行pnpm test:integrationExpress / Fastify / Hono / Koa 适配器集成测试pnpm test:docker先docker-compose up -d再跑测试并回收容器集成测试如 express.integration.test.ts展示了完整的端到端姿势启动 Web 服务挂载/inngest/apicreateRun()后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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考