自进化 AI Agent 全景解析：Hermes 107K 星背后的三大技术路线与混合架构实战

"""
GenericAgent 最小化种子实现
展示技能树生长机制的核心逻辑
"""

import json
import os
from typing import Dict, List, Optional
from dataclasses import dataclass, field


@dataclass
class SkillNode:
    """技能树节点"""
    name: str
    description: str
    code: str
    dependencies: List[str] = field(default_factory=list)
    usage_count: int = 0
    confidence: float = 0.0  # 0-1，技能可靠性评分


class SkillTree:
    """技能树管理器"""
    
    def __init__(self, seed_skills: List[SkillNode]):
        self.skills: Dict[str, SkillNode] = {}
        for skill in seed_skills:
            self.skills[skill.name] = skill
    
    def add_skill(self, skill: SkillNode) -> bool:
        """验证并添加新技能"""
        # 检查依赖是否满足
        for dep in skill.dependencies:
            if dep not in self.skills:
                return False
        
        # 检查是否与现有技能冲突
        if skill.name in self.skills:
            return False
        
        self.skills[skill.name] = skill
        return True
    
    def find_relevant_skills(self, task: str) -> List[SkillNode]:
        """根据任务找到相关技能"""
        # 简化版：基于关键词匹配
        relevant = []
        for skill in self.skills.values():
            keywords = task.lower().split()
            if any(kw in skill.description.lower() for kw in keywords):
                relevant.append(skill)
        return sorted(relevant, key=lambda s: s.confidence, reverse=True)
    
    def export_for_llm(self, skill_names: List[str]) -> str:
        """导出技能描述给 LLM（Token 优化版）"""
        lines = ["Available Skills:"]
        for name in skill_names:
            if name in self.skills:
                skill = self.skills[name]
                lines.append(f"  - {name}: {skill.description}")
                skill.usage_count += 1
        return "\n".join(lines)


class GenericAgent:
    """自进化 Agent 核心"""
    
    def __init__(self, llm_client, skill_tree: SkillTree):
        self.llm = llm_client
        self.skills = skill_tree
        self.experience_log: List[Dict] = []
    
    def execute_task(self, task: str) -> str:
        """执行任务的主循环"""
        # 1. 找到相关技能
        relevant = self.skills.find_relevant_skills(task)
        
        # 2. 生成执行计划
        skill_context = self.skills.export_for_llm(
            [s.name for s in relevant]
        )
        plan = self.llm.generate(
            f"Task: {task}\n\n{skill_context}\n\n"
            f"Generate execution plan using available skills."
        )
        
        # 3. 执行计划
        result = self._execute_plan(plan)
        
        # 4. 反思并记录经验
        reflection = self._reflect(task, result)
        self.experience_log.append({
            "task": task,
            "plan": plan,
            "result": result,
            "reflection": reflection
        })
        
        # 5. 尝试发现新技能
        new_skill = self._discover_new_skill(task, reflection)
        if new_skill:
            self.skills.add_skill(new_skill)
            print(f"🌱 新技能已生长: {new_skill.name}")
        
        return result
    
    def _reflect(self, task: str, result: str) -> str:
        """反思：提取可复用的经验"""
        return self.llm.generate(
            f"Task: {task}\nResult: {result}\n\n"
            f"Extract reusable patterns and lessons learned."
        )
    
    def _discover_new_skill(self, task: str, reflection: str) -> Optional[SkillNode]:
        """发现新技能：从经验中提炼可复用模式"""
        discovery = self.llm.generate(
            f"Task: {task}\nReflection: {reflection}\n\n"
            f"Based on this experience, propose a new reusable skill. "
            f"Return JSON with: name, description, code, dependencies."
        )
        # 解析并验证新技能
        try:
            skill_data = json.loads(discovery)
            return SkillNode(
                name=skill_data["name"],
                description=skill_data["description"],
                code=skill_data["code"],
                dependencies=skill_data.get("dependencies", []),
                confidence=0.5  # 新技能初始置信度
            )
        except json.JSONDecodeError:
            return None


