Agent 工程体系 · 第 95/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

Python 实现 Agent Runtime：类型、异步、工具、状态和测试

Agent Runtime 不是“调用一次大模型的函数包装器”，而是负责驱动一次 Agent 执行的运行时系统。它至少要回答六个问题：

  1. 当前 Agent 观察到了什么？
  2. 下一步应该继续思考、调用工具，还是结束？
  3. 工具调用如何被解析、校验和执行？
  4. 异步任务如何调度、取消和隔离？
  5. 中断或重启后，状态如何恢复？
  6. 不调用真实模型时，如何测试这些行为？

PydanticAI 将 Agent 描述为一个类型化的 Agent Loop，并把依赖、工具和结构化输出纳入类型系统；AutoGen 则把 Runtime 明确为消息投递、Agent 生命周期和状态保存的基础设施。两者抽象重点不同，但都围绕同一个核心：把模型生成的决策转换为可控的状态转换。(ai.pydantic.dev)

一、先定义 Agent Runtime

1. Agent Runtime 不等于 Agent

可以把一个 Agent 看成策略函数：

π:(Ot,St,Ct)→At\pi: (O_t, S_t, C_t) \rightarrow A_t

其中：

  • OtO_t：第 tt 步观察到的输入，例如用户消息、工具返回值；
  • StS_t：当前状态，例如消息历史、任务进度、重试次数；
  • CtC_t：运行上下文，例如数据库连接、用户身份、取消令牌；
  • AtA_t：动作，可以是工具调用，也可以是最终答案。

Runtime 的职责不是决定业务答案，而是执行状态转换：

(St,At)→execute(Ot+1,St+1)(S_t, A_t) \xrightarrow{\text{execute}} (O_{t+1}, S_{t+1})

因此，Runtime 至少包含以下组件：

┌────────────┐
│  输入消息   │
└─────┬──────┘
      ▼
┌────────────┐      ┌──────────────┐
│  状态存储   │◄────►│  运行上下文   │
└─────┬──────┘      └──────────────┘
      ▼
┌────────────┐
│ 模型适配器  │
└─────┬──────┘
      │ 决策：文本 / 工具调用
      ▼
┌────────────┐      ┌──────────────┐
│ 动作校验器  │─────►│  工具执行器   │
└─────┬──────┘      └──────┬───────┘
      │                    │
      └────────反馈─────────┘

如果代码只有：

answer = await model.generate(prompt)

那么它只是模型客户端调用，不是完整 Runtime。因为它没有定义工具调用、状态恢复、终止条件和故障路径。

2. Agent Loop 的六个阶段

一个可测试的运行循环可以形式化为：

观察 Observe
   │
   ▼
决策 Decide ────────► 最终输出 Finish
   │
   ▼
行动 Act
   │
   ▼
反馈 Feedback
   │
   ▼
状态更新 State Update
   │
   └──────────────► 回到观察

对应的伪代码如下：

while not state.terminated:
    observation = state.observe()
    decision = await policy.decide(observation, context)

    if decision.kind == "final":
        state.finish(decision.output)
        break

    if decision.kind == "tool":
        result = await tools.execute(decision.call, context)
        state.record_tool_result(decision.call, result)

    state.step += 1

这里最重要的不是 while，而是几个不变量：

0≤step≤max_steps0 \leq \text{step} \leq \text{max\_steps}

每一次工具返回都必须进入下一轮模型输入\text{每一次工具返回都必须进入下一轮模型输入}

最终输出只能在通过输出校验后提交\text{最终输出只能在通过输出校验后提交}

缺少第二条时，Agent 会调用工具，但模型看不到工具结果；缺少第三条时，模型输出的 JSON 只能被“看起来像结构化数据”，而不能被当作可靠业务对象。

二、用类型建模 Runtime 的边界

1. 类型不是装饰，而是运行时协议

Agent 系统中至少存在四类数据：

from dataclasses import dataclass
from typing import Literal, Any


@dataclass
class ToolCall:
    name: str
    arguments: dict[str, Any]


@dataclass
class FinalAnswer:
    content: str


Decision = ToolCall | FinalAnswer


@dataclass
class ToolResult:
    tool_name: str
    value: Any
    is_error: bool = False

Decision 是模型适配器和 Runtime 之间的协议。Runtime 不应该接收任意 dict，然后通过字符串判断：

if response.get("type") == "tool":
    ...

这种写法的问题是：字段拼写错误、参数类型错误和未知动作都会在很晚的阶段暴露。

