Subagents

添加于： @mastra/core@1.8.0

子 Agents 是可以让另一个 Agent 委派任务的专门 Agent。将它们添加到父 Agent 的 agents 属性，然后调用 Agent.stream() 或 Agent.generate()。父 Agent 会根据自身 instructions 和每个子 Agent 的 description 决定何时以及如何委派任务。

何时使用 subagents

当任务需要具有不同专长的 agents 协同工作时，使用子 Agents。父 Agent 决定何时委派，并将上下文传给每个子 Agent，然后综合它们的结果。

常见用例：

• 研究与写作流程：一个 Agent 收集数据，另一个 Agent 产出内容
• 需要在每个阶段使用不同专长的多步骤任务
• 需要对委派行为进行精细控制的任务

备注

协调子 Agents 的父 Agent 通常称为监督者（supervisor）。监督者模式是在 Mastra 中构建多 Agent 系统的一种方式。其他模式参见概念概览。

快速开始

用清晰的描述定义子 Agents，然后将它们添加到父 Agent：

src/mastra/agents/parent-agent.ts

import { Agent } from '@mastra/core/agent'
import { Memory } from '@mastra/memory'
import { LibSQLStore } from '@mastra/libsql'

const researchAgent = new Agent({
  id: 'research-agent',
  description: 'Gathers factual information and returns bullet-point summaries.',
  model: 'openai/gpt-5-mini',
})

const writingAgent = new Agent({
  id: 'writing-agent',
  description: 'Transforms research into well-structured articles.',
  model: 'openai/gpt-5-mini',
})

const parentAgent = new Agent({
  id: 'parent-agent',
  instructions: `You coordinate research and writing using specialized agents.
    Delegate to research-agent for facts, then writing-agent for content.`,
  model: 'openai/gpt-5.6-sol',
  agents: { researchAgent, writingAgent },
  memory: new Memory({
    storage: new LibSQLStore({ id: 'storage', url: 'file:mastra.db' }),
  }),
})

const stream = await parentAgent.stream('Research AI in education and write an article', {
  maxSteps: 10,
})

for await (const chunk of stream.textStream) {
  process.stdout.write(chunk)
}

委派 hooks

委派 hooks 让你在委派发生时拦截、修改或拒绝委派。在 delegation 选项下配置它们，可放在 Agent 的 defaultOptions 中，也可按调用配置。

onDelegationStart

在父 Agent 委派给子 Agent 之前调用。返回一个对象以控制委派：

• proceed: true：允许委派（默认行为）
• proceed: false：以 rejectionReason 拒绝委派
• modifiedPrompt：重写发送给子 Agent 的 prompt
• modifiedMaxSteps：限制子 Agent 的迭代次数
• threadMetadata：在子 Agent 线程创建时附加元数据（例如 tenantId），以便通过 memory.listThreads({ filter: { metadata } }) 过滤线程

src/mastra/agents/parent-agent.ts

const stream = await parentAgent.stream('Research AI trends', {
  maxSteps: 10,
  delegation: {
    onDelegationStart: async context => {
      console.log(`Delegating to: ${context.primitiveId}`)

      // Modify the prompt for a specific agent
      if (context.primitiveId === 'research-agent') {
        return {
          proceed: true,
          modifiedPrompt: `${context.prompt}\n\nFocus on 2024-2025 data.`,
          modifiedMaxSteps: 5,
        }
      }

      // Reject delegation after too many iterations
      if (context.iteration > 8) {
        return {
          proceed: false,
          rejectionReason: 'Max iterations reached. Synthesize current findings.',
        }
      }

      return { proceed: true }
    },
  },
})

context 对象包含：

属性	说明
primitiveId	被委派的子 Agent ID
prompt	父 Agent 正在发送的 prompt
iteration	当前迭代编号
requestContext	子 Agent 运行将收到的请求上下文

委派边界处的请求上下文

每次委派都会收到一个请求上下文，其条目从父运行浅拷贝而来，但不包含运行作用域的身份键。在子 Agent 运行期间设置或删除条目不会影响父上下文。在 onDelegationStart 中对 context.requestContext 设置条目，可将值传给被委派的运行：

src/mastra/agents/parent-agent.ts

const stream = await parentAgent.stream('Research AI trends', {
  maxSteps: 10,
  delegation: {
    onDelegationStart: async context => {
      context.requestContext.set('audience', 'technical')
    },
  },
})

