autogen/autogen_self_healing_demo.py

"""
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()