更严格的做法是使用 Pydantic：

from typing import Any, Literal
from pydantic import BaseModel, Field


class ToolCall(BaseModel):
    kind: Literal["tool"] = "tool"
    name: str
    arguments: dict[str, Any] = Field(default_factory=dict)


class FinalAnswer(BaseModel):
    kind: Literal["final"] = "final"
    content: str


class ToolResult(BaseModel):
    tool_name: str
    value: Any
    is_error: bool = False

kind 是判别字段。它让 tool 和 final 成为两个明确的协议分支，而不是依靠调用方猜测字典结构。

2. 输出类型与工具参数类型是两条不同的边界

工具参数描述的是：

Model→Tool\text{Model} \rightarrow \text{Tool}

结构化输出描述的是：

Model→Application\text{Model} \rightarrow \text{Application}

例如，模型可以调用：

{
  "name": "get_balance",
  "arguments": {
    "user_id": 42
  }
}

工具返回后，模型再生成：

{
  "risk": 2,
  "should_block_card": false,
  "explanation": "..."
}

这两个 JSON 的校验责任不同：

  • user_id 由工具参数模型校验；
  • risk 和 should_block_card 由最终输出模型校验。

PydanticAI 的函数工具会根据函数签名和文档字符串生成工具 schema，并在工具执行前校验参数；Agent 的 output_type 则用于约束最终运行结果。(ai.pydantic.dev)

三、实现一个最小但完整的 Python Runtime

下面的示例不依赖任何模型供应商。我们用一个确定性的 FakeModel 模拟模型决策，重点展示 Runtime 的执行语义。

1. 定义状态和模型协议

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from typing import Any, Protocol

from pydantic import BaseModel, Field


class ToolCall(BaseModel):
    kind: str = "tool"
    name: str
    arguments: dict[str, Any] = Field(default_factory=dict)


class FinalAnswer(BaseModel):
    kind: str = "final"
    content: str


Decision = ToolCall | FinalAnswer


class AgentState(BaseModel):
    messages: list[dict[str, Any]] = Field(default_factory=list)
    step: int = 0
    terminated: bool = False
    final_output: str | None = None


@dataclass
class RunContext:
    state: AgentState
    deps: dict[str, Any]
    cancellation: asyncio.Event = field(default_factory=asyncio.Event)


class Model(Protocol):
    async def decide(
        self,
        state: AgentState,
        tools: list[str],
    ) -> Decision:
        ...

这里的 AgentState 是可持久化状态，RunContext 是一次运行期间的上下文。

两者不能混淆：

  • state 应该可以序列化并在重启后恢复；
  • deps 通常包含连接池、客户端、密钥或进程内对象，不应该直接写入状态；
  • cancellation 属于本次执行，不属于业务记忆。

2. 定义工具协议

from collections.abc import Awaitable, Callable


ToolHandler = Callable[[dict[str, Any], RunContext], Awaitable[Any]]


@dataclass
class Tool:
    name: str
    handler: ToolHandler

再定义一个显式工具注册表：

class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        if tool.name in self._tools:
            raise ValueError(f"duplicate tool: {tool.name}")
        self._tools[tool.name] = tool

    def names(self) -> list[str]:
        return list(self._tools)

    async def execute(
        self,
        call: ToolCall,
        context: RunContext,
    ) -> Any:
        tool = self._tools.get(call.name)
        if tool is None:
            raise ValueError(f"unknown tool: {call.name}")

        return await tool.handler(call.arguments, context)

工具注册表承担两个职责：

  1. 将模型输出的工具名解析成真实函数；
  2. 拒绝未知工具。

第二点不能省略。若模型可以直接生成任意 Python 函数名，Runtime 就可能把“不受信任的模型输出”变成任意代码执行路径。

3. 实现可取消的 Runtime

