AI Agent 上线就翻车？问题多半不在模型，而在工具接口

一个 270 行、零依赖的可运行示例，讲清参数校验、错误回传、幂等和循环熔断。

做 Agent 的团队，路径往往惊人地相似。

Demo 阶段，接一个查询接口，模型表现惊艳：能理解意图、能选对工具、能把结果组织成人话。于是信心满满地接上真实系统，加上写操作——退款、下单、发工单——成功率立刻掉下来。第一反应通常是换模型、加 few-shot、把 System Prompt 写到两千字。折腾一圈，该错还是错。

我的观察是：大部分所谓"模型不听话"，其实是接口没设计好。

模型看不到你的数据库、看不到你的状态机，它只能看到你给的工具描述和一段 JSON Schema。它调错，很多时候是因为你让它猜。把不确定性收在模型这一层，外面的世界必须是确定的——这件事只能靠工程手段解决。

本文用一个售后 Agent 的完整示例，把四件事讲透。代码单文件、无第三方依赖，可以直接跑：

py -3 agent_loop.py

一、先给失败分个类

在动手改 Prompt 之前，先把线上失败日志翻出来分个类。绝大多数能归到下面五类：

只有第二类真正跟"模型聪不聪明"有关。剩下四类，换成再强的模型也救不回来——因为问题出在模型之外。

下面按这四类逐条拆。

二、把 Schema 当接口契约，而不是文档

很多人写工具定义，是把 OpenAPI 文档翻译一遍，字段能少写就少写，description 能省就省。但这份 Schema 是模型决策的唯一依据，它值得按"给新同事看的接口文档"标准来写。

三条最有效的约束：

1. 一个工具只做一件事。
"查订单并退款"这种复合工具，模型一定会漏参数。拆成 get_order 和 create_refund，让它一步步来，步数多一点，但可控。

2. 枚举优于自由文本。

反面例子：

{
  "reason": { "type": "string", "description": "退款原因" }
}

模型会给你写出"不想要了"“客户反悔”"七天无理由"等十几种说法，你的下游系统一个都接不住。

正面例子：

{
  "reason": {
    "type": "string",
    "enum": ["user_cancel", "out_of_stock", "price_error"],
    "description": "退款原因。用户主动取消选 user_cancel；缺货选 out_of_stock；价格标错选 price_error"
  }
}

枚举不只是校验，它同时是给模型的选项提示。三个选项摆在这里，模型就不需要发明第四个。

3. description 写"什么时候用"，而不是"这是什么"。

"订单号" 这种描述没有信息量。写成 "订单号，形如 A1001；用户只给了手机号时先用 search_order 查"，模型选错工具的概率会明显下降。

还有一个容易被忽略的细节：命名风格要全库统一。同一套工具里一半 order_id、一半 orderId，模型一定会串。这不是模型笨，是你在给它挖坑。

三、错误要能被模型读懂

这是投入产出比最高的一处改造。

大多数后端接口报错是这样的：

raise HTTPException(status_code=500, detail="internal error")

模型收到这行字，唯一能做的决策是"再试一次"，而且大概率用一模一样的参数再试。三次之后，你的步数预算就烧完了。

正确的做法是把错误当成给模型看的输入，结构化地告诉它三件事：哪个字段、错在哪、正确选项是什么。

{
  "ok": false,
  "error": {
    "code": "INVALID_ARGS",
    "message": "参数校验未通过，请按 problems 修正后重试",
    "problems": [
      {"field": "order_id", "reason": "missing", "hint": "订单号，例如 A1001"},
      {"field": "orderId", "reason": "unknown_field", "hint": "可用字段: ['order_id']"}
    ]
  }
}

注意这里同时点出了 order_id 缺失和 orderId 是未知字段——模型下一轮基本能自己改对。示例代码跑出来的真实效果：

step 1  get_order({"orderId": "A1001"})  → INVALID_ARGS
step 2  get_order({"order_id": "A1001"}) → ok
step 3  生成最终答复

一次无效调用，成本可以接受。如果没有这条 hint，它会一直错下去。

两个边界要注意：

