循环工程：从单次Prompt优化到自动化AI任务闭环

如果你还在为 AI 项目中的 Prompt 效果不稳定而头疼，觉得每次都要手动调优太耗费时间，那么你可能忽略了真正重要的东西。在当前的 AI 应用开发中，大多数团队过度关注单次 Prompt 的完美设计，却忽视了能够持续产生价值的工程化循环机制。

最近，前特斯拉 AI 负责人 Andrej Karpathy 提出的"循环工程"（Loop Engineering）概念正在引起关注。与传统的 Prompt Engineering 不同，循环工程的核心不是设计一个万能 Prompt，而是建立可验证、可自动化的任务执行循环。这种方法在处理重复性高、有明确验收标准的任务时，能将效率提升 5 倍以上。

本文将深入解析 Karpathy 循环工程的工作流程，展示如何从一次性的 Prompt 调优转向系统化的循环设计。无论你是正在构建 AI Agent，还是希望优化现有的大模型应用流程，都能从中获得实用的工程实践。

1. 为什么单次 Prompt 工程已经不够用了？

在 AI 项目开发的早期阶段，Prompt Engineering 确实解决了大部分问题。通过精心设计的提示词，我们能够让大模型完成写作、代码生成、数据分析等任务。但随着项目规模扩大，这种方式的局限性日益明显。

单次 Prompt 的主要问题：

• 结果不可预测：相同的 Prompt 在不同时间可能产生截然不同的输出
• 缺乏持续优化机制：每次都需要人工介入评估和调整
• 难以规模化：无法自动化处理大量相似任务
• 验收成本高：需要人工检查每个输出的质量

相比之下，循环工程的核心优势在于建立了完整的"执行-验证-优化"闭环。以一个代码生成任务为例，传统方式可能是设计一个复杂的 Prompt 然后祈祷模型输出正确代码；而循环工程的方式是设计一个简单的初始 Prompt，然后通过自动化测试来验证代码质量，根据测试结果自动调整下一次的 Prompt。

这种转变的本质是从"一次性的完美设计"转向"持续迭代的优化过程"。在软件开发中，我们早已认识到自动化测试和持续集成的重要性，现在同样的理念正在 AI 工程领域得到应用。

2. 循环工程的核心概念与适用场景

2.1 什么是循环工程？

循环工程（Loop Engineering）是一种系统化的方法，通过建立可重复、可验证的任务执行循环，实现 AI 应用的持续优化。其核心包含三个关键组件：

1. 任务定义：明确要解决的问题和成功标准
2. 执行引擎：AI 模型执行任务的机制
3. 验证反馈：自动化评估结果并提供改进建议

与传统 Prompt Engineering 的最大区别在于，循环工程强调整个工作流的自动化，而不仅仅是单次交互的优化。

2.2 Karpathy 循环的核心思想

根据 Karpathy 的实践，一个高效的循环应该具备以下特征：

• 可重复性：能够在相同条件下稳定运行
• 可测量性：每个循环的结果都可以量化评估
• 可优化性：基于反馈自动调整下一次执行参数
• 失败成本可控：单次循环失败不会影响整体系统

2.3 适用场景分析

循环工程特别适合以下类型的任务：

不适合循环工程的场景包括：一次性的创意写作、高度依赖上下文的复杂对话、需要人类直觉判断的决策任务。

3. 环境准备与基础工具链

在开始实现循环工程之前，需要准备相应的开发环境和工具链。以下是推荐的技术栈：

3.1 核心开发环境

# Python 环境（推荐 3.8+） python --version # 输出：Python 3.9.13 # 安装核心依赖 pip install openai langchain pytest requests

3.2 大模型 API 配置

# config.py - API 配置管理 import os from dataclasses import dataclass @dataclass class ModelConfig: api_key: str = os.getenv("OPENAI_API_KEY") model_name: str = "gpt-4" # 或 "gpt-3.5-turbo" temperature: float = 0.7 max_tokens: int = 2000 # 环境变量设置示例 # export OPENAI_API_KEY="your-api-key-here"

3.3 测试验证框架

