Agent 工具调用工程化：从 Function Calling 到可控执行系统

核心观点：Function Calling 解决的是“模型能提出工具调用”，Agent 工具调用工程化要解决的是“这些调用如何被安全、正确、可靠且经济地执行”。

刚接触 Agent 时，我们很容易被这样的 Demo 打动：用户说一句“帮我查下订单”，模型选中 query_order，传入参数，工具返回结果，模型顺利回答。整个过程像极了给 LLM 装上了手脚。

然后我们把 Demo 接上生产环境。

很快，现实开始贡献素材：同一个查询重复执行五次；工具已经返回成功，Agent 还在继续调用；接口超时后模型重试，创建了两张订单；取消订单尚未完成，通知短信却先发出去了；一次简单查询调了十几轮模型，成本比订单运费还高。

问题不在于 Function Calling 没用，而在于**“能调用”只是能力边界，“可控执行”才是工程边界**。

这篇文章不重复介绍如何定义一个函数工具，而是从一个订单 Agent 出发，系统讨论八类常见故障，以及如何通过 Tool Execution Runtime 把工具调用变成一个可管理、可恢复、可评测的执行过程。讨论不绑定具体框架，关键设计可以迁移到 LangGraph、OpenAI Agents SDK、自研 Agent 或基于 MCP 的工具系统。

1. 从 Function Calling 到 Tool Execution：究竟差了什么？

先看一个常见的 Agent Loop：

while True:
    response = await llm.generate(messages, tools=tool_schemas)

    if response.tool_calls:
        messages.append(response.assistant_message)  # 保留工具调用及其 ID
        results = await execute(response.tool_calls)
        messages.extend(format_tool_results(response.tool_calls, results))
    else:
        return response.output

这个结构足以跑通 Demo，但它悄悄假设了很多事情：

1. 模型选的工具是对的，参数也是对的。

2. 工具执行结束，就代表业务任务完成。

3. 失败后再试一次，大概率无害。

4. 模型能正确记住已经做过什么。

5. 模型知道什么时候该停。

这些假设在真实业务中都可能失效。尤其要注意：模型给出的 Tool Call 是一个“执行请求”，不是一张可以绕过校验的通行证。

我把 Agent 工具调用的质量分成三个相互制约的目标：

三个目标存在张力。例如，缩短超时可能降低响应延迟，却增加误判超时和重试；一味减少工具调用可能省钱，却牺牲任务成功率。因此，不能把“调用次数越少”当成唯一优化目标。

1.1 先搞清楚三个“成功”

假设 create_order() 返回 HTTP 200，Agent 就宣布：“订单创建成功！”

问题是：HTTP 200 有时只表示网关接收了请求，或者异步任务已受理。订单可能仍在处理中，甚至最终创建失败。

例如，订单创建任务的后置条件可以是：获得合法的 order_id，且通过业务系统确认订单进入允许的目标状态。如果工具只返回 ACCEPTED，那就应当进入等待或状态查询分支，而不是用“成功”两个字提前庆祝。

2. 八类常见问题：先识别故障，再选择机制

为了让后面的方案不悬空，我们先定义一个订单 Agent。它拥有以下工具：

get_order(order_id)                  # 查询订单
search_orders(user_id, filters)      # 查找订单
create_order(items, address)         # 创建订单
cancel_order(order_id, reason)       # 取消订单
get_operation_status(operation_id)   # 查询异步操作状态
send_notification(order_id, channel) # 发送通知

需求很普通：用户能够查询、创建、取消订单，并收到必要的通知。但就是这些普通操作，足以暴露典型的工程问题。

以下按“正确性 → 可靠性 → 经济性”的顺序拆解。记住一个原则：同一现象未必只有一个原因，所以不要拿一种补丁解决所有问题。

3. 正确性：工具不是函数名列表，而是一组执行契约

3.1 Tool Description 是面向模型的 API 设计

先看一个不太负责任的工具定义：

{
  "name": "order",
  "description": "处理订单",
  "parameters": {"type": "object"}
}

“处理订单”四个字涵盖查询、创建、取消、支付、退款。模型只好凭感觉猜，猜错了我们还怪模型不聪明。

