Loop Engineering深度解析：从Prompt到循环——2026年AI工程第四次范式跃迁的完整实战指南

2026年6月，一条推文引爆了整个AI开发圈："你不该再给编程Agent写提示词了，你应该设计一套循环机制（Loops），让这些循环去提示你的Agent。"——这就是Loop Engineering（循环工程），继Prompt Engineering、Context Engineering、Harness Engineering之后，AI工程方法论的第四次范式跃迁。

一、你和AI的协作方式，该升级了

1.1 现状：人是最大的瓶颈

2026年的今天，AI编程Agent已经强大到令人惊叹——Claude Code能读懂整个代码库，Codex能自主完成端到端的功能开发，Cursor能在你打字之前就预测你要写什么。但如果你仔细观察任何一个"高效使用AI"的开发者的工作流，你会发现一个荒诞的事实：

人，依然是整个流程中最慢的环节。

典型的一天是这样的：

你：帮我重构这个模块的错误处理
AI：（花了30秒生成了200行代码）
你：（花了5分钟review）嗯...这个try-catch的位置不太对
你：把第47行的catch移到外层
AI：（花了10秒修改）
你：（花了3分钟检查）现在跑一下测试
AI：（跑了2分钟测试）3个测试失败了
你：（花了8分钟分析失败原因）修一下第3个测试的断言
AI：（花了5秒修改）
你：再跑一遍
AI：全部通过
你：（松了口气，看了看时间——40分钟过去了，改了12行代码）

这就是2026年大多数开发者使用AI的真实写照。AI的执行速度以秒计，但人的review、决策、指令传递以分钟甚至小时计。你不是在用AI，你是在排队等AI等你。

1.2 四次范式跃迁：从"怎么问"到"怎么让它自己跑"

回顾过去三年，AI工程经历了四次根本性的范式转变：

每一次跃迁都不是对前一次的否定，而是在更高维度上的叠加。Prompt Engineering没有消亡——它被吸收进了Context Engineering的System Prompt设计中；Context Engineering没有消亡——它变成了Harness Engineering的上下文管理层；Harness Engineering也没有消亡——它成为了Loop Engineering中"循环环境"的核心组成部分。

Loop Engineering不是"更好的提示词"，而是"不再需要你亲自写提示词"。

二、Loop Engineering：定义、起源与核心思想

2.1 定义

Loop Engineering（循环工程）是一种AI工程范式，其核心主张是：

人类退出持续提示的位置，转去设计一个能够持续调度、约束和验证AI的"循环系统"。

在这个系统中，AI像流水线一样自主运转——自动发现工作、分配任务、检查结果并决定下一步。开发者的角色从"逐句指挥者"升级为"系统设计者"。

用一个比喻来说：

• Prompt Engineering = 你开车，手动挡，每一步都要踩离合换挡
• Context Engineering = 你给车装了GPS，但它还是手动挡
• Harness Engineering = 你给车装了自动变速箱，但方向盘还在你手里
• Loop Engineering = 自动驾驶。你设定目的地和安全规则，车自己开

2.2 起源

2026年6月，Loop Engineering的概念在AI社区迅速传播，几位关键人物几乎同时提出了相同的观点：

• Addy Osmani（Google工程师）：系统整理了Loop Engineering的理论框架
• Boris Cherny（Anthropic Claude Code负责人）：在公开场合阐述了Agent循环的设计理念
• Peter Steinberger（OpenClaw创始人）：在一条引发800万次浏览的推文中正式提出了这一范式

他们不约而同地指向同一个结论：当AI Agent已经能独立完成单个任务后，新的瓶颈不再是Agent的能力，而是人-Agent之间的交互模式本身。

2.3 核心思想：三个转变

转变一：从"单次调用"到"持续循环"

传统模式是请求-响应式的：你问一句，AI答一句。Loop模式是持续运转的：你启动一个循环，它自己跑到目标达成。

转变二：从"人工判断"到"可验证的终止条件"

传统模式中，每一次AI输出都需要人来判断"好了没有"。Loop模式中，你预先定义好"什么叫做完了"——通过测试、lint检查、类型检查、性能基准等可机器验证的条件来自动判断。

转变三：从"人在回路中"到"人在回路上"

传统模式中，人是循环的一部分（必须参与每一轮迭代）。Loop模式中，人站在循环之外，只负责设计循环规则和处理异常情况。

三、Loop的五大核心模块

一个完整的Loop Engineering系统由五个核心模块组成，它们形成一个闭合的循环：

┌─────────────────────────────────────────┐
│                                         │
│   ┌──────────┐                          │
│   │ Discover │ ←── 任务发现             │
│   └────┬─────┘                          │
│        ↓                                │
│   ┌──────────┐                          │
│   │   Plan   │ ←── 策略规划             │
│   └────┬─────┘                          │
│        ↓                                │
│   ┌──────────┐                          │
│   │ Execute  │ ←── 执行实施             │
│   └────┬─────┘                          │
│        ↓                                │
│   ┌──────────┐                          │
│   │  Verify  │ ←── 验证评估             │
│   └────┬─────┘                          │
│        ↓                                │
│   ┌──────────┐                          │
│   │ Iterate  │ ←── 迭代优化             │
│   └────┬─────┘                          │
│        │                                │
│        └──── 目标达成？──→ 否 → 回到 Discover
│                  │                       │
│                  是 → 结束               │
└─────────────────────────────────────────┘

3.1 Discover（发现）：让AI自己找活干

Discover模块负责识别当前需要完成的工作。它不是简单的"读取任务列表"，而是能够从多个信息源自动发现待处理的工作：

• 代码库分析：扫描TODO、FIXME、HACK注释
• 测试覆盖分析：识别未被测试覆盖的代码路径
• Issue/PR监控：从GitHub Issues中提取可执行的任务
• 依赖更新检测：发现过时的依赖和安全漏洞
• 代码质量扫描：通过静态分析发现潜在问题

import subprocess
import json
from pathlib import Path

