Harness工程学习--Learn Claude Code从0到1--(2)

一、第一阶段

1.1 S01 最小智能体（一个工具 + 一个循环）

Agent Loop把工具的结果送回模型继续推理。

1.1.1 TOOLS 定义里只有一个bash工具：

TOOLS = [{ "name": "bash", "description": "Run a shell command.", "input_schema": { "type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"], }, }]

1.1.2 循环代码，后面章节都在这个循环上叠加机制，循环本身始终不变：

while True: # 把消息历史、system prompt 和工具定义一起发给模型 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": # 调用bash output = run_bash(block.input["command"]) results.append({ "type": "tool_result", "tool_use_id": block.id, "content": output, }) # 把工具结果作为新消息追加，继续循环 messages.append({"role": "user", "content": results})

SYSTEM = f"You are a coding agent at {目录}. Use bash to solve tasks. Act, don't explain."

1.1.3 CC源码

1.1.4 测试prompt

a. Create a file called hello.py that prints "Hello, World!"
b. List all Python files in this directory
c. What is the current git branch?

1.2 S02 工具箱

1.2.1 TOOLS 定义了5个工具：bash、read_file、write_file、edit_file、glob

文件工具加了safe_path，只能操作项目目录里的文件，防止智能体越界访问系统文件。

1.2.2 循环：run_bash被替换为查找分发。

results = [] for block in response.content: if block.type == "tool_use": # 查找工具 handler = TOOL_HANDLERS.get(block.name) # 找到工具则调用 output = handler(**block.input) if handler else f"Unknown: {block.name}" results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})

SYSTEM = f"You are a coding agent at {目录}. Use tools to solve tasks. Act, don't explain."

注意：这里的工具是顺序调用的。CC 的做法更复杂：按原始顺序切成连续 batch，batch 内并发安全的工具并行执行，batch 间严格顺序。

1.2.3 CC源码

1.2.4 测试prompt

a. Read the file README.md and tell me what this project is about
b. Create a file called test.py that prints "hello", then read it back
c. Find all Python files in this directory
d. Read both README.md and requirements.txt, then create a summary file

1.3 S03 工具权限

权限系统是为了让 agent 工具调用前先经过一道可靠的安全判断。

1.3.1 循环：工具执行前插入check_permission——每个工具调用经过三道闸门，顺序固定：硬拒绝优先（拒绝列表），软询问次之（规则匹配、用户审批），都没命中就放行。

results = [] for block in response.content: if block.type != "tool_use": continue # s03 change: 工具执行前检查权限 if not check_permission(block): results.append({"type": "tool_result", "tool_use_id": block.id, "content": "Permission denied."}) continue handler = TOOL_HANDLERS.get(block.name) output = handler(**block.input) if handler else f"Unknown: {block.name}" results.append({"type": "tool_result", "tool_use_id": block.id, "content": output})

SYSTEM = f"You are a coding agent at {目录}. All destructive operations require user approval."

1.3.2 CC源码

1.3.3 测试prompt

a. Create a file called test.txt in the current directory（应该直接通过）
b. Delete all temporary files in /tmp（bash + rm 会触发闸门 2）
c. What files are in the current directory?（只读，全部通过）
d. Try to write a file to /etc/something（写工作区外，触发闸门 2）

1.4 S04 工具调用前后的钩子

Hook 是主循环在固定时机对外发出的调用。

1.4.1 循环：把check_permission从循环体内移到了 hook 上，循环不再直接调用任何检查函数，这样扩展行为时，不需要改动循环体代码。

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": # 调用停止钩子 force = trigger_hooks("Stop", messages) if force: messages.append({"role": "user", "content": force}) continue return results = [] for block in response.content: if block.type != "tool_use": continue # s04 change: 调用工具使用前钩子 blocked = trigger_hooks("PreToolUse", block) if blocked: results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(blocked)}) continue handler = TOOL_HANDLERS.get(block.name) output = handler(**block.input) if handler else f"Unknown: {block.name}" # 调用工具使用后钩子 trigger_hooks("PostToolUse", block, output) # s04: post hook results.append({"type": "tool_result", "tool_use_id": block.id, "content": output}) messages.append({"role": "user", "content": results})

SYSTEM = f"You are a coding agent at {目录}. Use tools to solve tasks. Act, don't explain."

1.4.2 CC源码

1.4.3 测试prompt

a. Read the file README.md（应该直接通过，观察 hook 日志）
b. Create a file called test.txt（通过后观察 PostToolUse 是否触发）
c. Delete all temporary files in /tmp（bash + rm 触发权限 hook）

二、第二阶段

2.1 S05 Todos先做计划，防止Agent注意力漂移

2.1.1 循环：工具箱增加了todo_write工具，循环体加了todo_write提醒todo_write本身不做任何实际工作（不能读文件、不能跑命令，只是展示任务每一步的状态）。注意SYSTEM的变化。

global rounds_since_todo while True: # s05: 3轮没调todo_write则注入一条提醒，CC中没有 if rounds_since_todo >= 3 and messages: messages.append({"role": "user", "content": "<reminder>Update your todos.</reminder>"}) rounds_since_todo = 0 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": force = trigger_hooks("Stop", messages) if force: messages.append({"role": "user", "content": force}) continue return rounds_since_todo += 1 results = [] for block in response.content: if block.type != "tool_use": continue blocked = trigger_hooks("PreToolUse", block) if blocked: results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(blocked)}) continue handler = TOOL_HANDLERS.get(block.name) output = handler(**block.input) if handler else f"Unknown: {block.name}" trigger_hooks("PostToolUse", block, output) # s05: 当todo_write被调用，重置轮次计数器 if block.name == "todo_write": rounds_since_todo = 0 results.append({"type": "tool_result", "tool_use_id": block.id, "content": output}) messages.append({"role": "user", "content": results})