# 使用示例
if __name__ == "__main__":
    # 1. 定义种子技能
    seed_skills = [
        SkillNode(
            name="read_file",
            description="Read content from a file",
            code="def read_file(path): return open(path).read()",
        ),
        SkillNode(
            name="write_file",
            description="Write content to a file",
            code="def write_file(path, content): open(path, 'w').write(content)",
        ),
    ]
    
    # 2. 初始化技能树和 Agent
    skill_tree = SkillTree(seed_skills)
    # agent = GenericAgent(llm_client=..., skill_tree=skill_tree)
    
    # 3. Agent 自主运行，技能树逐步生长
    # agent.execute_task("分析 project 目录中的 Python 文件")
    # agent.execute_task("将分析结果写入 report.md")
    # ... 随着任务积累，新技能自动生长
    
    print(f"初始技能数: {len(skill_tree.skills)}")
    print("🚀 GenericAgent 种子代码就绪")

"""
Evolver GEP (Genome Evolution Protocol) 最小实现
展示基因组编码 + 进化循环的核心逻辑
"""

import numpy as np
from typing import List, Tuple, Callable
from dataclasses import dataclass, field
import random


@dataclass
class Genome:
    """Agent 基因组——所有能力参数化"""
    # 行为策略基因：决定 Agent 的决策倾向
    exploration_rate: float = 0.3    # 探索 vs 利用的平衡
    risk_tolerance: float = 0.5       # 风险承受度
    patience: float = 0.7             # 任务坚持度
    
    # 工具使用基因：对不同工具的偏好
    tool_preferences: np.ndarray = field(
        default_factory=lambda: np.random.rand(5)
    )
    
    # 记忆策略基因
    memory_compression: float = 0.5   # 记忆压缩强度
    recall_threshold: float = 0.3     # 记忆召回阈值
    
    def encode(self) -> np.ndarray:
        """将基因组编码为向量"""
        return np.array([
            self.exploration_rate,
            self.risk_tolerance,
            self.patience,
            self.memory_compression,
            self.recall_threshold,
            *self.tool_preferences
        ])
    
    @classmethod
    def decode(cls, vector: np.ndarray) -> 'Genome':
        """从向量解码为基因组"""
        return cls(
            exploration_rate=float(np.clip(vector[0], 0, 1)),
            risk_tolerance=float(np.clip(vector[1], 0, 1)),
            patience=float(np.clip(vector[2], 0, 1)),
            memory_compression=float(np.clip(vector[3], 0, 1)),
            recall_threshold=float(np.clip(vector[4], 0, 1)),
            tool_preferences=np.clip(vector[5:10], 0, 1),
        )


class Evolver:
    """GEP 进化引擎"""
    
    def __init__(
        self,
        population_size: int = 100,
        elite_count: int = 10,
        mutation_rate: float = 0.1,
        crossover_rate: float = 0.7,
    ):
        self.population_size = population_size
        self.elite_count = elite_count
        self.mutation_rate = mutation_rate
        self.crossover_rate = crossover_rate
        self.generation = 0
        self.best_fitness = 0.0
        self.fitness_history: List[float] = []
    
    def initialize_population(self) -> List[Genome]:
        """初始化随机群体"""
        return [Genome() for _ in range(self.population_size)]
    
    def evaluate_fitness(
        self, 
        genome: Genome, 
        task_fn: Callable[[Genome], float]
    ) -> float:
        """评估基因组适应度"""
        return task_fn(genome)
    
    def select_elites(
        self, 
        population: List[Tuple[Genome, float]]
    ) -> List[Genome]:
        """精英选择：保留适应度最高的个体"""
        sorted_pop = sorted(population, key=lambda x: x[1], reverse=True)
        return [ind for ind, _ in sorted_pop[:self.elite_count]]
    
    def crossover(
        self, 
        parent1: Genome, 
        parent2: Genome
    ) -> Genome:
        """单点交叉繁殖"""
        v1, v2 = parent1.encode(), parent2.encode()
        # 随机选择交叉点
        crossover_point = random.randint(1, len(v1) - 1)
        child_vector = np.concatenate([
            v1[:crossover_point],
            v2[crossover_point:]
        ])
        return Genome.decode(child_vector)
    
    def mutate(self, genome: Genome) -> Genome:
        """基因变异"""
        vector = genome.encode()
        # 对每个基因以 mutation_rate 概率施加高斯变异
        mask = np.random.rand(len(vector)) < self.mutation_rate
        vector[mask] += np.random.randn(mask.sum()) * 0.1
        vector = np.clip(vector, 0, 1)
        return Genome.decode(vector)
    
    def evolve(
        self, 
        task_fn: Callable[[Genome], float],
        generations: int = 100
    ) -> Tuple[Genome, float]:
        """执行进化循环"""
        population = self.initialize_population()
        
        for gen in range(generations):
            self.generation = gen
            
            # 评估所有个体
            evaluated = [
                (ind, self.evaluate_fitness(ind, task_fn))
                for ind in population
            ]
            
            # 记录最佳适应度
            best_ind, best_fit = max(evaluated, key=lambda x: x[1])
            self.best_fitness = max(self.best_fitness, best_fit)
            self.fitness_history.append(self.best_fitness)
            
            if gen % 10 == 0:
                print(f"Generation {gen}: "
                      f"Best Fitness = {self.best_fitness:.4f}")
            
            # 精英保留
            elites = self.select_elites(evaluated)
            
            # 生成新一代
            new_population = list(elites)
            while len(new_population) < self.population_size:
                # 选择父母（轮盘赌选择）
                weights = [fit for _, fit in evaluated]
                total = sum(weights)
                probs = [w / total for w in weights]
                parent1 = random.choices(
                    [ind for ind, _ in evaluated], 
                    weights=probs, k=1
                )[0]
                parent2 = random.choices(
                    [ind for ind, _ in evaluated], 
                    weights=probs, k=1
                )[0]
                
                # 交叉 + 变异
                if random.random() < self.crossover_rate:
                    child = self.crossover(parent1, parent2)
                else:
                    child = parent1
                
                child = self.mutate(child)
                new_population.append(child)
            
            population = new_population
        
        # 返回最佳个体
        evaluated = [
            (ind, self.evaluate_fitness(ind, task_fn))
            for ind in population
        ]
        best_ind, best_fit = max(evaluated, key=lambda x: x[1])
        return best_ind, best_fit