class DiscoverModule:
    """Loop的第一阶段：自动发现待处理任务"""
    
    def __init__(self, repo_path: str):
        self.repo_path = Path(repo_path)
    
    def discover_tasks(self) -> list[dict]:
        """从多个来源聚合待处理任务"""
        tasks = []
        
        # 1. 扫描代码中的TODO/FIXME
        tasks.extend(self._scan_code_markers())
        
        # 2. 分析测试覆盖率缺口
        tasks.extend(self._find_uncovered_code())
        
        # 3. 检查失败的测试
        tasks.extend(self._find_failing_tests())
        
        # 4. Lint错误
        tasks.extend(self._find_lint_errors())
        
        # 5. 按优先级排序
        return self._prioritize(tasks)
    
    def _scan_code_markers(self) -> list[dict]:
        """扫描TODO、FIXME、HACK等标记"""
        tasks = []
        result = subprocess.run(
            ["grep", "-rn", "--include=*.py", 
             "-E", "TODO|FIXME|HACK|XXX", 
             str(self.repo_path)],
            capture_output=True, text=True
        )
        for line in result.stdout.strip().split('\n'):
            if not line:
                continue
            parts = line.split(':', 2)
            if len(parts) >= 3:
                tasks.append({
                    'type': 'code_marker',
                    'priority': 'medium',
                    'file': parts[0],
                    'line': parts[1],
                    'description': parts[2].strip(),
                    'source': 'grep'
                })
        return tasks
    
    def _find_failing_tests(self) -> list[dict]:
        """运行测试套件，捕获失败的测试"""
        result = subprocess.run(
            ["python", "-m", "pytest", "--tb=no", "-q"],
            capture_output=True, text=True,
            cwd=self.repo_path
        )
        tasks = []
        for line in result.stdout.split('\n'):
            if 'FAILED' in line:
                test_name = line.split('::')[1] if '::' in line else line
                tasks.append({
                    'type': 'failing_test',
                    'priority': 'high',
                    'description': f'修复失败的测试: {test_name.strip()}',
                    'source': 'pytest'
                })
        return tasks
    
    def _find_uncovered_code(self) -> list[dict]:
        """通过覆盖率报告找到未覆盖的代码"""
        # 简化示例：实际项目中会解析coverage.json
        coverage_file = self.repo_path / 'coverage.json'
        if not coverage_file.exists():
            return []
        with open(coverage_file) as f:
            data = json.load(f)
        tasks = []
        for file_path, file_data in data.get('files', {}).items():
            missing = file_data.get('missing_lines', [])
            if missing:
                tasks.append({
                    'type': 'coverage_gap',
                    'priority': 'medium',
                    'file': file_path,
                    'description': f'{file_path} 有 {len(missing)} 行未覆盖',
                    'source': 'coverage'
                })
        return tasks
    
    def _find_lint_errors(self) -> list[dict]:
        """运行linter找到代码质量问题"""
        result = subprocess.run(
            ["ruff", "check", "--output-format=json", str(self.repo_path)],
            capture_output=True, text=True
        )
        tasks = []
        try:
            errors = json.loads(result.stdout) if result.stdout else []
            for error in errors[:10]:  # 限制每次最多处理10个
                tasks.append({
                    'type': 'lint_error',
                    'priority': 'low',
                    'file': error.get('filename', ''),
                    'line': error.get('location', {}).get('row', 0),
                    'description': f"{error.get('code', '')}: {error.get('message', '')}",
                    'source': 'ruff'
                })
        except json.JSONDecodeError:
            pass
        return tasks
    
    def _prioritize(self, tasks: list[dict]) -> list[dict]:
        """按优先级排序：high > medium > low"""
        priority_order = {'high': 0, 'medium': 1, 'low': 2}
        return sorted(tasks, key=lambda t: priority_order.get(t.get('priority', 'low'), 3))

3.2 Plan（计划）：让AI自己制定执行策略

Plan模块接收Discover阶段发现的任务，制定具体的执行策略。这不是简单的"把任务丢给AI"，而是需要考虑：

• 任务分解：将复杂任务拆解为可执行的原子操作
• 依赖分析：识别任务之间的依赖关系
• 资源评估：估算每个任务需要的时间和计算资源
• 风险评估：识别可能出错的环节并制定应对策略