• 不要把堆栈、SQL、内网地址返回给模型。 既有注入风险，也会把上下文搅乱。
• 区分"可重试"和"不可重试"。 参数错是可重试的（改参数），权限不足、资源不存在重试多少次都没用，应该在 error 里明确 retryable: false，让模型尽早转人工。

四、副作用工具必须幂等

读工具重试没有代价，写工具重试就是事故。

模型重试的原因很多：网络超时、上游 502、你自己返回的错误不够清楚、甚至只是它在同一步里并发发了两个调用。只要写接口不是幂等的，"退款发两次"就是时间问题。

三个层次的防护，建议都上：

第一层：幂等键。 用"工具名 + 规范化后的参数"生成 key，参数先按 key 排序再序列化，保证 {"a":1,"b":2} 和 {"b":2,"a":1} 命中同一个 key。

cache_key = name + ":" + json.dumps(args, sort_keys=True, ensure_ascii=False)

第二层：存储层兜底。 内存字典只能挡住单进程内的重复。生产环境换成 Redis 的 SETNX，或者在业务表上建唯一索引（比如 refund_order_id 唯一）。即使 Agent 服务重启、多副本部署，也不会重复扣款。

第三层：状态机 + 人工闸门。 高金额、不可逆的操作，不要让模型"一步到位"。让它先创建一条"待确认"记录，由规则或人来点最后一下。Agent 负责发起，人负责授权——这不是能力妥协，是责任边界。

五、给循环装刹车

Agent 跑飞的典型画面：模型在两个工具之间来回横跳，每一步都"合理"，但就是不给结论，直到步数耗尽、账单爆炸。

三道刹车，成本都很低：

1. 步数预算。 max_steps 给 6～8 步，超了就返回 budget_exhausted，带上完整 trace 转人工。别设 50，设 50 等于没设。
2. 重复调用检测。 同一工具、同一参数出现第三次，直接熔断。正常任务不会需要三次一模一样的调用，出现即异常。
3. 工具级超时与重试上限。 单次调用 3～5 秒，重试最多 2 次，且只对幂等工具重试。

这三条不需要模型配合，纯编排层就能实现。示例代码里对应 escalate 和 budget_exhausted 两个终止状态——Agent 必须有能力"承认自己搞不定"，否则它只会用更贵的方式继续错。

下面是完整实现，单文件、无第三方依赖。接真实模型时，把 FakeModel 换成一个 OpenAI 兼容的调用即可（华为云 MaaS 上的模型服务提供 OpenAI 兼容接口，地址和模型名以控制台文档为准）。

"""agent_loop.py —— 一个带参数校验、错误回传、幂等与循环熔断的工具调用循环。

不依赖任何 SDK，可直接 python agent_loop.py 运行离线演示。
接真实模型时，把 FakeModel 换成 OpenAIChatModel 即可。
"""
from __future__ import annotations

import json
import time
from dataclasses import dataclass
from typing import Any, Callable

SYSTEM_PROMPT = (
    "你是一个售后助手。只能通过工具获取事实，不要凭空猜测订单信息。"
    "工具返回 ok=false 时，请阅读 error 中的 problems/hint 修正参数后重试。"
)

# ---------------------------------------------------------------- 工具定义
@dataclass
class Tool:
    name: str
    description: str
    schema: dict
    fn: Callable[..., dict]
    idempotent: bool = False


TYPE_MAP = {
    "string": str,
    "integer": int,
    "number": (int, float),
    "boolean": bool,
    "object": dict,
    "array": list,
}


