第一章：范式革命 – 从"造智能"到"造世界"

1.1 被滥用的"Agent"概念

2024-2025年，“Agent"这个词被彻底滥用了。拖拽式工作流构建器、无代码"AI Agent"平台、Prompt 链编排库……它们共享一个致命幻觉：把 LLM API 调用用 if-else 分支、节点图和硬编码路由逻辑串起来，就是在"构建 Agent”。

learn-claude-code 仓库开篇就给了这个行业一记响亮的耳光：

“You cannot brute-force intelligence by stacking procedural logic… Agency is learned, not coded.”（你无法通过堆叠过程逻辑来暴力生成智能……智能是被学出来的，不是被编码出来的。）

这不是一个教你如何"写智能"的教程。恰恰相反，它教你的是：放弃幻想，回归工程本质，你不是在造大脑，你是在造让大脑发挥作用的"身体"和"环境"。

1.2 核心公式：Agent = Model + Harness

仓库建立了一个极其清晰的认知框架：

Agent = Model（模型） + Harness（驾驭层）

• Model（模型）：是"智能"本身。它通过数十亿次梯度更新，在感知、推理、行动的序列中被训练出来。Claude、GPT、Gemini 的"智能"不是 harness 代码赋予的，而是训练赋予的。
• Harness（驾驭层）：是"环境"和"身体"。它提供工具、知识、观察接口、行动接口和权限边界。模型是司机，套具是车辆。

这个区分直接否定了当前市场上 90% 的"Agent 框架"，那些本质上只是"带 LLM 的 shell 脚本"的 Rube Goldberg 机器（过度工程化的脆弱规则管道）。

1.3 历史铁证：智能来自训练，不是代码

年份	里程碑	核心事实
2013	DeepMind DQN 玩 Atari	单一神经网络，接收原始像素和分数，学会 49 款游戏，超越人类专家
2019	OpenAI Five 征服 Dota 2	5 个神经网络自我对弈 45,000 年，击败 TI8 世界冠军 OG
2019	DeepMind AlphaStar 掌握星际争霸 II	达到欧服 Grandmaster（前 0.15%），无脚本策略
2019	腾讯绝悟统治王者荣耀	1v1 模式下职业玩家 15 场仅赢 1 场，一天训练 = 440 人类年
2024-25	LLM Coding Agents	Claude、GPT、Gemini 作为编码 Agent，读取代码库、编写实现、调试失败

每一个里程碑都指向同一个事实：智能是训练出来的，不是代码堆出来的。

1.4 Harness 的六大组件

Harness = Tools + Knowledge + Observation + Action Interfaces + Permissions
Tools:          文件 I/O、Shell、网络、数据库、浏览器
Knowledge:      产品文档、领域参考、API 规范、风格指南
Observation:    Git diff、错误日志、浏览器状态、传感器数据
Action:         CLI 命令、API 调用、UI 交互
Permissions:    沙箱隔离、审批工作流、信任边界

作为 Harness Engineer，你的实际工作是：

• 实现工具：给 Agent 手。文件读写、Shell 执行、API 调用、浏览器控制。每个工具都是原子化、可组合、描述清晰的。
• 策展知识：给 Agent 领域 expertise。按需加载，而非一次性灌入。
• 管理上下文：给 Agent 干净的记忆。子 Agent 隔离防止噪声泄漏，上下文压缩防止历史淹没当下，任务系统让目标超越单轮对话。
• 控制权限：给 Agent 边界。沙箱文件访问、破坏性操作需审批、信任边界强制执行。
• 收集轨迹数据：每个动作序列都是训练信号，是下一代 Agent 模型的微调原料。

“You are not writing intelligence. You are building the world that intelligence inhabits.”（你不是在编写智能，你是在构建智能栖息的世界。）

第二章：核心模式 – 一个永不变的循环（s01）

2.1 30 行代码的宇宙

整个仓库的教学围绕一个极简但永恒的模式：

THE AGENT PATTERN
=================
User --> messages[] --> LLM --> response
|
stop_reason == "tool_use"?
/                          \
yes                           no
|                             |
execute tools                    return text
append results
loop back -----------------> messages[]

代码实现只有 20 行：

def agent_loop(messages):
while True:
response = client.messages.create(
model=MODEL, system=SYSTEM,
messages=messages, tools=TOOLS,
max_tokens=8000,
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
return
results = []
for block in response.content:
if block.type == "tool_use":
output = TOOL_HANDLERS[block.name](**block.input)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": output,
})
messages.append({"role": "user", "content": results})

不到 30 行，这就是最小可运行的 Agent Harness 内核。它不是智能本身，而是让模型能持续行动的最小运行框架。模型负责决策（要不要调工具、调哪个），Harness 负责执行（调了就跑、结果喂回去）。

2.2 与 Claude Code 源码的精确对照

教学版的 30 行while True就是 Claude Code 1729 行query.ts的核心。CC 的复杂性不是"另一个 agent 大脑"，而是生产级保护机制：

一、循环结构差异

教学版检查 response.stop_reason。CC 不把它作为循环继续的唯一依据，流式响应中 stop_reason 可能还没更新但内容里已经有 tool_use 块了。CC 用 needsFollowUp 标志：

// query.ts:554-558
// stop_reason === 'tool_use' is unreliable.
// Set during streaming whenever a tool_use block arrives.
let needsFollowUp = false

二、State 对象的 10 个字段

#	字段	用途	对应章节
1	messages	当前迭代的消息数组	s01
2	toolUseContext	工具、信号、权限上下文	s02
3	autoCompactTracking	压缩状态追踪	s08
4	maxOutputTokensRecoveryCount	token 恢复尝试次数（上限 3）	s11
5	hasAttemptedReactiveCompact	本轮是否已尝试响应式压缩	s08
6	maxOutputTokensOverride	8K→64K 的升级覆盖	s11
7	pendingToolUseSummary	后台 Haiku 生成的 tool use 摘要	s08
8	stopHookActive	停止钩子是否产生阻塞错误	s04
9	turnCount	轮次计数（maxTurns 检查）	s01
10	transition	上一次继续原因	s11

三、多条退出和继续路径