SYSTEM = ( f"You are a coding agent at {目录}. " "Before starting any multi-step task, use todo_write to plan your steps. " "Update status as you go." )

2.1.2 CC源码：有两套任务系统并存。

2.1.3 测试prompt

a. Refactor s05_todo_write/example/hello.py: add type hints, docstrings, and a main guard（先列 3 步再执行）
b. Create a Python package under s05_todo_write/example/demo_pkg with __init__.py, utils.py, and tests/test_utils.py
c. Review Python files under s05_todo_write/example and fix any style issues

2.2 S06 子智能体 — 大任务拆小，小任务有独立的上下文

子智能体的核心，不是多一个角色，而是多一个干净上下文。

2.2.1 工具：增加了task工具，用于生成子智能体

TOOLS.append({ "name": "task", "description": "Launch a subagent to handle a complex subtask. Returns only the final conclusion.", "input_schema": {"type": "object", "properties": {"description": {"type": "string"}}, "required": ["description"]}, })

2.2.2 主循环：代码没有变化，和S05一样，但SYSTEM有变化。

SYSTEM = ( f"You are a coding agent at {目录}. " "For complex sub-problems, use the task tool to spawn a subagent." )

2.2.3 子Agent循环

def spawn_subagent(description: str) -> str: """Spawn a subagent with fresh messages[], return summary only.""" messages = [{"role": "user", "content": description}] # 新的messages,上下文隔离 for _ in range(30): # 最多30次循环 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": # Issue 1: subagent保留安全策略 blocked = trigger_hooks("PreToolUse", block) if blocked: results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(blocked)}) continue handler = SUB_HANDLERS.get(block.name) output = handler(**block.input) if handler else f"Unknown: {block.name}" trigger_hooks("PostToolUse", block, output) results.append({"type": "tool_result", "tool_use_id": block.id, "content": output}) messages.append({"role": "user", "content": results}) # Issue 5: fallback if safety limit hit during tool_use result = extract_text(messages[-1]["content"]) if not result: # last message is tool_result, look backwards for assistant text for msg in reversed(messages): if msg["role"] == "assistant": result = extract_text(msg["content"]) if result: break if not result: result = "Subagent stopped after 30 turns without final answer." return result # 只返回最后的文本结论

SUB_SYSTEM = ( f"You are a coding agent at {目录}. " "Complete the task you were given, then return a concise summary. " "Do not delegate further." )

SUB_TOOLS不包含工具task，防止子Agent再生子Agent，禁止递归。

2.2.4 CC源码

2.2.5 测试prompt

a. Use a subtask to find what testing framework this project uses（子 Agent 去读文件，主 Agent 只收结论）
b. Delegate: read all .py files in agents/ and summarize what each one does
c. Use a task to create s06_subagent/example/string_tools.py with a slugify(text: str) function, then verify it from the parent agent

2.3 S08 上下文压缩

上下文压缩的核心，是让模型在更短的活跃上下文里，仍然保住继续工作的连续性。

2.3.1 循环：增加压缩调用 L3(tool_result_budget落盘，保留完整内容) -> L1(snip裁掉无关的旧对话，裁中间，保留头尾)-> L2(micro旧工具结果占位) ->L4(主动调LLM压缩)

reactive_retries = 0 while True: # s08 change: 3个预处理，0 api调用 # 顺序与 CC 相同: budget → snip → micro messages[:] = tool_result_budget(messages) # L3: 大结果落盘 messages[:] = snip_compact(messages) # L1: 裁中间 messages[:] = micro_compact(messages) # L2: 旧工具结果占位 # s08 change: tokens超过阈值 → LLM 摘要(1 API调用) if estimate_size(messages) > CONTEXT_LIMIT: # L4: 调用LLM 压缩 messages[:] = compact_history(messages) try: response = client.messages.create(model=MODEL, system=SYSTEM, messages=messages, tools=TOOLS, max_tokens=8000) reactive_retries = 0 # reset on successful API call except Exception as e: if ("prompt_too_long" in str(e).lower() or "too many tokens" in str(e).lower()) and reactive_retries < MAX_REACTIVE_RETRIES: # 应急压缩 messages[:] = reactive_compact(messages) reactive_retries += 1 continue raise 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": continue # s08: 模型主动调用时触发compact工具 if block.name == "compact": messages[:] = compact_history(messages) results.append({"type": "tool_result", "tool_use_id": block.id, "content": "[Compacted. Conversation history has been summarized.]"}) messages.append({"role": "user", "content": results}) break # 结束当前轮次，用压缩后的上下文开始新一轮 blocked = trigger_hooks("PreToolUse", block) if blocked: results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(blocked)}) continue handler = TOOL_HANDLERS.get(block.name) output = handler(**block.input) if handler else f"Unknown: {block.name}" trigger_hooks("PostToolUse", block, output) results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)}) else: # 正常路径: 无压缩 messages.append({"role": "user", "content": results}) continue # compact was called: results already appended above continue

2.3.2 CC源码

2.3.3 测试prompt

a. Read the file README.md, then read code.py, then read s01_agent_loop/README.md（连续读多个文件，观察 L2 压缩旧结果）
b. Read every file in s08_context_compact/（一次性读大量内容，观察 L3 落盘）
c. 反复对话 20+ 轮，观察是否出现 [auto compact] 或 [reactive compact]