class PlanModule:
    """Loop的第二阶段：AI自主制定执行计划"""
    
    def __init__(self, llm_client, repo_context: str):
        self.llm = llm_client
        self.repo_context = repo_context
    
    async def create_plan(self, task: dict) -> dict:
        """为单个任务创建详细的执行计划"""
        
        prompt = f"""你是一个资深软件工程师。根据以下任务信息，制定一个详细的执行计划。

## 仓库上下文
{self.repo_context}

## 任务信息
- 类型: {task['type']}
- 描述: {task['description']}
- 相关文件: {task.get('file', '未知')}
- 优先级: {task.get('priority', 'medium')}

## 要求
请输出一个JSON格式的执行计划，包含：
1. steps: 具体的执行步骤列表
2. estimated_time: 预估耗时（分钟）
3. risk_level: 风险等级（low/medium/high）
4. rollback_plan: 如果失败的回滚策略
5. verification: 如何验证执行结果

只输出JSON，不要其他内容。"""

        response = await self.llm.chat.completions.create(
            model="claude-sonnet-4-20250514",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        plan = json.loads(response.choices[0].message.content)
        
        # 注入原始任务信息
        plan['original_task'] = task
        plan['plan_created_at'] = __import__('datetime').datetime.now().isoformat()
        
        return plan
    
    async def create_batch_plan(self, tasks: list[dict], max_parallel: int = 3) -> list[dict]:
        """为多个任务创建批量执行计划，支持并行"""
        plans = []
        batch = []
        
        for task in tasks:
            plan = await self.create_plan(task)
            
            # 检查是否与当前批次有依赖冲突
            if self._has_dependency_conflict(plan, batch):
                plans.append({'type': 'deferred', 'plan': plan, 'reason': 'dependency_conflict'})
            else:
                batch.append(plan)
                
                # 达到并行上限时，开始新批次
                if len(batch) >= max_parallel:
                    plans.append({'type': 'batch', 'plans': batch.copy()})
                    batch = []
        
        if batch:
            plans.append({'type': 'batch', 'plans': batch})
        
        return plans
    
    def _has_dependency_conflict(self, new_plan: dict, current_batch: list) -> bool:
        """检查新计划是否与当前批次中的任务存在文件冲突"""
        new_files = set()
        for step in new_plan.get('steps', []):
            if 'file' in step:
                new_files.add(step['file'])
        
        for existing_plan in current_batch:
            existing_files = set()
            for step in existing_plan.get('steps', []):
                if 'file' in step:
                    existing_files.add(step['file'])
            if new_files & existing_files:  # 文件交集
                return True
        
        return False

3.3 Execute（执行）：安全的AI自主操作

Execute模块是Loop的核心执行引擎。它需要在安全的环境中执行AI生成的操作，并确保每一步都可追踪、可回滚。

import asyncio
import shutil
from datetime import datetime
from dataclasses import dataclass, field

@dataclass
class ExecutionResult:
    """执行结果数据结构"""
    step_id: str
    status: str  # 'success' | 'failed' | 'skipped'
    output: str
    error: str | None = None
    files_changed: list[str] = field(default_factory=list)
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    duration_ms: int = 0

class ExecuteModule:
    """Loop的第三阶段：安全执行AI生成的操作"""
    
    def __init__(self, repo_path: str, sandbox: bool = True):
        self.repo_path = Path(repo_path)
        self.sandbox = sandbox
        self.execution_log: list[ExecutionResult] = []
    
    async def execute_plan(self, plan: dict) -> list[ExecutionResult]:
        """执行一个完整的计划"""
        results = []
        
        # 如果启用沙箱，先创建Git checkpoint
        if self.sandbox:
            checkpoint = self._create_checkpoint()
        
        try:
            for i, step in enumerate(plan.get('steps', [])):
                result = await self._execute_step(step, step_id=f"step_{i}")
                results.append(result)
                self.execution_log.append(result)
                
                # 如果某步失败且没有容错标记，停止执行
                if result.status == 'failed' and not step.get('continue_on_error', False):
                    break
            
            return results
            
        except Exception as e:
            # 执行出错，回滚到checkpoint
            if self.sandbox:
                self._rollback_to_checkpoint(checkpoint)
            raise
    
    def _create_checkpoint(self) -> str:
        """创建Git checkpoint用于回滚"""
        # 先暂存所有修改
        subprocess.run(["git", "stash", "push", "-m", "loop-checkpoint"],
                      cwd=self.repo_path, capture_output=True)
        
        # 记录当前HEAD
        result = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            capture_output=True, text=True, cwd=self.repo_path
        )
        return result.stdout.strip()
    
    def _rollback_to_checkpoint(self, checkpoint: str):
        """回滚到指定的checkpoint"""
        subprocess.run(
            ["git", "reset", "--hard", checkpoint],
            cwd=self.repo_path, capture_output=True
        )
        subprocess.run(
            ["git", "stash", "pop"],
            cwd=self.repo_path, capture_output=True
        )
    
    async def _execute_step(self, step: dict, step_id: str) -> ExecutionResult:
        """执行单个步骤"""
        start_time = __import__('time').time()
        action = step.get('action', '')
        
        try:
            if action == 'edit_file':
                return await self._edit_file(step, step_id, start_time)
            elif action == 'run_command':
                return await self._run_command(step, step_id, start_time)
            elif action == 'run_tests':
                return await self._run_tests(step, step_id, start_time)
            elif action == 'ai_generate':
                return await self._ai_generate(step, step_id, start_time)
            else:
                return ExecutionResult(
                    step_id=step_id, status='failed',
                    output='', error=f'Unknown action: {action}',
                    duration_ms=int((__import__('time').time() - start_time) * 1000)
                )
        except Exception as e:
            return ExecutionResult(
                step_id=step_id, status='failed',
                output='', error=str(e),
                duration_ms=int((__import__('time').time() - start_time) * 1000)
            )
    
    async def _edit_file(self, step: dict, step_id: str, start_time: float) -> ExecutionResult:
        """执行文件编辑操作"""
        file_path = step.get('file', '')
        old_text = step.get('old_text', '')
        new_text = step.get('new_text', '')
        
        full_path = self.repo_path / file_path
        if not full_path.exists():
            return ExecutionResult(
                step_id=step_id, status='failed',
                output='', error=f'File not found: {file_path}',
                duration_ms=int((__import__('time').time() - start_time) * 1000)
            )
        
        content = full_path.read_text()
        if old_text not in content:
            return ExecutionResult(
                step_id=step_id, status='failed',
                output='', error=f'Text not found in {file_path}',
                duration_ms=int((__import__('time').time() - start_time) * 1000)
            )
        
        new_content = content.replace(old_text, new_text, 1)
        full_path.write_text(new_content)
        
        return ExecutionResult(
            step_id=step_id, status='success',
            output=f'Edited {file_path}',
            files_changed=[file_path],
            duration_ms=int((__import__('time').time() - start_time) * 1000)
        )
    
    async def _run_command(self, step: dict, step_id: str, start_time: float) -> ExecutionResult:
        """执行shell命令"""
        command = step.get('command', '')
        timeout = step.get('timeout', 300)
        
        proc = await asyncio.create_subprocess_shell(
            command,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            cwd=self.repo_path
        )
        
        try:
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
            return ExecutionResult(
                step_id=step_id,
                status='success' if proc.returncode == 0 else 'failed',
                output=stdout.decode()[:5000],
                error=stderr.decode()[:2000] if proc.returncode != 0 else None,
                duration_ms=int((__import__('time').time() - start_time) * 1000)
            )
        except asyncio.TimeoutError:
            proc.kill()
            return ExecutionResult(
                step_id=step_id, status='failed',
                output='', error=f'Command timed out after {timeout}s',
                duration_ms=int((__import__('time').time() - start_time) * 1000)
            )
    
    async def _run_tests(self, step: dict, step_id: str, start_time: float) -> ExecutionResult:
        """运行测试套件"""
        test_path = step.get('test_path', '')
        command = f"python -m pytest {test_path} -v --tb=short" if test_path else "python -m pytest -v --tb=short"
        return await self._run_command({'command': command, 'timeout': step.get('timeout', 600)}, step_id, start_time)
    
    async def _ai_generate(self, step: dict, step_id: str, start_time: float) -> ExecutionResult:
        """调用AI生成代码或内容"""
        # 这里会调用LLM来生成代码
        prompt = step.get('prompt', '')
        # 实际实现中会调用LLM API
        return ExecutionResult(
            step_id=step_id, status='success',
            output='AI generation completed',
            duration_ms=int((__import__('time').time() - start_time) * 1000)
        )

