12.2 代码生成 Agent
生活类比：装修队
想象你请了一支装修队来装修新房。装修队不会一上来就砸墙，而是先量房、出图、列清单（规划）；然后按图纸施工（生成）；施工完要验收，水电通不通、墙面平不平（测试）；发现问题就返工修（修复）；最后交付给你入住（完成）。
代码生成 Agent 就是这支"数字装修队"。它的工作流和装修惊人地相似：先理解需求出方案，再写代码，再跑测试，发现问题就修，直到能交付。区别在于，这支队伍是 AI 驱动的，能 7x24 小时干活，但也需要严格的安全管理——毕竟装修队可能误砸承重墙，代码 Agent 也可能执行危险代码。
一、代码生成 Agent 的工作流设计
代码生成 Agent 与普通聊天 Agent 不同，它需要执行代码、验证结果、根据错误进行迭代修复。一个典型的代码生成 Agent 工作流如下：
用户需求 -> 需求分析 -> 代码生成 -> 沙箱执行 -> 结果验证
↑ ↓
└─── 错误修复 ←─── 是否有错误？核心设计原则有四条：
- Plan-then-Execute（先规划再执行）：避免盲目生成。就像装修前要出图纸，代码生成前要先拆解需求。
- Sandbox First（沙箱优先）：所有代码必须在沙箱中运行。就像装修要围挡施工，不能影响邻居。
- Iterative Refinement（迭代改进）：通过错误反馈迭代改进代码。就像验收发现问题就返工。
- Safety Guardrails（安全护栏）：禁止危险操作，限制资源使用。就像装修不准动承重墙。
下面的代码定义了任务状态机和 Agent 主体，每一步都有注释：
python
from dataclasses import dataclass, field # 数据类装饰器
from typing import List, Optional, Dict, Callable # 类型注解
from enum import Enum # 枚举类型
import asyncio # 异步IO
import re # 正则表达式
class CodeLanguage(Enum): # 支持的编程语言
PYTHON = "python" # Python
SQL = "sql" # SQL
SHELL = "bash" # Shell脚本
JAVASCRIPT = "javascript" # JavaScript
class TaskStatus(Enum): # 任务状态枚举
PLANNING = "规划中" # 正在拆解需求
GENERATING = "生成中" # 正在写代码
EXECUTING = "执行中" # 正在跑代码
TESTING = "测试中" # 正在跑测试
FIXING = "修复中" # 正在修bug
COMPLETED = "完成" # 全部通过
FAILED = "失败" # 超过最大重试
@dataclass
class CodeGenerationTask: # 代码生成任务
"""代码生成任务""" # 封装一次完整的生成请求
task_id: str # 任务唯一ID
requirement: str # 用户原始需求
language: CodeLanguage # 目标语言
plan: List[str] = field(default_factory=list) # 规划步骤列表
code: str = "" # 生成的代码
execution_result: Optional[Dict] = None # 执行结果
test_results: List[Dict] = field(default_factory=list) # 测试结果
iterations: int = 0 # 修复迭代次数
status: TaskStatus = TaskStatus.PLANNING # 当前状态
error_history: List[str] = field(default_factory=list) # 错误历史
class CodeGenerationAgent: # 代码生成Agent
"""代码生成Agent""" # 完整的生成-执行-修复流水线
MAX_ITERATIONS = 5 # 最大修复次数，防止死循环
def __init__(self, llm_client, sandbox, code_reviewer, test_runner): # 依赖注入
self.llm = llm_client # LLM客户端
self.sandbox = sandbox # 代码沙箱
self.reviewer = code_reviewer # 代码审查器
self.tester = test_runner # 测试运行器
async def generate_code(self, requirement: str, language: CodeLanguage) -> CodeGenerationTask:
"""完整的代码生成流程""" # 主入口
task = CodeGenerationTask( # 创建任务对象
task_id=f"task_{id(requirement)}", # 用id做简单唯一标识
requirement=requirement, # 存需求
language=language # 存语言
)
# Step 1: 需求分析与规划
task.plan = await self._plan_task(requirement, language) # 让LLM拆解需求
task.status = TaskStatus.GENERATING # 状态切换到生成
# Step 2: 代码生成
task.code = await self._generate_code(requirement, task.plan, language) # 生成代码
# Step 3-5: 执行-测试-修复循环
while task.iterations < self.MAX_ITERATIONS: # 不超过最大迭代
# 执行代码
task.status = TaskStatus.EXECUTING # 切到执行态
exec_result = await self.sandbox.execute(task.code, language) # 沙箱执行
task.execution_result = exec_result # 存执行结果
if exec_result["error"]: # 有错误
task.error_history.append(exec_result["error"]) # 记错误
task.status = TaskStatus.FIXING # 切到修复态
task.code = await self._fix_code(task) # 修复代码
task.iterations += 1 # 迭代数+1
continue # 重新执行
# 测试代码
task.status = TaskStatus.TESTING # 切到测试态
test_results = await self.tester.run_tests(task.code, requirement) # 跑测试
task.test_results = test_results # 存测试结果
if all(t["passed"] for t in test_results): # 全部通过
task.status = TaskStatus.COMPLETED # 标记完成
break # 跳出循环
else: # 有测试失败
task.error_history.extend([ # 记失败信息
t["error"] for t in test_results if not t["passed"]
])
task.status = TaskStatus.FIXING # 切到修复态
task.code = await self._fix_code(task) # 修复
task.iterations += 1 # 迭代数+1
if task.status != TaskStatus.COMPLETED: # 超过最大迭代仍未通过
task.status = TaskStatus.FAILED # 标记失败
return task # 返回任务对象二、需求规划与代码生成
规划阶段让 LLM 先拆解需求，生成一个步骤列表；生成阶段再根据步骤列表写代码。这种"先想清楚再动手"的方式，比直接让 LLM 一次性写完要可靠得多。
python
async def _plan_task(self, requirement: str, language: CodeLanguage) -> List[str]:
"""需求分析与任务规划""" # 让LLM拆解需求
prompt = f"""请分析以下代码需求，并给出实现步骤规划。 # Prompt开头
需求：{requirement} # 用户需求
语言：{language.value} # 目标语言
请以JSON格式返回规划，每个步骤包含描述和预期输出：
```json
{{
"steps": [
{{"step": 1, "description": "步骤描述", "expected_output": "预期输出"}}
]
}}
```"""
response = await self.llm.chat(prompt) # 调LLM
import json # 导入json
plan_data = json.loads(self._extract_json(response)) # 解析JSON
return [step["description"] for step in plan_data["steps"]] # 提取步骤描述
async def _generate_code(self, requirement: str, plan: List[str], language: CodeLanguage) -> str:
"""代码生成""" # 根据需求+规划生成代码
plan_text = "\n".join([f"{i+1}. {step}" for i, step in enumerate(plan)]) # 编号拼接
prompt = f"""请根据以下需求和规划生成{language.value}代码。
## 需求
{requirement}
## 实现规划
{plan_text}
## 要求
1. 代码必须完整可运行 # 不准写半截
2. 包含必要的错误处理 # 要有try-except
3. 添加适当的注释 # 关键逻辑要注释
4. 只输出代码，不要额外解释 # 不要废话
请直接输出代码："""
response = await self.llm.chat(prompt) # 调LLM生成
return self._extract_code(response, language) # 提取代码块
async def _fix_code(self, task: CodeGenerationTask) -> str:
"""根据错误修复代码""" # 把错误信息喂给LLM修复
errors = "\n".join(task.error_history[-3:]) # 只取最近3个错误，避免上下文过长
prompt = f"""以下代码执行时出现错误，请修复。
## 原始需求
{task.requirement}
## 当前代码
```{task.language.value}
{task.code}错误信息
要求
请修复以上错误，输出完整的修复后代码。只输出代码。""" response = await self.llm.chat(prompt) # 调LLM修复 return self._extract_code(response, task.language) # 提取修复后的代码
def _extract_code(self, text: str, language: CodeLanguage) -> str:
"""从LLM回复中提取代码块""" # LLM经常包在```里
pattern = rf'```{language.value}\s*\n(.*?)```' # 正则匹配代码块
match = re.search(pattern, text, re.DOTALL) # 搜索
if match: # 找到了
return match.group(1).strip() # 返回代码内容
return text.strip() # 没找到就返回原文
def _extract_json(self, text: str) -> str:
"""从LLM回复中提取JSON""" # 和上面类似，提取JSON
match = re.search(r'```json\s*\n(.*?)```', text, re.DOTALL) # 先找代码块
if match: # 找到了
return match.group(1).strip() # 返回JSON
match = re.search(r'\{.*\}', text, re.DOTALL) # 再找裸JSON
return match.group(0) if match else "{}" # 没有就返回空对象
#### 三、代码执行沙箱
安全是代码生成 Agent 的生命线。沙箱必须隔离执行环境，防止恶意代码影响宿主系统。就像化学实验在通风橱里做，不管反应多剧烈，都不会伤到实验室外面的人。
沙箱有三道防线：**危险模式拦截**（正则匹配禁止 `os.system`、`subprocess` 等）、**模块白名单**（只允许 `math`、`json` 等安全模块）、**资源限制**（超时、内存、输出大小都设上限）。
```python
import subprocess # 子进程管理
import tempfile # 临时文件
import os # 操作系统接口
import resource # 资源限制（Unix）
import signal # 信号处理
from typing import Dict, Optional # 类型注解
class CodeSandbox: # 代码执行沙箱
"""代码执行沙箱""" # 隔离执行环境
LIMITS = { # 安全限制配置
"timeout_seconds": 30, # 最大执行30秒
"max_memory_mb": 512, # 最大内存512MB
"max_output_bytes": 102400, # 最大输出100KB
"max_file_size_mb": 10, # 最大文件10MB
}
FORBIDDEN_PATTERNS = [ # 禁止的危险操作正则
r'os\.system\s*\(', # 系统命令调用
r'subprocess\.', # 子进程
r'eval\s*\(', # 动态求值
r'exec\s*\(', # 动态执行
r'__import__\s*\(', # 动态导入
r'open\s*\([^)]*[\'"]w', # 写文件
r'shutil\.rmtree', # 删目录树
r'os\.remove\s*\(', # 删文件
r'os\.rmdir\s*\(', # 删目录
r'requests\.', # 网络请求
r'socket\.', # 原始网络
r'import\s+ctypes', # C扩展
r'import\s+multiprocessing', # 多进程
]
ALLOWED_MODULES = { # 允许的安全模块白名单
'math', 'json', 're', 'datetime', 'collections', # 数学、JSON、正则等
'itertools', 'functools', 'typing', 'dataclasses', # 函数式工具
'random', 'statistics', 'decimal', 'fractions', # 随机、统计
'string', 'textwrap', 'unicodedata', 'hashlib', # 字符串处理
'base64', 'binascii', 'csv', 'io', 'pathlib', # IO
'copy', 'enum', 'bisect', 'heapq' # 数据结构
}
def __init__(self, workspace_dir: Optional[str] = None): # 初始化工作目录
self.workspace = workspace_dir or tempfile.mkdtemp(prefix="sandbox_") # 临时目录
os.makedirs(self.workspace, exist_ok=True) # 确保目录存在
def security_check(self, code: str) -> tuple[bool, str]: # 安全检查
"""安全检查""" # 返回(是否安全, 错误信息)
for pattern in self.FORBIDDEN_PATTERNS: # 遍历危险模式
if re.search(pattern, code): # 命中
return False, f"代码包含禁止操作：{pattern}" # 拒绝
import_pattern = r'(?:from\s+(\S+)\s+import|import\s+(\S+))' # 匹配import语句
imports = re.findall(import_pattern, code) # 提取所有导入
for imp in imports: # 遍历
module = imp[0] or imp[1] # 取模块名
base_module = module.split('.')[0] # 取顶层模块
if base_module not in self.ALLOWED_MODULES: # 不在白名单
return False, f"不允许导入模块：{base_module}" # 拒绝
return True, "" # 全部通过
async def execute(self, code: str, language: CodeLanguage) -> Dict: # 执行入口
"""在沙箱中执行代码""" # 先检查再执行
is_safe, error_msg = self.security_check(code) # 安全检查
if not is_safe: # 不安全
return { # 返回错误
"success": False,
"error": f"安全检查失败：{error_msg}",
"stdout": "", "stderr": "", "exit_code": -1
}
if language == CodeLanguage.PYTHON: # Python
return await self._execute_python(code) # 走Python执行
elif language == CodeLanguage.SHELL: # Shell
return await self._execute_shell(code) # 走Shell执行
else: # 不支持的语言
return {"success": False, "error": f"不支持的语言：{language.value}"}
async def _execute_python(self, code: str) -> Dict: # Python执行
"""执行Python代码""" # 写文件→subprocess→收集结果
file_path = os.path.join(self.workspace, "script.py") # 临时脚本路径
with open(file_path, "w") as f: # 写入文件
f.write(code) # 写代码
try: # try-finally确保清理
process = await asyncio.create_subprocess_exec( # 创建子进程
"python3", file_path, # 执行python3 script.py
stdout=subprocess.PIPE, # 捕获标准输出
stderr=subprocess.PIPE, # 捕获标准错误
cwd=self.workspace, # 工作目录设为沙箱
preexec_fn=self._set_limits if os.name != 'nt' else None # Unix设资源限制
)
try: # 带超时的等待
stdout, stderr = await asyncio.wait_for( # 异步等待
process.communicate(), # 读取输出
timeout=self.LIMITS["timeout_seconds"] # 超时30秒
)
except asyncio.TimeoutError: # 超时了
process.kill() # 杀进程
return { # 返回超时错误
"success": False,
"error": f"执行超时（{self.LIMITS['timeout_seconds']}秒）",
"stdout": "", "stderr": "", "exit_code": -1
}
stdout_text = stdout.decode('utf-8', errors='replace')[:self.LIMITS["max_output_bytes"]] # 截断输出
stderr_text = stderr.decode('utf-8', errors='replace')[:self.LIMITS["max_output_bytes"]]
return { # 返回执行结果
"success": process.returncode == 0, # 返回码0=成功
"error": stderr_text if process.returncode != 0 else "",
"stdout": stdout_text,
"stderr": stderr_text,
"exit_code": process.returncode
}
finally: # 无论成败都清理
if os.path.exists(file_path): # 文件还在
os.remove(file_path) # 删掉
@staticmethod
def _set_limits(): # 设置资源限制（仅Unix）
"""设置资源限制""" # 在子进程fork后调用
resource.setrlimit(resource.RLIMIT_AS, (512 * 1024 * 1024, 512 * 1024 * 1024)) # 内存512MB
resource.setrlimit(resource.RLIMIT_CPU, (30, 30)) # CPU时间30秒
def cleanup(self): # 清理沙箱
"""清理沙箱""" # 删除工作目录
import shutil # 导入shutil
if os.path.exists(self.workspace): # 目录还在
shutil.rmtree(self.workspace, ignore_errors=True) # 强制删除四、代码审查与自动化测试
代码审查和自动化测试是保证生成代码质量的关键环节。就像装修完验房，不能只看"墙刷白了没有"，还要检查水电通不通、开关灵不灵、防水漏不漏。
python
class CodeReviewer: # 代码审查器
"""代码审查器""" # 多维度评估代码质量
REVIEW_PROMPT = """请审查以下{language}代码，从以下维度评估： # 审查提示词
1. **正确性**：代码是否满足需求
2. **安全性**：是否存在安全漏洞
3. **性能**：是否有性能问题
4. **可读性**：代码是否清晰易读
5. **最佳实践**：是否符合{language}最佳实践
## 需求
{requirement}
## 代码
```{language}
{code}请以JSON格式返回审查结果：
json
{{
"overall_score": 1-10,
"issues": [
{{"severity": "critical|warning|info", "category": "分类", "description": "描述", "suggestion": "建议"}}
],
"passed": true/false
}}
```"""
async def review(self, code: str, requirement: str, language: CodeLanguage, llm_client) -> Dict:
"""审查代码""" # 调LLM做审查
prompt = self.REVIEW_PROMPT.format( # 填充模板
code=code, requirement=requirement, language=language.value
)
response = await llm_client.chat(prompt) # 调LLM
import json # 导入json
return json.loads(response) # 返回审查结果
class TestRunner: # 自动化测试运行器
"""自动化测试运行器""" # 生成测试→执行→解析
async def run_tests(self, code: str, requirement: str) -> List[Dict]:
"""运行测试用例""" # 生成测试并在沙箱执行
test_code = await self._generate_tests(code, requirement) # 先生成测试代码
full_code = f"{code}\n\n{test_code}" # 拼接源码+测试
result = await self.sandbox.execute(full_code, CodeLanguage.PYTHON) # 沙箱执行
if result["error"]: # 执行报错
return [{"passed": False, "test": "execution", "error": result["error"]}] # 返回失败
return self._parse_test_results(result["stdout"]) # 解析测试输出
async def _generate_tests(self, code: str, requirement: str) -> str:
"""基于代码和需求生成测试用例""" # 让LLM写pytest
prompt = f"""为以下代码生成pytest测试用例。
## 需求
{requirement}
## 代码
```python
{code}请生成完整的测试代码，包含：
- 至少3个测试用例 # 正常用例
- 边界条件测试 # n=0, n=1等
- 异常情况测试 # 非法输入
只输出测试代码：""" response = await self.llm.chat(prompt) # 调LLM return self._extract_code(response, CodeLanguage.PYTHON) # 提取测试代码
#### 五、迭代修复策略
代码生成 Agent 最核心的能力是"自我修复"——从错误中学习并改进代码。就像装修队第一次贴瓷砖贴歪了，第二次知道要先用水平仪。`IterativeFixer` 定义了五种修复策略，针对不同类型的错误用不同的 Prompt：
```python
class IterativeFixer: # 迭代修复器
"""迭代修复器""" # 分策略修复代码
FIX_STRATEGIES = [ # 修复策略列表
"syntax_fix", # 语法错误修复
"logic_fix", # 逻辑错误修复
"edge_case_fix", # 边界条件修复
"performance_fix", # 性能优化
"refactor", # 重构提升可读性
]
async def fix_with_strategy(self, # 用特定策略修复
task: CodeGenerationTask,
strategy: str,
llm_client) -> str:
"""使用特定策略修复代码""" # 不同策略不同Prompt
prompts = { # 策略-Prompt映射
"syntax_fix": f"修复以下代码的语法错误：\n```python\n{task.code}\n```\n错误：{task.error_history[-1]}",
"logic_fix": f"修复以下代码的逻辑错误，确保满足需求：\n需求：{task.requirement}\n```python\n{task.code}\n```\n错误：{task.error_history[-1]}",
"edge_case_fix": f"为以下代码添加边界条件处理：\n```python\n{task.code}\n```\n失败的测试：{task.error_history[-1]}",
"performance_fix": f"优化以下代码的性能，保持功能不变：\n```python\n{task.code}\n```",
"refactor": f"重构以下代码，提高可读性和可维护性：\n```python\n{task.code}\n```"
}
prompt = prompts.get(strategy, prompts["logic_fix"]) # 默认用逻辑修复
response = await llm_client.chat(prompt) # 调LLM
return self._extract_code(response, CodeLanguage.PYTHON) # 提取修复后的代码六、常见误区
误区一：不设最大迭代次数。如果 Agent 生成的代码一直有 bug，没有上限就会无限循环下去，烧光 Token 和算力。
MAX_ITERATIONS = 5是个合理的起点。误区二：沙箱只拦截 Python 关键字。有人只在 Python 层面做正则过滤，但
subprocess可以通过__import__('subprocess')绕过。必须同时做模块白名单和危险模式正则双重检查。误区三：错误信息一股脑全塞给 LLM。修复时把所有历史错误都塞进 Prompt，会导致上下文爆炸。应该只取最近 2-3 个错误，聚焦当前问题。
误区四：测试用例和代码用同一个 LLM 生成。如果生成代码的 LLM 和生成测试的 LLM 是同一个，可能存在相同的认知盲区。建议用不同模型或不同 Prompt 生成测试，形成交叉验证。
七、本节小结
| 要点 | 说明 |
|---|---|
| 工作流设计 | Plan→Generate→Execute→Test→Fix 循环是代码生成 Agent 的核心范式 |
| 沙箱安全 | 白名单过滤、资源限制、危险操作拦截是沙箱的三大安全支柱 |
| 迭代修复 | 通过错误反馈驱动代码改进，设置最大迭代次数防止无限循环 |
| 代码审查 | 多维度评估（正确性、安全性、性能、可读性）确保代码质量 |
| 测试自动化 | 自动生成测试用例并验证，形成从生成到验证的闭环 |
代码生成 Agent 是 Agent 最"硬核"的场景之一——它不仅要生成文本，还要执行代码、验证结果、自我修复。掌握这套架构后，你可以把它迁移到 SQL 生成、脚本自动化、API 测试代码生成等各种"生成-执行-验证"型场景。