Learn-Claude-Code 学习教程
Learn-Claude-Code 学习教程
从零开始学习 Agent Harness 工程 - 12 个渐进式会话
教程日期：2026-03-27
目录
- 环境准备
- 快速开始
- 会话 1: Agent Loop
- 会话 2: Tool Use
- 会话 3: Todo/Planning
- 会话 4: Subagent
- 会话 5: Skill Loading
- 会话 6: Context Compact
- 会话 7-12: 高级功能
- 最佳实践
- 常见问题
一、环境准备
1.1 前置知识
学习本教程前建议了解：
| 知识领域 | 重要程度 | 说明 |
|---|---|---|
| Python 基础 | ⭐⭐⭐⭐⭐ | 异步编程、类和数据类 |
| LLM API | ⭐⭐⭐⭐ | 理解 API 调用方式 |
| Claude API | ⭐⭐⭐ | Anthropic API 基础 |
| 工具调用概念 | ⭐⭐⭐ | Function Calling |
1.2 系统要求
Python 版本：3.9+
操作系统：macOS / Linux / Windows
需要：
├── Anthropic API Key
└── 网络访问
1.3 安装
# 克隆仓库
git clone https://github.com/shareAI-lab/learn-claude-code.git
cd learn-claude-code
# 创建虚拟环境
python -m venv venv
source venv/bin/activate # Linux/macOS
# 或 venv\Scripts\activate # Windows
# 安装依赖
pip install -r requirements.txt
1.4 配置 API Key
# 方式 1: 环境变量
export ANTHROPIC_API_KEY=your_api_key
# 方式 2: .env 文件
echo "ANTHROPIC_API_KEY=your_api_key" > .env
1.5 验证安装
import anthropic
client = anthropic.Anthropic()
message = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=100,
messages=[{"role": "user", "content": "Hello!"}]
)
print(message.content)
二、快速开始
2.1 运行第一个 Agent
# 运行会话 1: Agent Loop
python agents/s01_agent_loop.py
2.2 核心概念速览
┌─────────────────────────────────────────────────────────────────────┐
│ Agent Harness 核心概念 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 1. Agent Loop: 一个循环处理所有交互 │
│ │
│ while True: │
│ response = call_llm() │
│ if has_tool_calls(response): │
│ results = execute_tools() │
│ add_to_history(results) │
│ else: │
│ return text │
│ │
│ 2. Tool: 每个工具一个处理器 │
│ │
│ class Tool: │
│ name: str │
│ description: str │
│ input_schema: dict │
│ execute(**kwargs) -> str │
│ │
│ 3. Context: 上下文管理和压缩 │
│ │
│ 4. Subagent: 干净上下文的子任务处理 │
│ │
└─────────────────────────────────────────────────────────────────────┘
三、会话 1: Agent Loop
3.1 核心概念
格言: "One loop & Bash is all you need"
Agent Loop 是所有 Agent 的基础。它是一个简单的循环： 1. 调用 LLM 2. 检查是否有工具调用 3. 如果有，执行工具并添加结果 4. 循环继续 5. 如果没有，返回文本
3.2 基础实现
import anthropic
MODEL = "claude-sonnet-4-20250514"
SYSTEM = "You are a helpful assistant."
def agent_loop(user_message: str) -> str:
"""基础 Agent 循环"""
client = anthropic.Anthropic()
messages = [{"role": "user", "content": user_message}]
while True:
# 1. 调用 LLM
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=SYSTEM,
messages=messages,
)
# 2. 添加助手响应
messages.append({
"role": "assistant",
"content": response.content
})
# 3. 检查停止原因
if response.stop_reason == "end_turn":
# 没有工具调用，返回文本
return response.content[0].text
# 如果有工具调用，继续循环
# (下一个会话会添加工具处理)
# 使用
result = agent_loop("Hello, how are you?")
print(result)
3.3 带历史记录的循环
def agent_loop_with_history(user_message: str, history: list = None) -> tuple[str, list]:
"""带历史记录的 Agent 循环"""
client = anthropic.Anthropic()
messages = history or []
messages.append({"role": "user", "content": user_message})
while True:
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=SYSTEM,
messages=messages,
)
messages.append({
"role": "assistant",
"content": response.content
})
if response.stop_reason == "end_turn":
return response.content[0].text, messages
# 多轮对话
history = []
response, history = agent_loop_with_history("My name is Alice", history)
print(response)
response, history = agent_loop_with_history("What's my name?", history)
print(response) # Should remember "Alice"
四、会话 2: Tool Use
4.1 核心概念
格言: "Adding a tool means adding one handler"
工具扩展了 Agent 的能力。每个工具需要： - 名称和描述 - 输入 Schema - 执行函数
4.2 定义工具
from dataclasses import dataclass
from typing import Any
@dataclass
class Tool:
"""工具基类"""
name: str
description: str
input_schema: dict[str, Any]
def to_api_format(self) -> dict:
"""转换为 Claude API 格式"""
return {
"name": self.name,
"description": self.description,
"input_schema": self.input_schema,
}
def execute(self, **kwargs) -> str:
"""执行工具"""
raise NotImplementedError
4.3 创建具体工具
import subprocess
class BashTool(Tool):
"""Bash 命令工具"""
name = "bash"
description = "Execute bash commands"
input_schema = {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The bash command to execute"
}
},
"required": ["command"]
}
def execute(self, command: str) -> str:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True
)
return result.stdout or result.stderr
class ReadFileTool(Tool):
"""读取文件工具"""
name = "read_file"
description = "Read the contents of a file"
input_schema = {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Path to the file to read"
}
},
"required": ["file_path"]
}
def execute(self, file_path: str) -> str:
try:
with open(file_path, 'r') as f:
return f.read()
except Exception as e:
return f"Error reading file: {str(e)}"
4.4 Agent 循环 + 工具
import json
def agent_loop_with_tools(user_message: str, tools: list[Tool]) -> str:
"""带工具的 Agent 循环"""
client = anthropic.Anthropic()
messages = [{"role": "user", "content": user_message}]
# 创建工具字典
tool_dict = {tool.name: tool for tool in tools}
while True:
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=SYSTEM,
messages=messages,
tools=[tool.to_api_format() for tool in tools],
)
messages.append({
"role": "assistant",
"content": response.content
})
# 检查是否有工具调用
if response.stop_reason == "tool_use":
# 提取工具调用
tool_calls = [block for block in response.content if block.type == "tool_use"]
# 执行每个工具
tool_results = []
for tool_call in tool_calls:
tool = tool_dict[tool_call.name]
args = tool_call.input
result = tool.execute(**args)
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_call.id,
"content": result,
})
# 添加工具结果
messages.append({
"role": "user",
"content": tool_results
})
# 继续循环
continue
# 没有工具调用，返回文本
return response.content[0].text
# 使用
tools = [BashTool(), ReadFileTool()]
result = agent_loop_with_tools("List files in current directory", tools)
print(result)
五、会话 3: Todo/Planning
5.1 核心概念
格言: "An agent without a plan drifts"
计划能力让 Agent 能够： - 追踪进度 - 分解任务 - 保持专注
5.2 Todo 工具实现
from typing import Literal
class TodoWriteTool(Tool):
"""Todo 写入工具"""
name = "TodoWrite"
description = "Write todos to track progress on tasks"
input_schema = {
"type": "object",
"properties": {
"todos": {
"type": "array",
"items": {
"type": "object",
"properties": {
"content": {"type": "string"},
"status": {
"type": "string",
"enum": ["pending", "in_progress", "completed"]
},
"priority": {"type": "string", "enum": ["high", "medium", "low"]}
},
"required": ["content", "status"]
}
}
},
"required": ["todos"]
}
def __init__(self):
self.todos = []
def execute(self, todos: list) -> str:
self.todos = todos
return f"Updated {len(todos)} todos:\n" + "\n".join(
f"- [{t['status']}] {t['content']}" for t in todos
)
class TodoReadTool(Tool):
"""Todo 读取工具"""
name = "TodoRead"
description = "Read the current todo list"
input_schema = {"type": "object", "properties": {}}
def __init__(self, todo_write_tool: TodoWriteTool):
self.todo_tool = todo_write_tool
def execute(self) -> str:
if not self.todo_tool.todos:
return "No todos currently"
return "\n".join(
f"- [{t['status']}] {t['content']}" for t in self.todo_tool.todos
)
5.3 使用示例
# 创建工具
todo_write = TodoWriteTool()
todo_read = TodoReadTool(todo_write)
# 创建 Agent
tools = [todo_write, todo_read, BashTool()]
# Agent 会自动使用 TodoWrite 来规划任务
result = agent_loop_with_tools(
"Create a simple Python web server",
tools
)
print(result)
六、会话 4: Subagent
6.1 核心概念
格言: "Break big tasks down; each subtask gets a clean context"
子 Agent 的优势： - 隔离上下文 - 专注单一任务 - 避免干扰
6.2 Subagent 实现
class SubagentTool(Tool):
"""子 Agent 工具"""
name = "subagent"
description = "Spawn a subagent to handle a specific task"
input_schema = {
"type": "object",
"properties": {
"task": {
"type": "string",
"description": "The task for the subagent"
},
"context": {
"type": "string",
"description": "Additional context for the subagent"
}
},
"required": ["task"]
}
def __init__(self, tools: list[Tool]):
self.tools = tools
def execute(self, task: str, context: str = "") -> str:
# 创建子 Agent
subagent_messages = [{
"role": "user",
"content": f"Task: {task}\n\n{context}"
}]
# 使用干净的上下文运行
result = agent_loop_with_tools(
subagent_messages[0]["content"],
self.tools
)
return result
6.3 使用示例
tools = [
BashTool(),
ReadFileTool(),
TodoWriteTool(),
SubagentTool([BashTool(), ReadFileTool()]) # 子 Agent 的工具
]
result = agent_loop_with_tools(
"Analyze the codebase structure and create a summary",
tools
)
print(result)
七、会话 5: Skill Loading
7.1 核心概念
格言: "Load knowledge when you need it, not upfront"
技能系统允许： - 按需加载知识 - 模块化组织 - 可扩展性
7.2 Skill 结构
skills/
├── code-review/
│ └── SKILL.md
├── testing/
│ └── SKILL.md
└── documentation/
└── SKILL.md
7.3 SKILL.md 格式
---
name: code-review
description: Use when reviewing code for quality, security, and best practices
---
# Code Review Skill
## Purpose
Review code and provide actionable feedback.
## Checklist
1. Code quality
2. Security issues
3. Performance concerns
4. Best practices
## Output Format
- Issue: [description]
- Severity: [high/medium/low]
- Recommendation: [fix]
7.4 Skill 加载器
from pathlib import Path
import yaml
class SkillLoader:
"""技能加载器"""
def __init__(self, skills_dir: Path):
self.skills_dir = skills_dir
self.loaded_skills: dict[str, dict] = {}
def load_skill(self, skill_name: str) -> dict:
"""加载技能"""
if skill_name in self.loaded_skills:
return self.loaded_skills[skill_name]
skill_path = self.skills_dir / skill_name / "SKILL.md"
with open(skill_path) as f:
content = f.read()
# 解析 YAML frontmatter
if content.startswith("---"):
parts = content.split("---", 2)
metadata = yaml.safe_load(parts[1])
body = parts[2].strip()
else:
metadata = {}
body = content
skill = {
"name": metadata.get("name", skill_name),
"description": metadata.get("description", ""),
"body": body,
}
self.loaded_skills[skill_name] = skill
return skill
def get_skill_prompt(self, skill_name: str) -> str:
"""获取技能提示词"""
skill = self.load_skill(skill_name)
return f"# {skill['name']}\n\n{skill['body']}"
八、会话 6: Context Compact
8.1 核心概念
格言: "Context will fill up; you need a way to make room"
上下文压缩策略： - 移除旧消息 - 保留关键信息 - 添加压缩通知
8.2 实现
class MessageHistory:
"""消息历史管理"""
def __init__(self, context_window: int = 100000):
self.messages: list[dict] = []
self.context_window = context_window
self.token_count = 0
def add(self, message: dict) -> None:
"""添加消息"""
self.messages.append(message)
self.token_count += self._count_tokens(message)
self._check_truncate()
def _count_tokens(self, message: dict) -> int:
"""估算 token 数量"""
content = str(message.get("content", ""))
return len(content) // 4 # 粗略估算
def _check_truncate(self) -> None:
"""检查是否需要压缩"""
if self.token_count > self.context_window:
self._truncate()
def _truncate(self) -> None:
"""压缩上下文"""
# 移除最旧的消息对
while len(self.messages) >= 2 and self.token_count > self.context_window * 0.7:
removed = self.messages.pop(0)
self.token_count -= self._count_tokens(removed)
# 添加压缩通知
self.messages.insert(0, {
"role": "user",
"content": "[Earlier conversation truncated for context limit]"
})
def get_messages(self) -> list[dict]:
"""获取消息列表"""
return self.messages
8.3 使用示例
history = MessageHistory(context_window=50000)
# 添加多轮对话
for i in range(100):
history.add({"role": "user", "content": f"Message {i}"})
history.add({"role": "assistant", "content": f"Response {i}"})
# 自动压缩
print(f"Total messages: {len(history.get_messages())}")
print(f"First message: {history.get_messages()[0]}")
九、会话 7-12: 高级功能
9.1 Task System (s07)
from dataclasses import dataclass
from enum import Enum
import json
from pathlib import Path
class TaskStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
@dataclass
class Task:
id: str
content: str
status: TaskStatus = TaskStatus.PENDING
priority: str = "medium"
class TaskManager:
"""任务管理器"""
def __init__(self, storage_path: Path):
self.storage_path = storage_path
self.tasks: list[Task] = []
self._load()
def add(self, task: Task) -> None:
self.tasks.append(task)
self._save()
def get_next(self) -> Task | None:
pending = [t for t in self.tasks if t.status == TaskStatus.PENDING]
return pending[0] if pending else None
def complete(self, task_id: str) -> None:
for task in self.tasks:
if task.id == task_id:
task.status = TaskStatus.COMPLETED
self._save()
def _save(self) -> None:
data = [{"id": t.id, "content": t.content, "status": t.status.value} for t in self.tasks]
with open(self.storage_path, 'w') as f:
json.dump(data, f)
def _load(self) -> None:
if self.storage_path.exists():
with open(self.storage_path) as f:
data = json.load(f)
self.tasks = [Task(**t) for t in data]
9.2 Background Tasks (s08)
import asyncio
class BackgroundTaskManager:
"""后台任务管理器"""
def __init__(self):
self.tasks: dict[str, asyncio.Task] = {}
async def start(self, task_id: str, coro) -> str:
"""启动后台任务"""
task = asyncio.create_task(coro)
self.tasks[task_id] = task
return f"Started background task: {task_id}"
async def status(self, task_id: str) -> str:
"""检查任务状态"""
task = self.tasks.get(task_id)
if not task:
return f"Task {task_id} not found"
if task.done():
try:
result = task.result()
return f"Task {task_id} completed: {result}"
except Exception as e:
return f"Task {task_id} failed: {e}"
return f"Task {task_id} running"
async def wait(self, task_id: str) -> str:
"""等待任务完成"""
task = self.tasks.get(task_id)
if task:
await task
return await self.status(task_id)
return f"Task {task_id} not found"
9.3 Agent Teams (s09-s10)
class AgentTeam:
"""Agent 团队"""
def __init__(self, name: str, role: str, tools: list[Tool]):
self.name = name
self.role = role
self.tools = tools
async def run(self, task: str) -> str:
"""执行任务"""
return agent_loop_with_tools(
f"[{self.role}] {task}",
self.tools
)
class TeamCoordinator:
"""团队协调器"""
def __init__(self):
self.agents: dict[str, AgentTeam] = {}
def add_agent(self, agent: AgentTeam) -> None:
self.agents[agent.name] = agent
async def delegate(self, agent_name: str, task: str) -> str:
agent = self.agents.get(agent_name)
if agent:
return await agent.run(task)
return f"Agent {agent_name} not found"
9.4 Autonomous Agents (s11)
class AutonomousAgent:
"""自主 Agent"""
def __init__(self, name: str, skills: list[str]):
self.name = name
self.skills = skills
self.claimed_tasks: list[str] = []
def can_handle(self, task: Task) -> bool:
"""检查能否处理任务"""
return any(skill in task.required_skills for skill in self.skills)
def claim(self, task: Task) -> bool:
"""认领任务"""
if self.can_handle(task) and task.status == TaskStatus.PENDING:
self.claimed_tasks.append(task.id)
return True
return False
async def work(self, task: Task) -> str:
"""处理任务"""
# 实现任务处理逻辑
pass
9.5 Worktree Isolation (s12)
import subprocess
from pathlib import Path
class WorktreeManager:
"""Git Worktree 管理器"""
def __init__(self, repo_path: Path):
self.repo_path = repo_path
self.worktrees: dict[str, Path] = {}
def create(self, agent_name: str, branch: str) -> Path:
"""创建隔离的 worktree"""
worktree_path = self.repo_path / f".worktrees/{agent_name}"
subprocess.run([
"git", "worktree", "add",
str(worktree_path), "-b", branch
], cwd=self.repo_path)
self.worktrees[agent_name] = worktree_path
return worktree_path
def remove(self, agent_name: str) -> None:
"""移除 worktree"""
worktree_path = self.worktrees.get(agent_name)
if worktree_path:
subprocess.run([
"git", "worktree", "remove", str(worktree_path)
], cwd=self.repo_path)
del self.worktrees[agent_name]
十、最佳实践
10.1 设计原则
- 保持简单: 从基础循环开始，逐步添加功能
- 工具独立: 每个工具只做一件事
- 上下文管理: 始终监控上下文使用
- 子 Agent 隔离: 给子 Agent 干净的上下文
10.2 调试技巧
# 添加日志
import logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
def agent_loop_debug(user_message: str, tools: list[Tool]) -> str:
logger.info(f"Starting agent loop with: {user_message}")
while True:
response = call_llm(...)
logger.debug(f"Response: {response}")
if has_tool_calls(response):
logger.info(f"Tool calls: {extract_tools(response)}")
results = execute_tools(...)
logger.debug(f"Tool results: {results}")
else:
logger.info("Returning final response")
return response.text
10.3 错误处理
async def safe_agent_loop(user_message: str, tools: list[Tool], max_iterations: int = 50) -> str:
"""带错误处理的 Agent 循环"""
iterations = 0
while iterations < max_iterations:
try:
iterations += 1
response = await call_llm_async(...)
if has_tool_calls(response):
results = await execute_tools_async(...)
add_to_history(results)
else:
return response.text
except Exception as e:
logger.error(f"Error in iteration {iterations}: {e}")
if iterations >= max_iterations:
raise RuntimeError(f"Max iterations ({max_iterations}) reached")
await asyncio.sleep(1)
raise RuntimeError("Unexpected loop exit")
十一、常见问题
Q1: Agent 一直在循环，没有返回
原因: 可能是工具调用陷入了无限循环
解决: 添加最大迭代次数限制
def agent_loop_with_limit(user_message: str, tools: list[Tool], max_iterations: int = 20) -> str:
iterations = 0
while iterations < max_iterations:
iterations += 1
# ... 正常循环逻辑
return "Max iterations reached"
Q2: 上下文超出了限制
原因: 消息历史太长
解决: 实现上下文压缩
def truncate_history(messages: list, max_tokens: int = 50000) -> list:
while count_tokens(messages) > max_tokens:
messages.pop(0)
return messages
Q3: 工具执行失败
原因: 工具参数不正确或执行环境问题
解决: 添加错误处理
def safe_execute(tool: Tool, **kwargs) -> str:
try:
return tool.execute(**kwargs)
except Exception as e:
return f"Error executing {tool.name}: {str(e)}"
Q4: 子 Agent 没有隔离上下文
原因: 直接传递了历史记录
解决: 创建新的消息列表
def dispatch_subagent(task: str) -> str:
# 不要传递历史！
fresh_messages = [{"role": "user", "content": task}]
return agent_loop(fresh_messages, subagent_tools)
参考资源
- GitHub: https://github.com/shareAI-lab/learn-claude-code
- 官网: https://learn.shareai.run
- Anthropic API 文档: https://docs.anthropic.com
- Claude Code 文档: https://docs.anthropic.com/en/docs/claude-code
教程生成时间：2026-03-27