更好的做法是将工具职责拆清楚，明确适用条件和禁用边界：

{
  "name": "cancel_order",
  "description": "取消用户已确认要取消的订单。仅用于取消，不用于查询；执行前需校验订单归属和可取消状态。",
  "parameters": {
    "type": "object",
    "properties": {
      "order_id": {"type": "string", "minLength": 1},
      "reason": {"type": "string", "minLength": 1}
    },
    "required": ["order_id", "reason"],
    "additionalProperties": false
  }
}

当然，仅有参数 Schema 还不够。order_id 是非空字符串，并不意味着用户有权取消这个订单。结构合法、业务合法和操作授权是三道不同的门。

一个成熟的工具注册表，除了输入输出 Schema，还应该承载执行策略。下面是概念模型，并非特定 SDK 的现成类型：

from dataclasses import dataclass

@dataclass(frozen=True)
class ToolSpec:
    name: str
    input_schema: dict
    output_schema: dict
    read_only: bool
    idempotent: bool
    timeout_seconds: float
    max_attempts: int
    approval_required: bool
    side_effect_type: str  # none / reversible / irreversible

执行时由 Runtime 进行确定性的参数校验和权限校验。模型可以提议执行 cancel_order，但不能自己给自己批准权限。模型不是公司的法务，更不是数据库的超级管理员。

对于 MCP，也可以利用工具的 inputSchema、outputSchema 和注解表达部分契约。MCP 定义了 readOnlyHint、destructiveHint、idempotentHint 等风险提示，但规范强调：不可信服务端提供的注解不能直接充当安全依据。[1]

3.2 Tool Result 必须可被机器判定

很多工具返回类似：

"操作成功，请稍后查看。"

人看着觉得没问题，程序却很难判断“成功”的确切含义：已经提交？正在执行？最终完成？

我更倾向于统一结果信封（Result Envelope）：

{
  "status": "IN_PROGRESS",
  "operation_id": "op_789",
  "data": {"order_id": null},
  "error": null,
  "retryable": false,
  "next_poll_after_ms": 1000
}

建议至少区分这些状态：

• SUCCEEDED：工具契约对应的操作已经成功。

• IN_PROGRESS：操作被接受，仍在进行。

• FAILED：已明确失败，并给出机器可读错误码。

• UNKNOWN：无法确定远端是否已经执行，尤其常见于写操作超时。

• BLOCKED：执行被权限、风险策略或预算约束拦截。

其中 UNKNOWN 非常重要。网络断开只说明我们没有收到答案，不说明远端没有干活。

结构化输出也不能盲信。Runtime 应根据工具的来源、Schema、状态和业务后置条件进行验证。MCP 对 structuredContent 和 outputSchema 已有相关规范，同时建议客户端验证工具结果。[1]

3.3 结果验证要从“接口返回”走到“业务事实”

取消订单可以定义为：

前置条件：用户有权限操作此订单，订单状态允许取消。
执行动作：调用 cancel_order。
后置条件：订单状态为 CANCELED，或进入明确的异步取消流程。
失败分支：不能取消时返回原因，不允许继续发送“取消成功”通知。

实际工程中，后置条件可能通过操作结果、订单查询、事件回执或业务数据库确认。验证渠道本身也要可信：不能因为模型总结了一句“已经取消”，就把它当成状态存储。

4. 可靠性（一）：重复调用、无进展循环和停止条件

4.1 重复调用未必是 Bug

看两段日志：

场景 A：重复搜索
search_orders({"user_id": "u1"}) -> 订单列表 [A, B]
search_orders({"user_id": "u1"}) -> 订单列表 [A, B]
search_orders({"user_id": "u1"}) -> 订单列表 [A, B]

场景 B：合法轮询
get_operation_status("op_789") -> RUNNING
get_operation_status("op_789") -> RUNNING
get_operation_status("op_789") -> SUCCEEDED

同一个工具、同一组参数，前者可能纯属浪费，后者却是正常业务流程。单纯“同一工具最多调用两次”，很容易把正常轮询一起干掉。

