实战：自动化代码重构 Agent Harness

引言

在现代软件开发的快速迭代中，代码质量和可维护性往往成为技术债务积累的牺牲品。随着时间推移，原本设计精良的代码库可能会变成难以理解和修改的"大泥球"。传统的代码重构方法依赖于开发人员的专业知识和大量时间投入，这在快节奏的开发环境中往往难以实现。

近年来，随着大语言模型（LLMs）和AI Agent技术的快速发展，我们看到了自动化代码重构的新可能性。想象一下，有一个智能Agent能够理解代码库的结构，识别代码异味（code smells），并自动应用经过验证的重构模式，同时确保代码的功能完整性。这就是我们今天要探讨的主题：构建一个自动化代码重构Agent Harness。

在这篇文章中，我们将深入探讨如何设计和实现一个完整的自动化代码重构系统。我们将从理论基础开始，逐步深入到架构设计、核心算法实现，最后通过一个完整的项目实战来展示如何从零开始构建这样一个系统。

无论你是希望改善代码质量的软件工程师，还是对AI应用于软件工程感兴趣的研究者，这篇文章都将为你提供有价值的见解和实践指导。

核心概念

在深入探讨技术细节之前，让我们先明确一些关键概念，这些概念将贯穿我们整个讨论。

代码重构

首先，我们需要理解什么是代码重构。根据Martin Fowler的经典定义：

“重构是一种对软件内部结构的改变，使其在不改变代码外部行为的前提下更易于理解和修改。”

重构不是重写代码，而是在保持功能不变的情况下改进代码的结构和设计。常见的重构操作包括：

• 提取方法（Extract Method）
• 内联方法（Inline Method）
• 重命名变量/方法/类（Rename）
• 提取类（Extract Class）
• 移动方法/字段（Move）
• 替换条件表达式为多态（Replace Conditional with Polymorphism）

AI Agent

AI Agent是一个能够自主感知环境、做出决策并执行行动的计算实体。在我们的上下文中，Agent需要：

1. 感知：理解代码库的结构、依赖关系和语义
2. 推理：识别需要重构的部分，并决定应用哪种重构策略
3. 行动：执行实际的代码修改
4. 验证：确保重构没有改变代码的外部行为

Agent Harness

Agent Harness是一个框架或基础设施，用于：

• 管理Agent的生命周期
• 提供Agent与环境（代码库）交互的接口
• 协调多个Agent之间的协作
• 提供监控、日志和调试功能
• 确保操作的安全性和可回滚性

代码表示

为了让Agent能够理解和操作代码，我们需要适当的代码表示方法：

• 抽象语法树（AST）：代码的结构化表示，保留了语法结构
• 控制流图（CFG）：表示程序执行路径的图结构
• 数据流图（DFG）：表示数据在程序中流动和依赖关系的图
• 代码嵌入（Code Embeddings）：将代码映射到高维向量空间，捕捉语义信息

这些概念构成了我们构建自动化代码重构Agent Harness的基础。接下来，让我们探讨为什么我们需要这样一个系统。

问题背景与挑战

技术债务的累积

在软件开发过程中，技术债务是一个普遍存在的问题。根据一项对500名软件开发人员的调查，超过60%的受访者表示他们的代码库存在中度到重度的技术债务问题。技术债务可能来自多种原因：

• 紧迫的项目期限迫使开发人员走捷径
• 开发人员对代码库不熟悉
• 需求变更导致原有设计不再适用
• 缺乏代码审查和质量保证流程
• 技术演进使得原有实现方式过时

技术债务的累积会导致：

• 开发速度下降
• 缺陷率增加
• 新功能开发困难
• 开发人员满意度降低
• 维护成本上升

传统代码重构的局限性

传统的代码重构方法主要依赖于：

1. 开发人员的专业知识：需要深入理解设计模式、重构原则和代码异味
2. 大量的时间投入：重构往往需要仔细分析、逐步修改和充分测试
3. 手动操作：即使有IDE的辅助，大部分重构工作仍然需要手动完成

这些局限性导致：

• 重构往往被推迟，直到代码质量问题变得严重
• 只有经验丰富的开发人员才能进行有效的重构
• 重构过程中容易引入错误
• 大规模重构风险高，难以在不影响开发进度的情况下进行

现有工具的不足

现有的代码重构工具和IDE插件虽然提供了一些自动化能力，但它们往往：