教学版只有 1 条退出路径（模型不调工具就结束）。生产版有多条退出和继续路径，覆盖 blocking limit、prompt too long、model error、abort、hook stop、max turns、token budget continuation、reactive compact retry 等场景。每种场景都有对应的恢复或退出策略。

四、流式工具执行

CC 的StreamingToolExecutor让工具在模型还在生成时就开始并行执行（根据工具是否 concurrency-safe 决定并发或独占）。教学版不实现这些：目标是概念清晰，不是性能极致。

一句话总结：1729 行的 query.ts 核心就是 30 行 while True。所有复杂字段和退出路径都是保护机制。先理解核心循环，后面的一切自然展开。

第三章：工具分发 – 从 1 个工具到 5 个工具（s02）

3.1 问题：只有 Bash 的局限

s01 的 Agent 只有一个 bash 工具。读文件要cat，写文件要 echo “…” > file.py，改文件要sed。模型想的是"读这个文件"，却要拼出cat path/to/file。多了一层翻译，浪费 token，还容易拼错。

3.2 解决方案：TOOL_HANDLERS 查表分发

s01 的循环完全保留。唯一的变动在工具执行那 1 行：run_bash() 替换为 TOOL_HANDLERSblock.name 查表分发。

给 Agent 加一个工具只需要做两件事：

• 定义工具：在 TOOLS 数组里加一条描述
• 注册处理函数：在 TOOL_HANDLERS 字典里加一个映射

3.3 从 1 个工具到 5 个工具

TOOLS = [
{"name": "bash",       "description": "Run a shell command.", ...},
{"name": "read_file",  "description": "Read file contents.",  ...},
{"name": "write_file", "description": "Write content to file.", ...},
{"name": "edit_file",  "description": "Replace text in file once.", ...},
{"name": "glob",       "description": "Find files by pattern.", ...},
]
def run_read(path, limit=None):
lines = safe_path(path).read_text().splitlines()
if limit:
lines = lines[:limit]
return "\n".join(lines)
def run_write(path, content):
safe_path(path).write_text(content)
return f"Wrote {len(content)} bytes to {path}"
def run_edit(path, old_text, new_text):
text = safe_path(path).read_text()
if old_text not in text:
return "Error: text not found"
safe_path(path).write_text(text.replace(old_text, new_text, 1))
return f"Edited {path}"
def run_glob(pattern):
import glob as g
return "\n".join(g.glob(pattern, root_dir=WORKDIR))
TOOL_HANDLERS = {
"bash":       run_bash,
"read_file":  run_read,
"write_file": run_write,
"edit_file":  run_edit,
"glob":       run_glob,
}

加一个工具 = 在 TOOLS 数组加一条 + 在 TOOL_HANDLERS 字典加一行。循环不变。

3.4 与 CC 源码的精确对照

一、工具定义方式

教学版：TOOLS 数组 +TOOL_HANDLERS 字典，定义和实现分开。 CC：每个工具是 buildTool() 创建的独立对象，包含 schema、验证、权限、执行。getAllBaseTools() 汇总所有工具。

二、并发安全判断：isConcurrencySafe()

教学版按原始顺序逐个执行，不做并发。CC 用 isConcurrencySafe(input) 判断能否并发，注意这不是简单的"只读 vs 写"，而是按具体输入判断：

工具	isReadOnly	isConcurrencySafe
FileRead	true	true
Glob	true	true
Bash ls	true	true ← 关键差异
Bash rm	false	false
TaskCreate	false	true ← 改状态但可并发

三、分区算法

CC 的partitionToolCalls() 不是分两组，而是把工具调用按连续块分批：

[read A, read B, glob *.py, bash "rm x", read C]
→ batch1(并发): [read A, read B, glob *.py]
→ batch2(串行): [bash "rm x"]
→ batch3(并发): [read C]

并发安全的连续块编入同一个 batch，batch 内真正并发执行。遇到非并发安全的就开新 batch 串行执行。batch 之间严格顺序。

四、验证管线

CC 的每个工具调用经过严格的 5 步验证：

• Zod schema 验证（参数类型/结构检查）
• 工具级 validateInput()（参数值验证，如路径是否在工作区内）
• PreToolUse hooks（钩子可以返回消息、修改输入、阻止执行）
• 权限检查（canUseTool+checkPermissions→ allow/deny/ask）
• 执行 tool.call()

五、工具结果持久化

每个工具有一个maxResultSizeChars字段。结果超过这个值就落盘，模型看到的是预览 + 文件路径。FileRead 特殊，设为 Infinity，防止读文件的输出又被当成文件落盘，导致无限循环（读文件 → 落盘 → 再读 → 再落盘 → …）。

第四章：权限系统 – 三道闸门（s03）

4.1 问题：安全不能靠信任模型

s02 的 Agent 有 5 个工具。file tools 受safe_path保护，但 bash 不受限制。让它"清理一下项目"，可能执行 rm -rf /。

安全不能靠信任模型，要靠代码，在工具执行之前做判断。

4.2 三道闸门管线

┌─────────────────────────────────────────────────────────┐
│工具调用 → 闸门1: 硬拒绝 → 闸门2: 规则匹配 → 闸门3: 用户审批  │
│          (deny list)    (permission rules)   (ask user) │
└─────────────────────────────────────────────────────────┘

闸门 1：硬拒绝表

DENY_LIST = [
"rm -rf /", "sudo", "shutdown", "reboot",
"mkfs", "dd if=", "> /dev/sda",
]
def check_deny_list(command: str) -> str | None:
for pattern in DENY_LIST:
if pattern in command:
return f"Blocked: '{pattern}' is on the deny list"
return None

闸门 2：规则匹配

PERMISSION_RULES = [
{
"tools": ["write_file", "edit_file"],
"check": lambda args: not (WORKDIR / args.get("path", "")).resolve().is_relative_to(WORKDIR),
"message": "Writing outside workspace",
},
{
"tools": ["bash"],
"check": lambda args: any(kw in args.get("command", "") for kw in ["rm ", "> /etc/", "chmod 777"]),
"message": "Potentially destructive command",
},
]
def check_rules(tool_name: str, args: dict) -> str | None:
for rule in PERMISSION_RULES:
if tool_name in rule["tools"] and rule["check"](args):
return rule["message"]
return None

