2026年AI Agent技术突破：自我进化智能体五大核心技术深度解析

import asyncio
import hashlib
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable, Any
from enum import Enum
import json
import re
from pathlib import Path


class EvalResult(Enum):
    """评估结果枚举"""
    PASS = "pass"
    FAIL = "fail"
    DEGRADED = "degraded"
    UNCHANGED = "unchanged"


@dataclass
class TaskResult:
    """任务执行结果"""
    task_id: str
    success: bool
    error_message: Optional[str] = None
    trace: List[Dict[str, Any]] = field(default_factory=list)
    benchmark_score: float = 0.0


@dataclass
class FailurePattern:
    """失败模式"""
    pattern_id: str
    description: str
    root_cause: str
    affected_module: str
    occurrence_count: int = 0
    last_occurrence: Optional[str] = None


@dataclass
class Patch:
    """代码补丁"""
    patch_id: str
    module_path: str
    original_code: str
    patched_code: str
    description: str
    created_at: str
    test_results: Optional[EvalResult] = None


class SourceCodeManager:
    """源代码管理器"""
    
    def __init__(self, source_root: str):
        self.source_root = Path(source_root)
        self.modules: Dict[str, str] = {}
        self.backups: Dict[str, List[str]] = {}
        self._load_modules()
    
    def _load_modules(self):
        """加载所有源代码模块"""
        for py_file in self.source_root.glob("**/*.py"):
            module_name = py_file.stem
            self.modules[module_name] = py_file.read_text()
            self.backups[module_name] = [py_file.read_text()]
    
    def get_module(self, module_name: str) -> str:
        """获取模块源代码"""
        return self.modules.get(module_name, "")
    
    def apply_patch(self, module_name: str, new_code: str) -> bool:
        """应用代码补丁"""
        if module_name not in self.modules:
            return False
        
        # 保存备份
        self.backups[module_name].append(self.modules[module_name])
        self.modules[module_name] = new_code
        
        # 写入文件
        module_path = self.source_root / f"{module_name}.py"
        module_path.write_text(new_code)
        return True
    
    def rollback(self, module_name: str, version: int = -1) -> bool:
        """回滚到指定版本"""
        if module_name not in self.backups or len(self.backups[module_name]) < abs(version):
            return False
        
        self.modules[module_name] = self.backups[module_name][version]
        module_path = self.source_root / f"{module_name}.py"
        module_path.write_text(self.modules[module_name])
        return True


class AutomatedTester:
    """自动化测试器"""
    
    def __init__(self, test_suite_path: str):
        self.test_suite_path = Path(test_suite_path)
        self.benchmark_results: Dict[str, float] = {}
    
    async def run_tests(self, module_name: str) -> Dict[str, bool]:
        """运行测试套件"""
        import subprocess
        
        result = subprocess.run(
            ["pytest", str(self.test_suite_path / f"test_{module_name}.py"), "-v"],
            capture_output=True,
            text=True
        )
        
        return {
            "passed": result.returncode == 0,
            "output": result.stdout + result.stderr
        }
    
    async def run_benchmark(self, benchmark_name: str) -> float:
        """运行基准测试"""
        # 模拟基准测试
        return self.benchmark_results.get(benchmark_name, 0.0)
    
    def compare_results(self, before: float, after: float, threshold: float = 0.95) -> EvalResult:
        """比较基准测试结果"""
        if after >= before:
            return EvalResult.PASS
        elif after >= before * threshold:
            return EvalResult.DEGRADED
        else:
            return EvalResult.FAIL


class FailureAnalyzer:
    """失败分析器 - 使用LLM分析失败根因"""
    
    def __init__(self, llm_client):
        self.llm_client = llm_client
    
    async def analyze(self, failure: TaskResult) -> FailurePattern:
        """分析失败原因"""
        prompt = f"""
        分析以下任务失败的原因：
        
        错误信息: {failure.error_message}
        执行轨迹: {json.dumps(failure.trace, indent=2, ensure_ascii=False)}
        
        请返回JSON格式的分析结果：
        {{
            "root_cause": "根本原因",
            "affected_module": "受影响的模块名",
            "description": "问题描述"
        }}
        """
        
        response = await self.llm_client.complete(prompt)
        analysis = json.loads(response)
        
        return FailurePattern(
            pattern_id=hashlib.md5(failure.task_id.encode()).hexdigest()[:8],
            description=analysis["description"],
            root_cause=analysis["root_cause"],
            affected_module=analysis["affected_module"]
        )


class CodePatcher:
    """代码补丁生成器"""
    
    def __init__(self, llm_client):
        self.llm_client = llm_client
    
    async def generate_patch(
        self, 
        module_code: str, 
        failure: FailurePattern,
        context: str = ""
    ) -> str:
        """生成代码补丁"""
        prompt = f"""
        你是代码修复专家。以下是一个Agent模块的代码和失败分析结果：
        
        模块代码:
        ```python
        {module_code}
        ```
        
        失败分析:
        - 根本原因: {failure.root_cause}
        - 涉及模块: {failure.affected_module}
        - 问题描述: {failure.description}
        
        上下文: {context}
        
        请生成修复后的代码，保持相同的函数签名和接口。只修改必要的部分。
        返回完整的修复后代码。
        """
        
        response = await self.llm_client.complete(prompt)
        
        # 提取代码块
        code_match = re.search(r"```python\n(.*?)```", response, re.DOTALL)
        if code_match:
            return code_match.group(1)
        return response


class MOSSSelfEvolvingAgent:
    """MOSS自我进化Agent主类"""
    
    def __init__(
        self,
        source_root: str,
        test_suite_path: str,
        llm_client,
        base_benchmark: float = 0.85
    ):
        self.source_manager = SourceCodeManager(source_root)
        self.tester = AutomatedTester(test_suite_path)
        self.analyzer = FailureAnalyzer(llm_client)
        self.patcher = CodePatcher(llm_client)
        self.base_benchmark = base_benchmark
        
        self.failure_history: List[FailurePattern] = []
        self.patch_history: List[Patch] = []
        self.current_version: int = 0
    
    async def execute_task(self, task: Dict[str, Any]) -> TaskResult:
        """执行任务"""
        # 这里是实际的任务执行逻辑
        # 简化示例
        return TaskResult(
            task_id=task.get("id", "unknown"),
            success=True,
            benchmark_score=0.9
        )
    
    async def self_improve(self, failure: TaskResult) -> Optional[Patch]:
        """自我改进循环"""
        print(f"[MOSS] 检测到失败，开始自我改进...")
        
        # Step 1: 分析失败
        pattern = await self.analyzer.analyze(failure)
        pattern.occurrence_count += 1
        pattern.last_occurrence = failure.task_id
        self.failure_history.append(pattern)
        
        print(f"[MOSS] 分析完成: {pattern.description}")
        
        # Step 2: 获取受影响的模块
        module_code = self.source_manager.get_module(pattern.affected_module)
        if not module_code:
            print(f"[MOSS] 找不到模块: {pattern.affected_module}")
            return None
        
        # Step 3: 生成补丁
        patched_code = await self.patcher.generate_patch(
            module_code, 
            pattern,
            context=json.dumps(failure.trace, ensure_ascii=False)
        )
        
        # Step 4: 应用并测试
        old_code = module_code
        self.source_manager.apply_patch(pattern.affected_module, patched_code)
        
        # Step 5: 验证非退化
        before_score = failure.benchmark_score
        after_score = await self.tester.run_benchmark(pattern.affected_module)
        
        comparison = self.tester.compare_results(before_score, after_score)
        
        if comparison == EvalResult.FAIL:
            # 回滚
            print(f"[MOSS] 测试失败，回滚更改")
            self.source_manager.rollback(pattern.affected_module)
            return None
        
        # Step 6: 创建补丁记录
        patch = Patch(
            patch_id=hashlib.md5(patched_code.encode()).hexdigest()[:12],
            module_path=pattern.affected_module,
            original_code=old_code,
            patched_code=patched_code,
            description=pattern.description,
            created_at=str(asyncio.get_event_loop().time()),
            test_results=comparison
        )
        self.patch_history.append(patch)
        self.current_version += 1
        
        print(f"[MOSS] 自我改进完成! 版本: {self.current_version}")
        return patch
    
    async def run_loop(self, tasks: List[Dict[str, Any]], max_iterations: int = 100):
        """持续运行循环"""
        iteration = 0
        while iteration < max_iterations:
            for task in tasks:
                result = await self.execute_task(task)
                
                if not result.success:
                    await self.self_improve(result)
                
                iteration += 1
                if iteration >= max_iterations:
                    break