def validate_args(args: Any, schema: dict):
    """只校验模型最容易出错的三件事：必填、类型、枚举。"""
    if not isinstance(args, dict):
        return [{"field": "$", "reason": "expected object", "got": type(args).__name__}]

    problems = []
    props = schema.get("properties", {})
    for name in schema.get("required", []):
        if name not in args:
            problems.append({
                "field": name,
                "reason": "missing",
                "hint": props.get(name, {}).get("description", ""),
            })
    for name, value in args.items():
        spec = props.get(name)
        if spec is None:
            problems.append({
                "field": name,
                "reason": "unknown_field",
                "hint": f"可用字段: {list(props)}",
            })
            continue
        expected = TYPE_MAP.get(spec.get("type"))
        if expected and isinstance(value, bool) and spec.get("type") != "boolean":
            problems.append({"field": name, "reason": "expected " + spec["type"], "got": "boolean"})
            continue
        if expected and not isinstance(value, expected):
            problems.append({
                "field": name,
                "reason": "expected " + spec["type"],
                "got": type(value).__name__,
            })
        if "enum" in spec and value not in spec["enum"]:
            problems.append({
                "field": name,
                "reason": "not_in_enum",
                "got": value,
                "hint": spec["enum"],
            })
    return problems


# ---------------------------------------------------------------- 执行器
class ToolRunner:
    def __init__(self, tools, max_attempts: int = 2):
        self.tools = {t.name: t for t in tools}
        self.max_attempts = max_attempts
        self._cache = {}
        self.stats = {"calls": 0, "invalid": 0, "errors": 0, "cache_hits": 0}

    def specs(self):
        return [{
            "type": "function",
            "function": {
                "name": t.name,
                "description": t.description,
                "parameters": t.schema,
            },
        } for t in self.tools.values()]

    def call(self, name: str, args: Any) -> dict:
        self.stats["calls"] += 1
        tool = self.tools.get(name)
        if tool is None:
            return {"ok": False, "error": {
                "code": "UNKNOWN_TOOL",
                "message": f"没有名为 {name} 的工具",
                "hint": f"可用工具: {list(self.tools)}",
            }}

        problems = validate_args(args, tool.schema)
        if problems:
            self.stats["invalid"] += 1
            return {"ok": False, "error": {
                "code": "INVALID_ARGS",
                "message": "参数校验未通过，请按 problems 修正后重试",
                "problems": problems,
            }}

        cache_key = None
        if tool.idempotent:
            cache_key = name + ":" + json.dumps(args, sort_keys=True, ensure_ascii=False)
            if cache_key in self._cache:
                self.stats["cache_hits"] += 1
                return {**self._cache[cache_key], "cached": True}

        last_error = {"code": "UNKNOWN", "message": "未执行"}
        for attempt in range(1, self.max_attempts + 1):
            try:
                result = {"ok": True, "data": tool.fn(**args)}
                if cache_key:
                    self._cache[cache_key] = result
                return result
            except Exception as exc:  # 生产环境按异常类型细分
                last_error = {
                    "code": "TOOL_ERROR",
                    "message": f"{type(exc).__name__}: {exc}",
                    "retryable": attempt < self.max_attempts,
                }
                time.sleep(0.2 * attempt)
        self.stats["errors"] += 1
        return {"ok": False, "error": last_error}


# ---------------------------------------------------------------- 主循环
def run_agent(model, runner, task: str, max_steps: int = 6):
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": task},
    ]
    trace = []
    seen = {}

    for step in range(1, max_steps + 1):
        msg = model(messages, runner.specs())
        messages.append(msg)
        calls = msg.get("tool_calls") or []
        trace.append({
            "step": step,
            "assistant": msg.get("content"),
            "calls": [c["function"]["name"] for c in calls],
        })
        if not calls:
            return {"status": "done", "answer": msg.get("content", ""),
                    "steps": step, "trace": trace, "stats": runner.stats}

        for call in calls:
            fn = call["function"]
            try:
                args = json.loads(fn.get("arguments") or "{}")
                if not isinstance(args, dict):
                    raise ValueError("arguments 必须是 JSON 对象")
            except (json.JSONDecodeError, ValueError) as exc:
                result = {"ok": False, "error": {
                    "code": "BAD_JSON",
                    "message": str(exc),
                    "hint": "arguments 需要是合法 JSON 对象",
                }}
            else:
                sig = fn["name"] + json.dumps(args, sort_keys=True, ensure_ascii=False)
                seen[sig] = seen.get(sig, 0) + 1
                if seen[sig] > 2:
                    return {"status": "escalate",
                            "reason": "同一工具同一参数被重复调用 3 次，转人工",
                            "steps": step, "trace": trace, "stats": runner.stats}
                result = runner.call(fn["name"], args)

            messages.append({
                "role": "tool",
                "tool_call_id": call["id"],
                "content": json.dumps(result, ensure_ascii=False),
            })
            trace.append({"step": step, "tool": fn["name"], "result": result})

    return {"status": "budget_exhausted", "steps": max_steps,
            "trace": trace, "stats": runner.stats}


