智能体编排框架横评（5）：轻量与类型化框架

智能体编排框架横评 第 5 站 · 上一站：多智能体框架 · 下一站：低代码平台 · 全程：总览

这一站讲什么

智能体编排框架横评（1）：LangChain 1.x 走大而全的路线：一个高层入口加中间件，底下是 智能体编排框架横评（2）：LangGraph 的图运行时，外面一大圈集成包。这一站的四个框架反过来，尽量少发明概念，用语言已有的东西表达 Agent：

• 类型：Pydantic AI 用 Pydantic 模型和泛型，Mastra 用 zod schema，LlamaIndex Workflows 用类型注解决定事件流向；
• 工具：带 docstring 的普通函数，加一个装饰器；
• 控制流：普通的 if 和 for，不需要先学图引擎。

框架	语言	特点
Pydantic AI	Python	类型安全，依赖注入，结构化输出
LlamaIndex Workflows	Python	事件驱动，step 之间传事件
smolagents	Python	模型写 Python 代码当动作
Mastra	TypeScript	Agent、Workflow、记忆、评测放在一个实例里

这一组共同缺的是外围能力：持久化、审批、权限。有的框架后来补上了，有的留给你自己写。读的时候可以一直问：哪些是框架给的，哪些仍要自己写。

报销场景沿用 主线：一个报销请求走完整个 Agent 系统。代码按各家 2026-11 的官方文档写，是示意代码，没有运行过。

Pydantic AI

Pydantic AI 是什么

Pydantic 团队做的 Python Agent 框架，主打类型安全：依赖、输出、工具参数都有类型，IDE 和类型检查器能在运行前发现错误。官方的看法是，一个带类型的 Agent 加普通 Python 控制流，就能做 LangChain 和 LangGraph 那一套。

安装

pip install pydantic-ai
export OPENAI_API_KEY="sk-..."

第一个 Agent

hello_pydantic_ai.py 文件代码：

from pydantic_ai import Agent

agent = Agent('openai:gpt-6.1-sol', instructions='你是公司的报销助手，回答要简短。')
result = agent.run_sync('报销一般需要哪些材料？')
print(result.output)

输出结果类似：

一般需要发票、行程或用途说明，以及审批人签字。

模型名要带厂商前缀，比如 openai:、anthropic:。异步代码里用 await agent.run(...)。

基本概念

工具。 @agent.tool 装饰的函数第一个参数是 ctx，可以读依赖；不需要依赖的用 @agent.tool_plain。docstring 就是工具说明。

@agent.tool_plain
def get_invoice(invoice_id: str) -> dict:
    """按发票号查金额和报销标准。"""
    return {"id": invoice_id, "amount": 118.5, "standard": 80}

依赖（deps）。 有些数据不该让模型填，比如当前登录的员工。用 deps_type 声明类型，运行时 run(..., deps=实例) 传进去，工具里用 ctx.deps 读：

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class Employee:
    emp_id: str

agent = Agent('openai:gpt-6.1-sol', deps_type=Employee)

@agent.tool
def my_invoices(ctx: RunContext[Employee]) -> list[str]:
    """列出当前员工的发票。"""
    return invoice_db.list(owner=ctx.deps.emp_id)

agent.run_sync('我有哪些发票？', deps=Employee('E-001'))

deps_type 只用于类型检查，真正的值在运行时传入。测试时换一个 deps 就能把外部系统隔离掉，还可以用 Agent('test') 这个离线模型。

结构化输出。 output_type 填一个 Pydantic 模型，Agent 返回的就是这个模型的实例。校验失败时，错误会交回模型让它重试。

from pydantic import BaseModel

class Verdict(BaseModel):
    reimbursable: bool
    over_amount: float
    clause: str

agent = Agent('openai:gpt-6.1-sol', output_type=Verdict)
r = agent.run_sync('打车 118.5 元，标准 80 元，能报吗？')
print(r.output.over_amount)   # 38.5