子 Agent 在其工具和动态配置（例如 instructions: ({ requestContext }) => ...）中读取这些条目。详情参见请求上下文。值必须可 JSON 序列化，才能与持久化 agents 配合使用。

onDelegationComplete

在委派完成后调用。可用它检查结果、提供反馈，或停止执行：

• context.bail()：立即停止父 Agent 的循环
• 返回 { feedback: '...' }：添加反馈，保存到父 Agent 的 memory，并在下一轮可用
• 返回 { resultText: '...' }：在当前运行中替换父模型看到的本次委派的工具结果文本

当子 Agent 自身的结果会给父 Agent 造成误导时，请设置 resultText。例如，子 Agent 在工具调用步骤停止时返回的空文本，在父模型看来像是一次成功但为空的委派。你也可以将失败委派的错误文本替换为更有用的消息。这并不会恢复委派，父 Agent 仍会收到失败的工具结果。与下一轮才生效的 feedback 不同，resultText 会立即影响父 Agent 的推理。

若要在失败时停止父 Agent 并为后续轮次保存反馈，请调用 bail() 并返回 feedback：

src/mastra/agents/parent-agent.ts

const stream = await parentAgent.stream('Research AI trends', {
  maxSteps: 10,
  delegation: {
    onDelegationComplete: async context => {
      console.log(`Completed: ${context.primitiveId}`)

      // Bail on errors
      if (context.error) {
        context.bail()
        return {
          feedback: `Delegation to ${context.primitiveId} failed: ${context.error}. Try a different approach.`,
        }
      }
    },
  },
})

context 对象包含：

属性	说明
primitiveId	已运行的子 Agent ID
result	子 Agent 的响应，在可用时包含 text、usage、finishReason、subAgentToolResults、subAgentPendingToolCalls、subAgentThreadId 和 subAgentResourceId
success	委派是否成功
error	委派失败时的原始错误
messages	收集到的子 Agent 对话记录，包括可用时的部分消息
bail()	用于停止父 Agent 循环的函数

失败的委派

当子 Agent 运行失败时，父模型会收到带有通用错误消息的失败工具结果。这不会自动停止父 Agent，它可以解释失败或选择其他方法。你也可以调用 context.bail() 停止其循环。

该 hook 会收到 success: false、原始 error，以及已收集的部分结果和消息。返回 resultText 会替换父模型看到的错误文本，但不会改变委派的失败状态，也不会丢弃原始原因。空字符串也是有效的替换值。

若要让父 Agent 使用经过整理的失败消息继续运行，而不是停止其循环，请返回 resultText 而不调用 bail()：

src/mastra/agents/parent-agent.ts

const stream = await parentAgent.stream('Research AI trends', {
  maxSteps: 10,
  delegation: {
    onDelegationComplete: ({ success }) => {
      if (!success) {
        return {
          resultText: 'The delegated task failed. Try another approach.',
        }
      }
    },
  },
})

请优先使用本示例中的整理后消息。返回 { resultText: error.message } 会明确将底层消息转发给父模型的提供商，模型可能将其重复给最终用户。提供商错误可能包含敏感信息。通用的 resultText 并不是针对日志、traces 或流式错误 payload 的脱敏策略。

对于作为后台任务运行的委派，配置的重试可能会在每次尝试时调用一次 onDelegationComplete。如果你的 hook 有副作用，请考虑重复调用的情况。

Hook 错误

如果委派 hook 抛出异常，委派默认继续：onDelegationStart 使用原始 prompt 继续，messageFilter 回退到未过滤的上下文，onDelegationComplete 保留子 Agent 的原始结果。设置 hookErrorStrategy: 'throw' 可改为让委派失败。

src/mastra/agents/parent-agent.ts

const stream = await parentAgent.stream('Research AI trends', {
  maxSteps: 10,
  delegation: {
    hookErrorStrategy: 'throw',
    onDelegationComplete: async ({ result }) => {
      // If this throws, the parent sees a failed delegation instead of a
      // successful one whose result was silently dropped.
      await reportToAnalytics(result)
    },
  },
})

无论哪种方式，每次 hook 失败都会记录在运行的请求上下文中的 __mastra_delegationHookErrors 下，形式为 { hook, primitiveId, toolCallId, runId, name, message } 条目列表，因此你可以通过编程方式检测 hook 失败：

src/mastra/agents/parent-agent.ts

const requestContext = new RequestContext()