• 只能处理简单的、结构化的重构操作
• 缺乏对代码语义的深入理解
• 无法主动发现需要重构的代码
• 不能提供上下文相关的重构建议
• 缺乏验证重构正确性的能力

AI驱动的自动化重构的机遇

随着大语言模型（如GPT-4、CodeLlama、StarCoder等）的出现，我们现在有了能够理解和生成代码的强大工具。这些模型在代码理解和生成方面展现出了令人印象深刻的能力，为自动化代码重构提供了新的可能性。

然而，直接使用LLM进行代码重构也面临着一些挑战：

• 上下文窗口限制：大型代码库往往超过模型的上下文窗口
• 缺乏结构化理解：LLM主要通过文本理解代码，缺乏对代码结构的深入理解
• 一致性问题：LLM生成的代码可能与现有代码库的风格和模式不一致
• 验证困难：确保重构后的代码功能正确是一个挑战
• 安全性考虑：自动修改生产代码存在风险

这些挑战正是我们需要构建一个专门的Agent Harness的原因。Harness可以提供必要的基础设施，帮助我们克服这些挑战，实现安全、可靠的自动化代码重构。

技术基础与相关工作

在设计我们的系统之前，让我们先了解一些相关的技术基础和现有研究工作。

代码分析技术

抽象语法树（AST）

抽象语法树是源代码的结构化表示，它将代码组织成树状结构，其中每个节点代表代码中的一个构造（如表达式、语句、声明等）。

# 简单的Python AST示例
import ast

code = """
def calculate_sum(a, b):
    result = a + b
    return result
"""

tree = ast.parse(code)
print(ast.dump(tree, indent=2))

这段代码会输出：

Module(
  body=[
    FunctionDef(
      name='calculate_sum',
      args=arguments(
        posonlyargs=[],
        args=[
          arg(arg='a'),
          arg(arg='b')],
        kwonlyargs=[],
        kw_defaults=[],
        defaults=[]),
      body=[
        Assign(
          targets=[
            Name(id='result', ctx=Store())],
          value=BinOp(
            left=Name(id='a', ctx=Load()),
            op=Add(),
            right=Name(id='b', ctx=Load()))),
        Return(
          value=Name(id='result', ctx=Load()))],
      decorator_list=[])],
  type_ignores=[])

AST是代码分析和转换的基础，几乎所有的静态分析工具和重构工具都建立在AST之上。

控制流图（CFG）

控制流图表示程序执行过程中可能的路径。在CFG中，节点表示基本代码块（没有分支的连续代码序列），边表示控制流的转移。

# 简化的控制流图构建示例
def build_cfg(func_ast):
    # 这里会有更复杂的实现
    # 基本思路是遍历AST，识别基本块和控制流转移
    pass

控制流图对于理解程序的执行流程、发现死代码、进行数据流分析等非常重要。

程序依赖图（PDG）

程序依赖图结合了控制依赖和数据依赖，是更全面的程序表示方法。在PDG中，节点表示程序语句，边表示语句之间的依赖关系。

代码异味检测

代码异味是指代码中可能指示更深层次问题的某些模式。常见的代码异味包括：

• 过长函数（Long Method）：函数太长，难以理解
• 过大类（Large Class）：类承担了太多职责
• 重复代码（Duplicated Code）：相同或相似的代码出现在多个地方
• 过长参数列表（Long Parameter List）：函数有太多参数
• 发散式变化（Divergent Change）：一个类因为不同的原因在不同的方向上变化
• 霰弹式修改（Shotgun Surgery）：一个修改需要同时改变多个不同的类

检测代码异味是自动化重构的第一步。传统的代码异味检测方法主要基于规则和度量，而现代方法则开始使用机器学习技术。

大语言模型在代码理解和生成中的应用

近年来，大语言模型在代码相关任务中展现出了惊人的能力：

• 代码补全：根据上下文预测下一个代码片段
• 代码生成：根据自然语言描述生成代码
• 代码解释：解释代码的功能和工作原理
• 代码翻译：将代码从一种编程语言翻译到另一种
• 代码审查：发现代码中的问题和潜在错误
• 代码重构：改进代码结构而不改变功能

像OpenAI的GPT系列、Google的Codey、Meta的CodeLlama、Hugging Face的StarCoder等模型，都在代码任务上进行了专门的训练和优化。

相关研究工作

在学术界和工业界，已经有一些关于自动化代码重构的研究工作：