延迟工具（审批）。 工具设 requires_approval=True，模型调用它时这次运行不执行工具，而是以 DeferredToolRequests 结束；你带着决定再运行一次。

Capability。 2.0 起的核心原语，一个 Capability 打包一组工具、钩子、指令和模型设置。另有 pydantic-ai-harness 包提供文件系统、Shell、子代理、权限、检查点等现成能力。

报销例子：建单要跑两次

员工身份放 deps，来自登录态，模型看不到也改不了；判断结果是结构化的 Verdict；create_ticket 设 requires_approval=True。

第 1 步，定义 Agent 和工具：

from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext, DeferredToolRequests

@dataclass
class Employee:                 # 来自登录态，不从对话里取
    emp_id: str
    level: str                  # 示意：P6
    city: str                   # 示意：上海

class Verdict(BaseModel):
    reimbursable: bool
    over_amount: float          # 示意：118.5 - 80 = 38.5
    clause: str                 # 示意：差旅费管理办法 2026 版第 4.2 条
    ticket_id: str | None = None

agent = Agent('openai:gpt-6.1-sol', deps_type=Employee,
              output_type=[Verdict, DeferredToolRequests],
              instructions='先查制度和发票再下结论；建单要等员工确认。')

@agent.tool
def get_invoice(ctx: RunContext[Employee], invoice_id: str) -> dict:
    """查当前员工名下的一张发票。"""
    return invoice_db.get(invoice_id, owner=ctx.deps.emp_id)

@agent.tool(requires_approval=True)
def create_ticket(ctx: RunContext[Employee], invoice_id: str, reason: str) -> str:
    """提交报销单（有副作用）。"""
    return ticket_api.create(ctx.deps.emp_id, invoice_id, reason)

search_policy 同理，按 ctx.deps.level 和 ctx.deps.city 过滤，这里省略。注意 output_type 写成了列表：运行可能以 Verdict 结束，也可能以待审批的 DeferredToolRequests 结束。

第 2 步，运行，停在建单前，拿到员工的决定后再运行一次：

from pydantic_ai import DeferredToolResults, ToolDenied

me = Employee('E-001', 'P6', '上海')                  # 示意
r = agent.run_sync('上周打车超标还能报吗，能就帮我提单', deps=me)
if isinstance(r.output, DeferredToolRequests):       # 停在 create_ticket 前
    results = DeferredToolResults()
    for call in r.output.approvals:
        ok = ask_employee(call.args)                 # 前端弹窗，员工点确认
        results.approvals[call.tool_call_id] = True if ok else ToolDenied('员工取消')
    r = agent.run_sync(message_history=r.all_messages(),
                       deferred_tool_results=results, deps=me)
print(r.output)  # 示意：Verdict(..., over_amount=38.5, ticket_id='R-55821')

员工取消时，ToolDenied('员工取消') 会作为这次工具调用的结果交给模型。

两次运行之间如果隔着一次 HTTP 往返，要把消息历史存起来：把 result.conversation 用 ConversationTypeAdapter 序列化，之后恢复，再用 conversation.deferred_tool_requests 构造结果。

下面这张图画的就是这两次运行和中间存下的消息历史：

进阶

• 持久执行。 框架默认不保存运行状态。需要崩溃恢复时，接 Temporal、DBOS、Prefect、Restate 这类持久执行引擎，写法是 capabilities=[DBOSDurability()] 这样的 Capability。文档提醒 "Durability is not storage"：只想存聊天记录，不需要上持久执行引擎。见 可靠执行：状态、幂等与恢复。
• 多 Agent。 在工具里调用另一个 Agent（传 usage=ctx.usage 合并用量），或者由代码决定交给哪个 Agent；harness 的 SubAgents 提供 delegate_task。
• pydantic-graph。 一个独立的类型化状态机库，节点 run() 的返回类型注解就是边。文档自己说 "Don't use a nail gun unless you need a nail gun"，普通控制流够用时不必用它。
• 可观测与 MCP。 原生支持 OpenTelemetry，logfire.instrument_pydantic_ai() 一行接入（见 可观测性：追踪一条任务的代价）；MCP(url=...) Capability 接 MCP 工具，Agent 本身也能当 MCP 服务器。

