agent.generate() 全链路
从用户调用 agent.generate() 到拿到 FullOutput 之间,Mastra 内部构造了 3 层嵌套 Workflow(prepare-stream → agentic-loop → agentic-execution × N),跑 Processor 管线,多次调用 LLM,直到模型不再返回 tool_call。下面逐步拆解。
逐步骤拆解
13 steps agent.generate(messages) User 用户把消息(字符串或 Message 数组)传给 Agent.generate(),可以传 options(maxSteps、tools、temperature、output 等)。
#execute(options) Agent Agent 私有方法做参数归一化、合并 agent 级默认配置、把 message 转成 AI SDK 兼容格式,准备 stream 选项。
createPrepareStreamWorkflow() PrepareStream Workflow 位于 agent/workflows/prepare-stream/。构造最外层 workflow,包含 prepare-memory-step(从 storage 加载历史消息,226 行)、prepare-tools-step(合并 agent tools + MCP tools + 内置 tools,83 行)、input processor 执行、map-results-step(把 LLM 输出映射成 MastraModelOutput,415 行)。这个 workflow 的输出是可以直接送进 LLM 的消息列表和工具集合。
processInput / processLLMRequest ProcessorRunner 按注册顺序跑每个 Processor 的 input 侧 hooks。Memory processor 注入历史消息;Skills processor 注入技能 prompt;Browser-context processor 注入当前页面上下文;Caching processor 检查命中缓存。每个 Processor 都可以增删改消息。
stream(loopOptions) MastraLLMVNext Agent 持有 MastraLLMVNext(LanguageModel 的包装),它不直接调 AI SDK,而是把 stream 请求转给 loop(),把 AI SDK 的 doStream 和 Mastra 自己的 agentic loop 接起来。
loop(options) MastraLLMVNext → loop() MastraLLMVNext(位于 llm/model/model.loop.ts,是 Agent → loop() 之间的胶水)接收 stream 请求,把它转给 loop()(loop/loop.ts,177 行)。loop() 构造 agentic-loop workflow 并通过 workflowLoopStream()(loop/workflows/stream.ts,378 行)驱动 ReadableStream。
createAgenticLoopWorkflow() AgenticLoop Workflow iteration loop 外层循环 workflow。负责维护迭代计数、检查 TripWire(maxSteps/token/cost)、判断终止条件。每一轮迭代都 spawn 一个 agentic-execution workflow。
createAgenticExecutionWorkflow() AgenticExecution Workflow 单次迭代:跑 processLLMRequest(per-step 再次让 processor 介入,比如 structured output 这里注入 response_format)→ 调 AI SDK LanguageModel.doStream() → 消费输出流 → 如果有 tool_call 则跑工具执行 step → 如果是纯文本则标记迭代结束。
LanguageModel.doStream() LLM Step 最终真正打到 AI SDK provider(openai/anthropic/...)的 HTTP 调用。文本 token、tool_call 分片在这里以流式事件形式输出,Mastra 把它们包成 MastraModelOutput chunks。
forEach tool call: validate + execute Tool Step tool branch 如果模型返回 tool_calls,AgenticExecution 按顺序(或 parallel)执行每个工具。工具支持 requireApproval(中断循环等人审)、suspend(睡眠等待外部事件)、background(由 tool-loop-agent 在后台跑)。工具结果追加到消息列表,下一轮迭代喂回 LLM。
no tool calls → stop iteration AgenticExecution Workflow terminate 模型这轮返回的只有文本,没有 tool_call → agentic-loop 判断终止条件成立 → 结束循环。
processLLMResponse / processOutput ProcessorRunner 循环结束后跑 output 侧 processors:把 memory 写入 storage、做 structured output 校验、触发 observability span。
MastraModelOutput → FullOutput Stream / Agent 流式 chunks 聚合为 FullOutput(text + toolResults + usage + steps)返回给用户。stream() 路径则把 ReadableStream 直接返回。
关键点
takeawaysprepare-stream → agentic-loop → agentic-execution × N。不是 while 循环,而是把迭代本身建模为可持久化、可中断、可恢复的工作流。
外层 prepare-stream 跑一次 processInput;每轮迭代内 agentic-execution 再跑一次 processLLMRequest。这就是为什么 memory/skills/caching 能在循环中途持续生效。
位于 llm/model/model.loop.ts,不从 llm/index.ts 导出,被 agent/agent.ts 直接 import。它把 AI SDK LanguageModel 的 doStream 路由进 Mastra 自己的 loop()——这是 agent.generate() 最终跑三层 workflow 的关键 indirection。
requireApproval、suspend、background execution 都不是 Tool 类里的标志位判断,而是 agentic-execution workflow 的分支节点——durable execution 天然支持中途停下等外部事件。