1. Getafix：Facebook开发的自动修复工具，使用机器学习学习人类开发人员的修复模式
2. SapFix：另一个Facebook的工具，与Sapienz结合，自动生成和验证修复
3. DeepFix：使用深度学习修复常见的编程错误
4. RLAssist：使用强化学习进行代码重构
5. CodeT5：Salesforce开发的代码理解和生成模型
6. AlphaCode：DeepMind开发的代码生成系统，在编程竞赛中表现出色

虽然这些工作取得了一些进展，但构建一个实用、通用的自动化代码重构系统仍然是一个开放的研究问题。

系统架构设计

现在，让我们设计我们的自动化代码重构Agent Harness的整体架构。一个好的架构设计应该是模块化的、可扩展的，并且能够处理我们前面讨论的各种挑战。

整体架构

我们的系统将采用分层架构，包括以下主要组件：

核心组件详细设计

1. 代码仓库接口 (Code Repository Interface)

这个组件负责与代码仓库进行交互，提供：

• 代码检出和提交功能
• 分支管理
• 差异比较
• 历史记录查询

我们将使用Git作为主要的版本控制系统，但设计时会考虑支持其他系统。

2. AST解析器 (AST Parser)

AST解析器负责将源代码解析为抽象语法树，并提供遍历和操作AST的功能。我们需要支持多种编程语言，但最初可以从Python开始。

# AST解析器的简化接口
class ASTParser:
    def parse(self, code: str, language: str = 'python') -> ast.AST:
        """解析代码为AST"""
        pass
    
    def unparse(self, tree: ast.AST, language: str = 'python') -> str:
        """将AST转换回代码"""
        pass
    
    def traverse(self, tree: ast.AST, visitor: ast.NodeVisitor) -> None:
        """遍历AST并应用访问者"""
        pass
    
    def transform(self, tree: ast.AST, transformer: ast.NodeTransformer) -> ast.AST:
        """转换AST并返回新的AST"""
        pass

3. 代码知识库 (Code Knowledge Base)

代码知识库存储关于代码库的结构化信息，包括：

• 代码实体（函数、类、变量等）的索引
• 依赖关系图
• 调用图
• 代码度量（复杂度、行数等）
• 历史变更记录

# 代码知识库的简化接口
class CodeKnowledgeBase:
    def index_codebase(self, code_dir: str) -> None:
        """索引整个代码库"""
        pass
    
    def get_function_info(self, func_name: str) -> dict:
        """获取函数信息"""
        pass
    
    def get_call_graph(self) -> dict:
        """获取调用图"""
        pass
    
    def find_dependencies(self, entity: str) -> list:
        """查找依赖项"""
        pass
    
    def search_code(self, query: str) -> list:
        """搜索代码"""
        pass

4. 重构模式库 (Refactoring Patterns Library)

重构模式库存储已知的重构模式和应用规则，包括：

• 重构模式的描述和适用条件
• 应用步骤
• 前后代码示例
• 潜在风险和注意事项

# 重构模式库的简化接口
class RefactoringPattern:
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description
        self.preconditions = []
        self.steps = []
        self.examples = []
    
    def check_preconditions(self, code_context: dict) -> bool:
        """检查前置条件"""
        pass
    
    def apply(self, code: str, context: dict) -> str:
        """应用重构"""
        pass

class RefactoringPatternLibrary:
    def __init__(self):
        self.patterns = {}
    
    def register_pattern(self, pattern: RefactoringPattern) -> None:
        """注册重构模式"""
        pass
    
    def get_pattern(self, name: str) -> RefactoringPattern:
        """获取重构模式"""
        pass
    
    def suggest_patterns(self, code_context: dict) -> list:
        """根据上下文建议重构模式"""
        pass

5. LLM接口 (LLM Interface)

LLM接口封装了与大语言模型的交互，提供：

• 提示工程（Prompt Engineering）
• 响应解析和验证
• 错误处理和重试
• 成本和使用量追踪

# LLM接口的简化实现
from abc import ABC, abstractmethod
from typing import List, Dict, Any

class LLMProvider(ABC):
    @abstractmethod
    def generate(self, prompt: str, **kwargs) -> str:
        """生成文本"""
        pass
    
    @abstractmethod
    def chat(self, messages: List[Dict[str, str]], **kwargs) -> str:
        """进行对话"""
        pass