# ---------------------------------------------------------------- 离线演示
ORDERS = {
    "A1001": {"status": "paid", "amount": 199.0, "shipped": False},
    "A1002": {"status": "shipped", "amount": 89.0, "shipped": True},
}


def get_order(order_id: str) -> dict:
    if order_id not in ORDERS:
        raise LookupError(f"订单 {order_id} 不存在")
    return {"order_id": order_id, **ORDERS[order_id]}


REFUNDS = {}


def create_refund(order_id: str, amount: float, reason: str) -> dict:
    REFUNDS[order_id] = amount
    return {"refund_id": "R" + order_id, "order_id": order_id, "amount": amount, "reason": reason}


TOOLS = [
    Tool(
        name="get_order",
        description="按订单号查询订单状态、金额与是否已发货",
        schema={
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "订单号，例如 A1001"},
            },
            "required": ["order_id"],
        },
        fn=get_order,
    ),
    Tool(
        name="create_refund",
        description="对未发货订单发起退款。幂等：同参数重复调用不会重复退款",
        schema={
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "订单号"},
                "amount": {"type": "number", "description": "退款金额，不得超过订单金额"},
                "reason": {
                    "type": "string",
                    "description": "退款原因",
                    "enum": ["user_cancel", "out_of_stock", "price_error"],
                },
            },
            "required": ["order_id", "amount", "reason"],
        },
        fn=create_refund,
        idempotent=True,
    ),
]


class FakeModel:
    """按剧本走：第一次用错字段名，第二次按错误提示修正。"""

    def __init__(self):
        self.turn = 0

    def __call__(self, messages, tools):
        self.turn += 1
        if self.turn == 1:
            return {"role": "assistant", "content": None, "tool_calls": [{
                "id": "call_1", "type": "function",
                "function": {"name": "get_order", "arguments": '{"orderId": "A1001"}'},
            }]}
        if self.turn == 2:
            return {"role": "assistant", "content": None, "tool_calls": [{
                "id": "call_2", "type": "function",
                "function": {"name": "get_order", "arguments": '{"order_id": "A1001"}'},
            }]}
        return {"role": "assistant", "content": "订单 A1001 未发货，可为您原路退款 199 元（演示中未真正提交）。"}


if __name__ == "__main__":
    result = run_agent(FakeModel(), ToolRunner(TOOLS), "订单 A1001 我不想要了，帮我退款")
    print(json.dumps(result, ensure_ascii=False, indent=2))

跑一下，会看到模型第一次把 order_id 写成了 orderId，被校验拦下；拿到结构化错误后自行修正；第三步给出结论。

{
  "status": "done",
  "answer": "订单 A1001 未发货，可为您原路退款 199 元（演示中未真正提交）。",
  "steps": 3,
  "trace": [
    {"step": 1, "tool": "get_order", "result": {"ok": false, "error": {"code": "INVALID_ARGS", "problems": [
      {"field": "order_id", "reason": "missing", "hint": "订单号，例如 A1001"},
      {"field": "orderId", "reason": "unknown_field", "hint": "可用字段: ['order_id']"}
    ]}}},
    {"step": 2, "tool": "get_order", "result": {"ok": true, "data": {"order_id": "A1001", "status": "paid", "amount": 199.0, "shipped": false}}},
    {"step": 3, "assistant": "订单 A1001 未发货，可为您原路退款 199 元（演示中未真正提交）。"}
  ],
  "stats": {"calls": 2, "invalid": 1, "errors": 0, "cache_hits": 0}
}

接真实模型只需要替换一个函数，其余逻辑完全不动：

from openai import OpenAI

client = OpenAI(base_url="https://<控制台给出的兼容地址>/v1", api_key="<你的 API Key>")

