附录 B · 代码与检查单

pip install anthropic          # 官方 SDK
pip install requests           # 第 17 节项目一
pip install pytest             # 第 17 节项目二
export ANTHROPIC_API_KEY=...   # Windows: setx ANTHROPIC_API_KEY "..."

import json

def run_agent(model, tools, task, system, max_steps=10):
    messages = [{"role": "system", "content": system},
                {"role": "user", "content": task}]
    for step in range(max_steps):
        reply = model.chat(messages)
        messages.append({"role": "assistant", "content": reply})
        try:
            call = json.loads(reply); assert "tool" in call
        except (json.JSONDecodeError, AssertionError):
            return reply                          # 最终答案
        fn = tools.get(call["tool"])
        result = fn(**call["args"]) if fn else f"错误：无 {call['tool']}"
        messages.append({"role": "user", "content": f"工具结果：\n{result}"})
    return "达到最大步数"                          # 护栏

import anthropic
client = anthropic.Anthropic()

TOOLS = [{
    "name": "search_logs",
    "description": "在服务日志中搜索匹配的行。返回匹配行及前后各2行上下文。",
    "input_schema": {
        "type": "object",
        "properties": {
            "pattern": {"type": "string", "description": "正则或关键词"},
            "max_lines": {"type": "integer", "description": "最多返回行数，默认50"},
        },
        "required": ["pattern"],
    },
}]

def agent_turn(messages):
    resp = client.messages.create(
        model="claude-sonnet-5", max_tokens=2048,
        tools=TOOLS, messages=messages)
    if resp.stop_reason == "tool_use":
        block = next(b for b in resp.content if b.type == "tool_use")
        result = dispatch(block.name, block.input)
        messages.append({"role": "assistant", "content": resp.content})
        messages.append({"role": "user", "content": [{
            "type": "tool_result", "tool_use_id": block.id, "content": result}]})
        return None                                # 继续循环
    return resp.content[0].text                    # 最终答案

import inspect

def tool(fn):
    sig = inspect.signature(fn)
    fn._schema = {
        "name": fn.__name__,
        "description": (fn.__doc__ or "").strip(),
        "input_schema": {
            "type": "object",
            "properties": {n: {"type": "string"} for n in sig.parameters},
            "required": [n for n, p in sig.parameters.items()
                         if p.default is inspect.Parameter.empty],
        },
    }
    return fn

@tool
def search_logs(pattern: str, max_lines: str = "50"):
    """在服务日志中搜索匹配的行，返回匹配行及上下文。"""
    ...

COMPACT_PROMPT = """总结以下 Agent 工作历史。必须显式保留：
1. 已完成的事
2. 当前状态
3. 下一步计划
4. 所有关键事实：文件路径、做过的决策、报错原文、数字

不要省略任何路径或报错——省略会导致重复犯错。

历史：
{history}"""

def compact(model, messages):
    history = "\n".join(f"{m['role']}: {m['content']}" for m in messages[1:])
    summary = model.chat([{"role": "user",
                           "content": COMPACT_PROMPT.format(history=history)}])
    return [messages[0],                                  # 保留 system
            {"role": "user", "content": f"【历史摘要】\n{summary}\n\n请继续。"}]

def est_tokens(messages):
    return sum(len(m["content"]) for m in messages) // 3   # 粗估，够用

def spawn_subagent(model, task, tools, system, depth=0, max_depth=2):
    if depth >= max_depth:
        return "已达子 Agent 深度上限"                     # 防无限递归
    return run_agent(model, tools, task, system, max_steps=15)
    # 只有返回值进入主上下文，子 Agent 的完整历史被丢弃

def calculator(expression: str) -> str:
    """计算数学表达式。只支持数字和 + - * / ** ( )"""
    ALLOWED = set("0123456789+-*/(). ")
    bad = set(expression) - ALLOWED
    if bad:
        return (f"错误：不支持字符 {sorted(bad)}。"
                f"只支持数字和 + - * / ** ( )。"
                f"示例：calculator(expression='3.14 * 2**2')")
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"计算失败：{e}。请检查括号是否配对。"

