Signal

添加于： @mastra/core@1.39.0

beta

此功能处于 beta 阶段。在 API 稳定之前，可能会在不提升主版本号的情况下发生破坏性变更。

Signal 是一种通过 thread 与 Agent 交互的方式。无需每次交互都从 agent.stream() 开始，你可以订阅 thread，然后发送消息或 Signal。Mastra 会在 thread 空闲时唤醒 Agent，将输入放入正在运行的 Agent 循环，或将输入排队到下一轮。

使用消息 API 处理用户输入。使用 sendSignal() 处理较低层级的系统上下文，例如后台任务通知、策略提醒或 processor 生成的上下文。

📹 观看

观看 Mastra Signal 概览，了解 Signal 如何唤醒和引导长时间运行的 Agent。

何时使用 Signal

当 Agent thread 需要原始 stream() 调用之外的新输入或上下文时，请使用 Signal。用户在运行活跃期间发送后续消息、后台系统需要向 thread 添加上下文，或外部事件应唤醒、更新或通知 Agent 时，Signal 都很有用。

用户输入请使用 sendMessage() 和 queueMessage()。较低层级的系统上下文请使用 sendSignal()。持久化状态通道请使用 sendStateSignal()；外部事件需要创建持久化通知 inbox 记录时，请使用 sendNotificationSignal()。

快速开始

创建 Agent，订阅 thread，然后向该 thread 发送消息。当消息唤醒 Agent 或进入正在运行的循环时，订阅会接收活跃 stream。

src/mastra/signals.ts

import { Agent } from '@mastra/core/agent'

const agent = new Agent({
  id: 'support-agent',
  name: 'Support Agent',
  instructions: 'Help the user compare options.',
  model: 'openai/gpt-5.6-sol',
})

const thread = {
  resourceId: 'user_123',
  threadId: 'thread_456',
}

const subscription = await agent.subscribeToThread(thread)

await agent.sendMessage('Compare that with the previous option.', thread)

for await (const chunk of subscription.stream) {
  console.log(chunk)
}

当 thread 有正在运行的 Agent stream 时，sendMessage() 会成为该 Agent 循环中的新输入。当 thread 空闲时，Mastra 会以该消息作为首个输入启动 stream。

消息输入

立即发送消息

当用户希望活跃 Agent 立即看到消息时，请使用 sendMessage()。

src/mastra/signals.ts

agent.sendMessage(
  {
    contents: 'Use the latest customer note too.',
    attributes: { name: 'Jane', sentFrom: 'slack' },
  },
  {
    resourceId: 'user_123',
    threadId: 'thread_456',
  },
)

模型会以 XML 包装的用户输入形式接收带属性的消息：

<user name="Jane" sentFrom="slack">Use the latest customer note too.</user>

没有属性的消息会作为纯用户输入发送。

将消息排队到下一轮

当用户发送后续消息，但当前模型调用应先完成时，请使用 queueMessage()。Mastra 会等待活跃运行完成，然后在同一 thread 上启动新运行。

src/mastra/signals.ts

agent.queueMessage('Also check whether the tests need updates.', {
  resourceId: 'user_123',
  threadId: 'thread_456',
})

当 thread 空闲时，queueMessage() 会立即启动运行。当 thread 活跃时，它会在活跃运行完成后启动新运行，以保持轮次顺序。

Signal 上下文

控制低层级 Signal 行为

需要发送系统生成的上下文而不是用户输入时，请使用 sendSignal()。对于外部事件，使用 type: 'notification'。默认情况下，Mastra 会将 Signal 传递到活跃运行并唤醒空闲 thread。使用 ifActive.behavior 和 ifIdle.behavior 可更改此行为。

src/mastra/signals.ts

const result = agent.sendSignal(
  {
    type: 'notification',
    contents: 'GitHub CI failed on PR #123: 3 tests failed.',
  },
  {
    resourceId: 'user_123',
    threadId: 'thread_456',
    ifIdle: {
      behavior: 'persist',
    },
  },
)

await result.persisted

当空闲唤醒 stream 需要模型设置、Tool 或运行时上下文等选项时，请传入 ifIdle.streamOptions。ifActive、ifIdle、分支属性和 streamOptions 请参阅 Agent.sendSignal() 参考。

发送通知上下文

Signal 具有语义 type 和面向 LLM 的 tagName。使用 type 描述 Signal 类别；使用 tagName 控制模型看到的 XML 标签。

对于外部事件，使用 type: 'notification'。Reactive Signal 保留给 processor 或运行时生成的上下文，例如策略指导、后台任务结果和自动加载的指令。

