AI 智能体开发的工程核心：工具契约、规划循环与可靠性

import json
from dataclasses import dataclass
from typing import Callable, Any
from pydantic import BaseModel, ValidationError

class QueryOrder(BaseModel):          # 参数契约
    order_id: str
    fields: list[str] = ["status", "amount"]

@dataclass
class Tool:
    name: str
    desc: str
    schema: type[BaseModel]
    fn: Callable[..., Any]
    idempotent: bool = True           # 是否可安全重试

REGISTRY: dict[str, Tool] = {}

def tool(schema: type[BaseModel], desc: str, idempotent=True):
    def deco(fn):
        REGISTRY[fn.__name__] = Tool(fn.__name__, desc, schema, fn, idempotent)
        return fn
    return deco

@tool(QueryOrder, "按订单号查询订单状态与金额")
def query_order(order_id: str, fields: list[str]) -> dict:
    return {"order_id": order_id, **{f: "PAID" for f in fields}}

def dispatch(name: str, raw_args: str) -> tuple[bool, Any, str]:
    """返回 (成功, 结果, 错误信息)。错误信息回灌给模型用于自纠。"""
    t = REGISTRY.get(name)
    if not t:
        return False, None, f"unknown tool {name}"
    try:
        args = t.schema.model_validate_json(raw_args)   # 严格校验
    except ValidationError as e:
        # 关键：把校验错误原文返回，模型据此修正参数
        return False, None, f"invalid args: {e.errors()[:2]}"
    try:
        return True, t.fn(**args.model_dump()), ""
    except Exception as e:
        return False, None, f"tool error: {type(e).__name__}"

MAX_STEPS, MAX_TOKENS = 8, 20000

def run_agent(llm, user_input: str, tools_spec: str) -> str:
    history = [{"role": "user", "content": user_input}]
    used_tokens = 0
    for step in range(MAX_STEPS):
        resp = llm.chat(history, tools=tools_spec)
        used_tokens += resp.usage
        if used_tokens > MAX_TOKENS:                 # 成本刹车
            return "任务超预算，已中止，请缩小范围"
        if resp.tool_call is None:                   # 无工具调用 = 收敛
            return resp.content
        ok, result, err = dispatch(resp.tool_call.name, resp.tool_call.args)
        history.append({"role": "assistant", "tool_call": resp.tool_call})
        history.append({"role": "tool",
                        "content": json.dumps(result if ok else {"error": err},
                                              ensure_ascii=False)})
        if not ok and step >= MAX_STEPS - 2:         # 临近预算仍失败，降级
            return "无法完成，请人工处理"
    return "达到最大步数，未收敛"

import uuid

def execute_safe(t: Tool, args: dict, attempt: int = 2):
    for i in range(attempt if t.idempotent else 1):
        try:
            return t.fn(**args)
        except TimeoutError:
            if i == attempt - 1:
                raise

def test_refund_flow(agent, mock_llm):
    mock_llm.stub(tool_seq=["query_order", "refund"], args_valid=True)
    out = agent.run("帮我把 A1001 退款")
    assert [c.name for c in mock_llm.calls] == ["query_order", "refund"]
    assert "成功" in out