11 · 可以搬到自己项目里的实现范式

这一章把前面所有分析压缩成能直接抄进你自己项目的东西。骨架代码用 Python 写（服务端岗位更常见），但结构是语言无关的。

11.1 骨架一：带恢复状态机的主循环

from dataclasses import dataclass, replace
from typing import Literal, Optional

# 所有可能的"继续原因"，穷举出来。对应第 4.2 节的 transition 字段
Reason = Literal["next_turn", "compact_retry", "output_truncated",
                 "hook_blocking", "budget_continue"]

@dataclass(frozen=True)          # frozen=True 表示这个结构体不可修改
class LoopState:
    messages: list               # 当前的完整消息历史
    turn: int = 1                # 已经进行了几轮

    # —— 每条恢复路径一个独立的幂等锁或计数器 ——
    compact_attempted: bool = False        # 本轮是否已经压缩过（真假开关）
    truncation_retries: int = 0            # 输出截断已重试几次（计数器）
    max_tokens_override: Optional[int] = None   # 是否已升过输出上限档位
    hook_active: bool = False
    transition: Optional[Reason] = None    # ★ 只为可观测和可测试而存在


MAX_TRUNCATION_RETRIES = 3

def run(state: LoopState, ctx) -> str:
    while state.turn <= ctx.max_turns and ctx.cost < ctx.max_cost:

        # ① 上下文治理流水线（见骨架二）
        msgs = govern_context(state.messages, ctx)

        # ② 调用模型。可恢复错误在这里被"扣住"，不向外抛（见第 4.3 节）
        resp, recoverable = call_model(msgs, ctx,
                                       max_tokens=state.max_tokens_override)

        # ③ 恢复路径 —— 每条都必须先检查自己的锁
        if recoverable == "context_overflow" and not state.compact_attempted:
            state = replace(state,
                messages=compact(state.messages, ctx),
                compact_attempted=True,               # ★ 上锁
                transition="compact_retry")
            continue

        if recoverable == "output_truncated":
            if state.max_tokens_override is None:     # 还没升过档 → 升一次
                state = replace(state, max_tokens_override=ctx.escalated_max,
                                transition="output_truncated")
                continue
            if state.truncation_retries < MAX_TRUNCATION_RETRIES:
                state = replace(state,
                    messages=state.messages + [resp, RESUME_NUDGE],
                    truncation_retries=state.truncation_retries + 1,
                    max_tokens_override=None,
                    transition="output_truncated")
                continue
            # 重试次数耗尽 —— 现在才把错误抛出去
            raise AgentError(recoverable)

        if recoverable:                # 是可恢复错误，但上面没有一条路能走
            raise AgentError(recoverable)

        # ④ 没有工具调用 = 任务完成（见第 1.5 节）
        if not resp.tool_calls:
            return resp.text

        # ⑤ 执行工具（见骨架三）
        results = execute_tools(resp.tool_calls, ctx)

        # ⑥ 正常推进：★ 只有这里才重置所有恢复计数器
        state = replace(state,
            messages=state.messages + [resp] + results,
            turn=state.turn + 1,
            compact_attempted=False,      # ★ 只在正常推进时重置
            truncation_retries=0,
            max_tokens_override=None,
            transition="next_turn")

    return "预算已耗尽"


RESUME_NUDGE = {
  "role": "user",
  "content": ("Output token limit hit. Resume directly — no apology, "
              "no recap of what you were doing. Pick up mid-thought if that "
              "is where the cut happened. Break remaining work into "
              "smaller pieces."),
}

11.2 骨架二：上下文治理阶梯

