Mastra 持久化 Agent 实战指南：基于 Redis 可续流与三种 Durable 执行模式

Mastra 持久化 Agent 实战指南基于 Redis 可续流与三种 Durable 执行模式【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra本篇技术指南以仓库内 examples/durable-agents 示例为蓝本完整讲解 Mastra 中可续流resumable streams与持久化执行durable execution的实现原理与实战方案。你将掌握createDurableAgent、createEventedAgent、createInngestAgent三种 Agent 工厂的适用场景与配置方法理解 Redis 缓存如何在断线重连时回放事件、observe()如何通过runId续接会话并能直接照抄示例完成本地搭建与联调。一、示例概览一个 Agent 的三种持久化形态示例 README.md 明确指出该项目用同一个基础 Agent包装出三种不同的持久化执行形态且全部基于 Redis 支撑的可续流事件流。三者对比如下Agent工厂函数可续流持久化执行durableResearchAgentcreateDurableAgentRedis无执行仍在 HTTP 请求内eventedResearchAgentcreateEventedAgentRedis内置工作流引擎fire-and-forgetinngestResearchAgentcreateInngestAgentRedisInngest 分布式执行引擎选型逻辑在源码注释中有清晰说明见 research-agent.tscreateDurableAgent只需要断线重连不丢事件的能力执行本身留在 HTTP 请求生命周期内适合单次请求内可完成的中等时长任务createEventedAgent在可续流基础上叠加发后即忘fire-and-forget执行——通过内置工作流引擎的startAsync()把执行移出请求线程适合单实例部署上的长耗时操作createInngestAgent把执行调度交给 Inngest 的分布式任务引擎支持跨进程、跨机器恢复适合生产环境的多实例分布式系统。三种 Agent 都继承Mastra实例上的cache与pubsub配置因此示例代码中创建 Agent 时几乎不需要重复传参eventedResearchAgent显式传入了一个共享的EventEmitterPubSub实例除外。二、环境搭建与启动按 README.md 的 Setup 章节完整步骤如下。1. 启动 Redis可续流依赖 Redis 持久化事件推荐使用 docker 一行启动docker run -d -p 6379:6379 redis示例仓库还提供了更完整的 docker-compose.yaml一次性拉起 Redisredis:8映射6379端口和 Inngest 开发服务器inngest/inngest:v1.34.0映射8288端口并自动把事件回调指到本机4111端口的/inngest/api路由services: redis: image: redis:8 ports: - 6379:6379 inngest: image: inngest/inngest:v1.34.0 command: inngest dev -p 8288 -u http://host.docker.internal:4111/inngest/api --poll-interval1 ports: - 8288:8288 extra_hosts: - host.docker.internal:host-gateway需要 Inngest 形态时也可以不借助 docker-compose单独运行npx inngest-clilatest devREADME 中推荐的方式开发服务器默认端口为8288。2. 安装依赖并启动开发服务器pnpm install pnpm dev开发服务器默认监听4111端口可通过curl直接调用。示例 package.json 中还提供了两个脚本mastra:devmastra dev与start:inngest:server用 inngest-cli 把开发服务器指向http://localhost:3000/inngest/api。该示例通过 pnpm overrides 将mastra/core、mastra/inngest、mastra/redis等全部链接到仓库内的本地源码方便直接阅读与调试。3. 启动 Inngest 开发服务器仅 Inngest Agent 需要npx inngest-clilatest dev三、快速使用发起流式请求与断线续接发起一个流式任务README 使用durable-research-agent的/stream端点发起研究类任务curl -X POST http://localhost:4111/api/agents/durable-research-agent/stream \ -H Content-Type: application/json \ -d {messages: [{role: user, content: Research quantum computing}]}断线后按 runId 续接可续流的核心价值在于连接中断并不等于任务丢失。stream返回的runId就是续接凭证通过/observe端点带着runId重新订阅curl -X POST http://localhost:4111/api/agents/durable-research-agent/observe \ -H Content-Type: application/json \ -d {runId: your-run-id, offset: 5}offset参数从 0 开始计数表示跳过前 N 个已消费的事件、只接收此后的增量省略offset则会把该runId的全部事件重放一遍。这一机制的接口定义同样体现在源码中——observe(runId, options?: { offset?: number; ... })见 create-inngest-agent.ts底层由CachingPubSub与 Redis 缓存协同完成。四、工作原理cache pubsub 如何支撑可续流README 的 How It Works 章节给出了最小化配置这也是理解整套机制的关键import { EventEmitterPubSub } from mastra/core/events; import { RedisServerCache } from mastra/redis; import Redis from ioredis; // Redis cache for resumable streams - events persist across reconnections const cache new RedisServerCache({ client: new Redis(redis://localhost:6379) }); // EventEmitter pubsub for real-time delivery (process-local) const pubsub new EventEmitterPubSub(); export const mastra new Mastra({ cache, pubsub, agents: { durableResearchAgent, // Inherits cache/pubsub eventedResearchAgent, // Inherits cache/pubsub inngestResearchAgent, // Inherits cache/pubsub }, });示例实际运行的 index.ts 比 README 更完整除了RedisServerCache与EventEmitterPubSub还配置了LibSQLStorefile:./mastra.db用于存储运行元数据、PinoLogger日志以及一个挂载在/inngest/api路径上的 Inngest serve 路由const storage new LibSQLStore({ id: mastra-storage, url: file:./mastra.db, }); const cache new RedisServerCache({ client: new Redis(redis://localhost:6379) }); const pubsub new EventEmitterPubSub(); export const mastra new Mastra({ agents: { durableResearchAgent, eventedResearchAgent, inngestResearchAgent, regularResearchAgent }, storage, cache, pubsub, server: { host: 0.0.0.0, apiRoutes: [ { path: /inngest/api, method: ALL, createHandler: async ({ mastra }) inngestServe({ mastra, inngest }), }, ], }, logger: new PinoLogger({ name: Mastra, level: info }), });事件流的核心模型整个可续流机制可以概括为一条规则事件按顺序索引写入缓存observe()调用时先从缓存回放错过的历史事件再无缝切换到实时事件流。即 README 所描述的Events are cached with sequential indices. Whenobserve()is called, missed events replay from cache before continuing with live events.拆开来看cache与pubsub各司其职cache此处为RedisServerCache负责存储给每个事件分配递增的index并持久化。断线重连或新观察者接入时从缓存按索引回放历史事件pubsub此处为EventEmitterPubSub负责实时投递让同进程内的消费者以订阅方式收到新事件。示例中它是进程内实现若需跨进程/跨机器实时投递可替换为基于 Redis Streams、Valkey Streams 或 Google Cloud Pub/Sub 等分布式 pubsub 后端仓库 pubsub 目录提供了redis-streams、valkey-streams、google-cloud-pubsub等实现。值得注意的工程细节可从 create-durable-agent.ts 的选项注释确认不显式传cache时Agent 会继承Mastra实例的serverCache兜底使用InMemoryServerCache显式传cache: false则完全关闭缓存流不可续不显式传pubsub时默认使用EventEmitterPubSubshouldCache?: (topic: string) boolean可以对单个 topic关闭缓存回放直接透传给底层 pubsub只收实时事件换取热 topic 的最小发布延迟run-local内部 topic 无论如何都不进缓存cleanupTimeoutMs默认 30000ms控制持久化流状态的自动清理设为0可关闭自动清理maxSteps限制 agentic loop 的最大步数。Inngest Agent 内部的缓存封装从 create-inngest-agent.ts 源码可以看到Inngest Agent 默认使用InngestPubSub通过 Inngest 的实时通道分发事件并总是用CachingPubSub包裹一层将缓存与 pubsub 打通——否则observe()只能收到订阅之后的实时事件无法回放历史。缓存解析顺序为用户显式传入的cache→ Mastra 实例的serverCache→InMemoryServerCache兜底。若需要在多进程/多机器上跨进程observe必须通过cache或mastra.serverCache提供 Redis 这类共享缓存后端。五、三种 Agent 的源码级拆解5.1 共享的基础 Agent 与工具示例把三个持久化 Agent 都建立在同一个基础配置之上见 research-agent.ts模型openai/gpt-5.5、研究助手指令以及一个演示用webSearch工具用createTool定义输入query模拟 500ms 延迟后返回两条搜索结果。同时导出一个普通 AgentregularResearchAgent用于对照。5.2createDurableAgent纯可续流export const durableResearchAgent createDurableAgent({ agent: new Agent({ id: durable-research-agent, name: Research Agent (Durable), ...baseAgentConfig, }), // cache and pubsub inherited from Mastra });这是官方推荐的最简接入方式。工厂实现create-durable-agent.ts直接构造一个DurableAgent包装原 Agent支持id/name覆盖、cache、pubsub、maxSteps、cleanupTimeoutMs、shouldCache等选项。适用场景仅需断线续接能力、执行留在 HTTP 请求内的场景。5.3createEventedAgent可续流 内置工作流引擎export const eventedResearchAgent createEventedAgent({ agent: new Agent({ id: evented-research-agent, name: Research Agent (Evented), ...baseAgentConfig, }), pubsub, // cache inherited from Mastra });工厂实现create-evented-agent.ts构造一个EventedAgent其执行方式为发后即忘——通过内置工作流引擎的startAsync()把 Agent 执行从 HTTP 请求中摘出来。适用场景单实例部署下的长耗时任务如深度研究报告、批量处理请求可立即返回结果通过事件流消费。5.4createInngestAgent可续流 Inngest 分布式执行export const inngestResearchAgent createInngestAgent({ agent: new Agent({ id: inngest-research-agent, name: Research Agent (Inngest), ...baseAgentConfig, }), inngest, // cache and pubsub inherited from Mastra });这是面向生产分布式系统的高级形态。Inngest 客户端在 inngest.ts 中配置import { realtimeMiddleware } from inngest/realtime/middleware; import { Inngest } from inngest; export const inngest: Inngest new Inngest({ id: durable-agents-example, baseUrl: http://localhost:8288, isDev: true, middleware: [realtimeMiddleware()], });要点解读baseUrl指向本地 Inngest 开发服务器8288端口isDev: true表示开发模式realtimeMiddleware()是必须的——它负责把 Mastra 的流式事件通过 Inngest 的实时通道回传给调用方是可续流 Inngest 执行两条链路能接在一起的粘合剂。从 create-inngest-agent.ts 的选项定义可以确认CreateInngestAgentOptions支持agent被包装的 Agent、inngestInngest 客户端、id/name覆盖、pubsub覆盖默认InngestPubSub、cache启用可续流时提供内部自动包裹CachingPubSub、mastra用于可观测性注册时自动设置。返回的InngestAgent具备完整的方法面stream()、resume()续接被挂起的工作流、prepare()、observe()、generate()/resumeGenerate()等待完整输出的非流式形态并覆盖了onChunk、onStepFinish、onFinish、onError、onSuspended、onAbort、onIterationComplete、abortSignal等回调与控制能力见 create-inngest-agent.ts。注册到Mastra后所需的工作流会自动注册getDurableWorkflows()并通过 Proxy 把listTools()、getMemory()、listAgents()等 Agent 常规方法转发到底层 Agent。六、如何选择三种模式与普通 Agent 的取舍需求推荐形态说明普通流式对话不关心断线AgentregularResearchAgent最简无额外基础设施需要断线重连不丢事件任务可在请求内完成createDurableAgent只引入 Redis 即可长任务 发后即忘 单实例部署createEventedAgent内置工作流引擎调度长任务 生产级分布式 跨进程恢复createInngestAgent依赖 Inngest 开发/生产服务器选择时还需注意基础设施成本纯createDurableAgent只需一个 RediscreateEventedAgent依赖内置工作流引擎单实例内有效createInngestAgent则需要额外运行 Inngest开发环境用npx inngest-clilatest dev或 docker-compose 中的inngest/inngest服务生产环境还需部署 Inngest 平台或自托管 Inngest Server。七、验证与深入阅读示例入口examples/durable-agents/src/mastra/index.tsMastra 装配、examples/durable-agents/src/mastra/agents/research-agent.ts三种 Agent 工厂、examples/durable-agents/src/mastra/workflows/inngest.tsInngest 客户端。核心实现packages/core/src/agent/durable/create-durable-agent.ts、packages/core/src/agent/durable/create-evented-agent.ts、workflows/inngest/src/durable-agent/create-inngest-agent.ts。测试佐证仓库为 durable agent 提供了大量单测与 e2e 覆盖例如 packages/core/src/agent/durable/tests/create-durable-agent.test.ts、packages/core/src/agent/durable/tests/durable-agent-stream.test.ts、packages/core/src/agent/durable/tests/observe-idle-timeout.test.ts、workflows/inngest/src/tests/create-inngest-agent.test.ts可结合测试深入验证断线重连、事件回放、挂起恢复等行为。pubsub 备选后端仓库 pubsub 目录下的redis-streams、valkey-streams、google-cloud-pubsub可替换示例中的进程内EventEmitterPubSub实现跨进程实时投递。【免费下载链接】mastraMastra is the modern TypeScript framework for AI-powered applications and agents.项目地址: https://gitcode.com/GitHub_Trending/ma/mastra创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考