await parentAgent.generate('Research AI trends', { maxSteps: 10, requestContext })

const hookErrors = requestContext.get('__mastra_delegationHookErrors') ?? []

当 hookErrorStrategy 为 'throw' 时，抛出异常的 onDelegationStart 会阻止子 Agent 运行；抛出异常的 messageFilter 或 onDelegationComplete 会作为失败的工具调用呈现给父 Agent。如果 onDelegationComplete 在处理失败委派时抛出异常，最终呈现的错误仍是原始委派错误。该 hook 不会因自身失败而再次被调用。

消息过滤

默认情况下，子 Agents 会收到父 Agent 的完整对话上下文。使用 messageFilter 控制共享哪些消息，例如移除敏感数据或限制上下文大小。

src/mastra/agents/parent-agent.ts

const stream = await parentAgent.stream('Research AI trends', {
  maxSteps: 10,
  delegation: {
    messageFilter: ({ messages, primitiveId, prompt }) => {
      // Remove messages containing sensitive data
      return messages
        .filter(msg => {
          const content =
            typeof msg.content === 'string' ? msg.content : JSON.stringify(msg.content)
          return !content.includes('confidential')
        })
        .slice(-10) // Only pass the last 10 messages
    },
  },
})

回调会收到 messages（完整对话历史）、primitiveId（子 Agent ID）和 prompt（委派 prompt）。返回过滤后的消息数组。

子 Agent 结果上下文

子 Agent 完成后，父 Agent 的模型会在后续迭代中收到该子 Agent 的文本响应。嵌套工具调用和子 Agent 元数据（例如 thread 和 resource IDs）不会添加到父 Agent 的模型上下文中。

应用代码和 UI 集成仍可在工具结果 payload 中检查 subAgentToolResults 以及原始委派结果的其他内容。

这样既能保留调试和展示数据，又不会把嵌套工具参数或输出送回父 Agent 的下一次模型调用。

子 Agents 无法运行客户端工具（没有 execute 函数的工具）。如果子 Agent 调用客户端工具，其运行会在该步骤停止，finishReason 为 'tool-calls'。委派结果会在 subAgentPendingToolCalls 中列出未解决的调用，父 Agent 的模型会被告知哪些工具从未运行，同时 Mastra 会记录一条警告。请将客户端工具交给父 Agent，或在子 Agent 的工具需要用户输入时使用 suspend()。

设置 includeSubAgentToolResultsInModelContext，可在父 Agent 的模型上下文中包含完整的子 Agent 结果，包括嵌套工具结果和子 Agent 元数据。

src/mastra/agents/parent-agent.ts

await parentAgent.generate('Research AI trends', {
  delegation: {
    includeSubAgentToolResultsInModelContext: true,
  },
})

复用先前的子 Agent 结果

默认情况下，子 Agent 结果会以工具结果的形式到达父 Agent，而这些结果在转发给后续子 Agents 时会从上下文中移除。父 Agent 必须在下一个委派 prompt 中重述先前的结果，这会消耗 tokens 并丢失细节。

设置 enableResultReferences，让后续委派逐字复用先前的结果：

src/mastra/agents/parent-agent.ts

await parentAgent.generate('Find and fix the token refresh bug', {
  delegation: {
    enableResultReferences: true,
  },
})

启用后：

• 每个成功且非空的子 Agent 结果都会获得一个引用 ID，例如 explorer-1。父 Agent 的模型会在子 Agent 文本之后看到一行 [ref: explorer-1]。
• 委派工具会获得 contextFromRefs 输入。父 Agent 可以传入先前的 ID，形式可以是字符串（["explorer-1"]），也可以是带可选标签和备注的对象（[{ ref: "explorer-1", as: "investigation", note: "bug location" }]）。
• 引用的文本会插入到委派 prompt 之前，每个结果位于各自带标签的块中，内容与先前子 Agent 产生的完全一致。onDelegationStart 和 messageFilter 会收到扩展后的 prompt。
• 如果 onDelegationComplete 返回 resultText，后续委派收到的将是替换后的文本。

引用在单次父 Agent 运行期间保存在内存中，不会被持久化。被拒绝、失败、为空以及后台任务委派不会获得引用 ID。未知 ID 会被跳过并记录警告，委派继续进行。

