Mastra 与 Inngest 集成实战构建带可观测性的持久化 AI 工作流【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra导读本文将围绕 examples/inngest/README.md 所演示的完整示例讲解如何在 Mastra现代 TypeScript AI 应用框架中集成 Inngest构建一个获取天气 → 按降水概率分支规划活动的多步骤 AI 工作流并覆盖三个进阶主题基于mastra/observability的追踪可观测性、Inngest 原生的流程控制并发、限流、节流、防抖、优先级队列以及基于 Inngest 持久化执行的 durable agent可跨崩溃恢复的 AI Agent 循环。读完本文你将掌握mastra/inngest包的init/createWorkflow/createStep/serve/createInngestAgent完整调用链能够把 Mastra 工作流无缝嵌入 Inngest 事件驱动架构。示例概览为什么用 Inngest 编排 Mastra 工作流Mastra 自带默认执行引擎但当工作流需要事件驱动、定时触发、重试、持久化执行、以及与外部系统深度集成时Inngest 是一个成熟的事件编排层。mastra/inngest包源码位于 workflows/inngest/src将 Mastra 的Workflow与 Inngest 的createFunction机制对接Mastra 的createStep定义的步骤会被编译为 Inngest function可独立重试、暂停、恢复Mastra 的Workflow图形编排.then、.branch、.parallel等映射为 Inngest 的 step 执行模型每个步骤LLM 调用、工具调用都被 Inngest 检查点化checkpoint服务崩溃后可自动恢复这正是后文durable agent的基础。从源码看init(inngest)返回{ createWorkflow, createStep, cloneStep, cloneWorkflow, createTool }见 workflows/inngest/src/index.ts而InngestWorkflow继承自核心的Workflow类并将引擎类型标记为inngest见 workflows/inngest/src/workflow.ts这意味着你可以在不改变步骤定义习惯的前提下获得 Inngest 的全部能力。环境准备与本地开发示例的依赖与启动脚本见 examples/inngest/package.json核心安装命令如下npm install mastra/inngest inngest mastra/core mastra/deployer hono/node-server同时需要启动 Inngest Dev Server端口 8288官方镜像方式docker run --rm -p 8288:8288 \ inngest/inngest:v1.18.0 \ inngest dev -u http://host.docker.internal:3000/inngest/api注意版本前提需要inngest^4以及 Inngest Dev Serverv1.18.0或更高版本。Inngest SDK v4 已将 realtime 内建因此示例不再使用inngest/realtime与realtimeMiddleware。在示例仓库中package.json也提供了两个便捷脚本通过 pnpm 运行注意仓库使用pnpm10.18.2# Terminal 1启动 Inngest Dev Server pnpm start:inngest:server # Terminal 2启动 Mastra dev server 与 Studio pnpm mastra:dev其中start:inngest:server实际执行npx inngest-clilatest dev -u http://localhost:3000/inngest/api即把 Inngest 事件投递地址指向运行在本机 3000 端口的 Mastra 服务。定义规划 AgentPlanning Agent工作流的核心智能来自一个复用 LLM 调用的 planning agent。示例代码位于agents/planning-agent.ts// agents/planning-agent.ts import { Agent } from mastra/core/agent; // Create a new planning agent that uses the OpenAI model const planningAgent new Agent({ id: planning-agent, name: planningAgent, model: openai/gpt-5.1, instructions: You are a local activities and travel expert who excels at weather-based planning. Analyze the weather data and provide practical activity recommendations. [Day, Month Date, Year] ═══════════════════════════ ️ WEATHER SUMMARY • Conditions: [brief description] • Temperature: [X°C/Y°F to A°C/B°F] • Precipitation: [X% chance] MORNING ACTIVITIES Outdoor: • [Activity Name] - [Brief description including specific location/route] Best timing: [specific time range] Note: [relevant weather consideration] AFTERNOON ACTIVITIES Outdoor: • [Activity Name] - [Brief description including specific location/route] Best timing: [specific time range] Note: [relevant weather consideration] INDOOR ALTERNATIVES • [Activity Name] - [Brief description including specific venue] Ideal for: [weather condition that would trigger this alternative] ⚠️ SPECIAL CONSIDERATIONS • [Any relevant weather warnings, UV index, wind conditions, etc.] Guidelines: - Suggest 2-3 time-specific outdoor activities per day - Include 1-2 indoor backup options - For precipitation 50%, lead with indoor activities - All activities must be specific to the location - Include specific venues, trails, or locations - Consider activity intensity based on temperature - Keep descriptions concise but informative Maintain this exact formatting for consistency, using the emoji and section headers as shown. , }); export { planningAgent };要点解析instructions中定义了严格的输出模板时段划分、emoji 小标题、建议数量与触发条件保证不同天气输入下输出格式一致该 Agent 后续通过mastra.getAgent(planningAgent)在工作流步骤内部被动态获取因此必须在 Mastra 实例中注册见下文注册一节。初始化 Inngest 并定义工作流工作流定义在workflows/inngest-workflow.ts。首先通过init()把 Mastra 的createWorkflow/createStep绑定到指定的 Inngest 实例// workflows/inngest-workflow.ts import { init } from mastra/inngest; import { Inngest } from inngest; import { z } from zod; const { createWorkflow, createStep } init( new Inngest({ id: mastra, baseUrl: http://localhost:8288, }), );baseUrl指向本地 Inngest Dev Server。示例还定义了一个工具函数把 Open-Meteo 的 WMO 天气代码转换为可读描述以及贯穿多个步骤的forecastSchema日期、最高/最低温、降水概率、天气描述、地点后续步骤的输入/输出都基于该 Zod schemafunction getWeatherCondition(code: number): string { const conditions: Recordnumber, string { 0: Clear sky, 1: Mainly clear, 2: Partly cloudy, 3: Overcast, 45: Foggy, 48: Depositing rime fog, 51: Light drizzle, 53: Moderate drizzle, 55: Dense drizzle, 61: Slight rain, 63: Moderate rain, 65: Heavy rain, 71: Slight snow fall, 73: Moderate snow fall, 75: Heavy snow fall, 95: Thunderstorm, }; return conditions[code] || Unknown; } const forecastSchema z.object({ date: z.string(), maxTemp: z.number(), minTemp: z.number(), precipitationChance: z.number(), condition: z.string(), location: z.string(), });Step 1抓取指定城市的天气数据const fetchWeather createStep({ id: fetch-weather, description: Fetches weather forecast for a given city, inputSchema: z.object({ city: z.string(), }), outputSchema: forecastSchema, execute: async ({ inputData }) { if (!inputData) { throw new Error(Trigger data not found); } // Get latitude and longitude for the city const geocodingUrl https://geocoding-api.open-meteo.com/v1/search?name${encodeURIComponent(inputData.city)}count1; const geocodingResponse await fetch(geocodingUrl); const geocodingData (await geocodingResponse.json()) as { results: { latitude: number; longitude: number; name: string }[]; }; if (!geocodingData.results?.[0]) { throw new Error(Location ${inputData.city} not found); } const { latitude, longitude, name } geocodingData.results[0]; // Fetch weather data using the coordinates const weatherUrl https://api.open-meteo.com/v1/forecast?latitude${latitude}longitude${longitude}currentprecipitation,weathercodetimezoneauto,hourlyprecipitation_probability,temperature_2m; const response await fetch(weatherUrl); const data (await response.json()) as { current: { time: string; precipitation: number; weathercode: number }; hourly: { precipitation_probability: number[]; temperature_2m: number[] }; }; const forecast { date: new Date().toISOString(), maxTemp: Math.max(...data.hourly.temperature_2m), minTemp: Math.min(...data.hourly.temperature_2m), condition: getWeatherCondition(data.current.weathercode), location: name, precipitationChance: data.hourly.precipitation_probability.reduce((acc, curr) Math.max(acc, curr), 0), }; return forecast; }, });该步骤展示了createStep的完整形态id、description、inputSchema、outputSchema与execute。从源码看createStep支持四种输入显式参数、Agent、Tool、Processor并通过类型守卫自动分派见 workflows/inngest/src/index.ts这里使用的是显式StepParams形式。注意execute的解构参数inputData严格受inputSchema约束Schema 校验贯穿整个工作流。Step 2基于天气建议活动室内/室外混合const planActivities createStep({ id: plan-activities, description: Suggests activities based on weather conditions, inputSchema: forecastSchema, outputSchema: z.object({ activities: z.string(), }), execute: async ({ inputData, mastra }) { const forecast inputData; if (!forecast) { throw new Error(Forecast data not found); } const prompt Based on the following weather forecast for ${forecast.location}, suggest appropriate activities: ${JSON.stringify(forecast, null, 2)} ; const agent mastra?.getAgent(planningAgent); if (!agent) { throw new Error(Planning agent not found); } const response await agent.stream([ { role: user, content: prompt, }, ]); let activitiesText ; for await (const chunk of response.textStream) { process.stdout.write(chunk); activitiesText chunk; } return { activities: activitiesText, }; }, });Step 3仅规划室内活动雨天备用const planIndoorActivities createStep({ id: plan-indoor-activities, description: Suggests indoor activities based on weather conditions, inputSchema: forecastSchema, outputSchema: z.object({ activities: z.string(), }), execute: async ({ inputData, mastra }) { const forecast inputData; if (!forecast) { throw new Error(Forecast data not found); } const prompt In case it rains, plan indoor activities for ${forecast.location} on ${forecast.date}; const agent mastra?.getAgent(planningAgent); if (!agent) { throw new Error(Planning agent not found); } const response await agent.stream([ { role: user, content: prompt, }, ]); let activitiesText ; for await (const chunk of response.textStream) { process.stdout.write(chunk); activitiesText chunk; } return { activities: activitiesText, }; }, });Step 2/3 的关键点是execute上下文中的mastra对象它由执行引擎注入允许步骤内按id动态获取已注册的 Agentmastra.getAgent(planningAgent)并通过agent.stream(...)流式获取 LLM 输出逐 chunk 写入process.stdout并拼接为最终文本。组装工作流分支branch实现 if-else 逻辑三个步骤通过createWorkflow(...).then(...).branch([...])编排。branch接收条件-步骤对数组按顺序匹配第一个返回true的分支const activityPlanningWorkflow createWorkflow({ id: activity-planning-workflow-step2-if-else, inputSchema: z.object({ city: z.string().describe(The city to get the weather for), }), outputSchema: z.object({ activities: z.string(), }), }) .then(fetchWeather) .branch([ [ // If precipitation chance is greater than 50%, suggest indoor activities async ({ inputData }) { return inputData?.precipitationChance 50; }, planIndoorActivities, ], [ // Otherwise, suggest a mix of activities async ({ inputData }) { return inputData?.precipitationChance 50; }, planActivities, ], ]); activityPlanningWorkflow.commit(); export { activityPlanningWorkflow };执行流程为输入{ city }→fetchWeather得到forecastSchema结构 → 依据precipitationChance是否大于 50 选择planIndoorActivities下雨或planActivities非下雨。commit()用于固化工作流定义必须被调用之后工作流方可注册与执行。注册到 Mastra 实例并暴露 Inngest API 路由在入口文件index.ts中把 Agent 与工作流注册到Mastra实例并在server.apiRoutes中挂载 Inngest 的 serve 处理器// index.ts import { Mastra } from mastra/core; import { serve as inngestServe } from mastra/inngest; import { PinoLogger } from mastra/loggers; import { Inngest } from inngest; import { activityPlanningWorkflow } from ./workflows/inngest-workflow; import { planningAgent } from ./agents/planning-agent; // Create an Inngest instance for workflow orchestration and event handling // Realtime is built into the SDK in v4, so no middleware is needed. const inngest new Inngest({ id: mastra, baseUrl: http://localhost:8288, // URL of your local Inngest server isDev: true, }); // Create and configure the main Mastra instance export const mastra new Mastra({ workflows: { activityPlanningWorkflow, }, agents: { planningAgent, }, server: { host: 0.0.0.0, apiRoutes: [ { path: /inngest/api, // API endpoint for Inngest to send events to method: ALL, createHandler: async ({ mastra }) inngestServe({ mastra, inngest }), }, ], }, logger: new PinoLogger({ name: Mastra, level: info, }), });此处inngestServe即mastra/inngest导出的 Hono 版 serve 函数源码见 workflows/inngest/src/serve.ts它会把 Mastra 中已注册工作流编译出的 Inngest functions 与你的inngestclient 绑定。若你的服务不是 HonocreateServe(adapter)支持接入任意 Inngest 适配器如 Express、Fastify、Next.js见 serve.ts。启动服务并执行工作流exec.ts演示了如何在本地启动 HTTP 服务端口 3000供 Inngest 投递事件并从 mastra 实例获取工作流、创建 run 并启动// exec.ts import { mastra } from ./; import { serve } from hono/node-server; import { createHonoServer, getToolExports } from mastra/deployer/server; import { tools } from #tools; const app await createHonoServer(mastra, { tools: getToolExports(tools), }); // Start the server on port 3000 so Inngest can send events to it const srv serve({ fetch: app.fetch, port: 3000, }); const workflow mastra.getWorkflow(activityPlanningWorkflow); const run await workflow.createRun(); // Start the workflow with the required input data (city name) // This will trigger the workflow steps and stream the result to the console const result await run.start({ inputData: { city: New York } }); console.dir(result, { depth: null }); // Close the server after the workflow run is complete srv.close();执行流程要点先启动 Hono 服务Inngest Dev Server 才能把事件 POST 到http://localhost:3000/inngest/apimastra.getWorkflow(activityPlanningWorkflow)按 id 取回工作流workflow.createRun()创建一次运行源码见 workflows/inngest/src/workflow.ts会在 storage 中持久化 pending 快照run.start({ inputData: { city: New York } })触发执行并流式输出步骤结果执行完成后关闭服务。运行期间可在 Inngest 控制台http://localhost:8288实时查看工作流运行的每个 step 及其状态。可观测性追踪工作流执行全过程控制台追踪输出示例内建了 Mastra observability工作流运行时会打印如下 trace 事件 SPAN_STARTED Type: workflow_run Name: activity-planning-workflow-step2-if-else ID: span-id Trace ID: trace-id ──────────────────────────────────────────────────────────────────────────────── SPAN_STARTED Type: workflow_step Name: fetch-weather ... ──────────────────────────────────────────────────────────────────────────────── ✅ SPAN_ENDED Type: workflow_step Name: fetch-weather Duration: 1234ms Output: { date: ..., maxTemp: 25, ... }该输出证明 Mastra 的 observability 能够捕获工作流执行workflow_runspan单个步骤执行workflow_stepspanAgent/模型调用agent_run、model_generationspan步骤输入与输出耗时信息从源码层面看mastra/observability包observability/mastra/src提供了完整的 span 生命周期管理开始/结束/错误span 类型包括workflow_run、workflow_step等见 observability/mastra/src/spans/base.ts其单元测试也对workflow_branching_trace、workflow_child_spans_trace等场景做了快照验证见 observability/mastra/src/snapshots佐证了上述追踪能力的实现。配置 Exporters在index.ts中通过mastra/observability的Observability类配置默认导出器serviceName、采样策略、exporters 列表import { Observability, ConsoleExporter, MastraStorageExporter } from mastra/observability; const observability new Observability({ configs: { default: { serviceName: inngest-workflow-example, sampling: { type: always }, // Sample all traces exporters: [ new ConsoleExporter(), // Logs traces to console new MastraStorageExporter(), // Persists traces to storage ], }, }, }); export const mastra new Mastra({ // ... other config observability, });sampling: { type: always }表示采样所有 trace另有基于 trace 级别的采样策略见 observability/mastra/src/trace-level-sampling.test.tsConsoleExporter把 span 输出到控制台实现见 observability/mastra/src/exporters/console.tsMastraStorageExporter把 trace 持久化到 Mastra storage实现见 observability/mastra/src/exporters/mastra-storage.ts。生产环境 Exporter 示例Langfuseimport { LangfuseExporter } from mastra/langfuse; new LangfuseExporter({ publicKey: process.env.LANGFUSE_PUBLIC_KEY, secretKey: process.env.LANGFUSE_SECRET_KEY, });Datadogimport { DatadogExporter } from mastra/datadog; new DatadogExporter({ mlApp: my-app, apiKey: process.env.DD_API_KEY, });OpenTelemetry以 SigNoz 为例import { OtelExporter } from mastra/otel-exporter; new OtelExporter({ provider: { signoz: { endpoint: https://ingest.signoz.io, apiKey: process.env.SIGNOZ_API_KEY, }, }, });对应 exporter 包在仓库中的位置Langfuse 见 observability/langfuseDatadog 见 observability/datadogOTel exporter 见 observability/otel-exporter。生产环境通常建议将sampling调整为按需采样并把 exporter 替换为上述云服务之一ConsoleExporter仅用于本地调试。Inngest 流程控制并发、限流、节流、防抖与优先级Inngest 工作流原生支持高级流程控制能力帮助在大规模执行时防止资源过载。mastra/inngest把这些配置直接透传给 Inngest 的createFunction从类型定义看InngestFlowControlConfig就是PickcreateFunction 参数, concurrency | rateLimit | throttle | debounce | priority见 workflows/inngest/src/types.ts并在InngestWorkflow构造函数中把非空项收集为flowControlConfig见 workflows/inngest/src/workflow.ts最终合并进 Inngest function 配置。此外配置还支持cron定时触发与inputData/initialState初始值见 types.ts。并发控制Concurrency限制同一时间并行执行的工作流实例数可按key粒度如按用户维度划分const workflow createWorkflow({ id: user-processing-workflow, inputSchema: z.object({ userId: z.string() }), outputSchema: z.object({ result: z.string() }), steps: [processUserStep], // Limit to 10 concurrent executions, scoped by user ID concurrency: { limit: 10, key: event.data.userId, // Per-user concurrency }, });限流Rate Limiting限制单位时间内的执行次数const workflow createWorkflow({ id: api-sync-workflow, inputSchema: z.object({ endpoint: z.string() }), outputSchema: z.object({ status: z.string() }), steps: [apiSyncStep], // Maximum 1000 executions per hour rateLimit: { period: 1h, limit: 1000, }, });节流Throttling保证两次执行之间的最小间隔常用于防止通知轰炸const workflow createWorkflow({ id: email-notification-workflow, inputSchema: z.object({ organizationId: z.string(), message: z.string() }), outputSchema: z.object({ sent: z.boolean() }), steps: [sendEmailStep], // Only one execution per 10 seconds per organization throttle: { period: 10s, limit: 1, key: event.data.organizationId, }, });防抖Debouncing延迟执行直到时间窗口内不再有新事件到达适合搜索索引、自动保存等场景const workflow createWorkflow({ id: search-index-workflow, inputSchema: z.object({ documentId: z.string() }), outputSchema: z.object({ indexed: z.boolean() }), steps: [indexDocumentStep], // Wait 5 seconds of no updates before indexing debounce: { period: 5s, key: event.data.documentId, }, });优先级队列Priority Queuing为工作流设置执行优先级数值越小越先执行支持运行时动态计算const workflow createWorkflow({ id: order-processing-workflow, inputSchema: z.object({ orderId: z.string(), priority: z.number().optional(), }), outputSchema: z.object({ processed: z.boolean() }), steps: [processOrderStep], // Higher priority orders execute first priority: { run: event.data.priority ?? 50, // Dynamic priority, default 50 }, });组合使用多种流程控制所有控制项均为可选且可叠加未配置时采用 Inngest 默认行为const workflow createWorkflow({ id: comprehensive-workflow, inputSchema: z.object({ userId: z.string(), organizationId: z.string(), priority: z.number().optional(), }), outputSchema: z.object({ result: z.string() }), steps: [comprehensiveStep], // Multiple flow control features concurrency: { limit: 5, key: event.data.userId, }, rateLimit: { period: 1m, limit: 100, }, throttle: { period: 10s, limit: 1, key: event.data.organizationId, }, priority: { run: event.data.priority ?? 0, }, });注意示例中的steps: [...]与前面createWorkflow({...}).then(...).branch(...)是两种等价的工作流定义方式——前者以数组声明步骤并附加流程控制配置后者以链式 API 描述执行图流程控制配置在两种方式下都生效。所有流程控制配置由 Inngest 原生实现校验确保兼容性与正确性。Durable Agents可跨崩溃恢复的持久化 AI Agent示例还包含两个durable agents——通过 Inngest 持久化执行实现的 AI Agent 循环每一步LLM 调用、工具执行都被检查点化服务崩溃后自动从断点恢复而非重新开始。示例自带的两个 Durable AgentResearch Agentresearch-agent——带网页搜索工具的简单 AgentFile Manager Agentfile-manager-agent——演示危险操作的工具审批delete-file需要人工批准。运行方式需同时启动两个终端# Terminal 1: Inngest dev server pnpm start:inngest:server # Terminal 2: Mastra dev server studio pnpm mastra:dev然后在 Mastra Studio 中像普通 Agent 一样与之交互并可在http://localhost:8288的 Inngest 控制台监控每次运行的执行轨迹。创建自己的 Durable Agent用createInngestAgent包装任意 Mastra Agent 即可获得持久化执行能力import { createInngestAgent } from mastra/inngest; import { Agent } from mastra/core/agent; const myAgent new Agent({ id: my-agent, model: openai/gpt-4o, instructions: You are a helpful assistant., tools: {/* your tools */}, }); // Wrap with durable execution export const durableAgent createInngestAgent({ agent: myAgent, inngest, }); // Register in mastra config - workflows auto-register export const mastra new Mastra({ agents: { durableAgent }, });从源码看createInngestAgent(options)返回InngestAgentTOutput并配套了isInngestAgent类型守卫见 workflows/inngest/src/durable-agent/create-inngest-agent.ts。其内部把 Agent 的每一步执行编译为 Inngest 持久化步骤并利用InngestPubSub见 workflows/inngest/src/pubsub.ts转发流式 chunk/finish 事件使 Agent 的observe()能够回放缓存的历史输出。仓库中对应测试覆盖了挂起元数据、恢复上下文、完成副作用等持久化场景见 workflows/inngest/src/tests可作为深入阅读的参考。关键结论mastra/inngest把 Mastra 工作流无缝映射到 Inngest 的事件驱动执行模型createStep定义的标准步骤即可获得重试、检查点与恢复能力分支、并行等图编排.then/.branch在 Inngest 引擎下正常运作inputData受 Zod schema 严格约束观察性开箱即用ObservabilityConsoleExporter/MastraStorageExporter可立即看到workflow_run/workflow_step/agent_run等 span生产环境可替换为 Langfuse、Datadog 或 OpenTelemetry 导出器流程控制并发、限流、节流、防抖、优先级直接透传 Inngest 原生能力全部可选、可组合durable agent 是这套集成的进阶价值LLM 调用与工具调用被检查点化天然抗服务崩溃适合长时运行的 AI Agent 场景。【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考