需要明确区分三个概念：

Repeat Detection（重复检测）：发现相似的调用请求。

Deduplication（去重）：判断是否可以复用既有结果或拒绝重复操作。

Progress Detection（进展检测）：判断任务是否实际向完成条件推进。

重复调用只是一个信号；没有进展，才是我们更需要处理的情况。

4.2 先为调用生成指纹，但别把指纹当圣旨

一个常见起点是对调用做规范化：

import hashlib
import json


def fingerprint(tool: str, args: dict, scope: dict) -> str:
    payload = {
        "tool": tool,
        "args": args,
        "scope": scope,  # 如 tenant_id、授权主体、状态版本/数据版本
    }
    raw = json.dumps(
        payload, ensure_ascii=False, sort_keys=True,
        separators=(",", ":"),
    ).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()

但是，指纹只是候选重复调用标识，不是通用缓存键，更不是写操作的幂等键。

• 对只读查询，要考虑租户、授权范围、数据版本和 TTL；否则可能复用过期或越权的数据。

• 对分页工具，必须纳入游标；对时间相关查询，必须考虑查询窗口。

• 对创建订单等写操作，不能因为参数相同就断言是重复请求。用户可能真的想再买一件。应使用同一业务操作的稳定 ID判断重试身份。

• 对状态轮询，应遵守 next_poll_after、最大轮询次数和整体截止时间，而不是无限自旋。

4.3 建立 Progress Monitor，而不是只数调用次数

Progress Monitor 可以观察几类信号：

这些信号并非总能用一个通用分数表达。业务关键任务应定义明确的完成条件；没有统一进展语义的开放任务，可以组合有限步数、证据变化和预算阈值。

下面是一段简化的“无进展重复”检测代码。注意它只负责检测，不负责判断读操作可否缓存、写操作可否重试。

from collections import defaultdict


class LoopGuard:
    def __init__(self, threshold: int = 2):
        self.threshold = threshold
        self.no_progress_hits = defaultdict(int)

    def observe(self, call_fp: str, progress_marker: str,
                made_progress: bool, is_polling: bool = False) -> str:
        # 合法轮询由独立的间隔、次数、截止时间策略控制
        if is_polling:
            return "FOLLOW_POLL_POLICY"

        key = (call_fp, progress_marker)
        if made_progress:
            self.no_progress_hits[key] = 0
            return "CONTINUE"

        self.no_progress_hits[key] += 1
        if self.no_progress_hits[key] >= self.threshold:
            return "REPLAN_OR_STOP"
        return "CONTINUE"

生产环境中，还需要让“进展状态”持久化，并考虑跨工具循环（A → B → A → B）、重复结果哈希、有限历史窗口、多 Agent 共享状态等问题。不同任务对“进展”的定义也不同，不能让一个泛化规则包打天下。

4.4 触发循环检测后，怎么处理？

建议的处理顺序是：

1. 可复用就复用：仅对满足权限与新鲜度条件的读操作返回已有结果。

2. 该等就等：对于轮询，按调度策略延迟，而不是立刻再次请求。

3. 需要改变策略就重新规划：向模型说明“已执行的调用、已知结果、未满足的条件”，限制继续使用无效路径。

4. 无法安全推进就停止：返回明确的受阻状态；不要为了凑答案编造任务成功。

特别需要强调：停止控制必须掌握在 Runtime 手里。只在 System Prompt 中写“请不要重复调用”是提醒，不是边界。

模型轮次限制也并不等于工具调用次数限制。以 OpenAI Agents SDK 为例，max_turns 定义的是模型调用轮次，一轮可能包含多个工具调用。[2] 因此至少应分别限制：模型轮次、总工具调用数、同类重试次数、执行耗时、Token 和费用。

5. 可靠性（二）：超时、重试和幂等，别让补救变成二次伤害

这是整篇文章最适合让后端开发者感到熟悉的一节：Agent 世界里的分布式系统问题，依然是分布式系统问题。

5.1 “超时”不等于“没有执行”

