""" AutoGen 自愈演示 - 展示测试失败→自动修复→测试通过的完整闭环 用于 Workshop 演示场景 """ import os import sys from pathlib import Path from typing import Dict, Any import time sys.path.insert(0, str(Path(__file__).parent)) from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager from config.llm_config import get_llm_config, DEV_PROMPT, QA_PROMPT class SelfHealingDemo: """自愈功能演示类""" def __init__(self, api_key: str, model: str = "qwen3.5-flash"): """ 初始化演示 Args: api_key: API Key model: 模型名称 """ self.api_key = api_key self.model = model self.llm_config = get_llm_config(model=model, api_key=api_key) self.workspace_dir = Path("workspace") self.workspace_dir.mkdir(exist_ok=True) print("=" * 70) print("🔄 AutoGen 自愈功能演示") print("=" * 70) def create_intentional_bug(self) -> str: """创建一个有 bug 的代码示例""" buggy_code = '''""" 电池健康状态 (SOH) 计算模块 - 包含故意引入的 bug """ from typing import Dict, Any def calculate_soh(voltage: float, current: float, temperature: float) -> Dict[str, Any]: """ 计算电池健康状态 Args: voltage: 电压 (V) current: 电流 (A) temperature: 温度 (°C) Returns: 包含 SOH 百分比的字典 """ # BUG 1: 除零错误 - 没有检查 voltage 是否为 0 nominal_voltage = 3.7 # 标称电压 voltage_ratio = voltage / nominal_voltage # 可能除零 # BUG 2: 温度范围未验证 # 应该检查 temperature 是否在合理范围内 (-20 到 60) temp_factor = 1.0 - (temperature - 25) * 0.005 # BUG 3: 电流方向未考虑（充电/放电） # 充电和放电时的 SOH 计算应该不同 current_load_factor = 1.0 - abs(current) * 0.01 # 计算 SOH soh = voltage_ratio * temp_factor * current_load_factor * 100 # BUG 4: 没有限制 SOH 在 0-100 范围内 return {"soh": soh, "status": "healthy"} def validate_sensor_data(voltage: float, current: float, temperature: float) -> bool: """ 验证传感器数据有效性 BUG 5: 逻辑错误 - 应该返回 False 当数据无效时 """ # 错误的逻辑：数据超出范围时返回 True if voltage < 0 or voltage > 5: return True # 应该是 False if current < -10 or current > 10: return True # 应该是 False if temperature < -20 or temperature > 60: return True # 应该是 False return False # 应该是 True if __name__ == "__main__": # 测试 result = calculate_soh(3.8, 2.0, 30) print(f"SOH: {result['soh']:.2f}%") ''' code_file = self.workspace_dir / "src_battery_health_buggy.py" with open(code_file, 'w', encoding='utf-8') as f: f.write(buggy_code) print(f"\n📝 已创建包含 bug 的代码：{code_file}") return str(code_file) def create_test_cases(self) -> str: """创建测试用例""" test_code = '''""" 电池健康状态测试用例 """ import pytest import sys from pathlib import Path # 导入被测试模块（buggy 版本） sys.path.insert(0, str(Path(__file__).parent)) from src_battery_health_buggy import calculate_soh, validate_sensor_data class TestCalculateSOH: """测试 SOH 计算功能""" def test_normal_operation(self): """测试正常工作情况""" result = calculate_soh(voltage=3.8, current=2.0, temperature=30) assert "soh" in result assert 0 <= result["soh"] <= 100, "SOH 应该在 0-100 范围内" assert result["status"] == "healthy" def test_zero_voltage(self): """测试零电压情况（应该触发除零保护）""" # 这个测试会暴露 BUG 1 result = calculate_soh(voltage=0, current=0, temperature=25) assert result["soh"] == 0, "零电压时 SOH 应为 0" def test_extreme_temperature(self): """测试极端温度""" # 这个测试会暴露 BUG 2 result_high = calculate_soh(voltage=3.8, current=1.0, temperature=70) result_low = calculate_soh(voltage=3.8, current=1.0, temperature=-30) # 极端温度应该被检测并处理 assert result_high["status"] != "healthy", "高温应标记为异常" assert result_low["status"] != "healthy", "低温应标记为异常" def test_high_current(self): """测试大电流情况""" result = calculate_soh(voltage=3.8, current=15.0, temperature=25) # 过大电流应该被检测 assert result["soh"] >= 0, "SOH 不应为负值" assert result["soh"] <= 100, "SOH 不应超过 100%" class TestValidateSensorData: """测试传感器数据验证""" def test_valid_data(self): """测试有效数据""" result = validate_sensor_data(voltage=3.8, current=2.0, temperature=30) assert result is True, "有效数据应返回 True" def test_invalid_voltage(self): """测试无效电压""" result = validate_sensor_data(voltage=6.0, current=2.0, temperature=30) assert result is False, "电压过高应返回 False" def test_invalid_current(self): """测试无效电流""" result = validate_sensor_data(voltage=3.8, current=15.0, temperature=30) assert result is False, "电流过大应返回 False" def test_invalid_temperature(self): """测试无效温度""" result = validate_sensor_data(voltage=3.8, current=2.0, temperature=80) assert result is False, "温度过高应返回 False" if __name__ == "__main__": pytest.main([__file__, "-v"]) ''' test_file = self.workspace_dir / "test_battery_health.py" with open(test_file, 'w', encoding='utf-8') as f: f.write(test_code) print(f"✅ 已创建测试用例：{test_file}") return str(test_file) def run_tests(self) -> Dict[str, Any]: """运行测试并收集错误日志""" import subprocess test_file = self.workspace_dir / "test_battery_health.py" print("\n🧪 正在执行测试...") result = subprocess.run( [sys.executable, "-m", "pytest", str(test_file), "-v", "--tb=short"], capture_output=True, text=True, timeout=60, cwd=str(self.workspace_dir) ) print("\n" + "=" * 70) print("测试结果:") print("=" * 70) print(result.stdout) if result.stderr: print("错误输出:") print(result.stderr) return { "success": result.returncode == 0, "stdout": result.stdout, "stderr": result.stderr, "returncode": result.returncode } def fix_bugs_with_agent(self, error_log: str) -> str: """使用 Agent 自动修复 bug""" print("\n🔧 启动 Dev_Agent 进行自动修复...") dev_agent = AssistantAgent( name="Dev_Fix_Agent", system_message="""你是一名资深软件工程师，擅长调试和修复代码。 你的任务是分析测试失败日志，找出代码中的 bug 并修复。 请： 1. 分析每个测试失败的原因 2. 定位代码中的具体 bug 3. 提供完整的修复后代码 4. 确保所有测试能够通过""", llm_config=self.llm_config, human_input_mode="NEVER" ) # 读取原始代码 buggy_file = self.workspace_dir / "src_battery_health_buggy.py" with open(buggy_file, 'r', encoding='utf-8') as f: buggy_code = f.read() prompt = f""" 请修复以下代码中的 bug： 【原始代码】 {buggy_code} 【测试失败日志】 {error_log[:2000]} # 限制长度 已知 bug 列表： 1. 除零错误 - 没有检查 voltage 是否为 0 2. 温度范围未验证 3. 电流方向未考虑 4. SOH 未限制在 0-100 范围 5. validate_sensor_data 逻辑反转 请输出完整的修复后代码，保存为 src_battery_health_fixed.py """ response = dev_agent.generate_reply( messages=[{"role": "user", "content": prompt}] ) fixed_code = response if isinstance(response, str) else str(response) # 提取代码块 if "```python" in fixed_code: fixed_code = fixed_code.split("```python")[1].split("```")[0] # 保存修复后的代码 fixed_file = self.workspace_dir / "src_battery_health_fixed.py" with open(fixed_file, 'w', encoding='utf-8') as f: f.write(fixed_code) print(f"✅ 已生成修复后的代码：{fixed_file}") return str(fixed_file) def verify_fix(self) -> Dict[str, Any]: """验证修复后的代码""" print("\n✅ 验证修复结果...") # 修改测试文件导入固定的版本 test_file = self.workspace_dir / "test_battery_health_fixed.py" original_test = self.workspace_dir / "test_battery_health.py" with open(original_test, 'r', encoding='utf-8') as f: test_content = f.read() # 替换导入 test_content = test_content.replace( "from src_battery_health_buggy import", "from src_battery_health_fixed import" ) with open(test_file, 'w', encoding='utf-8') as f: f.write(test_content) # 运行测试 import subprocess result = subprocess.run( [sys.executable, "-m", "pytest", str(test_file), "-v", "--tb=short"], capture_output=True, text=True, timeout=60, cwd=str(self.workspace_dir) ) print("\n" + "=" * 70) print("修复后测试结果:") print("=" * 70) print(result.stdout) return { "success": result.returncode == 0, "stdout": result.stdout, "stderr": result.stderr } def run_demo(self): """运行完整演示""" print("\n🎬 开始自愈演示流程...\n") # 步骤 1: 创建有 bug 的代码 print("步骤 1: 创建包含 bug 的代码") self.create_intentional_bug() # 步骤 2: 创建测试用例 print("\n步骤 2: 创建测试用例") self.create_test_cases() # 步骤 3: 运行测试（预期失败） print("\n步骤 3: 运行测试（预期会失败）") test_result = self.run_tests() if not test_result["success"]: print("\n❌ 测试失败（符合预期），开始自动修复...") # 步骤 4: 使用 Agent 修复 bug print("\n步骤 4: AI 自动修复 bug") self.fix_bugs_with_agent(test_result["stdout"] + "\n" + test_result["stderr"]) # 步骤 5: 验证修复 print("\n步骤 5: 验证修复结果") verify_result = self.verify_fix() if verify_result["success"]: print("\n" + "=" * 70) print("✅ 自愈演示成功！") print("=" * 70) print("\n演示亮点：") print("1. ✅ 自动检测测试失败") print("2. ✅ AI 分析错误原因") print("3. ✅ 自动生成修复代码") print("4. ✅ 自动验证修复结果") print("\n这展示了 TDD + AI 的强大能力：测试驱动开发，AI 辅助修复！") else: print("\n⚠️ 修复未完全成功，这是正常现象（复杂 bug 可能需要多轮修复）") else: print("\n⚠️ 测试意外通过，可能是 bug 不够明显") def main(): """主函数""" api_key = os.getenv("DASHSCOPE_API_KEY") if not api_key: print("❌ 错误：请设置 DASHSCOPE_API_KEY 环境变量") return demo = SelfHealingDemo(api_key=api_key) demo.run_demo() if __name__ == "__main__": main()