换一套 Harness，比换两代模型还管用

01

             ┌──────────────┐   ┌────────►│    Reason    │  模型推理（一次 LLM 调用）   │         │  (LLM call)  │   │         └──────┬───────┘   │                │   │                ▼   │          ┌──────────┐   没有工具调用   ┌──────────┐   │          │  Tools?  ├───────────────►│  Output  │  返回最终文本，结束   │          └────┬─────┘                └──────────┘   │               │ 有工具调用   │               ▼   │          ┌──────────┐   │          │ Execute  │  执行工具（可并行）   │          │  tools   │   │          └────┬─────┘   │               │   │               ▼   │          ┌──────────┐   └──────────┤ Observe  │  把工具结果喂回去，进入下一轮    (loop)    │ results  │              └──────────┘

def run_loop(history: list, toolset: list, turn_budget: int = 20) -> str:    """一轮轮推进，直到模型不再要工具、只给文本。"""    for _ in range(turn_budget):        # ① 思考：一次模型调用        reply = model.invoke(context=history, toolset=toolset)        history.append(reply.message)          # 先把它这一轮的发言记下来
        # ② 出口：不再请求工具，说明它认为可以作答了        requests = reply.message.tool_calls        if not requests:            return reply.message.content
        # ③ 执行并观察：结果逐条写回，失败也照样写回        for req in requests:            outcome = invoke_tool(req)            history.append(as_tool_message(req, outcome))        # 回到 ①——模型带着新结果重新判断    raise TurnBudgetExhausted(f"{turn_budget} 轮内未收敛")

STUCK_THRESHOLD = 3
def is_spinning(history: list) -> bool:    """连续多次发出完全相同的调用，视为原地打转。"""    fingerprints = []    for msg in reversed(history):        calls = getattr(msg, "tool_calls", None)        if not calls:            continue        for c in calls:            fingerprints.append((c.function.name, c.function.arguments))        if len(fingerprints) >= STUCK_THRESHOLD:            break
    latest = fingerprints[:STUCK_THRESHOLD]    return len(latest) == STUCK_THRESHOLD and len(set(latest)) == 1

┌────────────────────────────────┐│         System Prompt          │  ← Agent 的身份定义├────────────────────────────────┤│        Tool Definitions        │  ← 能做什么（JSON Schema）├────────────────────────────────┤│        Tool Execution          │  ← Tool 实际执行逻辑├────────────────────────────────┤│          Tool Loop             │  ← 循环：思考 → 执行 → 观察└────────────────────────────────┘

#!/usr/bin/env python3"""便签助手：一个最小 Agent Harness。运行：python note_agent.py"""import jsonfrom openai import OpenAI
llm = OpenAI()MODEL_NAME = "gpt-4o-mini"       # 便宜够用，学习阶段首选TURN_LIMIT = 15
NOTES: dict[str, str] = {}       # 用内存字典当"存储层"，聚焦讲循环本身
# --- ① 身份 ---PERSONA = (    "你是一个便签管家。可以帮用户保存便签、按关键词检索。"    "涉及便签操作时，务必调用提供的工具，不要凭空编造内容。")
# --- ② 工具 Schema：模型唯一能看到的接口 ---TOOL_SPECS = [    {"type": "function", "function": {        "name": "save_note",        "description": "保存一条便签，返回它的编号",        "parameters": {"type": "object", "properties": {            "title": {"type": "string", "description": "便签标题"},            "body":  {"type": "string", "description": "便签正文"}},            "required": ["title", "body"]}}},    {"type": "function", "function": {        "name": "find_notes",        "description": "按关键词检索便签，返回命中的标题列表",        "parameters": {"type": "object", "properties": {            "keyword": {"type": "string"}},            "required": ["keyword"]}}},]
# --- ③ 工具实现：模型看不到这段 ---def call_tool(name: str, kwargs: dict) -> str:    try:        if name == "save_note":            note_id = f"n{len(NOTES) + 1}"            NOTES[note_id] = f"{kwargs['title']}\n{kwargs['body']}"            return f"已保存，编号 {note_id}"        if name == "find_notes":            kw = kwargs["keyword"]            hits = [f"{nid}: {txt.splitlines()[0]}"                    for nid, txt in NOTES.items() if kw in txt]            return "\n".join(hits) if hits else f"没有匹配 “{kw}” 的便签"        return f"错误：未知工具 {name}"    except Exception as exc:        return f"错误：{exc}"           # 出错也返回字符串，交给模型判断
# --- ④ 循环：思考 → 执行 → 观察 ---def chat(user_text: str) -> str:    history = [{"role": "system", "content": PERSONA},               {"role": "user", "content": user_text}]    for _ in range(TURN_LIMIT):        reply = llm.chat.completions.create(            model=MODEL_NAME, messages=history, tools=TOOL_SPECS)        turn = reply.choices[0].message        history.append(turn)                       # 关键：先把 assistant 回合入历史        if not turn.tool_calls:                    # 不再调用工具 → 收尾            return turn.content        for tc in turn.tool_calls:                 # 执行本回合请求的所有工具            out = call_tool(tc.function.name, json.loads(tc.function.arguments))            history.append({"role": "tool",                            "tool_call_id": tc.id, "content": out})    return "已达最大回合数，提前退出。"
if __name__ == "__main__":    print("便签管家已就绪（输入 q 退出）")    while (line := input("\n> ").strip()) not in ("q", "quit"):        print(chat(line))