3.4 Verify（验证）：可机器验证的终止条件

Verify模块是Loop Engineering最关键的创新之一。它用可机器验证的条件替代了人工判断，让循环能够"知道自己什么时候该停"。

import subprocess
import json
from dataclasses import dataclass

@dataclass
class VerificationResult:
    """验证结果"""
    check_name: str
    passed: bool
    details: str
    severity: str  # 'critical' | 'warning' | 'info'

class VerifyModule:
    """Loop的第四阶段：多维度自动验证"""
    
    def __init__(self, repo_path: str, config: dict = None):
        self.repo_path = Path(repo_path)
        self.config = config or {}
    
    async def verify(self, execution_results: list[ExecutionResult]) -> tuple[bool, list[VerificationResult]]:
        """
        运行所有验证检查，返回(是否通过, 检查结果列表)
        只有所有critical检查都通过，才算整体通过
        """
        checks = [
            self._check_tests(),
            self._check_lint(),
            self._check_type_check(),
            self._check_build(),
            self._check_no_secrets(),
            self._check_code_coverage(),
        ]
        
        results = []
        for check_coro in checks:
            result = await check_coro
            results.append(result)
        
        # 只有所有critical检查都通过，才算整体通过
        all_critical_passed = all(
            r.passed for r in results 
            if r.severity == 'critical'
        )
        
        return all_critical_passed, results
    
    async def _check_tests(self) -> VerificationResult:
        """运行测试套件"""
        result = subprocess.run(
            ["python", "-m", "pytest", "-x", "--tb=short", "-q"],
            capture_output=True, text=True, cwd=self.repo_path
        )
        passed = result.returncode == 0
        return VerificationResult(
            check_name='tests',
            passed=passed,
            details=result.stdout[-2000:] if not passed else 'All tests passed',
            severity='critical'
        )
    
    async def _check_lint(self) -> VerificationResult:
        """运行代码风格检查"""
        result = subprocess.run(
            ["ruff", "check", "."],
            capture_output=True, text=True, cwd=self.repo_path
        )
        passed = result.returncode == 0
        return VerificationResult(
            check_name='lint',
            passed=passed,
            details=result.stdout[-1000:] if not passed else 'No lint errors',
            severity='critical'
        )
    
    async def _check_type_check(self) -> VerificationResult:
        """运行类型检查"""
        result = subprocess.run(
            ["mypy", ".", "--ignore-missing-imports"],
            capture_output=True, text=True, cwd=self.repo_path
        )
        # mypy的返回码：0=成功，1=发现错误，2=内部错误
        passed = result.returncode == 0
        return VerificationResult(
            check_name='type_check',
            passed=passed,
            details=result.stdout[-1000:] if not passed else 'Type check passed',
            severity='warning'
        )
    
    async def _check_build(self) -> VerificationResult:
        """验证项目是否能成功构建"""
        # 根据项目类型选择构建命令
        if (self.repo_path / 'package.json').exists():
            cmd = ['npm', 'run', 'build']
        elif (self.repo_path / 'Makefile').exists():
            cmd = ['make', 'build']
        elif (self.repo_path / 'Cargo.toml').exists():
            cmd = ['cargo', 'build']
        else:
            return VerificationResult(
                check_name='build', passed=True,
                details='No build system detected, skipping',
                severity='info'
            )
        
        result = subprocess.run(
            cmd, capture_output=True, text=True, cwd=self.repo_path
        )
        return VerificationResult(
            check_name='build',
            passed=result.returncode == 0,
            details=result.stderr[-1000:] if result.returncode != 0 else 'Build successful',
            severity='critical'
        )
    
    async def _check_no_secrets(self) -> VerificationResult:
        """检查是否意外引入了密钥或敏感信息"""
        patterns = ['password', 'secret', 'api_key', 'token', 'private_key']
        found = []
        
        for pattern in patterns:
            result = subprocess.run(
                ["grep", "-rn", "--include=*.py", "--include=*.js", 
                 "--include=*.ts", "--include=*.env",
                 "-i", pattern, "."],
                capture_output=True, text=True, cwd=self.repo_path
            )
            if result.stdout.strip():
                for line in result.stdout.strip().split('\n')[:3]:
                    # 排除注释和文档
                    if not any(skip in line for skip in ['#', '//', 'example', 'test', 'mock']):
                        found.append(line)
        
        return VerificationResult(
            check_name='no_secrets',
            passed=len(found) == 0,
            details=f'Found potential secrets: {found}' if found else 'No secrets detected',
            severity='critical'
        )
    
    async def _check_code_coverage(self) -> VerificationResult:
        """检查代码覆盖率是否达标"""
        min_coverage = self.config.get('min_coverage', 80)
        
        result = subprocess.run(
            ["python", "-m", "pytest", f"--cov-fail-under={min_coverage}", "--cov=."],
            capture_output=True, text=True, cwd=self.repo_path
        )
        
        passed = result.returncode == 0
        return VerificationResult(
            check_name='code_coverage',
            passed=passed,
            details=f'Coverage check {"passed" if passed else "failed"} (minimum: {min_coverage}%)',
            severity='warning'
        )

3.5 Iterate（迭代）：智能的循环控制

Iterate模块根据Verify阶段的结果，决定下一步的行动。这是Loop的"大脑"——它决定了循环何时继续、何时调整策略、何时终止。

from enum import Enum
from dataclasses import dataclass

class LoopAction(Enum):
    """循环动作枚举"""
    CONTINUE = "continue"      # 继续下一轮循环
    RETRY = "retry"            # 重试当前任务（可能用不同策略）
    ADJUST = "adjust"          # 调整策略后继续
    ESCALATE = "escalate"      # 升级到人工处理
    TERMINATE = "terminate"    # 终止循环

