文章摘要：Ralph Loop 把 Coding Agent 放进持续迭代中，直到测试等外部条件通过。它真正有价值的不是 while true，而是状态外置、单轮增量、验证反馈压缩、无进展检测和明确预算。
所属系列：ReAct 与 Agent Loop 工程（4/6）
上一篇用 Plan-and-Execute 管理多个步骤；本篇换一个更窄的场景：目标和验收命令已经明确，怎样让 Coding Agent 一轮轮修改代码，直到外部验证通过。
第 2 篇的 Harness 管单次 Coding Agent 调用内部的工具、预算和权限；Ralph runner 管多次调用之间的反馈、验证、检查点和停止条件。
最原始的循环可以短到三行：
while :; do
agent-cli run –prompt-file PROMPT.md
done
但如果只抄这个形式，得到的很可能是烧钱和失控。真正有效的是循环外面的工程条件。
为什么重复运行可能持续进步
普通聊天的状态主要在上下文里。Coding Loop 的状态更多地留在真实环境：代码写进仓库，测试结果进入日志，Git 保存差异，下一轮读取更新后的项目。
读取仓库 -> 选择一个增量 -> 修改 -> 外部验证 -> 压缩失败反馈
^ |
+——————————————–+
模型并不是重新回答同一道题，而是在新的工作区状态和新的验证结果上继续推进。
验证失败应该怎样回传
把 2000 行 stdout 原样塞回下一轮通常会污染上下文。更稳妥的做法是分两层保存：
- 完整日志落盘，供人排查和审计；
- 下一轮只收到本轮失败命令、失败类型、关键首尾日志、重复错误签名和完整日志路径；
- 历史轮次只保留结构化摘要，不重复注入所有旧日志；
- 相同错误签名连续出现时触发“无进展”或人工接管。
默认可把反馈控制在约 100 到 200 行、8 到 12 KB。这个数不是协议标准，而是工程起点，应该按模型上下文、日志噪声和错误密度调整。
一个可落地的运行器骨架
下面的代码把三个原本容易悬空的适配器补齐：工作区指纹、Coding Agent 子进程和验证器。替换命令后即可接入自己的项目。
import asyncio
import hashlib
import os
import shlex
from dataclasses import dataclass
from enum import StrEnum
from pathlib import Path
from time import monotonic
class FailureType(StrEnum):
TEST = "test"
LINT = "lint"
BUILD = "build"
TIMEOUT = "timeout"
DEPENDENCY = "dependency"
AUTH = "auth"
PERMISSION = "permission"
INFRA = "infra"
UNKNOWN = "unknown"
@dataclass
class VerifierResult:
passed: bool
failure_type: FailureType | None
summary: str
full_log_path: str
signature: str
@dataclass
class LoopBudget:
max_attempts: int = 20
max_seconds: int = 3600
max_no_progress: int = 3
agent_timeout: int = 900
verifier_timeout: int = 600
EXCLUDED_PARTS = {
".git", ".agent-loop", ".venv", "node_modules",
"dist", "build", "coverage", "__pycache__",
}
async def run_process(
args: list[str],
cwd: Path,
timeout: int,
stdin_text: str | None = None,
) –> tuple[int, bytes]:
process = await asyncio.create_subprocess_exec(
*args,
cwd=cwd,
stdin=asyncio.subprocess.PIPE if stdin_text else None,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
)
try:
stdout, _ = await asyncio.wait_for(
process.communicate(
stdin_text.encode("utf-8") if stdin_text else None
),
timeout=timeout,
)
except TimeoutError:
process.kill()
await process.wait()
raise
return process.returncode or 0, stdout
async def workspace_fingerprint(root: Path) –> str:
"""哈希 tracked diff 与未跟踪源码；排除构建产物和循环日志。"""
digest = hashlib.sha256()
code, diff = await run_process(
["git", "diff", "–binary", "HEAD"], root, timeout=60
)
if code != 0:
raise RuntimeError("无法读取 git diff")
digest.update(diff)
code, raw_names = await run_process(
["git", "ls-files", "–others", "–exclude-standard", "-z"],
root,
timeout=60,
)
if code != 0:
raise RuntimeError("无法读取未跟踪文件")
for raw_name in sorted(filter(None, raw_names.split(b"\\0"))):
relative = Path(raw_name.decode("utf-8", errors="surrogateescape"))
if any(part in EXCLUDED_PARTS for part in relative.parts):
continue
target = (root / relative).resolve()
if target.is_file() and root.resolve() in target.parents:
digest.update(raw_name)
digest.update(target.read_bytes())
return digest.hexdigest()
def build_increment_prompt(
goal: str,
attempt: int,
previous: VerifierResult | None,
) –> str:
feedback = "尚无失败反馈"
if previous:
feedback = (
f"失败类型：{previous.failure_type}\\n"
f"错误签名：{previous.signature}\\n"
f"关键日志：\\n{previous.summary}\\n"
f"完整日志：{previous.full_log_path}"
)
return (
f"总目标：{goal}\\n"
f"当前轮次：{attempt}\\n"
"本轮只完成一个最小、可验证增量；先检查现有 diff。"
"不要重写无关模块，不要覆盖用户已有修改，验证通过后停止。\\n"
f"上一轮反馈：\\n{feedback}"
)
async def run_coding_agent(
root: Path,
prompt: str,
timeout: int,
) –> None:
command = os.getenv("CODING_AGENT_CMD")
if not command:
raise RuntimeError("请设置 CODING_AGENT_CMD 为你实际使用的 Agent CLI")
code, output = await run_process(
shlex.split(command), root, timeout=timeout, stdin_text=prompt
)
if code != 0:
tail = output.decode("utf-8", errors="replace")[–2000:]
raise RuntimeError(f"Coding Agent 退出码为 {code}: {tail}")
def classify_failure(command: list[str], text: str) –> FailureType:
lowered = text.lower()
if "permission denied" in lowered:
return FailureType.PERMISSION
if "unauthorized" in lowered or "invalid api key" in lowered:
return FailureType.AUTH
if "could not resolve" in lowered or "no matching distribution" in lowered:
return FailureType.DEPENDENCY
executable = Path(command[0]).name.lower()
if executable in {"pytest", "jest", "vitest"}:
return FailureType.TEST
if executable in {"ruff", "eslint", "mypy"}:
return FailureType.LINT
if executable in {"npm", "pnpm", "cargo", "gradle"}:
return FailureType.BUILD
return FailureType.UNKNOWN
def compact_log(text: str, max_lines: int = 160, max_chars: int = 12_000) –> str:
lines = text.splitlines()
if len(lines) > max_lines:
half = max_lines // 2
lines = lines[:half] + ["… 中间日志已截断 …"] + lines[–half:]
return "\\n".join(lines)[:max_chars]
async def run_verifier(
root: Path,
attempt: int,
commands: list[list[str]],
timeout: int,
) –> VerifierResult:
log_dir = root / ".agent-loop" / "logs"
log_dir.mkdir(parents=True, exist_ok=True)
log_path = log_dir / f"attempt-{attempt:03d}-verify.log"
chunks: list[str] = []
for command in commands:
label = shlex.join(command)
try:
code, raw = await run_process(command, root, timeout=timeout)
text = raw.decode("utf-8", errors="replace")
except TimeoutError:
text = f"$ {label}\\nTIMEOUT after {timeout}s"
chunks.append(text)
log_path.write_text("\\n\\n".join(chunks), encoding="utf-8")
return VerifierResult(
False,
FailureType.TIMEOUT,
compact_log(text),
str(log_path),
hashlib.sha256(text.encode()).hexdigest()[:16],
)
chunks.append(f"$ {label}\\n{text}")
if code != 0:
full = "\\n\\n".join(chunks)
log_path.write_text(full, encoding="utf-8")
failure_type = classify_failure(command, text)
signature_source = f"{failure_type}:{label}:{compact_log(text, 40, 3000)}"
return VerifierResult(
False,
failure_type,
compact_log(chunks[–1]),
str(log_path),
hashlib.sha256(signature_source.encode()).hexdigest()[:16],
)
full = "\\n\\n".join(chunks)
log_path.write_text(full, encoding="utf-8")
return VerifierResult(True, None, "all checks passed", str(log_path), "passed")
async def run_loop(root: Path, goal: str, budget: LoopBudget) –> str:
started = monotonic()
no_progress = 0
previous: VerifierResult | None = None
verifier_commands = [
["pytest", "-q"],
["ruff", "check", "."],
]
for attempt in range(1, budget.max_attempts + 1):
if monotonic() – started > budget.max_seconds:
return "stopped: time_budget"
before = await workspace_fingerprint(root)
prompt = build_increment_prompt(goal, attempt, previous)
await run_coding_agent(root, prompt, budget.agent_timeout)
verdict = await run_verifier(
root, attempt, verifier_commands, budget.verifier_timeout
)
if verdict.passed:
return "completed: verification_passed"
if verdict.failure_type in {
FailureType.AUTH,
FailureType.PERMISSION,
FailureType.DEPENDENCY,
FailureType.INFRA,
}:
return f"stopped: requires_human:{verdict.failure_type}"
after = await workspace_fingerprint(root)
repeated_error = previous and previous.signature == verdict.signature
no_progress = no_progress + 1 if after == before or repeated_error else 0
if no_progress >= budget.max_no_progress:
return "stopped: no_progress"
previous = verdict
return "stopped: attempt_budget"
这不是“复制就能适配所有仓库”的框架，但关键行为已经明确：完整日志去哪、摘要怎样生成、什么错误继续循环、什么错误应该停，以及工作区变化怎样进入停止条件。
工作区指纹不是 git diff –stat
git diff –stat 只反映文件和行数概况，不足以判断两轮内容是否相同。上面的最小实现哈希了：
- 相对 HEAD 的 tracked 二进制 diff；
- git ls-files –others –exclude-standard 找到的未跟踪文件内容；
- 排除了 .agent-loop、构建目录、虚拟环境等会自行变化的产物。
大型仓库不一定适合每轮哈希所有未跟踪文件，可以限制到任务涉及目录。但不能把验证日志本身算进指纹，否则每轮都会看起来“有进展”。
单轮只做一个增量
反面提示：
重构整个认证模块，顺便提高性能并补齐测试。
更可执行的提示：
只修复 test_token_expiry。
优先检查 src/auth/token.py；除非测试证明必要，不改其他模块。
运行该测试，记录结果，通过后停止本轮。
增量越小，失败反馈越容易归因，Git 回退越安全，下一轮上下文也越干净。
Git 检查点怎么做
生产环境更推荐独立 worktree 或任务分支：
“每轮失败都 reset”会丢掉有价值的中间修复；“从不设检查点”又会让错误长期累积。两者都不理想。
Ralph 适合什么任务
它适合目标明确、验证便宜、失败可恢复的工作，例如修复测试、迁移 API、升级依赖和清理静态检查错误。
不适合直接套循环的场景包括付款退款、删除生产数据、开放式战略判断、真实设备控制，以及任何失败不可逆或权限边界模糊的任务。
小结
Ralph Loop 不是神奇提示词，也不是让模型永远别停。它更像一个由外部验证驱动的迭代运行器：
关键结论：环境保存状态，Agent 提交小增量，验证器压缩反馈并决定是否结束。
参考资料
下一篇：当工具不再只是本地函数，MCP 应该负责哪一层，又不该负责哪一层？
