自进化 AI Agent 实战：GenericAgent 架构解析与技能树自主构建系统

# 自进化 Agent 的种子代码核心
from dataclasses import dataclass, field
from typing import Dict, List, Callable, Optional, Any
from enum import Enum
import hashlib
import json

class SkillStatus(Enum):
    DISCOVERING = "discovering"   # 正在发现中
    VALIDATING = "validating"     # 验证中（需要多次成功）
    ACTIVE = "active"             # 活跃可用
    DEPRECATED = "deprecated"     # 已废弃（被更好的技能替代）

@dataclass
class Skill:
    """Agent 的一个技能"""
    name: str
    description: str
    parameters: Dict[str, str]  # 参数名 -> 类型描述
    pattern: Callable           # 技能的实际执行模式
    success_count: int = 0
    failure_count: int = 0
    token_cost_avg: float = 0.0  # 平均 token 消耗
    status: SkillStatus = SkillStatus.DISCOVERING
    dependencies: List[str] = field(default_factory=list)
    abstraction_level: int = 1   # 抽象层级（1=具体操作，5=高度抽象）

    @property
    def success_rate(self) -> float:
        total = self.success_count + self.failure_count
        return self.success_count / total if total > 0 else 0.0

    @property
    def is_reliable(self) -> bool:
        """技能是否足够可靠，可以投入生产使用"""
        return (self.success_count >= 5
                and self.success_rate >= 0.8
                and self.status == SkillStatus.ACTIVE)

class SeedAgent:
    """自进化 Agent 的种子实现"""

    def __init__(self):
        self.skill_tree: Dict[str, Skill] = {}
        self.discovery_log: List[Dict] = []
        self._register_builtin_skills()

    def _register_builtin_skills(self) -> None:
        """注册种子技能——最基本的系统交互能力"""
        self.skill_tree["file_read"] = Skill(
            name="file_read",
            description="读取本地文件内容",
            parameters={"path": "文件路径"},
            pattern=self._file_read,
            abstraction_level=1,
        )
        self.skill_tree["file_write"] = Skill(
            name="file_write",
            description="写入内容到本地文件",
            parameters={"path": "文件路径", "content": "文件内容"},
            pattern=self._file_write,
            abstraction_level=1,
        )
        self.skill_tree["exec_command"] = Skill(
            name="exec_command",
            description="执行系统命令",
            parameters={"command": "要执行的命令"},
            pattern=self._exec_command,
            abstraction_level=1,
        )

    def attempt_skill(self, skill_name: str, **kwargs) -> Any:
        """尝试执行一个技能，记录成功/失败"""
        if skill_name not in self.skill_tree:
            return {"error": f"技能 '{skill_name}' 不存在"}

        skill = self.skill_tree[skill_name]
        try:
            result = skill.pattern(**kwargs)
            skill.success_count += 1
            self._check_for_new_skill()  # 每次操作后检查是否有新技能可发现
            return result
        except Exception as e:
            skill.failure_count += 1
            self._analyze_failure(skill, e)
            return {"error": str(e)}

    def _check_for_new_skill(self) -> None:
        """检查是否有新的技能模式可以被发现和注册"""
        # 分析最近的操作序列，寻找可复用的模式
        patterns = self._detect_patterns()
        for pattern in patterns:
            if self._is_novel_pattern(pattern):
                new_skill = self._abstract_pattern(pattern)
                if new_skill:
                    self.skill_tree[new_skill.name] = new_skill
                    self.discovery_log.append({
                        "event": "skill_discovered",
                        "skill": new_skill.name,
                        "abstraction": new_skill.abstraction_level,
                    })

    def _detect_patterns(self) -> List[Dict]:
        """从操作历史中检测可复用模式"""
        # 简化实现：查找连续使用的技能组合
        recent = self.discovery_log[-20:]
        sequences = self._find_repeated_sequences(recent)
        return sequences

    def _find_repeated_sequences(self, log: List[Dict]) -> List[Dict]:
        """查找重复的操作序列"""
        sequences = []
        for i in range(len(log) - 2):
            if i + 2 < len(log):
                seq = log[i:i+3]
                if seq in log[i+3:]:
                    sequences.append({"sequence": seq, "count": 2})
        return sequences

    def _is_novel_pattern(self, pattern: Dict) -> bool:
        """判断一个模式是否是新的（不与已有技能重复）"""
        pattern_sig = hashlib.md5(
            json.dumps(pattern, sort_keys=True).encode()
        ).hexdigest()[:8]
        return not any(
            hashlib.md5(json.dumps({"name": s.name, "params": s.parameters},
                                   sort_keys=True).encode()).hexdigest()[:8] == pattern_sig
            for s in self.skill_tree.values()
        )

    def _abstract_pattern(self, pattern: Dict) -> Optional[Skill]:
        """将具体操作序列抽象为新技能"""
        # 分析操作序列，提取通用模式
        sequence = pattern.get("sequence", [])
        if not sequence:
            return None

        skill_name = f"composite_{hashlib.md5(json.dumps(sequence).encode()).hexdigest()[:6]}"
        return Skill(
            name=skill_name,
            description=f"复合技能：{' → '.join(s.get('skill', '?') for s in sequence)}",
            parameters={"input": "通用输入"},
            pattern=self._create_composite(sequence),
            abstraction_level=2,
        )

    def _create_composite(self, sequence: List[Dict]) -> Callable:
        """创建复合技能的执行函数"""
        def composite_fn(**kwargs):
            results = []
            for step in sequence:
                skill_name = step.get("skill")
                if skill_name in self.skill_tree:
                    result = self.attempt_skill(skill_name, **kwargs)
                    results.append(result)
            return results
        return composite_fn

    # === 种子技能的具体实现 ===
    def _file_read(self, path: str) -> str:
        with open(path, 'r') as f:
            return f.read()

    def _file_write(self, path: str, content: str) -> str:
        with open(path, 'w') as f:
            f.write(content)
        return f"Written {len(content)} chars to {path}"

    def _exec_command(self, command: str) -> str:
        import subprocess
        result = subprocess.run(command, shell=True, capture_output=True, text=True)
        return result.stdout or result.stderr

    def _analyze_failure(self, skill: Skill, error: Exception) -> None:
        """分析失败，为技能改进提供信息"""
        self.discovery_log.append({
            "event": "failure",
            "skill": skill.name,
            "error": str(error),
        })