@dataclass
class LoopState:
    """循环状态"""
    iteration: int = 0
    max_iterations: int = 10
    consecutive_failures: int = 0
    max_consecutive_failures: int = 3
    total_tasks_completed: int = 0
    total_tasks_failed: int = 0
    current_strategy: str = "default"
    history: list = None
    
    def __post_init__(self):
        if self.history is None:
            self.history = []

class IterateModule:
    """Loop的第五阶段：智能循环控制与策略调整"""
    
    def __init__(self, llm_client, state: LoopState):
        self.llm = llm_client
        self.state = state
    
    async def decide_next_action(
        self,
        task: dict,
        execution_results: list[ExecutionResult],
        verification_results: list[VerificationResult]
    ) -> tuple[LoopAction, dict]:
        """
        根据执行和验证结果，决定下一步行动
        返回: (动作, 附加信息)
        """
        self.state.iteration += 1
        
        # 检查是否超过最大迭代次数
        if self.state.iteration >= self.state.max_iterations:
            return LoopAction.ESCALATE, {
                'reason': f'达到最大迭代次数 ({self.state.max_iterations})',
                'suggestion': '需要人工介入检查'
            }
        
        # 分析验证结果
        critical_failures = [
            r for r in verification_results 
            if not r.passed and r.severity == 'critical'
        ]
        warnings = [
            r for r in verification_results 
            if not r.passed and r.severity == 'warning'
        ]
        
        # 如果所有critical检查都通过
        if not critical_failures:
            self.state.consecutive_failures = 0
            self.state.total_tasks_completed += 1
            
            # 如果还有warnings，记录但继续
            if warnings:
                return LoopAction.CONTINUE, {
                    'reason': '关键检查通过，有非关键警告',
                    'warnings': [w.details for w in warnings]
                }
            
            return LoopAction.TERMINATE, {
                'reason': '所有检查通过，任务完成',
                'iterations_used': self.state.iteration
            }
        
        # 有critical失败
        self.state.consecutive_failures += 1
        self.state.total_tasks_failed += 1
        
        # 连续失败次数过多，升级到人工
        if self.state.consecutive_failures >= self.state.max_consecutive_failures:
            return LoopAction.ESCALATE, {
                'reason': f'连续 {self.state.consecutive_failures} 次失败',
                'failures': [f.details for f in critical_failures]
            }
        
        # 分析失败原因，决定是重试还是调整策略
        failure_analysis = await self._analyze_failures(critical_failures)
        
        if failure_analysis['strategy_change_needed']:
            return LoopAction.ADJUST, {
                'reason': '当前策略无效，需要调整',
                'new_strategy': failure_analysis['suggested_strategy'],
                'failure_analysis': failure_analysis
            }
        
        return LoopAction.RETRY, {
            'reason': '失败但可重试',
            'fix_hints': failure_analysis.get('fix_hints', [])
        }
    
    async def _analyze_failures(self, failures: list[VerificationResult]) -> dict:
        """用AI分析失败原因并建议策略"""
        failure_details = "\n".join([
            f"- [{f.check_name}] {f.details}" for f in failures
        ])
        
        prompt = f"""分析以下验证失败的原因，并建议下一步策略。

## 失败详情
{failure_details}

## 当前策略
{self.state.current_strategy}

## 历史尝试
{json.dumps(self.state.history[-3:], ensure_ascii=False) if self.state.history else '无'}

请用JSON格式回答：
{{
    "root_cause": "根本原因分析",
    "strategy_change_needed": true/false,
    "suggested_strategy": "建议的新策略（如果需要调整）",
    "fix_hints": ["修复提示1", "修复提示2"],
    "confidence": 0.0-1.0
}}"""

        response = await self.llm.chat.completions.create(
            model="claude-sonnet-4-20250514",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        
        return json.loads(response.choices[0].message.content)

四、完整的Loop Engine：把五大模块串起来

现在我们把五个模块组合成一个完整的Loop Engine：

import asyncio
import json
from datetime import datetime
from pathlib import Path

class LoopEngine:
    """
    Loop Engineering的核心引擎
    将Discover → Plan → Execute → Verify → Iterate 串联成完整的自主循环
    """
    
    def __init__(self, config: dict):
        self.repo_path = config['repo_path']
        self.llm_client = config['llm_client']
        self.max_iterations = config.get('max_iterations', 10)
        self.max_consecutive_failures = config.get('max_consecutive_failures', 3)
        
        # 初始化五大模块
        self.discover = DiscoverModule(self.repo_path)
        self.plan = PlanModule(self.llm_client, self._get_repo_context())
        self.execute = ExecuteModule(self.repo_path, sandbox=True)
        self.verify = VerifyModule(self.repo_path, config.get('verify', {}))
        self.iterate = IterateModule(self.llm_client, LoopState(
            max_iterations=self.max_iterations,
            max_consecutive_failures=self.max_consecutive_failures
        ))
        
        # 循环状态
        self.running = False
        self.results_log = []
    
    def _get_repo_context(self) -> str:
        """获取仓库上下文信息"""
        context_parts = []
        
        # 读取README
        readme = Path(self.repo_path) / 'README.md'
        if readme.exists():
            context_parts.append(f"## README\n{readme.read_text()[:2000]}")
        
        # 读取项目结构
        result = subprocess.run(
            ["find", ".", "-type", "f", "-name", "*.py", "|", "head", "-20"],
            capture_output=True, text=True, shell=True, cwd=self.repo_path
        )
        context_parts.append(f"## 项目文件\n{result.stdout}")
        
        return "\n\n".join(context_parts)
    
    async def run(self, target_task: str = None) -> dict:
        """
        启动Loop循环
        
        Args:
            target_task: 可选的特定目标任务。如果不指定，自动从Discover阶段获取。
        
        Returns:
            循环执行的总结报告
        """
        self.running = True
        start_time = datetime.now()
        
        print(f"🚀 Loop Engine 启动 - {start_time.strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"📁 仓库: {self.repo_path}")
        print(f"🔄 最大迭代次数: {self.max_iterations}")
        print("-" * 60)
        
        try:
            while self.running:
                iteration = self.iterate.state.iteration + 1
                print(f"\n{'='*60}")
                print(f"📍 迭代 #{iteration}")
                print(f"{'='*60}")
                
                # Phase 1: Discover
                print("\n🔍 [Discover] 发现任务...")
                if target_task:
                    tasks = [{'type': 'manual', 'description': target_task, 'priority': 'high'}]
                else:
                    tasks = self.discover.discover_tasks()
                
                if not tasks:
                    print("✅ 没有发现待处理任务，循环结束")
                    break
                
                task = tasks[0]  # 处理最高优先级的任务
                print(f"  📋 任务: {task['description'][:80]}...")
                
                # Phase 2: Plan
                print("\n📝 [Plan] 制定执行计划...")
                plan = await self.plan.create_plan(task)
                steps_count = len(plan.get('steps', []))
                print(f"  📊 计划包含 {steps_count} 个步骤")
                
                # Phase 3: Execute
                print("\n⚡ [Execute] 执行计划...")
                exec_results = await self.execute.execute_plan(plan)
                success_count = sum(1 for r in exec_results if r.status == 'success')
                print(f"  ✅ 成功: {success_count}/{len(exec_results)}")
                
                # Phase 4: Verify
                print("\n✅ [Verify] 验证结果...")
                all_passed, verify_results = await self.verify.verify(exec_results)
                
                for vr in verify_results:
                    status = "✅" if vr.passed else "❌"
                    print(f"  {status} {vr.check_name}: {vr.details[:60]}...")
                
                # Phase 5: Iterate
                print("\n🔄 [Iterate] 评估下一步...")
                action, info = await self.iterate.decide_next_action(
                    task, exec_results, verify_results
                )
                
                print(f"  🎯 决策: {action.value}")
                print(f"  💬 原因: {info.get('reason', '未知')}")
                
                # 记录本轮结果
                self.results_log.append({
                    'iteration': iteration,
                    'task': task,
                    'action': action.value,
                    'info': info,
                    'exec_results': len(exec_results),
                    'verify_passed': all_passed
                })
                
                # 根据决策执行
                if action == LoopAction.TERMINATE:
                    print(f"\n🎉 任务完成！共迭代 {iteration} 轮")
                    break
                elif action == LoopAction.ESCALATE:
                    print(f"\n⚠️ 需要人工介入: {info.get('reason', '未知原因')}")
                    self.running = False
                elif action == LoopAction.ADJUST:
                    print(f"\n🔧 调整策略: {info.get('new_strategy', '未指定')}")
                    self.iterate.state.current_strategy = info.get('new_strategy', 'default')
                # RETRY 和 CONTINUE 自动进入下一轮
                
        except KeyboardInterrupt:
            print("\n\n⏹️ 用户中断循环")
        except Exception as e:
            print(f"\n\n💥 循环异常: {e}")
            raise
        finally:
            self.running = False
        
        # 生成总结报告
        end_time = datetime.now()
        duration = (end_time - start_time).total_seconds()
        
        report = {
            'start_time': start_time.isoformat(),
            'end_time': end_time.isoformat(),
            'duration_seconds': duration,
            'total_iterations': self.iterate.state.iteration,
            'tasks_completed': self.iterate.state.total_tasks_completed,
            'tasks_failed': self.iterate.state.total_tasks_failed,
            'final_strategy': self.iterate.state.current_strategy,
            'history': self.results_log
        }
        
        # 保存报告
        report_path = Path(self.repo_path) / '.loop' / 'report.json'
        report_path.parent.mkdir(exist_ok=True)
        report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False))
        
        print(f"\n📊 报告已保存: {report_path}")
        return report


