Self-Evolving AI Agent（二）：自主进化的范式跃迁

# Claude-Mem 记忆注入伪代码
class ClaudeMemInjector:
    def __init__(self, compression_model="claude-sonnet-4"):
        self.memory_store = VectorStore()  # 语义向量存储
        self.compressor = LLM(compression_model)
        
    def capture_session(self, session):
        """捕获完整会话，包括文件变更、命令执行、对话历史"""
        raw_events = session.get_all_events()
        # 用 LLM 进行多层压缩：事件->关键决策->抽象经验
        compressed = self.compressor.compress(
            raw_events,
            prompt="提取本次会话的核心决策、有效模式和失败教训"
        )
        return CompressedMemory(
            summary=compressed.summary,
            patterns=compressed.code_patterns,
            lessons=compressed.lessons_learned,
            embedding=compressed.vector
        )
    
    def inject_context(self, current_task, k=5):
        """为当前任务注入最相关的历史经验"""
        relevant = self.memory_store.semantic_search(
            query=current_task, top_k=k
        )
        context = "基于历史经验的相关记忆：\n"
        for mem in relevant:
            if mem.relevance_score > 0.75:
                context += f"- {mem.summary}\n"
                for lesson in mem.lessons:
                    context += f"  教训: {lesson}\n"
        return context

# GenericAgent 技能树生长伪代码
class GenericAgentSkillTree:
    def __init__(self, seed_skills):
        self.tree = SkillTree()
        for skill in seed_skills:
            self.tree.add_node(skill)
    
    def attempt_task(self, task):
        """尝试用现有技能树完成任务"""
        matched = self.tree.find_relevant_skills(task)
        
        # 能力评估：现有技能能否覆盖？
        coverage = self._evaluate_coverage(matched, task)
        
        if coverage >= 0.8:
            return self._execute_with_skills(task, matched)
        else:
            # 能力不足，触发技能合成
            new_skill = self._synthesize_skill(task, matched)
            if self._validate_skill(new_skill):
                self.tree.add_node(new_skill)
                return self._execute_with_skills(task, [new_skill])
            return TaskResult(success=False, reason="技能合成失败")
    
    def _synthesize_skill(self, task, base_skills):
        """基于现有技能和任务需求，合成新技能"""
        gap = self._analyze_gap(task, base_skills)
        # 用 LLM 生成技能定义和执行代码
        skill_def = llm.generate(
            prompt=f"基于以下技能缺口创建新技能: {gap}",
            output_schema=SkillDefinition
        )
        return Skill(
            name=skill_def.name,
            code=skill_def.code,
            dependencies=skill_def.depends_on,
            description=skill_def.description
        )

import random
from typing import List, Callable

class GEPEvolver:
    """基于 GEP 的 Agent 策略进化引擎"""
    
    FUNCTIONS = {
        '+': lambda a, b: a + b,
        '-': lambda a, b: a - b,
        '*': lambda a, b: a * b,
        'max': lambda a, b: max(a, b),
        'min': lambda a, b: min(a, b),
    }
    TERMINALS = ['x', 'y', 'z']
    
    def __init__(self, head_length=10, population_size=50):
        self.head_length = head_length
        self.tail_length = head_length * 1 + 1
        self.gene_length = self.head_length + self.tail_length
        self.population_size = population_size
        self.population = self._initialize()
    
    def _initialize(self):
        symbols = list(self.FUNCTIONS.keys()) + self.TERMINALS
        return [''.join(random.choices(symbols, k=self.gene_length))
                for _ in range(self.population_size)]
    
    def evaluate_fitness(self, gene, test_cases):
        """适应度评估：在测试用例集上的平均表现"""
        total = 0.0
        for case in test_cases:
            try:
                env = {'x': case['x'], 'y': case['y'], 'z': case.get('z', 0)}
                expr = self._gene_to_expr(gene)
                result = eval(expr, {"__builtins__": {}}, env)
                total += 1.0 / (1.0 + abs(result - case['expected']))
            except Exception:
                total += 0.01
        return total / len(test_cases)
    
    def evolve(self, generations=100, test_cases=None):
        """主进化循环"""
        if test_cases is None:
            test_cases = self._gen_cases()
        
        for gen in range(generations):
            fitness = [self.evaluate_fitness(g, test_cases) for g in self.population]
            selected = self._tournament_select(fitness)
            
            new_pop = []
            for i in range(0, len(selected) - 1, 2):
                c1, c2 = self._crossover(selected[i], selected[i + 1])
                new_pop.extend([c1, c2])
            
            self.population = [self._mutate(g) for g in new_pop]
            
            if gen % 20 == 0:
                best = max(fitness)
                print(f"Gen {gen}: best={best:.4f}")
        
        final = [self.evaluate_fitness(g, test_cases) for g in self.population]
        return self.population[final.index(max(final))]
    
    def _crossover(self, g1, g2):
        point = random.randint(1, self.head_length - 1)
        return g1[:point] + g2[point:], g2[:point] + g1[point:]
    
    def _mutate(self, gene):
        symbols = list(self.FUNCTIONS.keys()) + self.TERMINALS
        chars = list(gene)
        for i in range(len(chars)):
            if random.random() < 0.05:
                chars[i] = random.choice(symbols)
        return ''.join(chars)
    
    def _gen_cases(self, n=20):
        return [{'x': random.uniform(-10, 10), 'y': random.uniform(-10, 10),
                 'expected': random.uniform(-20, 20)} for _ in range(n)]

