从零打造 AI 编程助手:12 步理解 Claude Code 的核心架构

作者注：最近在学Claude Code的基本原理。本文通过 12 个渐进式示例，带你从零开始理解 AI 编程代理（Agent）的工作原理。包含大量真实代码实现，适合想深入了解技术细节的开发者。

引子：AI 是怎么”写代码”的？

你有没有好奇过，像 Claude、ChatGPT 这样的 AI，是怎么帮你写代码、改文件、执行命令的？

答案并不神秘——核心只是一个简单的循环。

今天，我们就通过 12 个循序渐进的例子，揭开 AI 编程助手的面纱。每个例子都在前一个的基础上增加一点功能，就像搭积木一样，最终构建出一个完整的 AI 编程代理系统。

本文特色：

• ✅ 通俗易懂的生活化类比
• ✅ 真实的 Python 代码实现
• ✅ 逐行代码详解
• ✅ 完整的执行流程示例

准备好了吗？我们开始！

第 1 步：Agent 循环——AI 的”心跳”

核心概念：让 AI 能反复使用工具，直到完成任务

想象一下你去餐厅点菜：

1. 你点菜（用户输入）
2. 厨师思考（AI 决定用什么工具）
3. 厨师做菜（执行工具）
4. 菜端上来（返回结果）
5. **你尝了说”再加个辣”**（继续下一轮）

这就是 Agent 循环的本质：

用户输入 → AI 思考 → 调用工具 → 返回结果 → 继续循环 → 直到完成

核心代码实现

让我们看看真实的代码实现。首先是工具定义：

# 告诉 AI 它有哪些工具可用
TOOLS = [{
"name": "bash",
"description": "Run a shell command.",
"input_schema": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"],
    },
}]

然后是工具的执行函数：

import subprocess

defrun_bash(command: str) -> str:
# 安全检查：阻止危险命令
    dangerous = ["rm -rf /", "sudo", "shutdown", "reboot"]
if any(d in command for d in dangerous):
return"Error: Dangerous command blocked"

try:
# 执行命令，捕获输出
        r = subprocess.run(
            command, 
            shell=True, 
            capture_output=True, 
            text=True, 
            timeout=120# 120 秒超时
        )
        out = (r.stdout + r.stderr).strip()
return out[:50000] if out else"(no output)"
except subprocess.TimeoutExpired:
return"Error: Timeout (120s)"

最核心的 Agent 循环：

defagent_loop(messages: list):
whileTrue:
# 1. 调用 AI，传入工具和对话历史
        response = client.messages.create(
            model=MODEL, 
            system=SYSTEM, 
            messages=messages,
            tools=TOOLS, 
            max_tokens=8000,
        )

# 2. 把 AI 的回复加入对话历史
        messages.append({
"role": "assistant", 
"content": response.content
        })

# 3. 如果 AI 没有调用工具，说明任务完成，退出循环
if response.stop_reason != "tool_use":
return

# 4. 执行 AI 调用的工具
        results = []
for block in response.content:
if block.type == "tool_use":
                print(f"$ {block.input['command']}")
                output = run_bash(block.input["command"])
                results.append({
"type": "tool_result", 
"tool_use_id": block.id,
"content": output
                })

# 5. 把工具结果加入对话历史，继续下一轮循环
        messages.append({"role": "user", "content": results})

执行流程示例

用户：列出当前目录的文件

第 1 轮循环：
  AI 决定调用 bash 工具
  → run_bash("ls -la")
  → 返回文件列表

第 2 轮循环：
  AI 看到结果，直接回复用户
  → "当前目录有以下文件：..."
  → stop_reason = "end_turn"（不是 tool_use）
  → 循环结束

关键洞察： AI 不是一次性回答，而是可以反复使用工具，直到任务完成。整个 AI 编程代理的秘密，就藏在这个 while True 循环里。

第 2 步：工具扩展——从”只能执行命令”到”全能选手”

核心概念：增加更多工具，让 AI 能做的事情更多

第一步里，AI 只能执行命令行指令（bash）。这就像只会用一把锤子的工人。

第二步，我们给 AI 增加了更多工具：

• bash：执行命令（锤子）
• read_file：读取文件（放大镜）
• write_file：写入文件（笔）
• edit_file：编辑文件（手术刀）