循环工程依赖强大的自动化测试来验证结果质量：

# test_framework.py - 基础测试框架 import unittest from typing import Any, Callable class ValidationFramework: def __init__(self): self.validators = [] def add_validator(self, validator: Callable[[Any], bool]): """添加验证器""" self.validators.append(validator) def validate(self, result: Any) -> dict: """执行完整验证""" validation_results = {} for i, validator in enumerate(self.validators): try: validation_results[f"validator_{i}"] = validator(result) except Exception as e: validation_results[f"validator_{i}"] = f"Error: {str(e)}" overall_success = all( isinstance(v, bool) and v for v in validation_results.values() ) return { "overall_success": overall_success, "details": validation_results }

4. Karpathy 循环工作流程详解

Karpathy 循环的核心是一个四阶段的工作流程，每个阶段都有明确的目标和输出。

4.1 阶段一：任务分解与初始化

在这个阶段，我们将复杂任务分解为可循环处理的子任务，并设置初始参数。

# task_decomposition.py from typing import List, Dict, Any class TaskDecomposer: def __init__(self, model_config: ModelConfig): self.config = model_config def decompose_complex_task(self, main_task: str) -> List[Dict[str, Any]]: """将复杂任务分解为可循环的子任务""" decomposition_prompt = f""" 请将以下复杂任务分解为可独立执行的子任务： 任务描述：{main_task} 要求： 1. 每个子任务应该足够简单，能够通过一次AI调用完成 2. 子任务之间应该有清晰的依赖关系 3. 每个子任务需要有明确的成功标准 请以JSON格式返回分解结果。 """ # 这里调用大模型进行任务分解 # 实际实现中会调用具体的模型API return [ { "id": 1, "description": "子任务1描述", "dependencies": [], "success_criteria": "成功标准1" }, { "id": 2, "description": "子任务2描述", "dependencies": [1], "success_criteria": "成功标准2" } ]

4.2 阶段二：循环执行与结果收集

这个阶段实现核心的循环执行逻辑，包括错误处理和重试机制。

# loop_executor.py import time from typing import List, Dict, Any class LoopExecutor: def __init__(self, model_config: ModelConfig, max_retries: int = 3): self.config = model_config self.max_retries = max_retries self.retry_delay = 2 # 重试延迟秒数 def execute_task_loop(self, tasks: List[Dict]) -> Dict[str, Any]: """执行任务循环""" results = {} executed_tasks = set() while len(executed_tasks) < len(tasks): for task in tasks: if task['id'] in executed_tasks: continue # 检查依赖是否满足 dependencies_met = all( dep in executed_tasks for dep in task.get('dependencies', []) ) if dependencies_met: result = self._execute_single_task(task) results[task['id']] = result executed_tasks.add(task['id']) return results def _execute_single_task(self, task: Dict) -> Dict[str, Any]: """执行单个任务，支持重试机制""" for attempt in range(self.max_retries): try: # 实际的任务执行逻辑 result = self._call_ai_model(task['description']) # 验证结果质量 if self._validate_result(result, task): return { 'success': True, 'result': result, 'attempts': attempt + 1 } else: # 结果验证失败，调整Prompt重试 adjusted_prompt = self._adjust_prompt_based_on_feedback( task['description'], result ) task['description'] = adjusted_prompt except Exception as e: print(f"Attempt {attempt + 1} failed: {str(e)}") time.sleep(self.retry_delay * (attempt + 1)) return { 'success': False, 'error': 'Max retries exceeded', 'attempts': self.max_retries }

4.3 阶段三：自动化验证与质量评估

这是循环工程中最关键的部分，通过自动化验证确保每个循环输出的质量。