闸门 3：用户审批

def ask_user(tool_name: str, args: dict, reason: str) -> str:
print(f"\n⚠  {reason}")
print(f"   Tool: {tool_name}({args})")
choice = input("   Allow? [y/N] ").strip().lower()
return "allow" if choice in ("y", "yes") else "deny"

三道闸门串在一起，插在工具执行之前：

def check_permission(block) -> bool:
# 闸门 1: 硬拒绝
if block.name == "bash":
reason = check_deny_list(block.input.get("command", ""))
if reason:
print(f"\n⛔ {reason}")
return False
# 闸门 2 + 3: 规则匹配 → 用户审批
reason = check_rules(block.name, block.input)
if reason:
decision = ask_user(block.name, block.input, reason)
if decision == "deny":
return False
return True
# 在 agent_loop 中——s02 的循环只加了一行：
for block in response.content:
if block.type == "tool_use":
if not check_permission(block):           # ← 新增
results.append({... "content": "Permission denied."})
continue
output = TOOL_HANDLERS[block.name](**block.input)
results.append(...)

4.3 与 CC 源码的精确对照

一、PermissionResult：不是 3 种，是 4 种

教学版的三道闸门（deny → ask → allow）和 CC 不完全对应。CC 的PermissionResult有 4 个 behavior：

behavior	含义	教学版对应
allow	直接允许	闸门 3 通过
deny	直接拒绝	闸门 1 命中
ask	弹出对话框问用户	闸门 2 命中
passthrough	工具不表态，交给通用管线决定	教学版无

二、生产版的验证阶段

CC 的工具调用不是经过三道闸门，而是经过多个阶段：

• Zod schema 验证— 参数类型检查
• validateInput()— 工具级语义验证
• backfillObservableInput()— 补全遗留字段
• PreToolUse hooks— 钩子可以返回 allow/deny/ask
• resolveHookPermissionDecision()— 协调钩子+管线决策
• hasPermissionsToUseToolInner()— 多层规则检查：

• 整个工具被 deny rule 禁用 → deny
• 整个工具被 ask rule 标记 → ask
• tool.checkPermissions() 工具自己的判断
• 工具自己返回 deny → deny
• requiresUserInteraction() → ask
• 内容相关的 ask 规则 → ask（不可绕过）
• 安全检查违规 → ask（不可绕过）
• bypassPermissions 模式 → allow
• 整个工具被 allow rule 放行 → allow
• passthrough → 转为 ask

三、拒绝列表：不是一个文件，是 8 个来源

CC 没有单一的 deny list。权限规则来自 8 个来源：

来源	配置位置
userSettings	~/.claude/settings.json
projectSettings	.claude/settings.json
localSettings	settings.local.json
flagSettings	Feature flags
policySettings	企业管理策略
cliArg	--allowedTools / --deniedTools
command	内联命令
session	会话内临时授权

每条规则格式：{ toolName: “Bash”, ruleBehavior: “deny”, ruleContent: “npm publish:*” }。多个来源的规则合并，高优先级来源覆盖低优先级。

四、YoloClassifier（自动审批）

CC 的 auto 模式下，不会每次都弹对话框。classifyYoloAction把工具调用 + 对话上下文发给一个分类器 LLM 判断是否安全。先尝试 acceptEdits 模式模拟，再查安全工具白名单，最后才调分类器。分类器连续拒绝太多次 → 回退到人工审批。

五、权限冒泡

子 Agent（通过 AgentTool fork 出来的）的permissionMode设为’bubble’。意思是权限弹窗冒泡到父 Agent 的终端，而不是在子 Agent 里静默拒绝。

第五章：Hook 系统 – 挂在循环上，不写进循环里（s04）

5.1 问题：循环膨胀症

s03 的 Agent 有权限检查了。但每次加一个新检查，比如"记录每次 bash 调用"、“操作后自动 git commit”，都要修改agent_loop函数。

循环很快就变成了这样：

def agent_loop(messages):
while True:
# ... LLM call ...
for block in response.content:
if block.type != "tool_use":
continue
log_to_file(block)          # 加一行
check_permission(block)     # 加一行
notify_slack(block)         # 又加一行
output = execute(block)
auto_git_add(block)         # 再加一行
# ... 很快循环就认不出来了

你想扩展的是 Agent 的行为，但你改的却是循环本身。循环应该是一个稳定的核心，扩展应该挂在外面。

5.2 四个事件，覆盖完整 Agent Cycle

事件	触发时机	典型用途
UserPromptSubmit	用户输入提交后、进入 LLM 前	输入验证、注入上下文
PreToolUse	工具执行前	权限检查、日志记录
PostToolUse	工具执行后	副作用（自动 git add 等）、输出检查
Stop	循环即将退出时	收尾清理（CC 还支持强制续跑）

5.3 Hook 注册表实现

HOOKS = {
"UserPromptSubmit": [],
"PreToolUse": [],
"PostToolUse": [],
"Stop": [],
}
def register_hook(event: str, callback):
HOOKS[event].append(callback)
def trigger_hooks(event: str, *args):
for callback in HOOKS[event]:
result = callback(*args)
if result is not None:   # 返回值 ≠ None → hook 说"停"
return result
return None

PreToolUse hook 示例（s03 的权限检查移到 hook 上）：

def permission_hook(block):
if block.name == "bash":
for pattern in DENY_LIST:
if pattern in block.input.get("command", ""):
return "Permission denied by deny list"
if block.name in ("write_file", "edit_file"):
path = block.input.get("path", "")
if not (WORKDIR / path).resolve().is_relative_to(WORKDIR):
choice = input("   Allow? [y/N] ").strip().lower()
if choice not in ("y", "yes"):
return "Permission denied by user"
return None
def log_hook(block):
print(f"[HOOK] {block.name}(...)")
def large_output_hook(block, output):
if len(str(output)) > 100000:
print(f"[HOOK] ⚠ Large output from {block.name}")
register_hook("PreToolUse", permission_hook)
register_hook("PreToolUse", log_hook)
register_hook("PostToolUse", large_output_hook)