class LLMInterface:
    def __init__(self, provider: LLMProvider):
        self.provider = provider
        self.prompt_templates = {}
    
    def register_prompt_template(self, name: str, template: str) -> None:
        """注册提示模板"""
        self.prompt_templates[name] = template
    
    def generate_with_template(self, template_name: str, **kwargs) -> str:
        """使用模板生成"""
        template = self.prompt_templates[template_name]
        prompt = template.format(**kwargs)
        return self.provider.generate(prompt)
    
    def analyze_code(self, code: str, task: str) -> str:
        """分析代码"""
        # 这里会使用特定的提示模板
        pass
    
    def suggest_refactoring(self, code: str, context: dict) -> str:
        """建议重构"""
        # 这里会使用特定的提示模板
        pass

6. 测试运行器 (Test Runner)

测试运行器负责执行测试并收集结果，用于验证重构是否改变了代码的外部行为：

• 运行单元测试、集成测试等
• 收集代码覆盖率信息
• 检测性能回归
• 生成测试报告

# 测试运行器的简化接口
class TestRunner:
    def __init__(self, test_dir: str):
        self.test_dir = test_dir
    
    def run_all_tests(self) -> dict:
        """运行所有测试"""
        pass
    
    def run_specific_tests(self, test_names: list) -> dict:
        """运行特定测试"""
        pass
    
    def get_coverage(self) -> dict:
        """获取覆盖率信息"""
        pass
    
    def detect_regressions(self, baseline: dict, current: dict) -> list:
        """检测回归"""
        pass

7. Agent协调器 (Agent Orchestrator)

Agent协调器是系统的中央控制器，负责：

• 接收和解析用户请求
• 规划重构任务
• 分配任务给各个Agent
• 监控任务执行
• 处理错误和异常
• 整合结果并提供反馈

# Agent协调器的简化实现
class AgentOrchestrator:
    def __init__(self):
        self.agents = {}
        self.task_queue = []
        self.state_manager = StateManager()
    
    def register_agent(self, agent_name: str, agent: 'BaseAgent') -> None:
        """注册Agent"""
        self.agents[agent_name] = agent
    
    def submit_task(self, task: 'RefactoringTask') -> str:
        """提交任务"""
        task_id = self.state_manager.create_task(task)
        self.task_queue.append(task_id)
        return task_id
    
    def execute_task(self, task_id: str) -> dict:
        """执行任务"""
        task = self.state_manager.get_task(task_id)
        results = {}
        
        # 1. 代码分析阶段
        analysis_result = self.agents['code_analysis'].execute(task)
        results['analysis'] = analysis_result
        
        # 2. 重构规划阶段
        refactoring_plan = self.agents['refactoring'].plan(analysis_result, task)
        results['plan'] = refactoring_plan
        
        # 3. 重构执行阶段
        for step in refactoring_plan['steps']:
            refactoring_result = self.agents['refactoring'].execute_step(step)
            # 验证每个步骤
            verification_result = self.agents['verification'].execute(refactoring_result)
            if not verification_result['success']:
                # 处理失败
                self.state_manager.update_task(task_id, status='failed', error=verification_result['error'])
                return results
            
            results[step['id']] = {
                'refactoring': refactoring_result,
                'verification': verification_result
            }
        
        # 4. 文档更新
        if task.get('update_docs', False):
            docs_result = self.agents['documentation'].execute(task, results)
            results['documentation'] = docs_result
        
        self.state_manager.update_task(task_id, status='completed', results=results)
        return results

8. 专门的Agent

我们的系统将包含多个专门的Agent，每个负责不同的任务：

代码分析Agent

• 分析代码库结构
• 检测代码异味
• 识别重构机会
• 构建代码依赖关系图

重构Agent

• 规划重构步骤
• 应用重构模式
• 生成重构后的代码
• 确保代码风格一致

验证Agent

• 运行测试
• 检查功能正确性
• 验证性能没有退化
• 确保没有引入新的错误

文档更新Agent

• 更新代码注释
• 生成或更新API文档
• 更新架构文档
• 记录重构决策

工作流程

现在，让我们看看一个典型的重构任务是如何在我们的系统中执行的：

这个工作流程确保了重构过程的安全性和可控性，每个步骤都经过验证，并且可以在出现问题时回滚。

核心算法与实现

在这一节中，我们将深入探讨系统中的一些核心算法和实现细节。

代码异味检测算法

代码异味检测是自动化重构的第一步。我们将结合基于规则的方法和机器学习方法来检测各种代码异味。

基于度量的检测

许多代码异味可以通过代码度量来检测：

