29.5 调试与优化模块
第 29 章 · 构建自己的编程 Agent · 第8部分:Claude Code 原理与技术
29.5.1 调试模块概述#
调试模块是编程 Agent 的重要能力,它能够自动识别、定位和修复代码中的错误。调试涉及错误检测、根因分析、修复建议等多个环节。
调试流程#
错误报告 ↓ 错误检测 ↓ 错误定位 ↓ 根因分析 ↓ 修复建议 ↓ 修复验证 ↓ 问题解决
29.5.2 错误检测#
错误检测器#
python复制python classErrorDetector:"""错误检测器""" def__init__(self): self.detectors ={'syntax': SyntaxErrorDetector(),'runtime': RuntimeErrorDetector(),'logic': LogicErrorDetector(),'performance': PerformanceErrorDetector(),'security': SecurityErrorDetector()} asyncdefdetect(self, code:str, execution_result: ExecutionResult =None)-> List[DetectedError]:"""检测错误""" errors =[] # 运行各种检测器for detector_name, detector in self.detectors.items():try: detected =await detector.detect(code, execution_result) errors.extend(detected)except Exception as e: logger.error(f"Error in {detector_name} detector: {e}") # 去重 errors = self._deduplicate_errors(errors) # 排序(按严重程度) errors.sort(key=lambda e: e.severity, reverse=True) return errors def_deduplicate_errors(self, errors: List[DetectedError])-> List[DetectedError]:"""去重错误""" seen =set() unique_errors =[] for error in errors: key =(error.type, error.location, error.message)if key notin seen: seen.add(key) unique_errors.append(error) return unique_errors classSyntaxErrorDetector:"""语法错误检测器""" asyncdefdetect(self, code:str, execution_result: ExecutionResult =None)-> List[DetectedError]:"""检测语法错误""" errors =[] try:# 尝试编译代码compile(code,'','exec')except SyntaxError as e: error = DetectedError(type='syntax', severity='high', location=f"Line {e.lineno}, Column {e.offset}", message=e.msg, code_snippet=self._get_code_snippet(code, e.lineno), suggestion=self._get_syntax_suggestion(e)) errors.append(error) return errors def_get_code_snippet(self, code:str, line_no:int)->str:"""获取代码片段""" lines = code.split('\n')if1 line_no len(lines):return lines[line_no -1]return"" def_get_syntax_suggestion(self, error: SyntaxError)->str:"""获取语法错误建议""" suggestions ={'invalid syntax':"检查语法是否正确,确保括号、引号等匹配",'unexpected EOF':"检查代码是否完整,确保所有括号都已闭合",'unterminated string literal':"检查字符串是否正确闭合",'expected an indented block':"检查缩进是否正确",'unindent does not match any outer indentation level':"检查缩进层级是否一致"} return suggestions.get(error.msg,"请检查语法错误") classRuntimeErrorDetector:"""运行时错误检测器""" asyncdefdetect(self, code:str, execution_result: ExecutionResult =None)-> List[DetectedError]:"""检测运行时错误""" errors =[] ifnot execution_result ornot execution_result.error:return errors # 分析错误信息 error_info = self._parse_error(execution_result.error) error = DetectedError(type='runtime', severity='high', location=error_info.get('location','Unknown'), message=error_info.get('message','Unknown error'), code_snippet=error_info.get('code_snippet',''), traceback=error_info.get('traceback',''), suggestion=self._get_runtime_suggestion(error_info)) errors.append(error) return errors def_parse_error(self, error:str)-> Dict[str,str]:"""解析错误信息""" error_info ={'message': error,'location':'Unknown','traceback':'','code_snippet':''} # 解析 tracebackif'Traceback'in error: lines = error.split('\n')for line in lines:if'File'in line and'line'in line:# 提取文件和行号 parts = line.split(',')iflen(parts)>=2: error_info['location']= parts[1].strip()elif line.strip()andnot line.startswith(' ')and'Error'in line: error_info['message']= line.strip() return error_info def_get_runtime_suggestion(self, error_info: Dict)->str:"""获取运行时错误建议""" message = error_info.get('message','').lower() suggestions ={'nameerror':"检查变量名是否正确,确保在使用前已定义",'typeerror':"检查数据类型是否匹配,确保操作适用于该类型",'valueerror':"检查值是否在有效范围内",'indexerror':"检查索引是否在有效范围内",'keyerror':"检查字典键是否存在",'attributeerror':"检查对象是否具有该属性或方法",'zerodivisionerror':"检查除数是否为零"} for error_type, suggestion in suggestions.items():if error_type in message:return suggestion return"请检查错误信息和代码逻辑" classLogicErrorDetector:"""逻辑错误检测器""" def__init__(self, llm_client: LLMClient): self.llm_client = llm_client asyncdefdetect(self, code:str, execution_result: ExecutionResult =None)-> List[DetectedError]:"""检测逻辑错误""" errors =[] # 使用 LLM 分析代码逻辑 prompt =f""" 分析以下代码中的潜在逻辑错误: {code} 请识别: 1. 无限循环 2. 空指针引用 3. 边界条件错误 4. 逻辑矛盾 5. 死代码 以 JSON 格式返回检测到的错误列表。 """ response =await self.llm_client.complete(prompt) detected_errors = self._parse_logic_errors(response) for error in detected_errors: detected_error = DetectedError(type='logic', severity='medium', location=error.get('location','Unknown'), message=error.get('message','Logic error'), code_snippet=error.get('code_snippet',''), suggestion=error.get('suggestion','Review the logic')) errors.append(detected_error) return errors def_parse_logic_errors(self, response:str)-> List[Dict]:"""解析逻辑错误"""try:return json.loads(response)except json.JSONDecodeError:return[] ```## 29.5.3 根因分析 ### 根因分析器 classRootCauseAnalyzer:"""根因分析器"""def__init__(self, llm_client: LLMClient):self.llm_client = llm_client asyncdefanalyze(self, error: DetectedError,code:str,context: Dict =None)-> RootCauseAnalysis:"""分析根因"""analysis = RootCauseAnalysis(error=error)# 收集相关信息relevant_code = self._collect_relevant_code(error, code)# 分析根本原因analysis.root_cause =await self._identify_root_cause(error,relevant_code,context )# 分析影响范围analysis.impact =await self._analyze_impact(error,relevant_code )# 分析修复难度analysis.difficulty =await self._estimate_difficulty(error,analysis.root_cause )return analysis def_collect_relevant_code(self, error: DetectedError,code:str)->str:"""收集相关代码"""lines = code.split('\n')# 提取错误位置周围的代码if error.location:try:line_no =int(error.location.split(',')[0].split()[-1])start =max(0, line_no -5)end =min(len(lines), line_no +5)return'\n'.join(lines[start:end])except(ValueError, IndexError):passreturn code asyncdef_identify_root_cause(self, error: DetectedError,relevant_code:str,context: Dict)->str:"""识别根本原因"""prompt =f""" 分析错误的根本原因: 错误类型:{error.type}错误信息:{error.message}错误位置:{error.location}相关代码: {relevant_code}上下文:{context}请分析并说明: 1. 错误的根本原因 2. 为什么会发生这个错误 3. 代码中的具体问题 """returnawait self.llm_client.complete(prompt)asyncdef_analyze_impact(self, error: DetectedError,relevant_code:str)-> ImpactAnalysis:"""分析影响范围"""prompt =f""" 分析错误的影响范围: 错误:{error.message}相关代码: {relevant_code}请分析: 1. 错误影响哪些功能 2. 影响的严重程度 3. 是否会影响其他代码 4. 是否会影响数据完整性 """response =await self.llm_client.complete(prompt)return self._parse_impact(response)asyncdef_estimate_difficulty(self, error: DetectedError,root_cause:str)->str:"""估计修复难度"""prompt =f""" 估计修复错误的难度: 错误:{error.message}根本原因:{root_cause}请评估修复难度(简单/中等/困难)并说明理由。 """returnawait self.llm_client.complete(prompt)
29.5.4 修复建议#
修复建议生成器#
python复制```python classFixSuggestionGenerator:"""修复建议生成器""" def__init__(self, llm_client: LLMClient): self.llm_client = llm_client asyncdefgenerate_suggestions(self, analysis: RootCauseAnalysis, code:str)-> List[FixSuggestion]:"""生成修复建议""" suggestions =[] # 生成多个修复方案 prompt =f""" 为以下错误生成修复建议: 错误:{analysis.error.message} 根本原因:{analysis.root_cause} 原始代码: {code} 请生成 2-3 个不同的修复方案,每个方案包括: 1. 修复方法 2. 修改后的代码 3. 优缺点分析 4. 适用场景 以 JSON 格式返回修复方案列表。 """ response =await self.llm_client.complete(prompt) fix_suggestions = self._parse_fix_suggestions(response) for suggestion in fix_suggestions: fix_suggestion = FixSuggestion( method=suggestion.get('method',''), fixed_code=suggestion.get('fixed_code',''), advantages=suggestion.get('advantages',[]), disadvantages=suggestion.get('disadvantages',[]), applicable_scenarios=suggestion.get('applicable_scenarios',''), confidence=suggestion.get('confidence',0.5)) suggestions.append(fix_suggestion) # 按置信度排序 suggestions.sort(key=lambda s: s.confidence, reverse=True) return suggestions ```## 29.5.5 性能优化 ### 性能分析器 classPerformanceAnalyzer:"""性能分析器"""def__init__(self, llm_client: LLMClient):self.llm_client = llm_client asyncdefanalyze(self, code:str,execution_result: ExecutionResult =None)-> PerformanceAnalysis:
"""分析性能""" analysis = PerformanceAnalysis()
分析时间复杂度
analysis.time_complexity = await self._analyze_time_complexity(code)
分析空间复杂度
analysis.space_complexity = await self._analyze_space_complexity(code)
识别性能瓶颈
analysis.bottlenecks = await self._identify_bottlenecks(code)
生成优化建议
analysis.optimization_suggestions = await self._generate_optimization_suggestions( code, analysis ) return analysis async def _analyze_time_complexity(self, code: str) -> str: """分析时间复杂度""" prompt = f""" 分析以下代码的时间复杂度: {code} 请分析并说明:
- 整体时间复杂度
- 关键部分的复杂度
- 影响复杂度的主要因素 """ return await self.llm_client.complete(prompt) async def _analyze_space_complexity(self, code: str) -> str: """分析空间复杂度""" prompt = f""" 分析以下代码的空间复杂度: {code} 请分析并说明:
- 整体空间复杂度
- 内存使用情况
- 可能的内存泄漏 """ return await self.llm_client.complete(prompt) async def _identify_bottlenecks(self, code: str) -> List[PerformanceBottleneck]: """识别性能瓶颈""" prompt = f""" 识别以下代码中的性能瓶颈: {code} 请识别:
- 循环嵌套
- 重复计算
- 不必要的内存分配
- 低效的数据结构使用
- I/O 操作 以 JSON 格式返回瓶颈列表。 """ response = await self.llm_client.complete(prompt) return self._parse_bottlenecks(response) async def _generate_optimization_suggestions(self, code: str, analysis: PerformanceAnalysis) -> List[OptimizationSuggestion]: """生成优化建议""" suggestions = [] for bottleneck in analysis.bottlenecks: prompt = f""" 为以下性能瓶颈生成优化建议: 瓶颈类型:{bottleneck.type} 瓶颈位置:{bottleneck.location} 瓶颈描述:{bottleneck.description} 请生成具体的优化建议,包括:
- 优化方法
- 优化后的代码
- 预期性能提升 """ response = await self.llm_client.complete(prompt) suggestion = OptimizationSuggestion( bottleneck_type=bottleneck.type, method=self._extract_method(response), optimized_code=self._extract_code(response), expected_improvement=self._extract_improvement(response) ) suggestions.append(suggestion) return suggestions
bash复制### 代码优化器 ```python ```python class CodeOptimizer: """代码优化器""" def __init__(self, llm_client: LLMClient): self.llm_client = llm_client async def optimize(self, code: str, suggestions: List[OptimizationSuggestion]) -> OptimizedCode: """优化代码""" optimized_code = code for suggestion in suggestions: # 应用优化 optimized_code = await self._apply_optimization( optimized_code, suggestion ) return OptimizedCode( original_code=code, optimized_code=optimized_code, applied_suggestions=suggestions ) async def _apply_optimization(self, code: str, suggestion: OptimizationSuggestion) -> str: """应用单个优化""" prompt = f""" 应用以下优化到代码: 原始代码: {code} 优化方法:{suggestion.method} 优化后的代码示例: {suggestion.optimized_code} 请返回应用优化后的完整代码。 """ return await self.llm_client.complete(prompt) ```## 29.5.6 调试工作流 ### 调试工作流管理器 class DebuggingWorkflow: """调试工作流""" def __init__(self, llm_client: LLMClient, tool_manager: ToolManager): self.llm_client = llm_client self.tool_manager = tool_manager # 初始化组件 self.error_detector = ErrorDetector() self.root_cause_analyzer = RootCauseAnalyzer(llm_client) self.fix_suggestion_generator = FixSuggestionGenerator(llm_client) self.performance_analyzer = PerformanceAnalyzer(llm_client) self.code_optimizer = CodeOptimizer(llm_client) async def debug(self, code: str, execution_result: ExecutionResult = None) -> DebuggingResult: """执行调试工作流""" result = DebuggingResult(original_code=code) # 1. 检测错误 result.errors = await self.error_detector.detect(code, execution_result) if not result.errors: result.status = "no_errors" return result # 2. 分析每个错误 for error in result.errors: # 根因分析 analysis = await self.root_cause_analyzer.analyze(error, code) result.root_cause_analyses.append(analysis) # 生成修复建议 suggestions = await self.fix_suggestion_generator.generate_suggestions( analysis, code ) result.fix_suggestions.extend(suggestions) # 3. 性能分析 result.performance_analysis = await self.performance_analyzer.analyze(code) # 4. 生成优化建议 if result.performance_analysis.optimization_suggestions: result.optimization_suggestions = result.performance_analysis.optimization_suggestions result.status = "completed" return result async def apply_fix(self, code: str, suggestion: FixSuggestion) -> FixedCode: """应用修复""" fixed_code = FixedCode( original_code=code, fixed_code=suggestion.fixed_code, applied_suggestion=suggestion ) # 验证修复 validation_result = await self._validate_fix(fixed_code) fixed_code.validation_result = validation_result return fixed_code async def _validate_fix(self, fixed_code: FixedCode) -> ValidationResult: """验证修复""" # 检查语法 try: compile(fixed_code.fixed_code, '', 'exec') syntax_valid = True except SyntaxError: syntax_valid = False # 运行测试(如果有) test_result = await self._run_tests(fixed_code.fixed_code) return ValidationResult( syntax_valid=syntax_valid, tests_passed=test_result.passed, test_output=test_result.output ) async def _run_tests(self, code: str) -> TestResult: """运行测试""" # 简化实现:执行代码并检查错误 try: exec(code) return TestResult(passed=True, output="Execution successful") except Exception as e: return TestResult(passed=False, output=str(e))