常见问题

审批通过就等于有权限吗？ 不是。文档原文 "Approval is not an authorization boundary against an untrusted client"。前端送回的批准要验身份，工具函数里仍要做权限检查。

同一轮里模型既给了结果又调了工具，工具被执行了。 2.0 的默认 end_strategy 是 'graceful'，和输出同一轮返回的工具调用也会执行。有副作用的工具要注意。

旧代码里的 DeferredToolCalls、mcp_servers= 找不到了。 2.0 改了名，分别是 DeferredToolRequests、toolsets=，对照官方迁移指南改。

LocalWorkspace('.') 能不能当沙箱用？ 不能，文档写明它 "is not a sandbox"。

LlamaIndex Workflows

LlamaIndex Workflows 是什么

LlamaIndex 原本是 RAG 框架，后来把 Agent 编排的底座单独做成了 Workflows，一个事件驱动的流程库，可以不装 LlamaIndex 的其他部分单独用。

它和图框架的区别是不先画图。每个 step 收一个事件、发一个事件，框架按类型注解把事件交给能处理它的 step。分支是普通 if，循环是发回一个前面处理过的事件，并发是一次发出多个事件。

安装

pip install llama-index-workflows

要用后面的预制 Agent，再装 pip install llama-index，它带上了 LlamaIndex 核心和 OpenAI 接入。

第一个 Workflow

hello_workflow.py 文件代码：

import asyncio
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent

class HelloFlow(Workflow):
    @step
    async def greet(self, ev: StartEvent) -> StopEvent:
        return StopEvent(result=f"你好，{ev.name}")

async def main():
    result = await HelloFlow().run(name="小王")
    print(result)

asyncio.run(main())

输出结果为：

你好，小王

run() 的关键字参数会放进 StartEvent，返回 StopEvent 时 Workflow 结束，result 就是返回值。

基本概念

Event。 事件是 Pydantic 模型。自定义事件继承 Event：

from workflows.events import Event

class InvoiceChecked(Event):
    amount: float
    standard: float

Step。 参数类型决定它接什么事件，返回类型决定它能发什么事件。下面两个 step 用 InvoiceChecked 连起来：

class CheckFlow(Workflow):
    @step
    async def check(self, ev: StartEvent) -> InvoiceChecked:
        return InvoiceChecked(amount=118.5, standard=80)

    @step
    async def judge(self, ev: InvoiceChecked) -> StopEvent:
        over = ev.amount - ev.standard
        return StopEvent(result="超标" if over > 0 else "未超标")

@step(num_workers=N) 控制一个 step 最多同时处理几个事件。

Context。 一次运行的共享状态。ctx.store.get/set 读写，async with ctx.store.edit_state() 做原子修改；ctx.to_dict() / Context.from_dict() 做快照。

Handler。 run() 返回一个 handler：handler.stream_events() 读事件流，handler.ctx.send_event() 往运行中的 Workflow 送事件，await handler 拿最终结果。

预制 Agent。 FunctionAgent、ReActAgent、CodeActAgent 是现成的 Agent；AgentWorkflow 把几个 Agent 组在一起，用 can_handoff_to 指定谁能交给谁。

报销例子：在工具里等员工回复

框架没有审批开关，确认用 ctx.wait_for_event 自己写：工具发出一个 InputRequiredEvent，然后等一个 HumanResponseEvent。身份放在 Workflow 的 state 里，工具从 ctx 读。

from llama_index.core.agent.workflow import AgentWorkflow
from llama_index.core.workflow import Context, InputRequiredEvent, HumanResponseEvent