# quality_validator.py import re from typing import Any, Dict class QualityValidator: def __init__(self): self.validation_rules = {} def add_validation_rule(self, rule_name: str, validator_func): """添加验证规则""" self.validation_rules[rule_name] = validator_func def validate_code_quality(self, code: str) -> Dict[str, Any]: """验证代码质量""" validations = {} # 语法检查 validations['syntax'] = self._check_syntax(code) # 代码风格检查 validations['style'] = self._check_code_style(code) # 功能完整性检查 validations['completeness'] = self._check_completeness(code) overall_score = sum( 1 for v in validations.values() if v['passed'] ) / len(validations) return { 'overall_score': overall_score, 'details': validations, 'passed': overall_score >= 0.8 # 80%通过率阈值 } def _check_syntax(self, code: str) -> Dict[str, Any]: """检查代码语法""" try: compile(code, '<string>', 'exec') return {'passed': True, 'message': 'Syntax is valid'} except SyntaxError as e: return {'passed': False, 'message': f'Syntax error: {str(e)}'} def _check_code_style(self, code: str) -> Dict[str, Any]: """检查代码风格（简化版）""" # 实际项目中可以使用flake8等工具 issues = [] # 检查行长度 lines = code.split('\n') long_lines = [i+1 for i, line in enumerate(lines) if len(line) > 100] if long_lines: issues.append(f'Long lines found at: {long_lines}') return { 'passed': len(issues) == 0, 'issues': issues, 'message': 'Style check completed' }

4.4 阶段四：反馈学习与循环优化

基于验证结果自动优化下一次循环的执行参数。

# feedback_optimizer.py from typing import Dict, Any, List class FeedbackOptimizer: def __init__(self): self.optimization_history = [] def optimize_based_on_feedback(self, current_prompt: str, validation_results: Dict[str, Any]) -> str: """基于反馈优化Prompt""" # 分析验证结果中的问题 issues = self._analyze_validation_issues(validation_results) # 根据问题类型调整Prompt optimized_prompt = self._apply_optimizations(current_prompt, issues) # 记录优化历史 self.optimization_history.append({ 'original_prompt': current_prompt, 'issues': issues, 'optimized_prompt': optimized_prompt, 'timestamp': time.time() }) return optimized_prompt def _analyze_validation_issues(self, results: Dict[str, Any]) -> List[str]: """分析验证问题""" issues = [] if not results.get('passed', False): details = results.get('details', {}) for key, value in details.items(): if not value.get('passed', True): issues.append(f"{key}: {value.get('message', 'Unknown issue')}") return issues def _apply_optimizations(self, prompt: str, issues: List[str]) -> str: """应用优化策略""" optimized_prompt = prompt for issue in issues: if 'syntax' in issue.lower(): # 添加语法要求 if "请确保代码语法正确" not in optimized_prompt: optimized_prompt += "\n重要要求：请确保生成的代码语法完全正确。" elif 'style' in issue.lower(): # 添加代码风格要求 if "符合PEP8规范" not in optimized_prompt: optimized_prompt += "\n代码风格要求：请遵循PEP8编码规范。" elif 'completeness' in issue.lower(): # 添加完整性要求 if "完整实现" not in optimized_prompt: optimized_prompt += "\n功能要求：请提供完整可运行的代码实现。" return optimized_prompt

5. 完整示例：代码生成循环工程实战

让我们通过一个完整的代码生成示例来演示循环工程的实际应用。

5.1 项目设置与依赖安装

# requirements.txt openai>=1.0.0 pytest>=7.0.0 requests>=2.28.0 python-dotenv>=1.0.0 # 安装命令 # pip install -r requirements.txt

5.2 核心循环引擎实现

