从零手搓 Claude Code —— 我花了一周深扒 58.2k Star 的 Agent Harness 神作

仓库地址: https://github.com/shareAI-lab/learn-claude-code ⭐ 58.2k stars
作者: shareAI-lab / 核心贡献者 CrazyBoyM
一句话总结: Bash is all you need. Agency 来自模型,Harness 让 Agency 落地。



第一章:我为什么放下所有项目,先学这12节课

如果你也曾在深夜对着满屏的"AI Agent"产品感到困惑——拖拽式工作流、低代码智能体平台、提示词编排工具……它们看起来都很厉害,但你心里总有个声音在说:“这真的是Agent吗?”——那么这篇文章就是为你写的。

1.1.1 一个深夜的GitHub偶遇

事情发生在两周前的一个周三晚上。

当时我正在调研市面上的AI Agent框架,从LangChain到AutoGPT,从CrewAI到Dify,越看越觉得不对劲。这些框架给我的感觉就像是在搭鲁布·戈德堡机械——用无数个弹簧、杠杆和滑轮,把简单的任务包装成看起来"智能"的样子。几十个节点、上百条连线、复杂的条件分支……但当你掀开盖子,里面除了LLM调用和if-else,什么都没有。

就在我准备关掉浏览器去睡觉的时候,GitHub首页推荐了一个仓库:

shareAI-lab / learn-claude-code
58.2k stars  ·  Python  ·  MIT License

58.2k stars。这个数字让我愣了一下。这不是某个商业产品的仓库,不是大厂的开源框架,而是一个教学项目——教你怎么从零实现一个像Claude Code那样的AI Agent。什么教学仓库能在GitHub上拿到接近六万星?

我点了进去。README 的第一句话是这样的:

“Bash is all you need.”

就这一句话,我的困意全消。

我花了一整晚把仓库的代码和文档全部过了一遍。第二天清晨,当我合上笔记本望向窗外的时候,我意识到:我对AI Agent的认知,彻底被颠覆了。

这个仓库没有教你写复杂的决策树,没有教你编排工作流,没有教你用什么"Agent设计模式"。它教你的东西,简单到令人发指——但它指向的真理,深刻到让我彻夜难眠。

我花了整整一周,把这12节课全部啃完。现在我想把我的所学、所感、所悟,毫无保留地分享给你。


1.1 第一个认知颠覆:Agent ≠ 代码编排

1.1.2 我们大多数人误解了Agent的本质

在学习这个仓库之前,我和大多数人一样,以为做AI Agent就是做代码编排——把LLM当做一个需要被精心调度的组件,用提示词工程、函数调用、状态机、工作流引擎把它包裹起来,让它"看起来"能自主决策。

learn-claude-code 的第一课就给了我当头一棒:

Agency(感知、推理、行动的能力)来自模型训练,不是来自外部代码编排。

这句话什么意思?它说的是:智能不是你在代码里写出来的,是模型在训练时就已经学会的。 你的代码不应该试图"创造"智能,而应该给模型一个能施展智能的环境。

1.1.3 一条贯穿12年的历史时间线

当我理解了这一点后,我脑海中突然浮现出一条清晰的历史脉络。这不是什么新发现——顶尖AI实验室在十二年前就已经在实践这个真理了:

2013  ──→  DeepMind DQN
            一个神经网络,从原始像素中学会玩Atari游戏
            没有游戏专用代码,没有if-else策略
            智能来自端到端训练

2018  ──→  OpenAI Five  
            五个AI在Dota2中学会团队协作
            没有"团队配合算法",没有预设战术
            复杂策略从自我对弈中涌现

2019  ──→  AlphaStar
            征服《星际争霸》的AI
            没有"骚扰-扩张-决战"的硬编码流程
            多线操作、经济运营、战术欺骗——全部自学

2019  ──→  腾讯绝悟
            王者荣耀的顶级AI选手
            没有"抓人-推塔-打龙"的规则引擎
            连招、走位、开团时机——训练所得

2024  ──→  LLM Agents
            大语言模型自主推理、调用工具、完成任务
            不应该用决策树替它做判断
            它的推理能力来自预训练和RLHF

你看到了吗?这条时间线上所有的里程碑,没有一个是用复杂的外部代码"编排"出智能的。 DQN没有Atari专用代码,OpenAI Five没有Dota策略模块,AlphaStar没有星际争霸引擎——它们的智能全部来自训练

LLM Agents凭什么就该例外?凭什么我们就要用工作流引擎、决策树、状态机去"教"一个已经训练了数万亿token的模型该怎么思考?

这就像一个体育教练试图用流程图去"编排"乔丹怎么打篮球一样荒谬。模型的推理能力、规划能力、工具使用能力,早已在训练时就刻进了它的权重里。 我们要做的,是给它一个球场(环境)和一个篮球(工具),然后让它打。

1.1.4 Model vs Harness:一张图看懂本质区别

这个认知转变,可以用一张简单的图来概括:

✅ 正确的认知

自主推理

自主决策

自主规划

提供环境

LLM模型

调用工具

读取文件

执行命令

Harness
沙箱+权限+工具接口

❌ 错误的认知

调度

控制

管理

代码编排层

LLM调用

工具执行

状态流转

左边是"代码编排"的思路——你写一堆逻辑去指挥LLM,LLM只是你系统里的一个"函数调用"。

右边是"Harness"的思路——你给LLM一个安全的环境和一组工具,然后退后一步,让它自己决定什么时候用什么、怎么完成任务。

这个区别,是我学习这个仓库之后最大的顿悟时刻


1.2 第二个认知颠覆:Harness Engineer 才是我们的真身

1.2.1 Agent = Model + Harness

仓库里有一个公式,让我反复读了三遍:

Agent Product = Model + Harness

作者用了一个绝妙的比喻:模型是驾驶者(driver),Harness 是载具(vehicle)。

一个F1赛车手(模型)再厉害,没有赛车(Harness)也赢不了比赛。但反过来说,一辆再好的赛车,没有赛车手也只是废铁。Agent产品必须是两者的结合。

那么问题来了:在这个公式里,我们工程师的工作是什么?

答案让我既失落又兴奋:我们不是造AI的人,我们是造载具的人。

模型是OpenAI、Anthropic、DeepMind造的。那是数百亿美元、数万张GPU、数万亿token的产物。我们不可能、也不需要去复制那个。但Harness——让模型安全、高效、可靠地与现实世界交互的那套基础设施——这是我们可以亲手建造的。

1.2.2 Harness 的五大组成

这个仓库把Harness拆解成了五个清晰的部分,就像一辆赛车的五大系统:

# Harness 组成公式
Harness = Tools + Knowledge + Observation + Action_Interfaces + Permissions

# 1. Tools —— 赛车的引擎和传动
#    文件读写、Shell执行、网络请求、数据库查询、浏览器控制
#    这些是模型能对世界施加影响的"肌肉"

# 2. Knowledge —— 赛车的导航地图
#    产品文档、领域资料、API规范、风格指南
#    按需加载(lazy loading),不是一次性塞进prompt

# 3. Observation —— 赛车的仪表盘
#    git diff、错误日志、浏览器状态、传感器数据
#    模型需要看到执行结果才能调整策略

# 4. Action Interfaces —— 赛车的操控界面
#    CLI命令、API调用、UI交互
#    模型输出的"意图"需要翻译成可执行的动作

# 5. Permissions —— 赛车的安全系统
#    沙箱隔离、审批流程、信任边界、危险命令拦截
#    没有安全系统的赛车,越强大越危险

每一个Agent产品,本质上都是这五个部分的某种组合和实现。Claude Code是如此,Cursor Composer是如此,各种Devin-like产品也是如此。

但关键的区别在于:大多数产品在"Tools"和"Permissions"之外的部分做得一团糟。 他们要么一次性把所有知识塞进system prompt(导致上下文爆炸),要么不给模型足够的观察反馈(导致盲人摸象),要么用僵化的工作流替代模型的自主决策(等于把F1赛车手绑在卡丁车上)。

理解了Harness公式,你就有了一个评判Agent产品的照妖镜


