Self-Evolving AI Agent 三强争霸：Hermes vs GenericAgent vs Evolver 深度解析

# Hermes-style 经验压缩与检索系统
import json
from typing import List, Dict
from dataclasses import dataclass, asdict
from datetime import datetime

@dataclass
class ExperienceCard:
    """经验卡片：Hermes 的核心知识单元"""
    task_type: str           # 任务类型标签
    strategy: str            # 成功策略摘要
    pitfalls: List[str]      # 常见陷阱
    tools_used: List[str]    # 使用的工具链
    timestamp: str           # 创建时间
    confidence: float        # 置信度（0-1）
    
    def to_dict(self):
        return asdict(self)

class ExperienceCompressor:
    """经验压缩器：将冗长交互历史提炼为经验卡片"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.cards: List[ExperienceCard] = []
    
    def compress_session(self, session_log: List[Dict]) -> ExperienceCard:
        """压缩一次完整的 Agent 会话为经验卡片"""
        # 提取关键信息
        prompt = f"""
        分析以下 Agent 会话日志，提炼为结构化经验卡片：
        
        会话日志（摘要）：
        {json.dumps(session_log[:20], ensure_ascii=False)[:2000]}
        
        请输出 JSON 格式，包含：
        - task_type: 任务类型（一句话）
        - strategy: 成功策略（3-5 句话）
        - pitfalls: 遇到的陷阱和解决方案（列表）
        - tools_used: 使用的工具（列表）
        - confidence: 本次执行的置信度（0-1）
        """
        
        result = self.llm.chat(prompt, response_format="json")
        card_data = json.loads(result)
        
        card = ExperienceCard(
            task_type=card_data["task_type"],
            strategy=card_data["strategy"],
            pitfalls=card_data.get("pitfalls", []),
            tools_used=card_data.get("tools_used", []),
            timestamp=datetime.now().isoformat(),
            confidence=card_data.get("confidence", 0.5)
        )
        
        self.cards.append(card)
        return card
    
    def retrieve_relevant(self, task_description: str, top_k: int = 3) -> List[ExperienceCard]:
        """根据任务描述检索最相关的经验卡片"""
        # 计算语义相似度（简化版：关键词匹配 + 任务类型匹配）
        task_words = set(task_description.lower().split())
        
        scored_cards = []
        for card in self.cards:
            score = 0
            # 任务类型匹配（最高权重）
            if any(w in card.task_type.lower() for w in task_words):
                score += 0.5
            # 策略关键词匹配
            strategy_words = set(card.strategy.lower().split())
            overlap = len(task_words & strategy_words)
            score += 0.3 * (overlap / max(len(task_words), 1))
            # 工具链匹配
            if any(w in " ".join(card.tools_used).lower() for w in task_words):
                score += 0.2
            
            scored_cards.append((score, card))
        
        scored_cards.sort(key=lambda x: x[0], reverse=True)
        return [card for _, card in scored_cards[:top_k]]

# 使用示例
if __name__ == "__main__":
    # 模拟 LLM 客户端
    class MockLLM:
        def chat(self, prompt, response_format=None):
            return json.dumps({
                "task_type": "数据清洗与特征工程",
                "strategy": "先处理缺失值，再编码分类变量，最后标准化数值特征",
                "pitfalls": ["未处理异常值导致标准化失真", "OneHot 编码产生过多稀疏列"],
                "tools_used": ["pandas", "scikit-learn", "feature-engine"],
                "confidence": 0.85
            })
    
    compressor = ExperienceCompressor(MockLLM())
    
    # 模拟一次会话
    session = [
        {"role": "user", "content": "帮我清洗这个数据集"},
        {"role": "assistant", "content": "先检查缺失值..."},
        # ... 更多交互
    ]
    
    card = compressor.compress_session(session)
    print(f"✅ 经验卡片已保存: {card.task_type}")
    print(f"   策略: {card.strategy}")
    print(f"   陷阱: {card.pitfalls}")
    
    # 检索相关经验
    relevant = compressor.retrieve_relevant("处理 CSV 数据中的缺失值和异常值")
    for r in relevant:
        print(f"📋 相关经验: {r.task_type} (置信度: {r.confidence})")

# GenericAgent-style 技能树自主生长系统
import ast
import json
import hashlib
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field

@dataclass
class SkillNode:
    """技能节点：GenericAgent 能力树的基本单元"""
    name: str                        # 技能名称
    description: str                 # 技能描述
    precondition: str               # 前置条件
    parameters: Dict[str, str]      # 参数定义
    implementation: str             # 实现代码（Python）
    success_rate: float = 0.0       # 历史成功率
    avg_tokens: float = 0.0         # 平均 token 消耗
    call_count: int = 0             # 调用次数
    dependencies: List[str] = field(default_factory=list)  # 依赖的其他技能
    
    def compute_hash(self) -> str:
        """计算技能指纹，用于去重"""
        content = f"{self.name}:{self.implementation}"
        return hashlib.sha256(content.encode()).hexdigest()[:12]

class SkillTree:
    """技能树：管理所有已固化的技能"""
    
    def __init__(self):
        self.nodes: Dict[str, SkillNode] = {}
        self.categories: Dict[str, List[str]] = {}  # 按类别组织
    
    def add_skill(self, skill: SkillNode) -> bool:
        """添加新技能（自动去重）"""
        skill_hash = skill.compute_hash()
        if skill_hash in {s.compute_hash() for s in self.nodes.values()}:
            return False  # 重复技能
        
        skill_id = f"{skill.name}_{skill_hash}"
        self.nodes[skill_id] = skill
        
        # 自动分类
        category = self._infer_category(skill)
        self.categories.setdefault(category, []).append(skill_id)
        return True
    
    def _infer_category(self, skill: SkillNode) -> str:
        """根据技能名称和内容推断类别"""
        name_lower = skill.name.lower()
        if any(kw in name_lower for kw in ['file', 'read', 'write', 'io']):
            return "文件操作"
        elif any(kw in name_lower for kw in ['web', 'http', 'request', 'fetch']):
            return "网络操作"
        elif any(kw in name_lower for kw in ['code', 'exec', 'run', 'compile']):
            return "代码执行"
        elif any(kw in name_lower for kw in ['search', 'find', 'query']):
            return "搜索查询"
        else:
            return "其他"
    
    def find_applicable_skills(self, task: str) -> List[SkillNode]:
        """找到可应用于当前任务的技能"""
        applicable = []
        for skill in self.nodes.values():
            # 检查前置条件
            if not self._check_precondition(skill, task):
                continue
            applicable.append(skill)
        return applicable
    
    def _check_precondition(self, skill: SkillNode, task: str) -> bool:
        """检查技能的前置条件是否满足（简化版）"""
        # 实际实现会用 LLM 判断前置条件
        return True
    
    def compose_skill(self, sub_skills: List[str], name: str, description: str) -> SkillNode:
        """将多个子技能组合为高阶技能"""
        impl_parts = []
        params = {}
        for skill_id in sub_skills:
            if skill_id in self.nodes:
                sub = self.nodes[skill_id]
                impl_parts.append(f"# 调用: {sub.name}")
                impl_parts.append(sub.implementation)
                params.update(sub.parameters)
        
        return SkillNode(
            name=name,
            description=description,
            precondition="所有子技能的前置条件均满足",
            parameters=params,
            implementation="
".join(impl_parts),
            dependencies=sub_skills,
        )

class GenericAgent:
    """GenericAgent 核心：自主发现 → 验证 → 固化技能"""
    
    def __init__(self, llm_client):
        self.llm = llm_client
        self.skill_tree = SkillTree()
        self.attempt_log: List[Dict] = []
    
    def attempt_operation(self, operation_desc: str, env_state: Dict) -> Dict:
        """尝试一个操作，返回结果"""
        # 让 LLM 生成执行代码
        code = self.llm.generate_code(f"""
        环境状态: {json.dumps(env_state)[:500]}
        目标操作: {operation_desc}
        
        请生成 Python 代码来执行这个操作。
        """)
        
        try:
            # 安全执行（实际应使用沙箱）
            result = self._safe_exec(code, env_state)
            success = self._evaluate_success(result, operation_desc)
            
            self.attempt_log.append({
                "operation": operation_desc,
                "code": code,
                "success": success,
                "result": str(result)[:200],
            })
            
            return {"success": success, "result": result, "code": code}
        except Exception as e:
            self.attempt_log.append({
                "operation": operation_desc,
                "code": code,
                "success": False,
                "error": str(e),
            })
            return {"success": False, "error": str(e)}
    
    def _safe_exec(self, code: str, env: Dict) -> dict:
        """安全执行代码（简化版，实际应用需沙箱）"""
        local_vars = {"env": env}
        exec(code, {"__builtins__": {}}, local_vars)
        return local_vars.get("result", {})
    
    def _evaluate_success(self, result: dict, goal: str) -> bool:
        """评估操作是否成功（简化版）"""
        # 实际应由 LLM 判断
        return bool(result)
    
    def consolidate_skill(self, successful_attempts: List[Dict]) -> Optional[SkillNode]:
        """将多次成功的尝试固化为技能"""
        if not successful_attempts:
            return None
        
        # 提取共性模式
        pattern = self.llm.extract_pattern(successful_attempts)
        
        skill = SkillNode(
            name=pattern["name"],
            description=pattern["description"],
            precondition=pattern["precondition"],
            parameters=pattern["parameters"],
            implementation=pattern["implementation"],
        )
        
        if self.skill_tree.add_skill(skill):
            print(f"🌱 新技能已固化: {skill.name}")
            return skill
        return None

# 使用示例
if __name__ == "__main__":
    class MockLLM:
        def generate_code(self, prompt):
            return "result = {'status': 'ok', 'data': 'processed'}"
        
        def extract_pattern(self, attempts):
            return {
                "name": "文件批量重命名",
                "description": "批量重命名指定目录下的文件",
                "precondition": "目录存在且有写权限",
                "parameters": {"directory": "str", "pattern": "str", "replacement": "str"},
                "implementation": "import os\nfor f in os.listdir(directory):\n    new_name = f.replace(pattern, replacement)\n    os.rename(os.path.join(directory, f), os.path.join(directory, new_name))",
            }
    
    agent = GenericAgent(MockLLM())
    
    # 模拟多次尝试
    env = {"cwd": "/tmp/test", "files": ["a.txt", "b.txt"]}
    for i in range(3):
        result = agent.attempt_operation(f"重命名文件 (尝试 {i+1})", env)
        print(f"尝试 {i+1}: {'✅' if result['success'] else '❌'}")
    
    # 固化技能
    successful = [a for a in agent.attempt_log if a.get("success")]
    skill = agent.consolidate_skill(successful)
    if skill:
        print(f"技能树现在有 {len(agent.skill_tree.nodes)} 个技能")
        print(f"类别: {list(agent.skill_tree.categories.keys())}")

# Evolver-style GEP 基因组进化协议
import random
import numpy as np
from typing import List, Tuple, Dict
from dataclasses import dataclass

@dataclass
class AgentGenome:
    """Agent 基因组：将 Agent 能力参数化为向量"""
    # 决策策略参数
    exploration_rate: float      # 探索率 (0-1)
    tool_selection_temp: float   # 工具选择温度 (0.1-5.0)
    max_iterations: int          # 最大迭代次数 (1-100)
    
    # 提示词策略参数
    prompt_style: int           # 提示词风格 (0=简洁, 1=详细, 2=结构化)
    reasoning_depth: int        # 推理深度 (1-5)
    
    # 记忆策略参数
    memory_window: int          # 记忆窗口大小 (3-20)
    compression_threshold: float # 压缩阈值 (0.1-0.9)
    
    # 适应度
    fitness: float = 0.0
    
    def to_vector(self) -> np.ndarray:
        """将基因组转换为向量"""
        return np.array([
            self.exploration_rate,
            self.tool_selection_temp,
            self.max_iterations / 100.0,
            self.prompt_style / 2.0,
            self.reasoning_depth / 5.0,
            self.memory_window / 20.0,
            self.compression_threshold,
        ])
    
    @staticmethod
    def from_vector(v: np.ndarray) -> 'AgentGenome':
        """从向量重建基因组"""
        return AgentGenome(
            exploration_rate=np.clip(v[0], 0, 1),
            tool_selection_temp=np.clip(v[1], 0.1, 5.0),
            max_iterations=int(np.clip(v[2] * 100, 1, 100)),
            prompt_style=int(np.clip(round(v[3] * 2), 0, 2)),
            reasoning_depth=int(np.clip(round(v[4] * 5), 1, 5)),
            memory_window=int(np.clip(round(v[5] * 20), 3, 20)),
            compression_threshold=np.clip(v[6], 0.1, 0.9),
        )
    
    def random_genome() -> 'AgentGenome':
        """生成随机初始基因组"""
        return AgentGenome(
            exploration_rate=random.uniform(0, 1),
            tool_selection_temp=random.uniform(0.1, 5.0),
            max_iterations=random.randint(1, 100),
            prompt_style=random.randint(0, 2),
            reasoning_depth=random.randint(1, 5),
            memory_window=random.randint(3, 20),
            compression_threshold=random.uniform(0.1, 0.9),
        )

class GEPEvolver:
    """GEP 进化引擎"""
    
    def __init__(self, population_size: int = 20, elite_count: int = 3):
        self.pop_size = population_size
        self.elite_count = elite_count
        self.generation = 0
        self.history: List[Dict] = []
        
        # 初始化种群
        self.population = [AgentGenome.random_genome() for _ in range(population_size)]
    
    def evaluate_fitness(self, genome: AgentGenome, tasks: List[Dict]) -> float:
        """评估基因组的适应度"""
        total_score = 0
        total_tokens = 0
        
        for task in tasks:
            # 模拟 Agent 执行（实际应运行真实 Agent）
            success, tokens = self._simulate_agent(genome, task)
            if success:
                total_score += 1
            total_tokens += tokens
        
        # 适应度 = 成功率 * (1 / 归一化token消耗)
        success_rate = total_score / len(tasks)
        token_efficiency = 1.0 / (1.0 + total_tokens / (len(tasks) * 100))
        
        return 0.7 * success_rate + 0.3 * token_efficiency
    
    def _simulate_agent(self, genome: AgentGenome, task: Dict) -> Tuple[bool, int]:
        """模拟 Agent 执行（简化版）"""
        # 实际应实例化真实 Agent 并运行
        base_success = 0.3
        bonus = (
            0.2 * genome.exploration_rate +
            0.1 * (genome.reasoning_depth / 5.0) +
            0.1 * (genome.memory_window / 20.0)
        )
        success = random.random() < min(base_success + bonus, 0.95)
        tokens = int(100 / max(genome.tool_selection_temp, 0.1))
        return success, tokens
    
    def select_elites(self) -> List[AgentGenome]:
        """选择精英个体"""
        sorted_pop = sorted(self.population, key=lambda g: g.fitness, reverse=True)
        return sorted_pop[:self.elite_count]
    
    def crossover(self, parent1: AgentGenome, parent2: AgentGenome) -> AgentGenome:
        """基因交叉：两点交叉"""
        v1, v2 = parent1.to_vector(), parent2.to_vector()
        n_genes = len(v1)
        
        # 随机选择两个交叉点
        points = sorted(random.sample(range(1, n_genes), 2))
        
        child_v = np.zeros(n_genes)
        child_v[:points[0]] = v1[:points[0]]
        child_v[points[0]:points[1]] = v2[points[0]:points[1]]
        child_v[points[1]:] = v1[points[1]:]
        
        return AgentGenome.from_vector(child_v)
    
    def mutate(self, genome: AgentGenome, mutation_rate: float = 0.1) -> AgentGenome:
        """基因变异"""
        v = genome.to_vector()
        for i in range(len(v)):
            if random.random() < mutation_rate:
                v[i] += random.gauss(0, 0.1)
                v[i] = np.clip(v[i], 0, 1)
        return AgentGenome.from_vector(v)
    
    def evolve_one_generation(self, tasks: List[Dict], mutation_rate: float = 0.1) -> Dict:
        """进化一代"""
        # 1. 评估所有个体
        for genome in self.population:
            genome.fitness = self.evaluate_fitness(genome, tasks)
        
        # 2. 记录统计
        fitnesses = [g.fitness for g in self.population]
        stats = {
            "generation": self.generation,
            "best_fitness": max(fitnesses),
            "avg_fitness": np.mean(fitnesses),
            "std_fitness": np.std(fitnesses),
            "elite_fitness": [g.fitness for g in self.select_elites()],
        }
        self.history.append(stats)
        
        # 3. 选择精英
        elites = self.select_elites()
        
        # 4. 生成新一代
        new_population = list(elites)  # 保留精英
        
        while len(new_population) < self.pop_size:
            parent1, parent2 = random.sample(elites + self.population[:5], 2)
            child = self.crossover(parent1, parent2)
            child = self.mutate(child, mutation_rate)
            new_population.append(child)
        
        self.population = new_population
        self.generation += 1
        return stats

# 使用示例
if __name__ == "__main__":
    # 创建测试任务集
    tasks = [
        {"type": "web_search", "difficulty": 0.3},
        {"type": "code_generation", "difficulty": 0.5},
        {"type": "data_analysis", "difficulty": 0.4},
        {"type": "document_writing", "difficulty": 0.6},
        {"type": "debugging", "difficulty": 0.7},
        {"type": "system_admin", "difficulty": 0.5},
        {"type": "api_integration", "difficulty": 0.4},
        {"type": "creative_writing", "difficulty": 0.3},
    ]
    
    evolver = GEPEvolver(population_size=15, elite_count=3)
    
    print("🧬 开始 GEP 进化...")
    print(f"{'世代':>4} | {'最佳适应度':>10} | {'平均适应度':>10} | {'标准差':>8}")
    print("-" * 45)
    
    for gen in range(20):
        stats = evolver.evolve_one_generation(tasks, mutation_rate=0.15)
        print(f"{stats['generation']:4d} | {stats['best_fitness']:10.4f} | {stats['avg_fitness']:10.4f} | {stats['std_fitness']:8.4f}")
    
    print(f"
🏆 最终精英个体基因：")
    for i, elite in enumerate(evolver.select_elites()):
        print(f"  Elite #{i+1}: 探索率={elite.exploration_rate:.2f}, "
              f"推理深度={elite.reasoning_depth}, 记忆窗口={elite.memory_window}")