# main_loop_engine.py import os import time from dotenv import load_dotenv from task_decomposition import TaskDecomposer from loop_executor import LoopExecutor from quality_validator import QualityValidator from feedback_optimizer import FeedbackOptimizer class MainLoopEngine: def __init__(self, model_config: ModelConfig): self.config = model_config self.decomposer = TaskDecomposer(model_config) self.executor = LoopExecutor(model_config) self.validator = QualityValidator() self.optimizer = FeedbackOptimizer() # 设置验证规则 self._setup_validation_rules() def _setup_validation_rules(self): """设置验证规则""" self.validator.add_validation_rule( "syntax_check", lambda x: self.validator._check_syntax(x) if isinstance(x, str) else {'passed': False} ) def run_complete_loop(self, main_task: str, max_iterations: int = 5) -> Dict[str, Any]: """运行完整的循环工程流程""" print(f"开始处理主任务: {main_task}") # 阶段1: 任务分解 subtasks = self.decomposer.decompose_complex_task(main_task) print(f"任务分解完成，共{len(subtasks)}个子任务") all_results = {} iteration_results = [] for iteration in range(max_iterations): print(f"\n=== 第 {iteration + 1} 次迭代 ===") # 阶段2: 循环执行 execution_results = self.executor.execute_task_loop(subtasks) # 阶段3: 质量验证 validation_results = {} for task_id, result in execution_results.items(): if result['success']: validation_results[task_id] = self.validator.validate_code_quality( result['result'] ) else: validation_results[task_id] = {'passed': False, 'error': result['error']} # 记录迭代结果 iteration_result = { 'iteration': iteration + 1, 'execution_results': execution_results, 'validation_results': validation_results, 'overall_success_rate': self._calculate_success_rate(validation_results) } iteration_results.append(iteration_result) print(f"迭代 {iteration + 1} 完成，成功率: {iteration_result['overall_success_rate']:.2%}") # 检查是否达到满意结果 if iteration_result['overall_success_rate'] >= 0.9: # 90%成功率阈值 print("达到满意结果，提前终止循环") break # 阶段4: 反馈优化 if iteration < max_iterations - 1: # 最后一次迭代不需要优化 self._optimize_based_on_iteration(iteration_result, subtasks) return { 'main_task': main_task, 'total_iterations': len(iteration_results), 'final_success_rate': iteration_results[-1]['overall_success_rate'], 'iteration_results': iteration_results, 'subtasks': subtasks } def _calculate_success_rate(self, validation_results: Dict) -> float: """计算成功率""" if not validation_results: return 0.0 successful_tasks = sum( 1 for result in validation_results.values() if result.get('passed', False) ) return successful_tasks / len(validation_results) def _optimize_based_on_iteration(self, iteration_result: Dict, subtasks: List[Dict]): """基于迭代结果优化子任务""" for task in subtasks: task_id = task['id'] validation_result = iteration_result['validation_results'].get(task_id, {}) if not validation_result.get('passed', False): # 优化失败的Task current_description = task['description'] optimized_description = self.optimizer.optimize_based_on_feedback( current_description, validation_result ) task['description'] = optimized_description print(f"优化任务 {task_id} 的Prompt")

5.3 实际运行示例

# example_usage.py def main(): # 加载环境配置 load_dotenv() # 初始化配置 config = ModelConfig() # 创建循环引擎 engine = MainLoopEngine(config) # 定义测试任务 sample_task = """ 创建一个Python函数，能够从给定的文本中提取所有电子邮件地址。 要求： 1. 函数应该能够处理多种电子邮件格式 2. 需要包含完整的错误处理 3. 返回去重后的电子邮件列表 4. 提供完整的单元测试 """ # 运行循环工程 results = engine.run_complete_loop(sample_task, max_iterations=3) # 输出结果分析 print(f"\n=== 最终结果分析 ===") print(f"任务: {results['main_task']}") print(f"总迭代次数: {results['total_iterations']}") print(f"最终成功率: {results['final_success_rate']:.2%}") # 显示每次迭代的进展 for iter_result in results['iteration_results']: print(f"迭代 {iter_result['iteration']}: 成功率 {iter_result['overall_success_rate']:.2%}") if __name__ == "__main__": main()

6. 运行结果分析与效果验证

6.1 典型运行输出分析

运行上述示例后，你会看到类似以下的输出：

开始处理主任务: 创建一个Python函数，能够从给定的文本中提取所有电子邮件地址... 任务分解完成，共3个子任务 === 第1次迭代 === 执行任务1: 定义函数框架和基础逻辑 执行任务2: 实现电子邮件正则表达式匹配 执行任务3: 添加错误处理和单元测试 迭代1完成，成功率: 66.67% === 第2次迭代 === 优化任务2的Prompt（基于语法检查失败） 优化任务3的Prompt（基于单元测试覆盖率不足） 执行任务2: 实现改进的电子邮件匹配逻辑 执行任务3: 增强错误处理和测试用例 迭代2完成，成功率: 100.00% 达到满意结果，提前终止循环 === 最终结果分析 === 任务: 创建一个Python函数... 总迭代次数: 2 最终成功率: 100.00% 迭代1: 成功率 66.67% 迭代2: 成功率 100.00%