Agent                 订单服务
  |                       |
  |--- create_order ----->|
  |                       | 订单已写入数据库
  |       X 网络断开      |
  |<--- 无响应 -----------|
  |                       |
  |--- create_order ----->| 可能创建第二张订单

这时，正确的结果不是 FAILED，而是 UNKNOWN：请求结果未知，需要确认。

对写操作，必须有明确的重试身份。例如，Runtime 为“本次任务中的创建订单步骤”分配稳定的 operation_id 或 idempotency_key；重试时沿用同一个键，真正的新业务操作才生成新键。

业务操作：task_42 / create_order_step_1
稳定幂等键：op_9fa...
第一次发送：超时
第二次发送：沿用 op_9fa...，或先按 op_9fa... 查询执行状态

真正的防重复保证必须由支持幂等的业务 API 或共享持久化协调层提供。只在 Agent 进程内放一个 set()，多实例部署或进程重启后照样失效。

Stripe 的幂等请求设计是可参考的真实案例：客户端通过幂等键标识重试请求，服务端识别相同业务请求。[3] 具体返回与缓存行为受其 API 版本和保留窗口约束，不能把某个供应商的规则直接当成所有工具的默认语义。

5.2 先分类错误，再决定是否重试

对于允许重试的操作，可使用指数退避加随机扰动（jitter），避免大量 Agent 在服务故障时一起重试、把服务压得更彻底。AWS 的相关工程实践也强调超时、有限重试、退避、jitter 以及副作用幂等的组合。[4]

# 伪代码：调用前须确认操作具备安全重试条件
for attempt in range(max_attempts):
    result = await invoke_with_timeout(tool_call)
    if result.status == "SUCCEEDED":
        break
    if result.status == "UNKNOWN":
        result = await reconcile_by_operation_id(operation_id)
        if result.status in {"SUCCEEDED", "IN_PROGRESS"}:
            break
        # UNKNOWN 不能默认视为可以再次写入
    if not is_retryable(result, tool_policy):
        break
    await sleep(full_jitter_delay(attempt, retry_after=result.retry_after))

上面特意将重试资格判断与重试等待算法分开。指数退避解决的是负载节奏，不解决重复副作用。别让一个精美的 Backoff 函数背上它根本不该背的锅。

5.3 处理副作用：需要业务系统一起配合

对于不能天然幂等的操作，工程上有几种常见组合：

• 业务侧唯一约束：为每次业务操作保存 operation_id，约束同一操作只能提交一次。

• 操作记录状态机：PENDING → RUNNING → SUCCEEDED/FAILED/UNKNOWN，允许对账、恢复和告警。

• Transactional Outbox：业务状态与待发事件在同一数据库事务中提交，再异步发送通知。

• 补偿操作（Saga）：跨系统流程已部分完成时，用业务可接受的补偿步骤回滚效果，而不是幻想存在一笔跨所有 API 的大事务。

这里要诚实地说明：跨网络、跨系统的副作用，不会因为我们用了 Agent 就自然拥有“严格的 Exactly-once 执行”。通常是通过至少一次投递 + 幂等处理 + 去重约束 + 对账与补偿，让业务结果达到可接受的一致性。

6. 可靠性（三）：状态、恢复、并发与安全边界

6.1 对话历史不是执行状态数据库

假设模型上下文里有一句“已创建订单”，但 Agent 重启后无法确定这句话对应哪个后端操作、是否已提交、是否已发通知。此时继续执行，就等于拿一段聊天记录当事务日志。

建议至少显式管理这些字段：

{
  "task_id": "task_42",
  "state_version": 7,
  "goal": "取消订单并通知用户",
  "steps": {
    "cancel_order": {
      "status": "SUCCEEDED",
      "operation_id": "op_cancel_1",
      "order_id": "order_123"
    },
    "send_notification": {
      "status": "PENDING",
      "operation_id": "op_notice_1"
    }
  },
  "remaining_budget": {
    "tool_calls": 5,
    "deadline_ms": 18000
  }
}

这个状态是由 Runtime 和业务系统维护的事实。模型可以读取它来作决策，但不能仅凭自己生成一句话就篡改 SUCCEEDED。

恢复执行时，通常需要：