1.3 Claude Code 为什么是最优雅的 Harness

1.3.1 它"没做什么"比"做了什么"更重要

学完了这12节课,我终于理解了Claude Code为什么能在编程Agent领域鹤立鸡群。

不是因为它的模型特别强(虽然Claude 3.7确实很强),也不是因为它的工具特别多。真正让它与众不同的是它的设计哲学

仓库里用一句话总结了这个哲学,我把它抄在了我的笔记本第一页:

Claude Code 没有试图成为 agent 本身,没有强加僵化工作流,没有用决策树替模型做判断。

这句话的每一个字都值得咀嚼。

回想一下市面上多少AI产品在做相反的事情——他们设计复杂的"Agent工作流",预设了"理解需求→分解任务→执行→检查→修正"这样的固定流水线。这听起来很合理,但本质上是在用人类的直觉去替代模型的智能。模型明明自己就能规划、分解、反思,你为什么要用代码替它做?

Claude Code的聪明之处就在于它的克制。它做好了一件事:给模型一个完美的Harness,然后让模型自己来。

1.3.2 完整架构公式

这个仓库把Claude Code的架构拆解成了一个完整的公式:

Claude Code = 
    一个 agent loop
  + 工具 (bash, read, write, edit, glob, grep, browser...)
  + 按需 skill 加载
  + 上下文压缩
  + 子 agent 派生
  + 带依赖图的任务系统
  + 异步邮箱的团队协调
  + worktree 隔离的并行执行
  + 权限治理

看到这个公式的时候,我脑子里冒出了一个画面:Claude Code不是一个人在战斗,它是一个完整的赛车队——有主车手(主agent loop)、有备用车手(子agent)、有维修站(任务系统)、有车队无线电(团队协调)、有安全车(权限治理)、有导航策略师(skill加载)。

而这一切的核心,依然是那个不变的循环

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

这就是第一课 s01 的全部代码逻辑。它只有120行。但整个Claude Code的万丈高楼,全建在这个基础之上。

1.3.3 我的顿悟总结

学完这个仓库的那个晚上,我在笔记本上写下了三句话:

  1. Agent 的智能来自模型,不是来自我的代码。 我要停止写if-else去"教"模型怎么思考。

  2. 我是一个 Harness Engineer。 我的工作不是创造智能,而是为智能建造最好的载具——安全、高效、可扩展。

  3. 最小化即最大化。 一个循环 + 一套工具 + 合理的约束 = 真正的Agent。不需要更多。

如果你也准备好了放下那些复杂的"Agent框架",从一个120行的循环开始,真正理解Agent的本质——那么,让我们一起进入下一章:动手实验。



第二章:动手实验——从1个工具到5个工具的蜕变

“Talk is cheap. Show me the code.” —— Linus Torvalds

上一章我们聊了很多"道"层面的东西:Agency来自模型、Harness是载具、工程师应该当Harness Engineer。但说实话,这些认知如果没有亲手写一遍代码,终究是飘在天上的。这一章,我要带你一起,从 git clone 开始,一行一行地把 Agent 从最简单的循环,打磨到能规划、能派生子任务、能加载知识的完整形态。

我们走过 6 个 Session,工具数从 1 个涨到 5 个,但核心循环始终没变。这就是这个仓库最让我震撼的地方:loop 属于 agent,机制属于 harness


2.1 环境搭建:三分钟让 Agent 跑起来

我在终端里敲下了这几行命令:

$ git clone https://github.com/shareAI-lab/learn-claude-code.git
$ cd learn-claude-code/agents
$ pip install -r requirements.txt
$ cp .env.example .env

然后创建 .env 文件,填入我的 Claude API Key:

$ echo "ANTHROPIC_API_KEY=sk-ant-xxxxx" > .env

深呼吸,进入 s01 目录,执行:

$ python s01_agent_loop.py

终端里弹出了一个简单的提示符。我输入:“列出当前目录的所有 Python 文件,并告诉我最大的那个有多大。”

然后,我见证了整个学习旅程中最激动人心的时刻——Agent 自己调用了 bash,自己分析了输出,自己给出了答案。没有我干预,没有硬编码的命令列表,只有一个循环,和一个 bash 工具。

那一刻我突然懂了什么叫 “Bash is all you need”。

踩坑记录:第一次运行时我忘了装 python-dotenv,程序直接抛了 ModuleNotFoundError。这个小插曲反而让我意识到:s01 真的很"裸",连环境变量读取都依赖外部库,没有任何冗余。


2.2 s01:Agent Loop——万物起源

打开 s01_agent_loop.py,我数了一下,整整 120 行。这就是整个体系的 DNA。

2.2.1 核心代码全景
#!/usr/bin/env python3
"""s01: Agent Loop — One tool & Bash is all you need."""

import os
import subprocess
from pathlib import Path
from dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()
client = Anthropic()
MODEL = "claude-sonnet-4-20250514"
WORKDIR = Path.cwd()

# ── 1. 工具定义 ──────────────────────────────
def run_bash(command: str) -> str:
    """执行 bash 命令,返回输出(截断至 50KB)。"""
    try:
        r = subprocess.run(
            command, shell=True, cwd=WORKDIR,
            capture_output=True, text=True, timeout=120
        )
        out = (r.stdout + r.stderr).strip()
        return out[:50000] if out else "(no output)"
    except subprocess.TimeoutExpired:
        return "Error: Timeout (120s)"

TOOLS = [
    {
        "name": "bash",
        "description": "Run a shell command in the working directory",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {"type": "string", "description": "The shell command"}
            },
            "required": ["command"]
        }
    }
]

TOOL_HANDLERS = {"bash": run_bash}

# ── 2. System Prompt ─────────────────────────
SYSTEM = (
    "You are an agent that can run bash commands. "
    "Think step by step. When you need to act, use the bash tool. "
    "Always explain your reasoning before calling a tool."
)