# 代码度量计算器
class CodeMetricsCalculator:
    def calculate_cyclomatic_complexity(self, func_node: ast.FunctionDef) -> int:
        """计算圈复杂度"""
        complexity = 1  # 基础复杂度
        
        # 遍历函数体，增加复杂度的节点
        for node in ast.walk(func_node):
            if isinstance(node, (ast.If, ast.For, ast.While, ast.And, ast.Or, 
                                ast.ExceptHandler, ast.With, ast.Assert)):
                complexity += 1
            elif isinstance(node, ast.BoolOp):
                # 对于BoolOp，每个操作符增加复杂度
                complexity += len(node.values) - 1
        
        return complexity
    
    def calculate_function_length(self, func_node: ast.FunctionDef) -> int:
        """计算函数长度（行数）"""
        # 这需要行号信息，实际实现会更复杂
        if hasattr(func_node, 'end_lineno') and hasattr(func_node, 'lineno'):
            return func_node.end_lineno - func_node.lineno + 1
        return 0
    
    def calculate_parameter_count(self, func_node: ast.FunctionDef) -> int:
        """计算参数数量"""
        args = func_node.args
        return len(args.args) + len(args.kwonlyargs) + len(args.posonlyargs)
    
    def calculate_class_cohesion(self, class_node: ast.ClassDef) -> float:
        """计算类内聚性（简化版）"""
        methods = [n for n in class_node.body if isinstance(n, ast.FunctionDef)]
        if not methods:
            return 0.0
        
        # 收集每个方法使用的属性
        method_attributes = {}
        for method in methods:
            attrs = set()
            for node in ast.walk(method):
                if isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name) and node.value.id == 'self':
                    attrs.add(node.attr)
            method_attributes[method.name] = attrs
        
        # 计算方法间共享属性的比例
        total_pairs = len(methods) * (len(methods) - 1) / 2
        if total_pairs == 0:
            return 1.0 if len(methods) == 1 else 0.0
        
        cohesive_pairs = 0
        method_names = list(method_attributes.keys())
        
        for i in range(len(method_names)):
            for j in range(i + 1, len(method_names)):
                attrs_i = method_attributes[method_names[i]]
                attrs_j = method_attributes[method_names[j]]
                if attrs_i & attrs_j:  # 如果有共享属性
                    cohesive_pairs += 1
        
        return cohesive_pairs / total_pairs

基于这些度量，我们可以定义规则来检测代码异味：

# 代码异味检测器
class CodeSmellDetector:
    def __init__(self):
        self.metrics_calculator = CodeMetricsCalculator()
        self.smell_rules = {
            'long_method': {
                'check': lambda metrics: metrics['function_length'] > 30,
                'severity': lambda metrics: min(10, (metrics['function_length'] - 30) / 10)
            },
            'complex_method': {
                'check': lambda metrics: metrics['cyclomatic_complexity'] > 10,
                'severity': lambda metrics: min(10, (metrics['cyclomatic_complexity'] - 10) / 5)
            },
            'long_parameter_list': {
                'check': lambda metrics: metrics['parameter_count'] > 4,
                'severity': lambda metrics: min(10, (metrics['parameter_count'] - 4) / 2)
            },
            'low_cohesion': {
                'check': lambda metrics: 'class_cohesion' in metrics and metrics['class_cohesion'] < 0.3,
                'severity': lambda metrics: min(10, (0.3 - metrics['class_cohesion']) * 20)
            }
        }
    
    def detect_function_smells(self, func_node: ast.FunctionDef) -> list:
        """检测函数级别的代码异味"""
        metrics = {
            'function_length': self.metrics_calculator.calculate_function_length(func_node),
            'cyclomatic_complexity': self.metrics_calculator.calculate_cyclomatic_complexity(func_node),
            'parameter_count': self.metrics_calculator.calculate_parameter_count(func_node)
        }
        
        smells = []
        for smell_name, rule in self.smell_rules.items():
            if rule['check'](metrics):
                smells.append({
                    'name': smell_name,
                    'severity': rule['severity'](metrics),
                    'metrics': metrics.copy()
                })
        
        return smells
    
    def detect_class_smells(self, class_node: ast.ClassDef) -> list:
        """检测类级别的代码异味"""
        metrics = {
            'class_cohesion': self.metrics_calculator.calculate_class_cohesion(class_node)
        }
        
        smells = []
        for smell_name, rule in self.smell_rules.items():
            if rule['check'](metrics):
                smells.append({
                    'name': smell_name,
                    'severity': rule['severity'](metrics),
                    'metrics': metrics.copy()
                })
        
        # 还可以检测类级别的其他异味，如过大类等
        
        return smells