class AgentRuntime:
    def __init__(
        self,
        model: Model,
        tools: ToolRegistry,
        max_steps: int = 8,
    ) -> None:
        self.model = model
        self.tools = tools
        self.max_steps = max_steps

    async def run(
        self,
        user_input: str,
        *,
        state: AgentState | None = None,
        deps: dict[str, Any] | None = None,
    ) -> AgentState:
        state = state or AgentState()
        deps = deps or {}

        if state.terminated:
            return state

        state.messages.append({
            "role": "user",
            "content": user_input,
        })

        context = RunContext(
            state=state,
            deps=deps,
        )

        while not state.terminated:
            if context.cancellation.is_set():
                raise asyncio.CancelledError

            if state.step >= self.max_steps:
                raise RuntimeError(
                    f"agent exceeded max_steps={self.max_steps}"
                )

            decision = await self.model.decide(
                state,
                self.tools.names(),
            )

            state.step += 1

            if isinstance(decision, FinalAnswer):
                state.final_output = decision.content
                state.terminated = True
                state.messages.append({
                    "role": "assistant",
                    "content": decision.content,
                })
                continue

            state.messages.append({
                "role": "assistant",
                "tool_call": decision.model_dump(),
            })

            try:
                value = await self.tools.execute(decision, context)
            except Exception as exc:
                state.messages.append({
                    "role": "tool",
                    "name": decision.name,
                    "is_error": True,
                    "content": str(exc),
                })
                continue

            state.messages.append({
                "role": "tool",
                "name": decision.name,
                "is_error": False,
                "content": value,
            })

        return state

这个 Runtime 的故障语义是明确的：

  • 未知工具：进入工具错误消息，不直接终止；
  • 工具异常：将异常转换成模型可观察的反馈；
  • 超过步数：抛出 Runtime 错误；
  • 取消：抛出 CancelledError，由上层决定是否重试；
  • 最终答案：设置 final_output 并终止。

工具异常是否应反馈给模型，需要根据工具风险决定。对于“查询天气”这类只读工具，可以反馈错误；对于“扣款”“删除数据”这类副作用工具，不能仅靠模型自行恢复，应该配合幂等键、审批或补偿逻辑。

4. 一个可运行的确定性模型

class FakeModel:
    async def decide(
        self,
        state: AgentState,
        tools: list[str],
    ) -> Decision:
        tool_messages = [
            message
            for message in state.messages
            if message.get("role") == "tool"
        ]

        if not tool_messages:
            return ToolCall(
                name="add",
                arguments={"a": 20, "b": 22},
            )

        result = tool_messages[-1]["content"]
        return FinalAnswer(content=f"计算结果是 {result}")


async def add_tool(
    arguments: dict[str, Any],
    context: RunContext,
) -> int:
    a = arguments["a"]
    b = arguments["b"]

    if not isinstance(a, int) or not isinstance(b, int):
        raise TypeError("a and b must be integers")

    await asyncio.sleep(0)
    return a + b


async def main() -> None:
    tools = ToolRegistry()
    tools.register(Tool(name="add", handler=add_tool))

    runtime = AgentRuntime(
        model=FakeModel(),
        tools=tools,
    )

    state = await runtime.run("请计算 20 + 22")
    print(state.final_output)
    print(state.step)
    print(state.messages)


if __name__ == "__main__":
    asyncio.run(main())

预期输出类似：

计算结果是 42
2
[
  {'role': 'user', 'content': '请计算 20 + 22'},
  {'role': 'assistant', 'tool_call': {...}},
  {'role': 'tool', 'name': 'add', 'is_error': False, 'content': 42},
  {'role': 'assistant', 'content': '计算结果是 42'}
]

step == 2 的原因是：

  1. 第一步：模型决定调用 add；
  2. 第二步：模型观察到工具结果，生成最终答案。

这也说明了一个常见误解：工具调用不是 Agent 的最终输出，而是中间状态转换。

四、异步不是把函数声明为 async

1. 异步 Runtime 的三个层次

Python Agent Runtime 中，异步通常出现在三个层次：

模型请求 ───── await ─────┐
                          │
工具 A ─────── await ─────┼── 调度层
                          │
工具 B ─────── await ─────┘

第一层是模型网络请求；第二层是工具的 I/O；第三层是多个独立任务的并发调度。

将函数写成 async def 只表示它可以返回协程，并不自动带来并发。如果这样写：

result_a = await tool_a()
result_b = await tool_b()

那么 tool_b 必须等待 tool_a 完成。

只有在工具之间互不依赖、且没有共享可变状态时，才适合：

result_a, result_b = await asyncio.gather(
    tool_a(),
    tool_b(),
)

2. 并发的正确条件

设工具集合为 T={t1,t2,...,tn}T = \{t_1, t_2, ..., t_n\}。只有当：

∀i≠j,write(ti)∩read/write(tj)=∅\forall i \neq j,\quad \text{write}(t_i) \cap \text{read/write}(t_j) = \varnothing

并且工具之间没有业务顺序依赖时，才可以并行执行。

以下两个查询通常可以并行：