1. 从持久化 Checkpoint 找到最后一次一致的任务状态。

2. 读取每个外部操作对应的稳定 ID、已提交结果及待确认状态。

3. 对 UNKNOWN 或执行中的操作发起查询或对账，先确认，再决定是否重试。

4. 只执行尚未成功提交的步骤，并通过幂等与唯一约束抵御重复触发。

LangGraph 的持久化机制使用 Checkpointer 保存线程状态、支持中断恢复；同时区分线程级 Checkpoint 与跨线程的 Store。[5] 但是，框架提供了恢复能力，不等于外部 API 的副作用自动变成幂等。这个边界必须由应用设计补齐。

6.2 一次返回多个 Tool Call，不代表可以一起执行

例如模型同时提出：

A. get_order(order_123)
B. cancel_order(order_123)
C. send_notification(order_123, "sms")

如果并发执行，可能出现：通知发出 → 订单取消失败 → 用户收到一条喜讯，随后发现只是个误会。

更可靠的方式是把依赖显式表达出来：

flowchart LR
    A[查询订单及权限] --> B{允许取消?}
    B -->|是| C[提交取消]
    C --> D{取消结果已确认?}
    D -->|是| E[发送取消通知]
    B -->|否| F[返回不能取消原因]
    D -->|否| G[等待或对账]

原则很简单：独立的只读操作可以受控并发；有依赖的写操作需要顺序屏障；共享数据的并发写入需要版本或锁语义。

可使用依赖 DAG、并发数上限、乐观锁（例如 state_version 或 If-Match）、任务级租约、唯一键和消息队列来实现。对于可能重复投递的通知，也应有稳定事件 ID 和消费侧去重。

6.3 模型有工具访问权，不等于有业务操作权

只读查订单与退款、删除数据、转账不是同一种风险等级。需要在执行前判断：

• 谁在操作：真实用户身份与租户上下文由后端注入，而非完全由模型参数决定。

• 能操作什么：按用户、资源、动作做服务端授权。

• 操作风险多高：敏感写入需要确认或审批，必要时重新验证用户身份。

• 输入从哪里来：网页、邮件、检索结果、工具输出都可能包含提示注入，不能把其中的“请忽略之前指令”当成授权命令。

• 工具的边界在哪里：对 MCP / 远程工具使用可信名单、网络出站限制、凭据隔离、速率限制和审计。

还要注意审批与执行之间的状态变化。如果用户审批的是“取消金额为 100 元的订单”，执行时订单内容已经变化，不能只凭旧审批票据继续完成新动作。高风险操作应校验批准时的关键参数与资源版本。

7. 架构落地：一个可控的 Tool Execution Runtime 长什么样？

前面讲了不少补丁。如果每个工具各写一份去重、重试和日志代码，Agent 很快会长成一棵谁也不想修剪的灌木。

更好的做法，是把通用执行控制收敛到一个 Tool Execution Runtime。它不一定是一个单独部署的服务；对小应用而言，可以先是 Agent Loop 与 Tool Executor 之间的一层模块。

flowchart TB
    U[用户目标] --> P[LLM / Planner]
    P --> A
    subgraph R[Tool Execution Runtime]
      direction TB
      A[Registry 与 Schema 校验] --> B[权限与审批策略]
      B --> C[重复检测与进展判断]
      C --> D[预算与依赖调度]
      D --> E[重试 幂等与执行]
      E --> F[结果验证与状态提交]
    end
    E --> T[外部 API / MCP / 业务服务]
    T --> E
    F --> S[(状态与操作日志)]
    F --> O[(Tracing / Metrics)]
    F --> P

逻辑图中的各模块可以共处一个进程；生产部署时，幂等登记、状态存储与并发协调则需要跨实例可靠地共享。

7.1 把执行控制放到统一入口

下面是简化的 Runtime 伪代码，重点展示控制顺序，不是可以原封不动上线的 SDK：