# 使用示例
async def main():
    """运行Loop Engine"""
    # 假设已初始化LLM客户端
    import anthropic
    
    config = {
        'repo_path': '/path/to/your/project',
        'llm_client': anthropic.AsyncAnthropic(),
        'max_iterations': 10,
        'max_consecutive_failures': 3,
        'verify': {
            'min_coverage': 80
        }
    }
    
    engine = LoopEngine(config)
    report = await engine.run()
    
    print(f"\n📈 最终统计:")
    print(f"  总迭代次数: {report['total_iterations']}")
    print(f"  任务完成数: {report['tasks_completed']}")
    print(f"  任务失败数: {report['tasks_failed']}")
    print(f"  总耗时: {report['duration_seconds']:.1f}秒")

if __name__ == '__main__':
    asyncio.run(main())

五、Loop Engineering的三大设计原则

5.1 原则一：可验证性优先（Verifiability First）

Loop Engineering的核心假设是：AI的输出必须能被机器验证，而不是依赖人工判断。

这意味着你在设计Loop时，必须优先考虑"如何验证结果"，而不是"如何让AI做得更好"。一个无法自动验证的Loop，本质上还是一个需要人工介入的Prompt Chain。

# ❌ 错误的Loop设计：依赖人工判断
class BadLoop:
    async def should_continue(self, result):
        # 这不是Loop，这只是把人工判断延后了
        return input("结果满意吗？(y/n): ") == 'y'

# ✅ 正确的Loop设计：可机器验证
class GoodLoop:
    async def should_continue(self, result):
        # 通过测试、lint、类型检查等自动化手段验证
        tests_pass = await self.run_tests()
        lint_clean = await self.run_lint()
        types_ok = await self.run_type_check()
        return tests_pass and lint_clean and types_ok

5.2 原则二：失败是常态，回滚是基本功（Failure is Normal, Rollback is Essential）

AI的输出天然具有不确定性。在Loop Engineering中，你必须假设每一轮迭代都可能失败，并为此做好准备：

class ResilientLoop:
    """具备回滚能力的韧性循环"""
    
    async def run_with_rollback(self, task):
        # 每轮开始前创建checkpoint
        checkpoint = self.create_checkpoint()
        
        try:
            result = await self.execute(task)
            
            if await self.verify(result):
                return result  # 成功，保留修改
            else:
                # 验证失败，回滚到checkpoint
                self.rollback(checkpoint)
                return None
                
        except Exception as e:
            # 执行异常，回滚到checkpoint
            self.rollback(checkpoint)
            raise