async def collect_profile(user_id: int) -> dict[str, Any]:
    profile, orders = await asyncio.gather(
        get_profile(user_id),
        get_orders(user_id),
    )
    return {
        "profile": profile,
        "orders": orders,
    }

以下操作不能简单并行：

await asyncio.gather(
    reserve_inventory(order_id),
    charge_payment(order_id),
)

因为支付成功但库存预留失败时，需要补偿；库存成功但支付超时时，也需要定义明确的事务语义。gather 只解决调度，不解决分布式事务。

3. 取消必须穿透调用链

一个可取消的工具不应吞掉取消异常：

async def fetch_data(
    client,
    cancellation: asyncio.Event,
) -> str:
    if cancellation.is_set():
        raise asyncio.CancelledError

    response = await client.get(
        "/data",
        timeout=5,
    )

    if cancellation.is_set():
        raise asyncio.CancelledError

    return response.text

实际生产中通常还需要：

  • 网络请求超时；
  • 单个工具的超时；
  • 整个 Agent Run 的截止时间；
  • 任务取消后的资源释放；
  • 重试时的幂等控制。

AutoGen 的消息处理接口显式携带 CancellationToken，用于将取消信号传递给 Agent 和工具；这比依赖全局变量或手工轮询更适合组合式 Runtime。(microsoft.github.io)

五、依赖注入：把资源放进上下文，不放进 Prompt

1. 依赖与状态的区别

依赖注入是把运行时资源显式传入 Agent：

from dataclasses import dataclass


@dataclass
class Dependencies:
    user_id: int
    database: Any
    http_client: Any

工具只通过上下文获取资源：

async def current_balance(
    arguments: dict[str, Any],
    context: RunContext,
) -> float:
    deps: Dependencies = context.deps["services"]
    return await deps.database.get_balance(deps.user_id)

不要把数据库连接序列化进消息：

# 错误方向
state.messages.append({
    "role": "system",
    "content": repr(database_connection),
})

数据库连接的生命周期属于进程或请求；消息历史属于业务状态。把两者混合，会导致：

  • 状态无法 JSON 序列化；
  • 恢复时连接对象已经失效；
  • 日志中泄露凭据或内部地址；
  • 测试无法替换依赖。

PydanticAI 使用 RunContext[DepsType] 向工具和指令函数提供类型化依赖；官方示例中，数据库连接和用户标识由依赖对象携带，而不是拼接到提示词中。(ai.pydantic.dev)

2. PydanticAI 的端到端示例

下面示例展示四个边界：

  • deps_type：依赖类型；
  • @agent.tool：工具注册；
  • 工具参数：由函数签名定义；
  • output_type：结构化输出。
from dataclasses import dataclass
from typing import Literal

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext


@dataclass
class SupportDeps:
    customer_id: int
    balance: float


class SupportOutput(BaseModel):
    answer: str
    risk: Literal["low", "medium", "high"]
    should_escalate: bool = Field(default=False)


agent = Agent(
    "test",
    deps_type=SupportDeps,
    output_type=SupportOutput,
    instructions=(
        "根据客户账户信息回答问题。"
        "如果余额不足以完成操作，应提高风险等级。"
    ),
)


@agent.tool
async def get_balance(
    ctx: RunContext[SupportDeps],
) -> float:
    """返回当前客户余额。"""
    return ctx.deps.balance


async def main() -> None:
    result = await agent.run(
        "我可以支付 100 元吗？",
        deps=SupportDeps(
            customer_id=7,
            balance=250.0,
        ),
    )

    output: SupportOutput = result.output
    print(output.answer)
    print(output.risk)
    print(output.should_escalate)

"test" 模型用于离线测试，不会访问真实 LLM；生产环境应替换为实际 provider model。PydanticAI 的设计目标是让 Agent 的依赖、工具和输出保持端到端类型一致，但模型本身仍可能生成错误决策，因此类型安全不能替代权限控制和业务校验。(ai.pydantic.dev)

六、工具执行：从模型调用到副作用控制

1. 工具调用的完整生命周期

一次工具调用至少经历以下阶段：

模型生成 tool_call
        │
        ▼
解析 JSON 参数
        │
        ▼
根据名称查找工具
        │
        ▼
校验参数类型和业务约束
        │
        ▼
鉴权、限流、幂等检查
        │
        ▼
执行外部副作用
        │
        ▼
规范化为 ToolResult
        │
        ▼
写入消息历史并反馈给模型