async def handle_tool_call(call, ctx):
    # ① 工具是否存在、参数结构与业务前置条件
    spec = registry.require(call.name)
    args = validate_and_normalize(spec.input_schema, call.arguments)

    # ② 每次都检查授权；命中缓存不能绕过权限
    await policy.authorize(ctx.actor, spec, args, ctx.resource_version)
    await policy.require_approval_if_needed(spec, args, ctx)

    # ③ 重复 / 循环 / 轮询判断
    decision = progress_monitor.decide(call, args, ctx)
    if decision.action == "WAIT":
        return tool_feedback(call.id, "IN_PROGRESS", decision.details)
    if decision.action == "REPLAN_OR_STOP":
        return tool_feedback(call.id, "BLOCKED", decision.details)

    # ④ 只读且满足新鲜度、授权和隔离条件才可复用
    if spec.read_only:
        cached = read_cache.get_valid(spec, args, ctx)
        if cached is not None:
            return tool_feedback(call.id, "SUCCEEDED", cached)

    # ⑤ 原子预留预算；并发执行还需调度和依赖控制
    reservation = await budget.reserve(ctx.task_id, spec)
    try:
        # 同一个逻辑操作必须使用稳定 ID，重试不得重建
        op_id = await operation_store.get_or_create_stable_id(
            ctx.task_id, ctx.step_id, spec.name, args
        )
        # executor 内部负责超时、有限重试和不确定结果对账
        result = await executor.execute(spec, args, op_id, ctx)
        validated = result_validator.check(spec, result, ctx)
        await state_store.commit_result(ctx, op_id, validated)
        return tool_feedback(call.id, validated.status, validated.data)
    finally:
        await budget.settle_or_release(reservation)

这段代码至少还省略了四类生产细节：

1. 原子操作记录：get_or_create_stable_id() 必须通过持久化唯一约束或事务实现；多个 Worker 不能各自生成不同 ID。

2. 竞态控制：检查状态和提交状态之间可能发生变化，需要锁、版本校验或事务边界。

3. 预算结算语义：失败、取消、超时都要正确结算，不能让预算永久被占住。

4. 协议消息关联：即便 Runtime 拒绝了调用，也要以模型/框架要求的格式响应对应的 tool_call_id；不能直接把 Tool Call 丢掉，导致消息协议状态不完整。

另外，result_validator.check() 需要区分“结构校验”和“业务确认”。如果任务需要跨服务确认，不能把这一步偷换成简单判断 status == 200。

7.2 不要把所有逻辑都塞进 Runtime

Runtime 应该负责通用规则，但不需要代替业务系统做所有判断。

这点尤其关键：Runtime 应当加强现有业务系统的安全边界，而不是试图在 LLM 前面复刻一套业务数据库。

现成框架也能帮忙。例如 OpenAI Agents SDK 提供工具级 Guardrails、工具启用控制、审批机制与运行轮次约束，但不同工具类型是否覆盖同一 Guardrail 管线，要按官方文档确认。[2][6] 选型时更应该问“我还缺哪些运行时保障”，而不是“哪个框架宣传页上的特性最多”。

8. 经济性：省的是完成任务的成本，不是机械地少调用一次

很多优化建议会说“尽量少调用工具”。这句话就像“后端服务尽量少访问数据库”，正确，但信息量并不够。

8.1 优先消除无效工作

可以从四个层面入手：

一是缓存。 对只读且稳定的查询，采用带租户和权限作用域的缓存，并明确 TTL、失效事件与数据版本；对价格、库存、订单状态这类高变化数据，不要只凭参数相同就无脑复用。

二是减少不必要的模型决策。 对确定性的流程，例如取消成功后发送固定模板通知，可以用工作流直接调度；不必每一步都请模型重新开会表决。

三是控制工具输出。 搜索工具返回 2 MB 日志，不应全部塞进模型上下文。可以结构化提取、摘要、分页或按需取字段，但要保留满足业务判断所需的证据与引用。

四是合理并发。 互不依赖的只读工具可以并发；有写入依赖的步骤要串行。吞吐提升不是把 asyncio.gather() 撒遍整个代码库。

8.2 不要只看 Token 成本

总执行成本可粗略拆成：

[ C_{total}=C_{model}+C_{tool}+C_{infra}+C_{failure} ]

其中：