src/mastra/signals.ts

agent.sendSignal(
  {
    type: 'notification',
    contents: 'PR #123 has a new review comment from User X about the API surface.',
    attributes: {
      source: 'github',
      pr: '123',
    },
  },
  {
    resourceId: 'user_123',
    threadId: 'thread_456',
  },
)

模型会按如下形式接收 Signal 上下文：

<notification source="github" pr="123">PR #123 has a new review comment from User X about the API surface.</notification>

使用符合 XML 规范的 tagName 和属性名称。它们可以包含字母、数字、下划线、句点和连字符，并且必须以字母或下划线开头。

存储支持

通知 inbox 存储可用于支持更丰富 memory 和 Signal Workflow 的 storage adapter：libSQL、PostgreSQL 和 MongoDB。这些 adapter 通过 getStore('notifications') 公开通知记录。

发送 processor 上下文

Processor 可以在运行期间发送 Reactive Signal。Processor 应检查聊天历史，对特定触发条件作出反应，并避免多次发送相同上下文。

以下示例演示一个 processor：Tool 调用读取 AGENTS.md 文件后，该 processor 会注入 AGENTS.md 指令。

src/mastra/processors/agents-md-reminder.ts

import type { Processor, ProcessInputStepArgs } from '@mastra/core/processors'

export const agentsMdReminderProcessor: Processor = {
  id: 'agents-md-reminder',
  async processInputStep({ messageList, sendSignal }: ProcessInputStepArgs) {
    const messages = messageList.get.all.db()
    const agentsMdPath = findAgentsMdPathFromToolCalls(messages)

    if (!agentsMdPath || hasAlreadySentAgentsMdReminder(messages, agentsMdPath)) {
      return messageList
    }

    await sendSignal?.({
      type: 'reactive',
      contents: readAgentsMdInstructions(agentsMdPath),
      attributes: {
        type: 'dynamic-agents-md',
        path: agentsMdPath,
      },
      metadata: {
        path: agentsMdPath,
      },
    })

    return messageList
  },
}

Reactive Signal 默认为 tagName: 'system-reminder'，因此模型会按如下形式接收此上下文：

<system-reminder type="dynamic-agents-md" path="packages/ui/AGENTS.md">
  $agentsMdFileContents
</system-reminder>

等待 sendSignal() 可在订阅的 thread 活跃时保持 stream echo 顺序。

条件属性

使用 ifActive.attributes 和 ifIdle.attributes，根据 Agent 在传递时处于活跃还是空闲状态，为输入标记上下文。顶层 attributes 始终应用；接受输入时，Mastra 会将所选分支的 attributes 合并进去。分支专用属性请参阅 Agent.sendMessage() 参考和 Agent.sendSignal() 参考。

状态与通知 Signal

状态 Signal

状态 Signal 会公开命名的、thread 范围的上下文通道。可用于随时间变化的持久化上下文，例如浏览器状态、编辑器状态或后台 watcher 结果。

当外部 producer 检测到状态变化时，使用 sendStateSignal()。每个状态 Signal 都会标识状态通道、由 producer 管理的 cache key，以及更新是 snapshot 还是 delta。

src/mastra/browser-watcher.ts

await agent.sendStateSignal(
  {
    id: 'browser',
    mode: 'snapshot',
    cacheKey: 'browser:https://example.com:3-tabs',
    contents: 'Browser is open. Active tab URL: https://example.com. 3 open tabs.',
    value: {
      activeUrl: 'https://example.com',
      tabCount: 3,
      open: true,
    },
  },
  {
    resourceId: 'user_123',
    threadId: 'thread_456',
  },
)

Mastra 接受状态 Signal 后，会在 thread 上存储紧凑的跟踪元数据。如果在该状态仍为当前状态时，producer 再次发送相同的 cacheKey 和 mode，Mastra 会跳过重复项。

当 processor 管理状态通道时，使用 computeStateSignal()。Mastra 会在每个模型输入步骤中，于 processInputStep() 之后调用一次。状态 Signal 字段和返回值请参阅 Agent.sendStateSignal() 参考。

src/mastra/processors/browser-state.ts

import type { ComputeStateSignalArgs, Processor } from '@mastra/core/processors'

