Self-Evolving AI Agent（一）：从固定指令到自主进化

"""
GenericAgent 风格的自生长技能树原型
从 3.3K 行种子代码自动发现并学习新技能
"""

from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional, Any
import hashlib


@dataclass
class Skill:
    """Agent 的一个技能"""
    name: str
    description: str
    executor: Callable
    usage_count: int = 0
    success_rate: float = 1.0
    dependencies: List[str] = field(default_factory=list)
    
    def execute(self, **kwargs) -> Any:
        """执行技能"""
        self.usage_count += 1
        try:
            result = self.executor(**kwargs)
            self.success_rate = (self.success_rate * (self.usage_count - 1) + 1) / self.usage_count
            return result
        except Exception as e:
            self.success_rate *= 0.9
            raise e


class SelfGrowingSkillTree:
    """自生长技能树"""
    
    def __init__(self):
        self.skills: Dict[str, Skill] = {}
        self.skill_attempts: Dict[str, int] = {}  # 尝试发现但未成功的技能
        self._register_seed_skills()
    
    def _register_seed_skills(self):
        """注册种子技能（最小初始集合）"""
        seed_skills = [
            Skill(
                name="execute_command",
                description="执行系统命令",
                executor=self._execute_cmd,
            ),
            Skill(
                name="read_file", 
                description="读取文件内容",
                executor=self._read_file,
            ),
            Skill(
                name="write_file",
                description="写入文件内容", 
                executor=self._write_file,
            ),
        ]
        for skill in seed_skills:
            self.register_skill(skill)
    
    def register_skill(self, skill: Skill):
        """注册新技能"""
        self.skills[skill.name] = skill
        print(f"  [+] 新技能: {skill.name} - {skill.description}")
    
    def discover_skill(self, task_description: str) -> Optional[Skill]:
        """
        从任务描述中自动发现需要的技能
        这是 GenericAgent 的核心：当现有技能无法完成任务时，
        Agent 会自动分析需求并生成新技能
        """
        # 模拟 LLM 生成新技能的逻辑
        skill_name = self._generate_skill_name(task_description)
        
        if skill_name in self.skills:
            return self.skills[skill_name]
        
        # 检查是否曾尝试过但失败了
        if skill_name in self.skill_attempts:
            self.skill_attempts[skill_name] += 1
            print(f"  [!] 技能 {skill_name} 已尝试 {self.skill_attempts[skill_name]} 次")
            return None
        
        print(f"  [*] 发现新技能需求: {skill_name}")
        self.skill_attempts[skill_name] = 1
        return None
    
    def _generate_skill_name(self, task: str) -> str:
        """从任务描述生成技能名称（模拟）"""
        # 实际中这里会调用 LLM 来生成技能定义
        keywords = task.lower().split()
        return "_".join(k for k in keywords if len(k) > 2)[:30]
    
    def execute_task(self, task: str, **kwargs) -> Any:
        """执行任务，自动选择或发现所需技能"""
        skill = self.discover_skill(task)
        if skill:
            print(f"  [✓] 使用技能: {skill.name}")
            return skill.execute(**kwargs)
        
        # 技能不存在，需要生长新技能
        print(f"  [?] 技能缺失，需要生长: {task}")
        return None
    
    # --- 种子技能的实现 ---
    
    @staticmethod
    def _execute_cmd(command: str, **kwargs):
        import subprocess
        return subprocess.run(command, shell=True, capture_output=True, text=True)
    
    @staticmethod
    def _read_file(path: str, **kwargs):
        with open(path, 'r') as f:
            return f.read()
    
    @staticmethod
    def _write_file(path: str, content: str, **kwargs):
        with open(path, 'w') as f:
            f.write(content)
        return f"写入成功: {path}"


