> Discover all available pages from the documentation index: https://mastra.zisheng.pro/llms.txt

# 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 概览](https://www.youtube.com/watch?v=7It2y89TVP4)，了解 Signal 如何唤醒和引导长时间运行的 Agent。

## 何时使用 Signal

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

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

## 快速开始

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

```typescript
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()`。

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

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

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

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

### 将消息排队到下一轮

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

```typescript
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` 可更改此行为。

```typescript
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()` 参考](https://mastra.zisheng.pro/reference/agents/agent)。

### 发送通知上下文

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

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

```typescript
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 上下文：

```xml
<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](https://mastra.zisheng.pro/reference/storage/libsql)、[PostgreSQL](https://mastra.zisheng.pro/reference/storage/postgresql) 和 [MongoDB](https://mastra.zisheng.pro/reference/storage/mongodb)。这些 adapter 通过 `getStore('notifications')` 公开通知记录。

### 发送 processor 上下文

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

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

```typescript
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'`，因此模型会按如下形式接收此上下文：

```xml
<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()` 参考](https://mastra.zisheng.pro/reference/agents/agent)和 [`Agent.sendSignal()` 参考](https://mastra.zisheng.pro/reference/agents/agent)。

## 状态与通知 Signal

### 状态 Signal

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

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

```typescript
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()` 参考](https://mastra.zisheng.pro/reference/agents/agent)。

```typescript
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()` 参考](https://mastra.zisheng.pro/reference/agents/agent)，`notifications.deliveryPolicy` 配置请参阅 [`Agent` 构造函数参考](https://mastra.zisheng.pro/reference/agents/agent)，inbox Tool 操作请参阅 [`createNotificationInboxTool()` 参考](https://mastra.zisheng.pro/reference/signals/create-notification-inbox-tool)。

```typescript
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',
  },
)
```

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

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

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

```xml
<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` 构造函数参考](https://mastra.zisheng.pro/reference/agents/agent)，运行时通知 dispatch 配置请参阅 [`Mastra` 类参考](https://mastra.zisheng.pro/reference/core/mastra-class)。

#### 通知 inbox Tool

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

`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 能够跨实例协调：

```typescript
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 指南](https://mastra.zisheng.pro/docs/server/pubsub)和 [`RedisStreamsPubSub` 参考](https://mastra.zisheng.pro/reference/pubsub/redis-streams)。

## 兼容性与 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 参考](https://mastra.zisheng.pro/reference/agents/agent)。

### 批准 Tool 调用

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

### 使用 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 路由参考](https://mastra.zisheng.pro/reference/server/routes)。

### 使用客户端 SDK

JavaScript 客户端公开 thread Signal API。

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

```typescript
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()` 参考](https://mastra.zisheng.pro/reference/client-js/agents)。

### 保持自定义 SSE 订阅活跃

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

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

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

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

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

## 相关内容

- [`Agent.sendMessage()`](https://mastra.zisheng.pro/reference/agents/agent)
- [`Agent.queueMessage()`](https://mastra.zisheng.pro/reference/agents/agent)
- [`Agent.sendSignal()`](https://mastra.zisheng.pro/reference/agents/agent)
- [`Agent.sendStateSignal()`](https://mastra.zisheng.pro/reference/agents/agent)
- [`Agent.subscribeToThread()`](https://mastra.zisheng.pro/reference/agents/agent)
- [`createNotificationInboxTool()`](https://mastra.zisheng.pro/reference/signals/create-notification-inbox-tool)
- [`client.getAgent().sendMessage()`](https://mastra.zisheng.pro/reference/client-js/agents)
- [`client.getAgent().queueMessage()`](https://mastra.zisheng.pro/reference/client-js/agents)
- [`client.getAgent().sendSignal()`](https://mastra.zisheng.pro/reference/client-js/agents)
- [Server Agent 路由](https://mastra.zisheng.pro/reference/server/routes)
- [`client.getAgent().subscribeToThread()`](https://mastra.zisheng.pro/reference/client-js/agents)
- [`client.getAgent().sendToolApproval()`](https://mastra.zisheng.pro/reference/client-js/agents)
- [`RedisStreamsPubSub`](https://mastra.zisheng.pro/reference/pubsub/redis-streams)
- 📹 [Mastra Signal workshop](https://www.youtube.com/watch?v=KLg6uFKz9aw\&t=3020s)