基于机器学习的检测

除了基于规则的方法，我们还可以使用机器学习方法来检测更复杂的代码异味。我们可以使用代码嵌入（Code Embeddings）技术将代码转换为向量，然后使用分类器来检测代码异味。

# 基于机器学习的代码异味检测器
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature_extraction.text import TfidfVectorizer
import joblib

class MLCodeSmellDetector:
    def __init__(self, model_path=None):
        self.vectorizer = TfidfVectorizer(max_features=1000, ngram_range=(1, 3))
        self.classifiers = {}
        
        if model_path:
            self.load_models(model_path)
    
    def extract_features(self, code: str) -> list:
        """提取代码特征"""
        # 这里简化处理，实际中会使用更复杂的特征提取
        return self.vectorizer.transform([code])
    
    def train(self, code_samples: list, labels: dict) -> None:
        """训练分类器"""
        # code_samples: 代码字符串列表
        # labels: {smell_name: [0/1, ...]} 字典
        
        # 训练TF-IDF向量化器
        self.vectorizer.fit(code_samples)
        features = self.vectorizer.transform(code_samples)
        
        # 为每种代码异味训练一个分类器
        for smell_name, smell_labels in labels.items():
            clf = RandomForestClassifier(n_estimators=100, random_state=42)
            clf.fit(features, smell_labels)
            self.classifiers[smell_name] = clf
    
    def detect(self, code: str) -> list:
        """检测代码异味"""
        features = self.extract_features(code)
        smells = []
        
        for smell_name, clf in self.classifiers.items():
            prediction = clf.predict(features)[0]
            if prediction == 1:
                # 获取置信度
                proba = clf.predict_proba(features)[0][1]
                smells.append({
                    'name': smell_name,
                    'confidence': proba,
                    'severity': proba * 10  # 简化的严重性计算
                })
        
        return smells
    
    def save_models(self, path: str) -> None:
        """保存模型"""
        joblib.dump({
            'vectorizer': self.vectorizer,
            'classifiers': self.classifiers
        }, path)
    
    def load_models(self, path: str) -> None:
        """加载模型"""
        data = joblib.load(path)
        self.vectorizer = data['vectorizer']
        self.classifiers = data['classifiers']

重构规划与优化

重构规划是确定应该应用哪些重构以及以什么顺序应用它们的过程。这是一个复杂的优化问题，因为：

1. 重构之间可能存在依赖关系
2. 某些重构可能会使其他重构更容易或更难应用
3. 我们需要最大化代码质量改进，同时最小化风险和工作量

我们可以将重构规划形式化为一个优化问题：

$$\text{最大化}Q(R)-C(R)$$

$$\text{约束条件}D(R)\le D_{max},T(R)\le T_{max}$$

其中：

• $R$ 是重构序列
• $Q(R)$ 是应用重构 $R$ 后的代码质量改进
• $C(R)$ 是应用重构 $R$ 的成本（时间、风险等）
• $D(R)$ 是应用重构 $R$ 后的破坏程度（测试失败数等）
• $T(R)$ 是应用重构 $R$ 所需的时间

我们可以使用搜索算法来找到近似最优的重构序列：

# 重构规划器
import random
from typing import List, Dict, Any

class RefactoringOpportunity:
    def __init__(self, location: str, smell: str, applicable_refactorings: List[str]):
        self.location = location
        self.smell = smell
        self.applicable_refactorings = applicable_refactorings

class RefactoringAction:
    def __init__(self, refactoring_type: str, target: str, params: Dict[str, Any]):
        self.refactoring_type = refactoring_type
        self.target = target
        self.params = params
    
    def __str__(self):
        return f"{self.refactoring_type}({self.target})"