# 技能树 DAG 架构
from collections import defaultdict, deque
from typing import Set, List, Optional

class SkillDAG:
    """有向无环图结构的技能树"""

    def __init__(self):
        self.skills: Dict[str, Skill] = {}
        self.dependencies: Dict[str, Set[str]] = defaultdict(set)  # skill -> deps
        self.dependents: Dict[str, Set[str]] = defaultdict(set)    # skill -> used_by

    def add_skill(self, skill: Skill) -> bool:
        """添加技能到 DAG"""
        if skill.name in self.skills:
            return False

        self.skills[skill.name] = skill
        for dep in skill.dependencies:
            self.dependencies[skill.name].add(dep)
            self.dependents[dep].add(skill.name)

        # 验证无环
        if self._has_cycle():
            del self.skills[skill.name]
            for dep in skill.dependencies:
                self.dependencies[skill.name].discard(dep)
                self.dependents[dep].discard(skill.name)
            return False
        return True

    def get_execution_order(self, target_skill: str) -> List[str]:
        """获取执行目标技能所需的所有技能的拓扑排序"""
        if target_skill not in self.skills:
            return []

        # BFS 收集所有依赖
        needed = set()
        queue = deque([target_skill])
        while queue:
            current = queue.popleft()
            if current in needed:
                continue
            needed.add(current)
            for dep in self.dependencies.get(current, set()):
                if dep not in needed:
                    queue.append(dep)

        # 拓扑排序
        result = []
        visited = set()
        for skill in needed:
            self._topo_visit(skill, visited, result)
        return result

    def _topo_visit(self, skill: str, visited: Set[str], result: List[str]) -> None:
        if skill in visited:
            return
        visited.add(skill)
        for dep in self.dependencies.get(skill, set()):
            self._topo_visit(dep, visited, result)
        result.append(skill)

    def _has_cycle(self) -> bool:
        """检测 DAG 中是否有环"""
        visited = set()
        rec_stack = set()

        def dfs(node: str) -> bool:
            visited.add(node)
            rec_stack.add(node)
            for dep in self.dependencies.get(node, set()):
                if dep not in visited:
                    if dfs(dep):
                        return True
                elif dep in rec_stack:
                    return True
            rec_stack.discard(node)
            return False

        return any(dfs(n) for n in self.skills if n not in visited)

    def get_skill_chain(self, target: str) -> Optional[List[Skill]]:
        """获取从叶子到目标的完整技能链"""
        order = self.get_execution_order(target)
        return [self.skills[s] for s in order if s in self.skills]

    def prune_unreliable(self, min_success_rate: float = 0.5,
                         min_executions: int = 10) -> int:
        """修剪不可靠技能"""
        pruned = []
        for name, skill in list(self.skills.items()):
            total = skill.success_count + skill.failure_count
            if (total >= min_executions
                and skill.success_rate < min_success_rate
                and skill.status != SkillStatus.DEPRECATED):
                skill.status = SkillStatus.DEPRECATED
                pruned.append(name)

        # 修剪被废弃技能的依赖者
        for p in pruned:
            for dependent in self.dependents.get(p, set()):
                if dependent in self.skills:
                    self.skills[dependent].status = SkillStatus.DEPRECATED

        return len(pruned)

    def skill_stats(self) -> Dict:
        """获取技能树统计"""
        levels = defaultdict(int)
        statuses = defaultdict(int)
        for skill in self.skills.values():
            levels[skill.abstraction_level] += 1
            statuses[skill.status.value] += 1

        total = len(self.skills)
        active = statuses.get("active", 0)
        return {
            "total_skills": total,
            "active_skills": active,
            "utilization": active / total if total > 0 else 0,
            "by_level": dict(levels),
            "by_status": dict(statuses),
        }

# Token 优化：技能编译与动态工具选择
import time

