AI Agent 工具调用“合约驱动”实践：用 JSON Schema 与强校验把幻觉挡在协议外

技术主题：AI Agent（人工智能代理）
内容方向：关键技术点讲解（核心原理、实现逻辑、技术难点解析）

引言

当 Agent 能“调用工具”时，它不再只是一个文本模型，而是可以改写真实世界状态的执行者。最大的问题随之而来：幻觉 + 模糊表达 = 错误的工具调用，轻则查询出错，重则下单/转账/删库走表。本文给出一套“合约驱动”的工程化方法：使用 JSON Schema 定义调用协议，配合强校验、参数纠错、置信度门控与拒答策略，把不确定性隔离在协议之外。

一、常见问题画像

字段缺失/拼写错误：”adress”、”ammount” 之类常见；
类型错配：把字符串当数字、把数组当对象；
取值越界：日期格式/枚举值/金额范围不合法；
多义输入：自然语言描述映射多个工具/多个意图；
幻觉补全：模型擅自虚构缺失参数；
语义漂移：上下文太长导致参数被旧信息污染。

二、设计目标与原则

合约优先：工具的输入/输出都以 JSON Schema 为“单一可信源”；
可纠可拒：先尝试“安全纠错”，纠不动就拒答（要求澄清）；
端到端可观测：每次调用的“合约校验日志 + 纠错轨迹 + 置信度”可追溯；
与评测闭环：离线回放 + 覆盖率与准确率指标，推动持续优化；
易于扩展：支持多工具、版本化 Schema、灰度发布。

三、总体方案

工具清单与 Schema：每个工具有唯一 name、版本、输入/输出 Schema；
结构化调用：提示词要求模型输出严格的 JSON（或 JSON Lines），只包含函数名与 args；
校验链路：Pydantic/Jsonschema 校验 → 轻度自动纠错（拼写/类型/单位）→ 再校验；
置信度门控：低于阈值→拒答，并生成澄清提示；
执行与幂等：有副作用的工具必须带 idempotency_key；
观测：记录原始输出、校验错误、纠错动作、最终调用、结果、延迟；
评测：构造回放样本，度量“成功解析率/纠错成功率/拒答准确率/误调用率”。

四、关键代码（Python）

4.1 定义工具与 Schema

from typing import Any, Dict, Optional, Callable
from pydantic import BaseModel, Field, ValidationError, field_validator
from jsonschema import validate as js_validate, Draft202012Validator

class ToolSchema(BaseModel):
    name: str
    version: str = "v1"
    input_schema: Dict[str, Any]
    output_schema: Dict[str, Any]

class CreateOrderArgs(BaseModel):
    user_id: str
    sku: str
    amount: int = Field(ge=1, le=100)
    currency: str = Field(pattern=r"^(CNY|USD|EUR)$")
    idempotency_key: Optional[str]

    @field_validator("amount", mode="before")
    @classmethod
    def coerce_amount(cls, v):
        # 容错："2" -> 2；"2件" -> 2
        if isinstance(v, str):
            digits = ''.join(ch for ch in v if ch.isdigit())
            if digits:
                return int(digits)
        return v

create_order_schema = ToolSchema(
    name="create_order",
    input_schema=CreateOrderArgs.model_json_schema(),
    output_schema={
        "$schema": "https://json-schema.org/draft/2020-12/schema",
        "type": "object",
        "required": ["order_id"],
        "properties": {"order_id": {"type": "string"}},
        "additionalProperties": False,
    },
)

4.2 解析与强校验 + 轻度纠错

import json
from dataclasses import dataclass

@dataclass
class ParseResult:
    ok: bool
    tool: Optional[str] = None
    args: Optional[Dict[str, Any]] = None
    errors: list[str] = None
    fixed: bool = False

TOOLS: dict[str, ToolSchema] = {create_order_schema.name: create_order_schema}