async def create_ticket(ctx: Context, invoice_id: str, reason: str) -> str:
    """提交报销单（有副作用），提交前要员工确认。"""
    emp = (await ctx.store.get("state"))["emp_id"]       # 身份来自 state，不来自模型
    ans = await ctx.wait_for_event(
        HumanResponseEvent,
        waiter_id=f"confirm-{invoice_id}",
        waiter_event=InputRequiredEvent(prefix=f"确认为 {invoice_id} 提单？"),
    )
    if ans.response.strip() != "确认":
        return "员工未确认，未建单"
    return ticket_api.create(emp, invoice_id, reason)    # 示意：R-55821

agent = AgentWorkflow.from_tools_or_functions(
    [search_policy, get_invoice, create_ticket], llm=llm,
    system_prompt="先查制度和发票再下结论，建单前等员工确认。",
    initial_state={"emp_id": "E-001", "level": "P6", "city": "上海"})  # 示意
handler = agent.run(user_msg="上周打车超标还能报吗，能就帮我提单", ctx=Context(agent))
async for ev in handler.stream_events():
    if isinstance(ev, InputRequiredEvent):
        handler.ctx.send_event(HumanResponseEvent(response=ask_employee(ev.prefix)))
print(await handler)

外面的循环读事件流，看到 InputRequiredEvent 就去问员工，把回复包成 HumanResponseEvent 送回去，工具里的 wait_for_event 收到后继续执行。

如果员工要很久才回复，进程不该一直等：把 ctx 快照存起来，调用 cancel_run()，回复到了再从快照恢复。

常见问题

进程重启后运行状态没了。 文档原文 "Workflows are ephemeral by default"，没有内置检查点。要自己在 step 结束时快照，或用 llama-agents-dbos 的 DBOSRuntime。

恢复后 wait_for_event 前面的代码又执行了一遍。 恢复时会重放等待之前的代码，那段代码必须可以重复执行。文档推荐把发问和处理回答拆成两个 step。上面的例子等待前只读了 state，可以重放。

恢复后带副作用的 step 执行了两次。 执行到一半的 step 恢复时会重跑，要做幂等，见 可靠执行：状态、幂等与恢复。

旧教程里的 AgentRunner、OpenAIAgent、QueryPipeline 找不到了。 已在 llama-index-core 0.13 删除，改用 FunctionAgent、ReActAgent 和 AgentWorkflow。

整体流程不好看清。 没有一张边表，要读完所有 step 的签名才知道拓扑，也不能静态检查出走不到的分支。

smolagents

smolagents 是什么

Hugging Face 做的极简 Agent 库。它的主要特点是用代码当动作：CodeAgent 让模型写一段 Python，工具以函数的形式提供给这段代码，一次动作里就能循环、组合多个工具、处理中间结果。另一个类 ToolCallingAgent 走常见的 JSON 工具调用。

文档的选择建议：需要推理、串联、动态组合时用 CodeAgent；工具是简单的原子操作、要高可靠和清楚的参数校验时用 ToolCallingAgent。

安装

pip install smolagents
export HF_TOKEN="hf_..."

默认的 InferenceClientModel 调用 Hugging Face 的推理服务。用 OpenAI、Anthropic 等模型时换成 LiteLLMModel 或 OpenAIServerModel。

第一个 Agent

hello_smolagents.py 文件代码：

from smolagents import CodeAgent, InferenceClientModel

agent = CodeAgent(tools=[], model=InferenceClientModel())
agent.run("118.5 减去 80 是多少？")

运行时控制台会逐步打印：模型写出的代码（比如 result = 118.5 - 80 和 final_answer(result)）、代码的执行结果，最后是答案：

Final answer: 38.5

基本概念

工具。 用 @tool 装饰函数，必须有类型注解和带 Args: 段的 docstring：

from smolagents import tool

@tool
def get_invoice(invoice_id: str) -> dict:
    """查发票金额和报销标准。

    Args:
        invoice_id: 发票号，例如 INV-102
    """
    return {"id": invoice_id, "amount": 118.5, "standard": 80}

复杂的工具可以继承 Tool 类，实现 forward 方法。