export const browserStateProcessor: Processor = {
  id: 'browser-state',
  stateId: 'browser',
  computeStateSignal(args: ComputeStateSignalArgs) {
    const browser = readCurrentBrowserState()
    const previous = readMostRecentBrowserState(args.activeStateSignals)
    const changed = previous ? diffBrowserState(previous, browser) : browser
    const shouldRefreshSnapshot = Boolean(args.lastSnapshot && !args.contextWindow.hasSnapshot)

    if (previous && Object.keys(changed).length === 0 && !shouldRefreshSnapshot) {
      return
    }

    const isDelta = Boolean(previous && !shouldRefreshSnapshot)

    return {
      mode: isDelta ? 'delta' : 'snapshot',
      cacheKey: stableBrowserStateCacheKey(browser),
      contents: isDelta ? describeBrowserDelta(changed) : describeBrowserSnapshot(browser),
      value: browser,
      ...(isDelta ? { delta: changed } : {}),
    }
  },
}

Mastra 会将 lastSnapshot 和 deltasSinceSnapshot 传入 computeStateSignal()。当当前消息列表不包含最新 snapshot 时，它会从消息历史解析这些值。Processor 仍负责合并和 diff 逻辑。

contextWindow.hasSnapshot 会告诉 processor，活跃消息窗口是否已包含该状态通道的 snapshot。如果为 false，请返回新的 snapshot，这样即使较旧状态消息已从上下文窗口裁剪，模型仍能看到当前状态。

内置浏览器上下文 processor 使用 browser id，以 snapshot 和 delta 模式发出状态。

通知 Signal

通知 Signal 表示 GitHub 活动、电子邮件、Slack mention、CI 状态、incident、录制内容或私信等外部事件。当事件应创建持久化 inbox 记录时，请使用 agent.sendNotificationSignal()。

通知传递分为两个阶段。Ingress 阶段，agent.sendNotificationSignal() 存储通知记录并解析 Agent 的传递策略。Dispatch 阶段，Mastra 消费到期记录并发出完整通知或摘要 Signal。

默认传递策略会感知优先级。紧急通知会立即传递，较低优先级通知则可能批量汇总为摘要，或等到 thread 空闲。通知字段请参阅 Agent.sendNotificationSignal() 参考，notifications.deliveryPolicy 配置请参阅 Agent 构造函数参考，inbox Tool 操作请参阅 createNotificationInboxTool() 参考。

src/mastra/notifications.ts

await agent.sendNotificationSignal(
  {
    source: 'github',
    kind: 'ci-status',
    priority: 'high',
    summary: 'CI failed on main: 3 tests failed.',
    payload: {
      repository: 'acme/app',
      branch: 'main',
    },
    dedupeKey: 'github:acme/app:main:ci',
  },
  {
    resourceId: 'user_123',
    threadId: 'thread_456',
  },
)

模型会以如下上下文形式接收完整通知：

<notification source="github" type="ci-status" priority="high" status="delivered">CI failed on main: 3 tests failed.</notification>

通知摘要会告诉模型，inbox 中有记录正在等待：

<notification-summary pending="10">github: 3, email: 5, slack: 2</notification-summary>

Mastra 发出摘要时，会清除每条已汇总记录的 summaryAt 并设置 summarySignalId。记录保持 pending 且可读取。Mastra 发出完整通知时，会设置 deliveredSignalId 并将记录标记为 delivered。如果 inbox Tool 先读取通知，它可以注入完整通知 Signal 并将记录标记为 seen，以防止重复完整传递。

当部分通知应等待其他 dispatch 窗口或摘要汇总时，请在 Agent 上配置传递策略。当延迟通知和摘要汇总应自动传递时，请在 Mastra 层级启用定时 dispatch。notifications.deliveryPolicy 请参阅 Agent 构造函数参考，运行时通知 dispatch 配置请参阅 Mastra 类参考。

通知 inbox Tool

使用 createNotificationInboxTool() 为 Agent 提供一个 inbox 操作 Tool，而不是多个 CRUD Tool。当 Agent 在收到 <notification-summary> Signal 后需要摘要背后的完整记录时，请使用 read。通知内容以 Signal 传递，而不是作为普通 Tool 输出。设置示例、输入 schema 和 action 行为请参阅 createNotificationInboxTool() 参考。

sendNotificationSignal() 需要支持 notifications 的 storage domain。仅当较低层级的通知形状上下文应绕过 inbox 存储时，才使用 sendSignal({ type: 'notification' })。

分布式与 serverless 部署

Signal 通过 pub/sub 后端协调运行。当 Signal 到达实现了 LeaseProvider 的后端时，Mastra 会获取目标 thread 的 lease，确保同一时间只有一个进程拥有该对话；随后唤醒 Agent，或将输入路由到正在运行的循环。没有 lease 的后端会回退到始终授予所有权的 no-op，这适合单进程，但不适合跨实例。