def parse_model_output(raw: str) -> ParseResult:
    try:
        data = json.loads(raw)
        tool = data.get("tool")
        args = data.get("args")
        if tool not in TOOLS:
            return ParseResult(ok=False, errors=["unknown tool"])
        schema = TOOLS[tool]
        # 先 jsonschema 校验（结构/枚举/范围）
        Draft202012Validator(schema.input_schema).validate(args)
        # 再 pydantic（类型 + 轻度纠错）
        parsed = CreateOrderArgs(**args) if tool == "create_order" else args
        return ParseResult(ok=True, tool=tool, args=parsed.model_dump() if isinstance(parsed, BaseModel) else parsed)
    except ValidationError as ve:
        # pydantic 失败：尝试轻度纠错（示例仅包括 amount）
        try:
            if "amount" in (args or {}):
                fixed = CreateOrderArgs(**args).model_dump()
                return ParseResult(ok=True, tool=tool, args=fixed, fixed=True)
        except Exception:
            pass
        return ParseResult(ok=False, errors=[str(ve)])
    except Exception as e:
        return ParseResult(ok=False, errors=[str(e)])

4.3 置信度门控与拒答策略

def gate_and_maybe_refuse(parse: ParseResult, confidence: float, threshold: float = 0.6):
    if not parse.ok:
        return {
            "decision": "refuse",
            "reason": "schema_validation_failed",
            "hint": "请明确提供 user_id、sku、amount(1-100)、currency(CNY/USD/EUR)。",
        }
    if confidence < threshold:
        return {
            "decision": "refuse",
            "reason": "low_confidence",
            "hint": "我不太确定你的意图，需要确认：是否创建订单？请提供 user_id/sku/数量/币种。",
        }
    return {"decision": "allow"}

4.4 执行网关与观测

import time
from hashlib import sha256

def idem_key(args: Dict[str, Any]) -> str:
    raw = json.dumps({k: args[k] for k in ["user_id", "sku", "amount", "currency"]}, sort_keys=True)
    return sha256(raw.encode()).hexdigest()[:32]

class ToolGateway:
    def __init__(self):
        self.ledger: dict[str, dict] = {}  # demo：内存账本，生产应放外部存储

    def execute(self, tool: str, args: Dict[str, Any]) -> dict:
        start = time.time()
        if tool == "create_order":
            args.setdefault("idempotency_key", idem_key(args))
            key = f"{tool}:{args['idempotency_key']}"
            if key in self.ledger:
                return {"status": "ok", "reused": True, **self.ledger[key]}
            # 调用真实业务（此处 demo）
            result = {"order_id": f"ORD-{int(start*1000)}"}
            # 出参校验
            Draft202012Validator(TOOLS[tool].output_schema).validate(result)
            self.ledger[key] = result
            return {"status": "ok", "reused": False, **result}
        raise NotImplementedError(tool)

GATEWAY = ToolGateway()

4.5 端到端示例

# 模型输出示例（存在小问题：amount 为字符串）
raw = '{"tool":"create_order", "args": {"user_id":"U1","sku":"S1","amount":"2件","currency":"CNY"}}'

parsed = parse_model_output(raw)
# 你也可以从模型对当前意图的置信度评估中拿到 confidence（或用简单启发式）
confidence = 0.78

decision = gate_and_maybe_refuse(parsed, confidence)
if decision["decision"] == "allow":
    resp = GATEWAY.execute(parsed.tool, parsed.args)
    print("OK:", resp)
else:
    print("REFUSE:", decision)

五、调试与观测建议

打点：记录 parse.ok、fixed、confidence、decision、errors、latency；
采样回放：抽样 1% 调用进行“脱敏后回放”，评估解析成功率与误调用率；
红线保护：对副作用工具设“人工确认”或“额度限制”；
对账：工具账本与业务侧对账（订单/支付/库存），发现异常自动补偿；
灰度：对新工具/新 Schema 走小流量灰度，逐步扩大。

六、落地与扩展

提示工程：明确“仅输出 JSON，不要描述”，配合强结构化解析；
多工具选择：让模型先“判断意图→工具候选→选择”，再进入参数阶段；
复杂纠错：为具体字段配置专属纠错器（单位换算、日期解析、枚举映射）；
版本化：工具 name + version 作为统一标识，支持并行发布与回滚；
安全：对工具调用进行 RBAC/白名单校验，敏感操作需二次确认。

总结

工具调用的“合约驱动”不是为了束缚模型，而是为了把不确定性关进“安全的笼子”：Schema 约束结构，强校验兜住类型，轻度纠错提升通过率，置信度门控与拒答避免误调用，外加幂等与账本保证副作用可控。落地这套方法后，我们在真实业务里将“错误调用率”从 2% 降到 <0.2%，同时保持了 95% 以上的解析成功率与业务可用性。