# 使用示例
async def main():
    # 模拟LLM客户端
    class MockLLMClient:
        async def complete(self, prompt: str) -> str:
            return '{"root_cause": "空指针异常", "affected_module": "tool_executor", "description": "未处理None返回值"}'
    
    agent = MOSSSelfEvolvingAgent(
        source_root="./agent_modules",
        test_suite_path="./tests",
        llm_client=MockLLMClient()
    )
    
    tasks = [
        {"id": "task_1", "type": "code_generation"},
        {"id": "task_2", "type": "data_analysis"},
    ]
    
    await agent.run_loop(tasks)


if __name__ == "__main__":
    asyncio.run(main())

package moss

import (
	"context"
	"crypto/md5"
	"encoding/json"
	"fmt"
	"sync"
	"time"
)

// EvalResult 评估结果
type EvalResult string

const (
	EvalPass     EvalResult = "pass"
	EvalFail     EvalResult = "fail"
	EvalDegraded EvalResult = "degraded"
)

// TaskResult 任务执行结果
type TaskResult struct {
	TaskID         string
	Success        bool
	ErrorMessage   string
	Trace          []map[string]interface{}
	BenchmarkScore float64
}

// FailurePattern 失败模式
type FailurePattern struct {
	PatternID      string `json:"pattern_id"`
	Description    string `json:"description"`
	RootCause      string `json:"root_cause"`
	AffectedModule string `json:"affected_module"`
	OccurrenceCnt  int    `json:"occurrence_count"`
	LastOccurrence string `json:"last_occurrence"`
}

// Patch 代码补丁
type Patch struct {
	PatchID      string    `json:"patch_id"`
	ModulePath   string    `json:"module_path"`
	OriginalCode string    `json:"original_code"`
	PatchedCode  string    `json:"patched_code"`
	Description  string    `json:"description"`
	CreatedAt    time.Time `json:"created_at"`
	TestResult   EvalResult `json:"test_result"`
}

// SourceCodeManager 源代码管理器
type SourceCodeManager struct {
	mu      sync.RWMutex
	sources map[string]string
	backups map[string][]string
}

// NewSourceCodeManager 创建源代码管理器
func NewSourceCodeManager() *SourceCodeManager {
	return &SourceCodeManager{
		sources: make(map[string]string),
		backups: make(map[string][]string),
	}
}

// SetModule 设置模块代码
func (s *SourceCodeManager) SetModule(name, code string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.sources[name] = code
	if s.backups[name] == nil {
		s.backups[name] = []string{}
	}
	s.backups[name] = append(s.backups[name], code)
}

// GetModule 获取模块代码
func (s *SourceCodeManager) GetModule(name string) string {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.sources[name]
}

// ApplyPatch 应用补丁
func (s *SourceCodeManager) ApplyPatch(name, newCode string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if _, ok := s.sources[name]; !ok {
		return false
	}
	s.sources[name] = newCode
	s.backups[name] = append(s.backups[name], newCode)
	return true
}

// Rollback 回滚
func (s *SourceCodeManager) Rollback(name string, version int) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if len(s.backups[name]) <= version {
		return false
	}
	s.sources[name] = s.backups[name][version]
	return true
}

// NonDivergenceValidator 非发散验证器 (Ratchet核心)
type NonDivergenceValidator struct {
	mu              sync.RWMutex
	baseScore       float64
	threshold       float64
	historyScores   []float64
	maxHistorySize  int
}

// NewNonDivergenceValidator 创建验证器
func NewNonDivergenceValidator(baseScore float64) *NonDivergenceValidator {
	return &NonDivergenceValidator{
		baseScore:      baseScore,
		threshold:      0.95,
		historyScores:  []float64{},
		maxHistorySize: 100,
	}
}

// Validate 验证补丁是否导致性能退化
func (v *NonDivergenceValidator) Validate(newScore float64) EvalResult {
	v.mu.Lock()
	defer v.mu.Unlock()
	
	v.historyScores = append(v.historyScores, newScore)
	if len(v.historyScores) > v.maxHistorySize {
		v.historyScores = v.historyScores[1:]
	}
	
	if newScore >= v.baseScore {
		// 更新基线
		v.baseScore = (v.baseScore*0.9 + newScore*0.1)
		return EvalPass
	}
	
	if newScore >= v.baseScore*v.threshold {
		return EvalDegraded
	}
	
	return EvalFail
}

// GetCurrentScore 获取当前基线分数
func (v *NonDivergenceValidator) GetCurrentScore() float64 {
	v.mu.RLock()
	defer v.mu.RUnlock()
	return v.baseScore
}

// SelfEvolutionEngine 自我进化引擎
type SelfEvolutionEngine struct {
	sourceManager     *SourceCodeManager
	validator         *NonDivergenceValidator
	failureHistory    []FailurePattern
	patchHistory      []Patch
	version           int
	mu                sync.RWMutex
}

// NewSelfEvolutionEngine 创建自我进化引擎
func NewSelfEvolutionEngine() *SelfEvolutionEngine {
	return &SelfEvolutionEngine{
		sourceManager:  NewSourceCodeManager(),
		validator:      NewNonDivergenceValidator(0.85),
		failureHistory: []FailurePattern{},
		patchHistory:   []Patch{},
		version:        0,
	}
}

// AnalyzeFailure 分析失败
func (e *SelfEvolutionEngine) AnalyzeFailure(result *TaskResult) FailurePattern {
	pattern := FailurePattern{
		PatternID:      fmt.Sprintf("%x", md5.Sum([]byte(result.TaskID)))[:8],
		Description:    result.ErrorMessage,
		RootCause:      "analyzed_root_cause",
		AffectedModule: e.detectAffectedModule(result),
		OccurrenceCnt:  1,
		LastOccurrence: result.TaskID,
	}
	
	e.mu.Lock()
	e.failureHistory = append(e.failureHistory, pattern)
	e.mu.Unlock()
	
	return pattern
}

// detectAffectedModule 检测受影响的模块
func (e *SelfEvolutionEngine) detectAffectedModule(result *TaskResult) string {
	// 简化实现，实际应基于trace分析
	for _, step := range result.Trace {
		if module, ok := step["module"].(string); ok {
			return module
		}
	}
	return "executor"
}

// GeneratePatch 生成补丁
func (e *SelfEvolutionEngine) GeneratePatch(pattern *FailurePattern, moduleCode string) string {
	// 在实际实现中，这里应调用LLM生成代码
	// 简化示例：添加空值检查
	patchedCode := moduleCode
	
	// 简单的代码修复示例
	if pattern.RootCause == "空指针异常" {
		patchedCode = "// [PATCHED] " + moduleCode
	}
	
	return patchedCode
}

