OpenAI Codex 深度解析：从代码生成到自主执行的范式跃迁

from agents import Agent, Tool, Runner, guardrail
from agents.tool import function_tool
import subprocess
import ast

# 定义代码审查工具
@function_tool
def run_linter(code: str, language: str = "python") -> str:
    """运行代码检查工具，返回问题列表"""
    if language == "python":
        try:
            # 使用 ast 模块进行静态分析
            tree = ast.parse(code)
            issues = []
            for node in ast.walk(tree):
                # 检查过长的函数
                if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    line_count = node.end_lineno - node.lineno if node.end_lineno else 0
                    if line_count > 50:
                        issues.append(f"⚠️ 函数 '{node.name}' 超过 50 行 ({line_count}行)")
                # 检查空 except 块
                if isinstance(node, ast.ExceptHandler) and len(node.body) == 1:
                    if isinstance(node.body[0], ast.Pass):
                        issues.append(f"⚠️ 第 {node.lineno} 行：空的 except 块")
            return "\n".join(issues) if issues else "✅ 代码检查通过"
        except SyntaxError as e:
            return f"❌ 语法错误：{e.msg} (第 {e.lineno} 行)"
    return "不支持的语言"

@function_tool
def run_tests(test_path: str) -> str:
    """运行测试用例，返回结果"""
    result = subprocess.run(
        ["pytest", test_path, "-v", "--tb=short"],
        capture_output=True, text=True, timeout=60
    )
    return result.stdout if result.returncode == 0 else result.stderr

# 构建代码审查 Agent
code_reviewer = Agent(
    name="CodeReviewer",
    instructions="""你是一个资深代码审查专家。请：
    1. 使用 run_linter 检查代码质量问题
    2. 使用 run_tests 验证测试通过情况
    3. 综合给出审查意见和改进建议
    4. 对严重问题标记为 🔴，建议性问题标记为 🟡""",
    tools=[run_linter, run_tests],
)

# 构建文档生成 Agent
doc_writer = Agent(
    name="DocWriter",
    instructions="""你是一个技术文档专家。根据代码内容生成：
    1. 函数/类的 docstring
    2. README 更新建议
    3. API 使用示例""",
)

# 多 Agent 协作审查流程
async def review_repository(repo_path: str):
    # 步骤1：代码审查
    code_files = get_python_files(repo_path)
    for file_path in code_files:
        code = open(file_path).read()
        review_result = await Runner.run(
            code_reviewer,
            f"审查文件 {file_path}:\n{code}"
        )
        print(f"📋 {file_path}: {review_result.final_output}")
    
    # 步骤2：文档生成
    doc_result = await Runner.run(
        doc_writer,
        f"为仓库 {repo_path} 生成技术文档"
    )
    print(f"📝 文档建议: {doc_result.final_output}")
    
    return {"review": "完成", "docs": "完成"}

# 添加安全护栏：禁止执行危险操作
@guardrail
def safety_guardrail(ctx, output):
    dangerous_patterns = ["rm -rf", "sudo", "DROP TABLE", "chmod 777"]
    for pattern in dangerous_patterns:
        if pattern.lower() in output.lower():
            return "⛔ 检测到危险操作，已阻止"
    return None

# 运行自主审查流程
# asyncio.run(review_repository("./my-project"))

# 用 Agents SDK 构建端到端项目分析 Agent
# 这是 Codex 自主执行的简化版本

from agents import Agent, Runner, trace
import ast
import os
from pathlib import Path
from dataclasses import dataclass, field
from typing import List

@dataclass
class CodeMetrics:
    """代码度量指标"""
    file_path: str
    total_lines: int
    code_lines: int
    blank_lines: int
    comment_lines: int
    num_functions: int
    num_classes: int
    avg_complexity: float = 0.0
    max_complexity: float = 0.0
    issues: List[str] = field(default_factory=list)

    @property
    def refactor_priority(self) -> str:
        """计算重构优先级"""
        if self.max_complexity > 20 or self.total_lines > 1000:
            return "🔴 高"
        elif self.max_complexity > 10 or self.total_lines > 500:
            return "🟡 中"
        return "🟢 低"

def calculate_complexity(node: ast.AST) -> float:
    """计算代码圈的复杂度"""
    complexity = 1
    for child in ast.walk(node):
        if isinstance(child, (
            ast.If, ast.While, ast.For, ast.ExceptHandler,
            ast.With, ast.Assert, ast.BoolOp, ast.comprehension
        )):
            complexity += 1
    return complexity