“参数能被解析”不等于“操作被授权”。例如：

class RefundRequest(BaseModel):
    order_id: int
    amount: float = Field(gt=0)

这只能保证 amount > 0，不能保证：

  • 订单属于当前用户；
  • 退款金额没有超过已支付金额；
  • 订单尚未退款；
  • 当前操作者拥有退款权限。

因此工具应采用两层校验：

async def refund(
    arguments: dict[str, Any],
    context: RunContext,
) -> dict[str, Any]:
    request = RefundRequest.model_validate(arguments)
    deps: SupportDeps = context.deps["support"]

    order = await deps.database.get_order(request.order_id)

    if order.customer_id != deps.customer_id:
        raise PermissionError("order does not belong to customer")

    if request.amount > order.paid_amount:
        raise ValueError("refund amount exceeds paid amount")

    refund_id = await deps.database.create_refund(
        order_id=request.order_id,
        amount=request.amount,
    )

    return {"refund_id": refund_id}

第一层是数据校验，第二层是业务与权限校验。两层都必须在工具内部完成，不能假设模型会提前完成。

2. 幂等是重试的前提

如果 Runtime 在网络超时后自动重试工具：

第一次请求：服务端已经成功扣款
客户端：因超时未收到响应
第二次请求：再次扣款

这不是“重试提高可靠性”，而是重复副作用。

有副作用的工具应接收幂等键：

class ChargeRequest(BaseModel):
    order_id: int
    amount: float
    idempotency_key: str

服务端根据 idempotency_key 保证同一逻辑操作最多提交一次。Runtime 可以生成：

idempotency_key = f"{run_id}:{tool_call_id}"

其中 run_id 标识一次 Agent Run，tool_call_id 标识该 Run 中的一次工具动作。

七、状态：消息历史不是全部状态

1. 三种状态

Agent Runtime 通常需要区分：

对话状态

messages: list[dict[str, Any]]

用于让模型知道此前发生了什么。

控制状态

step: int
retry_count: int
terminated: bool
pending_tool_call: str | None

用于驱动 Runtime 本身。

业务状态

order_id: int
approval_status: str
refund_id: str | None

用于表达任务领域事实。

如果只保存消息历史，恢复时可能无法判断：

  • 当前工具是否已经成功执行；
  • 当前是否等待人工审批；
  • 当前步骤是否已经提交；
  • 是否可以安全重试。

2. 可恢复状态必须可序列化

AutoGen 的 Agent 接口要求 save_state() 返回 JSON 可序列化的数据，并提供 load_state() 恢复状态；其 AgentChat 状态模型还包含类型和版本字段，用于区分不同状态结构。(microsoft.github.io)

一个简单的状态仓库可以这样写：

import json
from pathlib import Path


class JsonStateStore:
    def __init__(self, path: str) -> None:
        self.path = Path(path)

    def save(self, state: AgentState) -> None:
        temporary = self.path.with_suffix(".tmp")
        temporary.write_text(
            json.dumps(
                state.model_dump(),
                ensure_ascii=False,
                indent=2,
            ),
            encoding="utf-8",
        )
        temporary.replace(self.path)

    def load(self) -> AgentState:
        if not self.path.exists():
            return AgentState()

        data = json.loads(
            self.path.read_text(encoding="utf-8")
        )
        return AgentState.model_validate(data)

先写临时文件，再原子替换，避免进程在写入中途崩溃导致主状态文件只保存了一半。

生产系统还需要：

  • 状态版本；
  • 数据库事务；
  • 乐观锁或租约；
  • 敏感字段加密；
  • 消息和副作用的提交顺序；
  • 恢复后的重放或补偿机制。

3. 状态恢复的关键边界

假设状态停在：

{
  "step": 3,
  "pending_tool": "charge_payment",
  "payment_status": "unknown"
}

这里不能直接再次调用支付工具，因为 unknown 可能意味着：

  1. 请求尚未发送；
  2. 请求已发送但响应丢失；
  3. 支付成功但状态写入失败；
  4. 支付服务正在异步处理。

正确恢复路径通常是先查询支付服务状态，再决定是否补偿或重试：

恢复
 │
 ▼
查询幂等键状态
 ├── 已成功 ──► 记录成功，继续
 ├── 已失败 ──► 重新决策
 ├── 处理中 ──► 等待或轮询
 └── 不存在 ──► 才允许首次提交

这就是 Runtime 与普通函数调用的根本差异：Runtime 必须面对“代码执行和状态提交不在同一个原子事务中”的事实。