两种 Agent。

Agent	动作是什么	适合
CodeAgent	一段 Python 代码	需要组合多个工具、处理中间结果
ToolCallingAgent	一次 JSON 工具调用	工具简单，要可靠的参数校验

步骤和记忆。 max_steps 限制最多几步；agent.memory.steps 记录每一步；step_callbacks 每步结束后回调；run(task, reset=False) 在上一次的记忆上继续对话；agent.interrupt() 让 Agent 在当前步结束后停下。

托管 Agent。 带 name 和 description 的 Agent 放进上级 Agent 的 managed_agents，上级就能把它当工具调用。

报销例子：在工具里确认

建单有副作用，按文档的建议用 ToolCallingAgent。框架没有审批机制，确认只能写在工具里；也没有依赖注入，这里用闭包把员工身份带进工具。

from smolagents import ToolCallingAgent, InferenceClientModel, tool

def build_agent(emp_id: str) -> ToolCallingAgent:     # 每个请求按登录态建一次
    @tool
    def create_ticket(invoice_id: str, reason: str) -> str:
        """提交报销单，有副作用。

        Args:
            invoice_id: 发票号，例如 INV-102
            reason: 超标事由
        """
        if not confirm_with_employee(emp_id, invoice_id, reason):  # 自己实现的确认
            return "员工未确认，未建单"
        return ticket_api.create(emp_id, invoice_id, reason)      # 示意：R-55821

    return ToolCallingAgent(tools=[search_policy, get_invoice, create_ticket],
                            model=InferenceClientModel(), max_steps=6)

build_agent("E-001").run("上周打车超标还能报吗，能就帮我提单")

如果换成 CodeAgent，模型可能写出 inv = get_invoice("INV-102"); create_ticket(...) 这样一段代码，一步里查完就建单。确认写在 create_ticket 内部，所以不会被绕过；但代码执行环境本身要隔离，见下一节。

CodeAgent 在哪里执行代码

背景见 安全边界：让权限落到工具层。

文档原文："By default, the CodeAgent runs LLM-generated code in your environment." 默认的 LocalPythonExecutor 是一个 AST 解释器，不是真正的 Python：import 走白名单（additional_authorized_imports 放行），会拦截间接访问系统模块，循环操作数有上限。但文档也承认 "no local python sandbox can ever be completely secure"，比如放行 Pillow 后代码可以写满磁盘。

要更强的隔离，用 executor_type 选远程执行环境："e2b"、"docker"、"modal"、"blaxel"，用完用 with CodeAgent(...) as agent: 或 cleanup() 释放。隔离方式有两种：

• 只把代码片段放进沙箱：API key 不进沙箱，但不支持托管 Agent；
• 整个系统放进沙箱：支持多 Agent，但可能要把 API key 带进去。

放到报销场景：模型能在一步代码里查发票、算超标、建单，审批要拦的正是这一串。所以确认必须写在工具实现里，执行环境按文档的建议断网、限资源、最小权限。

常见问题

能不能从 Hub 直接加载别人的工具？ 可以，但要 trust_remote_code=True，相当于执行别人的代码，先看过代码再用。

stdio 型 MCP 服务器安全吗？ 文档警告它 "will always execute code on your machine"。

进程重启后对话没了。 只有进程内的 agent.memory，没有内置检查点，要自己保存。

最近没有新版本。 PyPI 上最新版 1.26 发布于 2026-05，GitHub 仓库仍有提交。用到新特性前先确认主分支和 PyPI 是否一致。

Mastra

Mastra 是什么

TypeScript 生态的 Agent 框架。它把 Agent、Workflow、记忆、评测、可观测注册在一个 Mastra 实例里，配一个本地调试界面 Studio。每一块仍然是普通的 TS 对象和 zod schema。

文档反复强调分工：步骤事先不知道的开放任务交给 Agent，步骤事先清楚的流程用 Workflow。

安装

新建项目最快的方式是用脚手架：

