自进化 AI Agent 架构深度解析：GenericAgent、Evolver 与 2026 年 Agent 进化范式

# 一个胶囊示例：数据预处理胶囊
class DataPreprocessingCapsule:
    """可复用的数据预处理模块"""
    genes = [
        Gene("normalize", ["minmax", "zscore"]),
        Gene("handle_missing", ["drop", "fill_mean", "fill_median"]),
        Gene("encode_categorical", ["onehot", "label", "target"]),
    ]
    
    def execute(self, genome, data):
        # 根据基因组合执行预处理流程
        pipeline = decode_genome(genome, self.genes)
        return pipeline.transform(data)

# 事件驱动进化
events = {
    "task_failed": "触发变异操作，探索新策略",
    "performance_degraded": "触发回滚到上一个稳定版本",
    "new_capability_needed": "触发新基因的创建",
    "environment_changed": "触发适应性重评估",
    "evolution_milestone": "记录进化轨迹，生成审计报告",
}

from dataclasses import dataclass, field
from typing import Callable, Optional
from collections import defaultdict
import json
import random

# ============ 技能系统 ============

@dataclass
class Skill:
    """一个技能 = 名称 + 执行函数 + 元数据"""
    name: str
    func: Callable
    category: str
    usage_count: int = 0
    success_count: int = 0
    fail_count: int = 0
    parent: Optional[str] = None
    children: list = field(default_factory=list)
    
    @property
    def success_rate(self) -> float:
        total = self.success_count + self.fail_count
        return self.success_count / total if total > 0 else 0.0
    
    @property
    def is_active(self) -> bool:
        """使用率低于阈值视为不活跃（用于剪枝）"""
        return self.usage_count > 0
    
    def record_result(self, success: bool):
        self.usage_count += 1
        if success:
            self.success_count += 1
        else:
            self.fail_count += 1


class SkillTree:
    """技能树管理器"""
    
    def __init__(self):
        self.skills: dict[str, Skill] = {}
        self.prune_threshold = 0.3  # 成功率低于此值触发剪枝
    
    def register(self, skill: Skill, parent: str = None):
        """注册新技能"""
        self.skills[skill.name] = skill
        if parent and parent in self.skills:
            skill.parent = parent
            self.skills[parent].children.append(skill.name)
    
    def find_skill(self, name: str) -> Optional[Skill]:
        """按名称查找技能"""
        return self.skills.get(name)
    
    def find_by_category(self, category: str) -> list[Skill]:
        """按分类查找技能"""
        return [s for s in self.skills.values() 
                if s.category == category]
    
    def prune(self) -> list[str]:
        """剪枝：移除低成功率且不活跃的技能"""
        to_remove = []
        for name, skill in self.skills.items():
            if (not skill.is_active and 
                skill.success_rate < self.prune_threshold):
                to_remove.append(name)
        for name in to_remove:
            del self.skills[name]
        return to_remove
    
    def get_active_skills(self) -> list[Skill]:
        """获取所有活跃技能（用于构建上下文）"""
        return [s for s in self.skills.values() 
                if s.is_active]
    
    def stats(self) -> dict:
        return {
            "total": len(self.skills),
            "active": len(self.get_active_skills()),
            "avg_success_rate": sum(
                s.success_rate for s in self.skills.values()
            ) / max(len(self.skills), 1),
        }

# ============ 经验系统 ============

@dataclass
class Experience:
    """一次交互经验"""
    task: str
    skill_used: str
    success: bool
    feedback: str = ""
    timestamp: float = 0.0


class ExperienceMemory:
    """经验记忆与模式提取"""
    
    def __init__(self, max_size: int = 1000):
        self.experiences: list[Experience] = []
        self.max_size = max_size
        self.patterns: dict[str, int] = defaultdict(int)
    
    def add(self, exp: Experience):
        self.experiences.append(exp)
        if len(self.experiences) > self.max_size:
            self.experiences = self.experiences[-self.max_size:]
        
        # 提取模式
        key = f"{exp.skill_used}:{exp.success}"
        self.patterns[key] += 1
    
    def get_relevant_experiences(self, task: str) -> list[Experience]:
        """根据任务关键词检索相关经验"""
        task_words = set(task.lower().split())
        return [
            e for e in self.experiences
            if any(w in e.task.lower() for w in task_words)
        ][:5]
    
    def suggest_improvement(self, skill_name: str) -> str:
        """基于经验数据给出改进建议"""
        skill_exps = [e for e in self.experiences 
                      if e.skill_used == skill_name]
        fails = [e for e in skill_exps if not e.success]
        
        if not fails:
            return "当前表现良好，无需改进"
        
        # 分析最常见的失败原因
        failure_feedback = [e.feedback for e in fails if e.feedback]
        if not failure_feedback:
            return "失败但未提供反馈"
        
        # 返回最常见的反馈
        from collections import Counter
        most_common = Counter(failure_feedback).most_common(1)
        return f"建议改进：{most_common[0][0]}"