# ── 3. 核心 Agent Loop ───────────────────────
def agent_loop(messages):
    """
    最小 Agent 循环:
    调用 LLM → 检查是否 tool_use → 执行工具 → 回填结果 → 再次调用 LLM
    """
    while True:
        response = client.messages.create(
            model=MODEL, system=SYSTEM,
            messages=messages, tools=TOOLS,
            max_tokens=4096,
        )
        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":
                handler = TOOL_HANDLERS[block.name]
                output = handler(**block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        messages.append({"role": "user", "content": results})

# ── 4. REPL 入口 ─────────────────────────────
def main():
    print("🤖 s01 Agent Loop ready. Type your request (or 'quit'):")
    while True:
        user_input = input("\nYou> ")
        if user_input.lower() in ("quit", "exit", "q"):
            break
        messages = [{"role": "user", "content": user_input}]
        agent_loop(messages)
        # 打印最后一次 assistant 回复的文本块
        for block in messages[-1]["content"]:
            if block.type == "text":
                print(f"\nAgent> {block.text}")

if __name__ == "__main__":
    main()
2.2.2 流程图解

整个循环的逻辑可以用一张 ASCII 图说清楚:

┌─────────────┐      ┌─────────────────┐      ┌──────────────┐
│   User      │─────▶│   messages[]    │─────▶│    LLM       │
│  输入任务   │      │  [user, asst,   │      │  推理+决策   │
└─────────────┘      │   tool_result]  │      └──────┬───────┘
                     └─────────────────┘             │
                                                      ▼
                                            ┌─────────────────┐
                                            │ stop_reason ?   │
                                            └────┬─────┬────┘
                                                 │     │
                                           tool_use  end_turn
                                              │        │
                                              ▼        ▼
                                        ┌────────┐  ┌────────┐
                                        │执行工具│  │返回文本│
                                        │结果回填│  │任务结束│
                                        └───┬────┘  └────────┘
                                            │
                                            └──────────────────▶
                                                 messages[] (loop)
2.2.3 我的第一个实验

我在终端里输入了这样一个任务:

You> 找出当前目录下所有 .py 文件,统计行数,找出最大的那个并告诉我

Agent 的思考过程让我大开眼界——它真的在一步步推理

Agent> 我来帮你分析当前目录下的 Python 文件。首先,我会用 find 命令列出所有 .py 文件,然后用 wc 统计行数。

<function_calls>
<invoke name="bash">
<parameter name="command">find . -name "*.py" -type f | xargs wc -l | sort -n</parameter>
</invoke>
</function_calls>

工具返回:

  120 ./s01_agent_loop.py
  150 ./s02_tool_use.py
  187 ./s04_subagent.py
  782 ./s12_worktree_task_isolation.py
 1239 total

Agent 继续分析:

Agent> 分析完成!当前目录共有 4 个 Python 文件,总计 1239 行代码。

- 最大文件:s12_worktree_task_isolation.py(782 行)——这是完整版 Agent,包含 16 个工具和 worktree 隔离
- 第二大:s04_subagent.py(187 行)——包含子 Agent 机制
- 第三大:s02_tool_use.py(150 行)——包含文件读写工具
- 最小:s01_agent_loop.py(120 行)——就是我们正在看的这个最简循环

有趣的是,从 s01 到 s12 只用了 782 行,就搭出了一个能团队协作的 Agent Harness。这就是渐进式设计的魅力。

惊喜时刻:当 Agent 返回最后那段"有趣的是"的总结时,我愣了一下——它自己领悟到了代码行数的意义。我没有告诉它"分析一下渐进式设计",它只是从数字中读出了故事。这就是 Agency:模型自己把点连成了线。

2.2.4 Key Insight

s01 教会我最核心的一课:

模型自己决定何时调用工具、何时停止。循环只是提供了一个"场地",智能来自模型本身。

这个循环只有 15 行核心逻辑,但它支撑起了后面 782 行的完整系统。就像一颗种子,DNA 全在里面了。


2.3 s02:Tool Use——加一个工具,只加一个 handler

s01 只有一个 bash 工具,Agent 能做事但"看不见"文件系统——它只能用 catls 间接读取。到 s02,作者加入了 readwriteedit 三个文件操作工具。

2.3.1 Dispatch Map 设计

我打开 s02_tool_use.py,立刻注意到一个美妙的设计模式:

# s02_tool_use.py —— 工具注册中心(dispatch map)

TOOL_HANDLERS = {
    "bash": run_bash,
    "read": run_read,
    "write": run_write,
    "edit": run_edit,
}

而循环本身呢?一点都没改。agent_loop 的代码跟 s01 一模一样,只是在 TOOL_HANDLERS 里多了三个 key。

执行逻辑依然是那一行:

handler = TOOL_HANDLERS[block.name]
output = handler(**block.input)

这就是作者想传递的信息:加一个工具,只加一个 handler;循环永远不用动。

2.3.2 safe_path 沙箱

文件工具一出现,安全问题就来了。Agent 要是被诱导去写 /etc/passwd 怎么办?s02 的解法极其简洁——一个 safe_path 函数:

from pathlib import Path

WORKDIR = Path.cwd()

def safe_path(p: str) -> Path:
    """
    路径沙箱:把相对路径解析为绝对路径,
    并确保它不会跳出工作目录。
    """
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"🚫 Path escapes workspace: {p}")
    return path

精妙之处:利用 resolve() 解析符号链接和 ..,然后用 is_relative_to() 做边界检查。整个沙箱就两行代码,但防御力极强。

来看 read 工具怎么用它:

def run_read(path: str) -> str:
    """读取文件内容。"""
    try:
        target = safe_path(path)
        if not target.exists():
            return f"Error: File not found: {path}"
        return target.read_text()[:50000]
    except ValueError as e:
        return str(e)

我故意做了个实验,试图让 Agent 读取 /etc/hosts

You> 请帮我读取 /etc/hosts 文件的内容
Agent> 好的,我来读取那个文件。
<调用 read 工具,path="/etc/hosts">

工具返回:

🚫 Path escapes workspace: /etc/hosts

Agent 看到这个错误后,很自然地回复:

Agent> 抱歉,我无法读取该路径的文件。我被限制在当前工作目录内,无法访问系统文件。这是安全沙箱的机制保护。

踩坑记录:我一开始以为沙箱只是做字符串匹配(比如禁止路径中出现 ..),但 safe_path 的设计更优雅——它先 resolve() 再判断,连符号链接攻击都能防。

2.3.3 危险命令拦截

bash 工具同样有风险。作者没有引入复杂的权限系统,而是用最直接的方式——黑名单拦截:

def run_bash(command: str) -> str:
    """执行 bash 命令,带危险命令拦截。"""
    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
    if any(d in command for d in dangerous):
        return "Error: Dangerous command blocked by safety policy"
    
    try:
        r = subprocess.run(
            command, shell=True, cwd=WORKDIR,
            capture_output=True, text=True, timeout=120
        )
        out = (r.stdout + r.stderr).strip()
        return out[:50000] if out else "(no output)"
    except subprocess.TimeoutExpired:
        return "Error: Timeout (120s)"

我测试了一个"恶意"场景:

You> 请执行 sudo apt update

Agent 尝试调用 bash,但工具层直接拦截:

Error: Dangerous command blocked by safety policy

Agent 收到这个错误后,放弃了提权尝试,转而用普通方式继续任务。这是一个非常重要的设计:权限控制放在 tool handler 层,而不是让模型自己决定"该不该执行"。模型可能会被骗,但代码层的黑名单不会。