npm create mastra@latest

在已有项目里手动安装：

npm install @mastra/core zod

第一个 Agent

hello-mastra.ts 文件代码：

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

const agent = new Agent({
  id: 'assistant',
  name: 'Assistant',
  instructions: '你是公司的报销助手，回答要简短。',
  model: 'openai/gpt-6.1-sol',
})

const result = await agent.generate('报销一般需要哪些材料？')
console.log(result.text)

输出结果类似：

一般需要发票、行程或用途说明，以及审批人签字。

模型写成 厂商/模型名。要边生成边输出用 agent.stream(...)。

基本概念

工具。 用 createTool 定义，输入输出用 zod 描述：

import { createTool } from '@mastra/core/tools'
import { z } from 'zod'

const getInvoice = createTool({
  id: 'get-invoice',
  description: '查发票金额和报销标准',
  inputSchema: z.object({ invoiceId: z.string() }),
  execute: async ({ invoiceId }) => ({ id: invoiceId, amount: 118.5, standard: 80 }),
})

文档提醒：不用 createTool、直接写普通对象定义的工具会 "silently fail to execute"。

RequestContext。 每次请求的上下文，工具和动态指令都能读，值不会自动进提示词，适合放登录身份。

Workflow。 用 createStep 定义步骤，用 createWorkflow(...) 串起来：

const wf = createWorkflow({ id: 'check', inputSchema, outputSchema })
  .then(checkInvoice)
  .then(judge)
  .commit()

还有 .parallel()、.branch()、.dountil()、.foreach()、.sleep()。step 里调用 suspend() 暂停，外部用 run.resume() 恢复。

记忆和评测。 @mastra/memory 管对话线程和工作记忆；@mastra/evals 的 scorer 可以按比例抽样给线上输出打分。

报销例子：工具审批

import { Agent } from '@mastra/core/agent'
import { createTool } from '@mastra/core/tools'
import { RequestContext } from '@mastra/core/request-context'
import { z } from 'zod'

const createTicket = createTool({
  id: 'create-ticket',
  description: '提交报销单，有副作用',
  inputSchema: z.object({ invoiceId: z.string(), reason: z.string() }),
  requireApproval: true,
  execute: async ({ invoiceId, reason }, ctx) =>
    ticketApi.create(ctx?.requestContext?.get('empId'), invoiceId, reason), // 示意 R-55821
})

const expenseAgent = new Agent({
  id: 'expense-agent', name: 'Expense Agent', model: 'openai/gpt-6.1-sol',
  instructions: '先查制度和发票再下结论，建单前等员工确认。',
  tools: { searchPolicy, getInvoice, createTicket },
})

const rc = new RequestContext(); rc.set('empId', session.empId)  // 来自登录态
const out = await expenseAgent.generate('上周打车超标还能报吗，能就帮我提单', { requestContext: rc })
if (out.finishReason === 'suspended') {
  const ids = { runId: out.runId, toolCallId: out.suspendPayload.toolCallId }
  const res = (await askEmployee(out.suspendPayload.args))
    ? await expenseAgent.approveToolCallGenerate(ids)
    : await expenseAgent.declineToolCallGenerate(ids)
}

流程是：requireApproval: true 的工具被调用时，generate 以 finishReason === 'suspended' 返回，suspendPayload 里有工具参数；员工确认后调 approveToolCallGenerate，取消调 declineToolCallGenerate。

审批依赖快照，Mastra 实例必须配持久存储，否则会报 "snapshot not found"。进程重启后用 agent.listSuspendedRuns({ threadId, resourceId }) 找回待审批的运行。RequestContext 里只有可序列化的值会跨暂停保留，文档建议审批请求进来时重新验身份、重新写 empId。

常见问题

.branch() 的两个分支都执行了。 .branch() 会评估所有条件，每个为真的都执行，不是命中第一个就停。条件要写成互斥的，各分支 step 的输入输出 schema 也必须一致。