# ============ 自进化引擎 ============

class SelfEvolvingAgent:
    """自进化 Agent：融合技能树+基因变异+经验累积"""
    
    def __init__(self):
        self.skill_tree = SkillTree()
        self.memory = ExperienceMemory()
        self.generation = 0
    
    def execute_task(self, task: str, skill_name: str, 
                     *args, **kwargs) -> bool:
        """执行任务并记录结果"""
        skill = self.skill_tree.find_skill(skill_name)
        if not skill:
            print(f"❌ 技能 '{skill_name}' 不存在")
            return False
        
        try:
            result = skill.func(*args, **kwargs)
            success = bool(result)
        except Exception as e:
            result = None
            success = False
            print(f"⚠️ 执行失败: {e}")
        
        # 记录技能结果
        skill.record_result(success)
        
        # 记录经验
        exp = Experience(
            task=task,
            skill_used=skill_name,
            success=success,
            feedback=str(result) if not success else "",
        )
        self.memory.add(exp)
        
        # 基因变异：如果连续失败，尝试生成新策略
        if not success and skill.fail_count >= 3:
            self._mutate(skill_name)
        
        return success
    
    def _mutate(self, skill_name: str):
        """基因变异：基于经验生成改进策略"""
        improvement = self.memory.suggest_improvement(skill_name)
        self.generation += 1
        print(f"🧬 [第 {self.generation} 代变异] "
              f"技能 '{skill_name}': {improvement}")
    
    def evolve_cycle(self):
        """执行一个进化周期：剪枝+统计"""
        pruned = self.skill_tree.prune()
        stats = self.skill_tree.stats()
        print(f"\n🔄 进化周期完成:")
        print(f"   总技能: {stats['total']}")
        print(f"   活跃技能: {stats['active']}")
        print(f"   平均成功率: {stats['avg_success_rate']:.1%}")
        if pruned:
            print(f"   已剪枝: {pruned}")


# ============ 使用示例 ============

if __name__ == "__main__":
    agent = SelfEvolvingAgent()
    
    # 注册初始技能
    agent.skill_tree.register(Skill(
        name="add",
        func=lambda a, b: a + b,
        category="math",
    ))
    agent.skill_tree.register(Skill(
        name="multiply",
        func=lambda a, b: a * b,
        category="math",
    ))
    agent.skill_tree.register(Skill(
        name="divide",
        func=lambda a, b: a / b if b != 0 else (_ for _ in ()).throw(
            ValueError("除数不能为零")
        ),
        category="math",
    ))
    
    # 执行任务
    agent.execute_task("计算 2+3", "add", 2, 3)
    agent.execute_task("计算 4*5", "multiply", 4, 5)
    agent.execute_task("计算 10/0", "divide", 10, 0)
    agent.execute_task("计算 10/0", "divide", 10, 0)
    agent.execute_task("计算 10/0", "divide", 10, 0)  # 触发变异
    
    # 执行进化周期
    agent.evolve_cycle()

# 种子技能：Agent 启动时就有的基础能力
SEED_SKILLS = [
    Skill("web_search", web_search_func, "network"),
    Skill("file_read", file_read_func, "filesystem"),
    Skill("code_execute", code_exec_func, "computation"),
]

# 进化空间：允许 Agent 自主探索的方向
EVOLUTION_SPACE = {
    "network": ["api_call", "websocket", "graphql_query"],
    "filesystem": ["file_write", "dir_list", "file_search"],
    "computation": ["math_optimize", "parallel_exec", "cache"],
}

class SafetyGuardrails:
    """进化安全边界"""
    
    MAX_SKILL_DEPTH = 5          # 技能树最大深度
    MAX_MUTATION_RATE = 0.1      # 每代最大变异率
    REQUIRED_AUDIT_TRAIL = True  # 必须保留审计日志
    BLACKLIST_OPERATIONS = [     # 禁止的操作
        "rm -rf /",
        "DROP DATABASE",
        "os.system",
    ]
    
    def validate_skill(self, skill: Skill) -> bool:
        # 检查技能是否在安全范围内
        if skill.name in self.BLACKLIST_OPERATIONS:
            return False
        # 检查技能深度
        depth = self._get_skill_depth(skill)
        if depth > self.MAX_SKILL_DEPTH:
            return False
        return True

# 监控关键指标
METRICS = {
    "success_rate": "> 0.85",      # 总体成功率
    "token_efficiency": "< 5000",  # 平均 token 消耗
    "evolution_stability": "> 0.9", # 进化稳定性
}

# 回滚机制
def rollback_to_generation(agent, target_gen: int):
    """回滚到指定代数"""
    snapshot = agent.evolution_history.get(target_gen)
    if snapshot:
        agent.skill_tree = snapshot.skill_tree
        agent.generation = target_gen
        print(f"✅ 已回滚到第 {target_gen} 代")
    else:
        print(f"❌ 找不到第 {target_gen} 代的快照")