class RefactoringPlanner:
    def __init__(self, knowledge_base, pattern_library):
        self.knowledge_base = knowledge_base
        self.pattern_library = pattern_library
    
    def identify_opportunities(self, code_smells: List[Dict]) -> List[RefactoringOpportunity]:
        """识别重构机会"""
        opportunities = []
        
        for smell in code_smells:
            # 根据代码异味类型确定适用的重构
            applicable_refactorings = self._get_applicable_refactorings(smell['name'])
            if applicable_refactorings:
                opportunity = RefactoringOpportunity(
                    location=smell['location'],
                    smell=smell['name'],
                    applicable_refactorings=applicable_refactorings
                )
                opportunities.append(opportunity)
        
        return opportunities
    
    def _get_applicable_refactorings(self, smell_type: str) -> List[str]:
        """获取适用于特定代码异味的重构类型"""
        # 这是一个简化的映射，实际中会更复杂
        smell_to_refactoring = {
            'long_method': ['extract_method', 'replace_method_with_method_object'],
            'complex_method': ['extract_method', 'decompose_conditional', 'replace_conditional_with_polymorphism'],
            'long_parameter_list': ['introduce_parameter_object', 'preserve_whole_object'],
            'duplicated_code': ['extract_method', 'form_template_method', 'substitute_algorithm'],
            'low_cohesion': ['extract_class', 'extract_subclass', 'hide_delegate']
        }
        
        return smell_to_refactoring.get(smell_type, [])
    
    def estimate_quality_improvement(self, action: RefactoringAction) -> float:
        """估计应用重构后的质量改进"""
        # 这是一个简化的评分函数，实际中会更复杂
        base_scores = {
            'extract_method': 5.0,
            'replace_method_with_method_object': 7.0,
            'decompose_conditional': 4.0,
            'replace_conditional_with_polymorphism': 8.0,
            'introduce_parameter_object': 4.0,
            'preserve_whole_object': 3.0,
            'extract_class': 8.0,
            'extract_subclass': 6.0,
            'hide_delegate': 3.0
        }
        
        return base_scores.get(action.refactoring_type, 3.0)
    
    def estimate_cost(self, action: RefactoringAction) -> float:
        """估计应用重构的成本"""
        # 简化的成本估计
        base_costs = {
            'extract_method': 2.0,
            'replace_method_with_method_object': 5.0,
            'decompose_conditional': 2.0,
            'replace_conditional_with_polymorphism': 6.0,
            'introduce_parameter_object': 3.0,
            'preserve_whole_object': 2.0,
            'extract_class': 7.0,
            'extract_subclass': 5.0,
            'hide_delegate': 2.0
        }
        
        return base_costs.get(action.refactoring_type, 3.0)
    
    def estimate_risk(self, action: RefactoringAction, context: Dict) -> float:
        """估计应用重构的风险"""
        # 风险与多个因素相关：重构类型、代码复杂度、测试覆盖率等
        base_risks = {
            'extract_method': 1.0,
            'replace_method_with_method_object': 3.0,
            'decompose_conditional': 2.0,
            'replace_conditional_with_polymorphism': 4.0,
            'introduce_parameter_object': 2.0,
            'preserve_whole_object': 1.0,
            'extract_class': 5.0,
            'extract_subclass': 3.0,
            'hide_delegate': 1.0
        }
        
        base_risk = base_risks.get(action.refactoring_type, 2.0)
        
        # 调整风险：测试覆盖率越低，风险越高
        coverage = context.get('test_coverage', {}).get(action.target, 0.8)
        coverage_factor = 1.0 / max(0.2, coverage)
        
        # 调整风险：代码复杂度越高，风险越高
        complexity = context.get('complexity', {}).get(action.target, 5)
        complexity_factor = 1.0 + (complexity / 10.0)
        
        return base_risk * coverage_factor * complexity_factor
    
    def check_dependencies(self, action: RefactoringAction, applied_actions: List[RefactoringAction]) -> bool:
        """检查重构依赖关系"""
        # 某些重构可能依赖于其他重构先完成
        # 这里简化处理，实际中会有更复杂的依赖规则
        
        # 示例：如果我们已经对同一个目标应用了extract_class，那么某些重构可能不再适用
        for applied in applied_actions:
            if applied.target == action.target and applied.refactoring_type == 'extract_class':
                if action.refactoring_type in ['extract_method', 'decompose_conditional']:
                    # 这些重构可能已经在extract_class中完成了
                    return False
        
        return True
    
    def plan_refactoring(self, opportunities: List[RefactoringOpportunity], 
                        context: Dict, max_iterations: int = 100) -> List[RefactoringAction]:
        """使用随机爬山算法规划重构序列"""
        # 生成初始计划
        current_plan = self._generate_initial_plan(opportunities)
        current_score = self._evaluate_plan(current_plan, context)
        
        best_plan = current_plan.copy()
        best_score = current_score
        
        # 随机爬山搜索
        for _ in range(max_iterations):
            # 生成邻居计划
            neighbor_plan = self._generate_neighbor_plan(current_plan, opportunities)
            
            # 评估邻居计划
            neighbor_score = self._evaluate_plan(neighbor_plan, context)
            
            # 如果邻居更好，移动到邻居
            if neighbor_score > current_score:
                current_plan = neighbor_plan
                current_score = neighbor_score
                
                # 更新最佳计划
                if current_score > best_score:
                    best_plan = current_plan.copy()
                    best_score = current_score
        
        return best_plan
    
    def _generate_initial_plan(self, opportunities: List[RefactoringOpportunity]) -> List[RefactoringAction]:
        """生成初始重构计划"""
        plan = []
        
        for opportunity in opportunities:
            if opportunity.applicable_refactorings:
                # 随机选择一个适用的重构
                refactoring_type = random.choice(opportunity.applicable_refactorings)
                action = RefactoringAction(
                    refactoring_type=refactoring_type,
                    target=opportunity.location,
                    params={}  # 这里可以添加特定重构的参数
                )
                plan.append(action)
        
        # 随机排序
        random.shuffle(plan)
        return plan
    
    def _generate_neighbor_plan(self, plan: List[RefactoringAction], 
                                opportunities: List[RefactoringOpportunity]) -> List[RefactoringAction]:
        """生成邻居计划（对当前计划进行小的修改）"""
        neighbor = plan.copy()
        
        if not neighbor:
            return self._generate_initial_plan(opportunities)
        
        # 随机选择一种修改方式
        modification_type = random.choice(['swap', 'replace', 'add', 'remove'])
        
        if modification_type == 'swap' and len(neighbor) > 1:
            # 交换两个重构的顺序
            i, j = random.sample(range(len(neighbor)), 2)
            neighbor[i], neighbor[j] = neighbor[j], neighbor[i]
        
        elif modification_type == 'replace':
            # 替换一个重构
            i = random.randint(0, len(neighbor) - 1)
            target = neighbor[i].target
            
            # 找到对应目标的重构机会
            opportunity = next((o for o in opportunities if o.location == target), None)
            if opportunity and len(opportunity.applicable_refactorings) > 1:
                # 选择一个不同的重构
                current_type = neighbor[i].refactoring_type
                other_types = [t for t in opportunity.applicable_refactorings if t != current_type]
                if other_types:
                    new_type = random.choice(other_types)
                    neighbor[i] = RefactoringAction(
                        refactoring_type=new_type,
                        target=target,
                        params={}
                    )
        
        elif modification_type == 'add' and len(opportunities) > len(neighbor):
            # 添加一个新的重构
            used_targets = {a.target for a in neighbor}
            unused_opportunities = [o for o in opportunities if o.location not in used_targets]
            
            if unused_opportunities:
                opportunity = random.choice(unused_opportunities)
                refactoring_type = random.choice(opportunity.applicable_refactorings)
                action = RefactoringAction(
                    refactoring_type=refactoring_type,
                    target=opportunity.location,
                    params={}
                )
                neighbor.append(action)
        
        elif modification_type == 'remove' and len(neighbor) > 1:
            # 移除一个重构
            i = random.randint(0, len(neighbor) - 1)
            neighbor.pop(i)
        
        return neighbor
    
    def _evaluate_plan(self, plan: List[RefactoringAction], context: Dict) -> float:
        """评估重构计划的质量"""
        if not plan:
            return 0.0
        
        total_quality = 0.0
        total_cost = 0.0
        total_risk = 0.0
        applied_actions = []
        
        for action in plan:
            # 检查依赖关系
            if not self.check_dependencies(action, applied_actions):
                continue  # 跳过有依赖问题的重构
            
            # 估计质量改进、成本和风险
            quality = self.estimate_quality_improvement(action)
            cost = self.estimate_cost(action)
            risk = self.estimate_risk(action, context)
            
            # 应用折扣因子（后面的重构可能效果降低）
            discount = 0.9 ** len(applied_actions)
            
            total_quality += quality * discount
            total_cost += cost
            total_risk += risk
            
            applied_actions.append(action)
        
        # 计算综合分数
        if total_cost == 0:
            return 0.0
        
        # 我们希望最大化质量，最小化成本和风险
        score = total_quality / (total_cost * (1 + total_risk / 10))
        return score

基于LLM的代码重构实现

现在，让我们看看如何使用大语言模型来实现具体的代码重构操作。我们将使用一个简化的"提取方法"重构作为示例。

# 基于LLM的重构执行器
import re
from typing