• C_model：模型输入/输出 Token、推理轮次等成本。

• C_tool：搜索、第三方 API、数据库与工具调用成本。

• C_infra：计算、队列、存储与可观测性成本。

• C_failure：失败重试、人工兜底、错误业务操作等带来的成本。

更值得关注的指标是：

[ \text{Cost per Successful Task} = \frac{\text{统计窗口内任务总成本}}{\text{统计窗口内成功完成的任务数}} ]

举例说：某优化让平均工具调用从 6 次降至 3 次，却让任务成功率从 95% 降至 70%。这很可能只是把成本藏在失败任务和用户重试里了，不能直接宣布“效率提升 50%”。

预算也要设置多个维度：

limits:                     # 示例配置，非某框架固定语法
  max_model_turns: 12
  max_tool_calls: 20
  max_retries_per_operation: 2
  max_parallel_read_calls: 4
  task_deadline_seconds: 60
  max_estimated_cost_usd: 0.20

阈值需要按真实业务基线和 SLO 调整。审批等待不宜机械计算为普通计算超时；批量并发时预算计数应原子预留，以免四个并行分支各自认为“预算还够”。

9. 把案例走完：订单 Agent 从 V0 到 V5

只讲抽象规则，读者容易听完点头、回到项目后继续挠头。让我们把同一个 Agent 连续升级：

其中最值得做一次“故障注入”的，是 V2：

[10:00:00.000] task_42 开始 create_order，operation_id=op_9fa
[10:00:00.120] 订单服务提交成功，order_id=order_123
[10:00:00.140] 模拟丢弃 HTTP 响应
[10:00:02.000] Agent Runtime 超时，记录 UNKNOWN
[10:00:02.020] 查询 op_9fa 的执行状态
[10:00:02.050] 返回 SUCCEEDED，order_id=order_123
[10:00:02.060] 状态持久化并完成任务；未重复调用 create_order

这段轨迹比一句“我们加了重试机制”更能说明架构价值。关键不是 Agent 最终回复得多自然，而是它真的只创造了一次业务效果，而且我们能拿出证据。

10. 如何证明你的 Runtime 真有用？

工程方案最怕两件事：指标只看模型“自我感觉良好”，以及把一次漂亮的 Demo 当成稳定性证据。

10.1 构建可重复的故障评测集

为每类问题准备固定测试：

建议固定数据集、工具模拟器、模型版本和运行配置，对同一批任务重复测试。随机模型输出可以多次采样；超时、乱序、断网、429/503、进程重启等外部故障则尽量使用可复现的故障注入。

10.2 观察的不只是“工具调用了多少次”

建议至少记录：

trace_id / task_id / agent_run_id
model_turn_id / tool_call_id / operation_id
tool_name / normalized_args_hash
state_version / status / error_code
retry_attempt / dedup_decision / progress_decision
latency_ms / token_usage / estimated_cost
business_postcondition_passed

这些 ID 不能混为一谈：tool_call_id 关联模型的一次请求及结果；operation_id 标识一个可能跨重试、跨进程持续存在的业务操作；task_id 标识用户任务。

指标上至少包括：

• Task Success Rate：满足后置条件的任务占比。

• No-progress Call Rate：没有产生可定义进展的工具调用占比（需排除合理等待）。

• Duplicate Side-effect Count：重复产生实际业务副作用的次数；高风险业务通常要求为零。

• Recovery Success Rate：故障后能够自动恢复到正确状态的比例。

• P95 End-to-end Latency：用户任务从开始到完成的第 95 百分位耗时。

• Cost per Successful Task：完成一项有效任务的实际单位成本。

可观测性层面可以参考 OpenTelemetry 的 GenAI 语义规范，其中定义了 invoke_agent、execute_tool 等操作语义。[7] 但 progress_delta、业务后置条件是否通过等指标，还需要应用自己定义并上报。

不要编一个“优化后成功率提升 37%”来充门面。 没有基线和真实对照实验，就先展示方法、测试用例与评估脚本；数据值得相信，比数据看起来漂亮更重要。

10.3 一个最小测试骨架

import pytest


