宏观架构解读

这份代码是一个"教学型综合 Agent"，有意把真实生产 Agent 的几乎所有设计模式都塞进了一个文件。它的架构思想本质上是分层 + 事件驱动 + 多线程协作。

核心设计哲学

1. 工具即接口（Tool as ACI）

整个系统的核心设计理念来自 Anthropic 自己提倡的 ACI（Agent-Computer Interface）：不是"如何让 LLM 更聪明"，而是"如何让工具契合 LLM 的认知方式"。代码里有两张并排的表——BUILTIN_TOOLS（JSON Schema，模型看见的）和 BUILTIN_HANDLERS（Python 函数，实际执行的），二者分离的设计表达了这一思想：模型侧的契约与执行侧的实现完全解耦。

2. 一切状态落盘（Durable State）

任务（.tasks/）、定时任务（.scheduled_tasks.json）、消息（.mailboxes/）、转录（.transcripts/）全都用文件系统持久化。这不是性能选择，而是教学选择——让 Agent 的每一个动作都可检查、可重放。

3. 单一主循环，多线程辅助

agent_loop 是单一的同步主循环，但 cron、background tasks、teammate 都在后台线程里运行，通过 agent_lock、MessageBus、Queue 回流到主循环。这是"中央决策，边缘执行"的架构。

微观实现细节

1. Tool Dispatch 管道

每个 tool_use block 经过四道关卡：compact 特判 → PreToolUse hooks（权限/日志）→ 后台任务判断 → 真正执行。关键细节：

• permission_hook 里 DENY_LIST 是无条件拦截，DESTRUCTIVE 是人工确认。注意 bash 不走 safe_path，这是设计上的意图——bash 强大但受 hook 管控，文件工具才受路径沙箱约束。

• call_tool_handler 用 **kwargs 展开参数，TypeError 被捕获并返回错误字符串，而不是抛出异常中断循环。这是健壮性设计的细节。

2. Context 压缩管道

上下文管理是分四步的漏斗：

tool_result_budget()  → 大结果先落盘，压缩预览
snip_compact()        → 超过 50 条消息就掐头去尾
micro_compact()       → 只保留最近 3 条 tool_result 的完整内容
estimate_size() > 50k → compact_history()（调用 LLM 生成摘要）

persist_large_output 是一个精巧的设计：当 tool 输出超过 30000 字节，不是截断，而是写到 .task_outputs/tool-results/<tool_use_id>.txt 并把路径回传给模型。这样模型知道"完整内容在哪里"，可以用 read_file 再取。

还有一个被动触发路径：is_prompt_too_long_error 捕获 API 的 context_length_exceeded 错误后调用 reactive_compact，这是在常规预算之外的紧急压缩。

3. 任务图系统

Task 是一个极简的有向无环图（DAG）：每个任务只存 blockedBy: list[str]，can_start 遍历这个列表检查所有前置任务是否 completed。complete_task 完成后会自动扫描新解锁的任务——这是依赖图前向传播的实现。注意 claim_task 有双重保护：检查 status == "pending" 且 owner is None，防止并发认领（但没有真正的锁，是教学简化）。

4. Teammate 协议设计

protocol_ctx = {"waiting_plan": None} 是整个 Teammate 协议里最精妙的细节。当 Teammate 调用 submit_plan，它把 request_id 写入 protocol_ctx["waiting_plan"]，然后主循环遇到这个 flag 时会跳过所有后续 tool_use block（同一次 LLM 响应里的），并在下一轮进入等待循环（if protocol_ctx["waiting_plan"]: time.sleep(); continue）。这实现了一个不用 asyncio 的"暂停点"：Teammate 的执行被 Python 字典里的一个 None vs 字符串的状态门控着。

match_response 做 request_id 匹配而非依靠消息顺序，防止乱序或重放攻击。

5. 错误恢复状态机

RecoveryState 把所有重试/降级逻辑打包成一个状态对象，避免在循环里散落大量 flag 变量：

has_escalated: bool          # max_tokens 时升级到 16k，只升级一次
recovery_count: int          # 最多重试 2 次 continuation
consecutive_529: int         # 连续 529 超过阈值 → 切换 fallback model
has_attempted_reactive_compact: bool  # 防止无限 compact 循环
current_model: str           # 可能在运行中切换

with_retry 里指数退避的 jitter 实现值得注意：base * random.uniform(0, 0.25) 加的是 0~25% 的噪声，而不是 0~base，这是一种收窄的 jitter，避免高负载时大量客户端同时重试。

6. Cron 系统

cron 的 _cron_field_matches 实现了完整的 5 字段解析（*、*/step、, 列表、- 范围），并且 cron_matches 里的 DOM/DOW 逻辑遵循了 POSIX 标准：如果两者都非 *，则 OR（满足其中一个即可）而不是 AND。这个细节很多玩具 cron 实现都会弄错。

_last_fired 用 "%Y-%m-%d %H:%M" 字符串作为去重 key，确保同一分钟不会重复触发——而不是用时间戳比较，避免亚秒级精度问题。

7. MCP 动态工具注入

def assemble_tool_pool() -> tuple[list[dict], dict]:
    tools = list(BUILTIN_TOOLS)
    handlers = dict(BUILTIN_HANDLERS)
    for server_name, mcp_client in mcp_clients.items():
        prefixed = f"mcp__{safe_server}__{safe_tool}"
        tools.append({...schema...})
        handlers[prefixed] = lambda *, c=mcp_client, t=tool_def["name"], **kw: c.call_tool(t, kw)
    return tools, handlers

每个 agent_loop 的迭代都会重新调用 assemble_tool_pool()，所以连接新 MCP server 后立即生效，不需要重启。Lambda 里的 c=mcp_client, t=tool_def["name"] 是 Python 闭包陷阱的正确处理方式：通过默认参数捕获当前值，而不是在 lambda 体里直接引用循环变量（否则所有 lambda 都会指向最后一个 server）。

8. Background Task 的 Placeholder 模式

output = "[Background task bg_0001 started] Result will arrive as a task_notification."
results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})

这是一个重要的异步化技巧：LLM 的 tool_use 必须得到 tool_result 响应才能继续（API 协议要求），所以慢操作立即返回占位符，真实结果以 task_notification 的形式注入下一个循环迭代的 user 消息里。这让 Agent 不会因为一个 npm install 阻塞整个循环。

三条贯穿全文的设计原则