# 使用示例：优化一个简单任务
if __name__ == "__main__":
    # 定义适应度函数（模拟 Agent 在任务上的表现）
    def simple_task_fitness(genome: Genome) -> float:
        """
        模拟任务适应度：
        - 探索率和耐心应该平衡
        - 工具偏好应该集中（不要均匀分布）
        """
        # 探索-耐心平衡得分
        balance = 1.0 - abs(genome.exploration_rate - genome.patience)
        
        # 工具偏好集中度（用方差衡量）
        preference_focus = np.var(genome.tool_preferences)
        
        # 综合适应度
        fitness = balance * 0.6 + preference_focus * 0.4
        return min(fitness, 1.0)
    
    # 运行进化
    evolver = Evolver(
        population_size=50,
        elite_count=5,
        mutation_rate=0.15,
        crossover_rate=0.8,
    )
    
    best_genome, best_fitness = evolver.evolve(
        task_fn=simple_task_fitness,
        generations=50
    )
    
    print(f"\n🏆 进化完成!")
    print(f"最佳适应度: {best_fitness:.4f}")
    print(f"最佳基因组: exploration={best_genome.exploration_rate:.3f}, "
          f"patience={best_genome.patience:.3f}")
    print(f"工具偏好: {best_genome.tool_preferences.round(3)}")

┌─────────────────────────────────────────────┐
│           Mixed Evolution Agent             │
├─────────────────────────────────────────────┤
│  Layer 1: Multi-Agent (Hermes-style)        │
│  ├── Planner Agent (规划)                   │
│  ├── Executor Agent (执行)                   │
│  ├── Critic Agent (评估)                     │
│  └── Memory Agent (记忆)                     │
├─────────────────────────────────────────────┤
│  Layer 2: Skill Tree (GenericAgent-style)   │
│  ├── 每个 Agent 拥有独立技能树               │
│  ├── 技能树间可共享和借鉴                    │
│  └── 新技能通过尝试-验证-固化生长            │
├─────────────────────────────────────────────┤
│  Layer 3: Genome Optimization (Evolver)     │
│  ├── Agent 行为策略由基因组编码              │
│  ├── 定期执行群体进化优化基因组              │
│  └── 进化结果应用到所有 Agent 实例           │
└─────────────────────────────────────────────┘

安全层架构：
┌─────────────────────────────┐
│ Layer 0: 人类监督           │ ← 最终决策权
├─────────────────────────────┤
│ Layer 1: 硬约束沙箱         │ ← 不可逾越的边界
├─────────────────────────────┤
│ Layer 2: 软约束奖励函数     │ ← 引导进化方向
├─────────────────────────────┤
│ Layer 3: 自进化 Agent       │ ← 自主进化空间
└─────────────────────────────┘