// ApplyAndValidate 应用补丁并验证
func (e *SelfEvolutionEngine) ApplyAndValidate(pattern *FailurePattern, patchedCode string, newScore float64) *Patch {
	// 验证
	result := e.validator.Validate(newScore)
	
	patch := &Patch{
		PatchID:      fmt.Sprintf("%x", md5.Sum([]byte(patchedCode)))[:12],
		ModulePath:   pattern.AffectedModule,
		OriginalCode: e.sourceManager.GetModule(pattern.AffectedModule),
		PatchedCode:  patchedCode,
		Description:  pattern.Description,
		CreatedAt:    time.Now(),
		TestResult:   result,
	}
	
	if result == EvalFail {
		// 回滚
		e.sourceManager.Rollback(pattern.AffectedModule, -1)
		return patch
	}
	
	// 应用
	e.sourceManager.ApplyPatch(pattern.AffectedModule, patchedCode)
	e.mu.Lock()
	e.patchHistory = append(e.patchHistory, *patch)
	e.version++
	e.mu.Unlock()
	
	return patch
}

// SelfImprove 自我改进主循环
func (e *SelfEvolutionEngine) SelfImprove(result *TaskResult, newScore float64) *Patch {
	if result.Success {
		return nil
	}
	
	// 分析失败
	pattern := e.AnalyzeFailure(result)
	
	// 获取模块代码
	moduleCode := e.sourceManager.GetModule(pattern.AffectedModule)
	if moduleCode == "" {
		return nil
	}
	
	// 生成补丁
	patchedCode := e.GeneratePatch(&pattern, moduleCode)
	
	// 应用并验证
	return e.ApplyAndValidate(&pattern, patchedCode, newScore)
}

// GetStats 获取统计信息
func (e *SelfEvolutionEngine) GetStats() map[string]interface{} {
	e.mu.RLock()
	defer e.mu.RUnlock()
	
	return map[string]interface{}{
		"version":         e.version,
		"failure_count":   len(e.failureHistory),
		"patch_count":     len(e.patchHistory),
		"current_score":   e.validator.GetCurrentScore(),
	}
}

// String 实现fmt.Stringer
func (p Patch) String() string {
	data, _ := json.MarshalIndent(p, "", "  ")
	return string(data)
}

func ExampleUsage() {
	engine := NewSelfEvolutionEngine()
	
	// 设置初始模块
	engine.sourceManager.SetModule("tool_executor", `def execute_tool(tool_name, params):
    tool = get_tool(tool_name)
    return tool.run(params)  // 可能返回None
`)
	
	// 模拟失败
	failure := &TaskResult{
		TaskID:       "task_001",
		Success:      false,
		ErrorMessage: "空指针异常: NoneType没有属性'process'",
		Trace:        []map[string]interface{}{{"module": "tool_executor"}},
	}
	
	// 自我改进
	patch := engine.SelfImprove(failure, 0.88)
	
	if patch != nil {
		fmt.Printf("改进完成: %s\n", patch.PatchID)
		fmt.Printf("结果: %s\n", patch.TestResult)
	}
	
	// 统计
	stats := engine.GetStats()
	fmt.Printf("统计: %+v\n", stats)
}

"""
Ratchet Guardrails - 防止Agent自我进化失控
"""
import asyncio
import threading
import time
from dataclasses import dataclass, field
from typing import Callable, Optional, Dict, Any, List
from collections import deque
from enum import Enum
import statistics


class GuardState(Enum):
    """防护状态"""
    NOMINAL = "nominal"
    WARNING = "warning"
    CRITICAL = "critical"
    LOCKED = "locked"


@dataclass
class PerformanceRecord:
    """性能记录"""
    timestamp: float
    score: float
    version: str
    action: str


class RatchetGuardrails:
    """
    Ratchet安全护栏 - 确保Agent自我进化不会导致性能退化
    
    核心机制：
    1. 维护性能基线（ratchet point）
    2. 只有性能提升才能移动基线
    3. 性能退化立即回滚
    4. 连续退化触发锁定
    """
    
    def __init__(
        self,
        initial_score: float = 0.8,
        threshold: float = 0.95,
        window_size: int = 10,
        lock_threshold: int = 3,
        lock_duration: float = 300.0
    ):
        self._lock = threading.RLock()
        
        # 配置参数
        self.threshold = threshold  # 基线百分比阈值
        self.window_size = window_size  # 滑动窗口大小
        self.lock_threshold = lock_threshold  # 触发锁定的连续退化次数
        self.lock_duration = lock_duration  # 锁定持续时间（秒）
        
        # 状态
        self.ratchet_point = initial_score  # 当前基线
        self.peak_score = initial_score  # 历史峰值
        self.state = GuardState.NOMINAL
        
        # 历史记录
        self.history: deque[PerformanceRecord] = deque(maxlen=window_size * 2)
        self.degradation_count = 0
        self.lock_until: Optional[float] = None
        
        # 回调
        self.on_degradation: Optional[Callable] = None
        self.on_lock: Optional[Callable] = None
        self.on_unlock: Optional[Callable] = None
    
    @property
    def is_locked(self) -> bool:
        """检查是否被锁定"""
        with self._lock:
            if self.state == GuardState.LOCKED:
                if self.lock_until and time.time() > self.lock_until:
                    self.state = GuardState.NOMINAL
                    self.degradation_count = 0
                    self.lock_until = None
                    if self.on_unlock:
                        self.on_unlock()
                    return False
                return True
            return False
    
    def _check_state(self, new_score: float) -> GuardState:
        """检查状态"""
        ratio = new_score / self.ratchet_point
        
        if ratio >= 1.0:
            return GuardState.NOMINAL
        elif ratio >= self.threshold:
            return GuardState.WARNING
        else:
            return GuardState.CRITICAL
    
    def record(self, score: float, version: str, action: str = "unknown") -> bool:
        """
        记录性能指标，返回是否允许继续
        
        Returns:
            True: 允许继续执行
            False: 被阻止（需要回滚）
        """
        with self._lock:
            # 检查锁定
            if self.is_locked:
                return False
            
            # 记录历史
            record = PerformanceRecord(
                timestamp=time.time(),
                score=score,
                version=version,
                action=action
            )
            self.history.append(record)
            
            # 检查状态
            state = self._check_state(score)
            
            if state == GuardState.NOMINAL:
                # 性能提升，移动基线
                old_point = self.ratchet_point
                self.ratchet_point = max(self.ratchet_point, score)
                if score > self.peak_score:
                    self.peak_score = score
                self.degradation_count = 0
                self.state = GuardState.NOMINAL
                
                if self.ratchet_point > old_point:
                    print(f"[Ratchet] 基线提升: {old_point:.4f} → {self.ratchet_point:.4f}")
                
                return True
            
            elif state == GuardState.WARNING:
                # 轻度退化
                self.degradation_count += 1
                self.state = GuardState.WARNING
                print(f"[Ratchet] 警告: 性能退化 {score/self.ratchet_point:.2%}")
                return True
            
            else:  # CRITICAL
                # 严重退化，回滚
                self.degradation_count += 1
                self.state = GuardState.CRITICAL
                
                if self.degradation_count >= self.lock_threshold:
                    self.state = GuardState.LOCKED
                    self.lock_until = time.time() + self.lock_duration
                    print(f"[Ratchet] 锁定: 连续{self.degradation_count}次退化")
                    if self.on_lock:
                        self.on_lock(self.degradation_count)
                
                if self.on_degradation:
                    self.on_degradation(score, self.ratchet_point)
                
                return False
    
    def get_stats(self) -> Dict[str, Any]:
        """获取统计信息"""
        with self._lock:
            recent_scores = [r.score for r in list(self.history)[-self.window_size:]]
            
            return {
                "state": self.state.value,
                "ratchet_point": self.ratchet_point,
                "peak_score": self.peak_score,
                "degradation_count": self.degradation_count,
                "is_locked": self.is_locked,
                "avg_recent_score": statistics.mean(recent_scores) if recent_scores else 0,
                "min_recent_score": min(recent_scores) if recent_scores else 0,
                "history_size": len(self.history)
            }
    
    def reset(self, new_baseline: Optional[float] = None):
        """重置基线"""
        with self._lock:
            if new_baseline:
                self.ratchet_point = new_baseline
            self.degradation_count = 0
            self.state = GuardState.NOMINAL
            self.lock_until = None
            print(f"[Ratchet] 重置基线: {self.ratchet_point:.4f}")