def analyze_file(file_path: str) -> CodeMetrics:
    """分析单个 Python 文件"""
    with open(file_path, 'r', encoding='utf-8') as f:
        source = f.read()
    
    lines = source.split('\n')
    metrics = CodeMetrics(
        file_path=file_path,
        total_lines=len(lines),
        code_lines=sum(1 for l in lines if l.strip() and not l.strip().startswith('#')),
        blank_lines=sum(1 for l in lines if not l.strip()),
        comment_lines=sum(1 for l in lines if l.strip().startswith('#')),
        num_functions=0,
        num_classes=0,
    )
    
    try:
        tree = ast.parse(source)
        complexities = []
        
        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                metrics.num_functions += 1
                complexities.append(calculate_complexity(node))
                # 检查函数长度
                func_lines = node.end_lineno - node.lineno if node.end_lineno else 0
                if func_lines > 50:
                    metrics.issues.append(f"函数 '{node.name}' 过长 ({func_lines}行)")
            elif isinstance(node, ast.ClassDef):
                metrics.num_classes += 1
        
        if complexities:
            metrics.avg_complexity = sum(complexities) / len(complexities)
            metrics.max_complexity = max(complexities)
            
    except SyntaxError as e:
        metrics.issues.append(f"语法错误: {e}")
    
    return metrics

# 构建项目分析 Agent
project_analyzer = Agent(
    name="ProjectAnalyzer",
    instructions="""你是一个代码质量分析专家。
    1. 扫描项目中所有 Python 文件
    2. 对每个文件计算代码度量指标
    3. 识别需要重构的高优先级模块
    4. 生成详细的分析报告
    5. 给出具体可行的改进建议""",
)

# 构建 Issue 创建 Agent
issue_creator = Agent(
    name="IssueCreator",
    instructions="""你是一个 GitHub Issue 编写专家。
    根据代码分析报告，创建一个结构清晰的 Issue：
    - 包含问题摘要
    - 列出优先级最高的重构项
    - 提供具体的重构建议
    - 使用 Markdown 表格展示指标""",
)

# 构建报告生成 Agent
report_writer = Agent(
    name="ReportWriter",
    instructions="""你是一个技术报告撰写专家。
    将代码分析结果整理为专业的 Markdown 报告：
    - 执行摘要（1 段）
    - 项目概览统计
    - 高优先级重构项（表格）
    - 详细分析（按优先级排序）
    - 改进建议""",
)

async def full_analysis_pipeline(project_path: str):
    """完整的端到端分析流水线"""
    with trace("project-analysis"):
        # 阶段1：代码扫描与分析
        print("🔍 阶段1：扫描项目结构...")
        py_files = list(Path(project_path).rglob("*.py"))
        print(f"   发现 {len(py_files)} 个 Python 文件")
        
        metrics_list = []
        for py_file in py_files:
            metrics = analyze_file(str(py_file))
            metrics_list.append(metrics)
        
        # 按重构优先级排序
        high_priority = [m for m in metrics_list if m.refactor_priority == "🔴 高"]
        medium_priority = [m for m in metrics_list if m.refactor_priority == "🟡 中"]
        
        print(f"   🔴 高优先级: {len(high_priority)} 个文件")
        print(f"   🟡 中优先级: {len(medium_priority)} 个文件")
        
        # 阶段2：生成分析报告
        print("\n📝 阶段2：生成分析报告...")
        analysis_data = {
            "total_files": len(metrics_list),
            "total_lines": sum(m.total_lines for m in metrics_list),
            "high_priority": high_priority,
            "medium_priority": medium_priority,
        }
        
        report = await Runner.run(
            report_writer,
            f"根据以下分析数据生成报告:\n{analysis_data}"
        )
        
        # 阶段3：创建 GitHub Issue
        print("\n🐙 阶段3：创建 GitHub Issue...")
        issue = await Runner.run(
            issue_creator,
            f"基于以下报告创建 Issue:\n{report.final_output}"
        )
        
        return {
            "metrics": metrics_list,
            "report": report.final_output,
            "issue": issue.final_output,
        }

# 执行分析
# result = asyncio.run(full_analysis_pipeline("./my-project"))
# print(result["report"])