八、终止条件：没有终止策略的循环就是故障

Agent 可以在以下条件下终止：

class TerminationReason(str):
    FINAL_OUTPUT = "final_output"
    MAX_STEPS = "max_steps"
    TIMEOUT = "timeout"
    USER_CANCELLED = "user_cancelled"
    FATAL_TOOL_ERROR = "fatal_tool_error"

最小实现至少要有：

if state.step >= max_steps:
    raise RuntimeError("maximum step count exceeded")

但 max_steps 只是安全阀，不是成功条件。更完整的终止判定可以写为：

terminate=has_valid_output∨cancelled∨timeout∨step≥N∨fatal_error\text{terminate} = \text{has\_valid\_output} \lor \text{cancelled} \lor \text{timeout} \lor \text{step} \geq N \lor \text{fatal\_error}

常见错误包括：

  • 模型不断重复调用同一个工具；
  • 工具返回错误，但错误消息没有进入下一轮；
  • 结构化输出校验失败后无限重试；
  • 达到最大步数却仍返回半成品；
  • 超时只取消模型请求，没有取消正在执行的工具。

终止结果应携带原因，而不是只返回字符串：

class RunResult(BaseModel):
    status: Literal["succeeded", "failed", "cancelled"]
    output: Any | None = None
    termination_reason: str
    steps: int

这样监控系统才能区分“正常完成”和“因为步数耗尽而停止”。

九、测试：不要用真实 LLM 测试控制流

1. 测试分层

Agent Runtime 至少应该分成四层测试：

层次 测试对象 是否调用真实模型
单元测试 工具参数、权限、状态转换 否
Runtime 测试 工具调用循环、异常、终止 否
集成测试 模型适配器、schema 转换 可选
评测测试 真实任务成功率、轨迹质量 通常是

把这四层混成一个“调用真实模型然后断言答案字符串”的测试，会同时引入网络、延迟、价格、模型漂移和非确定性。

PydanticAI 官方建议使用 pytest，并通过 TestModel 或 FunctionModel 替代真实模型；还可以用 Agent.override 在测试范围内替换模型和依赖，并设置 ALLOW_MODEL_REQUESTS=False 防止误发真实请求。(pydantic.dev)

2. 测试最小 Runtime

import pytest


@pytest.mark.asyncio
async def test_runtime_executes_tool_then_finishes():
    tools = ToolRegistry()
    tools.register(Tool(name="add", handler=add_tool))

    runtime = AgentRuntime(
        model=FakeModel(),
        tools=tools,
        max_steps=4,
    )

    state = await runtime.run("计算")

    assert state.terminated is True
    assert state.final_output == "计算结果是 42"
    assert state.step == 2

    assert state.messages[1]["tool_call"]["name"] == "add"
    assert state.messages[2]["content"] == 42

这个测试没有验证模型是否“聪明”，而是验证 Runtime 的确定性协议：

  1. 模型产生工具调用；
  2. 工具被正确找到；
  3. 工具结果被记录；
  4. 第二轮产生最终答案；
  5. 状态被标记为终止。

3. 测试工具失败路径

class BrokenModel:
    async def decide(
        self,
        state: AgentState,
        tools: list[str],
    ) -> Decision:
        if state.step == 0:
            return ToolCall(
                name="missing_tool",
                arguments={},
            )
        return FinalAnswer(content="已处理失败")


@pytest.mark.asyncio
async def test_unknown_tool_becomes_feedback():
    tools = ToolRegistry()

    runtime = AgentRuntime(
        model=BrokenModel(),
        tools=tools,
        max_steps=4,
    )

    state = await runtime.run("执行")

    assert state.terminated is True
    assert any(
        message.get("is_error") is True
        for message in state.messages
        if message.get("role") == "tool"
    )

这个测试揭示了一个设计选择：未知工具是否作为反馈继续运行。对于严格系统，也可以把未知工具定义为致命错误：

if call.name not in self.tools.names():
    raise FatalRuntimeError(...)

选择哪一种，取决于错误是否可能由模型纠正，以及该错误是否代表安全边界被突破。

4. 用 FunctionModel 测试具体工具参数

TestModel 能验证工具是否被调用，但它生成的参数通常是为了通过 schema 校验，并不一定覆盖业务分支。PydanticAI 文档明确区分了 TestModel 和 FunctionModel：后者允许测试代码控制模型在每一轮返回什么工具调用或文本。(pydantic.dev)