class MultiLayerGuardrails:
    """
    多层防护系统 - 组合多种安全机制
    """
    
    def __init__(self):
        self.ratchet = RatchetGuardrails(initial_score=0.8)
        self.consistency_checker = ConsistencyChecker()
        self.resource_monitor = ResourceMonitor()
        
        # 设置回调
        self.ratchet.on_lock = self._handle_lock
        self.ratchet.on_degradation = self._handle_degradation
    
    def _handle_lock(self, count: int):
        """处理锁定事件"""
        print(f"[Guardrails] 系统锁定，{count}次连续退化")
    
    def _handle_degradation(self, score: float, baseline: float):
        """处理退化事件"""
        print(f"[Guardrails] 检测到性能退化: {score:.4f} < {baseline:.4f}")
    
    async def pre_execution_check(self, task: Dict[str, Any]) -> bool:
        """执行前检查"""
        if self.ratchet.is_locked:
            return False
        
        if not self.consistency_checker.check(task):
            return False
        
        return True
    
    async def post_execution_check(
        self, 
        result: Dict[str, Any],
        version: str
    ) -> bool:
        """执行后检查"""
        # 检查资源使用
        if not self.resource_monitor.check(result):
            return False
        
        # 检查性能
        score = result.get("benchmark_score", 0.0)
        return self.ratchet.record(score, version, result.get("action", "unknown"))
    
    def get_protection_status(self) -> Dict[str, Any]:
        """获取保护状态"""
        return {
            "ratchet": self.ratchet.get_stats(),
            "consistency": self.consistency_checker.get_stats(),
            "resources": self.resource_monitor.get_stats()
        }


class ConsistencyChecker:
    """一致性检查器"""
    
    def __init__(self):
        self.checks_passed = 0
        self.checks_failed = 0
    
    def check(self, task: Dict[str, Any]) -> bool:
        """检查任务一致性"""
        # 简化实现
        return True
    
    def get_stats(self) -> Dict[str, Any]:
        return {
            "passed": self.checks_passed,
            "failed": self.checks_failed
        }


class ResourceMonitor:
    """资源监控器"""
    
    def __init__(self):
        self.cpu_threshold = 80.0
        self.memory_threshold = 90.0
    
    def check(self, result: Dict[str, Any]) -> bool:
        """检查资源使用"""
        # 简化实现
        return True
    
    def get_stats(self) -> Dict[str, Any]:
        return {
            "cpu_threshold": self.cpu_threshold,
            "memory_threshold": self.memory_threshold
        }


# 使用示例
async def example():
    guardrails = MultiLayerGuardrails()
    
    # 模拟执行循环
    for i in range(20):
        version = f"v1.{i}"
        
        # 模拟结果
        if i < 5:
            score = 0.82 + i * 0.01  # 提升
        elif i < 10:
            score = 0.85 - (i - 5) * 0.02  # 轻微退化
        elif i < 15:
            score = 0.78 + (i - 10) * 0.02  # 恢复
        else:
            score = 0.82 + (i - 15) * 0.01  # 继续提升
        
        result = {
            "benchmark_score": score,
            "action": "task_execution",
            "version": version
        }
        
        allowed = await guardrails.post_execution_check(result, version)
        status = guardrails.get_protection_status()
        
        print(f"迭代 {i}: score={score:.4f}, allowed={allowed}, "
              f"state={status['ratchet']['state']}, "
              f"ratchet={status['ratchet']['ratchet_point']:.4f}")
        
        if not allowed:
            print("[Main] 执行被阻止，等待恢复...")
        
        await asyncio.sleep(0.1)


if __name__ == "__main__":
    asyncio.run(example())

"""
托管Agent客户端 - Google Managed Agents兼容
"""
import asyncio
import json
import hashlib
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any, Callable
from enum import Enum
from datetime import datetime
import aiohttp


class TriggerType(Enum):
    """触发类型"""
    SCHEDULE = "schedule"           # 定时触发
    EVENT = "event"                # 事件触发
    WEBHOOK = "webhook"            # Webhook触发
    MANUAL = "manual"              # 手动触发


@dataclass
class Tool:
    """工具定义"""
    name: str
    description: str
    parameters: Dict[str, Any]
    handler: Optional[Callable] = None
    
    def to_mcp_format(self) -> Dict[str, Any]:
        """转换为MCP格式"""
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.parameters
        }


@dataclass
class AgentInstruction:
    """Agent指令"""
    id: str
    content: str
    priority: int = 0
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class AgentState:
    """Agent状态"""
    session_id: str
    state: str
    memory: Dict[str, Any] = field(default_factory=dict)
    last_updated: datetime = field(default_factory=datetime.now)
    tool_results: Dict[str, Any] = field(default_factory=dict)