默认的内存 pub/sub 无法跨越实例边界。在 Vercel 等 serverless 平台或任何多实例部署中，后续 Signal 可能被路由到不同于正在运行 Agent 的实例。

如果没有共享 pub/sub，该实例无法访问活跃运行，只能自行启动一个运行，使原运行不受影响，并导致 thread 被处理两次。

在 Mastra 实例上配置由 Redis Streams 支持的共享 pub/sub，使 lease 和 Signal 能够跨实例协调：

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { RedisStreamsPubSub } from '@mastra/redis-streams'

export const mastra = new Mastra({
  agents: { agent },
  pubsub: new RedisStreamsPubSub({
    url: process.env.REDIS_URL,
    keyPrefix: 'mastra:my-app',
  }),
})

RedisStreamsPubSub 同时实现事件传递约定和分布式 lease，因此一个后端即可处理跨实例 Signal 传递和 lease 所有权。Vercel 托管的 Redis 集成和 Upstash Redis 都很合适。有关何时需要分布式 pub/sub，请参阅 PubSub 指南和 RedisStreamsPubSub 参考。

兼容性与 API

兼容性

Mastra 仍接受 type: 'user-message' 和 type: 'system-reminder' 等旧版 Signal payload，并在内部将其规范化为新的类别和标签结构：

• type: 'user-message'：规范化为 type: 'user' 和 tagName: 'user'
• type: 'system-reminder'：规范化为 type: 'reactive' 和 tagName: 'system-reminder'

已存储的现有 Signal 行和旧客户端会继续通过兼容层加载。Server 支持时，新客户端会调用消息路由；检测到旧版 server 时，React 的 thread Signal 路径会回退到旧版 /signals 路由。完整消息、Signal 和订阅类型请参阅 Agent Signal 参考。

批准 Tool 调用

订阅运行因 Tool 审批暂停时，请使用订阅原生方法批准或拒绝 Tool 调用。恢复后的 chunk 会通过现有 thread 订阅到达。请求和响应结构请参阅 client.getAgent().sendToolApproval() 参考和 server Agent 路由。

使用 HTTP 路由

如果通过 HTTP 直接调用 Mastra，请使用 POST /api/agents/:agentId/send-message 发送即时消息，使用 POST /api/agents/:agentId/queue-message 发送下一轮消息。订阅原生 Tool 审批请使用 POST /api/agents/:agentId/send-tool-approval。请求和响应 schema 请参阅 Server 路由参考。

使用客户端 SDK

JavaScript 客户端公开 thread Signal API。

发送 thread 输入前使用 subscribeToThread()，使客户端可以呈现接收输入或因输入而唤醒的 stream。

src/app/chat.ts

const agent = client.getAgent('supportAgent')

const subscription = await agent.subscribeToThread({
  resourceId: 'user_123',
  threadId: 'thread_456',
})

await agent.sendMessage({
  message: 'Show the shorter version.',
  resourceId: 'user_123',
  threadId: 'thread_456',
})

await subscription.processDataStream({
  onChunk: chunk => {
    console.log(chunk)
  },
  reconnect: true,
})

长时间订阅请使用 reconnect: true。重连选项请参阅 client.getAgent().subscribeToThread() 参考。

保持自定义 SSE 订阅活跃

如果为 thread 订阅公开自己的 Server-Sent Events（SSE）endpoint，请在 stream 空闲时定期发送 heartbeat frame。这样可以防止浏览器、proxy 和 load balancer 在下一个 Signal 或模型 chunk 到达前关闭连接。

以下示例每 25 秒发送一条 SSE 注释：

src/api/subscribe.ts

const heartbeat = setInterval(() => {
  controller.enqueue(encoder.encode(': keep-alive\n\n'))
}, 25_000)

request.signal.addEventListener('abort', () => {
  clearInterval(heartbeat)
})

将 heartbeat 与客户端重连逻辑结合使用。Heartbeat 可减少空闲断开，而重连可在网络或运行时仍关闭 stream 时进行恢复。

相关内容

• Agent.sendMessage()
• Agent.queueMessage()
• Agent.sendSignal()
• Agent.sendStateSignal()
• Agent.subscribeToThread()
• createNotificationInboxTool()
• client.getAgent().sendMessage()
• client.getAgent().queueMessage()
• client.getAgent().sendSignal()
• Server Agent 路由
• client.getAgent().subscribeToThread()
• client.getAgent().sendToolApproval()
• RedisStreamsPubSub
• 📹 Mastra Signal workshop