@pytest.mark.asyncio
async def test_create_order_timeout_does_not_duplicate(runtime, fake_backend):
    # 后端已提交，但客户端没有收到成功响应
    fake_backend.commit_then_drop_response_once()

    result = await runtime.run(
        task_id="task-42",
        goal="创建一笔订单",
        inputs={"items": [{"sku": "book", "qty": 1}]},
    )

    assert result.business_status == "SUCCEEDED"
    assert fake_backend.count_orders_for_operation("task-42/create") == 1

这段是测试契约示意，runtime 和 fake_backend 是测试夹具。真正的回归集还应验证：同一个幂等键不允许被复用到不同请求参数；两个并发 Worker 提交同一操作时只产生一次业务效果；过期或无权限的缓存不能被复用。

11. 真正落地时，我会按这个优先级做

不建议上来就打造一个几十个模块的“全能 Agent 操作系统”。先把风险最高、性价比最大的能力补齐：

第一阶段：先让系统不会胡来。

落实输入输出 Schema、服务端权限校验、工具超时、总调用预算、错误分类；对写操作定义可确认的业务后置条件。此时先保证“能拒绝危险和错误的请求”。

第二阶段：让系统不会把故障放大。

引入稳定操作 ID、业务端幂等、结果 UNKNOWN 状态、有限重试与对账；同时增加无进展循环熔断。此时重点保证“故障不会轻易产生第二次业务效果”。

第三阶段：让系统可以恢复和追责。

持久化状态、操作日志、Checkpoint、Trace 关联和执行审计；再支持并发依赖调度、审批与补偿。此时重点保证“哪一步发生了什么，我们能查、能恢复、能解释”。

第四阶段：用证据优化经济性。

在上述正确性和可靠性不变量成立的前提下，再做缓存、动态工具暴露、批量化、结果裁剪和成本优化。先跑评测，再看账单，而不是先砍调用次数再祈祷一切正常。

结语：把模型的不确定性关进工程的笼子里

Function Calling 让 LLM 从“会说”走向“会做”。这是 Agent 能力层面的进步，但也意味着模型产生的请求开始影响真实系统：数据库会被更新，消息会被发出，订单会被创建，钱也可能真的被扣掉。

当工具调用不再只是演示，我们就必须重新拿起一些经典但可靠的工程武器：Schema、权限、幂等、状态机、事务边界、重试预算、故障恢复、Tracing 和测试。

它们并不新潮，甚至有些“不够 AI”。但生产系统最怕的不是不够新潮，而是凌晨三点出现“Agent 一晚上替用户创建了 800 张订单”的事故单。

所以，我更愿意这样划分职责：

LLM 负责提出“下一步可能该做什么”；Runtime 负责判断“能不能做、如何执行、结果是否成立，以及什么时候必须停”。

衡量一个 Agent 的终点，不是工具被调用了多少次，而是用户目标是否以可证明的方式被正确、可靠、经济地完成。

让 Agent 会调用工具不难。让它调用之后的每一步都经得起追问，才是工程化真正开始的地方。

参考资料

以下为文章涉及的规范和官方工程文档，链接便于读者进一步查看具体实现与版本差异。

1. Model Context Protocol — Tools（2025-11-25 规范）：工具 Schema、structuredContent、outputSchema、工具注解与安全建议。

2. OpenAI Agents SDK — Runner：Agent Loop 与 max_turns 的定义。

3. Stripe — Idempotent requests：使用幂等键处理请求重试与重复业务操作。

4. AWS Builders' Library — Timeouts, retries, and backoff with jitter：分布式服务的超时、重试、退避与抖动策略。

5. LangGraph — Persistence：Checkpointer、Store、线程状态与恢复。

6. OpenAI Agents SDK — Guardrails：工具级 Guardrails、审批前后检查及不同工具类型的边界。

7. OpenTelemetry — GenAI Spans Semantic Conventions：Agent 与工具执行的 Trace 语义。

说明：本文代码以工程设计与测试契约为主；标记为“伪代码”或“示意”的片段刻意省略了框架适配、分布式原子性与鉴权基础设施，不能未经补充直接用于生产环境。