例如要覆盖“过去日期调用历史接口、未来日期调用预测接口”，应由测试模型明确返回日期：

from datetime import date


def fake_weather_model(messages, info):
    if len(messages) == 1:
        return ModelResponse(
            parts=[
                ToolCallPart(
                    "weather_forecast",
                    {
                        "location": "Hangzhou",
                        "forecast_date": "2032-01-01",
                    },
                )
            ]
        )

    return ModelResponse(
        parts=[
            TextPart("The forecast is rainy")
        ]
    )

测试的重点不是模拟一个“像人一样”的模型，而是精确构造控制流输入：

第一轮：调用 weather_forecast(2032-01-01)
第二轮：观察工具结果
第三轮：生成最终文本

十、AutoGen 的 Runtime 抽象：消息驱动而非函数驱动

PydanticAI 更强调单个 Agent 的类型化循环；AutoGen Core 更强调 Agent、消息和 Runtime 的解耦。

AutoGen Core 的 Agent Runtime 通过消息向指定 Agent 发送请求，Agent 由 Runtime 调用其消息处理方法；Agent 还提供 save_state()、load_state() 和 close() 等生命周期接口。(microsoft.github.io)

一个简化的 AutoGen 思路是：

@dataclass
class UserMessage:
    content: str


class MyAgent(RoutedAgent):
    @message_handler
    async def handle(
        self,
        message: UserMessage,
        ctx: MessageContext,
    ) -> UserMessage:
        result = await self._model_client.create(
            messages=[...],
            tools=self._tools,
            cancellation_token=ctx.cancellation_token,
        )
        return UserMessage(content=str(result.content))

工具 schema 来自 Python 函数签名和描述：

async def get_stock_price(
    ticker: str,
    date: str,
) -> float:
    """返回指定股票在某日期的价格。"""
    return 140.05

然后包装为 FunctionTool：

from autogen_core.tools import FunctionTool

stock_tool = FunctionTool(
    get_stock_price,
    description="Get the stock price.",
)

AutoGen 的 Core API 不会替你完成完整 Agent Loop；开发者需要自行处理模型响应、工具调用、工具结果回填和最终响应。官方文档也明确指出 Core API 保持最小化，完整工具 Agent 逻辑需要由应用构建。(microsoft.github.io)

这带来两个取舍：

  • PydanticAI：单 Agent 业务代码较短，类型和校验集成较紧；
  • AutoGen Core：消息路由、Agent 生命周期和多 Agent 拓扑更显式，适合构造分布式或事件驱动系统。

AutoGen AgentChat 则提供更高层的 AssistantAgent。其 run() 会更新 Agent 内部消息历史，run_stream() 会以异步生成器形式产生中间事件和最终结果；默认情况下，工具通常由同一个 Agent 在一次运行中执行。(microsoft.github.io)

使用 Agent 作为工具时必须特别注意并发。因为 Agent 和 Team 持有内部状态，若模型并行发出多个 AgentTool 调用，就可能产生状态竞争；AutoGen 文档要求这类场景关闭并行工具调用。(microsoft.github.io)

十一、Pydantic Graph 什么时候比 while 更合适

普通 Agent Loop 适合：

观察 → 模型决策 → 工具 → 再观察

当业务流程出现固定分支、并行节点、汇聚节点或人工审批时，显式图更容易验证：

┌─────────┐
│ Validate│
└────┬────┘
     ▼
┌─────────┐       ┌──────────┐
│ Search  │──────►│ Synthesize│
└────┬────┘       └────┬─────┘
     ▼                 ▼
┌─────────┐       ┌──────────┐
│ Approve │──────►│  Finish  │
└─────────┘       └──────────┘

图模型的优势不是“更智能”，而是把隐式控制流变成显式节点和边：

G=(V,E)G = (V, E)

  • VV：节点，例如查询、校验、审批；
  • EE：边，例如成功、失败、需要人工介入。

如果一个流程必须满足：

支付前必须完成风控
退款前必须完成身份验证
高风险订单必须进入人工审批

那么把这些约束编码到图中，比将它们写进系统提示词更可靠。Pydantic Graph 提供步骤、决策、并行执行以及汇聚能力，并延续类型化节点和数据传递模型。(pydantic.dev)

但图也有成本：

  • 节点状态需要设计；
  • 图版本需要兼容历史运行；
  • 动态任务不一定适合静态拓扑；
  • 图上的每个边界都可能成为恢复点。