被引用的文本是另一个 Agent 的输出。每个块使用全新的、不可预测的标签，并告知接收方子 Agent 将内容视为数据；但如果子 Agents 会处理不受信任的输入，请在 onDelegationStart 中或通过 processors 添加你自己的检查。

迭代监控

onIterationComplete 在父 Agent 循环的每次迭代后调用。可用它监控执行或指导下一次迭代，也可以提前停止执行。

src/mastra/agents/parent-agent.ts

const stream = await parentAgent.stream('Research AI trends', {
  maxSteps: 10,
  onIterationComplete: async context => {
    console.log(`Iteration ${context.iteration}/${context.maxIterations}`)
    console.log(`Finish reason: ${context.finishReason}`)

    // Inject feedback to guide the agent
    if (!context.text.includes('recommendations')) {
      return {
        continue: true,
        feedback: 'Please include specific recommendations in your analysis.',
      }
    }

    // Stop early when the response is sufficient
    if (context.text.length > 1000 && context.finishReason === 'stop') {
      return { continue: false }
    }

    return { continue: true }
  },
})

返回 { continue: true } 继续迭代，或返回 { continue: false } 停止。可包含可选的 feedback，向对话注入指导。当 feedback 与 continue: false 同时使用时，模型可能会获得最后一次文本回复机会，以整合该反馈。只有当前迭代仍处于活跃状态（例如工具调用之后）时，才会给予这次额外轮次。

记忆隔离

Mastra 在委派期间隔离子 Agent 的记忆。子 Agents 会收到完整对话上下文以便更好地决策，但只有其特定的委派 prompt 和响应会保存到其记忆中。

工作方式：

1. 转发完整上下文：父 Agent 委派时，子 Agent 会收到父 Agent 对话中的所有消息
2. 限定范围内的记忆保存：只有委派 prompt 和子 Agent 的响应会保存到子 Agent 的记忆中
3. 每次调用使用新线程：每次委派都使用唯一的线程 ID，确保清晰隔离

因此，子 Agents 既拥有所需上下文，又不会让父 Agent 的完整对话弄乱其记忆。详情参见多 Agent 系统中的记忆。

工具审批传递

工具审批会沿委派链传递。当子 Agent 使用带有 requireApproval: true 的工具或调用 suspend() 时，审批请求会出现在父 Agent 的流中。

const sensitiveDataTool = createTool({
  id: 'get-user-data',
  requireApproval: true,
  execute: async input => {
    return await database.getUserData(input.userId)
  },
})

const dataAgent = new Agent({
  id: 'data-agent',
  tools: { sensitiveDataTool },
})

const parentAgent = new Agent({
  id: 'parent-agent',
  agents: { dataAgent },
  memory: new Memory(),
})

const stream = await parentAgent.stream('Get data for user 123')

for await (const chunk of stream.fullStream) {
  if (chunk.type === 'tool-call-approval') {
    console.log('Tool requires approval:', chunk.payload.toolName)
  }
}

取消

当你向父 Agent 的 stream() 或 generate() 调用传入 abortSignal 时，Mastra 会将同一个信号转发给被委派的子 Agents。调用 AbortController.abort() 会在进行中的子 Agent 运行到达下一步时取消它们，而不是让它们运行到完成。

const controller = new AbortController()

const stream = await parentAgent.stream('Research AI trends', {
  abortSignal: controller.signal,
})

// Cancel the parent agent and any in-flight subagents
controller.abort()

任务完成度评分

Agents 并不总能在第一次尝试时就产出完整、正确的输出。任务完成度 scorers 可以在每次迭代后验证任务是否完成。如果验证失败，父 Agent 会继续迭代。失败 scorers 的反馈会包含在对话上下文中，让子 Agents 了解缺失了什么。

import { createScorer } from '@mastra/core/evals'

const taskCompleteScorer = createScorer({
  id: 'task-complete',
  name: 'Task Completeness',
}).generateScore(async context => {
  const text = (context.run.output || '').toString()
  const hasAnalysis = text.includes('analysis')
  const hasRecommendations = text.includes('recommendation')
  return hasAnalysis && hasRecommendations ? 1 : 0
})

const stream = await parentAgent.stream('Research AI in education', {
  maxSteps: 10,
  isTaskComplete: {
    scorers: [taskCompleteScorer],
    strategy: 'all',
    onComplete: async result => {
      console.log('Task complete:', result.complete)
    },
  },
})

Rubric scorer

内置 rubric scorer 让你以清单形式定义何为“正确”，并让 Agent 自我评估并迭代，直到满足每项标准或达到 maxSteps。