class CompiledSkill:
    """编译后的技能——用极少的 token 引用代替完整描述"""

    def __init__(self, skill: Skill, compile_threshold: int = 10):
        self.skill = skill
        self.compiled = False
        self.macro_id = f"0x{hash(skill.name) & 0xFFFF:04x}"
        self.original_token_cost = 200  # 估算原始描述 token 数
        self.compiled_token_cost = 8    # 编译后引用 token 数
        self.compile_threshold = compile_threshold

    def should_compile(self) -> bool:
        """判断是否应该编译"""
        return self.skill.success_count >= self.compile_threshold

    def compile(self) -> str:
        """编译为宏指令"""
        self.compiled = True
        return f"SKILL_MACRO:{self.macro_id}"

    def token_savings(self) -> float:
        """单次执行的 token 节省"""
        if self.compiled:
            return self.original_token_cost - self.compiled_token_cost
        return 0

class DynamicToolSelector:
    """动态工具选择器——根据任务选择最相关的工具"""

    def __init__(self, all_skills: Dict[str, CompiledSkill]):
        self.all_skills = all_skills
        self.usage_history: Dict[str, int] = defaultdict(int)

    def select_tools(self, task_description: str,
                     max_tools: int = 10) -> List[str]:
        """根据任务描述选择最相关的工具"""
        # 1. 关键词匹配
        task_words = set(task_description.lower().split())
        scored = {}
        for name, compiled in self.all_skills.items():
            skill_words = set(
                compiled.skill.name.lower().split()
                + compiled.skill.description.lower().split()
            )
            relevance = len(task_words & skill_words) / max(len(task_words), 1)
            # 2. 历史使用频率加权
            history_boost = self.usage_history.get(name, 0) * 0.1
            scored[name] = relevance + history_boost

        # 3. 选择 Top-K
        selected = sorted(scored.items(), key=lambda x: x[1], reverse=True)
        return [name for name, _ in selected[:max_tools]]

    def record_usage(self, skill_name: str) -> None:
        self.usage_history[skill_name] += 1

    def estimate_token_cost(self, task_description: str,
                            max_tools: int = 10) -> Dict[str, float]:
        """估算不同策略的 token 成本"""
        all_described = len(self.all_skills) * 200  # 全部工具描述
        selected = self.select_tools(task_description, max_tools)
        dynamic_cost = len(selected) * 200

        compiled_count = sum(1 for s in selected if self.all_skills[s].compiled)
        mixed_cost = compiled_count * 8 + (len(selected) - compiled_count) * 200

        return {
            "all_tools": all_described,
            "dynamic_selection": dynamic_cost,
            "dynamic_with_compiled": mixed_cost,
            "savings_vs_all": f"{(1 - mixed_cost/all_described)*100:.1f}%",
        }

#!/usr/bin/env python3
"""自进化 Agent 原型演示"""

def demo_self_evolving_agent():
    """完整的自进化 Agent 演示"""
    import tempfile, os

    # 1. 初始化 Agent
    agent = SeedAgent()
    dag = SkillDAG()

    # 注册种子技能到 DAG
    for name, skill in agent.skill_tree.items():
        dag.add_skill(skill)

    print(f"初始技能树: {len(agent.skill_tree)} 个技能")
    print(f"技能: {list(agent.skill_tree.keys())}")
    print()

    # 2. 创建一个临时文件用于演示
    with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
        f.write("Hello, Self-Evolving Agent!\n")
        temp_path = f.name

    # 3. 执行一些操作，触发技能发现
    print("=== 执行操作序列 ===")
    result = agent.attempt_skill("file_read", path=temp_path)
    print(f"读取文件: {result[:50]}..." if isinstance(result, str) else result)

    result = agent.attempt_skill("file_write", path=temp_path, content="Updated content")
    print(f"写入文件: {result}")

    result = agent.attempt_skill("exec_command", command="echo 'done'")
    print(f"执行命令: {result.strip()}")

    # 4. 查看技能树状态
    print()
    print(f"=== 技能树状态 ===")
    stats = dag.skill_stats()
    print(f"总技能: {stats['total_skills']}")
    print(f"活跃技能: {stats['active_skills']}")
    print(f"按层级: {stats['by_level']}")

    # 5. 演示 token 优化
    print()
    print("=== Token 优化演示 ===")
    compiled_skills = {
        name: CompiledSkill(skill)
        for name, skill in agent.skill_tree.items()
    }
    selector = DynamicToolSelector(compiled_skills)

    # 模拟多次使用
    for _ in range(12):
        agent.attempt_skill("file_read", path=temp_path)
        selector.record_usage("file_read")

    # 编译频繁使用的技能
    for name, cs in compiled_skills.items():
        if cs.should_compile():
            cs.compile()
            print(f"编译技能: {name} -> {cs.macro_id}")

    # 估算 token 成本
    costs = selector.estimate_token_cost("读取和写入配置文件")
    print(f"\nToken 成本对比:")
    for strategy, cost in costs.items():
        print(f"  {strategy}: {cost}")

    # 清理
    os.unlink(temp_path)
    print("\n✅ 自进化 Agent 演示完成")

if __name__ == "__main__":
    demo_self_evolving_agent()