def govern_context(messages: list, ctx) -> list:
    """成本递增的四级流水线。越靠前越便宜，能在前面解决就不往后走。
       对应第 6 章。"""

    # ① 落盘：单条工具结果超限 → 存文件，只回预览 + 路径。成本 0
    messages = spill_oversized_results(messages, ctx)

    # ② 精确删除：旧的可压缩工具结果，保留最近 N 条。成本 0
    #    ★ 关键：缓存状态是决策输入（见第 6.3 节）
    if ctx.cache_is_warm:
        # 缓存是热的 —— 本地一个字都不改，让服务端删（如果接口支持）
        ctx.queue_cache_edits(pick_stale_tool_ids(messages, keep=ctx.keep_recent))
    else:
        # 缓存已经凉了 —— 前缀反正要全部重写，直接就地清空内容
        messages = clear_stale_tool_results(messages, keep=max(1, ctx.keep_recent))
        #                                              ★ 至少留 1 条，别删光

    # ③ 结构化折叠：把整段交互折叠成可展开摘要，保留粒度。成本低
    if estimate_tokens(messages) > ctx.collapse_threshold:
        messages = apply_collapses(messages, ctx)

    # ④ 整段摘要：一次完整模型调用，有损不可逆。最后手段
    if estimate_tokens(messages) > ctx.compact_threshold:
        messages = summarize_and_replace(messages, ctx)

    return messages

压缩用的提示词模板

COMPACT_PROMPT = """CRITICAL: Respond with TEXT ONLY. Do NOT call any tools.
- Do NOT use Read, Bash, Grep, Glob, Edit, Write, or ANY other tool.
- You already have all the context you need in the conversation above.
- Tool calls will be REJECTED and will waste your only turn — you will fail.
- Your entire response must be plain text: an <analysis> block followed
  by a <summary> block.

Before the summary, wrap your analysis in <analysis> tags. Chronologically
go through each section of the conversation and identify:
  - the user's explicit requests and intents
  - your approach to addressing them
  - key decisions, technical concepts, and code patterns
  - specific details: file names, full code snippets, function signatures

Then produce <summary>.
"""

def summarize_and_replace(messages, ctx):
    raw = call_model_no_tools(COMPACT_PROMPT, messages, max_turns=1)
    summary = strip_analysis_block(raw)     # ★ analysis 是草稿纸，用完就扔
    return [protected_head(messages),        # 保留开头
            summary_message(summary),        # 中间换成摘要
            *protected_tail(messages, n=ctx.protect_last_n)]   # 保留结尾

11.3 骨架三：工具抽象 + 并发分区

from typing import Protocol, Any

class Tool(Protocol):            # Protocol 是 Python 里定义"接口"的方式
    name: str
    input_schema: dict
    max_result_chars: int        # 结果超过这个长度就落盘

    def call(self, args: dict, ctx) -> Any: ...

    # —— 给调度器看的安全谓词，全部 fail-closed（失败倒向保守）——
    def is_concurrency_safe(self, args: dict) -> bool: return False
    def is_read_only(self, args: dict)        -> bool: return False
    def is_destructive(self, args: dict)      -> bool: return False
    def check_permissions(self, args, ctx)    -> "Decision": ...
    #   ★ 注意这些方法都接收 args ——
    #     同一个 Bash 工具，跑 ls 安全，跑 rm 不安全


def partition(calls: list, tools: dict) -> list[tuple[bool, list]]:
    """贪心分区：相邻的安全工具合并并行，遇到不安全的就切断。
       ★ 完整保留模型隐含的顺序语义。见第 5.4 节。"""
    batches = []
    for c in calls:
        tool = tools.get(c.name)
        try:
            args = validate(tool.input_schema, c.args)
            safe = bool(tool.is_concurrency_safe(args))
        except Exception:
            safe = False              # ★ fail-closed：解析或判定失败都当不安全
        if safe and batches and batches[-1][0]:
            batches[-1][1].append(c)  # 上一批也安全 → 并进去
        else:
            batches.append((safe, [c]))   # 否则 → 开新批
    return batches


def execute_tools(calls, ctx):
    results = []
    for is_safe, batch in partition(calls, ctx.tools):
        if is_safe:
            # ★ 两级中止作用域：批内失败杀兄弟进程，但不杀整个轮次
            #   见第 5.5 节
            sibling = ChildCancelScope(parent=ctx.cancel)
            with ThreadPoolExecutor(max_workers=10) as pool:
                results += list(pool.map(
                    lambda c: run_one(c, ctx, cancel=sibling), batch))
        else:
            for c in batch:
                results.append(run_one(c, ctx, cancel=ctx.cancel))

        if ctx.cancel.is_set():
            # ★ 中断兜底：为每个还没有结果的工具调用补一条合成结果，
            #   否则下一轮 API 会因为"便条与回复不配对"直接报错
            #   见第 4.4 节
            results += synth_results_for_missing(calls, results,
                                                 "Interrupted by user")
            break
    return results