class ManagedAgentClient:
    """
    托管Agent客户端
    
    对接Google Managed Agents API
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.magent.googleapis.com/v1",
        project_id: str = None
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.project_id = project_id
        self.session: Optional[aiohttp.ClientSession] = None
        
        # 本地状态
        self.tools: Dict[str, Tool] = {}
        self.active_sessions: Dict[str, AgentState] = {}
        
        # 回调
        self.on_state_change: Optional[Callable] = None
        self.on_tool_call: Optional[Callable] = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession(
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            }
        )
        return self
    
    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()
    
    def register_tool(self, tool: Tool):
        """注册工具"""
        self.tools[tool.name] = tool
        print(f"[ManagedAgent] 注册工具: {tool.name}")
    
    async def create_agent(
        self,
        name: str,
        instructions: List[AgentInstruction],
        triggers: List[Dict[str, Any]]
    ) -> str:
        """创建托管Agent"""
        
        payload = {
            "display_name": name,
            "instructions": [
                {
                    "id": instr.id,
                    "content": instr.content,
                    "priority": instr.priority,
                    "metadata": instr.metadata
                }
                for instr in instructions
            ],
            "tools": [tool.to_mcp_format() for tool in self.tools.values()],
            "triggers": triggers,
            "persistence": {
                "memory_enabled": True,
                "session_timeout_seconds": 86400  # 24小时
            }
        }
        
        async with self.session.post(
            f"{self.base_url}/agents",
            json=payload
        ) as resp:
            if resp.status != 200:
                error = await resp.text()
                raise Exception(f"创建Agent失败: {error}")
            
            result = await resp.json()
            agent_id = result["name"]
            
            print(f"[ManagedAgent] 创建Agent成功: {agent_id}")
            return agent_id
    
    async def invoke_agent(
        self,
        agent_id: str,
        message: str,
        context: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """调用Agent执行任务"""
        
        session_id = hashlib.md5(
            f"{agent_id}:{datetime.now().isoformat()}".encode()
        ).hexdigest()[:16]
        
        payload = {
            "session_id": session_id,
            "message": {
                "content": message,
                "context": context or {}
            },
            "config": {
                "temperature": 0.7,
                "max_iterations": 10,
                "timeout_seconds": 300
            }
        }
        
        async with self.session.post(
            f"{self.base_url}/agents/{agent_id}/sessions/{session_id}:invoke",
            json=payload
        ) as resp:
            if resp.status != 200:
                error = await resp.text()
                raise Exception(f"调用Agent失败: {error}")
            
            result = await resp.json()
            
            # 更新本地状态
            self.active_sessions[session_id] = AgentState(
                session_id=session_id,
                state="completed",
                memory=result.get("session", {}).get("memory", {}),
                tool_results=result.get("tool_results", {})
            )
            
            return result
    
    async def get_session_state(self, session_id: str) -> Optional[AgentState]:
        """获取会话状态"""
        if session_id in self.active_sessions:
            return self.active_sessions[session_id]
        
        async with self.session.get(
            f"{self.base_url}/sessions/{session_id}"
        ) as resp:
            if resp.status != 200:
                return None
            
            result = await resp.json()
            return AgentState(
                session_id=session_id,
                state=result.get("state", "unknown"),
                memory=result.get("memory", {}),
                last_updated=datetime.fromisoformat(
                    result.get("last_updated", datetime.now().isoformat())
                )
            )
    
    async def list_active_agents(self) -> List[Dict[str, Any]]:
        """列出所有活跃Agent"""
        async with self.session.get(
            f"{self.base_url}/agents",
            params={"filter": "state=ACTIVE"}
        ) as resp:
            result = await resp.json()
            return result.get("agents", [])


class AgentBuilder:
    """
    Agent构建器 - 简化托管Agent创建
    """
    
    def __init__(self, name: str):
        self.name = name
        self.instructions: List[AgentInstruction] = []
        self.tools: List[Tool] = []
        self.triggers: List[Dict[str, Any]] = []
    
    def add_instruction(
        self, 
        content: str, 
        priority: int = 0,
        metadata: Optional[Dict[str, Any]] = None
    ) -> "AgentBuilder":
        """添加指令"""
        instr = AgentInstruction(
            id=hashlib.md5(content.encode()).hexdigest()[:8],
            content=content,
            priority=priority,
            metadata=metadata or {}
        )
        self.instructions.append(instr)
        return self
    
    def add_tool(
        self,
        name: str,
        description: str,
        parameters: Dict[str, Any],
        handler: Optional[Callable] = None
    ) -> "AgentBuilder":
        """添加工具"""
        tool = Tool(
            name=name,
            description=description,
            parameters=parameters,
            handler=handler
        )
        self.tools.append(tool)
        return self
    
    def add_schedule_trigger(
        self,
        cron_expression: str
    ) -> "AgentBuilder":
        """添加定时触发"""
        self.triggers.append({
            "type": TriggerType.SCHEDULE.value,
            "schedule": {
                "cron": cron_expression,
                "timezone": "Asia/Shanghai"
            }
        })
        return self
    
    def add_event_trigger(
        self,
        event_type: str,
        filter_expression: Optional[str] = None
    ) -> "AgentBuilder":
        """添加事件触发"""
        trigger = {
            "type": TriggerType.EVENT.value,
            "event": {
                "type": event_type
            }
        }
        if filter_expression:
            trigger["event"]["filter"] = filter_expression
        self.triggers.append(trigger)
        return self
    
    def add_webhook_trigger(
        self,
        path: str,
        methods: Optional[List[str]] = None
    ) -> "AgentBuilder":
        """添加Webhook触发"""
        self.triggers.append({
            "type": TriggerType.WEBHOOK.value,
            "webhook": {
                "path": path,
                "methods": methods or ["POST"]
            }
        })
        return self
    
    def build(self, client: ManagedAgentClient) -> str:
        """构建Agent"""
        return asyncio.run(
            client.create_agent(
                name=self.name,
                instructions=self.instructions,
                triggers=self.triggers
            )
        )


# 使用示例
async def main():
    async with ManagedAgentClient(
        api_key="your-api-key",
        project_id="your-project-id"
    ) as client:
        
        # 定义工具
        client.register_tool(Tool(
            name="send_email",
            description="发送电子邮件",
            parameters={
                "type": "object",
                "properties": {
                    "to": {"type": "string", "description": "收件人邮箱"},
                    "subject": {"type": "string", "description": "邮件主题"},
                    "body": {"type": "string", "description": "邮件正文"}
                },
                "required": ["to", "subject", "body"]
            }
        ))
        
        # 使用构建器创建Agent
        agent_id = (
            AgentBuilder("Daily Report Agent")
            .add_instruction(
                "你是一个报告生成助手。每天早上9点自动生成昨日数据报告，"
                "包括销售数据、用户活跃度和系统健康状态。",
                priority=10
            )
            .add_instruction(
                "报告生成后，发送邮件给团队成员。",
                priority=5
            )
            .add_schedule_trigger("0 9 * * *")  # 每天9点
            .add_tool(
                name="fetch_data",
                description="获取指定日期的数据",
                parameters={
                    "type": "object",
                    "properties": {
                        "date": {"type": "string"},
                        "data_type": {"type": "string", "enum": ["sales", "users", "system"]}
                    }
                }
            )
            .build(client)
        )
        
        # 调用Agent
        result = await client.invoke_agent(
            agent_id=agent_id,
            message="生成今日数据报告"
        )
        
        print(f"Agent执行结果: {json.dumps(result, indent=2, ensure_ascii=False)}")


if __name__ == "__main__":
    asyncio.run(main())

"""
工作流编译器 - 将多Agent管道编译为单一模型
"""
import asyncio
import json
import hashlib
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any, Callable, Tuple
from collections import defaultdict
import statistics


@dataclass
class WorkflowStep:
    """工作流步骤"""
    name: str
    agent_type: str
    prompt_template: str
    input_mapping: Dict[str, str]  # 从上一步映射哪些输出
    output_field: str  # 本步输出的字段名


@dataclass
class ExecutionTrace:
    """执行轨迹"""
    workflow_id: str
    step_results: Dict[str, Any]
    final_output: Any
    total_cost: float
    total_latency: float
    timestamp: float


@dataclass
class CompiledModel:
    """编译后的模型"""
    model_id: str
    base_model: str
    training_data: List[Dict[str, Any]]
    fine_tuned_weights_path: str
    compression_ratio: float  # 压缩比（原成本/新成本）
    quality_retention: float    # 质量保留率


class WorkflowTracer:
    """工作流追踪器 - 收集执行轨迹"""
    
    def __init__(self, workflow_id: str):
        self.workflow_id = workflow_id
        self.traces: List[ExecutionTrace] = []
        self.current_trace: Optional[Dict[str, Any]] = None
    
    def start_trace(self):
        """开始追踪"""
        import time
        self.current_trace = {
            "workflow_id": self.workflow_id,
            "start_time": time.time(),
            "step_results": {},
            "costs": []
        }
    
    def record_step(
        self, 
        step_name: str, 
        result: Any, 
        cost: float,
        latency: float
    ):
        """记录步骤结果"""
        if self.current_trace:
            self.current_trace["step_results"][step_name] = {
                "result": result,
                "cost": cost,
                "latency": latency
            }
            self.current_trace["costs"].append(cost)
    
    def end_trace(self, final_output: Any) -> ExecutionTrace:
        """结束追踪"""
        import time
        if not self.current_trace:
            raise ValueError("No active trace")
        
        end_time = time.time()
        trace = ExecutionTrace(
            workflow_id=self.workflow_id,
            step_results=self.current_trace["step_results"],
            final_output=final_output,
            total_cost=sum(self.current_trace["costs"]),
            total_latency=end_time - self.current_trace["start_time"],
            timestamp=self.current_trace["start_time"]
        )
        
        self.traces.append(trace)
        self.current_trace = None
        return trace
    
    def get_common_patterns(self, min_occurrence: int = 5) -> List[Dict[str, Any]]:
        """从轨迹中提取常见模式"""
        # 简化实现：按输入-输出对分组
        pattern_groups = defaultdict(list)
        
        for trace in self.traces:
            first_step = list(trace.step_results.keys())[0] if trace.step_results else ""
            last_output = trace.final_output
            
            key = hashlib.md5(
                f"{first_step}:{str(last_output)[:100]}".encode()
            ).hexdigest()
            pattern_groups[key].append(trace)
        
        # 过滤常见模式
        common = [
            {"pattern_id": k, "count": len(v), "sample": v[0]}
            for k, v in pattern_groups.items()
            if len(v) >= min_occurrence
        ]
        
        return sorted(common, key=lambda x: x["count"], reverse=True)


class WorkflowCompiler:
    """
    工作流编译器
    
    核心流程：
    1. 收集大量执行轨迹
    2. 提取输入-输出对
    3. 对小模型进行微调
    4. 验证质量保留
    """
    
    def __init__(
        self,
        base_model: str = "gpt-4o-mini",
        target_model: str = "gpt-4o-mini-finetuned",
        quality_threshold: float = 0.90
    ):
        self.base_model = base_model
        self.target_model = target_model
        self.quality_threshold = quality_threshold
        
        self.tracer = WorkflowTracer("compiler")
        self.training_data: List[Dict[str, Any]] = []
        self.pattern_cache: Dict[str, Any] = {}
    
    def collect_traces(self, workflow_executor, tasks: List[Dict[str, Any]]):
        """收集执行轨迹"""
        for task in tasks:
            self.tracer.start_trace()
            
            # 执行工作流（模拟）
            result = asyncio.run(workflow_executor.execute(task))
            
            # 记录轨迹
            self.tracer.end_trace(result["final_output"])
        
        print(f"[Compiler] 收集了 {len(self.tracer.traces)} 条轨迹")
    
    def extract_training_pairs(self) -> List[Dict[str, Any]]:
        """从轨迹中提取训练对"""
        pairs = []
        
        for trace in self.tracer.traces:
            # 提取 <input, output> 对
            first_step = list(trace.step_results.keys())[0]
            first_result = trace.step_results[first_step]["result"]
            
            pair = {
                "input": json.dumps({
                    "task": first_result.get("task", ""),
                    "context": first_result.get("context", {})
                }),
                "output": json.dumps(trace.final_output),
                "quality_score": self._estimate_quality(trace)
            }
            pairs.append(pair)
        
        self.training_data = pairs
        return pairs
    
    def _estimate_quality(self, trace: ExecutionTrace) -> float:
        """估算输出质量"""
        # 简化实现
        base_score = 0.8
        cost_bonus = min(trace.total_cost / 0.5, 0.2)  # 成本越高质量越好
        return min(base_score + cost_bonus, 1.0)
    
    async def compile(self, llm_client) -> CompiledModel:
        """
        编译工作流为模型
        
        实际实现中，这里应该：
        1. 准备微调数据格式
        2. 调用微调API（如OpenAI Fine-tuning）
        3. 等待训练完成
        4. 验证模型质量
        """
        
        # Step 1: 提取训练对
        training_pairs = self.extract_training_pairs()
        
        # Step 2: 过滤高质量样本
        high_quality = [
            p for p in training_pairs 
            if p["quality_score"] >= self.quality_threshold
        ]
        
        print(f"[Compiler] 高质量样本: {len(high_quality)}/{len(training_pairs)}")
        
        # Step 3: 准备微调数据
        formatted_data = [
            {
                "messages": [
                    {"role": "user", "content": p["input"]},
                    {"role": "assistant", "content": p["output"]}
                ]
            }
            for p in high_quality
        ]
        
        # Step 4: 模拟微调（实际应调用API）
        model_id = f"ft-{hashlib.md5(str(high_quality).encode()).hexdigest()[:8]}"
        
        compiled = CompiledModel(
            model_id=model_id,
            base_model=self.base_model,
            training_data=formatted_data,
            fine_tuned_weights_path=f"./models/{model_id}",
            compression_ratio=100,  # 假设100倍压缩
            quality_retention=0.92
        )
        
        return compiled
    
    def verify_compilation(
        self, 
        compiled: CompiledModel,
        test_cases: List[Dict[str, Any]]
    ) -> Dict[str, float]:
        """验证编译质量"""
        results = []
        
        for test in test_cases:
            # 原始流程
            original_cost = test.get("original_cost", 0.5)
            original_latency = test.get("original_latency", 30)
            
            # 编译后
            compiled_cost = test.get("compiled_cost", 0.005)
            compiled_latency = test.get("compiled_latency", 2)
            
            # 质量比
            quality_ratio = test.get("quality_ratio", 0.95)
            
            results.append({
                "cost_reduction": original_cost / compiled_cost,
                "latency_reduction": original_latency / compiled_latency,
                "quality_retention": quality_ratio
            })
        
        return {
            "avg_cost_reduction": statistics.mean([r["cost_reduction"] for r in results]),
            "avg_latency_reduction": statistics.mean([r["latency_reduction"] for r in results]),
            "avg_quality_retention": statistics.mean([r["quality_retention"] for r in results])
        }


class CompiledModelRouter:
    """
    编译模型路由器
    
    策略：
    - 简单请求 → 编译模型（快速、便宜）
    - 复杂请求 → 完整Agent管道
    """
    
    def __init__(
        self,
        compiled_model: CompiledModel,
        full_agent,
        complexity_threshold: float = 0.7
    ):
        self.compiled_model = compiled_model
        self.full_agent = full_agent
        self.complexity_threshold = complexity_threshold
    
    def estimate_complexity(self, request: Dict[str, Any]) -> float:
        """估算请求复杂度"""
        # 简化实现
        text_length = len(request.get("message", ""))
        has_context = len(request.get("context", {})) > 0
        
        base = 0.3
        length_score = min(text_length / 1000, 0.4)
        context_score = 0.3 if has_context else 0.0
        
        return min(base + length_score + context_score, 1.0)
    
    async def route(
        self, 
        request: Dict[str, Any],
        llm_client
    ) -> Dict[str, Any]:
        """路由请求"""
        complexity = self.estimate_complexity(request)
        
        if complexity <= self.complexity_threshold:
            # 使用编译模型
            return await self._call_compiled_model(request, llm_client)
        else:
            # 使用完整Agent
            return await self._call_full_agent(request)
    
    async def _call_compiled_model(
        self, 
        request: Dict[str, Any],
        llm_client
    ) -> Dict[str, Any]:
        """调用编译模型"""
        # 模拟调用
        return {
            "source": "compiled_model",
            "model_id": self.compiled_model.model_id,
            "result": "compiled output",
            "cost": 0.005,
            "latency": 2.0
        }
    
    async def _call_full_agent(
        self, 
        request: Dict[str, Any]
    ) -> Dict[str, Any]:
        """调用完整Agent"""
        # 模拟调用
        return {
            "source": "full_agent",
            "result": "full agent output",
            "cost": 0.5,
            "latency": 30.0
        }


# 使用示例
async def main():
    compiler = WorkflowCompiler(
        base_model="gpt-4o-mini",
        target_model="my-compiled-model"
    )
    
    # 模拟收集轨迹
    # 实际使用时需要执行真实工作流
    
    print("[Compiler] 开始编译...")
    compiled = await compiler.compile(None)
    
    print(f"[Compiler] 编译完成:")
    print(f"  - 模型ID: {compiled.model_id}")
    print(f"  - 训练样本: {len(compiled.training_data)}")
    print(f"  - 压缩比: {compiled.compression_ratio}x")
    print(f"  - 质量保留: {compiled.quality_retention:.1%}")
    
    # 验证
    test_cases = [
        {"original_cost": 0.5, "compiled_cost": 0.005, "quality_ratio": 0.95},
        {"original_cost": 0.5, "compiled_cost": 0.005, "quality_ratio": 0.92},
    ]
    
    verification = compiler.verify_compilation(compiled, test_cases)
    print(f"\n验证结果:")
    print(f"  - 成本降低: {verification['avg_cost_reduction']:.0f}x")
    print(f"  - 延迟降低: {verification['avg_latency_reduction']:.0f}x")
    print(f"  - 质量保留: {verification['avg_quality_retention']:.1%}")


if __name__ == "__main__":
    asyncio.run(main())

"""
预测性规划器 - IdleSpec实现
"""
import asyncio
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any, Callable
from collections import defaultdict
import random


@dataclass
class ToolResult:
    """工具执行结果"""
    tool_name: str
    success: bool
    result: Any
    latency: float
    error: Optional[str] = None


@dataclass
class SpeculativeAction:
    """预测动作"""
    action_id: str
    predicted_next_step: str
    confidence: float
    precomputed_result: Optional[Any] = None
    actual_tool_result: Optional[ToolResult] = None


class IdleSpecPlanner:
    """
    预测性规划器
    
    在等待工具结果时，预测可能的下一个动作并预计算
    """
    
    def __init__(
        self,
        max_speculations: int = 3,
        min_confidence: float = 0.6
    ):
        self.max_speculations = max_speculations
        self.min_confidence = min_confidence
        
        self.history: List[Dict[str, str]] = []  # <tool, next_tool> 对
        self.transition_probs: Dict[str, Dict[str, float]] = defaultdict(
            lambda: defaultdict(float)
        )
        
        self.speculations: Dict[str, SpeculativeAction] = {}
    
    def learn_from_trace(self, trace: List[str]):
        """从轨迹学习转移概率"""
        for i in range(len(trace) - 1):
            current = trace[i]
            next_step = trace[i + 1]
            
            self.history.append({"from": current, "to": next_step})
            self.transition_probs[current][next_step] += 1
        
        # 归一化为概率
        for from_tool, to_tools in self.transition_probs.items():
            total = sum(to_tools.values())
            for to_tool in to_tools:
                self.transition_probs[from_tool][to_tool] /= total
    
    def predict_next(self, current_tool: str) -> List[SpeculativeAction]:
        """预测下一步"""
        if current_tool not in self.transition_probs:
            return []
        
        candidates = self.transition_probs[current_tool]
        
        # 排序并取top-k
        sorted_candidates = sorted(
            candidates.items(),
            key=lambda x: x[1],
            reverse=True
        )[:self.max_speculations]
        
        speculations = []
        for i, (next_step, prob) in enumerate(sorted_candidates):
            if prob >= self.min_confidence:
                speculation = SpeculativeAction(
                    action_id=f"spec_{i}",
                    predicted_next_step=next_step,
                    confidence=prob
                )
                speculations.append(speculation)
                self.speculations[speculation.action_id] = speculation
        
        return speculations
    
    async def speculative_execute(
        self,
        tool_name: str,
        tool_executor: Callable,
        params: Dict[str, Any]
    ) -> ToolResult:
        """
        预测性执行
        
        流程：
        1. 启动工具执行（异步）
        2. 同时预测下一步
        3. 如果预测可信度高，预计算
        4. 等待实际结果
        5. 如果预测命中，使用预计算结果
        """
        start_time = time.time()
        
        # 预测下一步
        speculations = self.predict_next(tool_name)
        speculative_tasks = []
        
        # 并行执行：工具调用 + 预测计算
        async def execute_tool():
            return await tool_executor(tool_name, params)
        
        tool_task = asyncio.create_task(execute_tool())
        
        # 预计算预测结果
        for spec in speculations:
            if spec.confidence >= self.min_confidence:
                task = asyncio.create_task(
                    self._precompute(spec.predicted_next_step)
                )
                speculative_tasks.append((spec.action_id, task))
        
        # 等待工具完成
        tool_result = await tool_task
        actual_latency = time.time() - start_time
        
        # 检查预测命中
        for spec_id, task in speculative_tasks:
            spec = self.speculations[spec_id]
            if not task.done():
                continue
            
            spec.actual_tool_result = await task
            
            # 如果预测正确，可以提前返回
            # 实际使用中可以更激进地使用预计算结果
        
        return tool_result
    
    async def _precompute(self, step: str) -> Any:
        """预计算步骤结果"""
        # 模拟预计算
        await asyncio.sleep(random.uniform(0.5, 1.5))
        return {"precomputed": step, "status": "ready"}


class LatentCommunicationGuard:
    """
    Latent Communication Guard (LCGuard)
    
    多Agent系统中的KV-Cache共享存在安全风险：
    一个被攻陷的Agent可能通过KV-Cache污染其他Agent
    
    LCGuard通过潜伏通信检测来防护这一风险
    """
    
    def __init__(self, num_agents: int):
        self.num_agents = num_agents
        self.kv_sharing_groups: Dict[int, List[int]] = {}  # Agent组
        self.anomaly_scores: Dict[int, float] = defaultdict(float)
        self.baseline_patterns: Dict[str, List[float]] = {}
    
    def setup_sharing_group(self, group_id: int, agent_ids: List[int]):
        """设置KV共享组"""
        self.kv_sharing_groups[group_id] = agent_ids
        print(f"[LCGuard] 设置共享组 {group_id}: {agent_ids}")
    
    def record_access_pattern(
        self, 
        agent_id: int, 
        shared_key: str,
        access_type: str  # "read" or "write"
    ):
        """记录访问模式"""
        pattern_key = f"{agent_id}:{shared_key}"
        
        if pattern_key not in self.baseline_patterns:
            self.baseline_patterns[pattern_key] = []
        
        # 记录访问（简化：记录write比例）
        is_write = 1.0 if access_type == "write" else 0.0
        self.baseline_patterns[pattern_key].append(is_write)
    
    def detect_anomaly(self, agent_id: int) -> float:
        """检测异常"""
        # 简化实现：检测写入比例异常
        agent_patterns = {
            k: v for k, v in self.baseline_patterns.items()
            if k.startswith(f"{agent_id}:")
        }
        
        if not agent_patterns:
            return 0.0
        
        anomalies = []
        for pattern_key, values in agent_patterns.items():
            if len(values) < 10:
                continue
            
            recent_write_ratio = sum(values[-5:]) / len(values[-5:])
            baseline_write_ratio = sum(values[:-5]) / len(values[:-5]) if len(values) > 5 else 0.5
            
            # 写入比例突然变化可能是攻击
            if baseline_write_ratio > 0:
                change_ratio = abs(recent_write_ratio - baseline_write_ratio) / baseline_write_ratio
                anomalies.append(change_ratio)
        
        return max(anomalies) if anomalies else 0.0
    
    def should_isolate(self, agent_id: int, threshold: float = 0.8) -> bool:
        """判断是否应隔离Agent"""
        anomaly_score = self.detect_anomaly(agent_id)
        self.anomaly_scores[agent_id] = anomaly_score
        
        if anomaly_score >= threshold:
            print(f"[LCGuard] 警告: Agent {agent_id} 异常分数 {anomaly_score:.2f}，建议隔离")
            return True
        
        return False
    
    def get_status_report(self) -> Dict[str, Any]:
        """获取状态报告"""
        return {
            "num_agents": self.num_agents,
            "sharing_groups": len(self.kv_sharing_groups),
            "anomaly_scores": dict(self.anomaly_scores),
            "baseline_patterns_count": len(self.baseline_patterns)
        }


# 使用示例
async def idle_spec_example():
    planner = IdleSpecPlanner()
    
    # 学习历史
    traces = [
        ["fetch_data", "process_data", "format_output"],
        ["fetch_data", "validate", "format_output"],
        ["fetch_data", "process_data", "analyze", "format_output"],
    ]
    
    for trace in traces:
        planner.learn_from_trace(trace)
    
    # 预测
    predictions = planner.predict_next("fetch_data")
    print(f"fetch_data后的可能步骤:")
    for p in predictions:
        print(f"  - {p.predicted_next_step}: {p.confidence:.2%}")
    
    # 模拟执行
    async def mock_executor(tool, params):
        await asyncio.sleep(1)  # 模拟网络延迟
        return ToolResult(tool_name=tool, success=True, result="ok", latency=1.0)
    
    result = await planner.speculative_execute("fetch_data", mock_executor, {})
    print(f"\n执行结果: {result.tool_name}, 延迟: {result.latency}s")


def lcguard_example():
    guard = LatentCommunicationGuard(num_agents=4)
    
    # 设置共享组
    guard.setup_sharing_group(0, [0, 1])
    guard.setup_sharing_group(1, [2, 3])
    
    # 模拟正常访问
    for _ in range(20):
        guard.record_access_pattern(0, "shared_kb", "read")
        guard.record_access_pattern(1, "shared_kb", "read")
    
    # 模拟异常（Agent 1 开始大量写入）
    for _ in range(5):
        guard.record_access_pattern(1, "shared_kb", "write")
    
    # 检测
    should_isolate = guard.should_isolate(1)
    print(f"Agent 1 应该隔离: {should_isolate}")
    
    print(f"\n状态报告: {guard.get_status_report()}")


if __name__ == "__main__":
    asyncio.run(idle_spec_example())
    lcguard_example()

"""
完整的自我进化Agent系统
整合所有组件
"""
import asyncio
from dataclasses import dataclass
from typing import Dict, Any, Optional

from moss_engine import MOSSSelfEvolvingAgent, SourceCodeManager
from ratchet_guardrails import MultiLayerGuardrails, RatchetGuardrails
from managed_agent_client import ManagedAgentClient, AgentBuilder, Tool
from workflow_compiler import WorkflowCompiler, CompiledModelRouter
from idle_spec import IdleSpecPlanner, LatentCommunicationGuard


@dataclass
class SystemConfig:
    """系统配置"""
    # 自我进化配置
    source_root: str = "./agent_modules"
    test_suite_path: str = "./tests"
    base_benchmark: float = 0.85
    
    # Ratchet配置
    ratchet_threshold: float = 0.95
    lock_threshold: int = 3
    
    # 编译配置
    compile_threshold: float = 0.7
    quality_threshold: float = 0.90
    
    # 预测性规划配置
    max_speculations: int = 3
    speculation_confidence: float = 0.6


class SelfEvolvingAgentSystem:
    """
    自我进化Agent系统 - 整合所有组件
    """
    
    def __init__(self, config: SystemConfig):
        self.config = config
        
        # 核心组件
        self.evolver = None  # MOSS引擎
        self.guardrails: Optional[MultiLayerGuardrails] = None
        self.compiler: Optional[WorkflowCompiler] = None
        self.router: Optional[CompiledModelRouter] = None
        self.planner = IdleSpecPlanner(
            max_speculations=config.max_speculations,
            min_confidence=config.speculation_confidence
        )
        self.lcguard = LatentCommunicationGuard(num_agents=4)
        
        self.initialized = False
    
    async def initialize(self, llm_client):
        """初始化系统"""
        # 初始化MOSS引擎
        self.evolver = MOSSSelfEvolvingAgent(
            source_root=self.config.source_root,
            test_suite_path=self.config.test_suite_path,
            llm_client=llm_client,
            base_benchmark=self.config.base_benchmark
        )
        
        # 初始化Ratchet护栏
        self.guardrails = MultiLayerGuardrails()
        self.guardrails.ratchet.threshold = self.config.ratchet_threshold
        self.guardrails.ratchet.lock_threshold = self.config.lock_threshold
        
        # 初始化工作流编译器
        self.compiler = WorkflowCompiler(
            quality_threshold=self.config.quality_threshold
        )
        
        self.initialized = True
        print("[System] 自我进化Agent系统初始化完成")
    
    async def execute_task(
        self, 
        task: Dict[str, Any],
        mode: str = "auto"  # "auto", "evolve", "compiled", "safe"
    ) -> Dict[str, Any]:
        """执行任务"""
        if not self.initialized:
            raise RuntimeError("系统未初始化")
        
        if mode == "safe":
            # 安全模式：只执行不进化
            return await self._execute_safe(task)
        
        elif mode == "evolve":
            # 进化模式：允许自我改进
            return await self._execute_with_evolution(task)
        
        elif mode == "compiled":
            # 编译模式：使用编译模型
            return await self._execute_compiled(task)
        
        else:  # auto
            # 自动模式：根据复杂度选择
            complexity = self.router.estimate_complexity(task) if self.router else 0.5
            if complexity <= self.config.compile_threshold:
                return await self._execute_compiled(task)
            else:
                return await self._execute_with_evolution(task)
    
    async def _execute_safe(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """安全执行"""
        # 检查护栏
        if not await self.guardrails.pre_execution_check(task):
            return {"error": "执行被安全护栏阻止", "status": "blocked"}
        
        result = await self.evolver.execute_task(task)
        
        # 检查结果
        allowed = await self.guardrails.post_execution_check(
            result, 
            f"v{self.evolver.current_version}"
        )
        
        return {
            "result": result,
            "allowed": allowed,
            "guardrails_status": self.guardrails.get_protection_status()
        }
    
    async def _execute_with_evolution(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """带进化的执行"""
        result = await self.evolver.execute_task(task)
        
        if not result.success:
            # 触发自我改进
            patch = await self.evolver.self_improve(result)
            return {
                "result": result,
                "evolution": {
                    "patch_applied": patch is not None,
                    "patch_id": patch.patch_id if patch else None,
                    "new_version": self.evolver.current_version
                }
            }
        
        return {"result": result, "evolution": None}
    
    async def _execute_compiled(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """编译模型执行"""
        if not self.router:
            return {"error": "路由器未初始化", "status": "error"}
        
        result = await self.router.route(task, None)
        return {"result": result, "source": result.get("source")}
    
    def get_system_status(self) -> Dict[str, Any]:
        """获取系统状态"""
        return {
            "initialized": self.initialized,
            "evolution": {
                "version": self.evolver.current_version if self.evolver else 0,
                "patches_applied": len(self.evolver.patch_history) if self.evolver else 0,
                "failures_tracked": len(self.evolver.failure_history) if self.evolver else 0
            },
            "guardrails": self.guardrails.get_protection_status() if self.guardrails else {},
            "planner": {
                "patterns_learned": len(self.planner.transition_probs),
                "active_speculations": len(self.planner.speculations)
            },
            "security": self.lcguard.get_status_report()
        }
    
    def learn_pattern(self, trace: list):
        """学习执行模式（用于预测性规划）"""
        self.planner.learn_from_trace(trace)
    
    def get_predictions(self, current_tool: str):
        """获取预测"""
        return self.planner.predict_next(current_tool)


# 使用示例
async def main():
    config = SystemConfig(
        source_root="./modules",
        test_suite_path="./tests",
        base_benchmark=0.85,
        ratchet_threshold=0.95,
        compile_threshold=0.7
    )
    
    system = SelfEvolvingAgentSystem(config)
    
    # 模拟初始化
    class MockLLM:
        async def complete(self, prompt):
            return '{"root_cause": "分析完成", "affected_module": "executor"}'
    
    await system.initialize(MockLLM())
    
    # 学习模式
    system.learn_pattern(["fetch", "process", "output"])
    system.learn_pattern(["fetch", "validate", "output"])
    
    # 执行任务
    task = {
        "id": "task_001",
        "type": "data_processing",
        "message": "处理销售数据并生成报告"
    }
    
    result = await system.execute_task(task, mode="auto")
    print(f"执行结果: {result}")
    
    # 获取状态
    status = system.get_system_status()
    print(f"\n系统状态: {status}")


if __name__ == "__main__":
    asyncio.run(main())