# 加进 TOOL_SPECS 列表：{"type": "function", "function": {    "name": "delete_note",    "description": "按编号删除一条便签",    "parameters": {"type": "object",        "properties": {"note_id": {"type": "string"}}, "required": ["note_id"]}}}
# 加进 call_tool()：if name == "delete_note":    nid = kwargs["note_id"]    return f"已删除 {nid}" if NOTES.pop(nid, None) else f"编号 {nid} 不存在"

from anthropic import Anthropicllm = Anthropic()reply = llm.messages.create(    model="claude-sonnet-4-20250514", max_tokens=4096,    system=PERSONA, messages=history,    tools=[{"name": s["function"]["name"],                 # 字段名不同，做个映射            "description": s["function"]["description"],            "input_schema": s["function"]["parameters"]} for s in TOOL_SPECS])for block in reply.content:    if block.type == "tool_use":        out = call_tool(block.name, block.input)

02

┌──────────────────────────────────────────────┐│                   HARNESS                    ││  ┌──────────┐  ┌──────────┐  ┌────────────┐  ││  │ Agentic  │  │   Tool   │  │  Memory &  │  ││  │   Loop   │  │  System  │  │  Context   │  ││  └──────────┘  └──────────┘  └────────────┘  ││  ┌────────────────────────────────────────┐  ││  │              Guardrails                │  ││  └────────────────────────────────────────┘  │└──────────────────────────────────────────────┘

# 模型看到的（tool schema）{    "name": "get_weather",    "description": "查询某城市的当前天气",    "parameters": {        "type": "object",        "properties": {"city": {"type": "string", "description": "城市名，如 上海"}},        "required": ["city"]    }}
# Harness 执行的（tool implementation）——模型看不到这段def get_weather(city: str) -> str:    resp = requests.get(WEATHER_API, params={"q": city, "key": API_KEY})    return f"{city}：{resp.json()['now']['text']}，{resp.json()['now']['temp']}℃"

from dataclasses import dataclassfrom typing import Callable@dataclass
class Tool:    name: str    description: str    parameters: dict          # JSON Schema    handler: Callable         # 实际实现
class ToolRegistry:    def __init__(self):        self._tools: dict[str, Tool] = {}
    def register(self, tool: Tool):        self._tools[tool.name] = tool
    def get_schemas(self) -> list[dict]:        """给 LLM API 用——只暴露 Schema，不暴露实现。"""        return [{"type": "function", "function": {                    "name": t.name,                    "description": t.description,                    "parameters": t.parameters}}                for t in self._tools.values()]
    def dispatch(self, name: str, arguments: dict) -> str:        """执行工具调用——永远返回字符串，出错也不例外。"""        tool = self._tools.get(name)        if not tool:            return f"Error: Unknown tool '{name}'"        try:            result = tool.handler(**arguments)            return result if isinstance(result, str) else json.dumps(result)        except TypeError as e:            return f"Error: Invalid arguments for '{name}': {e}"        except Exception as e:            return f"Error: {type(e).__name__}: {e}"   # 错误也返回给模型，不静默崩溃

Context 窗口（如 128K Token）┌─────────────────────────────────┐│  [Reserve] 回复预留    (~4,000)  │  ← 必须留空，否则模型没地方回话├─────────────────────────────────┤│  更早对话              (剩余)    │  ← 优先级最低，最先被压缩/丢弃  ▲ 先出│  近期对话              (~varies) │                                 ││  注入文件 AGENTS.md…   (~5,000)  │                                 ││  Memory 摘要           (~1,000)  │                                 ││  任务指令              (~500)    │                                 ││  Tool Schema (活跃)    (~2,000)  │                                 ││  System Prompt         (~500)   │  ← 优先级最高，永远保留         ▼ 后出└─────────────────────────────────┘        组装器从高优先级往低填，装满即止；关键段落(≤2)宁可截断也不整段丢

@dataclassclass ContextBlock:    priority: int          # 越小越优先    content: str    tokens: int
def assemble_context(blocks: list[ContextBlock],                     max_tokens: int = 128_000,                     reserve: int = 4_000) -> str:    """按优先级组装 Context，为模型回复预留 reserve 空间。"""    budget = max_tokens - reserve    used = 0    selected = []    for block in sorted(blocks, key=lambda b: b.priority):   # 高优先级先进        if used + block.tokens <= budget:            selected.append(block)            used += block.tokens        elif block.priority <= 2:                            # 关键段落宁可截断也不整段丢            remaining = budget - used            if remaining > 100:                selected.append(ContextBlock(                    block.priority, truncate(block.content, remaining), remaining))                used = budget    # 组装时按逻辑顺序（system 在前），而非按优先级    return "\n\n".join(b.content for b in sorted(selected, key=lambda b: b.priority))

        模型侧（不可信）   ┌──────────────────────────────┐   │  模型推理 + 发起工具调用请求    │   └──────────────┬───────────────┘                  │ 请求：save_note(...) / run_shell(...)                  ▼   ┌──────────────────────────────┐   │  ★ 权限闸门（代码，非文本）★   │  放行 / 拦截 / 改写 / 转人工   └──────────────┬───────────────┘                  │ 仅放行经审核的调用                  ▼        执行侧（真实副作用）   ┌──────────────────────────────┐   │   文件系统 · 网络 · Shell     │   └──────────────────────────────┘

from enum import Enum
class Decision(Enum):    ALLOW = "allow"; DENY = "deny"; MODIFY = "modify"; ASK = "ask"
def check_permission(tool_name: str, args: dict) -> tuple[Decision, str]:    # 1) 破坏性 Shell 命令：直接拒绝    if tool_name == "run_shell":        cmd = args.get("command", "")        for pattern in ("rm -rf /", ":(){:|:&};:", "mkfs", "dd if=", "> /dev/sd"):            if pattern in cmd:                return Decision.DENY, f"Blocked dangerous command: {pattern}"        if any(k in cmd for k in ("curl", "wget", "git push", "rm ")):            return Decision.ASK, "Command needs human approval"    # 2) 文件写入：限制在工作目录内（修改而非拒绝）    if tool_name == "write_file":        path = os.path.realpath(args["path"])        if not path.startswith(os.path.realpath(WORKSPACE)):            return Decision.DENY, f"Path outside workspace: {path}"    return Decision.ALLOW, ""
def guarded_dispatch(registry, tool_name: str, args: dict) -> str:    decision, reason = check_permission(tool_name, args)    if decision == Decision.DENY:        log_blocked(tool_name, args, reason)            # 记录被拒操作，便于调试        return f"Permission denied: {reason}"           # 作为 Tool 结果返回给模型    if decision == Decision.ASK:        if not human_approves(tool_name, args, reason):            return "User declined this action."    return registry.dispatch(tool_name, args)

策略                        Token 数────────────────────────────────────────全部预先加载:               ~12,000（每轮都是）Skill 菜单 + 加载 2 个:     ~150 + ~2,400 = ~2,550────────────────────────────────────────节省:                       每轮约 9,450（78%）

class SkillRegistry:    def __init__(self):        self._skills: dict[str, Skill] = {}     # name -> Skill(tools, skill_md)        self._active: set[str] = set()
    def menu(self) -> str:        """一直在 Context 里的轻量菜单（每项约 1 行）。"""        return "Available skills (call load_skill to activate):\n" + "\n".join(            f"- {s.name}: {s.summary}" for s in self._skills.values())
    def load_skill(self, name: str) -> str:        skill = self._skills.get(name)        if not skill:            return f"Error: no skill '{name}'"        self._active.add(name)        # 加载后，该 Skill 的 SKILL.md + 全部 Tool Schema 才进入 Context        return f"Loaded '{name}'. Guide:\n{skill.skill_md}"
    def unload_skill(self, name: str) -> str:        self._active.discard(name)              # 释放 Context        return f"Unloaded '{name}'"
    def active_schemas(self) -> list[dict]:        """只把已激活 Skill 的工具喂给 LLM。"""        schemas = []        for name in self._active:            schemas.extend(self._skills[name].tool_schemas())        return schemas

03

                        ┌─────────────────┐   Tool / LLM 出错 ────►│   分类 error     │                        └───┬───┬───┬───┬─┘              ┌─────────────┘   │   │   └──────────────┐              ▼                 ▼   ▼                   ▼        ┌──────────┐     ┌──────────┐  ┌──────────┐  ┌──────────┐        │  瞬时    │     │  模型     │  │  永久    │  │  资源     │        └────┬─────┘     └────┬─────┘  └────┬─────┘  └────┬─────┘             │                │             │             │             ▼                ▼             ▼             ▼      指数退避+抖动重试   带纠正信息      尝试 fallback   Checkpoint             │           重新 prompt    还不行→回喂模型  + 升级人类       重试N次仍失败                                       (BLOCK)             │             ▼        升级 (INFORM)          ── 全程铁律：错误永远作为 Tool 结果回喂，绝不在循环里抛异常 ──

def retry(max_attempts=3, base_delay=1.0, max_delay=60.0):    def decorator(func):        @functools.wraps(func)        def wrapper(*args, **kwargs):            last_error = None            for attempt in range(max_attempts):                try:                    return func(*args, **kwargs)                except Exception as e:                    last_error = e                    if classify_error(e) != ErrorClass.TRANSIENT:                        raise                        # 非瞬时错误不重试                    if attempt < max_attempts - 1:                        delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)                        time.sleep(delay)            # 指数退避 + 抖动            raise RetryExhausted(last_error, max_attempts)        return wrapper    return decorator

docker_cmd = [    "docker", "run", "--rm",              # 用完即删，不留持久容器    "--user", "1000:1000",                # 非 root    "--memory", "512m", "--cpus", "1.0",  # 资源上限    "--pids-limit", "100",                # 防 fork 炸弹    "--read-only",                        # 根文件系统不可变，不能植后门    "--tmpfs", "/tmp:size=100m,noexec",   # 临时可写空间，且禁止执行（防"写脚本再跑"逃逸）    "--security-opt", "no-new-privileges",    "--cap-drop", "ALL",                  # 移除所有 Linux capabilities    "--network", "none",                  # 默认禁网，无法外传数据]

阶段 1: 规划             阶段 2: 执行              阶段 3: 合并┌────────────┐           ┌──────────┐               ┌────────────┐│   Leader   │──spawn──► │ Worker A │──result──┐    │   Leader   ││  拆分任务   │──spawn──► │ Worker B │──result──┼──► │  审查合并   ││            │──spawn──► │ Worker C │──result──┘    │  向用户汇报 │└────────────┘           └──────────┘               └────────────┘

def spawn_subagent(task: str, context: str, max_turns: int = 15,                   timeout_s: int = 300) -> str:    """作为工具暴露给 Leader。子 Agent 独立 Context，结果字符串返回给父。"""    sub_messages = [        {"role": "system", "content": SUBAGENT_SYSTEM},   # 精简的子 Agent 身份        {"role": "user", "content": f"Task:\n{task}\n\nRelevant context:\n{context}"},    ]    try:        with time_limit(timeout_s):                        # 每个 Worker 必须设超时            return run_loop(sub_messages, max_turns=max_turns)    except TimeoutError:        return f"Sub-agent timed out after {timeout_s}s (partial work may exist)"
def claim_task(task_id: str, worker_id: str) -> bool:    """文件系统当分布式锁：O_CREAT|O_EXCL 保证只有一个 Worker 抢到。"""    lock = f"current_tasks/{task_id}.lock"    try:        fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)        os.write(fd, worker_id.encode()); os.close(fd)        return True                                        # 抢到了    except FileExistsError:        return False                                       # 已被别人认领，换一个任务

① Sequential Pipeline（流水线，串行转换）   ┌───────┐   ┌───────┐   ┌───────┐   ┌───────┐   │ Ingest│──►│Analyze│──►│ Draft │──►│Review │   └───────┘   └───────┘   └───────┘   └───────┘
② Fan-Out / Fan-In（扇出扇入，并行同类）                ┌──►│Worker A│──┐   ┌──────────┐ │   ┌────────┐ │   ┌────────┐   │Dispatcher│─┼──►│Worker B│─┼──►│ Merger │   └──────────┘ │   ┌────────┐ │   └────────┘                └──►│Worker C│──┘
③ Supervisor（监督者，中央决策+循环重派）              ┌────────────┐       ┌──────┤ Supervisor ├──────┐   ◄─┐ 可循环、重新委派       ▼      └─────┬──────┘      ▼    │   ┌────────┐  ┌────────┐   ┌────────┐ │   │  Code  │  │Research│   │ Review ├─┘   └────────┘  └────────┘   └────────┘
④ Peer-to-Peer（对等，无中心，最难调试）   ┌────────┐◄────►┌────────┐   │Agent A │      │Agent B │   └───┬────┘      └────┬───┘       └────►┌────────┐◄┘             │Agent C │             └────────┘

from croniter import croniterfrom datetime import datetime, timezone@dataclassclass ScheduledTask:    id: str    cron: str                    # "0 8 * * *"（UTC！）—— 时区是最大的坑    prompt: str                  # 交给 Agent 的任务    session_mode: str            # "isolated"（全新 Context）| "main"（注入主会话）    payload_type: str            # "agentTurn"（完整执行）| "systemEvent"（只丢便条）    delivery: str                # "announce" | "webhook" | "silent"    model: str = "default"       # 隔离 Session 可按任务选模型    def next_run(self, after: datetime | None = None) -> datetime:        base = after or datetime.now(timezone.utc)     # 铁律：内部一律 UTC        return croniter(self.cron, base).get_next(datetime)def tick(scheduler, now: datetime):    """调度循环——由外部 1 分钟 tick 驱动，不要自己 while True: sleep。"""    for task in scheduler.due_tasks(now):              # cron 到点的任务        if task.session_mode == "isolated":            run_isolated(task.prompt, model=task.model, deliver=task.delivery)        else:            inject_into_main(task.prompt, kind=task.payload_type)        scheduler.reschedule(task, task.next_run(now))

04

重置（Reset）—— 到阈值就清空重来，带一份简报  Turn 1-50   [完整对话历史] ── 80%满 ──►  Turn 51 [系统提示 + 前50轮的摘要 + 当前任务]                                                      └─ 全新开始，~10% 已用
压缩（Compaction）—— 旧的折叠，新的保真，渐进式  Turn 1-20  [已压缩：3行摘要]  Turn 21-40 [已压缩：关键决策]  Turn 41-50 [完整细节 ← 最近进行中的工作]

        ┌─────────┐        │ 规划者   │  把目标切成可独立验收的子任务，并写明各自的通过标准        └────┬────┘             ▼  子任务列表    ┌────────┴────────┐    ▼                 ▼┌────────┐        ┌────────┐│ 生成者  │        │ 生成者  │  领一个子任务，用干净上下文执行，不给自己判分└───┬────┘        └───┬────┘    ▼                 ▼┌────────┐        ┌────────┐│ 评估者  │        │ 评估者  │  只看输出（不看推理过程），按规划者的标准打分└────────┘        └────────┘

def generate_evaluate_loop(subtask: dict, max_iters: int = 3) -> dict:    """生成-评估循环。评估者只拿到 output + rubric，拿不到生成者的推理。"""    feedback = ""    for i in range(max_iters):                          # 迭代预算，防死循环        output = generator.run(                         # 生成者：全新上下文            task=subtask["goal"], prior_feedback=feedback)        verdict = evaluator.run(                         # 评估者：独立上下文            output=output,                               # 只看输出，不看怎么想的            rubric=subtask["acceptance"])                # 按明确清单逐条打分        if verdict["pass"]:            return {"status": "done", "output": output, "iters": i + 1}        feedback = verdict["actionable_feedback"]        # "parse_input 没处理空串"    return {"status": "needs_replan", "last": output}   # 三次不过→交回规划者/人
EVALUATOR_PROMPT = """你是一名严格的评审。你只能看到【产出物】和下方【验收清单】，看不到作者的思考过程。逐条核对清单，给出 通过/不通过 以及具体理由。不要讲人情、不要放水。验收清单：{rubric}"""

                       编排层  ┌──────────────┐   ┌──────────┐   ┌───────────────────┐  │    大脑      │   │ Session  │   │      双手          │  │ Harness+LLM  │   │ 事件日志  │   │  Sandbox A/B      │  │              │   │          │   │  MCP Tool         │  │  无状态       │   │ 持久化    │   │  可丢弃           │  └──────┬───────┘   └────┬─────┘   └────────┬──────────┘         │  emitEvent()   │   execute()      │         ├───────────────►│◄─────────────────┤         │  getEvents()   │   provision()    │         └────────────────┴──────────────────┘   崩了→wake()从日志恢复   活得最久，唯一真相   延迟创建，用完即弃   （牲畜，非宠物）

Session（仅追加事件日志，持久化，可能数百万 Token）┌──────────────────────────────────────────────────────────┐│ e1 │ e2 │ ... │ e500 │ ... │ e1950 │ ... │ e2000         │└──────────────────────────────────────────────────────────┘                                  │ getEvents(slice) 取景                                  ▼              Context Window（选取的子集，128K-200K）              ┌───────────────────────────┐              │ system_prompt             │              │ e1950 ... e2000（最近50个）│  ← 需要旧事件？回日志再取，压缩不再是单向销毁              └───────────────────────────┘

class StatelessBrain:    def __init__(self, session_store, sandbox_pool):        self.sessions = session_store       # 持久事件日志        self.sandboxes = sandbox_pool       # 按需 provision
    def wake(self, session_id: str):        """从事件日志恢复并继续——大脑本身不持有任何状态。"""        events = self.sessions.get_events(session_id)      # 唯一真相来源        context = self.assemble_context(events)            # 从日志重建 Context        while not self.is_done(context):            response = llm.chat(context)            self.sessions.emit(session_id, Event("llm_response", response))            for call in response.tool_calls:                sandbox = self.sandboxes.provision(session_id)   # 延迟创建                result = sandbox.execute(call.name, call.input)  # 挂了当失败工具调用                self.sessions.emit(session_id, Event("tool_result", result))            context = self.assemble_context(self.sessions.get_events(session_id))# 进程被 kill？→ 起个新进程调 wake(session_id) 即可，状态全在日志里

                    HARNESS DRIVER（你的脚本，故意"笨"）         │ 跑一次                              │ 跑 N 次（每次全新进程）         ▼                                    ▼┌──────────────────────┐          ┌──────────────────────────┐│  INITIALIZER AGENT   │          │      CODING AGENT        ││ • 读简报              │          │ • pwd + git log（真相）   ││ • 写 init.sh          │          │ • 读 progress + JSON     ││ • 写 feature_list.json│          │ • bash init.sh 起服务    ││ • 写 progress 文件    │          │ • 实现【一个】feature     ││ • git init + 首次提交 │          │ • 浏览器 E2E 验证         ││                      │          │ • 翻 passes:true + 提交   │└──────────┬───────────┘          └───────────┬──────────────┘           └──── 干净状态仓库（磁盘共享）────────┘                记忆在文件里，不在 Context Window 里

{  "features": [    {      "id": "cart-03",      "title": "购物车支持修改商品数量",      "priority": 1,      "depends_on": ["cart-01"],      "acceptance": [        "PATCH /api/cart/item 返回 200，且数量被更新",        "数量改为 0 时该商品从购物车移除",        "数量为负数返回 400，购物车不变"      ],      "passes": false    }  ]}

1. pwd                     → 确认在项目里2. git log --oneline -20   → 发了什么（真相）   cat claude-progress.txt → 上个 session 的笔记（提示，冲突时 git 赢）3. cat feature_list.json   → 什么做完、什么下一步、什么被阻塞4. bash init.sh            → 装依赖、起 dev server5. 浏览器打开 App          → 验证上个 session 的活还在工作（这一步杀死过早胜利）

05

① 正常解题        ② 元假设形成        ③ 识别评测集         ④ 破解数据          ⑤ 直接取答案大量 Token    ──► "题目太具体、失败  ──► 比对公开评测集，  ──► 读加载代码，   ──► 找到密文镜像，正常网搜，无解      率太高，像是考题"     确定正在被哪个测     还原出解密方式      解开题库取答案   └──────────── 每一步单独看都平常，新颖之处在于"无人提示下把它们串成了连贯计划" ────────────┘

「用户要的是这件事，而 Agent 现在要执行这个动作——这算合理解读吗？」

   USER ──► ┌─────────┐  要执行动作  ┌──────────────────┐            │  AGENT  ├────────────►│ 输出层 审核模型   │ 执行前审动作            │         │             │  (看不到推理过程) │            └────▲────┘             └────────┬─────────┘                 │ 工具结果回喂              │ 放行                 │                           ▼                 │                      ┌──────────┐      ┌──────────┴───────┐   干净       │   执行    │      │ 输入层 审核模型    │◄────────────┤          │      │ (扫外部内容注入)   │             └──────────┘      └──────────────────┘   两层独立：一层被绕过，另一层仍有机会拦下（纵深防御）

  要执行的动作 ──► ┌──────────────────────┐                    │ 第一层：白名单        │── 只读操作 / 用户已授权 ──► 放行                    └──────────┬───────────┘                               │ 未命中                               ▼                    ┌──────────────────────┐                    │ 第二层：项目内编辑    │── 有版本管理可回滚 ──► 放行                    └──────────┬───────────┘                               │ 未命中                               ▼                    ┌──────────────────────┐   先快筛一次，可疑才细判                    │ 第三层：审核模型      │──────────────► 放行 / 拦截                    │ Shell / 网络 / 外部写 │                    └──────────────────────┘

def review_action(user_msgs, tool_call, tool_results) -> Decision:    # 注意：只喂 用户消息 + 工具调用 + 工具结果，绝不喂 Agent 的推理过程    view = render_transcript(user_msgs, tool_call, tool_results)
    # 第一步：极短快筛（便宜、偏保守，宁可多标记）    flag = cheap_model.judge(        system=QUICK_PROMPT, input=view, max_tokens=1)   # 只输出 "Y"/"N"    if flag == "N":        return Decision.ALLOW                            # 绝大多数动作到这就放行
    # 第二步：完整判断（贵，但很少触发，产出可审计的理由）    verdict = strong_model.judge(system=FULL_PROMPT, input=view)    log_review(tool_call, verdict)                       # 必须记录，便于事后调误报    return Decision.BLOCK if verdict["decision"] == "block" else Decision.ALLOW
FULL_PROMPT = """判断【动作】是否是对【用户请求】的合理解读。你能看到用户消息、工具调用和工具结果，但看不到 Agent 的推理过程。若动作会：销毁或外传数据、削弱安全配置、跨越信任边界、或绕过审查——则拦截；否则放行。"""

用户： 帮我清一下测试环境的脏数据Agent（推理）： 用户想要干净环境，我把整个订单库删了最省事，反正应该没人用。Agent（动作）： DROP TABLE orders;

06