def OpenAIChatModel(messages, tools):
    resp = client.chat.completions.create(
        model="<模型名>",
        messages=messages,
        tools=tools,
        temperature=0,
    )
    return resp.choices[0].message.model_dump(exclude_none=True)

注意 temperature=0。工具调用场景不需要创造力，参数生成越确定越好；需要发散的是最后那段自然语言，那部分可以单独用一次高温调用生成。

六、怎么证明它真的变好了

"感觉稳多了"不能作为上线依据。四个指标，一张表就够：

方法很朴素：准备 20～50 条用例，覆盖三类场景——

• 正常流：标准请求，验证主路径；
• 边界流：不存在的订单号、金额为 0、超长备注；
• 对抗流：故意让它选错工具、给错字段名、要求它做权限外的事。

每次改 Prompt、换模型、加工具，先把这套用例跑一遍再发布。用例不用多，20 条能覆盖主路径就够了；关键是要固定下来反复跑，而不是每次上线前手动点几下。

def evaluate(make_model, cases):
    agg = {"pass": 0, "calls": 0, "invalid": 0, "steps": 0, "escalated": 0}
    for case in cases:
        runner = ToolRunner(TOOLS)
        out = run_agent(make_model(), runner, case["task"])
        agg["calls"] += runner.stats["calls"]
        agg["invalid"] += runner.stats["invalid"]
        agg["steps"] += out["steps"]
        agg["escalated"] += out["status"] != "done"
        agg["pass"] += out["status"] == "done" and case["check"](out)
    n = len(cases)
    return {
        "任务成功率": agg["pass"] / n,
        "无效调用率": agg["invalid"] / max(agg["calls"], 1),
        "平均步数": agg["steps"] / n,
        "人工接管率": agg["escalated"] / n,
    }

顺带一个实践建议：把每次评测的 trace 都存下来。模型升级后如果成功率下降，回放 trace 能直接定位到是哪一步的入参开始漂的，比看汇总数字快得多。

部署时这几块放在哪

• 模型侧接 MaaS（ModelArts Studio）。是否使用 OpenAI 兼容协议、endpoint 和模型名，以控制台文档为准；业务代码只依赖一个 model 函数，换模型不用改编排。
• Agent 服务本身跑在 CCE 上，工具调用建议拆成独立服务，方便单独限流、扩缩容和灰度。
• trace 统一写 LTS，用同一个 trace_id 串起模型轮次和工具调用，出问题能直接回放。
• 成功率、无效调用率、人工接管率上报 CES 做告警，比事后翻日志快得多。

七、上线前 checklist

打印出来，逐条打勾：

• [ ] 每个工具都有完整的 JSON Schema，包含 required、enum 和带示例的 description
• [ ] 全部工具命名风格统一，没有 orderId / order_id 混用
• [ ] 所有错误都是结构化 JSON，含 code / message / hint / retryable
• [ ] 错误信息里没有堆栈、SQL、内网地址和密钥
• [ ] 每个写工具都有幂等键，且存储层有唯一约束兜底
• [ ] 高金额、不可逆操作有人工确认闸门
• [ ] 步数预算、重复调用检测、工具超时三件套齐全
• [ ] 每一步的入参、返回、耗时都进了 trace，可回放
• [ ] 有 20 条以上固定回归用例，换模型先跑再发
• [ ] 监控上能看到成功率、无效调用率、人工接管率三条曲线

写在最后

Agent 的可靠性，很少是"换了个更聪明的模型"换来的，更多是"把接口做得更笨、更确定"换来的。

模型本身是概率的，这一点改不了。但你完全可以决定：它拿到的是模糊的自然语言，还是带枚举和示例的 Schema；它收到的是 500，还是"order_id 缺失，可用字段是 [‘order_id’]"；它的重试会重复扣款，还是命中幂等键直接返回上次结果。

把不确定性收在模型这一层，外面的世界保持确定——这是我目前看到的、最靠谱的工程化路径。

代码已放在文中，直接复制即可运行。如果你正在把 Agent 往生产推，欢迎在评论区聊聊你踩过的坑。