# --- 使用示例 ---
if __name__ == "__main__":
    agent = SelfGrowingSkillTree()
    print("=== 种子技能 ===")
    print(f"  初始技能数: {len(agent.skills)}")
    print(f"  技能列表: {list(agent.skills.keys())}")
    
    print("
=== 执行任务 ===")
    agent.execute_task("read file config.yaml")
    
    print("
=== 技能树统计 ===")
    total_uses = sum(s.usage_count for s in agent.skills.values())
    print(f"  技能总数: {len(agent.skills)}")
    print(f"  总执行次数: {total_uses}")

"""
GEP (Genome Evolution Protocol) 进化引擎原型
模拟 EvoMap Evolver 的自我进化机制
"""

import random
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Any, Callable
import json


@dataclass
class Gene:
    """Agent 基因组中的一个基因"""
    name: str
    value: float  # 基因值，范围 [0, 1]
    mutation_rate: float = 0.1  # 变异率
    
    def mutate(self) -> 'Gene':
        """基因变异"""
        if random.random() < self.mutation_rate:
            delta = random.gauss(0, 0.15)
            new_value = max(0.0, min(1.0, self.value + delta))
            return Gene(self.name, new_value, self.mutation_rate)
        return self


@dataclass
class AgentGenome:
    """Agent 的完整基因组"""
    genes: Dict[str, Gene] = field(default_factory=dict)
    fitness: float = 0.0
    generation: int = 0
    
    def to_config(self) -> Dict[str, Any]:
        """将基因组解码为 Agent 配置"""
        config = {}
        # 行为基因 → Agent 行为参数
        config['max_retries'] = int(1 + self.genes['max_retries'].value * 9)
        config['temperature'] = 0.1 + self.genes['temperature'].value * 0.9
        config['max_tokens'] = int(500 + self.genes['max_tokens'].value * 3500)
        
        # 知识基因 → RAG 参数
        config['chunk_size'] = int(100 + self.genes['chunk_size'].value * 900)
        config['top_k'] = int(1 + self.genes['top_k'].value * 9)
        
        # 技能基因 → 工具调用策略
        config['tool_first'] = self.genes['tool_first'].value > 0.5
        config['parallel_tools'] = self.genes['parallel_tools'].value > 0.5
        
        return config
    
    @classmethod
    def create_random(cls, generation: int = 0) -> 'AgentGenome':
        """创建随机初始基因组"""
        genes = {
            'max_retries': Gene('max_retries', random.random()),
            'temperature': Gene('temperature', random.random()),
            'max_tokens': Gene('max_tokens', random.random()),
            'chunk_size': Gene('chunk_size', random.random()),
            'top_k': Gene('top_k', random.random()),
            'tool_first': Gene('tool_first', random.random()),
            'parallel_tools': Gene('parallel_tools', random.random()),
        }
        return cls(genes=genes, generation=generation)
    
    def crossover(self, other: 'AgentGenome') -> Tuple['AgentGenome', 'AgentGenome']:
        """基因交叉（两个基因组的重组）"""
        child1_genes = {}
        child2_genes = {}
        
        for name in self.genes:
            if random.random() < 0.5:
                child1_genes[name] = Gene(name, self.genes[name].value)
                child2_genes[name] = Gene(name, other.genes[name].value)
            else:
                child1_genes[name] = Gene(name, other.genes[name].value)
                child2_genes[name] = Gene(name, self.genes[name].value)
        
        next_gen = self.generation + 1
        return (
            AgentGenome(genes=child1_genes, generation=next_gen),
            AgentGenome(genes=child2_genes, generation=next_gen),
        )


class EvolutionEngine:
    """GEP 进化引擎"""
    
    def __init__(self, population_size: int = 20, 
                 mutation_rate: float = 0.1,
                 elite_ratio: float = 0.2):
        self.population_size = population_size
        self.mutation_rate = mutation_rate
        self.elite_count = int(population_size * elite_ratio)
        self.population: List[AgentGenome] = []
        self.best_fitness_history: List[float] = []
    
    def initialize(self):
        """初始化种群"""
        self.population = [
            AgentGenome.create_random() 
            for _ in range(self.population_size)
        ]
    
    def evaluate_fitness(self, genome: AgentGenome, 
                        eval_func: Callable) -> float:
        """评估基因组适应度"""
        config = genome.to_config()
        score = eval_func(config)
        genome.fitness = score
        return score
    
    def evolve(self, eval_func: Callable, generations: int = 50):
        """执行进化循环"""
        for gen in range(generations):
            # 1. 评估所有个体
            for genome in self.population:
                self.evaluate_fitness(genome, eval_func)
            
            # 2. 排序
            self.population.sort(key=lambda g: g.fitness, reverse=True)
            
            # 3. 记录最优
            best = self.population[0]
            self.best_fitness_history.append(best.fitness)
            
            # 4. 精英保留
            elites = self.population[:self.elite_count]
            
            # 5. 选择、交叉、变异生成新种群
            new_population = list(elites)
            
            while len(new_population) < self.population_size:
                # 锦标赛选择
                parent1 = random.choice(self.population[:10])
                parent2 = random.choice(self.population[:10])
                
                # 交叉
                child1, child2 = parent1.crossover(parent2)
                
                # 变异
                for gene in child1.genes.values():
                    gene.mutate()
                for gene in child2.genes.values():
                    gene.mutate()
                
                new_population.extend([child1, child2])
            
            self.population = new_population[:self.population_size]
            
            if gen % 10 == 0:
                print(f"  第 {gen} 代 | 最优适应度: {best.fitness:.4f} | "
                      f"配置: {best.to_config()}")
        
        return self.population[0]


# --- 使用示例 ---
if __name__ == "__main__":
    # 模拟适应度评估函数（实际中基于任务表现计算）
    def mock_eval(config: Dict) -> float:
        """模拟评估：假设最优温度在 0.6-0.8 之间"""
        temp_score = 1.0 - abs(config['temperature'] - 0.7) * 2
        retry_score = min(1.0, config['max_retries'] / 5)
        chunk_score = 1.0 - abs(config['chunk_size'] - 500) / 500
        return (temp_score * 0.4 + retry_score * 0.3 + chunk_score * 0.3)
    
    engine = EvolutionEngine(population_size=30, generations=50)
    engine.initialize()
    
    print("=== GEP 进化引擎开始 ===")
    print(f"  种群大小: {engine.population_size}")
    print(f"  进化代数: 50")
    print()
    
    best = engine.evolve(mock_eval, generations=50)
    
    print(f"
=== 进化结果 ===")
    print(f"  最优适应度: {best.fitness:.4f}")
    print(f"  第 {best.generation} 代")
    print(f"  最优配置: {json.dumps(best.to_config(), indent=2, ensure_ascii=False)}")

"""
Self-Evolving Agent 完整实现
融合 Hermes Agent + GenericAgent + Evolver 三种思路

功能：
1. 经验捕获：记录每次任务执行轨迹
2. 技能生长：自动发现并注册新技能
3. 进化循环：定期评估并优化 Agent 配置
"""

import json
import time
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional, Any, Callable
from datetime import datetime


@dataclass
class Experience:
    """一条经验记录"""
    task: str
    tools_used: List[str]
    success: bool
    duration: float
    error: Optional[str] = None
    key_decisions: List[str] = field(default_factory=list)
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    
    def compress(self) -> Dict:
        """经验压缩：提取关键模式"""
        return {
            "task_pattern": self.task.split()[0] if self.task else "",
            "tools": self.tools_used,
            "success_rate_hint": 1.0 if self.success else 0.0,
            "duration_avg": self.duration,
        }


@dataclass 
class AgentConfig:
    """Agent 配置（进化目标）"""
    max_retries: int = 3
    temperature: float = 0.7
    max_tool_calls: int = 10
    use_reflection: bool = True
    
    def mutate(self) -> 'AgentConfig':
        """配置变异"""
        import random
        new = AgentConfig()
        new.max_retries = max(1, self.max_retries + random.choice([-1, 0, 1]))
        new.temperature = max(0.1, min(1.0, self.temperature + random.gauss(0, 0.1)))
        new.max_tool_calls = max(3, self.max_tool_calls + random.choice([-2, 0, 2]))
        new.use_reflection = self.use_reflection if random.random() > 0.2 else not self.use_reflection
        return new


class SelfEvolvingAgent:
    """自我进化 Agent"""
    
    def __init__(self, name: str = "EvoAgent"):
        self.name = name
        self.config = AgentConfig()
        self.experiences: List[Experience] = []
        self.skills: Dict[str, Callable] = {}
        self.success_history: List[float] = []
        self.generation = 0
        
        # 注册种子技能
        self._seed_skills()
    
    def _seed_skills(self):
        """注册初始种子技能"""
        self.skills = {
            "calculate": self._skill_calculate,
            "echo": self._skill_echo,
            "sleep": self._skill_sleep,
        }
    
    def execute(self, task: str) -> Dict:
        """执行任务（含经验捕获）"""
        start = time.time()
        tools_used = []
        
        try:
            # 1. 从经验库检索相关经验
            relevant = self._retrieve_experience(task)
            
            # 2. 选择并执行技能
            result = self._dispatch_task(task, tools_used)
            
            duration = time.time() - start
            
            # 3. 记录成功经验
            exp = Experience(
                task=task, tools_used=tools_used,
                success=True, duration=duration,
            )
            self.experiences.append(exp)
            self.success_history.append(1.0)
            
            return {"status": "success", "result": result, "duration": duration}
            
        except Exception as e:
            duration = time.time() - start
            
            # 3. 记录失败经验
            exp = Experience(
                task=task, tools_used=tools_used,
                success=False, duration=duration,
                error=str(e),
            )
            self.experiences.append(exp)
            self.success_history.append(0.0)
            
            return {"status": "error", "error": str(e), "duration": duration}
    
    def _retrieve_experience(self, task: str) -> List[Dict]:
        """从经验库检索相关经验"""
        keyword = task.split()[0].lower() if task else ""
        relevant = []
        for exp in self.experiences:
            if keyword in exp.task.lower():
                relevant.append(exp.compress())
        return relevant[:3]  # 最多返回 3 条
    
    def _dispatch_task(self, task: str, tools_used: list) -> Any:
        """分发任务到对应技能"""
        words = task.lower().split()
        
        for skill_name in self.skills:
            if skill_name in words:
                tools_used.append(skill_name)
                return self.skills[skill_name](task)
        
        # 技能不存在 → 生长新技能
        new_skill = self._grow_skill(task)
        if new_skill:
            tools_used.append(new_skill)
            return f"[新技能 {new_skill}] 已创建并执行"
        
        raise ValueError(f"无法处理任务: {task}")
    
    def _grow_skill(self, task: str) -> Optional[str]:
        """技能生长：从任务中发现新技能"""
        skill_name = task.split()[0].lower()
        if len(skill_name) < 2 or skill_name in self.skills:
            return None
        
        # 注册通用技能
        def generic_handler(t):
            return f"执行通用任务: {t}"
        
        self.skills[skill_name] = generic_handler
        print(f"  🌱 技能生长: {skill_name}")
        return skill_name
    
    def evolve(self, eval_window: int = 10):
        """进化循环：评估并优化配置"""
        if len(self.success_history) < eval_window:
            return
        
        recent = self.success_history[-eval_window:]
        current_rate = sum(recent) / len(recent)
        
        self.generation += 1
        old_config = asdict(self.config)
        
        # 如果成功率低于阈值，进行变异
        if current_rate < 0.6:
            self.config = self.config.mutate()
            print(f"  🧬 进化 #{self.generation} | "
                  f"成功率: {current_rate:.1%} → 变异配置")
        elif current_rate > 0.9:
            print(f"  ✅ 进化 #{self.generation} | "
                  f"成功率: {current_rate:.1%} → 配置稳定")
        else:
            print(f"  📊 进化 #{self.generation} | "
                  f"成功率: {current_rate:.1%} → 观察中")
    
    def get_stats(self) -> Dict:
        """获取 Agent 统计信息"""
        total = len(self.experiences)
        success = sum(1 for e in self.experiences if e.success)
        return {
            "name": self.name,
            "generation": self.generation,
            "total_tasks": total,
            "success_rate": success / total if total > 0 else 0,
            "skills_count": len(self.skills),
            "skills": list(self.skills.keys()),
            "config": asdict(self.config),
        }
    
    # --- 种子技能实现 ---
    
    @staticmethod
    def _skill_calculate(task: str) -> float:
        import re
        expr = re.sub(r'[^0-9+-*/(). ]', '', task)
        return eval(expr) if expr else 0
    
    @staticmethod
    def _skill_echo(task: str) -> str:
        return f"Echo: {task}"
    
    @staticmethod
    def _skill_sleep(task: str) -> str:
        import re
        match = re.search(r'(d+)', task)
        secs = int(match.group(1)) if match else 1
        time.sleep(min(secs, 3))
        return f"休眠 {secs} 秒"


# --- 演示 ---
if __name__ == "__main__":
    agent = SelfEvolvingAgent("MyEvoAgent")
    
    print("=" * 60)
    print("Self-Evolving Agent 演示")
    print("=" * 60)
    
    # 第一批任务
    tasks = [
        "calculate 2 + 3 * 4",
        "echo hello world",
        "sleep 1 second",
        "calculate 100 / 7",
        "search quantum computing",  # 新技能 → 自动生长
        "calculate 50 * 3",
        "analyze sentiment",         # 新技能 → 自动生长  
        "translate to English",      # 新技能 → 自动生长
        "calculate 1 + 1",
        "echo test message",
    ]
    
    for i, task in enumerate(tasks, 1):
        print(f"
任务 {i}: {task}")
        result = agent.execute(task)
        status = "✅" if result["status"] == "success" else "❌"
        print(f"  {status} 耗时: {result['duration']:.3f}s")
    
    # 触发进化
    print(f"
{'=' * 60}")
    print("触发进化评估")
    print(f"{'=' * 60}")
    agent.evolve()
    
    # 打印统计
    stats = agent.get_stats()
    print(f"
{'=' * 60}")
    print("Agent 统计")
    print(f"{'=' * 60}")
    print(json.dumps(stats, indent=2, ensure_ascii=False))