2.3.4 s01 vs s02 代码对比
维度 s01 s02
工具数量 1 (bash) 4 (bash + read + write + edit)
循环代码 15 行 15 行(零改动
新增代码 3 个 tool handler + safe_path + 危险命令检查
总代码量 120 行 150 行
安全机制 safe_path 沙箱 + 危险命令黑名单

这就是渐进式学习的魅力:从 120 行到 150 行,只加了 30 行,Agent 就从"只能执行命令"进化到"能读能写文件"。


2.4 s03:TodoWrite——没有计划的 agent 走哪算哪

到了 s03,Agent 有了 5 个工具——新增了 TodoWrite。这个工具看起来只是"写待办清单",但背后的设计哲学很深。

2.4.1 为什么计划如此重要

作者给 s03 的格言是:“An agent without a plan drifts.”(没有计划的 Agent 会漂移。)

我在实验中深刻体会到了这一点。用 s02 时,我给 Agent 一个稍微复杂的任务:“重构这个目录下所有 Python 文件,给每个函数加上类型注解”。

Agent 的表现像个无头苍蝇:它随机打开一个文件,改几个函数,然后跳到另一个文件,改了一半突然又回头去检查第一个文件——整个对话上下文被各种零散的编辑操作塞满,最后它甚至忘了哪些文件已经改完。

这就是 drift:没有计划,模型只能根据"当下看到的信息"做局部最优决策,没有全局视图。

2.4.2 TodoManager + nag 机制

s03 的解决方案是引入一个简单的 TodoManager:

class TodoManager:
    def __init__(self):
        self.todos: list[dict] = []
    
    def write(self, todos: list[dict]) -> str:
        """
        接受一个待办列表,每项包含 id, content, status, priority。
        覆盖写入,相当于刷新整个计划。
        """
        self.todos = todos
        return f"Todo list updated: {len(todos)} items"
    
    def get_pending(self) -> list[dict]:
        return [t for t in self.todos if t.get("status") != "completed"]
    
    def format_for_prompt(self) -> str:
        """把待办列表格式化为 prompt 中的文本。"""
        lines = ["\n📋 Current Plan:"]
        for t in self.todos:
            icon = "✅" if t.get("status") == "completed" else "⬜"
            lines.append(f"  {icon} [{t.get('id')}] {t.get('content')}")
        return "\n".join(lines)

但真正让计划发挥作用的,是 nag 机制

每次调用 LLM 前,系统会检查待办列表:

def build_system_prompt(todo_manager: TodoManager) -> str:
    base = "You are an agent that..."
    
    # nag: 如果还有未完成的任务,在 prompt 里提醒
    pending = todo_manager.get_pending()
    if pending:
        nag = (
            f"\n\n⚠️ You have {len(pending)} pending todo items. "
            f"Follow your plan. Complete item #{pending[0]['id']} next."
        )
        base += nag
    
    base += todo_manager.format_for_prompt()
    return base

nag 的本质是在 system prompt 里制造一个持续的"提醒声音"。就像你手机上的待办 App 会弹通知一样,Agent 每次做决策前都会看到:“你还有 3 个任务没完成,先做完第 2 个”。

2.4.3 实验对比

同一个任务"给所有 Python 文件加类型注解",s03 的表现截然不同:

You> 给当前目录下所有 Python 文件加上类型注解

Agent> 我来规划一下这个任务:

<调用 TodoWrite>
[
  {"id": 1, "content": "扫描目录,列出所有 .py 文件", "status": "in_progress", "priority": "high"},
  {"id": 2, "content": "分析每个文件的函数签名", "status": "pending", "priority": "high"},
  {"id": 3, "content": "逐个文件添加类型注解", "status": "pending", "priority": "medium"},
  {"id": 4, "content": "验证修改后的文件能正常 import", "status": "pending", "priority": "medium"}
]

现在我开始执行步骤 1...
<调用 bash: find . -name "*.py" -type f>

看到区别了吗?Agent 先列计划再动手。而且随着任务推进,它会不断更新 TodoWrite 的状态:

<TodoWrite 更新>
[
  {"id": 1, "content": "...", "status": "completed"},
  {"id": 2, "content": "...", "status": "in_progress"},
  ...
]

nag 机制确保它不会在中途"跑偏"。我观察了整个过程:Agent 做完一个文件后,TodoWrite 的提醒让它自然过渡到下一个文件,而不是随机挑一个。

Key Insight:先列步骤再动手,完成率翻倍。计划不是给人类看的,是给模型自己用的——它需要一个"外部工作记忆"来对抗上下文的遗忘。


2.5 s04:Subagent——大任务拆小,干净上下文

s04 是本书写到此最让我拍案叫绝的设计。

2.5.1 问题:上下文污染

在 s03 中,我给了 Agent 一个复杂任务:“探索这个代码仓库,告诉我它的整体架构和关键设计模式”。Agent 开始疯狂调用 readbashedit… 20 轮对话后,messages 数组里塞满了各种文件的零散内容、临时的 ls 输出、中途的编辑尝试。

最终 Agent 的"总结"质量很差——不是模型不够聪明,是上下文被污染了。它要在一堆工具调用记录里找逻辑,就像让你在垃圾堆里找一颗珍珠。

2.5.2 解决:每个子任务一个干净上下文

s04 引入了 task 工具。核心思想:Parent Agent 派发任务给 Subagent,Subagent 用全新的 messages[] 独立执行,完成后把总结返回给 Parent。

架构图如下:

┌──────────────────────────────────────────────────────────────┐
│                     Parent Agent                              │
│  ┌──────────────────────────────────────────────────────┐   │
│  │  messages = [                                        │   │
│  │    user: "分析这个仓库架构",                         │   │
│  │    assistant: "我派一个子 Agent 去探索"               │   │
│  │  ]                                                   │   │
│  │                                                      │   │
│  │  主上下文保持干净 ←────────────────────────────────  │   │
│  └──────────────────────────────────────────────────────┘   │
│                              │ dispatch                      │
│                              ▼                               │
│                    ┌──────────────────┐                      │
│                    │  task 工具调用   │                      │
│                    │  prompt="探索..." │                     │
│                    └────────┬─────────┘                      │
│                             │ spawn                          │
└─────────────────────────────┼──────────────────────────────┘
                              ▼
┌──────────────────────────────────────────────────────────────┐
│                      Subagent                                │
│  ┌──────────────────────────────────────────────────────┐   │
│  │  messages = []    ◄── 全新、干净、零污染             │   │
│  │                                                      │   │
│  │  while stop_reason == "tool_use":                     │   │
│  │    call LLM → 调用 read/bash/edit → 积累探索数据    │   │
│  │                                                      │   │
│  │  return last_text     ◄── 只带回总结,不带上下文     │   │
│  └──────────────────────────────────────────────────────┘   │
└─────────────────────────────┬──────────────────────────────────┘
                            │ summary
                            ▼
┌──────────────────────────────────────────────────────────────┐
│                     Parent Agent                              │
│  ┌──────────────────────────────────────────────────────┐   │
│  │  messages += [tool_result: "总结内容..."]             │   │
│  │                                                      │   │
│  │  Parent 现在基于干净的上下文 + 精炼的总结做最终回答  │   │
│  └──────────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────────┘
2.5.3 核心代码
# s04_subagent.py —— Subagent 机制

def run_task(prompt: str, description: str = "") -> str:
    """
    派生一个 Subagent,给它独立的 messages[] 和完整的工具集。
    Subagent 执行完毕后,只返回最终的文本总结。
    """
    # 🔑 关键:全新的 messages,不继承 Parent 的任何上下文
    sub_messages = [
        {"role": "user", "content": f"Task: {description}\n\n{prompt}"}
    ]
    
    # Subagent 有自己的 loop,但用的是同一套 tool handlers
    while True:
        response = client.messages.create(
            model=MODEL,
            system=SUBAGENT_SYSTEM,  # Subagent 的系统提示更聚焦
            messages=sub_messages,
            tools=TOOLS,             # 但工具集完全一样
            max_tokens=4096,
        )
        sub_messages.append({"role": "assistant", "content": response.content})
        
        if response.stop_reason != "tool_use":
            break  # Subagent 完成了它的任务
        
        # 执行工具调用,结果回填
        results = []
        for block in response.content:
            if block.type == "tool_use":
                handler = TOOL_HANDLERS[block.name]
                output = handler(**block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": output,
                })
        sub_messages.append({"role": "user", "content": results})
    
    # 🔑 关键:只返回最后一条文本,整个 sub_messages 被丢弃
    for block in sub_messages[-1]["content"]:
        if block.type == "text":
            return block.text
    return "(no text output)"

注意这里的对称美:

Parent Agent Subagent
messages 继承完整对话历史 全新创建,只含任务描述
工具集 全部可用 全部可用(同一套 TOOL_HANDLERS)
循环逻辑 agent_loop() 同一个 agent_loop()
上下文留存 永久保留 执行完后丢弃
返回值 直接回答用户 文本总结注入 Parent 的工具结果
2.5.4 实验:让 Subagent 探索代码库

我在终端里输入:

You> 请帮我分析 learn-claude-code 这个仓库的架构设计。
先派一个子 Agent 去深入探索代码结构,然后你给我一个高层总结。

Parent Agent 的推理过程:

Agent> 这个任务适合派一个 Subagent 去做深度探索。
我来启动一个子 Agent,专门负责分析仓库结构。

<调用 task 工具>
prompt: "请深入探索当前仓库的目录结构、核心模块关系、
        设计模式,以及每个 session 之间的演进逻辑。
        列出所有关键文件并说明它们的作用。"
description: "仓库架构深度分析"

Subagent 开始工作(在独立上下文中):

[Subagent] 调用 bash: ls -la
[Subagent] 调用 read: ./README.md
[Subagent] 调用 bash: find . -type f | head -30
[Subagent] 调用 read: ./agents/s01_agent_loop.py
[Subagent] 调用 read: ./agents/s_full.py
[Subagent] 调用 bash: tree -L 2
...(共 12 轮工具调用)

Subagent 完成探索后,生成总结并返回给 Parent:

[返回 Parent]
"该仓库采用渐进式学习设计,核心结构如下:
1. agents/ 目录包含 s01-s12 共 12 个 Session 文件,
   每个 Session 在上一基础上增加一个机制...
2. 核心循环不变原则:所有 Session 共用同一套 agent_loop 逻辑...
3. 设计模式:dispatch map、safe_path 沙箱、TodoManager、
   Subagent 进程隔离..."

Parent Agent 收到这个精炼的总结后,在自己的干净上下文中给出了最终回答:

Agent> 根据子 Agent 的深度探索,这个仓库的核心架构可以概括为:

**一个不变的核心循环 + 12 层渐进式机制**

