agent-evolve 是一个面向智能体工程化实践的教学与实验项目。
项目通过对比一个故意保留八大故障点的 Naive Agent，逐步构建具备循环控制、工具治理、上下文管理、外部验证、权限门禁、预算控制、进度审计和断点恢复能力的 Harness Agent。
当前工程不是简单复制示例代码，而是把各项机制真正接入同一条 Agent 主循环，并通过自动化测试验证关键闭环。
Naive Agent 能够调用大模型和工具，但缺少稳定运行所需的工程约束。本项目重点治理以下八类故障：
| 故障编号 | 故障模式 | Naive Agent 中的表现 | Harness 治理方向 |
|---|---|---|---|
| ① | 循环失控 | while True 无最大轮次，可能无限调用模型和工具 |
最大轮次、明确终止状态 |
| ② | Context 溢出 | 消息持续追加，没有压缩、摘要和窗口控制 | Token 估算、历史摘要、完整消息块保留 |
| ③ | 上下文基线漂移 | 每轮修改 System Prompt，破坏提示词一致性 | 固定 System Prompt、恢复一致性校验 |
| ④ | Tool 错误吞没 | 工具异常被转换为空字符串，模型无法判断失败原因 | 结构化错误、异常类型和原因回写 |
| ⑤ | 状态丢失 | 进程中断后无法恢复，已完成步骤可能重放 | 原子 Checkpoint、resume=True |
| ⑥ | 权限缺口 | 工具参数、文件路径和危险命令没有独立检查 | Permission Gate、参数校验、Workspace 沙箱 |
| ⑦ | 缺少自动化评审 | 模型自行声称完成，没有外部验证 | pytest Verifier、Generator-Evaluator |
| ⑧ | 成本失控 | 没有 Token、费用、子 Agent 深度和轮次限制 | Token Budget、费用上限、子 Agent 硬限制 |
flowchart LR
User["用户任务"] --> Demo["demo.py"]
Demo --> Agent["HarnessAgent"]
Agent --> Context["Context Management"]
Agent --> Budget["Token Budget"]
Agent --> Hooks["Hooks"]
Agent --> Tools["Tool Registry"]
Agent --> Verifier["Verification Loop"]
Agent --> Checkpoint["Checkpoint Recovery"]
Hooks --> Permission["Permission Gate"]
Hooks --> Progress["Progress Tracking"]
Tools --> Planner["Feature List"]
Tools --> Workspace["Workspace Tools"]
Agent -. 可选扩展 .-> Subagent["Subagents"]
Agent -. 可选扩展 .-> Evaluator["Generator-Evaluator"]
- 单一主循环：真正的 Agent Loop 只存在于
HarnessAgent.run()。 - 机制必须接入闭环：上下文压缩、预算、权限和验证不是孤立工具函数。
- 安全判断外置：模型不能自行决定是否有权限执行危险操作。
- 完成必须可验证：模型声称完成不等于任务完成。
- 进度与恢复分离：Progress 用于审计，Checkpoint 用于精确恢复。
- 扩展状态实例化：Planner 使用会话实例，不使用模块全局变量。
- 依赖可注入：模型客户端、Verifier 和 Hooks 均可替换，便于测试。
agent-evolve/
├── .env # DeepSeek/OpenAI-compatible 配置
├── demo.py # 主工程运行入口
├── pyproject.toml # Python 项目与依赖配置
├── uv.lock # uv 依赖锁文件
├── README.md # 项目总说明
├── HARNESS_MECHANISM_MATRIX.md # 机制—故障—文件详细矩阵
├── HARNESS_REVIEW.md # 示例代码审查记录
│
├── naive_agent/
│ ├── __init__.py
│ ├── naive_agent_demo.py # 故意保留八大故障点的对照实现
│ └── test_calculator.py # 教学用计算器与测试
│
├── harness_agent/
│ ├── __init__.py # 公共 API
│ ├── agent.py # 唯一 Agent 主循环
│ ├── budget.py # Token 与费用硬预算
│ ├── checkpoint.py # 原子快照与断点恢复
│ ├── config.py # 集中配置与环境变量加载
│ ├── context.py # 上下文估算、分组和压缩
│ ├── core.py # 函数式兼容入口
│ ├── evaluator.py # Generator-Evaluator
│ ├── hooks.py # 生命周期事件与安全 Gate
│ ├── permission.py # 危险工具和命令权限策略
│ ├── planner.py # 会话级 Feature List
│ ├── progress.py # JSONL 进度与审计日志
│ ├── subagent.py # 子 Agent 委托
│ ├── tools.py # 工具注册、Schema、参数校验和沙箱
│ └── verifier.py # 独立 pytest 验证器
│
└── tests/
├── __init__.py
├── test_harness_agent.py # 八大故障治理测试
├── test_harness_extensions.py # 扩展机制及主循环集成测试
└── test_verifier_real.py # 真实 pytest 子进程测试
当前工程在参考表的 11 个机制基础上，增加了独立的 Checkpoint Recovery，共计 12 个机制。
| 编号 | 机制名称 | 主要治理故障 | 实现文件 | 实现状态 |
|---|---|---|---|---|
| ① | Agent Loop | ① 循环失控、③ Prompt 漂移 | agent.py、core.py |
已接入主循环 |
| ② | Tool Use | ④ Tool 错误吞没、⑥ 参数缺口 | tools.py、agent.py |
已接入主循环 |
| ③ | Progress Tracking | ⑤ 执行过程不可见、缺少审计 | progress.py、hooks.py |
已通过 Hook 接入 |
| ④ | Context Management | ② Context 溢出、③ 基线漂移 | context.py、agent.py |
已接入每轮模型调用前 |
| ⑤ | Feature List | ② 长任务膨胀、⑤ 任务进度丢失 | planner.py、demo.py |
已注册为 todo 工具 |
| ⑥ | Verification Loop | ⑦ 模型自判完成、④ 下游错误 | verifier.py、agent.py |
已作为完成门禁 |
| ⑦ | Subagents | ② 父上下文膨胀、⑧ 子任务失控 | subagent.py、tools.py |
可选扩展 |
| ⑧ | Generator-Evaluator | ⑦ 缺少自动化候选评审 | evaluator.py |
可选扩展 |
| ⑨ | Permission Gate | ⑥ 权限缺口 | permission.py、tools.py、agent.py |
已通过 Gate 接入 |
| ⑩ | Hooks | 贯穿全部机制 | hooks.py、agent.py |
已接入生命周期 |
| ⑪ | Token Budget | ⑧ 成本失控 | budget.py、agent.py |
已接入调用前后 |
| ⑫ | Checkpoint Recovery | ⑤ 状态丢失 | checkpoint.py、agent.py |
已支持恢复 |
更详细的机制矩阵见 HARNESS_MECHANISM_MATRIX.md。
下面的代码分别展示 Naive Agent 中八类故障的典型表现，以及当前 Harness 的解决方式。 示例默认已经准备好以下基础对象：
from harness_agent import (
HarnessAgent,
HarnessConfig,
HookManager,
PermissionGate,
create_workspace_tools,
pytest_verifier,
)
config = HarnessConfig.from_env()
tools = create_workspace_tools(
config.workspace,
timeout=config.tool_timeout_seconds,
)
verifier = pytest_verifier(
config.workspace,
"tests",
timeout=config.tool_timeout_seconds,
)# 没有最大轮次；如果模型一直返回 tool_calls，循环不会结束。
while True:
response = client.chat.completions.create(...)
execute_tool_calls(response)config.max_steps = 10
agent = HarnessAgent(
config=config,
tools=tools,
verifier=verifier,
)
result = agent.run("检查并修复测试")
if result["status"] == "max_steps":
print("达到最大轮次，已安全停止")
print("已执行轮次：", result["steps"])HarnessAgent.run() 的主循环为：
while steps < self.config.max_steps:
...达到上限后返回结构化状态，而不是继续消耗 API：
{
"status": "max_steps",
"answer": "达到最大执行轮次，任务尚未通过外部验证",
"steps": 10
}# 每轮都追加 Assistant 和 Tool 消息，但从不清理历史。
messages.append(assistant_message)
messages.append(tool_message)Agent 每次调用模型前都会执行上下文压缩：
messages = compress_messages(
messages,
config.context_token_limit,
config.keep_recent_blocks,
)也可以单独使用上下文模块：
from harness_agent.context import compress_messages, estimate_tokens
before_tokens = estimate_tokens(messages)
compressed_messages = compress_messages(
messages,
token_limit=12_000,
keep_recent_blocks=6,
)
after_tokens = estimate_tokens(compressed_messages)
print("压缩前：", before_tokens)
print("压缩后：", after_tokens)压缩过程会固定保留：
System Prompt
原始 User Goal
较早历史摘要
最近若干完整交互消息块
Assistant Tool Calls 和对应 Tool Results 会被视为一个不可拆分的消息块：
[
{
"role": "assistant",
"tool_calls": [
{
"id": "call-1",
"function": {
"name": "read_file",
"arguments": "{\"path\": \"demo.py\"}",
},
}
],
},
{
"role": "tool",
"tool_call_id": "call-1",
"content": "...",
},
]因此不会因为压缩而产生缺失 tool_call_id 的协议错误。
while True:
# 每轮修改 System Prompt，使模型行为基线不断变化。
messages[0]["content"] = (
f"你是编程助手。[session_iter={iteration}]"
)System Prompt 只在创建 Agent 时设置一次：
FIXED_SYSTEM_PROMPT = """你是一个严谨的编程智能体。
只能在配置的工作区中操作。
必须通过外部验证后才能声称任务完成。"""
agent = HarnessAgent(
config=config,
tools=tools,
verifier=verifier,
system_prompt=FIXED_SYSTEM_PROMPT,
)
result = agent.run("修复测试")主循环只读取 self.system_prompt，不会在每轮修改它。恢复 Checkpoint 时还会校验：
if (
state.get("task") != task
or state.get("system_prompt") != self.system_prompt
):
raise ValueError("checkpoint 与当前任务或系统提示词不匹配")这能避免使用其他任务或其他 Prompt 的旧状态继续运行。
try:
result = tool(**arguments)
except Exception:
# 模型无法区分空文件、权限错误和工具异常。
result = ""所有工具统一返回 ToolResult：
from harness_agent import create_workspace_tools
registry = create_workspace_tools(config.workspace)
result = registry.execute(
"read_file",
{"path": "not-exists.txt"},
)
print(result.ok)
print(result.error_type)
print(result.error)
print(result.to_json())预期结果类似：
{
"ok": false,
"value": null,
"error": "[Errno 2] No such file or directory: 'not-exists.txt'",
"error_type": "FileNotFoundError"
}模型可以根据 error_type 选择正确动作，例如：
FileNotFoundError：重新确认路径；InvalidArguments：修正工具参数；PermissionDenied：停止危险操作；TimeoutExpired：缩小任务范围或调整策略。
工具名和参数也会在调用前进行校验：
unknown = registry.execute("unknown_tool", {})
wrong_args = registry.execute(
"read_file",
{"path": "README.md", "unexpected": True},
)
assert unknown.error_type == "UnknownTool"
assert wrong_args.error_type == "InvalidArguments"messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": task},
]
# 进程退出后 messages、steps 和 token 全部丢失。
run_loop(messages)配置 Checkpoint 路径：
from pathlib import Path
config.checkpoint_path = (
Path(config.workspace)
/ ".harness"
/ "checkpoint.json"
)首次运行：
agent = HarnessAgent(
config=config,
tools=tools,
verifier=verifier,
)
result = agent.run("完成一个多轮代码任务")
print(result["status"])如果任务因为最大轮次、预算或异常而中断，Checkpoint 会保存：
任务描述
固定 System Prompt
完整 Messages
已执行轮次
输入和输出 Token
外部验证失败次数
恢复同一个任务：
resumed_result = agent.run(
"完成一个多轮代码任务",
resume=True,
)
print(resumed_result["steps"])
print(resumed_result["status"])Checkpoint 使用临时文件和原子替换：
temporary_path.write_text(checkpoint_json, encoding="utf-8")
os.replace(temporary_path, checkpoint_path)因此进程在写入期间退出时，不会轻易留下半个 JSON 文件。
def write_file(path: str, content: str) -> str:
# 模型可以传入任意绝对路径或 ../ 路径。
with open(path, "w", encoding="utf-8") as file:
file.write(content)
return "done"registry = create_workspace_tools(config.workspace)
escaped = registry.execute(
"write_file",
{
"path": "../outside.txt",
"content": "不允许写出工作区",
},
)
assert escaped.ok is False
assert escaped.error_type == "PermissionError"路径会先执行 resolve()，然后检查真实路径是否仍位于 Workspace 内：
candidate = (workspace / requested_path).resolve()
candidate.relative_to(workspace.resolve())该检查同时覆盖：
../路径逃逸；- 工作区外绝对路径；
- 已存在符号链接指向工作区外部。
hooks = HookManager()
gate = PermissionGate()
gate.register_to(hooks)
allowed, reason = hooks.trigger_gate(
"pre_tool_use",
"run_bash",
{"command": "rm -rf /"},
)
assert allowed is False
print(reason)将 Hooks 接入 Agent：
agent = HarnessAgent(
config=config,
tools=tools,
verifier=verifier,
hooks=hooks,
)实际工具执行顺序为：
JSON 解析
→ pre_tool_use Gate
→ PermissionGate
→ 函数签名绑定
→ 基础类型校验
→ Workspace 路径检查
→ 实际执行
如果 Permission Handler 自身异常，Gate 会采用 Fail-Closed：
allowed = False
reason = "权限 Hook 执行失败"if not message.tool_calls:
# 模型没有调用工具，不代表代码和测试真的正确。
return message.content创建独立 pytest Verifier：
verifier = pytest_verifier(
config.workspace,
test_path="tests",
timeout=60,
)
agent = HarnessAgent(
config=config,
tools=tools,
verifier=verifier,
)
result = agent.run("修复所有失败测试")完成判定变为：
flowchart LR
A["模型声称完成"] --> B["执行独立 pytest"]
B --> C{"pytest 是否通过?"}
C -- 否 --> D["把失败输出反馈给模型"]
D --> E["模型继续修复"]
E --> B
C -- 是 --> F["返回 completed"]
验证失败时，Harness 会追加新的 User 消息：
{
"role": "user",
"content": (
"外部验证失败，请根据以下客观结果继续修复，"
"验证通过前不要声称任务完成：..."
),
}只有 Verifier 返回：
VerificationResult(
passed=True,
summary="pytest 验证通过",
details={"exit_code": 0},
)Agent 才返回：
{
"status": "completed",
"verification": {
"passed": true,
"summary": "pytest 验证通过"
}
}while True:
# 没有轮次、Token 或费用上限。
response = client.chat.completions.create(...)配置 Token 硬上限：
config.max_steps = 20
config.max_total_tokens = 50_000可选配置费用上限：
# 以下价格仅为配置示例，请按实际模型价格填写。
config.max_cost_usd = 0.50
config.input_cost_per_1k = 0.001
config.output_cost_per_1k = 0.002Agent 在模型调用前检查预计输入：
budget.check_before_call(
estimate_tokens(messages)
)模型返回后再记录真实用量：
budget.record(
input_tokens=response.usage.prompt_tokens,
output_tokens=response.usage.completion_tokens,
)如果供应商没有返回 usage，Harness 仍会执行本地估算，不能借此绕过预算。
超出预算时不会继续调用模型：
result = agent.run("执行受预算控制的任务")
if result["status"] == "budget_exceeded":
print(result["budget"])
print(result["errors"])返回示例：
{
"status": "budget_exceeded",
"budget": {
"input_tokens": 42000,
"output_tokens": 9000,
"total_tokens": 51000,
"cost_usd": 0.06,
"max_total_tokens": 50000,
"max_cost_usd": 0.5
}
}子 Agent 同样可以设置独立限制：
from harness_agent import delegate
subtask_result = delegate(
"分析测试覆盖缺口",
config=config,
tools=tools,
allowed_tools=["read_file"],
max_steps=5,
max_total_tokens=10_000,
max_depth=2,
)下面的代码将循环、上下文、工具、安全、进度、验证、预算和恢复机制组合到同一个 Agent 中：
from harness_agent import (
HarnessAgent,
HarnessConfig,
HookManager,
PermissionGate,
PlanTracker,
ProgressTracker,
create_workspace_tools,
pytest_verifier,
)
# 1. 加载并设置硬限制
config = HarnessConfig.from_env()
config.max_steps = 20
config.max_total_tokens = 100_000
config.context_token_limit = 12_000
config.keep_recent_blocks = 6
config.verification_attempts = 2
# 2. 创建 Workspace 沙箱工具
tools = create_workspace_tools(
config.workspace,
timeout=config.tool_timeout_seconds,
)
# 3. 注册会话级 Feature List
planner = PlanTracker()
tools.register(
planner.tool,
name="todo",
description="查询、更新或清空当前任务计划",
)
# 4. 创建生命周期 Hooks
hooks = HookManager()
# 5. 注册权限门禁
permission_gate = PermissionGate()
permission_gate.register_to(hooks)
# 6. 注册结构化进度日志
progress = ProgressTracker(
config.workspace / ".harness" / "progress.jsonl"
)
progress.register_to(hooks)
# 7. 创建独立 pytest 验证器
verifier = pytest_verifier(
config.workspace,
test_path="tests",
timeout=config.tool_timeout_seconds,
)
# 8. 创建并运行 Harness Agent
agent = HarnessAgent(
config=config,
tools=tools,
verifier=verifier,
hooks=hooks,
)
result = agent.run(
"先制定计划，再检查当前工程测试；如果失败则修复，"
"直到外部 pytest 验证通过。",
resume=False,
)
print("状态：", result["status"])
print("轮次：", result["steps"])
print("预算：", result["budget"])
print("验证：", result["verification"])
print("错误：", result["errors"])如果任务之前中断，可以使用完全相同的任务描述恢复：
result = agent.run(
"先制定计划，再检查当前工程测试；如果失败则修复，"
"直到外部 pytest 验证通过。",
resume=True,
)该组合对应关系如下：
| 故障点 | 生效机制 | 关键配置或对象 |
|---|---|---|
| ① 循环失控 | Agent Loop | config.max_steps |
| ② Context 溢出 | Context Management | context_token_limit、keep_recent_blocks |
| ③ Prompt 漂移 | Stable System Prompt | HarnessAgent.system_prompt |
| ④ Tool 错误吞没 | Tool Use | ToolRegistry、ToolResult |
| ⑤ 状态丢失 | Checkpoint Recovery、Progress | checkpoint_path、ProgressTracker |
| ⑥ 权限缺口 | Permission Gate、Workspace Sandbox | PermissionGate、create_workspace_tools |
| ⑦ 模型自判完成 | Verification Loop | pytest_verifier |
| ⑧ 成本失控 | Token Budget | max_total_tokens、max_cost_usd |
flowchart TD
Start["启动任务"] --> Init["初始化消息、预算和错误状态"]
Init --> Resume{"resume=True 且存在 Checkpoint?"}
Resume -- 是 --> Load["恢复 Messages、轮次、预算和验证次数"]
Resume -- 否 --> Session["触发 session_start"]
Load --> Session
Session --> Limit{"steps < max_steps?"}
Limit -- 否 --> Max["返回 max_steps"]
Limit -- 是 --> Pre["触发 pre_iteration"]
Pre --> Compress["压缩过长上下文"]
Compress --> PreBudget["调用前 Token 预算检查"]
PreBudget --> LLM["调用大模型"]
LLM --> Usage["记录真实或估算 Token"]
Usage --> ToolDecision{"存在 Tool Calls?"}
ToolDecision -- 是 --> Parse["解析 JSON 参数"]
Parse --> Gate["触发 pre_tool_use Gate"]
Gate --> Allowed{"权限是否放行?"}
Allowed -- 否 --> Denied["生成 PermissionDenied"]
Allowed -- 是 --> Execute["签名校验并执行工具"]
Denied --> ToolResult["结构化 ToolResult 回写"]
Execute --> ToolResult
ToolResult --> Progress["触发 post_tool_use"]
Progress --> Save["原子保存 Checkpoint"]
Save --> Limit
ToolDecision -- 否 --> Verify{"配置了 Verifier?"}
Verify -- 否 --> Complete["返回 completed"]
Verify -- 是 --> External["执行独立外部验证"]
External --> Passed{"验证通过?"}
Passed -- 是 --> Clear["清理 Checkpoint"]
Clear --> Complete
Passed -- 否 --> Attempts{"超过验证重试次数?"}
Attempts -- 是 --> Failed["返回 verification_failed"]
Attempts -- 否 --> Feedback["把客观失败结果反馈给模型"]
Feedback --> Save
| 状态 | 含义 | 是否保留 Checkpoint |
|---|---|---|
completed |
模型完成且外部验证通过，或配置允许不验证 | 验证成功时清理 |
max_steps |
达到最大执行轮次 | 是 |
budget_exceeded |
Token 或费用超过硬预算 | 是 |
verification_failed |
达到外部验证失败次数上限 | 是 |
error |
未预期异常 | 是 |
flowchart LR
Call["模型 Tool Call"] --> Json["JSON 解析"]
Json --> Hook["pre_tool_use Hook"]
Hook --> Permission["Permission Gate"]
Permission --> Signature["函数签名绑定"]
Signature --> Type["基础类型检查"]
Type --> Sandbox["Workspace 路径沙箱"]
Sandbox --> Run["执行工具"]
Run --> Result["ToolResult JSON"]
Result --> Model["结果回写模型"]
无论工具成功还是失败，都会返回统一结构：
{
"ok": false,
"value": null,
"error": "文件不存在",
"error_type": "FileNotFoundError"
}这使模型可以区分：
- 工具成功；
- 参数不合法；
- 工具不存在；
- 权限被拒绝；
- 文件不存在；
- 运行超时；
- 其他明确异常。
create_workspace_tools() 当前提供：
| 工具 | 功能 | 安全措施 |
|---|---|---|
read_file |
读取 UTF-8 文本文件 | Workspace 边界和真实路径检查 |
write_file |
原子写入 UTF-8 文本文件 | Workspace 边界、临时文件、os.replace |
run_pytest |
执行 pytest | Workspace cwd、超时、输出截断、当前解释器 |
文件路径在 resolve() 后再次检查，因此能够拦截：
../路径逃逸；- 工作区外绝对路径；
- 已存在符号链接指向工作区外部；
- 模型构造的路径绕过尝试。
当前没有内置 Shell 工具。如果后续增加命令执行能力，不应直接使用 shell=True，应继续增加 argv 白名单、环境变量过滤和人工审批策略。
HookManager 用于把权限、进度、指标和审计从主循环解耦。
| Hook 事件 | 触发时机 | 典型用途 |
|---|---|---|
session_start |
会话初始化或恢复后 | Session 日志、Tracing |
pre_iteration |
每轮模型调用前 | 轮次指标、取消检查 |
post_iteration |
模型响应和 Token 统计完成后 | Token、费用和延迟统计 |
pre_tool_use |
工具执行前 | Permission Gate、审批 |
post_tool_use |
工具执行后 | 工具审计、结果记录 |
verification |
外部验证完成后 | 测试结果和质量指标 |
checkpoint_saved |
Checkpoint 落盘后 | 恢复点审计 |
session_stop |
任意终止状态返回前 | 最终状态、错误和预算汇总 |
普通 Hook 异常会被隔离，不会中断其他 Handler；pre_tool_use 是安全 Gate，Handler 异常时采用 Fail-Closed，默认拒绝工具执行。
这两个机制职责不同，不能互相替代。
| 机制 | 文件 | 面向对象 | 保存内容 | 主要用途 |
|---|---|---|---|---|
| Progress Tracking | .harness/progress.jsonl |
人类和分析程序 | 生命周期事件、工具、验证、终止状态 | 审计、观察、统计 |
| Checkpoint Recovery | .harness/checkpoint.json |
Agent 恢复逻辑 | Messages、轮次、预算、验证失败次数 | 中断恢复、避免重放 |
{
"version": 1,
"task": "用户任务",
"system_prompt": "固定系统提示词",
"messages": [],
"steps": 3,
"budget": {
"input_tokens": 1000,
"output_tokens": 300,
"total_tokens": 1300
},
"verification_failures": 1
}Checkpoint 使用临时文件和 os.replace() 原子替换，避免程序在写入中途退出后留下残缺 JSON。
恢复执行：
result = agent.run(
"继续原来的任务",
resume=True,
)恢复时会校验任务和 System Prompt，防止把其他任务的快照错误加载到当前会话。
Python >= 3.13
| 依赖 | 用途 |
|---|---|
openai |
调用 DeepSeek/OpenAI-compatible Chat Completions |
python-dotenv |
从 .env 加载配置 |
pytest |
自动化测试和外部验证 |
cd /Users/jean/PycharmProjects/agent-evolve
uv sync如果虚拟环境已创建，可直接使用：
.venv/bin/python --version
.venv/bin/pytest --version在工程根目录创建或修改 .env：
DEEPSEEK_API_KEY=your_api_key
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_MODEL_NAME=deepseek-chat
# 以下为可选 Harness 配置
HARNESS_WORKSPACE=/Users/jean/PycharmProjects/agent-evolve
HARNESS_MAX_STEPS=20
HARNESS_MAX_TOTAL_TOKENS=100000
HARNESS_CONTEXT_TOKEN_LIMIT=12000
HARNESS_CHECKPOINT_PATH=/Users/jean/PycharmProjects/agent-evolve/.harness/checkpoint.json| 配置项 | 默认值 | 说明 |
|---|---|---|
model_name |
deepseek-chat |
模型名称 |
workspace |
当前目录 | 文件和测试工具允许操作的根目录 |
max_steps |
20 |
Agent 最大循环轮次 |
max_total_tokens |
100000 |
单 Session 累计 Token 上限 |
max_cost_usd |
None |
可选美元费用上限 |
input_cost_per_1k |
0.0 |
每千输入 Token 费用 |
output_cost_per_1k |
0.0 |
每千输出 Token 费用 |
context_token_limit |
12000 |
触发上下文压缩的估算阈值 |
keep_recent_blocks |
6 |
压缩时保留的近期完整消息块数量 |
tool_timeout_seconds |
60 |
工具和 pytest 超时 |
verification_attempts |
2 |
外部验证失败重试次数 |
require_verification |
True |
是否强制要求 Verifier |
checkpoint_path |
.harness/checkpoint.json |
Checkpoint 路径 |
temperature |
0.0 |
主 Agent 模型温度 |
input_cost_per_1k和output_cost_per_1k不会自动读取供应商实时价格，需要根据实际模型自行配置。
cd /Users/jean/PycharmProjects/agent-evolve
.venv/bin/python demo.pydemo.py 已接入：
- Workspace Tools；
- PlanTracker；
- HookManager；
- PermissionGate；
- ProgressTracker；
- pytest Verifier；
- Token Budget；
- Checkpoint Recovery。
from harness_agent import (
HarnessAgent,
HarnessConfig,
create_workspace_tools,
pytest_verifier,
)
config = HarnessConfig.from_env()
tools = create_workspace_tools(
config.workspace,
timeout=config.tool_timeout_seconds,
)
verifier = pytest_verifier(
config.workspace,
"tests",
timeout=config.tool_timeout_seconds,
)
agent = HarnessAgent(
config=config,
tools=tools,
verifier=verifier,
)
result = agent.run("检查并修复测试，直到 pytest 通过")
print(result["status"])
print(result["answer"])core.py 仅提供函数式入口，内部仍然复用 HarnessAgent：
from harness_agent import HarnessConfig, create_workspace_tools, run_agent
config = HarnessConfig.from_env()
config.require_verification = False
tools = create_workspace_tools(config.workspace)
result = run_agent(
"分析当前工程",
tools,
config,
verifier=None,
)如果不传 Verifier，必须在创建 Agent 前显式设置
config.require_verification = False。
from harness_agent import (
HarnessAgent,
HarnessConfig,
HookManager,
PermissionGate,
PlanTracker,
ProgressTracker,
create_workspace_tools,
pytest_verifier,
)
config = HarnessConfig.from_env()
tools = create_workspace_tools(config.workspace)
# 会话级计划工具
planner = PlanTracker()
tools.register(
planner.tool,
name="todo",
description="查询、更新或清空当前任务计划",
)
# 生命周期扩展
hooks = HookManager()
PermissionGate().register_to(hooks)
ProgressTracker(
config.workspace / ".harness" / "progress.jsonl"
).register_to(hooks)
agent = HarnessAgent(
config=config,
tools=tools,
verifier=pytest_verifier(config.workspace, "tests"),
hooks=hooks,
)
result = agent.run("先制定计划，再检查并修复测试")PlanTracker 是会话级计划管理器，不使用全局 TODOS。
planner = PlanTracker()
tools.register(planner.tool, name="todo")planner.tool(
action="update",
items=[
{
"id": "1",
"content": "检查测试失败原因",
"status": "in_progress",
},
{
"id": "2",
"content": "修改实现并重新验证",
"status": "pending",
},
],
)planner.tool(
action="update",
merge=True,
items=[
{
"id": "1",
"content": "检查测试失败原因",
"status": "completed",
}
],
)有效状态：
pending
in_progress
completed
blocked
from harness_agent import HarnessConfig, evaluate
config = HarnessConfig.from_env()
result = evaluate(
candidates=[
"方案 A：直接修改原函数",
"方案 B：提取独立服务并增加测试",
],
rubric="正确性、可维护性、安全性和改动范围",
config=config,
)
print(result["best_index"])
print(result["scores"])
print(result["reasoning"])返回结构：
{
"best_index": 1,
"scores": [0.5, 0.9],
"reasoning": "第二个方案具备更好的测试边界",
"error": null
}Evaluator 会校验：
- 分数数量是否与候选数量一致；
- 分数是否位于
0到1； best_index是否越界；- 返回内容是否为合法 JSON 对象。
解析失败时会显式设置 error，不会把回退结果伪装成可靠评分。
from harness_agent import HarnessConfig, create_workspace_tools, delegate
config = HarnessConfig.from_env()
tools = create_workspace_tools(config.workspace)
result = delegate(
"只读分析 tests 目录中的测试覆盖缺口",
config=config,
tools=tools,
context="父任务正在评估 Harness 测试完整性",
allowed_tools=["read_file"],
max_steps=5,
max_total_tokens=10000,
max_depth=2,
)子 Agent 具备：
- 独立消息历史；
- 独立 System Prompt；
- 工具白名单；
- 最大轮次限制；
- Token 硬上限；
- 递归深度限制；
- 不覆盖父 Agent Checkpoint。
当前子 Agent 使用独立预算上限，但尚未自动聚合到父 Agent 的 BudgetGuard，详见“当前限制”。
cd /Users/jean/PycharmProjects/agent-evolve
.venv/bin/pytest -q -p no:cacheprovider当前测试结果：
19 passed
| 测试文件 | 覆盖内容 |
|---|---|
test_harness_agent.py |
最大轮次、上下文压缩、Prompt 稳定、结构化错误、Checkpoint、沙箱、Verifier、Budget、Tool Schema |
test_harness_extensions.py |
Hooks、Permission Gate、Progress、Planner、Evaluator、Subagent、core 兼容入口 |
test_verifier_real.py |
使用当前 Python 解释器执行真实 pytest 子进程 |
.venv/bin/python -m compileall -q demo.py harness_agent tests| 维度 | 参考示例 | 当前工程 |
|---|---|---|
| 主循环 | core.py 中的大函数 |
agent.py 中的 HarnessAgent，core.py 仅兼容 |
| 上下文压缩 | 模块存在，但未接入主循环 | 每轮模型调用前强制接入 |
| 完成验证 | 主要依赖 Verification Prompt | 独立 Verifier 是完成门禁 |
| pytest 解释器 | 硬编码 python |
使用 sys.executable |
| Progress | Markdown 人类日志 | 结构化 JSONL 事件 |
| 恢复能力 | 没有真正的会话恢复 | 原子 JSON Checkpoint 和 resume=True |
| Planner | 模块全局 TODOS |
实例级 PlanTracker |
| Permission | 命令正则，需要人工接线 | Hook Gate + 命令策略 + ToolRegistry + Workspace 沙箱 |
| Evaluator | 自己创建 Client，解析失败默认均分 | Client 可注入、结果严格校验、显式错误 |
| Subagent | 直接递归调用主函数 | 独立上下文、工具白名单、深度和预算限制 |
| Budget | usage 缺失时可能绕过 | usage 缺失时也会估算并计入 |
当前版本已经覆盖基础 Harness 和主要扩展机制，但仍有以下边界：
- 没有自动重试策略：429、网络中断和部分 5xx 尚未进行有限重试。
- Token 为启发式估算：模型未返回 usage 时使用字符数除以 4，不等同于精确 tokenizer。
- 价格不会自动更新：美元费用依赖手动配置的千 Token 单价。
- Planner 未自动写入 Agent Checkpoint：
PlanTracker提供snapshot()和restore()，但尚未注册为通用状态提供者。 - 子 Agent 预算未聚合到父预算：子 Agent 有独立硬限制，但父子总成本尚未统一汇总。
- Hooks 为同步执行：长耗时 Hook 会增加主循环延迟。
- 没有内置 Shell 工具：当前 PermissionGate 已准备命令策略，但工程默认不暴露命令执行能力。
- 没有文件变更回滚：原子写入避免半文件，但验证失败时不会自动恢复旧版本。
- 没有并发 Subagent 调度器：当前
delegate()为同步、单子任务执行。 - 没有真实 API 自动化测试：测试默认使用 FakeClient，避免产生外部费用。
flowchart LR
A["当前 Harness"] --> B["模型调用 Retry Policy"]
B --> C["统一 Model Gateway"]
C --> D["Planner 状态接入 Checkpoint"]
D --> E["父子 Agent 共享总预算"]
E --> F["文件 Diff 与失败回滚"]
F --> G["并发 Subagent 调度"]
G --> H["Tracing 与指标面板"]
建议优先级：
- 增加带指数退避和 jitter 的有限重试；
- 将所有模型调用统一封装为 Model Gateway；
- 建立可注册的 Session State Provider；
- 将 Planner 和其他扩展状态纳入 Checkpoint；
- 建立父子 Agent 共享预算和取消信号；
- 增加文件 Diff、变更集和失败回滚；
- 加入并发 Subagent 上限和任务调度；
- 接入 OpenTelemetry 或结构化指标系统。
- HARNESS_MECHANISM_MATRIX.md：十二机制与八大故障详细对应关系。
- HARNESS_REVIEW.md：参考示例代码、测试和架构问题审查。
- naive_agent/naive_agent_demo.py：保留八大故障点的对照代码。
- demo.py：当前完整 Harness 的运行入口。
该项目当前用于智能体工程探索和教学验证。即使已经加入 Workspace 沙箱、Permission Gate、预算和验证器，也不应在未经额外审计的情况下直接赋予智能体：
- 生产服务器权限；
- 云平台管理员凭据；
- 数据库写权限；
- 无限制 Shell；
- 工作区之外的文件权限；
- 无上限 API 额度。
在生产场景中，应继续加入人工审批、最小权限凭据、隔离容器、网络出口控制、审计存储和事故回滚机制。