Stop hook 示例（强制续跑）：

def summary_hook(messages: list) -> str | None:
tool_count = sum(1 for m in messages
for b in (m.get("content") if isinstance(m.get("content"), list) else [])
if isinstance(b, dict) and b.get("type") == "tool_result")
print(f"\033[90m[HOOK] Stop: session used {tool_count} tool calls\033[0m")
return None   # return None = allow stop, return string = force continuation
register_hook("Stop", summary_hook)

在 agent_loop 中：

if response.stop_reason != "tool_use":
force = trigger_hooks("Stop", messages)   # ← 退出之前
if force:
messages.append({"role": "user", "content": force})
continue
return

5.4 与 CC 源码的精确对照

一、Hook 事件：不止 4 个，而是 27 个

类别	事件
工具相关	PreToolUse , PostToolUse, PostToolUseFailure
会话相关	SessionStart , SessionEnd, Stop, StopFailure, Setup
用户交互	UserPromptSubmit , Notification, PermissionRequest, PermissionDenied
子 Agent	SubagentStart , SubagentStop
压缩相关	PreCompact , PostCompact
团队相关	TeammateIdle , TaskCreated, TaskCompleted
其他	Elicitation , ElicitationResult, ConfigChange, WorktreeCreate, WorktreeRemove, InstructionsLoaded, CwdChanged, FileChanged

二、HookResult 的 14 个字段

字段	类型	用途
message	Message	可选 UI 消息
blockingError	HookBlockingError	阻塞错误 → 注入对话让模型自纠
outcome	success/blocking/non_blocking_error/cancelled	执行结果
preventContinuation	boolean	阻止后续执行
stopReason	string	停止原因描述
permissionBehavior	allow/deny/ask/passthrough	hook 返回权限决策
updatedInput	Record	修改工具输入
additionalContext	string	附加上下文
updatedMCPToolOutput	unknown	MCP 工具输出修改

三、关键安全不变式：Hook ‘allow’ 不能绕过 deny/ask 规则

这是 CC 权限系统最重要的安全设计：hook 返回 allow 时，仍然要检查 settings.json 的 deny/ask 规则。即使用户的 hook 脚本说"允许"，如果在 settings.json 中禁用了这个工具，操作仍然会被阻止。

四、stopHookActive防无限循环机制

当 stop hooks 产生 blockingError 时，循环带stopHookActive: true重入下一轮。后续迭代中 stop hooks 看到这个标志就不会再次触发。这防止了一个永不停机的 bug：模型自纠后 stop hook 再次报错 → 模型再自纠 → stop hook 再报错…

第六章：TodoWrite – 没有计划的 Agent 会漂移（s05）

6.1 问题：上下文淹没导致目标遗忘

给 Agent 一个复杂任务：“把所有 Python 文件改成 snake_case 命名，然后跑测试，修好失败。”

Agent 开始干活，改了 3 个文件，跑了个测试，发现 2 个失败，开始修。修着修着，它忘了最初是"改成 snake_case"，测试失败把注意力全吸走了。

对话越长越严重：工具结果不断填满上下文，系统提示的影响力被稀释。一个 10 步重构，做完 1-3 步就开始即兴发挥，因为 4-10 步已经被挤出注意力了。

6.2 解决方案：todo_write 工具 + Nag Reminder

todo_write 本身不做任何实际工作，不能读文件、不能跑命令，只是让 Agent 在动手之前先理清思路。

CURRENT_TODOS: list[dict] = []
def run_todo_write(todos: list) -> str:
global CURRENT_TODOS
CURRENT_TODOS = todos
lines = ["\n## Current Tasks"]
for t in CURRENT_TODOS:
icon = {"pending": " ", "in_progress": "▸", "completed": "✓"}[t["status"]]
lines.append(f"  [{icon}] {t['content']}")
print("\n".join(lines))
return f"Updated {len(CURRENT_TODOS)} tasks"

Nag Reminder：模型连续 3 轮没调 todo_write 时，自动注入一条提醒：

if rounds_since_todo >= 3 and messages:
messages.append({
"role": "user",
"content": "<reminder>Update your todos.</reminder>",
})
rounds_since_todo = 0

关键洞察：todo_write 不给 Agent 增加任何执行能力。它增加的是规划能力。

6.3 与 CC 源码的精确对照

CC 中有两套任务系统并存：

• TodoWrite（V1）：一个简单的列表工具，数据在内存 AppState 中维护。教学版也保存在进程内存里，退出后清空
• Task System（V2 = s12）：文件持久化、依赖图、并发锁、ownership

切换由 isTodoV2Enabled() 控制。当前源码的实现逻辑：交互式会话中 V2 默认启用，非交互式会话（SDK）中 V1 默认启用。

教学版的 nag reminder（3 轮未更新就注入提醒）是教学机制。CC 源码中没有固定的"3 轮"逻辑，更接近的是当 3 个以上 todo 全部完成但没有 verification 项时，追加 verification nudge。

第七章：Subagent – 大任务拆小，干净上下文（s06）

7.1 问题：上下文污染

Agent 在修一个 bug。它读了 30 个文件来追踪调用链，中间聊了 60 轮。messages 列表涨到 120 条，其中大部分是"追踪调用链"的中间过程，和"修 bug"这个最终目标无关。

这些中间过程占着上下文位置，让 Agent 越来越"健忘"。

7.2 解决方案：独立 messages[] + 只回传结论