它作为 LLM-as-judge scorer 工作。每次迭代后，一个独立的评分模型会根据 rubric 审查 Agent 的输出。当所有必需标准都通过时循环结束。未通过的标准会将其反馈加入对话，让 Agent 重试。

这最适合具有清晰、可验证成功标准的任务。用法如下：

src/mastra/agents/rubric-scorer.ts

import { Agent } from '@mastra/core/agent'
import { createRubricScorer } from '@mastra/evals/scorers/prebuilt'

const parentAgent = new Agent({
  id: 'parent-agent',
  instructions: 'You coordinate research and writing using specialized agents.',
  model: 'openai/gpt-5.6-sol',
  agents: { researchAgent, writingAgent },
})

const rubricScorer = createRubricScorer({
  model: 'openai/gpt-5-mini',
  criteria: [
    { description: 'The response includes an analysis section' },
    { description: 'The response includes concrete recommendations' },
  ],
})

const stream = await parentAgent.stream('Research AI in education', {
  maxSteps: 10,
  isTaskComplete: {
    scorers: [rubricScorer],
    strategy: 'all',
  },
})

完整 API 细节参见 rubric scorer 参考。

编写有效的 instructions

清晰的 instructions 对有效委派至关重要。

父 Agent 的 instructions 应说明可用资源以及何时使用每种资源，还应定义协调行为和成功标准。

每个子 Agent 都应有清晰的 description，说明其用途和返回格式，包括父 Agent 应在何时使用它。

父 Agent 利用这些描述做出委派决策。

const parentAgent = new Agent({
  id: 'parent-agent',
  instructions: `You coordinate research and writing tasks.

Available resources:
- researchAgent: Gathers factual data and sources (returns bullet points)
- writingAgent: Transforms research into narrative content (returns full paragraphs)

Delegation strategy:
1. For research requests: Delegate to researchAgent first
2. For writing requests: Delegate to writingAgent
3. For complex requests: Delegate to researchAgent first, then writingAgent

Success criteria:
- All user questions are fully answered
- Response is well-formatted and complete`,
  agents: { researchAgent, writingAgent },
})

在后台运行子 Agents

子 Agent 调用会作为工具调用分派，因此可以作为后台任务运行。当一个或多个委派耗时较长，而你又不希望它们阻塞父 Agent 的响应时，这很有用。

在 Mastra 实例上启用 backgroundTasks 管理器，然后在父 Agent 上让子 Agents 选择启用：

src/mastra/agents/parent-agent.ts

const parentAgent = new Agent({
  id: 'parent-agent',
  instructions: 'Coordinate research and writing using the available agents.',
  model: 'openai/gpt-5.6-sol',
  agents: { researchAgent, writingAgent },
  backgroundTasks: {
    tools: {
      researchAgent: { enabled: true, timeoutMs: 900_000 },
      writingAgent: { enabled: true, timeoutMs: 900_000 },
    },
  },
})

const stream = await parentAgent.streamUntilIdle('Research AI in education and write an article', {
  memory: { thread: 't1', resource: 'u1' },
})

请使用 streamUntilIdle() 而不是 stream()，这样流会保持打开，直到子 Agents 完成且父 Agent 有机会对其结果做出响应。

如果某个子 Agent 未列在父 Agent 的 backgroundTasks.tools 下，但拥有自己的后台可用工具，父 Agent 仍会将该子 Agent 作为后台任务分派，并继承其配置。详情参见从子 Agent 继承。

子 Agent 版本控制

使用 editor 时，你可以控制父 Agent 在运行时使用每个子 Agent 的哪个已存储版本。可在 Mastra 实例上或按调用设置版本覆盖：

const result = await parentAgent.generate('Research and write about AI safety', {
  versions: {
    agents: {
      'research-agent': { status: 'published' },
      'writing-agent': { versionId: 'draft-456' },
    },
  },
})

版本覆盖会自动通过委派传递。解析顺序和服务器 API 用法细节参见子 Agent 版本控制。

相关内容

• 后台任务
• 子 Agent 版本控制
• 指南：研究协调器
• Agent.stream() 参考
• Agent.streamUntilIdle() 参考
• Agent.generate() 参考
• Agent 审批
• 多 Agent 系统中的记忆
• 概念：多 Agent 系统
• 📹 Mastra 监督者 agents 研讨会