现在 AI 不仅能执行命令，还能读写和编辑文件了！

核心代码实现

1. 安全路径检查

防止 AI 读写工作目录以外的文件：

from pathlib import Path

WORKDIR = Path.cwd()  # 当前工作目录

defsafe_path(p: str) -> Path:
"""确保路径在工作目录内"""
    path = (WORKDIR / p).resolve()  # 解析为绝对路径
ifnot path.is_relative_to(WORKDIR):
raise ValueError(f"Path escapes workspace: {p}")
return path

# 示例：
safe_path("../../etc/passwd")  # ❌ 报错！路径逃逸
safe_path("README.md")         # ✅ 返回绝对路径

2. 文件读取函数

defrun_read(path: str, limit: int = None) -> str:
try:
        text = safe_path(path).read_text()
        lines = text.splitlines()  # 按行分割

# 如果指定了 limit，只返回前 N 行
if limit and limit < len(lines):
            lines = lines[:limit] + [f"... ({len(lines) - limit} more lines)"]

return"\n".join(lines)[:50000]  # 最多返回 5 万字
except Exception as e:
returnf"Error: {e}"

# 示例：
run_read("big_file.py", limit=10)
# → 返回前 10 行 + "... (990 more lines)"

3. 文件写入函数

defrun_write(path: str, content: str) -> str:
try:
        fp = safe_path(path)
# 如果父目录不存在，自动创建
        fp.parent.mkdir(parents=True, exist_ok=True)
        fp.write_text(content)
returnf"Wrote {len(content)} bytes to {path}"
except Exception as e:
returnf"Error: {e}"

4. 文件编辑函数（精确替换）

defrun_edit(path: str, old_text: str, new_text: str) -> str:
try:
        fp = safe_path(path)
        content = fp.read_text()

# 检查要替换的文本是否存在
if old_text notin content:
returnf"Error: Text not found in {path}"

# 只替换第一次出现（最后一个参数 1）
        fp.write_text(content.replace(old_text, new_text, 1))
returnf"Edited {path}"
except Exception as e:
returnf"Error: {e}"

# 示例：
# 原文件：print("world")
run_edit("test.py", 'print("world")', 'print("hello world")')
# → 修改后：print("hello world")

5. 工具分发器（关键设计）

# 工具名称 → 执行函数的映射
TOOL_HANDLERS = {
"bash":       lambda **kw: run_bash(kw["command"]),
"read_file":  lambda **kw: run_read(kw["path"], kw.get("limit")),
"write_file": lambda **kw: run_write(kw["path"], kw["content"]),
"edit_file":  lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),
}

lambda **kw 是匿名函数，接收任意关键字参数。比如 AI 调用 read_file(path="test.py", limit=10) 时，kw = {"path": "test.py", "limit": 10}。

6. 工具定义（告诉 AI 有哪些工具）