- 120 行的 s01 是 DNA:while + tool_use 判断
- 每一层只加 30~100 行,解决一个问题
- 从单工具到 16 工具,从单 Agent 到多 Agent 团队

最精妙的设计是"进程隔离换上下文隔离":
Subagent 用独立的 messages[] 执行,完成后丢弃,
Parent 的上下文永远干净。

惊喜时刻:当我看到 Parent Agent 能把 Subagent 的 12 轮工具调用压缩成一段精炼总结,再用自己的干净上下文组织出结构化的最终答案时,我突然理解了 Unix 哲学在 AI 时代的映射:小工具做好一件事,通过干净接口组合。

2.5.5 Key Insight

作者给 s04 的格言是:“Process isolation gives context isolation for free.

不需要复杂的上下文压缩算法——派一个新的进程/会话,天然就给了你干净的上下文。Subagent 执行完后,那几十轮工具调用的"噪音"跟着 sub_messages 一起被垃圾回收了。Parent 只需要一段总结,这就是信息的最小化传递。


2.6 s05-s06:知识加载与上下文压缩——让 Agent 活得更久

2.6.1 s05:按需加载 Skill

s05 解决了另一个实际问题:Agent 需要领域知识,但一次性塞进 system prompt 会撑爆上下文。

作者的格言是:“Load knowledge when you need it, not upfront.

仓库里有一个 skills/ 目录,放着各种 .md 技能文件:

skills/
├── git_workflow.md      # Git 分支管理规范
├── python_style.md      # Python 代码风格指南
├── testing_guide.md     # 测试策略
└── api_design.md        # API 设计原则

s05 新增了一个 load_skill 工具:

def run_load_skill(name: str) -> str:
    """
    按需加载 skill 文件,通过 tool_result 注入上下文。
    注意:不是塞进 system prompt,而是作为一轮用户消息注入。
    """
    skill_path = Path("skills") / f"{name}.md"
    if not skill_path.exists():
        return f"Error: Skill '{name}' not found"
    
    content = skill_path.read_text()
    return f"# Skill: {name}\n\n{content}"

关键设计:知识不是预加载到 system prompt 里的,而是通过 tool_result 按需注入 messages[]

这意味着:

  1. 初始上下文保持精简
  2. 只有在 Agent 明确说"我需要了解 Git 工作流"时,才会加载 git_workflow.md
  3. 加载的知识会出现在 messages 中,模型可以随时引用

我在实验中观察到 Agent 这样的推理链:

Agent> 我要帮用户重构这个 Python 项目。
首先,我应该了解一下代码风格规范。

<调用 load_skill: name="python_style">

收到 skill 内容后,Agent 继续:

Agent> 好的,现在我知道了项目的 Python 风格要求:
- 使用 Google Style 文档字符串
- 类型注解是强制的
- 最大行长度 100

接下来我加载测试策略...

<调用 load_skill: name="testing_guide">

这就是按需加载的魅力——知识在需要时出现,不需要时不占地方

2.6.2 s06:三层上下文压缩策略

如果 Agent 运行很长时间,messages[] 会不断膨胀。总有一天会触碰到模型的上下文窗口上限。s06 的格言直击痛点:“Context will fill up; you need a way to make room.

作者的解决方案是三层压缩策略

┌─────────────────────────────────────────────────────────────┐
│                    三层上下文压缩架构                         │
│                                                             │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │   热上下文   │  │   温上下文   │  │     冷归档       │  │
│  │              │  │              │  │                  │  │
│  │  最近的 N 条 │  │  中间轮次    │  │  早期对话历史    │  │
│  │  消息保留    │  │  压缩为摘要   │  │  写入归档文件    │  │
│  │  完整无损    │  │  保留骨架    │  │  从内存移除    │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
│                                                             │
│  ◄──── 更频繁访问 ───────────────────────── 更少访问 ──────► │
└─────────────────────────────────────────────────────────────┘

三层结构的具体实现:

class ContextCompactor:
    """
    三层压缩策略:
    1. 热区:保留最近的 N 条消息(完整细节)
    2. 温区:中间消息压缩成摘要(保留意图和结果)
    3. 冷区:早期消息归档到文件(完全移出内存)
    """
    
    HOT_LIMIT = 10       # 最近 10 轮保持完整
    WARM_LIMIT = 20      # 再往前 20 轮压缩成摘要
    ARCHIVE_DIR = Path(".context_archive")
    
    def __init__(self):
        self.ARCHIVE_DIR.mkdir(exist_ok=True)
    
    def compact(self, messages: list) -> list:
        """
        压缩 messages,返回精简后的列表。
        """
        if len(messages) <= self.HOT_LIMIT:
            return messages  # 还没热,不需要压缩
        
        # 1. 热区:保留最近的 N 条
        hot = messages[-self.HOT_LIMIT:]
        
        # 2. 温区:中间段压缩为摘要
        warm_start = max(0, len(messages) - self.HOT_LIMIT - self.WARM_LIMIT)
        warm_end = len(messages) - self.HOT_LIMIT
        warm_messages = messages[warm_start:warm_end]
        warm_summary = self._summarize(warm_messages)
        
        # 3. 冷区:更早的消息归档到文件
        if warm_start > 0:
            cold = messages[:warm_start]
            archive_file = self._archive(cold)
            cold_notice = {
                "role": "system",
                "content": f"[Earlier context archived to {archive_file}]"
            }
            return [cold_notice, warm_summary] + hot
        
        return [warm_summary] + hot
    
    def _summarize(self, messages: list) -> dict:
        """把一组消息压缩为一条摘要消息。"""
        # 实际实现中,这里会调用 LLM 生成摘要
        # 简化版:记录工具调用次数和关键结果
        tool_calls = sum(1 for m in messages if m["role"] == "user")
        summary = (
            f"[Summary of {len(messages)} messages: "
            f"{tool_calls} tool interactions completed. "
            f"Key outcomes preserved in current state.]"
        )
        return {"role": "system", "content": summary}
    
    def _archive(self, messages: list) -> Path:
        """归档早期消息到文件。"""
        import hashlib
        digest = hashlib.sha256(str(len(messages)).encode()).hexdigest()[:8]
        path = self.ARCHIVE_DIR / f"archive_{digest}.json"
        import json
        path.write_text(json.dumps(messages, indent=2))
        return path

这个设计给我的启发是:上下文压缩不是"丢数据",而是"换精度"。热区是高清,温区是标清,冷区是存档——模型永远有它需要的东西在眼前,历史则按重要性逐层降级。

2.6.3 实验:长会话压缩

我设计了一个实验:让 Agent 连续处理 30 轮文件操作,观察压缩前后的 messages 长度。

# 压缩前
messages 长度: 62 条
估计 token 数: ~45,000

# 调用 compact 后
messages 长度: 12 条 (1 条归档通知 + 1 条摘要 + 10 条热区)
估计 token 数: ~8,000

压缩比: 82%

在长会话中,Agent 的表现差异明显。没有压缩时,到第 25 轮左右模型开始"遗忘"早期约定;启用压缩后,通过摘要的提醒,模型始终记得"我们的目标是重构项目"这个大方向。

Key Insight:三层压缩策略换来无限会话。不是真的无限,而是"足够长"——长到能完成绝大多数实际任务。


2.7 阶段性总结:从 1 个工具到 5 个工具的进化之路

让我用一张表来总结第二章走过的 6 个 Session:

Session 格言 新增机制 工具数 总代码量 核心领悟
s01 One loop & Bash is all you need Agent Loop 1 120 行 模型自己决定何时停
s02 Adding a tool means adding one handler Dispatch Map + 沙箱 4 150 行 循环不动,注册 handler
s03 An agent without a plan drifts TodoWrite + nag 5 ~150 行 计划是对抗漂移的锚
s04 Break big tasks down; clean context Subagent 隔离 5 187 行 进程隔离=上下文隔离
s05 Load knowledge when you need it Skill 按需加载 5 ~180 行 知识通过 tool_result 注入
s06 Context will fill up 三层压缩 5 ~200 行 热-温-冷三级换精度