def spawn_subagent(description: str) -> str:
# 子 Agent 的工具：基础工具，但没有 task（禁止递归）
sub_tools = [
{"name": "bash", ...}, {"name": "read_file", ...},
{"name": "write_file", ...}, {"name": "edit_file", ...},
{"name": "glob", ...},
]
messages = [{"role": "user", "content": description}]  # 全新 messages[]
for _ in range(30):  # safety limit
response = client.messages.create(
model=MODEL, system=SUB_SYSTEM,
messages=messages, tools=sub_tools, max_tokens=8000,
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
break
results = []
for block in response.content:
if block.type == "tool_use":
blocked = trigger_hooks("PreToolUse", block)
if blocked:
results.append({... "content": str(blocked)})
continue
handler = SUB_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown"
trigger_hooks("PostToolUse", block, output)
results.append({... "content": output})
messages.append({"role": "user", "content": results})
# 只返回最后的文本结论，中间过程全部丢弃
return extract_text(messages[-1]["content"])

三个关键设计决策：

决策	选择	原因
上下文隔离	全新 messages[]	子 Agent 的中间过程不污染主 Agent 的上下文
只回传结论	extract_text(last_message)	不是回传整个 messages 列表
禁止递归	子 Agent 无 task 工具	防止子 Agent 再 spawn 新的子 Agent
安全策略不跳过	子 Agent 工具调用也走 PreToolUse hook	上下文隔离不代表权限隔离

7.3 与 CC 源码的精确对照

一、不是一种模式，是三种

模式	触发条件	上下文
Normal Subagent	指定了 subagent_type	全新 messages[]，只有 prompt
Fork Subagent	没指定 subagent_type，fork gate 开启	通过 buildForkedMessages() 构造 cache-friendly 前缀，共享 prompt cache
General-Purpose	没指定 subagent_type，fork gate 关闭	同 Normal

二、Fork 模式：为了共享 Prompt Cache

Fork 模式不创建全新上下文，而是通过buildForkedMessages()构造 cache-friendly 消息前缀，保留父 assistant message 并生成 placeholder tool results。目的不是隔离，而是让 Anthropic API 的 prompt cache 命中：父子 Agent 的 system prompt、tools、messages 前缀完全一致，API 端不需要重算。

缓存命中的五个关键组件：system prompt、tools、model、messages 前缀、thinking config，必须字节级一致。

三、Context Isolation 的精确粒度

createSubagentContext()创建子 Agent 的ToolUseContext：

字段	行为
abortController	新的 child controller，父 abort 向下传播
setAppState	默认 no-op；但 sync agent 通过 shareSetAppState 共享
readFileState	从父克隆 （避免重复读相同文件）
queryTracking	新 chainId，depth = parentDepth + 1

子 Agent 不是完全隔离的：文件读取状态是共享的。

四、递归 Fork 防护

教学版用"子 Agent 不给 task 工具"表达递归保护。真实实现更精细：isInForkChild()检查对话历史中是否有 FORK_BOILERPLATE_TAG，有就拒绝。但 Agent工具默认在所有 agent 的禁用集合里，针对 fork child 有专门的递归保护，teammate 场景下有特殊放行。不是简单的"禁止新的子 Agent"。

五、Permission Bubbling

Fork Agent 的 permissionMode: ‘bubble’ 意味着子 Agent 的权限弹窗冒泡到父终端，用户在主终端里审批子 Agent 的操作。

六、Async vs Sync

教学版只展示了同步子 Agent（父等着子跑完）。CC 还支持异步路径：run_in_background: true时异步启动，返回{ status: ‘async_launched’ }立即给父 Agent，子 Agent 完成后通过通知机制告知父 Agent。

第八章：Context Compact – 四层压缩策略（s08）

8.1 问题：上下文窗口是有限的

Agent 读了一个 1000 行的文件（~4000 token），又读了 30 个文件，跑了 20 条命令。每条命令的输出、每个文件的内容，全都堆在messages列表里。上下文窗口满了之后，API 直接拒绝：prompt_too_long。

8.2 核心设计：便宜的先跑，贵的后跑

┌─────────────────────────────────────────────────────────────┐
│  L1: snip_compact    → 裁掉无关的旧对话（0 API）              │
│  L2: micro_compact   → 旧工具结果占位（0 API）                │
│  L3: tool_result_budget → 大结果落盘（0 API）                 │
│  L4: compact_history → LLM 全量摘要（1 API）                  │
│  应急: reactive_compact → 更激进的裁剪（1 API）                │
└─────────────────────────────────────────────────────────────┘

8.3 L1: snip_compact 裁掉无关的旧对话

消息数超过 50 条 → 保留头部 3 条（初始上下文）和尾部 47 条（当前工作），中间裁掉。唯一额外边界条件：不能把 assistant(tool_use) 和后面的 user(tool_result) 拆开：

def snip_compact(messages, max_messages=50):
if len(messages) <= max_messages:
return messages
head_end, tail_start = 3, len(messages) - (max_messages - 3)
if head_end > 0 and _message_has_tool_use(messages[head_end - 1]):
while head_end < len(messages) and _is_tool_result_message(messages[head_end]):
head_end += 1
if (tail_start > 0 and tail_start < len(messages)
and _is_tool_result_message(messages[tail_start])
and _message_has_tool_use(messages[tail_start - 1])):
tail_start -= 1
snipped = tail_start - head_end
placeholder = {"role": "user", "content": f"[snipped {snipped} messages from conversation middle]"}
return messages[:head_end] + [placeholder] + messages[tail_start:]

8.4 L2: micro_compact — 旧工具结果占位

只保留最近 3 条tool_result的完整内容，更旧的替换为一行占位符：

KEEP_RECENT_TOOL_RESULTS = 3
def micro_compact(messages):
tool_results = collect_tool_result_blocks(messages)
if len(tool_results) <= KEEP_RECENT_TOOL_RESULTS:
return messages
for _, _, block in tool_results[:-KEEP_RECENT_TOOL_RESULTS]:
if len(block.get("content", "")) > 120:
block["content"] = "[Earlier tool result compacted. Re-run if needed.]"
return messages

8.5 L3: tool_result_budget — 大结果落盘

统计最后一条 user 消息里所有tool_result的总大小。超过 200KB → 按大小排序，从最大的开始落盘到.task_outputs/tool-results/，上下文里只留标记 + 前 2000 字符预览：

def tool_result_budget(messages, max_bytes=200_000):
last = messages[-1]
blocks = [(i, b) for i, b in enumerate(last["content"])
if b.get("type") == "tool_result"]
total = sum(len(str(b.get("content", ""))) for _, b in blocks)
if total <= max_bytes:
return messages
ranked = sorted(blocks, key=lambda p: len(str(p[1].get("content", ""))), reverse=True)
for idx, block in ranked:
if total <= max_bytes:
break
block["content"] = persist_large_output(block["tool_use_id"], str(block["content"]))
total = recalculate_total(blocks)
return messages

8.6 L4: compact_history — LLM 全量摘要

三步流程：

• 保存 transcript：完整对话写入.transcripts/，JSONL 格式
• LLM 生成摘要：把对话历史发给 LLM，要求保留当前目标、重要发现、已改文件、剩余工作、用户约束等关键信息
• 替换消息列表：所有旧消息被替换为一条摘要

def compact_history(messages):
transcript_path = write_transcript(messages)  # 先保存完整对话
summary = summarize_history(messages)          # LLM 生成摘要
return [{"role": "user",
"content": f"[Compacted]\n\n{summary}"}]

熔断器：连续失败 3 次后停止重试，防止死循环浪费 API 调用。

8.7 应急: reactive_compact

API 返回 prompt_too_long（413）时触发，比 compact_history 更激进：

def reactive_compact(messages):
transcript = write_transcript(messages)
tail_start = max(0, len(messages) - 5)
if (tail_start > 0 and tail_start < len(messages)
and _is_tool_result_message(messages[tail_start])
and _message_has_tool_use(messages[tail_start - 1])):
tail_start -= 1
summary = summarize_history(messages[:tail_start])
return [{"role": "user",
"content": f"[Reactive compact]\n\n{summary}"}, *messages[tail_start:]]

8.8 完整的压缩管线

def agent_loop(messages):
reactive_retries = 0
while True:
# 三个预处理器（0 API 调用）
# 顺序：budget 先跑，确保大内容落盘后再做占位和裁剪
messages[:] = tool_result_budget(messages)    # L3: 大结果落盘
messages[:] = snip_compact(messages)          # L1: 裁中间
messages[:] = micro_compact(messages)         # L2: 旧结果占位
# 还不够？LLM 摘要（1 API 调用）
if estimate_token_count(messages) > THRESHOLD:
messages[:] = compact_history(messages)
try:
response = client.messages.create(...)
except PromptTooLongError:
if reactive_retries < MAX_REACTIVE_RETRIES:
messages[:] = reactive_compact(messages)  # 应急
reactive_retries += 1
continue
raise  # 超过重试上限，抛出异常
# ... 工具执行 ...
# compact 工具：模型主动调用时触发 compact_history
if block.name == "compact":
messages[:] = compact_history(messages)
results.append({..., "content": "[Compacted. History summarized.]"})
messages.append({"role": "user", "content": results})
break

顺序不能换。L3（budget）在 L2（micro）前面，因为 micro 会把旧的大 tool_result 替换成一行占位符，budget 必须在那之前把完整内容落盘。

8.9 与 CC 源码的精确对照

执行顺序对照：

维度	教学版	Claude Code
执行顺序	budget → snip → micro → auto	budget → snip → micro → collapse → auto
snip_compact	保留头 3 + 尾 47	CC 仅主线程启用；HISTORY_SNIP feature gate
micro_compact	文本占位符替换	两条路径：time-based 直接清内容，cached 走 API cache_edits
tool_result_budget	200KB 字符	200,000 字符
compact_history 阈值	字符数估算	精确 token：contextWindow - maxOutputTokens - 13_000
摘要要求	5 类信息	9 个部分 + <analysis>/<summary> 双标签
压缩 prompt	简单 prompt	首尾双重防呆禁止调工具
PTL retry	有（简化）	truncateHeadForPTLRetry() 按消息组回退
后压缩恢复	无（教学版只保留摘要）	自动重新读取最近文件、计划、agent/skill/tool 等
熔断器	3 次	3 次
reactive 重试	1 次	CC 有更精细的分级重试

完整常量参考：

常量	值	源文件
AUTOCOMPACT_BUFFER_TOKENS	13,000	autoCompact.ts:62
MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES	3	autoCompact.ts:70
MAX_OUTPUT_TOKENS_FOR_SUMMARY	20,000	autoCompact.ts:30
POST_COMPACT_TOKEN_BUDGET	50,000	compact.ts:123
POST_COMPACT_MAX_FILES_TO_RESTORE	5	compact.ts:122
POST_COMPACT_MAX_TOKENS_PER_FILE	5,000	compact.ts:124
时间 micro_compact 间隔	60 分钟	timeBasedMCConfig.ts
MAX_COMPACT_STREAMING_RETRIES	2	compact.ts:131

第九章：Task System – 文件持久化的任务图（s12）

9.1 问题：TodoWrite 的局限

s05 的 TodoWrite 是当前任务的执行清单，保存在会话内存中。但"重构整个后端"涉及认证模块、数据库层、API 路由、测试，任务之间有先后依赖。需要任务系统：每个任务是一个 JSON 文件，任务之间有blockedBy依赖，跨会话持久化在磁盘上。

9.2 数据结构

@dataclass
class Task:
id: str
subject: str
description: str
status: str          # pending | in_progress | completed
owner: str | None    # Agent 名（多 Agent 场景）
blockedBy: list[str] # 依赖的任务 ID 列表

9.3 五个工具操作

create_task：创建任务，自动 save_task 到 .tasks/{id}.json

def create_task(subject: str, description: str = "",
blockedBy: list[str] | None = None) -> Task:
task = Task(
id=f"task_{int(time.time())}_{random_hex(4)}",
subject=subject, description=description,
status="pending", owner=None,
blockedBy=blockedBy or [],
)
save_task(task)
return task

can_start：依赖检查

def can_start(task_id: str) -> bool:
task = load_task(task_id)
for dep_id in task.blockedBy:
if not _task_path(dep_id).exists():
return False
dep = load_task(dep_id)
if dep.status != "completed":
return False
return True

claim_task：认领任务

def claim_task(task_id: str, owner: str = "agent") -> str:
task = load_task(task_id)
if task.status != "pending":
return f"Task {task_id} is {task.status}, cannot claim"
if not can_start(task_id):
deps = [d for d in task.blockedBy
if load_task(d).status != "completed"]
return f"Blocked by: {deps}"
task.owner = owner
task.status = "in_progress"
save_task(task)
return f"Claimed {task_id} ({task.subject})"

complete_task：完成与解锁

def complete_task(task_id: str) -> str:
task = load_task(task_id)
task.status = "completed"
save_task(task)
unblocked = [t.subject for t in list_tasks()
if t.status == "pending" and t.blockedBy
and can_start(t.id)]
msg = f"Completed {task_id} ({task.subject})"
if unblocked:
msg += f"\nUnblocked: {', '.join(unblocked)}"
return msg

状态机：

pending ──claim──→ in_progress ──complete──→ completed

9.4 与 CC 源码的精确对照

TaskRecord 的完整 9 个字段：

字段	类型	用途
id	string	递增整数 ID
subject	string	简短标题
description	string	自由格式描述
activeForm	string?	进行时态，in_progress 时在 spinner 显示
owner	string?	分配的 agent ID
status	pending/in_progress/completed	生命周期
blocks	string[]	此任务阻塞的任务 ID（下游）
blockedBy	string[]	阻塞此任务的任务 ID（上游）
metadata	Record?	任意扩展键值对

并发认领的锁机制：

claimTask() 用双重锁防竞争：

• 任务文件锁：proper-lockfile 锁住 {taskId}.json（最多重试 30 次，指数退避 5-100ms）。锁内重新读取任务、检查已被他人认领、检查已完成、检查上游未完成、设置 owner。
• 列表级锁：.lock 文件，原子性扫描所有任务并检查该 agent 是否已有其他 open task。

高水位标防 ID 重用：

.highwatermark 文件记录曾分配过的最高任务 ID。即使任务被删除，ID 也不会被重用。

四个 Task 工具：

CC 的任务系统有四个工具（不是教学版的一个通用 Task 工具）：TaskCreate、TaskGet、TaskUpdate、TaskList。全部设置isConcurrencySafe: true和shouldDefer: true。

第十章：Agent Teams – 多 Agent 协作（s15）

10.1 问题：单个 Agent 的注意力覆盖不了所有模块

"重构整个后端"涉及认证模块、数据库层、API 路由、测试。一个 Agent 在修 API 路由时，认证模块的细节已经不在上下文里了。

s06 的子 Agent 是临时工，叫来干一件事就走了。但有些任务需要能通信、能协作的队友。

10.2 解决方案：MessageBus + 队友线程

MessageBus：文件收件箱

class MessageBus:
def send(self, from_agent: str, to_agent: str,
content: str, msg_type: str = "message"):
msg = {"from": from_agent, "to": to_agent,
"content": content, "type": msg_type,
"ts": time.time()}
inbox = MAILBOX_DIR / f"{to_agent}.jsonl"
with open(inbox, "a") as f:
f.write(json.dumps(msg) + "\n")
def read_inbox(self, agent: str) -> list[dict]:
inbox = MAILBOX_DIR / f"{agent}.jsonl"
if not inbox.exists():
return []
msgs = [json.loads(line) for line in inbox.read_text().splitlines()]
inbox.unlink()  # 消费式：读完删除
return msgs

spawn_teammate_thread：启动队友

def spawn_teammate_thread(name: str, role: str, prompt: str) -> str:
system = f"You are '{name}', a {role}. Use tools to complete tasks."
def run():
messages = [{"role": "user", "content": prompt}]
sub_tools = [bash, read_file, write_file, send_message]
for _ in range(10):           # 最多 10 轮
inbox = BUS.read_inbox(name)
if inbox:
messages.append({"role": "user",
"content": f"<inbox>{json.dumps(inbox)}</inbox>"})
response = client.messages.create(
model=MODEL, system=system, messages=messages[-20:],
tools=sub_tools, max_tokens=8000)
# ... 执行工具、处理结果
BUS.send(name, "lead", summary, "result")
threading.Thread(target=run, daemon=True).start()

Lead 的 inbox 注入：

inbox = BUS.read_inbox("lead")
if inbox:
inbox_text = "\n".join(
f"From {m['from']}: {m['content'][:200]}" for m in inbox)
history.append({"role": "user",
"content": f"[Inbox]\n{inbox_text}"})

10.3 与 CC 源码的精确对照

15 种消息类型：

类型	方向	用途
plain text	双向	普通队友间通信
idle_notification	队友→Lead	队友完成一轮工作，进入空闲
permission_request	队友→Lead	队友需要操作审批
permission_response	Lead→队友	Lead 审批结果
plan_approval_request	队友→Lead	队友提交计划待审
plan_approval_response	Lead→队友	Lead 审批计划
shutdown_request	Lead→队友	请求体面关机
shutdown_approved	队友→Lead	确认关机
shutdown_rejected	队友→Lead	拒绝关机（附原因）
task_assignment	Lead→队友	分配任务
team_permission_update	Lead→队友	广播权限变更
mode_set_request	Lead→队友	修改队友的权限模式
sandbox_permission_*	双向	网络权限请求/回复
teammate_terminated	系统	队友被移除通知

权限冒泡：双向轮询

• 队友遇到需要审批的操作 → 发 permission_request 到 Lead 的收件箱
• Lead 的 useInboxPoller（每 1 秒轮询）检测到请求 → 路由到 ToolUseConfirmQueue
• Lead 的 UI 显示审批对话框，带队友名字和颜色
• 用户审批后 → Lead 发 permission_response 回队友的收件箱
• 队友的 useSwarmPermissionPoller（每 500ms 轮询）收到回复 → 继续或拒绝执行

队友生命周期：

• Spawn：创建 tmux 窗格（或进程内），分配颜色，写入 team config
• Work：useInboxPoller每 1 秒检查收件箱 → 有消息就提交为新的 turn
• Idle：Stop hook 触发 → 发idle_notification给 Lead
• Shutdown：Lead 发shutdown_request→ 队友回复 shutdown_approved→ Lead 清理

Team Config：

{
"name": "my-team",
"leadAgentId": "lead@my-team",
"members": [{
"agentId": "researcher@my-team",
"name": "researcher",
"agentType": "general-purpose",
"color": "blue",
"isActive": true
}]
}

第十一章：Comprehensive Agent – 全部机制归位（s20）

11.1 问题：真实 Agent 不会只带一个机制运行

一个能长期工作的 coding agent 需要同时拥有：工具分发和权限边界、hooks 扩展点、todo 计划和任务图、技能、记忆、系统 prompt 组装、压缩和错误恢复、后台任务和 cron 调度、团队、协议、自治认领、worktree 隔离、MCP 外部工具接入。

难点不是把功能堆起来，而是看清楚它们都挂在循环的哪个位置。

11.2 完整的 Harness 架构

用户输入
→ UserPromptSubmit hooks
→ cron/background 通知注入
→ context compact
→ memory + skills + MCP 状态组装 system prompt
→ LLM
→ has tool_use block?
否 → Stop hooks → 返回
是 → PreToolUse hooks + permission
→ TOOL_HANDLERS / MCP handlers / background dispatch
→ PostToolUse hooks
→ tool_result / task_notification 回 messages
→ 下一轮

11.3 27 个内置工具

bash, read_file, write_file, edit_file, glob
todo_write, task, load_skill, compact
create_task, list_tasks, get_task, claim_task, complete_task
schedule_cron, list_crons, cancel_cron
spawn_teammate, send_message, check_inbox
request_shutdown, request_plan, review_plan
create_worktree, remove_worktree, keep_worktree
connect_mcp

assemble_tool_pool()每轮组装：BUILTIN_TOOLS + connected MCP tools

11.4 组件在循环中的精确位置

位置	组件	作用
用户输入前后	UserPromptSubmit hooks	记录、注入、审计用户输入
LLM 前	cron queue	把定时触发的 prompt 注入 messages
LLM 前	background notifications	后台任务完成后以 <task_notification> 注入
LLM 前	compaction pipeline	先压大输出，再裁历史，再压旧 tool_result，必要时摘要
LLM 前	memory / skills / MCP state	组装 system prompt，让模型看到当前能力和长期上下文
LLM 调用	error recovery	429/529 重试，max_tokens 升级，prompt too long 触发 reactive compact
工具执行前	PreToolUse hooks + permission	拦截危险命令、写越界、破坏性 MCP 工具
工具分发	assemble_tool_pool	组装内置工具和 MCP 动态工具
工具执行时	background dispatch	慢 bash 操作放 daemon thread，主循环先返回占位结果
工具执行后	PostToolUse hooks	大输出告警、日志等后处理
返回循环	tool_result	每个 tool_use 对应一个 tool_result，再回到下一轮
本轮没有 tool_use / 停止时	Stop hooks	统计、清理、审计

11.5 两层计划系统同时存在

• todo_write：当前会话内的轻量计划，保存在内存中
• task graph：跨会话、可依赖、可认领的任务文件，写入.tasks/task_*.json

前者帮助单个 Agent 不漂移；后者支撑团队协作。

11.6 两种 Delegation 同时存在

• task：一次性 subagent。独立messages[]，中间过程丢弃，只返回最终摘要。
• spawn_teammate：持久队友线程。通过 MessageBus 收发消息，能 idle 轮询任务板并自动认领。

一次性 subagent 解决"上下文隔离"；持久队友解决"长期并行协作"。

11.7 完整的 system prompt 组装

def assemble_system_prompt(context):
# 身份和工具说明
# workspace
# skills catalog
# .memory/MEMORY.md
# 已连接 MCP server
...

技能只在 system prompt 里放目录。完整内容通过load_skill(name)按需加载。

11.8 完整的错误恢复

• 429：指数退避重试
• 529：指数退避，连续失败可切 fallback model
• max_tokens：先提高 max_tokens，再要求 continuation
• prompt too long：reactive compact 后重试

11.9 后台与 cron

慢 bash 操作不会阻塞主循环：

should_run_background → start_background_task → placeholder tool_result
后台完成 → task_notification → 下一轮注入 messages

cron 调度器独立 daemon thread 每秒检查一次。CLI 会监听cron_queue，命中后主动把[Scheduled] …注入并运行一轮 Agent。

11.10 worktree 与 MCP

worktree 负责隔离目录：

• create_worktree(name, task_id)创建独立分支和目录
• task 的worktree字段绑定目录
• 队友 claim 到带 worktree 的 task 后，bash/read/write 自动在对应目录下执行

MCP 负责外部能力：

• connect_mcp(name) 连接 mock server
• assemble_tool_pool() 把 MCP 工具组装进工具池
• 工具名统一为 mcp__server__tool

第十二章：范式总结 – 从"用完即弃"到"永远在线"

仓库最后提出了一个极具前瞻性的对比：

learn-claude-code                   claw0
(Agent 套具内部机制:                 (永远在线套具:
循环、工具、规划、                    心跳、定时、IM 通道、
团队、工作树隔离)                     记忆、Soul 人格)

• learn-claude-code 教的是"用完即弃"型驾驭层， 打开终端、给任务、做完关闭、下轮重来。Claude Code 属于此类。
• claw0（OpenClaw）展示了另一种可能：在相同 Agent 核心上，增加Heartbeat（心跳）和 Cron（定时任务）两个驾驭层机制，让 Agent 从"戳一下动一下"变成"每 30 秒自己醒来检查工作"。

加上 IM 多通道路由（WhatsApp/Telegram/Slack/Discord 等 13+ 平台）、持久化上下文记忆、Soul 人格系统，Agent 就从一次性工具变成了永远在线的个人 AI 助手。

第十三章：结语 – 机制很多，循环一个

从 s01 到 s20，代码表面越来越复杂，但核心始终没变：

while True:
response = LLM(messages, tools)
if not has_tool_use(response.content):
return
results = execute_tools(response.content)
messages.append(tool_results)

Claude Code 的复杂性不是"另一个 agent 大脑"，而是一个成熟 harness 的复杂性。模型负责判断和行动选择；harness 负责把环境、工具、权限、记忆、团队和外部能力组织好。

“Bash is all you need. Real agents are all the universe needs.”（Bash 就是你所需要的一切。真正的 Agent 是宇宙所需要的一切。）

“This is not ‘copy the source code.’ This is ‘grasp the key designs and build it yourself.’”（这不是"复制源码"，这是"掌握关键设计，自己构建"。）

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】