TOOLS = [
    {"name": "bash", "description": "Run a shell command.",
"input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},
    {"name": "read_file", "description": "Read file contents.",
"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},
    {"name": "write_file", "description": "Write content to file.",
"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
    {"name": "edit_file", "description": "Replace exact text in file.",
"input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
]

每个工具包含：

• name：工具名字
• description：工具干什么
• input_schema：参数定义（JSON Schema 格式）

7. Agent 循环的变化

defagent_loop(messages: list):
whileTrue:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=TOOLS, max_tokens=8000,  # ← 工具从 1 个变成了 4 个
        )
        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.get(block.name)
                output = handler(**block.input) if handler elsef"Unknown tool: {block.name}"
                results.append({
"type": "tool_result", 
"tool_use_id": block.id, 
"content": output
                })

        messages.append({"role": "user", "content": results})

关键洞察： 循环结构完全没变，只是工具变多了。添加新工具？只需在 TOOL_HANDLERS 里加一行映射！工具分发器模式让系统极易扩展。

第 3 步：任务管理——让 AI 学会”做计划”

核心概念：让 AI 自己跟踪进度，管理多步骤任务

假设你让 AI：”帮我重构这个模块，包括：1. 读取代码 2. 分析依赖 3. 重写函数 4. 更新测试”

如果没有任务管理，AI 可能会做一步忘一步。

第三步，我们给 AI 一个待办清单（TODO List）：

☑ 读取代码 - 完成
☑ 分析依赖 - 完成
⬜ 重写函数 - 进行中
⬜ 更新测试 - 待开始

AI 现在可以：

• 创建任务
• 更新状态（待开始→进行中→已完成）
• 查看进度

类比： 就像你去旅游前写的行程单，完成一项划掉一项，不会迷路。

关键洞察： 结构化的任务列表让 AI 能处理复杂的多步骤任务，不会遗忘或混乱。

第 4 步：子代理——AI 的”分身术”

核心概念：上下文隔离，让子任务独立运行

想象你在写论文，突然需要查一个资料。如果查资料的过程和你的写作混在一起，你的思路会被打断。

解决办法？派个助手去查，你继续写。

这就是子代理（Subagent）的作用：

主 AI（写论文）
  └─ 派生子 AI（查资料）
       ├─ 搜索
       ├─ 阅读
       └─ 返回摘要
  ← 收到摘要，继续写

核心代码实现

1. 子代理函数（核心）

defrun_subagent(prompt: str) -> str:
# 子 agent 的对话历史从零开始
    sub_messages = [{"role": "user", "content": prompt}]

# 最多循环 30 次，防止 AI 一直用工具不停
for _ in range(30):
# 调用 AI，使用子 agent 自己的对话历史和工具
        response = client.messages.create(
            model=MODEL, 
            system=SUBAGENT_SYSTEM, 
            messages=sub_messages,
            tools=CHILD_TOOLS,  # 子 agent 的工具集（没有 task 工具）
            max_tokens=8000,
        )

# 把 AI 回复加入子 agent 的对话历史
        sub_messages.append({
"role": "assistant", 
"content": response.content
        })

# 如果 AI 直接说话了（没用工具），退出循环
if response.stop_reason != "tool_use":
break

# 执行工具，收集结果
        results = []
for block in response.content:
if block.type == "tool_use":
                handler = TOOL_HANDLERS.get(block.name)
                output = handler(**block.input) if handler elsef"Unknown tool: {block.name}"
# 截断输出，防止撑爆上下文
                results.append({
"type": "tool_result", 
"tool_use_id": block.id, 
"content": str(output)[:50000]
                })

# 把工具结果加入子 agent 的对话历史
        sub_messages.append({"role": "user", "content": results})

# 提取 AI 最后的文字回复作为摘要返回
return"".join(
        b.text for b in response.content 
if hasattr(b, "text")
    ) or"(no summary)"

关键点： 函数返回后，sub_messages 这个变量被 Python 回收——子 agent 的所有中间步骤都消失了，只有摘要留给了父 agent。

2. 子 Agent 的工具集

# 子 agent 只能用基础工具
CHILD_TOOLS = [
    {"name": "bash", ...},
    {"name": "read_file", ...},
    {"name": "write_file", ...},
    {"name": "edit_file", ...},
]
# 注意：没有 task 工具！

为什么没有 task 工具？ 防止无限递归嵌套：父→子→孙→曾孙→…

3. 父 Agent 的工具集

# 父 agent = 子 agent 的工具 + 一个额外的 task 工具
PARENT_TOOLS = CHILD_TOOLS + [
    {"name": "task", 
"description": "Spawn a subagent with fresh context.",
"input_schema": {
"type": "object", 
"properties": {
"prompt": {"type": "string"},
"description": {"type": "string"}
         }, 
"required": ["prompt"]
     }
]

4. Agent 循环的变化

defagent_loop(messages: list):
whileTrue:
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=PARENT_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.name == "task":
# ← 关键：调用子 agent
                desc = block.input.get("description", "subtask")
                prompt = block.input.get("prompt", "")
                print(f"> task ({desc}): {prompt[:80]}")
                output = run_subagent(prompt)
else:
# 普通工具
                handler = TOOL_HANDLERS.get(block.name)
                output = handler(**block.input)

            results.append({
"type": "tool_result", 
"tool_use_id": block.id, 
"content": str(output)
            })

        messages.append({"role": "user", "content": results})

完整流程示例

用户：帮我找出项目里所有的 TODO 注释

父 agent 收到 → 决定用 task 工具
  └─ task(prompt="搜索整个项目，找出所有 TODO 注释")
      │
      ▼
  子 agent 启动（空白上下文！不知道前面聊了什么）
      ├─ 调用 bash("grep -r 'TODO' .")
      ├─ 收到结果：s01.py:10 TODO: xxx, s03.py:20 TODO: yyy
      └─ 回复："Found 2 TODO comments in: s01.py, s03.py"
      │
      ▼
  父 agent 收到摘要（只有这一行！子 agent 的中间步骤全消失了）
  └─ 回复用户："找到了 2 个 TODO..."

对比没有子 agent 的情况：

• 父 agent 自己做：对话历史会多出 10 轮（grep 命令、结果分析…），占满上下文窗口
• 用子 agent：对话历史只多了一行摘要

关键洞察： 通过进程隔离实现上下文隔离，父任务不会被子任务的细节污染。函数返回后，子 agent 的所有中间状态自动销毁。

第 5 步：技能加载——AI 的”按需学习”

核心概念：不要把所有知识都塞给 AI，按需加载

假设你有 10 个专业领域的知识（PDF 处理、代码审查、数据库优化…），每个有 5000 字的说明。

错误做法： 启动时全部塞给 AI → 5 万字，又贵又慢

正确做法： 两层架构

核心代码实现

1. 技能文件格式（SKILL.md）

---
name: pdf
description: Process PDF files with text extraction
tags: file-processing
---

# PDF Processing Skill

Step 1: Use pdfplumber to open the file
Step 2: Extract text from each page
Step 3: ...

文件分为两部分：

• YAML frontmatter（--- 之间）：元数据（名称、描述）
• 正文（--- 之后）：详细操作指南

2. 解析 YAML Frontmatter

import re
import yaml

classSkillLoader:
def_parse_frontmatter(self, text: str) -> tuple:
"""解析 YAML frontmatter"""
# 正则表达式匹配两个 --- 之间的内容
        match = re.match(r"^---\n(.*?)\n---\n(.*)", text, re.DOTALL)
ifnot match:
return {}, text

try:
# 把 YAML 文本解析成字典
            meta = yaml.safe_load(match.group(1)) or {}
except yaml.YAMLError:
            meta = {}

# 返回元数据和正文
return meta, match.group(2).strip()

3. 扫描所有技能

from pathlib import Path

classSkillLoader:
def__init__(self, skills_dir: Path):
        self.skills_dir = skills_dir
        self.skills = {}
        self._load_all()

def_load_all(self):
ifnot self.skills_dir.exists():
return

# 递归查找所有 SKILL.md 文件
for f in sorted(self.skills_dir.rglob("SKILL.md")):
            text = f.read_text()
            meta, body = self._parse_frontmatter(text)

# 优先用 YAML 里的 name，没有就用文件夹名
            name = meta.get("name", f.parent.name)

# 保存技能信息
            self.skills[name] = {
"meta": meta,    # 元数据
"body": body,    # 正文
"path": str(f)   # 文件路径
            }

4. 第一层：获取技能描述

defget_descriptions(self) -> str:
"""Layer 1: 简短描述，嵌入 System Prompt"""
ifnot self.skills:
return"(no skills available)"

    lines = []
for name, skill in self.skills.items():
        desc = skill["meta"].get("description", "No description")
        tags = skill["meta"].get("tags", "")
        line = f"  - {name}: {desc}"
if tags:
            line += f" [{tags}]"
        lines.append(line)

return"\n".join(lines)

输出示例：

  - pdf: Process PDF files with text extraction [file-processing]
  - code-review: Review code for bugs and style issues [quality]

5. 第二层：获取技能正文

defget_content(self, name: str) -> str:
"""Layer 2: 完整技能正文，通过 tool_result 返回"""
    skill = self.skills.get(name)
ifnot skill:
returnf"Error: Unknown skill '{name}'. Available: {', '.join(self.skills.keys())}"

returnf'<skill name="{name}">\n{skill["body"]}\n</skill>'

输出示例：

<skill name="pdf">
# PDF Processing Skill

Step 1: Use pdfplumber to open the file
Step 2: Extract text from each page
...
</skill>

6. System Prompt 的组装

SKILL_LOADER = SkillLoader(SKILLS_DIR)

SYSTEM = f"""You are a coding agent at {WORKDIR}.
Use load_skill to access specialized knowledge before tackling unfamiliar topics.

Skills available:
{SKILL_LOADER.get_descriptions()}"""

启动时： 扫描所有技能，只把描述列表嵌入 System Prompt（几百 token）

7. 工具注册

TOOL_HANDLERS = {
    ...
"load_skill": lambda **kw: SKILL_LOADER.get_content(kw["name"]),
}

TOOLS = [
    ...
    {"name": "load_skill", 
"description": "Load specialized knowledge by name.",
"input_schema": {
"type": "object", 
"properties": {"name": {"type": "string"}}, 
"required": ["name"]
     }
]

完整流程示例

用户：帮我分析这个 PDF 文件

AI 看到 System Prompt：
  Skills available:
    - pdf: Process PDF files...
    - code-review: Review code...

AI 决定：我需要 PDF 处理的知识
  └─ 调用 load_skill(name="pdf")
      │
      ▼
  返回完整的 PDF 处理指南（几千字）

AI 收到技能正文 → 根据指南处理 PDF 文件
  └─ 调用 bash("python extract_pdf.py file.pdf")
  └─ 返回结果

对比：

• 全放 System Prompt：启动时加载 5 万 token（10 个技能 × 5000 token）
• 两层架构：启动时只加载几百字的描述，需要时才加载具体知识

关键洞察：“描述先行、按需加载” 避免了 token 浪费。AI 启动时只知道有哪些技能，遇到不懂的领域再加载详细指南。

第 6 步：上下文压缩——AI 的”战略性遗忘”

核心概念：让 AI 可以忘记旧内容，从而无限期工作

AI 的记忆（上下文窗口）是有限的，比如 20 万字。当对话越来越长，旧内容会超出限制，AI 会”忘记”之前做了什么。

怎么办？压缩！

我们设计了三层压缩机制：

核心代码实现

1. 估算 Token 数

defestimate_tokens(messages: list) -> int:
"""粗略估算 token 数：约 4 个字符 = 1 个 token"""
return len(str(messages)) // 4

# 示例：
estimate_tokens(messages)  # → 60000（超过 5 万，需要压缩）

2. 第一层：微压缩（Micro Compact）

每轮循环都做，轻量级压缩：

KEEP_RECENT = 3# 保留最近 3 个工具结果
PRESERVE_RESULT_TOOLS = {"read_file"}  # 保留 read_file 的输出

defmicro_compact(messages: list) -> list:
# 收集所有 tool_result 条目
    tool_results = []
for msg_idx, msg in enumerate(messages):
if msg["role"] == "user"and isinstance(msg.get("content"), list):
for part_idx, part in enumerate(msg["content"]):
if isinstance(part, dict) and part.get("type") == "tool_result":
                    tool_results.append((msg_idx, part_idx, part))

# 如果不超过 KEEP_RECENT 个，不需要压缩
if len(tool_results) <= KEEP_RECENT:
return messages

# 构建工具 ID 到工具名的映射
    tool_name_map = {}
for msg in messages:
if msg["role"] == "assistant":
            content = msg.get("content", [])
if isinstance(content, list):
for block in content:
if hasattr(block, "type") and block.type == "tool_use":
                        tool_name_map[block.id] = block.name

# 替换旧的结果（保留最近 KEEP_RECENT 个）
    to_clear = tool_results[:-KEEP_RECENT]
for _, _, result in to_clear:
# 跳过小结果或 read_file 结果
ifnot isinstance(result.get("content"), str) or len(result["content"]) <= 100:
continue

        tool_id = result.get("tool_use_id", "")
        tool_name = tool_name_map.get(tool_id, "unknown")

# 保留 read_file 的输出
if tool_name in PRESERVE_RESULT_TOOLS:
continue

# 替换成占位符
        result["content"] = f"[Previous: used {tool_name}]"

return messages

压缩前后对比：

压缩前：

[tool_result: ls 输出的几千行文件列表...]
[tool_result: cat 输出的几百行代码...]
[tool_result: git diff 的大段变更...]

压缩后：

[Previous: used bash]
[Previous: used bash]
[Previous: used edit_file]
[tool_result: read_file 输出（保留）]  ← read_file 被保留
[tool_result: 最新的工具结果]
[tool_result: 最新的工具结果]

3. 第二层：自动压缩（Auto Compact）

当对话超过阈值时触发，重量级压缩：

import json
import time
from pathlib import Path

TRANSCRIPT_DIR = Path(".transcripts")
THRESHOLD = 50000# 5 万 token

defauto_compact(messages: list) -> list:
# 1. 保存完整对话记录到磁盘
    TRANSCRIPT_DIR.mkdir(exist_ok=True)
    transcript_path = TRANSCRIPT_DIR / f"transcript_{int(time.time())}.jsonl"

with open(transcript_path, "w") as f:
for msg in messages:
            f.write(json.dumps(msg, default=str) + "\n")

    print(f"[transcript saved: {transcript_path}]")

# 2. 让 AI 总结对话
    conversation_text = json.dumps(messages, default=str)[-80000:]  # 最多 8 万字符

    response = client.messages.create(
        model=MODEL,
        messages=[{"role": "user", "content":
"Summarize this conversation for continuity. Include: "
"1) What was accomplished, 2) Current state, 3) Key decisions made. "
"Be concise but preserve critical details.\n\n" + conversation_text
        }],
        max_tokens=2000,
    )

# 提取总结
    summary = next(
        (block.text for block in response.content if hasattr(block, "text")), 
""
    )
ifnot summary:
        summary = "No summary generated."

# 3. 替换所有消息为一条总结
return [
        {"role": "user", "content": 
f"[Conversation compressed. Transcript: {transcript_path}]\n\n{summary}"
        }
    ]

压缩效果：

压缩前（几百条消息）：

messages = [
    {"role": "user", "content": "帮我重构这个模块"},
    {"role": "assistant", "content": [tool_use: read_file]},
    {"role": "user", "content": [tool_result: 2000行代码...]},
    ... 几百条消息 ...
]

压缩后（只剩一条）：

messages = [
    {"role": "user", "content": 
"[Conversation compressed. Transcript: .transcripts/transcript_1713600000.jsonl]\n\n"
"总结：已完成数据库迁移脚本的编写和测试，当前正在编写前端适配层。"
"关键决策：选择 PostgreSQL 作为数据库，使用 SQLAlchemy 作为 ORM。"
    }
]

4. 第三层：手动压缩（Manual Compact）

AI 可以主动触发压缩：

TOOL_HANDLERS = {
    ...
"compact": lambda **kw: "Manual compression requested.",
}

# 在 agent_loop 中：
if block.name == "compact":
    print("[manual compact]")
    messages[:] = auto_compact(messages)
return

5. Agent 循环的变化

defagent_loop(messages: list):
whileTrue:
# Layer 1: 微压缩（每轮都做）
        micro_compact(messages)

# Layer 2: 自动压缩（超限时触发）
if estimate_tokens(messages) > THRESHOLD:
            print("[auto_compact triggered]")
            messages[:] = auto_compact(messages)

# 调用 AI
        response = client.messages.create(...)

# ... 执行工具 ...

# Layer 3: 手动压缩（AI 主动触发）
if manual_compact:
            print("[manual compact]")
            messages[:] = auto_compact(messages)
return

三层压缩的执行顺序：

每次循环：
  1. micro_compact() → 替换旧工具结果为占位符（轻量，每轮都做）
  2. 检查 token 数 → 如果超过 5 万，触发 auto_compact()（重量级）
  3. 调用 AI → 执行工具
  4. 如果 AI 调用了 compact 工具 → 立即触发 auto_compact() 并返回

关键洞察： 三层压缩机制让 agent 不会撑爆上下文窗口：

• 微压缩：每轮清理旧工具结果
• 自动压缩：上下文过大时让 AI 总结对话
• 手动压缩：AI 主动触发总结

旧的完整对话记录保存到磁盘，不会丢失。AI 可以有策略地遗忘，从而永远持续工作。

第 7 步：任务持久化——断电也不忘

核心概念：任务目标存储在文件中，不随对话压缩而消失

第六步的压缩虽然好，但有个问题：压缩后，AI 可能忘记”我本来要做什么”。

第七步，我们把任务持久化到文件：

.tasks/
  task_1.json  {"标题": "重构模块", "状态": "进行中", ...}
  task_2.json  {"标题": "修复 bug", "状态": "待开始", ...}

现在即使对话被压缩，AI 也能从文件里读取任务状态。

类比： 就像你把待办事项写在纸上，即使大脑忘了，看纸就行。

关键洞察：文件存储让任务目标独立于对话历史，压缩也不会丢失目标。

第 8 步：后台任务——AI 学会”多线程工作”

核心概念：AI 可以同时运行多个任务，不用等一个完成再开始下一个

假设你让 AI 运行一个需要 10 分钟的测试。

没有后台任务： AI 傻傻等 10 分钟，啥也不干。

有后台任务： AI 启动测试后，去做别的事情，时不时回来看看测试进度。

实现方式：

# 启动后台任务
task_id = start_background("python test.py")

# 继续做其他事...
read_file("code.py")
edit_file("config.py")

# 检查任务状态
status = check_task(task_id)
# → "运行中" 或 "已完成"

类比： 就像你用洗衣机洗衣服——启动后不用守在旁边，可以去做饭、看电视。

关键洞察： AI 从单线程变成多线程，效率大幅提升。

第 9 步：团队代理——AI 的”团队协作”

核心概念：多个不同角色的 AI 组成团队，分工合作

现在我们有多个 AI 代理了，但它们还是各干各的。

第九步，让它们组成团队：

产品经理 AI → 提需求
  ↓
开发者 AI   → 写代码
  ↓
测试员 AI   → 跑测试

每个 AI 有自己的：

• 角色（产品经理、开发者、测试员）
• 性格（严谨、创新、细致）
• 专长（需求分析、编码、测试）

它们通过任务系统分配和完成任务，就像一个真实团队。

类比： 就像公司的部门协作——产品提需求，开发写代码，测试找 bug。

关键洞察： 多角色 AI 团队可以处理更复杂的协作任务，各司其职。

第 10 步：团队协议——AI 的”沟通规范”

核心概念：结构化的握手协议，实现关机和审批流程

团队有了，但怎么沟通？怎么确保消息不丢失？怎么安全关机？

第十步，引入结构化协议：

请求-响应模式

AI-A 发送：{"request_id": "abc123", "内容": "请审查代码"}
AI-B 回复：{"request_id": "abc123", "内容": "审查通过"}

通过 request_id 关联请求和响应，就像快递单号追踪包裹。

审批流程

开发者 AI：请求部署到生产环境
  ↓
管理员 AI：审批通过/拒绝

安全关机

主 AI：所有人，准备关机
  ↓
各 AI：保存状态 → 确认 → 退出

类比： 就像公司的规章制度——请假要审批，下班要打卡。

关键洞察： 结构化的协议让 AI 团队可靠通信，不会丢消息或混乱。

第 11 步：自主 Agent——AI 学会”自己找活干”

核心概念：AI 空闲时自动扫描任务板，主动认领任务

之前的 AI 都是被动接受任务。第十一步，让 AI 学会主动工作：

whileTrue:
    任务 = 扫描任务板()

if 有未分配的任务:
        认领任务()
        执行任务()
else:
        休息一会儿()

现在 AI 会：

• 空闲时自动扫描任务板
• 主动认领未分配的任务
• 执行完成后继续扫描

类比： 就像餐厅服务员——空闲时主动找客人点单，而不是傻傻站着。

关键洞察： AI 从被动执行变成主动工作，实现真正的自主性。

第 12 步：工作树隔离——AI 的”平行宇宙”

核心概念：每个任务在独立目录中执行，互不干扰

假设两个 AI 同时修改同一个文件，会发生什么？冲突！

最后一步，我们引入Git Worktree，给每个任务一个独立的工作目录：

主仓库/
  worktree_1/  ← 任务 1 在这里修改
  worktree_2/  ← 任务 2 在那里修改
  worktree_3/  ← 任务 3 在另一个地方修改

每个 worktree 都是独立的”平行宇宙”：

• 任务 1 改文件 A，不影响任务 2
• 任务完成后，合并到主仓库
• 临时目录自动清理

类比： 就像多人编辑文档——每个人在自己的副本上改，最后合并。

关键洞察：目录级隔离让多个任务可以安全并行，互不干扰。

总结：12 步构建完整 AI 编程代理

让我们回顾一下这 12 步的演进：

核心架构模式

这 12 步揭示了 AI 编程代理的核心设计模式：

1. 循环驱动：Agent 的本质是一个 while 循环
2. 工具扩展：通过工具让 AI 与世界交互
3. 上下文管理：压缩、隔离、持久化，让 AI 不会”失忆”
4. 多 Agent 协作：团队、协议、自主性，实现复杂任务
5. 安全隔离：工作树、子代理，防止冲突

为什么这很重要？

理解这些模式，你就能：

• 看懂 Claude Code、Cursor 等 AI 编程工具的底层原理
• 构建自己的 AI 代理系统
• 优化现有 AI 工具的使用方式

完整代码

所有示例代码都在这个开源项目中： 👉 learn-claude-code on GitHub

每个文件都可以直接运行，体验 AI Agent 的完整工作流程。

写在最后

AI 编程助手并不神秘——核心只是一个循环，加上精心设计的工具、任务管理和协作机制。

通过这 12 步的渐进式学习，希望你不仅能”用”好 AI 工具，更能”理解”它们的工作原理。

毕竟，理解原理，才能更好地驾驭工具。

如果你觉得这篇文章有帮助，欢迎点赞、转发、关注！

延伸阅读：

• s01-s12 的详细技术文档：agents/explanations/ 目录
• 每个文件都包含逐行代码解释和 Python 基础速查表
• 适合 Python 新手，即使不会编程也能看懂

技术交流： 欢迎在 GitHub 提 Issue 或 PR！

附录：核心代码速查索引

为方便开发者快速查阅，以下是文章中涉及的核心代码片段索引：

s01 – Agent 循环

• 工具定义：TOOLS 数组，JSON Schema 格式
• 工具执行：run_bash() 函数，使用 subprocess.run()
• 核心循环：agent_loop() 函数，while True + stop_reason 判断
• 安全检查：dangerous 列表，any() 函数

s02 – 工具扩展

• 安全路径：safe_path() 函数，Path.is_relative_to() 检查
• 文件读取：run_read() 函数，limit 参数限制行数
• 文件写入：run_write() 函数，mkdir(parents=True) 自动创建目录
• 文件编辑：run_edit() 函数，replace(old, new, 1) 只替换第一次
• 工具分发器：TOOL_HANDLERS 字典，lambda **kw 匿名函数

s03 – 任务管理

• 任务类：TodoItem dataclass
• 任务状态：pending → in_progress → completed
• 工具函数：todo_add(), todo_update(), todo_list()

s04 – 子代理

• 子代理函数：run_subagent()，独立 sub_messages 列表
• 工具隔离：CHILD_TOOLS 不包含 task 工具
• 上下文清理：函数返回后 sub_messages 自动销毁
• 循环限制：for _ in range(30) 防止无限循环

s05 – 技能加载

• YAML 解析：re.match() + yaml.safe_load()
• 技能扫描：Path.rglob("SKILL.md") 递归查找
• 两层架构：get_descriptions() + get_content()
• System Prompt：只嵌入描述，按需加载正文

s06 – 上下文压缩

• Token 估算：len(str(messages)) // 4
• 微压缩：micro_compact() 替换旧工具结果为占位符
• 自动压缩：auto_compact() 让 AI 总结对话
• 对话保存：json.dumps() 写入 .transcripts/ 目录
• 阈值检查：estimate_tokens() > THRESHOLD

s07 – 任务持久化

• JSON 存储：json.load() / json.dump()
• 文件扫描：glob.glob("*.json")
• 任务依赖：blockedBy 字段
• 状态同步：对话压缩后从文件读取任务

s08 – 后台任务

• 进程启动：subprocess.Popen()
• 状态检查：process.poll()
• 输出读取：process.stdout.read()
• 线程安全：threading.Lock()
• 并行执行：threading.Thread()

s09 – 团队代理

• Agent 类：class Agent: 定义角色
• 角色属性：name, role, personality
• 文件邮箱：通过文件通信
• 任务分配：共享任务系统

s10 – 团队协议

• 请求 ID：uuid.uuid4() 生成唯一标识
• 请求响应：通过 request_id 关联
• 审批流程：approve / reject
• 安全关机：保存状态 → 确认 → 退出

s11 – 自主 Agent

• 自动扫描：while True 循环
• 任务认领：扫描未分配任务
• 随机选择：random.choice()
• 进程管理：os.getpid(), sys.exit()

s12 – 工作树隔离

• Git Worktree：git worktree add
• 目录隔离：每个任务独立 worktree
• 临时目录：tempfile.mkdtemp()
• 自动清理：atexit.register()
• 并行安全：互不干扰