class Guard:
    """确定性规则。模型碰不到这里——这才叫硬边界"""
    def __init__(self):
        self.whitelist = {"user_alice": {"user_bob"}}   # 运行时不可变更

    def check_transfer(self, actor, to_u, amt):
        if to_u not in self.whitelist.get(actor, set()):
            return False, f"拒绝：{to_u} 不在白名单收款人中"
        if amt > 1000:
            return False, "需要用户二次确认"
        return True, "ok"

def transfer(actor, to_u, amt, guard, confirmed=False):
    ok, reason = guard.check_transfer(actor, to_u, amt)
    if not ok and not confirmed:
        return f"转账被拒：{reason}"      # 模型说破天也没用
    ...

from collections import defaultdict

def evaluate(agent, tasks, code_scorer, runs=1):
    results = []
    for t in tasks:
        for _ in range(runs):
            out = agent.run(t["input"])
            score, note = code_scorer(out, t["reference"])
            results.append({"id": t["id"], "type": t["type"],
                            "score": score, "note": note})
    return results

def summarize(results):
    """⭐ 必须分维度看，总分会骗人"""
    by = defaultdict(list)
    for r in results:
        by[r["type"]].append(r["score"])
    return {k: round(sum(v)/len(v), 3) for k, v in by.items()}

def pass_hat_k(results_per_task, k):
    """k 次全部成功的比例。面向用户的 Agent 该看这个"""
    return sum(all(s == 1.0 for s in scores[:k])
               for scores in results_per_task.values()) / len(results_per_task)

RUBRIC = """按以下维度给输出打分，每项 0-25 分：
- 准确性：结论与事实/参考解一致吗
- 完整性：要求的要素都覆盖了吗
- 边界处理：异常/模糊输入处理得当吗
- 恰当拒绝：该说"不知道"时说了吗（不该拒绝的别拒绝）

输出 JSON：{"准确":分,"完整":分,"边界":分,"拒绝":分,"理由":"一句话"}

参考解：{reference}
实际输出：{output}"""
# ⭐ 用前先校准：让它给参考解打分，应该接近满分，否则评委坏了

【架构】
☐ 做过降级测试，有数据证明当前复杂度必要（第6节）
☐ 能说清为什么用工作流/Agent、为什么用这种多Agent模式
☐ 多Agent的15倍token溢价对得起它的价值

【上下文】
☐ 系统提示词砍到最小，删掉了为旧模型写的补丁
☐ 长任务有压缩机制，且摘要保留关键事实
☐ 工具超过20个上了Tool Search
☐ 大量阅读类任务派给了子Agent

【工具】
☐ 工具按意图分组，不是API的一比一镜像
☐ 返回里没有模型用不上的字段
☐ 错误信息是可行动的（告诉模型怎么改）
☐ 危险操作做了防呆

【验证】
☐ Agent有能自己跑的验证（测试/编译/lint）
☐ 验证者不是干活的那个Agent
☐ 主观标准转成了可回答的rubric条目

【评测】
☐ 至少20个来自真实失败的任务，每个有参考解
☐ 评分器评产出不评路径
☐ 读过转录，人工核对过评分器
☐ 明确该看pass@k还是pass^k
☐ 对小于3个百分点的差距保持怀疑

【安全】
☐ 做过威胁建模：Agent被完全控制最坏会怎样
☐ 至少一道环境层硬边界（代码里的if，不是提示词）
☐ 高危动作权限分离（能写≠能部署）
☐ 只在高危动作要确认，避免审批疲劳
☐ 外部内容当不可信数据处理

【运维】
☐ 有max_steps/超时/预算三道护栏，且优雅退出
☐ 状态可从出错点恢复，不用从头重跑
☐ 有结构化trace能回放决策路径
☐ 部署不打断运行中的Agent（彩虹部署）

【长期】
☐ 日历上有"模型升级后重审harness"的提醒
☐ 定期问：我可以停止做什么？