工具里 suspend() 之后的代码还在执行。 suspend() 不会抛异常，要写成 return await suspend(...)。

直接 import 的 Agent 没有记忆和追踪。 通过 mastra.getAgentById() 拿到的 Agent 才共享实例里注册的存储、记忆和 tracing。

多 Agent 用 .network() 有弃用警告。 改为在父 Agent 的 agents: {...} 里注册子 Agent。

同一件事有两种暂停方式。 工具审批和 Workflow step 的 suspend() 都能暂停，事先定好用哪一种。

Agno 与 DSPy

这两个只简单介绍定位。

Agno 的自我定位是 "framework and runtime for building agent platforms"：用 Agno SDK 写 Agent、团队和 Workflow，跑在 AgentOS 运行时上，由它负责认证、权限、恢复、人工审批和调度。工具审批写法是 @tool(requires_confirmation=True)，运行返回 is_paused，再 continue_run()。它比本站其他几个更靠近平台产品，可以和下一站 智能体编排框架横评（6）：低代码与可视化平台 对照着看。

DSPy 不是编排框架，口号是 "Program, don't prompt, your LLMs."。你用 Signature 声明输入输出，用 Module（如 dspy.ReAct）决定怎么执行，再用优化器按评测指标自动调提示词和示例。它解决的是提示词怎么调稳，可以嵌在任何框架的某一步里用。评测指标怎么设计见 评测与可观测性。

版本说明

本文按 2026-11 初的文档写，查阅时的版本：Pydantic AI 2.66，llama-index-workflows 2.26，smolagents 1.26，@mastra/core 1.82。对照旧资料时注意：

• Pydantic AI 2.0（2026-06）改了不少名字，加入 Capability 和 harness；1.x 仍在出补丁。
• LlamaIndex Workflows 1.0（2025-06）起是独立包，旧的 llama_index.core.workflow 只是重新导出。
• Mastra 1.0 在 2026-01 发布，之后小版本更新很频繁，durable agent 等功能仍是 Beta。

和手写循环的对照

跟读 learn-claude-code：17 个 harness 机制怎么挂到一个循环上 系列手写的 harness 有这几样：跟读 learn-claude-code（1）：循环内核 的循环和三条规矩（没有工具调用就停；每个工具调用按 id 配一个结果；被拒也要回填结果），权限检查和钩子，计划、子代理与上下文预算 的压缩与子代理，任务图、后台与定时：让工作撑过一轮对话 的检查点和后台任务。拿这四个框架对一下：

手写 harness 的部分	这一组框架的情况
循环与停止条件	四个都写好了
调用与结果配对	都由框架维护，被拒时回填的内容可定制（ToolDenied('...')、Mastra 的 decline）
工具 schema 与参数校验	都从类型生成：Pydantic 模型、函数注解、zod
权限与审批	Pydantic AI、Mastra 有审批；LlamaIndex 用事件自己拼；smolagents 没有。按规则自动放行或拒绝基本都要自己写，见 安全边界：让权限落到工具层
压缩与上下文预算	Mastra 有记忆处理器；其他几个主要靠手动处理消息历史
检查点与恢复	Pydantic AI、LlamaIndex 交给外部引擎或手动快照；Mastra 有存储快照；smolagents 没有
子代理	都有，形式是把 Agent 当工具，或注册子 Agent

这一组离手写 harness 很近。框架替你做的基本就是 跟读 learn-claude-code（1）：循环内核 那一站的内容，再加类型校验和追踪。读懂了手写循环，再看它们的源码不难。

仍要自己做的集中在循环外围：身份从哪来、谁有权批准、批准是否绑定到具体参数、建单的幂等键、被打断后从哪续跑。审批不等于授权，存储不等于持久执行，这些边界要你自己定，参见 可靠执行：状态、幂等与恢复 和 Coding / Harness。

自测

Pydantic AI 里 @agent.tool 和 @agent.tool_plain 有什么区别？