因此，while 不是低级实现，Graph 也不是自动升级。选择标准是：控制流是否已经稳定到值得显式建模。

十二、诊断 Runtime 的正确顺序

当 Agent 表现异常时，不要先改 Prompt。应按执行链逐层检查：

1. 检查模型输出

确认模型到底返回了：

  • 普通文本；
  • 工具调用；
  • 多个工具调用；
  • 无效 JSON；
  • 未注册工具；
  • 缺少必填参数。

2. 检查 schema

工具 schema 是否准确描述：

  • 参数名称；
  • 参数类型；
  • 枚举值；
  • 必填字段；
  • 业务限制。

“模型没有调用工具”可能不是模型能力问题，而是工具描述让它无法判断何时调用。

3. 检查 Runtime 状态

打印每一轮：

print({
    "step": state.step,
    "last_message": state.messages[-1],
    "terminated": state.terminated,
})

需要确认工具结果是否真的被追加到了下一次模型输入中。

4. 检查工具副作用

区分：

  • 工具根本没有执行；
  • 工具执行了但参数错误；
  • 工具执行成功但结果未写入状态；
  • 工具执行超时；
  • 工具已成功但 Runtime 重试造成重复副作用。

5. 检查终止原因

不要只记录：

agent failed

至少记录：

{
  "status": "failed",
  "reason": "max_steps",
  "step": 8,
  "last_tool": "search_orders",
  "run_id": "..."
}

PydanticAI 支持对运行过程进行消息捕获和测试断言；AutoGen 则通过消息事件、运行结果和 Agent 状态接口暴露执行轨迹。它们都说明了同一个工程事实：最终答案不足以诊断 Agent，必须保存中间轨迹。(pydantic.dev)

十三、常见错误及其边界

错误一：把 Agent 当成无状态函数

async def answer(prompt: str) -> str:
    return await model(prompt)

这种函数无法自然表达工具结果、重试、恢复和人工审批。它可以是 Runtime 的一个模型适配器，但不能代表 Runtime 本身。

错误二：把所有内容都放进消息历史

连接池、缓存、密钥、取消令牌和事务对象不属于对话状态。它们应通过依赖或运行上下文注入。

错误三：把类型校验当成权限控制

Pydantic 可以验证：

amount > 0

但不能单独验证：

当前用户是否有权为该订单退款

权限必须由工具访问的业务服务执行。

错误四：所有工具调用都并行

并行只适用于无共享状态、无顺序依赖、无冲突副作用的任务。带内部会话状态的 AgentTool 或 TeamTool 尤其不能无条件并行。(microsoft.github.io)

错误五：测试只断言最终文本

最终文本可能碰巧正确，但工具调用路径已经错误。例如：

  • 调用了不该调用的工具；
  • 使用了错误的用户 ID；
  • 走了历史数据接口而不是实时接口；
  • 重复执行了支付操作。

测试还应断言工具名称、参数、调用顺序、错误反馈和状态变化。

十四、一个可落地的 Runtime 最小契约

在进入生产前，Runtime 至少应该稳定以下契约：

class RuntimeContract(Protocol):
    async def run(
        self,
        input_text: str,
        *,
        state: AgentState | None,
        deps: Any,
    ) -> RunResult:
        ...

    async def cancel(self, run_id: str) -> None:
        ...

    async def save_state(self, run_id: str) -> dict[str, Any]:
        ...

    async def load_state(
        self,
        run_id: str,
        state: dict[str, Any],
    ) -> None:
        ...

对应的验证条件是：

  1. 所有模型动作都经过 schema 校验；
  2. 所有工具名称都经过注册表解析；
  3. 所有工具异常都有明确分类；
  4. 所有循环都有步数或时间上限；
  5. 所有可恢复状态都可序列化；
  6. 所有副作用工具都有幂等策略；
  7. 所有异步任务都能被取消；
  8. 所有关键轨迹都可测试和观测；
  9. 真实模型请求可以在单元测试中被强制禁止；
  10. 最终输出必须经过结构化校验。

Agent Runtime 的核心不是让模型“多想几步”，而是建立一套可证明、可恢复、可取消、可测试的执行协议。类型系统约束数据边界，异步调度管理等待和并发，工具系统连接外部世界，状态系统承担恢复，测试系统验证每一次状态转换。只有这些部分同时成立，观察—决策—行动—反馈—状态—终止 才真正成为一个工程运行时，而不是一段偶尔成功的模型调用代码。

系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。