《AI Agent 高级编程》11.2 真实 API 冒烟：假模型 vs 真模型量化差距

# gap_harness.py
import argparse
import os
import sys
import time

from pydantic_ai import Agent
from pydantic_ai.models.function import FunctionModel, AgentInfo
from pydantic_ai.messages import ModelResponse, TextPart, ToolCallPart

EVAL = [
    {"id": "q1", "query": "smolagents", "tool": "search", "expect": "CodeAgent"},
    {"id": "q2", "query": "langgraph", "tool": "search", "expect": "StateGraph"},
    {"id": "q3", "query": "pydanticai", "tool": "search", "expect": "FunctionModel"},
    {"id": "q4", "query": "a2a protocol", "tool": "search", "expect": "Agent Card"},
    {"id": "q5", "query": "durable execution", "tool": "search", "expect": "重放"},
]

DOCS = {
    "smolagents": "smolagents 用 CodeAgent 把动作写成 Python 代码执行。",
    "langgraph": "LangGraph 用 StateGraph 与检查点做有状态编排。",
    "pydanticai": "PydanticAI 提供 FunctionModel / TestModel 做确定性测试。",
    "a2a protocol": "A2A 用 Agent Card 描述能力，用 Task 承载协作。",
    "durable execution": "Durable Execution 通过确定性重放恢复长任务。",
}

RESULTS = {"tool_args": []}


def search(query: str) -> str:
    """检索本地知识库。"""
    RESULTS["tool_args"].append(query)
    return DOCS.get(query, "[miss]")


def scripted(task):
    def fn(messages, info: AgentInfo) -> ModelResponse:
        returned = sum(
            1 for m in messages for p in getattr(m, "parts", [])
            if type(p).__name__ == "ToolReturnPart"
        )
        if returned == 0:
            return ModelResponse(parts=[ToolCallPart("search", {"query": task["query"]})])
        return ModelResponse(parts=[TextPart(f"{task['expect']}：{DOCS.get(task['query'], '')}")])
    return FunctionModel(fn)


def score_fake() -> list[dict]:
    rows = []
    for task in EVAL:
        RESULTS["tool_args"].clear()
        agent = Agent(scripted(task), instructions="先检索再给结论")
        agent.tool_plain(search)
        t0 = time.perf_counter()
        r = agent.run_sync(f"查 {task['query']} 并给结论")
        dt = (time.perf_counter() - t0) * 1000
        rows.append({
            "id": task["id"],
            "run_ok": True,
            "tool_sel": task["query"] in RESULTS["tool_args"],
            "answer_hit": task["expect"] in r.output,
            "steps": r.usage.requests,
            "tokens": r.usage.total_tokens,
            "ms": round(dt, 1),
        })
    return rows


def print_table(rows: list[dict]) -> None:
    hdr = f"{'id':4} {'run_ok':6} {'tool_sel':8} {'answer_hit':10} {'steps':5} {'tokens':6} {'ms':>8}"
    print(hdr)
    print("-" * len(hdr))
    for r in rows:
        print(f"{r['id']:4} {str(r['run_ok']):6} {str(r['tool_sel']):8} "
              f"{str(r['answer_hit']):10} {r['steps']:<5} {r['tokens']:<6} {r['ms']:>8}")
    n = len(rows)
    print(f"\n聚合: tool_sel={sum(r['tool_sel'] for r in rows) / n:.0%}  "
          f"answer_hit={sum(r['answer_hit'] for r in rows) / n:.0%}  n={n}")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--real", action="store_true", help="用真模型跑（需用户授权，会消耗额度）")
    args = ap.parse_args()

    if args.real:
        if os.environ.get("AGENT_ALLOW_REAL") != "1":
            print("[拒绝] --real 需要用户明确授权：请先设置 AGENT_ALLOW_REAL=1。")
            return 2
        if not os.environ.get("ANTHROPIC_API_KEY"):
            print("[拒绝] 缺少 ANTHROPIC_API_KEY。")
            return 2
        print("[已授权] 真模型列将用同一评测集运行……（本节未执行此分支）")
        return 0

    print("== 假模型列（本次真跑）==")
    print_table(score_fake())
    print("\n真模型列：未运行（需 --real 且 AGENT_ALLOW_REAL=1 授权）")
    return 0


if __name__ == "__main__":
    sys.exit(main())

== 假模型列（本次真跑）==
id   run_ok tool_sel answer_hit steps tokens       ms
-----------------------------------------------------
q1   True   True     True       2     128        23.9
q2   True   True     True       2     124         3.5
q3   True   True     True       2     128         2.9
q4   True   True     True       2     135         2.9
q5   True   True     True       2     126         3.3

聚合: tool_sel=100%  answer_hit=100%  n=5

真模型列：未运行（需 --real 且 AGENT_ALLOW_REAL=1 授权）

# 仅示意：需要活的 API 会话才能运行，本节未执行
from pydantic_ai.models.anthropic import AnthropicModel

def score_real(model_name: str) -> list[dict]:   # model_name 用你账户可用的模型 id
    model = AnthropicModel(model_name)           # 需要 ANTHROPIC_API_KEY
    rows = []
    for task in EVAL:
        RESULTS["tool_args"].clear()
        agent = Agent(model, instructions="先检索再给结论")
        agent.tool_plain(search)
        t0 = time.perf_counter()
        r = agent.run_sync(f"查 {task['query']} 并给结论")   # <- 真实网络调用
        rows.append({"id": task["id"], "tool_sel": task["query"] in RESULTS["tool_args"],
                     "answer_hit": task["expect"] in r.output,
                     "steps": r.usage.requests, "tokens": r.usage.total_tokens,
                     "ms": round((time.perf_counter() - t0) * 1000, 1)})
    return rows