参考：@agent.tool 的函数第一个参数是 ctx: RunContext[...]，可以通过 ctx.deps 读依赖；@agent.tool_plain 不带 ctx，适合不需要依赖的工具。

Pydantic AI 的 deps_type 在运行时起什么作用？员工身份为什么放在 deps 而不是让模型填？

参考：deps_type 只用于类型检查，真正的依赖在 run(..., deps=实例) 传入，工具通过 ctx.deps 读。身份来自登录态，放在 deps 里模型看不到也改不了；如果作为工具参数让模型填，提示词注入或模型出错都可能换成别人的工号。

LlamaIndex Workflows 里，框架怎么知道一个事件该交给哪个 step？

参考：看 step 的参数类型注解。哪个 step 的参数类型和事件类型一致，事件就交给它；step 的返回类型决定它能发出什么事件。

报销例子里员工点了"取消"，Pydantic AI 和 Mastra 分别怎么把这个决定交回模型？

参考：Pydantic AI 在 DeferredToolResults.approvals[tool_call_id] 里填 ToolDenied('员工取消')，带着原消息历史再跑一次，这段话作为工具结果交给模型。Mastra 调 declineToolCallGenerate({ runId, toolCallId })。两者都保证这次工具调用有对应的结果，符合手写循环里被拒也要回填的规矩。

smolagents 的 CodeAgent 默认在哪里执行代码？只把代码片段放进 E2B 沙箱有什么代价？

参考：默认在本机，由 LocalPythonExecutor 这个 AST 解释器执行，它有导入白名单和操作数上限，但文档明说本地沙箱不可能完全安全。只放代码片段进远程沙箱时，API key 不必进沙箱，但不支持托管 Agent；要多 Agent 就得把整个系统放进沙箱，可能要带上 API key。

Mastra 的 .branch() 两个条件同时为真会怎样？

参考：两个分支都会执行。.branch() 会评估所有条件，每个为真的都运行对应的 step，所以条件要写成互斥的。

参考资料

以下链接均查阅于 2026-11。

• Pydantic AI 文档首页
• Pydantic AI：Deferred Tools
• Pydantic AI：Durable Execution
• Pydantic AI：Upgrade Guide
• LlamaIndex Workflows 文档
• LlamaIndex：Agents 教程（含 State、Human in the loop、Multi-agent）
• smolagents：Guided tour
• smolagents：Secure code execution
• Mastra：Agent Human-in-the-loop
• Mastra：Agent lifecycle
• Mastra：Workflow Control flow
• Mastra：从 .network() 迁移到 supervisor agents
• Agno 文档
• DSPy 首页

连接

• 智能体编排框架横评：同一个报销 Agent 换六种写法：系列总览，这一站是第 5 站。
• 智能体编排框架横评（4）：多智能体框架：上一站，角色团队式的多智能体框架，可对比这里把 Agent 当工具的委托。
• 智能体编排框架横评（6）：低代码与可视化平台：下一站，低代码平台；Agno 的 AgentOS 处在两者之间。
• 智能体编排框架横评（1）：LangChain 1.x：大而全路线的代表，和本站的少抽象路线对照。
• 智能体编排框架横评（2）：LangGraph：显式图运行时，对照 LlamaIndex 的事件驱动和 pydantic-graph。
• 主线：一个报销请求走完整个 Agent 系统：报销场景的出处。
• 跟读 learn-claude-code：17 个 harness 机制怎么挂到一个循环上：手写 harness 的参照系，框架替你写好的基本是其中循环内核那部分。
• 安全边界：让权限落到工具层：CodeAgent 执行隔离，审批不等于授权。
• 人在回路：四个框架的建单前确认写法。
• 可靠执行：状态、幂等与恢复：持久执行、恢复重放与建单幂等。
• 可观测性：追踪一条任务的代价：Pydantic AI 等框架的 OpenTelemetry 接入。
• 评测与可观测性：DSPy 优化器依赖的评测指标。
• 工具调用：用类型生成工具 schema 的共同做法。