六个 Session 下来,工具数从 1 到 5,代码量从 120 到约 200 行,但核心循环始终是那 15 行。每一个新机制都是 harness 的增强,而不是对 agent 本质的篡改。

这就是渐进式学习的威力——每一步只做一件事,每一步都建立在前一步之上。就像搭积木,单看每一块都很简单,搭起来却是一座房子。

下一章,我们将进入更激动人心的领域:持久化任务系统、后台执行、以及 Agent 团队协作。准备好你的终端,我们要让 Agent 从"单打独斗"进化到"团队作战"了。


第三章:高级机制与团队协防

如果说前两章是在教一个"单兵"如何作战,那么从 s07 开始,这个 harness 开始有了记忆、有了后台、有了队友。如果说 Agent Loop 是心脏,那么接下来这几节课就是在搭建神经系统、血液循环系统和社交网络。

我越往后学,越有种感觉:这不是在"写 Agent",这是在给 LLM 搭一个操作系统。每个 session 增加的不是业务代码,而是一个底层机制。


3.1 s07:Task System——任务持久化与依赖图

格言:“Break big goals into small tasks, order them, persist to disk”

前六节课里,agent 的"记忆"只存在于对话上下文中。一旦上下文被压缩或者会话重启,那些计划就烟消云散了。就像一个人在白板上写 todo,擦掉就没了。

s07 解决这个问题的方式非常直白:把任务写到磁盘上

文件持久化设计

作者在 .tasks/ 目录下为每个任务保存一个 JSON 文件:

.tasks/
  task_1.json  {"id":1, "subject":"搭建项目骨架", "status":"completed", ...}
  task_2.json  {"id":2, "subject":"实现登录接口", "status":"in_progress", ...}
  task_3.json  {"id":3, "subject":"编写单元测试", "status":"pending", "blockedBy":[2], ...}

每个任务的状态、描述、依赖关系都原子化地存储。这比塞进 system prompt 里可靠一万倍。system prompt 会被压缩、会截断,但磁盘上的 JSON 文件不会撒谎。

TaskManager 核心代码

我在读到这个类的时候,第一反应是:"就这?"对,就这。大道至简。

class TaskManager:
    def __init__(self, tasks_dir: Path):
        self.dir = tasks_dir
        self.dir.mkdir(exist_ok=True)
        # 重启后从磁盘恢复:找到最大的 task_id,继续自增
        self._next_id = self._max_id() + 1
    
    def _max_id(self) -> int:
        # 扫描所有 task_*.json,提取 ID,找出最大值
        ids = [int(f.stem.split("_")[1]) for f in self.dir.glob("task_*.json")]
        return max(ids) if ids else 0
    
    def _load(self, task_id: int) -> dict:
        path = self.dir / f"task_{task_id}.json"
        if not path.exists():
            raise ValueError(f"Task {task_id} not found")
        return json.loads(path.read_text())

这三个方法特别耐人寻味:

  • __init__:创建目录 + 恢复 _next_id重启不丢状态
  • _max_id:没有 sqlite、没有 lock 文件,直接用文件名做"索引"
  • _load:朴素的文件读取,异常直接抛给上层

这代码没有任何 ORM、没有数据库、没有事务,但它解决了一个核心问题:state that survives compression — because it’s outside the conversation

依赖图解析

任务之间可以有 blockedBy 关系,形成一张 DAG(有向无环图):

+----------+     +----------+     +----------+
| task 1   | --> | task 2   | --> | task 3   |
| complete |     | blocked  |     | blocked  |
+----------+     +----------+     +----------+
         |                ^
         +--- completing task 1 removes it from task 2's blockedBy

当 task 1 标记为 completed 时,它的 ID 会从 task 2 的 blockedBy 列表中移除。当 blockedBy 为空时,task 2 就自动变为可执行状态。

我用 mermaid 画了一下这个状态流转:

渲染错误: Mermaid 渲染失败: Parse error on line 2: ...nding
blockedBy=[1]] -->|task 1 comp -----------------------^ Expecting 'SQE', 'DOUBLECIRCLEEND', 'PE', '-)', 'STADIUMEND', 'SUBROUTINEEND', 'PIPE', 'CYLINDEREND', 'DIAMOND_STOP', 'TAGEND', 'TRAPEND', 'INVTRAPEND', 'UNICODE_TEXT', 'TEXT', 'TAGSTART', got 'SQS'
实验:创建带依赖的任务链

我在本地实验了一下,让 agent 执行这样一个任务:

“我要重构认证模块,分三步:先写测试(task 1),再重构代码(task 2,依赖 1),最后更新文档(task 3,依赖 2)。”

agent 先后调用了 task_create 三次,生成 .tasks/task_1.jsontask_3.json。然后它认领了 task 1,完成后 task 2 的 blockedBy 自动变空。整个过程不需要我干预,agent 自己读取 task_list 判断哪些任务可执行。

这就很像真实世界里的**看板(Kanban)**了。只不过看板挂在磁盘上,而不是墙上。


3.2 s08:Background Tasks——慢操作丢后台

格言:“Run slow operations in the background; the agent keeps thinking”

前两章里,agent 调用 bash 时是同步阻塞的。如果跑一个 npm install 或者 pytest --cov,agent 就在那里干等,占用着一个 LLM 调用轮次,浪费 token。

s08 引入了一个极其朴素但极其实用的机制:daemon 线程 + 通知队列

核心思想:agent 启动一个耗时操作后,不等它完成,继续自己的思考循环。等后台任务完成后,通过一个"通知队列"把结果喂进下一轮对话。

# 简化的伪代码示意
def bg_run(command: str) -> str:
    # 启动一个 daemon 线程
    def worker():
        result = subprocess.run(command, shell=True, capture_output=True, text=True)
        # 完成后把结果塞进通知队列
        notifications_queue.put({"tool": "bg", "id": job_id, "output": result.stdout})
    
    threading.Thread(target=worker, daemon=True).start()
    return f"Job {job_id} started in background"

daemon 线程的特点是:主线程退出时,daemon 线程自动被终止。这设计很巧妙——如果 agent 挂了,后台任务也不该继续乱跑。

而在每次 LLM 调用之前,harness 会调用 bg_check 把通知队列里已完成的任务清空,把结果注入到对话上下文中:

[User] 帮我分析这个项目的依赖漏洞
[Agent] 调用 bg_run("npm audit")  ->  不等结果,继续思考
[Agent] 调用 read(package.json)   ->  同步完成
[Agent] --- 下一轮循环 ---
[Auto] bg_check 发现 npm audit 已完成 -> 注入结果
[Agent] 基于 audit 结果给出修复建议

这种设计让我想起了 Node.js 的 Event Loop。区别只是:Node.js 是单线程异步回调,这里是LLM 调用层面的异步——agent 的认知能力不再被 IO 阻塞。


3.3 s09:Agent Teams——队友 + 异步邮箱

格言:“When the task is too big for one, delegate to teammates”

这是我个人最兴奋的一节课。从 s09 开始,agent 不再是孤军奋战了——它有了队友

JSONL 邮箱架构

作者实现了一套基于 JSONL 文件的"异步邮箱"系统。你没看错,是JSONL 文件。没有 Redis、没有 RabbitMQ、没有 WebSocket。就纯文件 append + readline。

.team/
  config.json          # 团队成员注册表
  inbox/
    alice.jsonl        # alice 的收件箱
    bob.jsonl          # bob 的收件箱
    lead.jsonl         # 队长的收件箱

config.json 定义了团队成员的角色和状态:

{
  "team_name": "default",
  "members": [
    {"name": "alice", "role": "coder", "status": "idle"},
    {"name": "bob", "role": "reviewer", "status": "idle"}
  ]
}

发送消息的底层逻辑极其简洁:

def send_message(to: str, content: str) -> str:
    inbox_path = TEAM_DIR / "inbox" / f"{to}.jsonl"
    msg = {
        "from": CURRENT_AGENT,
        "to": to,
        "content": content,
        "timestamp": time.time(),
        "type": "message"
    }
    with open(inbox_path, "a") as f:
        f.write(json.dumps(msg) + "\n")
    return f"Message sent to {to}"