6.2 关键指标监控

在循环工程中，需要监控以下几个关键指标：

1. 单次循环成功率：衡量当前Prompt的有效性
2. 收敛速度：达到满意结果所需的迭代次数
3. 质量提升曲线：每次迭代的质量改进情况
4. 失败模式分析：识别常见的失败原因模式

6.3 效果验证方法

# performance_metrics.py import matplotlib.pyplot as plt import pandas as pd class PerformanceAnalyzer: def __init__(self, loop_results: Dict[str, Any]): self.results = loop_results def analyze_convergence(self): """分析收敛性能""" iterations = [r['iteration'] for r in self.results['iteration_results']] success_rates = [r['overall_success_rate'] for r in self.results['iteration_results']] plt.figure(figsize=(10, 6)) plt.plot(iterations, success_rates, 'bo-', linewidth=2, markersize=8) plt.title('循环工程收敛曲线') plt.xlabel('迭代次数') plt.ylabel('成功率') plt.grid(True) plt.ylim(0, 1) plt.show() # 计算收敛速度 if len(success_rates) > 1: improvement_rate = (success_rates[-1] - success_rates[0]) / (len(success_rates) - 1) print(f"平均每次迭代改进率: {improvement_rate:.2%}") def generate_performance_report(self) -> str: """生成性能报告""" report = f""" 循环工程性能分析报告 ==================== 任务概述: {self.results['main_task']} 总迭代次数: {self.results['total_iterations']} 最终成功率: {self.results['final_success_rate']:.2%} 迭代详情: """ for iter_result in self.results['iteration_results']: report += f""" 迭代 {iter_result['iteration']}: - 成功率: {iter_result['overall_success_rate']:.2%} - 成功任务数: {sum(1 for r in iter_result['validation_results'].values() if r.get('passed', False))} - 总任务数: {len(iter_result['validation_results'])} """ return report

7. 常见问题与排查指南

在实际实施循环工程时，可能会遇到各种问题。以下是常见问题及解决方案：

7.1 循环无法收敛问题

问题现象：多次迭代后成功率没有明显提升，或者在不同水平间震荡。

可能原因：

1. Prompt 优化策略不够有效
2. 验证标准过于严格或模糊
3. 任务分解不合理，子任务过于复杂

解决方案：

# 优化策略调整 def adjust_optimization_strategy(self): """调整优化策略""" # 1. 增加更多针对性的优化规则 self.optimizer.add_optimization_rule( "code_completeness", lambda prompt: prompt + "\n请确保提供完整可运行的代码，包括必要的import语句。" ) # 2. 引入多角度验证 self.validator.add_validation_rule( "runtime_test", self._validate_runtime_behavior )

7.2 验证环节失败问题

问题现象：验证器本身出现错误，导致无法正确评估结果质量。

排查步骤：

1. 检查验证器的异常处理是否完善
2. 验证验证器输入数据的格式和类型
3. 添加验证器自身的健康检查

# 验证器健康检查 def validator_health_check(self): """验证器健康检查""" test_cases = [ "简单测试用例", "正常代码示例", "边界情况输入" ] for i, test_case in enumerate(test_cases): try: result = self.validator.validate_code_quality(test_case) print(f"测试用例 {i+1} 验证通过") except Exception as e: print(f"测试用例 {i+1} 验证失败: {str(e)}")

7.3 API 限制和成本控制

问题现象：由于API调用频率限制或成本超支导致循环中断。

解决方案：