5.3 原则三：人在回路上，不在回路中（Human on the Loop, not in the Loop）

Loop Engineering的目标不是完全取代人类，而是把人类从"每一步都需要参与"的模式中解放出来。人类的角色变成了：

• 设计者：设计Loop的规则和约束
• 监督者：监控Loop的运行状态
• 仲裁者：处理Loop无法自动解决的异常情况

class SupervisedLoop:
    """带监督的循环系统"""
    
    async def run(self, task):
        while not self.is_complete():
            result = await self.iterate(task)
            
            # 自动处理的情况
            if result.status == 'success':
                continue
            elif result.status == 'retryable':
                await self.adjust_strategy()
                continue
            
            # 需要人工介入的情况
            if result.status == 'escalate':
                notification = self.format_notification(result)
                await self.notify_human(notification)
                
                # 等待人工决策
                human_decision = await self.wait_for_human_input()
                
                if human_decision.action == 'override':
                    self.apply_override(human_decision)
                elif human_decision.action == 'abort':
                    break

六、实战：用Loop Engineering重构一个真实项目

让我们用一个完整的实战案例来演示Loop Engineering的威力。假设我们有一个老旧的Python项目，需要进行全面的代码质量提升。

6.1 场景设定

# 项目结构
"""
legacy_project/
├── src/
│   ├── api/
│   │   ├── handlers.py      # 有很多TODO和FIXME
│   │   ├── middleware.py     # 缺少类型注解
│   │   └── routes.py        # 有重复代码
│   ├── models/
│   │   ├── user.py          # 缺少测试
│   │   └── order.py         # 有潜在的N+1查询
│   └── utils/
│       ├── validators.py    # 有安全漏洞
│       └── formatters.py    # 代码风格混乱
├── tests/
│   └── test_api.py          # 覆盖率只有40%
├── requirements.txt         # 有已知漏洞的依赖
└── README.md                # 过时的文档
"""

6.2 启动Loop

async def refactor_legacy_project():
    """用Loop Engineering重构遗留项目"""
    
    config = {
        'repo_path': './legacy_project',
        'llm_client': anthropic.AsyncAnthropic(),
        'max_iterations': 20,
        'max_consecutive_failures': 3,
        'verify': {
            'min_coverage': 85  # 目标覆盖率85%
        }
    }
    
    engine = LoopEngine(config)
    
    # 不指定具体任务，让Discover自动发现
    report = await engine.run()
    
    # 分析报告
    print("\n📊 重构报告:")
    print(f"  总迭代次数: {report['total_iterations']}")
    print(f"  完成的任务: {report['tasks_completed']}")
    print(f"  失败的任务: {report['tasks_failed']}")
    
    # 查看每轮迭代的详情
    for entry in report['history']:
        print(f"\n  迭代 #{entry['iteration']}:")
        print(f"    任务: {entry['task']['description'][:50]}...")
        print(f"    决策: {entry['action']}")
        print(f"    原因: {entry['info'].get('reason', '未知')}")

6.3 预期的Loop运行过程

🚀 Loop Engine 启动 - 2026-07-06 01:30:00
📁 仓库: ./legacy_project
🔄 最大迭代次数: 20
------------------------------------------------------------

============================================================
📍 迭代 #1
============================================================

🔍 [Discover] 发现任务...
  📋 任务: src/utils/validators.py 存在SQL注入漏洞...

📝 [Plan] 制定执行计划...
  📊 计划包含 4 个步骤

⚡ [Execute] 执行计划...
  ✅ 成功: 4/4

✅ [Verify] 验证结果...
  ✅ tests: All tests passed
  ✅ lint: No lint errors
  ✅ type_check: Type check passed
  ✅ no_secrets: No secrets detected

🔄 [Iterate] 评估下一步...
  🎯 决策: terminate
  💬 原因: 所有检查通过，任务完成

🎉 任务完成！共迭代 1 轮

============================================================
📍 迭代 #2
============================================================

🔍 [Discover] 发现任务...
  📋 任务: tests/test_api.py 覆盖率仅40%，目标85%...

📝 [Plan] 制定执行计划...
  📊 计划包含 8 个步骤

⚡ [Execute] 执行计划...
  ✅ 成功: 7/8
  ❌ 失败: 1/8

✅ [Verify] 验证结果...
  ❌ tests: 3 tests FAILED
  ✅ lint: No lint errors
  ⚠️ code_coverage: Coverage check failed (minimum: 85%)

🔄 [Iterate] 评估下一步...
  🎯 决策: retry
  💬 原因: 失败但可重试
  📝 修复提示: 检查test_user_creation中的断言逻辑...

============================================================
📍 迭代 #3
============================================================

🔍 [Discover] 沿用上一轮任务...

📝 [Plan] 根据失败分析调整策略...

⚡ [Execute] 执行计划...
  ✅ 成功: 3/3

✅ [Verify] 验证结果...
  ✅ tests: All 45 tests passed
  ✅ lint: No lint errors
  ⚠️ code_coverage: Coverage check passed (minimum: 85%)

🔄 [Iterate] 评估下一步...
  🎯 决策: terminate
  💬 原因: 所有检查通过，任务完成

🎉 任务完成！共迭代 1 轮

七、Loop Engineering vs 其他方案

7.1 与传统CI/CD的区别

7.2 与Agent Framework的区别

7.3 与Harness Engineering的关系

Loop Engineering是Harness Engineering的自然演进：

• Harness Engineering关注的是"如何组织AI的能力"——工具注册、权限管理、执行流程
• Loop Engineering关注的是"如何让AI持续产出"——自动发现、自动执行、自动验证

Harness是Loop的基础设施，Loop是Harness的上层应用。

八、生产环境中的Loop Engineering

8.1 监控与可观测性

在生产环境中运行Loop，你需要完善的监控体系：