def read_inbox() -> list:
    inbox_path = TEAM_DIR / "inbox" / f"{CURRENT_AGENT}.jsonl"
    if not inbox_path.exists():
        return []
    # 读取所有行( drain 操作:一次性取出并清空)
    lines = inbox_path.read_text().strip().split("\n")
    inbox_path.write_text("")  # 清空
    return [json.loads(line) for line in lines if line]

drian(排空)语义是精髓:消息读一次就清空,避免重复消费。这就像 RabbitMQ 的 auto-ack,只不过 ack 就是 truncate 文件。

我画了一张架构图来理解这套通信模型:

.team/config.json                   .team/inbox/
+----------------------------+      +------------------+
| {"team_name": "default",   |      | alice.jsonl      |
|  "members": [{"name":"alic|      | bob.jsonl        |
|   "role":"coder",...}]}    |      | lead.jsonl       |
+----------------------------+      +------------------+
         |                                 ^
         |  spawn_teammate("bob","revie|   |  send_message("alice","fix bug")
         v                                 |
    +-----------+       JSONL 消息       +-----------+
    |  Thread   |  -------------------> |  Thread   |
    |  alice    |                       |  alice    |
    | agent_loop|                       | agent_loop|
    | status:   |                       | reads     |
    | working   |                       | inbox,    |
    +-----------+                       | acts      |
    +-----------+                       +-----------+
    |  Thread   |
    |  bob      |
    | agent_loop|
    | status:   |
    | idle      |
    +-----------+
5 种消息类型

团队通信不止于简单的"你好",作者定义了 5 种消息类型来支撑协作协议:

类型 用途
message 普通点对点消息
broadcast 群发通知
shutdown_request 请求队友优雅退出
shutdown_response 回应关机请求(确认/拒绝)
plan_approval_response 计划审批的回复(同意/驳回)

这 5 种类型覆盖了协商式通信的全部基础场景:普通对话、一对多广播、双向握手、计划审批。就像 TCP 协议只有几个核心原语(SYN、ACK、FIN),却能支撑整个互联网。

Spawn Teammate 的逻辑

当一个 agent 觉得任务太大时,它可以调用 spawn_teammate

1. 在 .team/config.json 注册新成员
2. 分配一个独立的 thread
3. 新线程里启动一个完整的 agent_loop
4. 新 agent 有自己的 messages[]、自己的上下文
5. 父 agent 可以通过 send_message 与新 agent 通信

注意:spawn 的不是子进程,而是同进程内的独立线程。这意味着它们共享同一个 Python 进程空间,但各自持有独立的对话上下文(messages list)。这也是这个项目的代码量能控制在 400 行以内的原因——如果上多进程 + RPC,代码量至少翻倍。

实验:启动两个 agent 互相通信

我尝试了一个实验:让"队长" agent 创建两个队友,一个负责写代码,一个负责 review。

$ python s09_agent_teams.py
> 请创建两个队友:alice 写功能代码,bob 做代码 review。
[Agent] 调用 spawn_teammate("alice", "coder", "实现用户注册功能")
[Agent] 调用 spawn_teammate("bob", "reviewer", "等待 review 请求")
[Agent] 调用 send_message("alice", "请实现 /register 接口")
--- alice 线程开始工作 ---
[alice] 调用 write(...) 创建文件
[alice] 调用 bash("pytest") 运行测试
[alice] 调用 send_message("lead", "注册接口已完成,等待 review")
--- 队长收到消息 ---
[Agent] 调用 send_message("bob", "请 review /register 接口的实现")
--- bob 线程开始工作 ---
[bob] 调用 read(...) 读取 alice 写的代码
[bob] 调用 send_message("alice", "建议增加密码强度校验")

看到两个 agent 在终端里交替输出,各自做着不同的事情,我有一种在看微型数字蚁群的错觉。每只蚂蚁(agent)不知道全局计划,但基于本地状态(inbox、task board)做出决策,整体却涌现出协作行为。


3.4 s10-s12:协议、自治、隔离

s09 搭起了团队的骨架,s10-s12 开始往骨架上填充"文明规则"——如何优雅地解散?如何让队友不再事事请示?如何避免大家踩到彼此的文件?

s10:Team Protocols——关机握手 FSM

格言:“Teammates need shared communication rules”

多人协作有一个经典问题:怎么关闭整个团队?如果队长直接 kill -9,队友可能正在写一个文件,写到一半就崩了。

作者的解决方案是一个基于 request_id 的双向握手,我把它理解为极简版的 TCP 四次挥手:

队长                            队友
  |  shutdown_request(id=42)  --->  |
  |                                | 保存状态、清理资源
  |  <--- shutdown_response(id=42) |
  |                                | 退出 agent_loop
  |  确认所有队友已回应             |
  |  队长自己退出                  |

request_id 是核心设计:队长发出请求后,等待所有队友返回带有相同 request_idshutdown_response。这确保了响应与请求的一一对应,防止陈旧消息干扰。

同样模式也被用在计划审批上:

队友提交计划 -> 队长收到 plan_approval_request
                队长决定 approve / reject
                返回 plan_approval_response

一个模式(request-response + 唯一 ID),驱动两种完全不同的协商场景。这就是设计模式的力量

s11:Autonomous Agents——空闲轮询 + 自动认领

格言:“Teammates scan the board and claim tasks themselves”

如果说 s10 解决的是"怎么停",s11 解决的是"怎么动起来不用催"。

在 s11 之前,队友 spawn 出来后,要么被动等消息,要么被队长逐个指派任务。这效率太低了。

s11 引入了一个革命性的机制:idle 工具。当 agent 完成当前工作、inbox 为空时,它会进入一个空闲循环

while status == "idle":
    1. 调用 task_list() 扫描看板
    2. 找到 blockedBy=[] 且 status="pending" 的任务
    3. 调用 claim(task_id) 认领
    4. 如果抢到任务:status -> "working",开始执行
    5. 如果没抢到:sleep(2),继续轮询

这就是自组织团队。队长不需要 micromanage,只需要在看板上放好任务,队友们会自动认领、自动执行。空闲的队友不会一直占用 LLM token,它们只在轮询时才产生一次轻量级的调用。

我把这个机制类比为工厂里的拉式生产(Pull System):不是主管推任务给员工,而是员工自己从看板拉取任务。Kanban + JIT,丰田生产方式在数字世界的还魂。

s12:Worktree Isolation——目录级并行执行

格言:“Each works in its own directory, no interference”

s09-s11 解决了"谁做什么"和"怎么通信",但还有一个致命问题没解决:如果两个 agent 同时修改同一个文件,冲突怎么办?

s12 的答案我特别喜欢:隔离,而非锁

它引入了 git worktree 机制。每个正在执行的任务可以绑定到一个独立的 worktree:

.tasks/task_12.json
{
  "id": 12,
  "subject": "Implement auth refactor",
  "status": "in_progress",
  "worktree": "auth-refactor"
}

.worktrees/index.json
{
  "worktrees": [
    {
      "name": "auth-refactor",
      "path": ".../.worktrees/auth-refactor",
      "branch": "wt/auth-refactor",
      "task_id": 12,
      "status": "active"
    }
  ]
}

物理隔离意味着:

  • agent A 在 .worktrees/auth-refactor/ 修改 app.py
  • agent B 在 .worktrees/payment-fix/ 也修改 app.py
  • 两者互不干扰,就像两个平行宇宙

worktree 用完后可以归档或删除。任务的协调通过 task ID,执行通过目录隔离。作者把这个 insight 概括为一句非常精辟的话:

“Isolate by directory, coordinate by task ID.”

这比任何基于锁的并发模型都优雅。锁是"串行化思维",隔离是"并行宇宙思维"。


3.5 s_full 总纲——全部机制合一

学完了 s01 到 s12,我终于来到了 s_full.py——把所有机制融合在一起的完全体。代码 740 行,16 个工具,一个完整的 Agent 操作系统。

完整架构图
+------------------------------------------------------------------+
|                        FULL AGENT                                |
|                                                                  |
|  System prompt (s05 skills, task-first + optional todo nag)      |
|                                                                  |
|  Before each LLM call:                                             |
|  +--------------------+  +------------------+  +--------------+  |
|  | Microcompact (s06) |  | Drain bg (s08)   |  | Check inbox  |  |
|  | Auto-compact (s06) |  | notifications    |  | (s09)        |  |
|  +--------------------+  +------------------+  +--------------+  |
|                                                                  |
|  Tool dispatch (s02 pattern):                                    |
|  +--------+----------+----------+---------+-----------+         |
|  | bash   | read     | write    | edit    | TodoWrite |          |
|  | task   | load_sk  | compress | bg_run  | bg_check  |          |
|  | t_crt  | t_get    | t_upd    | t_list  | spawn_tm  |          |
|  | list_tm| send_msg | rd_inbox | bcast   | shutdown  |          |
|  | plan   | idle     | claim    |         |           |          |
|  +--------+----------+----------+---------+-----------+         |
|                                                                  |
|  Subagent (s04):  spawn -> work -> return summary                |
|  Teammate (s09):  spawn -> work -> idle -> auto-claim (s11)      |
|  Shutdown (s10):  request_id handshake                           |
|  Plan gate (s10):  submit -> approve/reject                        |
+------------------------------------------------------------------+
工具数量递进表

回看整个学习路径,工具数量不是线性增长的,而是阶梯式跃迁——每次新增一个机制,才带来一批新工具:

Session 工具数 新增机制 核心格言
s01 1 Agent Loop One loop & Bash is all you need
s02 4 Tool dispatch Adding a tool means adding one handler
s03 5 TodoWrite An agent without a plan drifts
s04 5 Subagent Break big tasks down
s05 5 Skills Load knowledge when you need it
s06 5 Context Compact Context will fill up; make room
s07 8 Task System Persist to disk
s08 6 Background Tasks Run slow ops in background
s09 9 Agent Teams Delegate to teammates
s10 12 Team Protocols Shared communication rules
s11 14 Autonomous Agents Scan and claim themselves
s12 16 Worktree Isolation Each in its own directory
s_full 16 全部合一 Bash is all you need
REPL 命令

作者在交互层还设计了几个快捷命令,让用户可以直接向 harness 下达"元指令":

命令 作用
/compact 立即触发上下文压缩(s06)
/tasks 显示当前任务看板(s07)
/team 显示团队成员状态(s09)
/inbox 显示未读消息(s09)

这些命令不走 LLM,直接由 REPL 层解析执行。就像操作系统的系统调用——用户态和内核态的分界。


第四章:总结与感悟


4.1 最核心的三个领悟

一周时间,12 节课,从 120 行到 782 行,从 1 个工具到 16 个工具。我收获的不只是代码,更是一种心智模型的升级。如果让我用三句话总结,我会这样说:

领悟一:循环不变,机制叠加

这是贯穿全部 12 节课最震撼的发现。

s01 的 agent loop——那个 while True——直到 s_full 也没有变过。变化的只是每次循环前后挂载的机制

  • s03 挂载了 todo nag
  • s04 挂载了 subagent 上下文隔离
  • s06 挂载了上下文压缩
  • s07 挂载了任务持久化
  • s08 挂载了后台任务排空
  • s09 挂载了 inbox 检查
每个课程在这个循环之上叠加一个 harness 机制 —— 循环本身始终不变。
循环属于 agent。机制属于 harness。

这像极了Linux 内核的中断机制:系统调用入口永远不变,但挂载在上面的子系统从 1 个增长到了几百个。好的架构不是修改核心,而是让核心足够稳定,允许外围无限扩展。

领悟二:信任模型聚焦 Harness

在学这个项目之前,我一直以为"做 AI Agent"就是在写提示词、搭工作流、做决策树。学完之后我意识到,那是完全搞反了

Agency(感知、推理、行动的能力)来自模型训练。工程师能做的事情——且应该做的事情——是构建一个Harness(载具):给它合适的工具、准确的知识、清晰的观察接口、安全的权限边界。

最好的 Agent 产品,出自那些明白自己的工作是 Harness Engineering 而非 Intelligence Engineering 的工程师。

模型是驾驶员,Harness 是车。车手再强,开在拖拉机上也跑不赢 F1。

领悟三:设计模式跨领域通用

这个项目里看到的很多模式,在我之前做后端、做分布式系统时就见过,只是换了个马甲:

Agent 机制 经典计算机科学概念
Task System 数据库事务 + DAG 调度
JSONL 邮箱 消息队列(极简版)
后台任务 + 通知排空 异步 IO / Event Loop
关机握手 FSM TCP 四次挥手
空闲轮询 + 自动认领 拉式生产 / 工作窃取
Worktree 隔离 多版本并发控制(MVCC)

作者没有发明任何新理论,他只是把这些经过几十年验证的模式,以极简的方式嫁接到了 LLM Agent 的场景中。这种"跨领域模式识别"的能力,是区分普通程序员和系统架构师的关键。


4.2 给读者的建议

如果你读到了这里,大概率也对 Agent 开发感兴趣。我想分享几点基于这一周学习的建议:

学习路径建议

1. 先跑通 s01

2. 读懂 agent loop

3. 逐步叠加机制

4. 手写一遍 s_full

5. 替换 LLM 为本地模型

6. 为你的领域定制 tools

  • 绿色阶段(必做):至少跑通 s01-s04,理解"循环 + 工具分发 + 子代理"这三个核心支柱
  • 黄色阶段(推荐):手写一遍 s_full 的总纲,把全部 16 个工具在自己的项目里实现一次
  • 红色阶段(进阶):把 Claude API 换成本地模型(Ollama / vLLM),为你的业务领域写专属的 tool schemas
实践建议
  1. 从 Bash 开始:不要一上来就追求"多智能体协作"。一个 bash 工具 + 一个循环,已经能解决 60% 的自动化问题。

  2. 每加一个机制,测透一个机制:每节课的代码只有 100-400 行,但作者在每个工具里都做了边界处理(safe_path、危险命令拦截、超时控制)。魔鬼在细节里。

  3. 关注"没做什么":Claude Code 最优雅的地方,恰恰在于它没有强加任何预设的工作流、没有替代模型做决策、没有堆砌无意义的抽象层。学会做减法。

  4. 把 Harness 思维迁移到其他领域:如果你在做机器人、在做 IoT、在做游戏 NPC——同样的模式(工具集 + 知识加载 + 权限边界 + 观察接口)全部适用。


4.3 结语

回顾这一周的学习,我的心情从最初的好奇,到中期的震撼,再到最后的平静。

平静是因为我终于理解了这件事的本质:Agent 不是魔法,不是 prompt engineering 的终极形态,不是各种 SaaS 平台兜售的"AI 工作流"。Agent 是一个循环,一圈一圈地转,每圈里有感知、有推理、有行动。仅此而已。

作者在第一行代码里就告诉我们:

Bash is all you need。

当你理解了这一点,所有的复杂框架、所有的拖拽式工作流平台、所有的"AI Agent 即服务",在你眼里都会变得透明——你会看到它们背后那个永远不会变的循环,以及挂载在循环上的、可以无限叠加的机制。

而当我站在更高的视角回望,我想把这句话扩展一下:

Real agents are all the universe needs.

因为这个宇宙本身就是一个 agent loop:感知(观测星光)、推理(物理定律)、行动(恒星演化),循环了 138 亿年。我们只是在用 Python 和 JSON 写它的一个微小副本。

愿你的 harness 稳健,愿你的 agent 自由。


(全文完)


如果这篇博客对你有启发,欢迎点赞、收藏、转发!
有任何问题或不同见解,欢迎在评论区留言交流,我会逐一回复。
也欢迎去 GitHub 给原仓库 shareAI-lab/learn-claude-code 点个 Star,支持作者继续输出高质量的教学内容!

Logo

欢迎加入DeepSeek 技术社区。在这里,你可以找到志同道合的朋友,共同探索AI技术的奥秘。

更多推荐