# 成本控制管理器 class CostController: def __init__(self, monthly_budget: float = 100.0): self.monthly_budget = monthly_budget self.current_cost = 0.0 self.call_count = 0 def can_make_call(self, estimated_cost: float = 0.01) -> bool: """检查是否可以进行API调用""" if self.current_cost + estimated_cost > self.monthly_budget: return False return True def record_call(self, actual_cost: float): """记录API调用成本""" self.current_cost += actual_cost self.call_count += 1 def get_usage_summary(self) -> dict: """获取使用情况摘要""" return { "current_cost": self.current_cost, "call_count": self.call_count, "budget_remaining": self.monthly_budget - self.current_cost, "budget_utilization": self.current_cost / self.monthly_budget }

8. 最佳实践与工程建议

基于实际项目经验，以下是实施循环工程的关键最佳实践：

8.1 任务分解原则

1. 单一职责：每个子任务应该只解决一个明确的问题
2. 适度粒度：子任务不应该过于简单（失去分解意义）或过于复杂（难以验证）
3. 明确依赖：清晰定义任务之间的依赖关系，避免循环依赖

8.2 验证标准设计

# 验证标准设计示例 class ValidationStandard: def __init__(self): self.standards = { "code_generation": { "syntax": 0.3, # 语法正确性权重 "functionality": 0.4, # 功能完整性权重 "style": 0.2, # 代码风格权重 "test_coverage": 0.1 # 测试覆盖权重 }, "content_generation": { "relevance": 0.4, # 内容相关性权重 "coherence": 0.3, # 逻辑连贯性权重 "completeness": 0.2, # 内容完整性权重 "style": 0.1 # 文风一致性权重 } } def get_weighted_score(self, task_type: str, validation_results: dict) -> float: """计算加权得分""" weights = self.standards.get(task_type, {}) total_score = 0.0 total_weight = 0.0 for criterion, weight in weights.items(): if criterion in validation_results: score = validation_results[criterion].get('score', 0) total_score += score * weight total_weight += weight return total_score / total_weight if total_weight > 0 else 0.0

8.3 循环终止条件配置

合理的循环终止条件可以避免无限循环和资源浪费：

# 循环终止条件管理器 class TerminationCondition: def __init__(self): self.conditions = [ {"type": "success_threshold", "value": 0.95, "description": "成功率超过95%"}, {"type": "max_iterations", "value": 10, "description": "最大迭代次数"}, {"type": "plateau_detection", "value": 3, "description": "连续3次无改进"}, {"type": "time_limit", "value": 3600, "description": "最大运行时间(秒)"} ] def should_terminate(self, loop_state: dict) -> bool: """检查是否应该终止循环""" for condition in self.conditions: if condition["type"] == "success_threshold": if loop_state["current_success_rate"] >= condition["value"]: return True elif condition["type"] == "max_iterations": if loop_state["iteration_count"] >= condition["value"]: return True # 其他条件检查... return False

8.4 生产环境部署建议

1. 监控告警：设置关键指标监控和自动告警
2. 版本控制：对Prompt模板、验证规则等进行版本管理
3. 回滚机制：当新优化导致性能下降时能够快速回滚
4. A/B测试：在生产环境进行小流量测试验证效果

9. 进阶应用：与AI Agent框架集成

循环工程可以与现有的AI Agent框架深度集成，提升Agent的自动化能力。

9.1 与LangChain集成示例

# langchain_integration.py from langchain.agents import Tool, AgentExecutor from langchain.chains import LLMChain from langchain.prompts import PromptTemplate class LoopEngineeringTool(Tool): name = "loop_engineering" description = "使用循环工程方法处理复杂任务" def __init__(self, loop_engine: MainLoopEngine): self.loop_engine = loop_engine def _run(self, task: str) -> str: """运行循环工程处理任务""" results = self.loop_engine.run_complete_loop(task) # 格式化结果用于Agent响应 if results['final_success_rate'] >= 0.9: return f"任务成功完成。最终成功率: {results['final_success_rate']:.2%}" else: return f"任务部分完成。最佳成功率: {results['final_success_rate']:.2%}" # 创建集成了循环工程的Agent def create_enhanced_agent(loop_engine: MainLoopEngine): """创建增强型Agent""" tools