11.4 骨架四：按信任边界配置工具面

# ★ 工具的"实现"和"投放"必须解耦。见第 3.2 节
CORE_TOOLS = ["web_search", "read_file", "write_file", "terminal",
              "patch", "search_files", "delegate", "memory"]

TOOLSETS = {
    # 交互式会话：有人在场，能弹确认框 → 全量工具
    "interactive": {"tools": CORE_TOOLS},

    # 后台任务：弹不出确认框 → 收窄到基本只读
    "background":  {"tools": ["web_search", "read_file", "search_files"]},

    # 公开 webhook：内容来自不可信第三方（PR 标题、issue 评论）
    # → 只给只读工具。
    #   ★ 防提示词注入靠"危险工具根本不在清单里"，
    #     不靠在提示词里写"请不要执行危险命令"
    "webhook":     {"tools": ["web_search", "read_file"]},
}

def tools_for(context) -> list[Tool]:
    preset = ("webhook"    if context.source == "webhook"
         else "background" if not context.can_prompt_user
         else "interactive")
    names = TOOLSETS[preset]["tools"]
    return [t for t in ALL_TOOLS
            if t.name in names and not denied_by_rule(t, context)]

11.5 骨架五：不可绕过的安全底座

import re

# 命令位置锚点：行首 / 分隔符后 / 子 shell 开启符后 / sudo|env|exec 包装后
# ★ 这个锚点是全部的关键。见第 7.3 节
_CMDPOS = r'(?:^|[\n;&|]|\$\(|`|\b(?:sudo|env|exec)\s+)\s*'

HARDLINE = [
    (_CMDPOS + r'rm\s+(-[^\s]*\s+)*(/(?:(?:\.\.?)?/)*(?:\.\.?)?\**|/ \*)(?=[\s;&|]|$)',
                                              "递归删除根目录"),
    (_CMDPOS + r'mkfs(\.[a-z0-9]+)?\b',       "格式化文件系统"),
    (_CMDPOS + r'dd\b[^\n]*\bof=/dev/(sd|nvme|vd)[a-z0-9]*', "写入裸磁盘设备"),
    (_CMDPOS + r'(shutdown|reboot|halt|poweroff)\b',         "关机或重启"),
]
# ★ 模块加载时就预编译。见第 7.3 节末尾（避免正则缓存被挤掉后反复重编）
HARDLINE_C = [(re.compile(p, re.IGNORECASE | re.DOTALL), d) for p, d in HARDLINE]

# 这两条在命令行任意位置都有效，没有命令名可以锚定 → 用引号遮蔽
POSITIONLESS = [
    (re.compile(r'>\s*/dev/(sd|nvme|vd)[a-z0-9]*\b'), "重定向到裸磁盘设备"),
    (re.compile(r':\(\)\s*\{\s*:\s*\|\s*:\s*&\s*\}\s*;\s*:'), "分叉炸弹"),
]

# 会把引号内容交给另一个 shell 执行的命令
SHELL_CARRIERS = {"eval", "sh", "bash", "zsh", "ksh", "dash", "source", "."}

def check_hardline(cmd: str) -> str | None:
    for rx, desc in HARDLINE_C:
        if rx.search(cmd):
            return desc
    # ★ 遇到 sh -c / eval 时，引号里是代码不是散文 → 必须扫原始字符串
    scan = cmd if has_shell_carrier(cmd, SHELL_CARRIERS) else mask_quoted(cmd)
    for rx, desc in POSITIONLESS:
        if rx.search(scan):
            return desc
    return None


def mask_quoted(cmd: str) -> str:
    """把引号内容清空（保留引号字符本身）。
    ★ 但双引号里的 $(...) 和反引号必须保持原样 —— shell 真的会执行它们。
    见第 7.3 节那张三行对照表。"""
    ...

11.6 自建智能体的决策清单

按建议的决策顺序排列。每一条都对应前面某一章的分析。

11.7 一页纸检查清单