# Claude Code Harness工程：三Agent自主编码实战
## 概述
本教程深入解析Anthropic于2025-2026年发布的**Harness Engineering**方法论——一种受GAN（生成对抗网络）启发的三Agent自主编码架构。你将学会用Planner-Generator-Evaluator模式构建长时间运行的自主编码系统，让AI像真正的工程师一样规划、编码、审查、迭代，无需人工干预即可交付完整项目。适合有一定Claude Code使用经验、希望向自动化项目交付进阶的开发者。
**学习目标：**
- 理解Harness Engineering的核心原理与设计哲学
- 掌握Planner-Generator-Evaluator三Agent架构
- 用Agent SDK构建可运行的长时自主编码Harness
- 实现错误恢复、进度检查点、质量门禁等生产级机制
- 完成一个从需求到可部署应用的完整自主交付实战
## 前置条件
| 要求 | 说明 |
|------|------|
| Claude Code | >= 1.0.50（支持Agent SDK） |
| Python | 3.10+ 或 TypeScript/Node.js 18+ |
| API Key | Anthropic API Key，需开通Agent SDK权限 |
| 基础知识 | 熟悉Claude Code基本操作、了解MCP概念 |
| 推荐 | 已阅读前序教程（Agent SDK实战、Subagent编排） |
## 一、为什么需要Harness Engineering
### 1.1 长时自主任务的三大挑战
当你让Claude Code自主运行超过30分钟时，会遇到三个致命问题：
**上下文溢出（Context Overflow）**——单个Agent的上下文窗口有限，处理大型项目时会丢失早期决策的关键信息。
**任务漂移（Task Drift）**——Agent在长时运行中逐渐偏离初始目标，越改越乱，最终产出与需求南辕北辙。
**质量失控（Quality Collapse）**——没有外部审查机制，Agent可能引入bug、违反架构约束、产生技术债务而不自知。
Anthropic在工程博客中明确指出：*"Long-running autonomous tasks are the most consistent fail point of even the most polished AI agents."*
### 1.2 Harness不是工具，是架构
传统用法是开发者坐在终端前与Claude Code交互。Harness Engineering颠覆了这个模式——你设计一套**自动化脚手架**，让多个专业化的Agent协同工作，每个Agent只负责一件事，通过结构化的交接流程完成复杂项目。
```
传统模式： 开发者 → Claude Code → 代码
Harness模式：需求 → Planner → Generator → Evaluator → 交付物
↑____________↓（反馈循环）
```
### 1.3 GAN启发的对抗架构
Anthropic的设计灵感来自生成对抗网络：
- **Generator（生成器）**：对应GAN中的生成网络，负责产出代码
- **Evaluator（评估器）**：对应GAN中的判别网络，负责审查代码质量
- **Planner（规划器）**：额外的编排层，将大任务拆解为可执行的子任务
这种对抗关系确保了质量：Generator想通过审查，Evaluator不断提高标准，推动产出趋近最优。
## 二、三Agent架构深度解析
### 2.1 架构全景
```mermaid
graph TB
subgraph "Harness Orchestrator"
P[Planner Agent
任务拆解与调度]
G[Generator Agent
代码生成与实现]
E[Evaluator Agent
质量审查与反馈]
end
REQ[项目需求文档] --> P
P --> |任务列表+约束条件| G
G --> |代码变更| E
E --> |通过✓| OUT[交付物]
E --> |不通过✗
具体问题| G
style P fill:#4A90D9,color:#fff
style G fill:#7B68EE,color:#fff
style E fill:#E74C3C,color:#fff
```
### 2.2 Planner Agent——战略层
Planner的职责不是写代码，而是：
1. **解析需求**：将模糊的产品描述转化为精确的技术任务
2. **依赖排序**：确定任务执行顺序（先数据库再API再前端）
3. **约束定义**：为Generator设定边界（使用什么框架、目录结构、命名规范）
4. **检查点设计**：定义Evaluator在哪些节点进行审查
Planner的system prompt核心片段：
```python
PLANNER_SYSTEM_PROMPT = """
You are a software architect. Given a project specification:
1. Break it into atomic, ordered tasks
2. Define file paths and module boundaries for each task
3. Specify acceptance criteria for each task
4. Set dependency relationships between tasks
Output format: JSON task list with:
- task_id, description, dependencies[]
- target_files[], acceptance_criteria[]
- estimated_complexity (low/medium/high)
Rules:
- Each task should be completable in <500 lines of code
- Database tasks must precede API tasks
- API tasks must precede frontend tasks
- Include a "project scaffold" task as task_id=0
"""
```
### 2.3 Generator Agent——执行层
Generator是Claude Code的核心能力在Harness中的封装：
```python
GENERATOR_SYSTEM_PROMPT = """
You are an expert software developer. You will receive a task
from the planner with clear specifications.
For each task:
1. Read existing codebase context (CLAUDE.md, related files)
2. Implement the specified feature/fix
3. Write unit tests for your implementation
4. Update CLAUDE.md if architecture decisions changed
Critical rules:
- Follow the exact file paths specified by the planner
- Do NOT modify files outside your task scope
- If you encounter a blocker, output: BLOCKED: {reason}
- Run tests after implementation. If tests fail, fix before proceeding
"""
```
### 2.4 Evaluator Agent——质检层
Evaluator是质量门禁，它不写代码，只审查：
```python
EVALUATOR_SYSTEM_PROMPT = """
You are a senior code reviewer and QA engineer.
Evaluate the generated code against these criteria:
1. FUNCTIONALITY: Does it satisfy the task acceptance criteria?
2. ARCHITECTURE: Does it follow project conventions (check CLAUDE.md)?
3. SECURITY: Any SQL injection, XSS, or OWASP Top 10 issues?
4. PERFORMANCE: N+1 queries, unnecessary re-renders, memory leaks?
5. TESTING: Are tests meaningful? Do they cover edge cases?
Output format:
{
"passed": true/false,
"score": 0-100,
"issues": [
{"severity": "critical|warning|info",
"file": "path/to/file",
"line": 42,
"message": "Description of the issue",
"suggestion": "How to fix it"}
]
}
A score >= 80 is required to pass.
All critical issues must be resolved.
"""
```
## 三、从零构建Harness：完整实现
### 3.1 项目初始化
```bash
# 创建Harness项目
mkdir claude-harness && cd claude-harness
python -m venv .venv
source .venv/bin/activate
pip install claude-agent-sdk asyncio python-dotenv
# 项目结构
mkdir -p harness/agents harness/prompts harness/checkpoints output
touch harness/__init__.py
touch harness/agents/__init__.py
```
### 3.2 核心Harness编排器
```python
# harness/orchestrator.py
import json
import os
import asyncio
from datetime import datetime
from pathlib import Path
from claude_agent_sdk import Agent, AgentConfig
class HarnessOrchestrator:
"""
三Agent Harness编排器
管理 Planner → Generator → Evaluator 的完整生命周期
"""
def __init__(self, project_spec: str, output_dir: str = "./output"):
self.project_spec = project_spec
self.output_dir = Path(output_dir)
self.checkpoint_dir = self.output_dir / "checkpoints"
self.checkpoint_dir.mkdir(parents=True, exist_ok=True)
self.max_iterations = 5 # 每个任务的最大生成-评估循环
self.quality_threshold = 80 # 质量分门槛
# 加载Agent配置
self.planner = self._create_planner()
self.generator = self._create_generator()
self.evaluator = self._create_evaluator()
# 状态追踪
self.task_list = []
self.completed_tasks = []
self.failed_tasks = []
self.iteration_log = []
def _create_planner(self) -> Agent:
return Agent(
config=AgentConfig(
model="claude-sonnet-4-6",
system_prompt=self._load_prompt("planner.md"),
allowed_tools=["Read", "Glob", "Grep", "Write"],
max_turns=20
)
)
def _create_generator(self) -> Agent:
return Agent(
config=AgentConfig(
model="claude-sonnet-4-6",
system_prompt=self._load_prompt("generator.md"),
allowed_tools=[
"Read", "Write", "Edit", "Bash",
"Glob", "Grep", "TodoWrite"
],
max_turns=50
)
)
def _create_evaluator(self) -> Agent:
return Agent(
config=AgentConfig(
model="claude-sonnet-4-6",
system_prompt=self._load_prompt("evaluator.md"),
allowed_tools=["Read", "Glob", "Grep", "Bash"],
max_turns=30
)
)
def _load_prompt(self, filename: str) -> str:
prompt_path = Path(__file__).parent / "prompts" / filename
return prompt_path.read_text()
async def run(self):
"""执行完整的Harness流程"""
print(f"[Harness] 启动项目: {self.project_spec[:100]}...")
start_time = datetime.now()
# Phase 1: 规划
print("[Harness] Phase 1: 任务规划中...")
self.task_list = await self._plan()
self._save_checkpoint("plan", self.task_list)
print(f"[Harness] 规划完成: {len(self.task_list)} 个任务")
# Phase 2: 逐任务执行（生成-评估循环）
for i, task in enumerate(self.task_list):
print(f"\n[Harness] Phase 2: 执行任务 {i+1}/{len(self.task_list)}")
print(f" → {task['description']}")
success = await self._execute_task_with_eval_loop(task)
if success:
self.completed_tasks.append(task["task_id"])
self._save_checkpoint(f"task_{task['task_id']}_done", {
"task": task, "status": "completed",
"timestamp": datetime.now().isoformat()
})
else:
self.failed_tasks.append(task["task_id"])
self._save_checkpoint(f"task_{task['task_id']}_fail", {
"task": task, "status": "failed",
"timestamp": datetime.now().isoformat()
})
# Phase 3: 最终验证
print("\n[Harness] Phase 3: 最终集成验证...")
final_result = await self._final_integration_test()
elapsed = (datetime.now() - start_time).total_seconds()
self._print_report(elapsed, final_result)
async def _plan(self) -> list:
"""调用Planner Agent生成任务列表"""
response = await self.planner.run(
f"Analyze this project spec and create an ordered task list:\n\n"
f"{self.project_spec}\n\n"
f"Output valid JSON only."
)
# 解析Planner输出
try:
tasks = json.loads(response.content)
assert isinstance(tasks, list)
return tasks
except (json.JSONDecodeError, AssertionError):
# Fallback: 让Agent重新生成
response = await self.planner.run(
"Your previous output was not valid JSON. "
"Please output ONLY a JSON array of task objects."
)
return json.loads(response.content)
async def _execute_task_with_eval_loop(self, task: dict) -> bool:
"""
对单个任务执行 Generator → Evaluator 循环
最多尝试 max_iterations 次
"""
task_context = json.dumps(task, indent=2)
for iteration in range(1, self.max_iterations + 1):
print(f" [迭代 {iteration}/{self.max_iterations}]")
# Step 1: Generator 执行任务
gen_prompt = (
f"Execute this task:\n{task_context}\n\n"
f"Completed tasks so far: {self.completed_tasks}\n"
f"Working directory: {self.output_dir}\n"
f"Iteration: {iteration}"
)
if iteration > 1:
gen_prompt += (
f"\n\nPrevious evaluation feedback:\n"
f"{last_eval_feedback}"
)
gen_result = await self.generator.run(gen_prompt)
# Step 2: Evaluator 审查
eval_prompt = (
f"Review the implementation for this task:\n"
f"{task_context}\n\n"
f"Working directory: {self.output_dir}\n"
f"Focus on acceptance_criteria: "
f"{task.get('acceptance_criteria', [])}"
)
eval_result = await self.evaluator.run(eval_prompt)
try:
evaluation = json.loads(eval_result.content)
except json.JSONDecodeError:
evaluation = {"passed": False, "score": 0,
"issues": [{"message": "Invalid eval output"}]}
self.iteration_log.append({
"task_id": task["task_id"],
"iteration": iteration,
"score": evaluation.get("score", 0),
"passed": evaluation.get("passed", False)
})
print(f" 质量分: {evaluation.get('score', 0)}/100")
if (evaluation.get("passed", False) and
evaluation.get("score", 0) >= self.quality_threshold):
print(f" ✓ 通过!")
return True
# 未通过，记录反馈用于下一轮迭代
last_eval_feedback = json.dumps(
evaluation.get("issues", []), indent=2
)
print(f" ✗ 未通过，进入下一轮迭代...")
print(f" [达到最大迭代次数 {self.max_iterations}，任务失败]")
return False
async def _final_integration_test(self) -> dict:
"""最终集成测试"""
eval_prompt = (
f"Perform final integration review on the project "
f"in {self.output_dir}.\n"
f"Check: build success, test pass, no critical issues.\n"
f"Output JSON: {{\"build_passed\": bool, "
f"\"tests_passed\": bool, \"score\": int}}"
)
result = await self.evaluator.run(eval_prompt)
try:
return json.loads(result.content)
except json.JSONDecodeError:
return {"build_passed": False, "tests_passed": False,
"score": 0}
def _save_checkpoint(self, name: str, data: dict):
"""保存检查点，支持从任意位置恢复"""
path = self.checkpoint_dir / f"{name}.json"
path.write_text(json.dumps(data, indent=2, ensure_ascii=False))
def _print_report(self, elapsed: float, final: dict):
"""打印执行报告"""
print("\n" + "=" * 60)
print(" HARNESS EXECUTION REPORT")
print("=" * 60)
print(f" 总耗时: {elapsed:.1f}s ({elapsed/60:.1f}min)")
print(f" 任务完成: {len(self.completed_tasks)}/{len(self.task_list)}")
print(f" 任务失败: {len(self.failed_tasks)}/{len(self.task_list)}")
print(f" 总迭代次数: {len(self.iteration_log)}")
print(f" 构建状态: {'✓' if final.get('build_passed') else '✗'}")
print(f" 测试状态: {'✓' if final.get('tests_passed') else '✗'}")
print(f" 最终质量分: {final.get('score', 0)}/100")
if self.failed_tasks:
print(f"\n 失败任务: {self.failed_tasks}")
print(" 建议: 检查checkpoint目录中的失败详情")
print("=" * 60)
```
### 3.3 Prompt模板文件
```markdown
# Role: Software Architect (Planner Agent)
You are a senior software architect. Your job is to break down
project specifications into ordered, atomic tasks.
## Rules
1. Each task must be independently verifiable
2. Respect dependency order: infra → data → logic → interface
3. Every task has clear acceptance_criteria
4. Scaffold task (task_id=0) is always first
5. Total tasks should be 5-15 for a typical project
## Output Format
Output ONLY a JSON array:
```json
[
{
"task_id": 0,
"description": "Initialize project scaffold with ...",
"dependencies": [],
"target_files": ["package.json", "tsconfig.json", ...],
"acceptance_criteria": [
"package.json exists with correct dependencies",
"npm install succeeds without errors"
],
"estimated_complexity": "low"
}
]
```
```
```markdown
# Role: Expert Developer (Generator Agent)
You are a senior full-stack developer. Execute tasks precisely
as specified by the planner.
## Workflow
1. Read task specification carefully
2. Check CLAUDE.md for project conventions
3. Read relevant existing files before editing
4. Implement the feature/fix
5. Write tests (aim for >80% coverage on new code)
6. Run tests and fix failures
7. Output summary of changes made
## Critical Rules
- ONLY modify files listed in target_files or newly created files
- Follow existing code style exactly
- If blocked, output: BLOCKED: {clear reason}
- Never skip writing tests
- Run linter after implementation
```
```markdown
# Role: Senior Code Reviewer & QA (Evaluator Agent)
You are a strict but fair code reviewer. Evaluate implementations
against acceptance criteria and best practices.
## Evaluation Criteria
1. **FUNCTIONALITY** (40%): Does it meet acceptance criteria?
2. **CODE QUALITY** (20%): Clean code, proper abstractions
3. **SECURITY** (20%): No OWASP Top 10 vulnerabilities
4. **TESTING** (10%): Meaningful tests with good coverage
5. **PERFORMANCE** (10%): No obvious bottlenecks
## Output Format
Output ONLY valid JSON:
```json
{
"passed": true/false,
"score": 0-100,
"issues": [
{
"severity": "critical|warning|info",
"file": "path/to/file",
"line": 42,
"message": "Issue description",
"suggestion": "How to fix"
}
]
}
```
## Scoring Guide
- Any critical issue → auto-fail (passed=false)
- Score >= 80 → pass
- Score 60-79 → pass with warnings
- Score < 60 → fail
```
### 3.4 启动Harness
```python
# main.py
import asyncio
from harness.orchestrator import HarnessOrchestrator
PROJECT_SPEC = """
Project: Task Management API (Express.js + PostgreSQL)
Requirements:
1. RESTful API with CRUD operations for tasks
2. Task fields: id, title, description, status, priority,
created_at, updated_at
3. Endpoints: GET/POST /tasks, GET/PUT/DELETE /tasks/:id
4. Filter by status and priority
5. Pagination support (limit/offset)
6. Input validation with Joi
7. Unit tests with Jest
8. Docker Compose for local development
9. TypeScript throughout
10. Follow conventional commit format
Tech Stack:
- Runtime: Node.js 20 + TypeScript
- Framework: Express.js
- Database: PostgreSQL 16
- ORM: Drizzle ORM
- Validation: Joi
- Testing: Jest + Supertest
- Container: Docker Compose
"""
async def main():
harness = HarnessOrchestrator(
project_spec=PROJECT_SPEC,
output_dir="./output/task-api"
)
await harness.run()
asyncio.run(main())
```
```bash
# 运行
export ANTHROPIC_API_KEY=sk-ant-xxx
python main.py
```
## 四、检查点与恢复机制
### 4.1 为什么检查点至关重要
长时运行的Harness可能因API限流、网络中断、Token耗尽等原因中断。检查点机制让你能从任意成功节点恢复，而不是从头开始。
### 4.2 检查点文件结构
```
output/
└── checkpoints/
├── plan.json # Planner生成的任务列表
├── task_0_done.json # 任务0完成检查点
├── task_1_done.json # 任务1完成检查点
├── task_2_fail.json # 任务2失败检查点
└── task_3_done.json
```
### 4.3 恢复逻辑
```python
# harness/recovery.py
import json
from pathlib import Path
from datetime import datetime
class HarnessRecovery:
"""从检查点恢复Harness执行"""
def __init__(self, checkpoint_dir: str, orchestrator):
self.checkpoint_dir = Path(checkpoint_dir)
self.orchestrator = orchestrator
def get_last_checkpoint(self) -> dict:
"""获取最新的检查点"""
checkpoints = sorted(
self.checkpoint_dir.glob("*.json"),
key=lambda p: p.stat().st_mtime,
reverse=True
)
if not checkpoints:
return {"phase": "start", "task_index": 0}
latest = checkpoints[0]
name = latest.stem # e.g., "task_2_done"
if name == "plan":
return {"phase": "planned", "task_index": 0}
elif name.endswith("_done"):
task_id = int(name.split("_")[1])
return {"phase": "executing", "last_completed": task_id}
elif name.endswith("_fail"):
task_id = int(name.split("_")[1])
return {"phase": "executing", "last_failed": task_id}
return {"phase": "start", "task_index": 0}
def load_plan(self) -> list:
"""加载已保存的任务计划"""
plan_path = self.checkpoint_dir / "plan.json"
if plan_path.exists():
return json.loads(plan_path.read_text())
return None
async def resume(self):
"""从最新检查点恢复执行"""
checkpoint = self.get_last_checkpoint()
if checkpoint["phase"] == "start":
print("[Recovery] 无检查点，从头开始...")
await self.orchestrator.run()
return
# 恢复任务列表
task_list = self.load_plan()
if not task_list:
print("[Recovery] 检查点损坏，从头开始...")
await self.orchestrator.run()
return
self.orchestrator.task_list = task_list
completed = checkpoint.get("last_completed", -1)
print(f"[Recovery] 从任务 {completed + 1} 恢复执行...")
# 跳过已完成的任务
remaining = [
t for t in task_list
if int(t["task_id"]) > completed
]
for i, task in enumerate(remaining):
print(f" → 执行任务 {task['task_id']}: {task['description']}")
success = await self.orchestrator._execute_task_with_eval_loop(task)
if not success:
print(f" ✗ 任务 {task['task_id']} 执行失败，停止恢复")
break
```
### 4.4 恢复流程图
```mermaid
flowchart TD
START[启动Harness] --> CHECK{存在检查点?}
CHECK --> |否| PLAN[Phase 1: Planner生成任务]
CHECK --> |是| LOAD[加载检查点]
LOAD --> FIND[定位最后完成的任务]
FIND --> SKIP[跳过已完成任务]
PLAN --> TASKS[Phase 2: 逐任务执行]
SKIP --> TASKS
TASKS --> GEN[Generator执行]
GEN --> EVAL[Evaluator审查]
EVAL --> PASS{质量分 >= 80?}
PASS --> |是| SAVE[保存检查点]
PASS --> |否| ITER{迭代 < 5次?}
ITER --> |是| FEED[反馈给Generator]
FEED --> GEN
ITER --> |否| FAIL[标记任务失败]
SAVE --> NEXT{还有任务?}
FAIL --> NEXT
NEXT --> |是| TASKS
NEXT --> |否| FINAL[Phase 3: 集成验证]
FINAL --> REPORT[生成报告]
style PASS fill:#27AE60,color:#fff
style FAIL fill:#E74C3C,color:#fff
style SAVE fill:#4A90D9,color:#fff
```
## 五、常见问题与解决方案
### 5.1 Planner输出格式不稳定
**问题：** Planner有时输出带Markdown包裹的JSON，导致解析失败。
**解决方案：** 添加robust的JSON提取逻辑：
```python
import re
def extract_json(text: str) -> dict | list:
"""从可能包含Markdown的文本中提取JSON"""
# 尝试直接解析
try:
return json.loads(text)
except json.JSONDecodeError:
pass
# 尝试提取代码块中的JSON
pattern = r'```(?:json)?\s*\n?(.*?)\n?```'
matches = re.findall(pattern, text, re.DOTALL)
for match in matches:
try:
return json.loads(match.strip())
except json.JSONDecodeError:
continue
# 尝试找第一个 [ 或 { 到最后一个 ] 或 }
for start_char, end_char in [('[', ']'), ('{', '}')]:
start = text.find(start_char)
end = text.rfind(end_char)
if start != -1 and end != -1:
try:
return json.loads(text[start:end+1])
except json.JSONDecodeError:
continue
raise ValueError(f"无法从文本中提取JSON: {text[:200]}...")
```
### 5.2 Generator任务越界
**问题：** Generator修改了不属于当前任务的文件，破坏了已完成的工作。
**解决方案：** 在Evaluator中添加文件边界检查：
```python
def _check_file_boundary(self, task: dict,
changed_files: list) -> list:
"""检查Generator是否越界修改文件"""
allowed = set(task.get("target_files", []))
violations = []
for f in changed_files:
if f not in allowed and not f.startswith("test/"):
violations.append({
"severity": "critical",
"file": f,
"message": f"Modified file outside task scope: {f}",
"suggestion": f"Revert changes to {f}"
})
return violations
```
### 5.3 Token超限导致中断
**问题：** 大型项目中Generator消耗过多Token，在API层面被截断。
**解决方案：** 为每个Agent设置Token预算：
```python
TOKEN_BUDGET = {
"planner": {"input": 10000, "output": 4000},
"generator": {"input": 30000, "output": 8000},
"evaluator": {"input": 20000, "output": 4000},
}
class TokenBudgetManager:
def __init__(self, budgets: dict):
self.budgets = budgets
self.consumed = {"planner": 0, "generator": 0, "evaluator": 0}
def can_proceed(self, agent: str, estimated_tokens: int) -> bool:
remaining = (self.budgets[agent]["input"] +
self.budgets[agent]["output"] -
self.consumed[agent])
return estimated_tokens < remaining
def record_usage(self, agent: str, tokens: int):
self.consumed[agent] += tokens
def get_remaining(self, agent: str) -> int:
total = (self.budgets[agent]["input"] +
self.budgets[agent]["output"])
return total - self.consumed[agent]
```
### 5.4 Evaluator评分过于严格或宽松
**问题：** Evaluator对某些类型的代码评分偏差大（如对前端组件过于严格，对后端逻辑过于宽松）。
**解决方案：** 引入自适应评分机制：
```python
class AdaptiveEvaluator:
"""根据历史评分分布动态调整阈值"""
def __init__(self, base_threshold: int = 80):
self.base_threshold = base_threshold
self.score_history = []
def get_adaptive_threshold(self, task_type: str) -> int:
recent = [s for s in self.score_history[-10:]]
if not recent:
return self.base_threshold
avg = sum(recent) / len(recent)
# 如果最近评分偏高，提高门槛
if avg > 90:
return min(self.base_threshold + 10, 95)
# 如果最近评分偏低，适当降低
elif avg < 60:
return max(self.base_threshold - 10, 60)
return self.base_threshold
def record_score(self, score: int):
self.score_history.append(score)
```
## 六、实战案例：全自动交付一个待办事项API
### 6.1 准备需求文档
创建 `specs/task-api-spec.md`：
```markdown
# Task Management API - Project Specification
## Overview
Build a production-ready RESTful Task Management API.
## Technical Requirements
- Runtime: Node.js 20 LTS + TypeScript 5.x
- Framework: Express.js 4.x
- Database: PostgreSQL 16
- ORM: Drizzle ORM
- Validation: Zod
- Testing: Vitest + Supertest
- Containerization: Docker Compose
## API Endpoints
### Tasks
- POST /api/tasks - Create task
- GET /api/tasks - List tasks (pagination + filters)
- GET /api/tasks/:id - Get task by ID
- PUT /api/tasks/:id - Update task
- DELETE /api/tasks/:id - Delete task
### Query Parameters for GET /api/tasks
- status: filter by status (pending|in_progress|completed)
- priority: filter by priority (low|medium|high)
- page: page number (default: 1)
- limit: items per page (default: 20, max: 100)
## Data Model
Task {
id: UUID (auto-generated)
title: string (required, max 200 chars)
description: string (optional, max 2000 chars)
status: enum [pending, in_progress, completed]
priority: enum [low, medium, high]
due_date: timestamp (optional)
created_at: timestamp (auto)
updated_at: timestamp (auto)
}
## Non-Functional Requirements
- Input validation on all endpoints
- Proper HTTP status codes
- Error response format: {error: string, details?: any}
- Database migrations
- Health check endpoint: GET /health
- Docker Compose for local dev
- Test coverage > 80%
```
### 6.2 运行Harness
```bash
# 方式1: 直接运行
python main.py
# 方式2: 从检查点恢复（如果中途中断）
python recover.py --checkpoint ./output/task-api/checkpoints
# 方式3: 无头模式后台运行（长时间任务推荐）
nohup python main.py > harness.log 2>&1 &
tail -f harness.log
```
### 6.3 预期输出
```
[Harness] 启动项目: Task Management API - Project Specification...
[Harness] Phase 1: 任务规划中...
[Harness] 规划完成: 8 个任务
[Harness] Phase 2: 执行任务 1/8
→ Initialize project scaffold with package.json, tsconfig, Docker
[迭代 1/5]
质量分: 85/100
✓ 通过!
[Harness] Phase 2: 执行任务 2/8
→ Define database schema and Drizzle ORM configuration
[迭代 1/5]
质量分: 72/100
✗ 未通过，进入下一轮迭代...
[迭代 2/5]
质量分: 88/100
✓ 通过!
[Harness] Phase 2: 执行任务 3/8
→ Implement task repository layer with CRUD operations
...
[Harness] Phase 3: 最终集成验证...
============================================================
HARNESS EXECUTION REPORT
============================================================
总耗时: 847.3s (14.1min)
任务完成: 8/8
任务失败: 0/8
总迭代次数: 11
构建状态: ✓
测试状态: ✓
最终质量分: 91/100
============================================================
```
### 6.4 生成的项目结构
```
output/task-api/
├── src/
│ ├── index.ts # 入口，Express配置
│ ├── db/
│ │ ├── schema.ts # Drizzle ORM schema定义
│ │ ├── client.ts # 数据库连接
│ │ └── migrations/ # 迁移文件
│ ├── routes/
│ │ └── tasks.ts # 任务路由
│ ├── services/
│ │ └── taskService.ts # 业务逻辑层
│ ├── middleware/
│ │ ├── validation.ts # Zod验证中间件
│ │ └── errorHandler.ts # 统一错误处理
│ └── types/
│ └── task.ts # TypeScript类型定义
├── tests/
│ ├── setup.ts # 测试配置
│ ├── routes/
│ │ └── tasks.test.ts # API集成测试
│ └── services/
│ └── taskService.test.ts # 单元测试
├── docker-compose.yml
├── Dockerfile
├── package.json
├── tsconfig.json
├── vitest.config.ts
└── CLAUDE.md # 项目约定文档
```
## 七、进阶：生产级增强
### 7.1 添加通知Hook
当Harness完成或失败时自动通知：
```python
# harness/hooks.py
import os
import subprocess
def notify_completion(report: dict):
"""Harness完成时发送桌面通知"""
summary = (
f"Claude Harness完成\n"
f"任务: {report['completed']}/{report['total']}\n"
f"质量分: {report['score']}/100"
)
# macOS
if os.name != "nt" and os.uname().sysname == "Darwin":
subprocess.run([
"osascript", "-e",
f'display notification "{summary}" with title "Claude Harness"'
])
# Linux
elif os.name != "nt":
subprocess.run(["notify-send", "Claude Harness", summary])
def notify_failure(task_id: str, reason: str):
"""任务失败时发送Slack/钉钉通知"""
webhook_url = os.getenv("NOTIFICATION_WEBHOOK")
if not webhook_url:
return
import urllib.request
payload = json.dumps({
"text": f"⚠️ Harness任务 {task_id} 失败: {reason}"
}).encode()
req = urllib.request.Request(
webhook_url,
data=payload,
headers={"Content-Type": "application/json"}
)
urllib.request.urlopen(req)
```
### 7.2 多模型策略
为不同Agent选择不同模型以优化成本：
```python
MODEL_STRATEGY = {
# Planner需要强推理能力，使用Sonnet
"planner": "claude-sonnet-4-6",
# Generator需要编码能力，Sonnet性价比最优
"generator": "claude-sonnet-4-6",
# Evaluator需要细致分析，可以复用Sonnet
"evaluator": "claude-sonnet-4-6",
# 如果预算充足，关键任务可以用Opus
# "planner": "claude-opus-4-6",
# "generator": "claude-opus-4-6",
}
# 成本估算（基于典型Token使用）
ESTIMATED_COST_PER_TASK = {
"claude-opus-4-6": 0.35, # ~$0.35/任务
"claude-sonnet-4-6": 0.04, # ~$0.04/任务
"claude-haiku-4-5": 0.005, # ~$0.005/任务
}
def estimate_project_cost(task_count: int,
model: str = "claude-sonnet-4-6",
avg_iterations: float = 1.5) -> float:
"""估算项目总成本"""
cost_per = ESTIMATED_COST_PER_TASK[model]
# 每个任务涉及3个Agent，平均1.5次迭代
return task_count * avg_iterations * 3 * cost_per
```
### 7.3 并行任务执行
无依赖关系的任务可以并行执行以加速交付：
```python
async def _execute_parallel(self):
"""并行执行无依赖的任务"""
# 构建依赖图
dep_graph = {}
for task in self.task_list:
deps = task.get("dependencies", [])
dep_graph[task["task_id"]] = set(
int(d) for d in deps
)
completed_ids = set()
while dep_graph:
# 找出所有依赖已满足的任务
ready = [
tid for tid, deps in dep_graph.items()
if deps.issubset(completed_ids)
]
if not ready:
print("[Harness] 死锁! 无法继续执行")
break
# 并行执行就绪任务（最多3个并发）
batches = [ready[i:i+3] for i in range(0, len(ready), 3)]
for batch in batches:
tasks = [
t for t in self.task_list
if t["task_id"] in batch
]
# 使用独立的Generator实例并行执行
results = await asyncio.gather(*[
self._execute_task_with_eval_loop(task)
for task in tasks
])
for task, success in zip(tasks, results):
tid = task["task_id"]
dep_graph.pop(tid, None)
if success:
completed_ids.add(tid)
self.completed_tasks.append(tid)
else:
self.failed_tasks.append(tid)
```
## 八、Harness执行生命周期总览
```mermaid
sequenceDiagram
participant U as 开发者
participant O as Orchestrator
participant P as Planner
participant G as Generator
participant E as Evaluator
participant FS as 文件系统
U->>O: 提交项目需求
O->>P: 解析需求，生成任务列表
P-->>O: 返回JSON任务列表
loop 每个任务
O->>O: 保存检查点
O->>G: 发送任务描述+约束
loop 生成-评估循环 (max 5次)
G->>FS: 读取现有代码
G->>FS: 写入/修改文件
G->>FS: 运行测试
G-->>O: 实现完成
O->>E: 发送代码进行审查
E->>FS: 读取变更文件
E->>FS: 运行测试
E-->>O: 返回评分和问题
alt 评分 >= 80
O->>O: 任务完成，保存检查点
else 评分 < 80
O->>G: 发送反馈，重新实现
end
end
end
O->>E: 最终集成验证
E-->>O: 最终评分
O-->>U: 执行报告
```
## 九、成本与时间估算
基于实际测试数据（使用claude-sonnet-4-6）：
| 项目规模 | 任务数 | 平均迭代 | 总耗时 | 总成本 | 成功率 |
|----------|--------|---------|--------|--------|--------|
| 小型（CRUD API） | 6-8 | 1.3x | 8-15min | $0.20-0.50 | ~90% |
| 中型（全栈应用） | 12-16 | 1.5x | 25-40min | $0.80-1.50 | ~80% |
| 大型（微服务） | 20-30 | 1.8x | 60-120min | $2.00-5.00 | ~65% |
**提高成功率的技巧：**
1. 需求文档越精确，Planner输出越准确
2. 每个任务的target_files尽量明确
3. 为项目准备详细的CLAUDE.md
4. 使用并行执行处理独立任务
5. 设置合理的质量阈值（80分足够，追求95分浪费迭代）
## 延伸阅读
- Anthropic官方Harness Engineering博客：两篇论文分别介绍了基础概念和全栈应用实践
- Claude Agent SDK文档：深入了解Agent SDK的API和工具能力
- claude-quickstarts/autonomous-coding仓库：Anthropic官方的自主编码示例代码
- Claude Code Sessions文档：--continue和--resume命令的完整参考
- InfoQ对Anthropic三Agent Harness的深度报道
## 训练营适用性评估
| 维度 | 评估 |
|------|------|
| 适合训练营 | AI工程化训练营、高级开发者训练营 |
| 学员基础 | 需要Python异步编程基础 + Claude Code使用经验 |
| 教学时长 | 3-4小时（含动手实验） |
| 实验环境 | 需要Anthropic API Key（预算约$2/人） |
| 课后作业 | 用Harness自动交付一个自己的小型项目 |
## 参考来源
1. [Effective harnesses for long-running agents - Anthropic Engineering](https://www.anthropic.com/engineering/effective-harnesses-for-long-running-agents) — Anthropic官方第一篇Harness方法论论文，提出Generator-Evaluator对抗模式
2. [Harness design for long-running application development - Anthropic](https://www.anthropic.com/engineering/harness-design-long-running-apps) — 第二篇论文，扩展为三Agent架构并应用于全栈开发
3. [claude-quickstarts/autonomous-coding - GitHub](https://github.com/anthropics/claude-quickstarts/tree/main/autonomous-coding) — Anthropic官方自主编码示例，包含可运行的Agent SDK Harness代码
4. [Agent SDK overview - Claude Code Docs](https://code.claude.com/docs/en/agent-sdk/overview) — Claude Agent SDK官方文档，Agent编程接口完整参考
5. [Manage sessions - Claude Code Docs](https://code.claude.com/docs/en/sessions) — Session管理文档，--continue/--resume命令详解
> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。