class LoopMonitor:
    """Loop运行监控"""
    
    def __init__(self):
        self.metrics = {
            'iterations_total': 0,
            'iterations_success': 0,
            'iterations_failure': 0,
            'tasks_completed': 0,
            'avg_iteration_duration': 0,
            'consecutive_failures': 0,
        }
    
    def on_iteration_start(self, iteration: int):
        """迭代开始时的回调"""
        self.metrics['iterations_total'] = iteration
        self._emit_metric('loop.iteration.started', iteration)
    
    def on_iteration_end(self, iteration: int, success: bool, duration: float):
        """迭代结束时的回调"""
        if success:
            self.metrics['iterations_success'] += 1
            self.metrics['consecutive_failures'] = 0
        else:
            self.metrics['iterations_failure'] += 1
            self.metrics['consecutive_failures'] += 1
        
        # 更新平均耗时
        total = self.metrics['iterations_success'] + self.metrics['iterations_failure']
        self.metrics['avg_iteration_duration'] = (
            (self.metrics['avg_iteration_duration'] * (total - 1) + duration) / total
        )
        
        self._emit_metric('loop.iteration.completed', {
            'iteration': iteration,
            'success': success,
            'duration': duration
        })
        
        # 连续失败告警
        if self.metrics['consecutive_failures'] >= 3:
            self._alert('Loop连续失败次数过多', self.metrics)
    
    def _emit_metric(self, name: str, value):
        """发送指标到监控系统"""
        # 集成Prometheus/Grafana/Datadog等
        pass
    
    def _alert(self, message: str, context: dict):
        """发送告警"""
        # 集成PagerDuty/Slack/邮件等告警渠道
        pass

8.2 安全边界

Loop Engineering必须有明确的安全边界：

class LoopSafetyGuard:
    """Loop安全守卫"""
    
    def __init__(self, config: dict):
        self.max_file_changes = config.get('max_file_changes', 50)
        self.max_lines_changed = config.get('max_lines_changed', 1000)
        self.forbidden_paths = config.get('forbidden_paths', [
            '.env', '.git/config', 'secrets/', 'credentials/'
        ])
        self.require_approval_for = config.get('require_approval_for', [
            'database_migration', 'api_contract_change', 'dependency_major_update'
        ])
    
    def check_safety(self, plan: dict) -> tuple[bool, str]:
        """检查计划是否符合安全约束"""
        
        # 检查文件变更数量
        files_changed = set()
        total_lines = 0
        for step in plan.get('steps', []):
            if 'file' in step:
                files_changed.add(step['file'])
                
                # 检查是否触及禁止路径
                for forbidden in self.forbidden_paths:
                    if step['file'].startswith(forbidden):
                        return False, f'禁止修改受保护路径: {forbidden}'
            
            if 'old_text' in step and 'new_text' in step:
                total_lines += abs(
                    len(step['new_text'].split('\n')) - len(step['old_text'].split('\n'))
                )
        
        if len(files_changed) > self.max_file_changes:
            return False, f'变更文件数({len(files_changed)})超过限制({self.max_file_changes})'
        
        if total_lines > self.max_lines_changed:
            return False, f'变更行数({total_lines})超过限制({self.max_lines_changed})'
        
        return True, '安全检查通过'

九、Loop Engineering的未来展望

9.1 从单Loop到多Loop协作

未来的Loop Engineering将不仅仅是单个循环的运行，而是多个Loop的协作：

• 代码质量Loop：持续监控和改进代码质量
• 安全审计Loop：持续扫描和修复安全漏洞
• 性能优化Loop：持续监控和优化系统性能
• 文档更新Loop：持续保持文档与代码的同步

这些Loop可以并行运行，也可以串联协作，形成一个完整的"自治开发系统"。

9.2 Loop之间的知识共享

不同Loop之间可以共享知识和经验：

class LoopKnowledgeBase:
    """Loop知识库：跨Loop的经验共享"""
    
    def __init__(self, storage_path: str):
        self.storage_path = Path(storage_path)
        self.patterns = self._load_patterns()
    
    def record_pattern(self, pattern: dict):
        """记录一个有效的修复模式"""
        self.patterns.append({
            'problem_type': pattern['problem_type'],
            'solution': pattern['solution'],
            'success_rate': pattern.get('success_rate', 1.0),
            'context': pattern.get('context', {}),
            'timestamp': datetime.now().isoformat()
        })
        self._save_patterns()
    
    def find_similar_patterns(self, problem: dict) -> list[dict]:
        """查找类似问题的历史解决方案"""
        # 使用向量相似度搜索
        similar = []
        for p in self.patterns:
            if self._calculate_similarity(problem, p) > 0.8:
                similar.append(p)
        return sorted(similar, key=lambda x: x['success_rate'], reverse=True)

9.3 从编程到其他领域

Loop Engineering的理念正在扩展到编程之外的领域：

• 内容创作Loop：自动发现热点 → 撰写文章 → 人工审核 → 发布
• 数据分析Loop：自动发现异常 → 分析原因 → 生成报告 → 建议行动
• 运维监控Loop：自动发现告警 → 诊断问题 → 执行修复 → 验证恢复

十、总结：设计你自己的第一个Loop

Loop Engineering的核心思想可以浓缩为三句话：

1. 别再手动Prompt了——设计一个循环，让循环去Prompt
2. 用可验证的条件替代人工判断——测试、lint、类型检查就是你的"眼睛"
3. 人站在Loop上面，不是Loop里面——设计规则，而不是执行步骤

如果你现在就想开始实践Loop Engineering，这里是一个最小化的起步方案：

#!/bin/bash
# 最简Loop：一行bash实现Ralph Loop
while :; do
    # 1. 读取任务
    cat TASK.md | claude-code
    
    # 2. 运行验证
    if pytest -x --tb=short && ruff check .; then
        echo "✅ 本轮迭代通过"
        
        # 检查是否还有待处理任务
        if [ ! -s TASK.md ]; then
            echo "🎉 所有任务完成"
            break
        fi
    else
        echo "❌ 验证失败，继续迭代"
    fi
    
    # 3. 防止无限循环
    sleep 5
done

这不是玩具——这就是Loop Engineering最纯粹的形态。当你理解了这个bash脚本的本质，你就理解了整个Loop Engineering的精髓。

2026年，AI工程的关键词不再是"更好的提示词"，而是"更好的循环"。

本文写作过程中参考了Addy Osmani、Boris Cherny、Peter Steinberger等人的公开分享，以及菜鸟教程、SegmentFault等社区的技术分析文章。