# 使用
if __name__ == "__main__":
    evolver = GEPEvolver(head_length=12, population_size=30)
    best = evolver.evolve(generations=100)
    print(f"进化完成！最优基因: {best}")

"""
SelfEvolvingAgent - 记忆+技能树混合自进化 Agent
特点：
1. 自动记录任务执行经验到记忆库
2. 基于经验自动创建和更新技能
3. 技能支持版本控制和回滚
"""
import json
import hashlib
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from collections import Counter

@dataclass
class Experience:
    task_description: str
    actions_taken: List[str]
    result: str
    success: bool
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    embedding: Optional[List[float]] = None

@dataclass
class Skill:
    name: str
    description: str
    trigger_pattern: str
    execution_template: str
    version: int = 1
    success_count: int = 0
    failure_count: int = 0
    created_from: str = ""
    
    @property
    def reliability(self) -> float:
        total = self.success_count + self.failure_count
        return self.success_count / total if total > 0 else 0.0

class SelfEvolvingAgent:
    def __init__(self, agent_id="default"):
        self.agent_id = agent_id
        self.memory: List[Experience] = []
        self.skills: Dict[str, Skill] = {}
        self.skill_threshold = 2
        
    def execute_task(self, task: str, actions: List[str]) -> str:
        # 1. 先尝试用现有技能
        matched = self._find_best_skill(task)
        if matched and matched.reliability > 0.7:
            result = f"[技能] {matched.name}: {matched.execution_template}"
            matched.success_count += 1
            return result
        
        # 2. 基础执行
        result = f"[基础] {task}: {' -> '.join(actions)}"
        success = "error" not in result.lower()
        
        # 3. 记录经验
        exp = Experience(task, actions, result, success)
        self.memory.append(exp)
        print(f"[记忆] {task[:30]}... (成功={success})")
        
        # 4. 检查是否创建新技能
        similar = [e for e in self.memory 
                   if self._similarity(e.task_description, task) > 0.8]
        if len(similar) >= self.skill_threshold:
            name = f"skill_{hash(task) % 10000:04d}"
            if name not in self.skills:
                common = " -> ".join(
                    [a for a, _ in Counter(
                        sum((e.actions_taken for e in similar), [])
                    ).most_common(3)]
                )
                self.skills[name] = Skill(
                    name=name, description=f"自动技能: {task[:60]}",
                    trigger_pattern=task, execution_template=common,
                    created_from=f"基于{len(similar)}条经验",
                    success_count=sum(1 for e in similar if e.success),
                    failure_count=sum(1 for e in similar if not e.success)
                )
                print(f"[进化] 创建技能: {name}")
            else:
                self.skills[name].version += 1
                print(f"[进化] 更新技能: {name} -> v{self.skills[name].version}")
        
        return result
    
    def _find_best_skill(self, task):
        best, score = None, 0.0
        for s in self.skills.values():
            sc = self._similarity(task, s.trigger_pattern)
            if sc > score:
                score, best = sc, s
        return best if score > 0.6 else None
    
    def _similarity(self, a, b):
        sa, sb = set(a.lower().split()), set(b.lower().split())
        if not sa or not sb:
            return 0.0
        return len(sa & sb) / len(sa | sb)

# 运行测试
if __name__ == "__main__":
    agent = SelfEvolvingAgent("my-agent")
    tasks = [
        ("查询用户订单状态", ["调订单API", "解析JSON", "返回状态"]),
        ("查询用户订单信息", ["调订单API", "解析JSON", "返回详情"]),
        ("查询用户订单历史", ["调订单API", "解析JSON", "返回列表"]),
        ("查询用户支付记录", ["调支付API", "解析JSON", "返回记录"]),
        ("查询用户支付历史", ["调支付API", "解析JSON", "返回列表"]),
    ]
    for task, actions in tasks:
        agent.execute_task(task, actions)
    
    print("\n=== 进化统计 ===")
    stats = {
        "memory": len(agent.memory),
        "skills": {k: {"v": v.version, "reliability": f"{v.reliability:.2f}"}
                   for k, v in agent.skills.items()}
    }
    print(json.dumps(stats, indent=2, ensure_ascii=False))