Thinking Model Enhancer

{
  "owner": "xqicxx",
  "slug": "thinking-model-enhancer",
  "displayName": "thinking-model-enhancer",
  "latest": {
    "version": "1.0.0",
    "publishedAt": 1770119100995,
    "commit": "https://github.com/clawdbot/skills/commit/738fff72134934f03fd620c62a0dd7ff7fbf5d8a"
  },
  "history": []
}

# 思维模型增强器技能包
# Thinking Model Enhancer Skill Package

from .thinking_model_core import (
    ThinkingModelCore,
    ThinkingModel,
    ThinkingResult,
    ProblemAnalysis,
    ProblemType,
    ConfidenceLevel,
    UrgencyLevel,
    get_thinking_core
)

from .thinking_memory import (
    ThinkingMemory,
    ModelSnapshot,
    get_thinking_memory
)

from .thinking_interface import (
    ThinkingInterface,
    get_thinking_interface
)

from .initialize import (
    initialize_thinking_enhancer
)

from .advanced_features import (
    PerformanceTracker,
    PerformanceMetrics,
    ThinkingVisualizer,
    BatchProcessor,
    ModelConfigManager,
    get_performance_tracker,
    get_visualizer,
    get_batch_processor,
    get_config_manager
)

from .p2_enhancements import (
    TemplateLibrary,
    TriggerManager,
    CustomModelManager,
    TriggerType,
    TriggerMode,
    ThinkingTemplate,
    TriggerRule,
    CustomModelConfig,
    get_template_library,
    get_trigger_manager,
    get_custom_model_manager
)

from .p3_optimization import (
    ContinuousLearning,
    ModelCollaborator,
    AdvancedAnalyzer,
    LearningPattern,
    SessionData,
    ModelCollaboration,
    get_continuous_learning,
    get_model_collaborator,
    get_advanced_analyzer
)

__all__ = [
    # 核心模块
    'ThinkingModelCore',
    'ThinkingModel',
    'ThinkingResult',
    'ProblemAnalysis',
    'ProblemType',
    'ConfidenceLevel',
    'UrgencyLevel',
    'get_thinking_core',
    
    # 记忆模块
    'ThinkingMemory',
    'ModelSnapshot',
    'get_thinking_memory',
    
    # 接口模块
    'ThinkingInterface',
    'get_thinking_interface',
    
    # 初始化
    'initialize_thinking_enhancer',
    
    # 高级功能模块
    'PerformanceTracker',
    'PerformanceMetrics',
    'ThinkingVisualizer',
    'BatchProcessor',
    'ModelConfigManager',
    'get_performance_tracker',
    'get_visualizer',
    'get_batch_processor',
    'get_config_manager',
    
    # P2增强模块
    'TemplateLibrary',
    'TriggerManager',
    'CustomModelManager',
    'TriggerType',
    'TriggerMode',
    'ThinkingTemplate',
    'TriggerRule',
    'CustomModelConfig',
    'get_template_library',
    'get_trigger_manager',
    'get_custom_model_manager',
    
    # P3优化模块
    'ContinuousLearning',
    'ModelCollaborator',
    'AdvancedAnalyzer',
    'LearningPattern',
    'SessionData',
    'ModelCollaboration',
    'get_continuous_learning',
    'get_model_collaborator',
    'get_advanced_analyzer'
]

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
思维模型增强器 - 高级功能模块
Thinking Model Enhancer - Advanced Features Module

实现高级功能：性能追踪、思维可视化、批量处理、自定义配置
"""

import time
import json
from pathlib import Path
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any, Callable
from dataclasses import dataclass, field
from collections import defaultdict
import statistics


@dataclass
class PerformanceMetrics:
    """性能指标"""
    total_runs: int = 0
    successful_runs: int = 0
    failed_runs: int = 0
    total_duration_ms: float = 0
    avg_duration_ms: float = 0
    min_duration_ms: float = float('inf')
    max_duration_ms: float = 0
    by_model: Dict[str, Dict] = field(default_factory=dict)
    by_problem_type: Dict[str, Dict] = field(default_factory=dict)
    recent_runs: List[Dict] = field(default_factory=list)
    
    def to_dict(self) -> Dict:
        return {
            "total_runs": self.total_runs,
            "successful_runs": self.successful_runs,
            "failed_runs": self.failed_runs,
            "success_rate": self.successful_runs / self.total_runs if self.total_runs > 0 else 0,
            "total_duration_ms": self.total_duration_ms,
            "avg_duration_ms": self.avg_duration_ms,
            "min_duration_ms": self.min_duration_ms if self.min_duration_ms != float('inf') else 0,
            "max_duration_ms": self.max_duration_ms,
            "by_model": self.by_model,
            "by_problem_type": self.by_problem_type,
            "recent_runs": self.recent_runs[-50:]  # 最近50条
        }


class PerformanceTracker:
    """性能追踪器"""
    
    def __init__(self, storage_dir: Optional[str] = None):
        """初始化性能追踪器"""
        if storage_dir is None:
            self.storage_dir = Path.home() / ".claude" / "thinking_models" / "performance"
        else:
            self.storage_dir = Path(storage_dir)
        
        self.storage_dir.mkdir(parents=True, exist_ok=True)
        self.metrics_file = self.storage_dir / "metrics.json"
        
        # 加载现有指标
        self.metrics = self._load_metrics()
        
        # 实时追踪
        self.current_run_start: Optional[float] = None
        self.current_run_data: Dict = {}
    
    def _load_metrics(self) -> PerformanceMetrics:
        """加载性能指标"""
        if self.metrics_file.exists():
            with open(self.metrics_file, 'r', encoding='utf-8') as f:
                data = json.load(f)
                metrics = PerformanceMetrics()
                metrics.total_runs = data.get('total_runs', 0)
                metrics.successful_runs = data.get('successful_runs', 0)
                metrics.failed_runs = data.get('failed_runs', 0)
                metrics.total_duration_ms = data.get('total_duration_ms', 0)
                metrics.avg_duration_ms = data.get('avg_duration_ms', 0)
                metrics.min_duration_ms = data.get('min_duration_ms', 0)
                metrics.max_duration_ms = data.get('max_duration_ms', 0)
                metrics.by_model = data.get('by_model', {})
                metrics.by_problem_type = data.get('by_problem_type', {})
                metrics.recent_runs = data.get('recent_runs', [])
                return metrics
        return PerformanceMetrics()
    
    def _save_metrics(self):
        """保存性能指标"""
        with open(self.metrics_file, 'w', encoding='utf-8') as f:
            json.dump(self.metrics.to_dict(), f, ensure_ascii=False, indent=2)
    
    def start_run(self, model: str, problem_type: str, problem: str):
        """开始一次运行追踪"""
        self.current_run_start = time.time()
        self.current_run_data = {
            "model": model,
            "problem_type": problem_type,
            "problem": problem[:100],
            "start_time": datetime.now().isoformat(),
            "stages": []
        }
    
    def log_stage(self, stage_name: str, duration_ms: float):
        """记录阶段信息"""
        if self.current_run_data:
            self.current_run_data["stages"].append({
                "stage": stage_name,
                "duration_ms": duration_ms
            })
    
    def end_run(self, success: bool, result_summary: str = ""):
        """结束运行追踪"""
        if self.current_run_start is None:
            return
        
        duration_ms = (time.time() - self.current_run_start) * 1000
        
        # 更新总体指标
        self.metrics.total_runs += 1
        if success:
            self.metrics.successful_runs += 1
        else:
            self.metrics.failed_runs += 1
        
        self.metrics.total_duration_ms += duration_ms
        self.metrics.avg_duration_ms = self.metrics.total_duration_ms / self.metrics.total_runs
        self.metrics.min_duration_ms = min(self.metrics.min_duration_ms, duration_ms)
        self.metrics.max_duration_ms = max(self.metrics.max_duration_ms, duration_ms)
        
        # 更新模型指标
        model = self.current_run_data.get("model", "unknown")
        if model not in self.metrics.by_model:
            self.metrics.by_model[model] = {
                "runs": 0, "success": 0, "total_duration": 0
            }
        self.metrics.by_model[model]["runs"] += 1
        if success:
            self.metrics.by_model[model]["success"] += 1
        self.metrics.by_model[model]["total_duration"] += duration_ms
        
        # 更新问题类型指标
        problem_type = self.current_run_data.get("problem_type", "unknown")
        if problem_type not in self.metrics.by_problem_type:
            self.metrics.by_problem_type[problem_type] = {
                "runs": 0, "success": 0
            }
        self.metrics.by_problem_type[problem_type]["runs"] += 1
        if success:
            self.metrics.by_problem_type[problem_type]["success"] += 1
        
        # 添加到最近记录
        self.metrics.recent_runs.append({
            "timestamp": datetime.now().isoformat(),
            "model": model,
            "problem_type": problem_type,
            "duration_ms": duration_ms,
            "success": success,
            "stages": len(self.current_run_data.get("stages", [])),
            "summary": result_summary[:50]
        })
        
        # 保存并重置
        self._save_metrics()
        self.current_run_start = None
        self.current_run_data = {}
    
    def get_metrics(self) -> PerformanceMetrics:
        """获取性能指标"""
        return self.metrics
    
    def get_model_performance(self, model: str) -> Dict:
        """获取特定模型的性能"""
        return self.metrics.by_model.get(model, {
            "runs": 0, "success": 0, "total_duration": 0
        })
    
    def get_summary_report(self) -> str:
        """生成性能摘要报告"""
        m = self.metrics
        success_rate = m.successful_runs / m.total_runs if m.total_runs > 0 else 0
        
        lines = [
            "📊 思维模型性能报告",
            "=" * 50,
            "",
            f"📈 总体统计:",
            f"   总运行次数: {m.total_runs}",
            f"   成功: {m.successful_runs} | 失败: {m.failed_runs}",
            f"   成功率: {success_rate*100:.1f}%",
            f"   平均耗时: {m.avg_duration_ms:.1f}ms",
            f"   最短耗时: {m.min_duration_ms:.1f}ms" if m.min_duration_ms != float('inf') else "",
            f"   最长耗时: {m.max_duration_ms:.1f}ms",
        ]
        
        if m.by_model:
            lines.extend(["", f"📂 按模型:"])
            for model, data in sorted(m.by_model.items(), key=lambda x: x[1]["runs"], reverse=True):
                model_success = data["success"] / data["runs"] if data["runs"] > 0 else 0
                lines.append(f"   • {model}: {data['runs']}次 (成功率{ model_success*100:.0f}%)")
        
        if m.by_problem_type:
            lines.extend(["", f"🏷️ 按问题类型:"])
            for ptype, data in sorted(m.by_problem_type.items(), key=lambda x: x[1]["runs"], reverse=True):
                lines.append(f"   • {ptype}: {data['runs']}次")
        
        return "\n".join(filter(None, lines))


class ThinkingVisualizer:
    """思维可视化器"""
    
    def __init__(self):
        """初始化可视化器"""
        pass
    
    def format_thinking_process(self, 
                                 problem: str,
                                 analysis: Dict,
                                 model: str,
                                 stages: List[Dict],
                                 result: Dict) -> str:
        """
        格式化思维过程为可读报告
        
        Args:
            problem: 问题描述
            analysis: 问题分析结果
            model: 使用的模型
            stages: 各阶段处理结果
            result: 最终结果
            
        Returns:
            格式化的报告字符串
        """
        lines = [
            f"🧠 思维过程报告",
            "=" * 60,
            "",
            f"📝 问题: {problem[:100]}{'...' if len(problem) > 100 else ''}",
            "",
            f"🔍 问题分析:",
            f"   类型: {analysis.get('type', '未知')}",
            f"   复杂度: {analysis.get('complexity', 'N/A')}/10",
            f"   置信度: {analysis.get('confidence', 'N/A')}",
            "",
            f"🎯 使用模型: {model}",
            "",
            f"🔄 处理过程 ({len(stages)} 阶段):"
        ]
        
        for i, stage in enumerate(stages, 1):
            lines.append(f"   {i}. {stage.get('name', 'Unknown')}")
            if stage.get("findings"):
                for finding in stage["findings"][:3]:
                    lines.append(f"      • {finding}")
            if stage.get("duration_ms"):
                lines.append(f"      ⏱️ {stage['duration_ms']:.1f}ms")
        
        lines.extend([
            "",
            f"✅ 处理结果:",
            f"   置信度: {result.get('confidence', 'N/A')}",
            f"   耗时: {result.get('duration_ms', 0):.1f}ms",
        ])
        
        if result.get("recommendations"):
            lines.extend([
                "",
                f"💡 建议:",
                *[f"   • {rec}" for rec in result["recommendations"][:5]]
            ])
        
        return "\n".join(lines)
    
    def generate_text_chart(self, 
                            data: List[Dict], 
                            value_key: str = "success_rate",
                            label_key: str = "label") -> str:
        """
        生成简单的文本图表
        
        Args:
            data: 数据列表
            value_key: 数值字段名
            label_key: 标签字段名
            
        Returns:
            文本图表字符串
        """
        if not data:
            return "暂无数据 📊"
        
        max_value = max(d.get(value_key, 0) for d in data)
        max_label_len = max(len(d.get(label_key, "")) for d in data)
        
        lines = []
        for item in data:
            label = item.get(label_key, "")[:20]
            value = item.get(value_key, 0)
            
            # 计算进度条长度
            bar_length = int((value / max_value) * 20) if max_value > 0 else 0
            bar = "█" * bar_length + "░" * (20 - bar_length)
            
            # 格式化行
            if isinstance(value, float):
                if value > 1:
                    line = f"{label:<{max_label_len}} {bar} {value:.0f}"
                else:
                    line = f"{label:<{max_label_len}} {bar} {value*100:.1f}%"
            else:
                line = f"{label:<{max_label_len}} {bar} {value}"
            
            lines.append(line)
        
        return "\n".join(lines)


class BatchProcessor:
    """批量处理器"""
    
    def __init__(self, 
                 processor_func: Callable,
                 max_concurrent: int = 3,
                 timeout_seconds: int = 60):
        """
        初始化批量处理器
        
        Args:
            processor_func: 处理函数
            max_concurrent: 最大并发数
            timeout_seconds: 超时时间
        """
        self.processor = processor_func
        self.max_concurrent = max_concurrent
        self.timeout = timeout_seconds
        
        self.results: List[Dict] = []
        self.stats = {
            "total": 0,
            "completed": 0,
            "failed": 0,
            "start_time": None,
            "end_time": None
        }
    
    def process(self, items: List[Dict]) -> List[Dict]:
        """
        批量处理项目
        
        Args:
            items: 项目列表，每个项目包含id和data
            
        Returns:
            处理结果列表
        """
        self.stats["total"] = len(items)
        self.stats["start_time"] = datetime.now().isoformat()
        
        results = []
        for item in items:
            try:
                result = self.processor(item["data"])
                results.append({
                    "id": item.get("id", ""),
                    "status": "success",
                    "result": result
                })
                self.stats["completed"] += 1
            except Exception as e:
                results.append({
                    "id": item.get("id", ""),
                    "status": "failed",
                    "error": str(e)
                })
                self.stats["failed"] += 1
        
        self.stats["end_time"] = datetime.now().isoformat()
        self.results = results
        
        return results
    
    def get_stats(self) -> Dict:
        """获取处理统计"""
        duration = None
        if self.stats["start_time"] and self.stats["end_time"]:
            start = datetime.fromisoformat(self.stats["start_time"])
            end = datetime.fromisoformat(self.stats["end_time"])
            duration = (end - start).total_seconds()
        
        return {
            **self.stats,
            "duration_seconds": duration,
            "success_rate": self.stats["completed"] / self.stats["total"] if self.stats["total"] > 0 else 0
        }


class ModelConfigManager:
    """模型配置管理器"""
    
    def __init__(self, config_dir: Optional[str] = None):
        """初始化配置管理器"""
        if config_dir is None:
            self.config_dir = Path.home() / ".claude" / "thinking_models" / "config"
        else:
            self.config_dir = Path(config_dir)
        
        self.config_dir.mkdir(parents=True, exist_ok=True)
        self.config_file = self.config_dir / "models.json"
        
        # 加载或创建默认配置
        self.config = self._load_or_create_config()
    
    def _load_or_create_config(self) -> Dict:
        """加载或创建默认配置"""
        if self.config_file.exists():
            with open(self.config_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        
        # 默认配置
        default_config = {
            "version": "1.0",
            "updated_at": datetime.now().isoformat(),
            "models": {
                "research_mode": {
                    "enabled": True,
                    "timeout_seconds": 120,
                    "max_retries": 3,
                    "priority": 1
                },
                "diagnostic_mode": {
                    "enabled": True,
                    "timeout_seconds": 90,
                    "max_retries": 2,
                    "priority": 2
                },
                "generic_pipeline": {
                    "enabled": True,
                    "timeout_seconds": 60,
                    "max_retries": 3,
                    "priority": 10
                }
            },
            "general": {
                "auto_store_results": True,
                "confidence_threshold": 0.6,
                "enable_visualization": True,
                "max_history_items": 100
            }
        }
        
        self._save_config(default_config)
        return default_config
    
    def _save_config(self, config: Dict):
        """保存配置"""
        config["updated_at"] = datetime.now().isoformat()
        with open(self.config_file, 'w', encoding='utf-8') as f:
            json.dump(config, f, ensure_ascii=False, indent=2)
    
    def get_model_config(self, model_name: str) -> Optional[Dict]:
        """获取模型配置"""
        return self.config.get("models", {}).get(model_name)
    
    def set_model_config(self, model_name: str, settings: Dict):
        """设置模型配置"""
        if model_name not in self.config["models"]:
            self.config["models"][model_name] = {}
        
        self.config["models"][model_name].update(settings)
        self._save_config(self.config)
    
    def get_general_config(self) -> Dict:
        """获取通用配置"""
        return self.config.get("general", {})
    
    def set_general_config(self, settings: Dict):
        """设置通用配置"""
        self.config["general"].update(settings)
        self._save_config(self.config)
    
    def is_model_enabled(self, model_name: str) -> bool:
        """检查模型是否启用"""
        model_config = self.get_model_config(model_name)
        return model_config.get("enabled", True) if model_config else True


def get_performance_tracker() -> PerformanceTracker:
    """获取性能追踪器实例"""
    return PerformanceTracker()


def get_visualizer() -> ThinkingVisualizer:
    """获取可视化器实例"""
    return ThinkingVisualizer()


def get_batch_processor(processor_func: Callable) -> BatchProcessor:
    """获取批量处理器实例"""
    return BatchProcessor(processor_func)


def get_config_manager() -> ModelConfigManager:
    """获取配置管理器实例"""
    return ModelConfigManager()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
思维模型增强器 - 初始化脚本
Thinking Model Enhancer - Initialization Script

初始化记忆目录和配置
"""

import os
from pathlib import Path
from datetime import datetime


def initialize_thinking_enhancer():
    """初始化思维模型增强器"""
    base_dir = Path.home() / ".claude" / "thinking_models"
    
    # 创建目录结构
    directories = [
        base_dir,  # 根目录
        base_dir / "memory",  # 记忆存储
        base_dir / "cache",  # 缓存
    ]
    
    for directory in directories:
        directory.mkdir(parents=True, exist_ok=True)
        print(f"✅ 创建目录: {directory}")
    
    # 创建示例配置文件
    config_file = base_dir / "config.json"
    if not config_file.exists():
        config = {
            "version": "1.0",
            "created_at": datetime.now().isoformat(),
            "settings": {
                "auto_store_results": True,
                "max_history_days": 90,
                "default_model": "generic_pipeline",
                "confidence_threshold": 0.6
            },
            "enabled_models": [
                "research_mode",
                "diagnostic_mode",
                "generic_pipeline"
            ]
        }
        
        with open(config_file, 'w', encoding='utf-8') as f:
            import json
            json.dump(config, f, ensure_ascii=False, indent=2)
        
        print(f"✅ 创建配置文件: {config_file}")
    
    print(f"\n🎉 思维模型增强器初始化完成！")
    print(f"📁 主目录: {base_dir}")
    
    return str(base_dir)


if __name__ == "__main__":
    initialize_thinking_enhancer()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
思维模型增强器 - P2增强功能模块
Thinking Model Enhancer - P2 Enhancement Module

实现P2增强功能：
- 思维模型模板库
- 自定义模型配置
- 自动化触发机制
"""

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


class TriggerType(Enum):
    """触发类型"""
    KEYWORD = "keyword"           # 关键词触发
    PATTERN = "pattern"           # 正则模式触发
    SCHEDULE = "schedule"         # 定时触发
    CONTEXT = "context"           # 上下文触发
    API = "api"                   # API调用触发


class TriggerMode(Enum):
    """触发模式"""
    MANUAL = "manual"             # 手动确认
    AUTO = "auto"                 # 自动执行
    SUGGEST = "suggest"           # 建议模式


@dataclass
class ThinkingTemplate:
    """思维模型模板"""
    template_id: str
    name: str
    description: str
    category: str  # 创建/修复/分析/决策/研究
    stages: List[Dict]  # 阶段定义
    conditions: List[str]  # 适用条件
    examples: List[str]  # 使用示例
    version: str = "1.0"
    author: str = "system"
    created_at: str = ""
    use_count: int = 0
    rating: float = 0.0
    tags: List[str] = field(default_factory=list)
    custom_fields: Dict = field(default_factory=dict)
    
    def to_dict(self) -> Dict:
        return {
            "template_id": self.template_id,
            "name": self.name,
            "description": self.description,
            "category": self.category,
            "stages": self.stages,
            "conditions": self.conditions,
            "examples": self.examples,
            "version": self.version,
            "author": self.author,
            "created_at": self.created_at or datetime.now().isoformat(),
            "use_count": self.use_count,
            "rating": self.rating,
            "tags": self.tags,
            "custom_fields": self.custom_fields
        }


@dataclass
class TriggerRule:
    """触发规则"""
    rule_id: str
    name: str
    trigger_type: TriggerType
    trigger_value: str  # 关键词/模式/时间/上下文
    mode: TriggerMode
    target_model: str  # 目标思维模型
    priority: int  # 优先级
    enabled: bool = True
    cooldown_seconds: int = 300  # 冷却时间
    last_triggered: Optional[str] = None
    conditions: List[str] = field(default_factory=list)
    actions: List[str] = field(default_factory=list)
    
    def to_dict(self) -> Dict:
        return {
            "rule_id": self.rule_id,
            "name": self.name,
            "trigger_type": self.trigger_type.value,
            "trigger_value": self.trigger_value,
            "mode": self.mode.value,
            "target_model": self.target_model,
            "priority": self.priority,
            "enabled": self.enabled,
            "cooldown_seconds": self.cooldown_seconds,
            "last_triggered": self.last_triggered,
            "conditions": self.conditions,
            "actions": self.actions
        }


@dataclass
class CustomModelConfig:
    """自定义模型配置"""
    model_id: str
    name: str
    base_model: str  # 基于哪个基础模型
    modifications: Dict  # 修改内容
    created_at: str = ""
    use_count: int = 0
    performance_notes: List[str] = field(default_factory=list)
    
    def to_dict(self) -> Dict:
        return {
            "model_id": self.model_id,
            "name": self.name,
            "base_model": self.base_model,
            "modifications": self.modifications,
            "created_at": self.created_at or datetime.now().isoformat(),
            "use_count": self.use_count,
            "performance_notes": self.performance_notes
        }


class TemplateLibrary:
    """思维模型模板库"""
    
    def __init__(self, library_dir: Optional[str] = None):
        """初始化模板库"""
        if library_dir is None:
            self.library_dir = Path.home() / ".claude" / "thinking_models" / "templates"
        else:
            self.library_dir = Path(library_dir)
        
        self.library_dir.mkdir(parents=True, exist_ok=True)
        self.templates_file = self.library_dir / "templates.json"
        
        # 加载或创建模板库
        self.templates = self._load_or_create_templates()
    
    def _load_or_create_templates(self) -> Dict:
        """加载或创建默认模板"""
        if self.templates_file.exists():
            with open(self.templates_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        
        # 默认模板
        default_templates = {
            "version": "1.0",
            "updated_at": datetime.now().isoformat(),
            "templates": [
                {
                    "template_id": "quick_decision",
                    "name": "快速决策模板",
                    "description": "用于需要快速做出决策的场景，简化流程",
                    "category": "decision",
                    "stages": [
                        {"name": "问题定义", "order": 1, "required": True},
                        {"name": "选项列举", "order": 2, "required": True},
                        {"name": "利弊分析", "order": 3, "required": True},
                        {"name": "最终决策", "order": 4, "required": True}
                    ],
                    "conditions": ["时间紧迫", "信息有限", "需要快速响应"],
                    "examples": ["快速决定是否采纳建议", "快速选择方案"],
                    "tags": ["快速", "决策", "简化"]
                },
                {
                    "template_id": "deep_analysis",
                    "name": "深度分析模板",
                    "description": "用于需要全面深入分析的场景",
                    "category": "analysis",
                    "stages": [
                        {"name": "问题分解", "order": 1, "required": True},
                        {"name": "数据收集", "order": 2, "required": True},
                        {"name": "多角度分析", "order": 3, "required": True},
                        {"name": "假设验证", "order": 4, "required": True},
                        {"name": "综合结论", "order": 5, "required": True},
                        {"name": "建议制定", "order": 6, "required": False}
                    ],
                    "conditions": ["复杂问题", "需要全面考虑", "时间充裕"],
                    "examples": ["深度分析系统架构", "全面评估方案"],
                    "tags": ["深度", "分析", "全面"]
                },
                {
                    "template_id": "creative_solving",
                    "name": "创意解决问题模板",
                    "description": "用于需要创新思维的问题解决",
                    "category": "solving",
                    "stages": [
                        {"name": "问题重构", "order": 1, "required": True},
                        {"name": "头脑风暴", "order": 2, "required": True},
                        {"name": "方案筛选", "order": 3, "required": True},
                        {"name": "可行性分析", "order": 4, "required": True},
                        {"name": "实施计划", "order": 5, "required": False}
                    ],
                    "conditions": ["常规方法无效", "需要创新", "开放性问题"],
                    "examples": ["创新解决方案", "突破性想法"],
                    "tags": ["创意", "创新", "解决问题"]
                }
            ]
        }
        
        self._save_templates(default_templates)
        return default_templates
    
    def _save_templates(self, templates: Dict):
        """保存模板库"""
        templates["updated_at"] = datetime.now().isoformat()
        with open(self.templates_file, 'w', encoding='utf-8') as f:
            json.dump(templates, f, ensure_ascii=False, indent=2)
    
    def get_template(self, template_id: str) -> Optional[Dict]:
        """获取模板"""
        for template in self.templates.get("templates", []):
            if template.get("template_id") == template_id:
                return template
        return None
    
    def search_templates(self, 
                         query: str = "", 
                         category: Optional[str] = None,
                         tags: Optional[List[str]] = None,
                         limit: int = 10) -> List[Dict]:
        """
        搜索模板
        
        Args:
            query: 搜索关键词
            category: 分类过滤
            tags: 标签过滤
            limit: 返回数量限制
            
        Returns:
            匹配的模板列表
        """
        results = []
        query_lower = query.lower()
        
        for template in self.templates.get("templates", []):
            # 关键词匹配
            if query_lower:
                if not any(query_lower in str(v).lower() 
                          for v in template.values() if isinstance(v, str)):
                    continue
            
            # 分类过滤
            if category and template.get("category") != category:
                continue
            
            # 标签过滤
            if tags:
                template_tags = template.get("tags", [])
                if not any(tag in template_tags for tag in tags):
                    continue
            
            results.append(template)
        
        return results[:limit]
    
    def add_template(self, template: Dict) -> bool:
        """添加新模板"""
        try:
            template["template_id"] = template.get("template_id") or f"custom_{datetime.now().strftime('%Y%m%d%H%M%S')}"
            template["created_at"] = datetime.now().isoformat()
            template["use_count"] = 0
            template["rating"] = 0.0
            
            self.templates["templates"].append(template)
            self._save_templates(self.templates)
            return True
        except Exception:
            return False
    
    def update_template(self, template_id: str, updates: Dict) -> bool:
        """更新模板"""
        for i, template in enumerate(self.templates.get("templates", [])):
            if template.get("template_id") == template_id:
                template.update(updates)
                self._save_templates(self.templates)
                return True
        return False
    
    def delete_template(self, template_id: str) -> bool:
        """删除模板"""
        templates = self.templates.get("templates", [])
        for i, template in enumerate(templates):
            if template.get("template_id") == template_id:
                templates.pop(i)
                self.templates["templates"] = templates
                self._save_templates(self.templates)
                return True
        return False
    
    def get_categories(self) -> List[str]:
        """获取所有分类"""
        categories = set()
        for template in self.templates.get("templates", []):
            if template.get("category"):
                categories.add(template["category"])
        return list(categories)
    
    def increment_use_count(self, template_id: str):
        """增加使用计数"""
        for template in self.templates.get("templates", []):
            if template.get("template_id") == template_id:
                template["use_count"] = template.get("use_count", 0) + 1
                self._save_templates(self.templates)
                break
    
    def get_popular_templates(self, limit: int = 5) -> List[Dict]:
        """获取最受欢迎的模板"""
        templates = self.templates.get("templates", [])
        return sorted(templates, key=lambda x: x.get("use_count", 0), reverse=True)[:limit]


class TriggerManager:
    """触发器管理器"""
    
    def __init__(self, rules_dir: Optional[str] = None):
        """初始化触发器管理器"""
        if rules_dir is None:
            self.rules_dir = Path.home() / ".claude" / "thinking_models" / "triggers"
        else:
            self.rules_dir = Path(rules_dir)
        
        self.rules_dir.mkdir(parents=True, exist_ok=True)
        self.rules_file = self.rules_dir / "rules.json"
        
        # 加载或创建默认规则
        self.rules = self._load_or_create_rules()
    
    def _load_or_create_rules(self) -> Dict:
        """加载或创建默认规则"""
        if self.rules_file.exists():
            with open(self.rules_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        
        # 默认规则
        default_rules = {
            "version": "1.0",
            "updated_at": datetime.now().isoformat(),
            "rules": [
                {
                    "rule_id": "skill_creation_trigger",
                    "name": "技能创建触发器",
                    "trigger_type": "keyword",
                    "trigger_value": "写skill|创建技能|实现功能",
                    "mode": "auto",
                    "target_model": "research_mode",
                    "priority": 1,
                    "enabled": True,
                    "cooldown_seconds": 60,
                    "conditions": [],
                    "actions": ["启动研究型思维模式", "准备技能创建模板"]
                },
                {
                    "rule_id": "system_repair_trigger",
                    "name": "系统修复触发器",
                    "trigger_type": "keyword",
                    "trigger_value": "修复|报错|错误|问题|启动失败",
                    "mode": "auto",
                    "target_model": "diagnostic_mode",
                    "priority": 1,
                    "enabled": True,
                    "cooldown_seconds": 30,
                    "conditions": [],
                    "actions": ["启动诊断型思维模式", "准备修复工具"]
                }
            ]
        }
        
        self._save_rules(default_rules)
        return default_rules
    
    def _save_rules(self, rules: Dict):
        """保存规则"""
        rules["updated_at"] = datetime.now().isoformat()
        with open(self.rules_file, 'w', encoding='utf-8') as f:
            json.dump(rules, f, ensure_ascii=False, indent=2)
    
    def check_triggers(self, 
                       input_text: str, 
                       context: Optional[Dict] = None) -> List[Dict]:
        """
        检查输入是否触发规则
        
        Args:
            input_text: 输入文本
            context: 上下文信息
            
        Returns:
            匹配的触发规则列表
        """
        matches = []
        now = datetime.now().isoformat()
        
        for rule in self.rules.get("rules", []):
            if not rule.get("enabled", True):
                continue
            
            # 检查冷却时间
            last_triggered = rule.get("last_triggered")
            if last_triggered:
                last_time = datetime.fromisoformat(last_triggered)
                cooldown = rule.get("cooldown_seconds", 300)
                if (datetime.now() - last_time).total_seconds() < cooldown:
                    continue
            
            # 检查触发条件
            trigger_type = rule.get("trigger_type")
            trigger_value = rule.get("trigger_value")
            
            if trigger_type == "keyword":
                if trigger_value and any(kw in input_text for kw in trigger_value.split("|")):
                    matches.append(rule)
            
            elif trigger_type == "pattern":
                try:
                    if re.search(trigger_value, input_text):
                        matches.append(rule)
                except Exception:
                    continue
        
        # 按优先级排序
        matches.sort(key=lambda x: x.get("priority", 100))
        
        # 更新最后触发时间
        for rule in matches:
            rule["last_triggered"] = now
        
        if matches:
            self._save_rules(self.rules)
        
        return matches
    
    def add_rule(self, rule: Dict) -> bool:
        """添加规则"""
        try:
            rule["rule_id"] = rule.get("rule_id") or f"rule_{datetime.now().strftime('%Y%m%d%H%M%S')}"
            self.rules["rules"].append(rule)
            self._save_rules(self.rules)
            return True
        except Exception:
            return False
    
    def update_rule(self, rule_id: str, updates: Dict) -> bool:
        """更新规则"""
        for i, rule in enumerate(self.rules.get("rules", [])):
            if rule.get("rule_id") == rule_id:
                rule.update(updates)
                self._save_rules(self.rules)
                return True
        return False
    
    def delete_rule(self, rule_id: str) -> bool:
        """删除规则"""
        rules = self.rules.get("rules", [])
        for i, rule in enumerate(rules):
            if rule.get("rule_id") == rule_id:
                rules.pop(i)
                self.rules["rules"] = rules
                self._save_rules(self.rules)
                return True
        return False
    
    def get_enabled_rules(self) -> List[Dict]:
        """获取所有启用的规则"""
        return [r for r in self.rules.get("rules", []) if r.get("enabled", True)]
    
    def disable_rule(self, rule_id: str) -> bool:
        """禁用规则"""
        return self.update_rule(rule_id, {"enabled": False})
    
    def enable_rule(self, rule_id: str) -> bool:
        """启用规则"""
        return self.update_rule(rule_id, {"enabled": True})


class CustomModelManager:
    """自定义模型管理器"""
    
    def __init__(self, models_dir: Optional[str] = None):
        """初始化自定义模型管理器"""
        if models_dir is None:
            self.models_dir = Path.home() / ".claude" / "thinking_models" / "custom_models"
        else:
            self.models_dir = Path(models_dir)
        
        self.models_dir.mkdir(parents=True, exist_ok=True)
        self.models_file = self.models_dir / "models.json"
        
        # 加载或创建模型配置
        self.models = self._load_or_create_models()
    
    def _load_or_create_models(self) -> Dict:
        """加载或创建模型配置"""
        if self.models_file.exists():
            with open(self.models_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        
        # 默认配置
        default_models = {
            "version": "1.0",
            "updated_at": datetime.now().isoformat(),
            "models": []
        }
        
        self._save_models(default_models)
        return default_models
    
    def _save_models(self, models: Dict):
        """保存模型配置"""
        models["updated_at"] = datetime.now().isoformat()
        with open(self.models_file, 'w', encoding='utf-8') as f:
            json.dump(models, f, ensure_ascii=False, indent=2)
    
    def create_model(self, 
                     name: str, 
                     base_model: str,
                     modifications: Dict) -> Optional[str]:
        """
        创建自定义模型
        
        Args:
            name: 模型名称
            base_model: 基础模型
            modifications: 修改内容
            
        Returns:
            模型ID或None
        """
        try:
            model_id = f"custom_{hashlib.md5((name + datetime.now().isoformat()).encode()).hexdigest()[:8]}"
            
            model = {
                "model_id": model_id,
                "name": name,
                "base_model": base_model,
                "modifications": modifications,
                "created_at": datetime.now().isoformat(),
                "use_count": 0,
                "performance_notes": []
            }
            
            self.models["models"].append(model)
            self._save_models(self.models)
            
            return model_id
        except Exception:
            return None
    
    def get_model(self, model_id: str) -> Optional[Dict]:
        """获取自定义模型"""
        for model in self.models.get("models", []):
            if model.get("model_id") == model_id:
                return model
        return None
    
    def get_models_by_base(self, base_model: str) -> List[Dict]:
        """获取基于特定基础模型的所有自定义模型"""
        return [m for m in self.models.get("models", []) if m.get("base_model") == base_model]
    
    def update_model(self, model_id: str, updates: Dict) -> bool:
        """更新自定义模型"""
        for i, model in enumerate(self.models.get("models", [])):
            if model.get("model_id") == model_id:
                model.update(updates)
                self._save_models(self.models)
                return True
        return False
    
    def delete_model(self, model_id: str) -> bool:
        """删除自定义模型"""
        models = self.models.get("models", [])
        for i, model in enumerate(models):
            if model.get("model_id") == model_id:
                models.pop(i)
                self.models["models"] = models
                self._save_models(self.models)
                return True
        return False
    
    def increment_use_count(self, model_id: str):
        """增加使用计数"""
        for model in self.models.get("models", []):
            if model.get("model_id") == model_id:
                model["use_count"] = model.get("use_count", 0) + 1
                self._save_models(self.models)
                break
    
    def add_performance_note(self, model_id: str, note: str):
        """添加性能备注"""
        for model in self.models.get("models", []):
            if model.get("model_id") == model_id:
                if "performance_notes" not in model:
                    model["performance_notes"] = []
                model["performance_notes"].append({
                    "timestamp": datetime.now().isoformat(),
                    "note": note
                })
                self._save_models(self.models)
                break


def get_template_library() -> TemplateLibrary:
    """获取模板库实例"""
    return TemplateLibrary()


def get_trigger_manager() -> TriggerManager:
    """获取触发器管理器实例"""
    return TriggerManager()


def get_custom_model_manager() -> CustomModelManager:
    """获取自定义模型管理器实例"""
    return CustomModelManager()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
思维模型增强器 - P3优化功能模块
Thinking Model Enhancer - P3 Optimization Module

实现P3优化功能：
- 跨会话持续学习
- 多模型协同
- 高级分析功能
- 智能优化建议
"""

import json
import hashlib
from pathlib import Path
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any, Tuple
from dataclasses import dataclass, field
from collections import defaultdict
import statistics


@dataclass
class LearningPattern:
    """学习模式"""
    pattern_id: str
    pattern_type: str  # successful_pattern, failed_pattern, improvement_area
    description: str
    conditions: List[str]
    actions: List[str]
    success_rate_impact: float  # 对成功率的影响
    occurrence_count: int
    last_observed: str
    confidence: float  # 0-1
    source_sessions: List[str]
    
    def to_dict(self) -> Dict:
        return {
            "pattern_id": self.pattern_id,
            "pattern_type": self.pattern_type,
            "description": self.description,
            "conditions": self.conditions,
            "actions": self.actions,
            "success_rate_impact": self.success_rate_impact,
            "occurrence_count": self.occurrence_count,
            "last_observed": self.last_observed,
            "confidence": self.confidence,
            "source_sessions": self.source_sessions
        }


@dataclass
class SessionData:
    """会话数据"""
    session_id: str
    start_time: str
    end_time: Optional[str]
    user_input: str
    selected_model: str
    stages_completed: List[str]
    result_summary: str
    success: bool
    user_rating: Optional[int]
    duration_ms: float
    insights: List[str] = field(default_factory=list)
    improvements: List[str] = field(default_factory=list)
    
    def to_dict(self) -> Dict:
        return {
            "session_id": self.session_id,
            "start_time": self.start_time,
            "end_time": self.end_time,
            "user_input": self.user_input,
            "selected_model": self.selected_model,
            "stages_completed": self.stages_completed,
            "result_summary": self.result_summary,
            "success": self.success,
            "user_rating": self.user_rating,
            "duration_ms": self.duration_ms,
            "insights": self.insights,
            "improvements": self.improvements
        }


@dataclass
class ModelCollaboration:
    """模型协作配置"""
    collaboration_id: str
    name: str
    description: str
    models: List[str]  # 参与的模型
    workflow: List[Dict]  # 工作流程定义
    conditions: List[str]  # 适用条件
    performance_metrics: Dict = field(default_factory=dict)
    use_count: int = 0
    
    def to_dict(self) -> Dict:
        return {
            "collaboration_id": self.collaboration_id,
            "name": self.name,
            "description": self.description,
            "models": self.models,
            "workflow": self.workflow,
            "conditions": self.conditions,
            "performance_metrics": self.performance_metrics,
            "use_count": self.use_count
        }


class ContinuousLearning:
    """跨会话持续学习系统"""
    
    def __init__(self, learning_dir: Optional[str] = None):
        """初始化持续学习系统"""
        if learning_dir is None:
            self.learning_dir = Path.home() / ".claude" / "thinking_models" / "learning"
        else:
            self.learning_dir = Path(learning_dir)
        
        self.learning_dir.mkdir(parents=True, exist_ok=True)
        
        # 数据文件
        self.patterns_file = self.learning_dir / "patterns.json"
        self.sessions_file = self.learning_dir / "sessions.json"
        
        # 加载数据
        self.patterns = self._load_patterns()
        self.sessions = self._load_sessions()
    
    def _load_patterns(self) -> Dict:
        """加载学习模式"""
        if self.patterns_file.exists():
            with open(self.patterns_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        return {
            "version": "1.0",
            "updated_at": datetime.now().isoformat(),
            "patterns": []
        }
    
    def _load_sessions(self) -> Dict:
        """加载会话数据"""
        if self.sessions_file.exists():
            with open(self.sessions_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        return {
            "version": "1.0",
            "updated_at": datetime.now().isoformat(),
            "sessions": []
        }
    
    def _save_patterns(self):
        """保存学习模式"""
        self.patterns["updated_at"] = datetime.now().isoformat()
        with open(self.patterns_file, 'w', encoding='utf-8') as f:
            json.dump(self.patterns, f, ensure_ascii=False, indent=2)
    
    def _save_sessions(self):
        """保存会话数据"""
        self.sessions["updated_at"] = datetime.now().isoformat()
        with open(self.sessions_file, 'w', encoding='utf-8') as f:
            json.dump(self.sessions, f, ensure_ascii=False, indent=2)
    
    def record_session(self, session: Dict) -> str:
        """
        记录一个会话
        
        Args:
            session: 会话数据字典
            
        Returns:
            会话ID
        """
        session_id = session.get("session_id") or f"session_{datetime.now().strftime('%Y%m%d%H%M%S')}_{hashlib.md5(str(datetime.now()).encode()).hexdigest()[:6]}"
        session["session_id"] = session_id
        
        self.sessions["sessions"].append(session)
        self._save_sessions()
        
        # 触发模式学习
        self._learn_from_session(session)
        
        return session_id
    
    def _learn_from_session(self, session: Dict):
        """从会话中学习模式"""
        # 分析成功模式
        if session.get("success", False):
            self._extract_successful_pattern(session)
        
        # 分析改进建议
        improvements = session.get("improvements", [])
        if improvements:
            self._extract_improvement_pattern(session, improvements)
    
    def _extract_successful_pattern(self, session: Dict):
        """提取成功模式"""
        # 简化实现：基于模型和问题类型创建模式
        model = session.get("selected_model", "")
        problem_type = session.get("problem_type", "general")
        
        pattern_desc = f"使用{model}处理{problem_type}类型问题成功"
        
        # 检查是否已存在类似模式
        existing = None
        for p in self.patterns.get("patterns", []):
            if p.get("description") == pattern_desc:
                existing = p
                break
        
        if existing:
            existing["occurrence_count"] += 1
            existing["last_observed"] = datetime.now().isoformat()
            existing["confidence"] = min(1.0, existing["occurrence_count"] / 10)
        else:
            new_pattern = {
                "pattern_id": f"pattern_{hashlib.md5((pattern_desc + datetime.now().isoformat()).encode()).hexdigest()[:8]}",
                "pattern_type": "successful_pattern",
                "description": pattern_desc,
                "conditions": [problem_type],
                "actions": [model],
                "success_rate_impact": 0.05,
                "occurrence_count": 1,
                "last_observed": datetime.now().isoformat(),
                "confidence": 0.3,
                "source_sessions": [session.get("session_id", "")]
            }
            self.patterns["patterns"].append(new_pattern)
        
        self._save_patterns()
    
    def _extract_improvement_pattern(self, session: Dict, improvements: List[str]):
        """提取改进模式"""
        model = session.get("selected_model", "")
        
        pattern_desc = f"{model}的改进建议: {', '.join(improvements[:2])}"
        
        new_pattern = {
            "pattern_id": f"pattern_{hashlib.md5((pattern_desc + datetime.now().isoformat()).encode()).hexdigest()[:8]}",
            "pattern_type": "improvement_area",
            "description": pattern_desc,
            "conditions": [session.get("problem_type", "")],
            "actions": improvements,
            "success_rate_impact": 0.02,
            "occurrence_count": 1,
            "last_observed": datetime.now().isoformat(),
            "confidence": 0.2,
            "source_sessions": [session.get("session_id", "")]
        }
        
        self.patterns["patterns"].append(new_pattern)
        self._save_patterns()
    
    def get_patterns_by_type(self, pattern_type: str) -> List[Dict]:
        """按类型获取模式"""
        return [p for p in self.patterns.get("patterns", []) if p.get("pattern_type") == pattern_type]
    
    def get_relevant_patterns(self, problem_type: str, model: str) -> List[Dict]:
        """获取相关模式"""
        relevant = []
        for p in self.patterns.get("patterns", []):
            # 检查条件匹配
            conditions = p.get("conditions", [])
            if problem_type in conditions or not conditions:
                # 检查动作匹配
                actions = p.get("actions", [])
                if model in actions or not actions:
                    relevant.append(p)
        
        # 按置信度排序
        relevant.sort(key=lambda x: x.get("confidence", 0), reverse=True)
        return relevant
    
    def get_learning_summary(self) -> Dict:
        """获取学习摘要"""
        patterns = self.patterns.get("patterns", [])
        sessions = self.sessions.get("sessions", [])
        
        successful_patterns = [p for p in patterns if p.get("pattern_type") == "successful_pattern"]
        improvement_patterns = [p for p in patterns if p.get("pattern_type") == "improvement_area"]
        
        return {
            "total_patterns": len(patterns),
            "successful_patterns": len(successful_patterns),
            "improvement_patterns": len(improvement_patterns),
            "total_sessions": len(sessions),
            "avg_session_duration": statistics.mean([s.get("duration_ms", 0) for s in sessions]) if sessions else 0,
            "top_patterns": sorted(patterns, key=lambda x: x.get("confidence", 0), reverse=True)[:5]
        }


class ModelCollaborator:
    """多模型协同管理器"""
    
    def __init__(self, collab_dir: Optional[str] = None):
        """初始化多模型协同管理器"""
        if collab_dir is None:
            self.collab_dir = Path.home() / ".claude" / "thinking_models" / "collaborations"
        else:
            self.collab_dir = Path(collab_dir)
        
        self.collab_dir.mkdir(parents=True, exist_ok=True)
        self.collaborations_file = self.collab_dir / "collaborations.json"
        
        # 加载或创建协作配置
        self.collaborations = self._load_or_create_collaborations()
    
    def _load_or_create_collaborations(self) -> Dict:
        """加载或创建协作配置"""
        if self.collaborations_file.exists():
            with open(self.collaborations_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        
        # 默认协作配置
        default_collabs = {
            "version": "1.0",
            "updated_at": datetime.now().isoformat(),
            "collaborations": [
                {
                    "collaboration_id": "research_diagnostic",
                    "name": "研究-诊断协同",
                    "description": "先进行研究分析，再进行诊断修复",
                    "models": ["research_mode", "diagnostic_mode"],
                    "workflow": [
                        {"order": 1, "model": "research_mode", "action": "分析问题背景"},
                        {"order": 2, "model": "research_mode", "action": "收集解决方案"},
                        {"order": 3, "model": "diagnostic_mode", "action": "诊断具体问题"},
                        {"order": 4, "model": "diagnostic_mode", "action": "制定修复方案"}
                    ],
                    "conditions": ["复杂问题", "需要全面分析"],
                    "performance_metrics": {"avg_duration_ms": 0, "success_rate": 0},
                    "use_count": 0
                }
            ]
        }
        
        self._save_collaborations(default_collabs)
        return default_collabs
    
    def _save_collaborations(self, collabs: Dict):
        """保存协作配置"""
        collabs["updated_at"] = datetime.now().isoformat()
        with open(self.collaborations_file, 'w', encoding='utf-8') as f:
            json.dump(collabs, f, ensure_ascii=False, indent=2)
    
    def get_collaboration(self, collab_id: str) -> Optional[Dict]:
        """获取协作配置"""
        for collab in self.collaborations.get("collaborations", []):
            if collab.get("collaboration_id") == collab_id:
                return collab
        return None
    
    def find_applicable_collaboration(self, 
                                        problem_type: str,
                                        complexity: int) -> Optional[Dict]:
        """查找适用的协作配置"""
        applicable = []
        
        for collab in self.collaborations.get("collaborations", []):
            conditions = collab.get("conditions", [])
            
            # 检查条件匹配
            if "复杂问题" in conditions and complexity >= 7:
                applicable.append(collab)
            elif "简单问题" in conditions and complexity <= 3:
                applicable.append(collab)
            elif not conditions:
                applicable.append(collab)
        
        # 按使用次数排序
        applicable.sort(key=lambda x: x.get("use_count", 0), reverse=True)
        
        return applicable[0] if applicable else None
    
    def execute_collaboration(self, 
                               collab_id: str,
                               executors: Dict[str, Any]) -> Dict:
        """
        执行协作流程
        
        Args:
            collab_id: 协作ID
            executors: 模型执行器字典 {model_name: executor_func}
            
        Returns:
            执行结果
        """
        collab = self.get_collaboration(collab_id)
        if not collab:
            return {"status": "error", "message": "协作配置不存在"}
        
        results = []
        workflow = sorted(collab.get("workflow", []), key=lambda x: x.get("order", 0))
        
        for step in workflow:
            model = step.get("model")
            action = step.get("action")
            
            if model in executors:
                try:
                    result = executors[model](action)
                    results.append({
                        "step": step.get("order"),
                        "model": model,
                        "action": action,
                        "result": result,
                        "success": True
                    })
                except Exception as e:
                    results.append({
                        "step": step.get("order"),
                        "model": model,
                        "action": action,
                        "error": str(e),
                        "success": False
                    })
            else:
                results.append({
                    "step": step.get("order"),
                    "model": model,
                    "action": action,
                    "error": f"执行器不存在: {model}",
                    "success": False
                })
        
        # 更新使用计数
        collab["use_count"] = collab.get("use_count", 0) + 1
        self._save_collaborations(self.collaborations)
        
        return {
            "status": "completed",
            "collaboration": collab.get("name"),
            "results": results,
            "overall_success": all(r.get("success") for r in results)
        }
    
    def create_collaboration(self, collab: Dict) -> Optional[str]:
        """创建新协作配置"""
        try:
            collab["collaboration_id"] = collab.get("collaboration_id") or f"collab_{datetime.now().strftime('%Y%m%d%H%M%S')}"
            collab["use_count"] = 0
            collab["performance_metrics"] = {}
            
            self.collaborations["collaborations"].append(collab)
            self._save_collaborations(self.collaborations)
            
            return collab["collaboration_id"]
        except Exception:
            return None


class AdvancedAnalyzer:
    """高级分析器"""
    
    def __init__(self):
        """初始化高级分析器"""
        pass
    
    def analyze_success_factors(self, sessions: List[Dict]) -> Dict:
        """
        分析成功因素
        
        Args:
            sessions: 会话列表
            
        Returns:
            成功因素分析
        """
        successful = [s for s in sessions if s.get("success", False)]
        failed = [s for s in sessions if not s.get("success", False)]
        
        # 分析模型成功率
        model_success = defaultdict(lambda: {"success": 0, "total": 0})
        for s in sessions:
            model = s.get("selected_model", "unknown")
            model_success[model]["total"] += 1
            if s.get("success", False):
                model_success[model]["success"] += 1
        
        model_rates = {
            model: data["success"] / data["total"] 
            for model, data in model_success.items() 
            if data["total"] > 0
        }
        
        return {
            "total_sessions": len(sessions),
            "successful_sessions": len(successful),
            "failed_sessions": len(failed),
            "overall_success_rate": len(successful) / len(sessions) if sessions else 0,
            "model_success_rates": model_rates,
            "best_model": max(model_rates, key=model_rates.get) if model_rates else None,
            "success_factors": self._extract_success_factors(successful),
            "failure_factors": self._extract_failure_factors(failed)
        }
    
    def _extract_success_factors(self, successful: List[Dict]) -> List[str]:
        """提取成功因素"""
        factors = []
        
        # 检查高评分会话
        high_rated = [s for s in successful if s.get("user_rating", 0) >= 4]
        if high_rated:
            factors.append(f"用户评分高的会话 ({len(high_rated)}个) 通常成功")
        
        # 检查平均耗时
        if successful:
            avg_duration = statistics.mean([s.get("duration_ms", 0) for s in successful])
            factors.append(f"成功会话平均耗时: {avg_duration:.0f}ms")
        
        return factors
    
    def _extract_failure_factors(self, failed: List[Dict]) -> List[str]:
        """提取失败因素"""
        factors = []
        
        if failed:
            failed_models = defaultdict(int)
            for s in failed:
                model = s.get("selected_model", "unknown")
                failed_models[model] += 1
            
            if failed_models:
                worst_model = max(failed_models, key=failed_models.get)
                factors.append(f"{worst_model} 模型失败次数最多 ({failed_models[worst_model]}次)")
        
        return factors
    
    def generate_optimization_suggestions(self, analysis: Dict) -> List[str]:
        """生成优化建议"""
        suggestions = []
        
        # 基于成功率建议
        model_rates = analysis.get("model_success_rates", {})
        if model_rates:
            worst_model = min(model_rates, key=model_rates.get)
            worst_rate = model_rates[worst_model]
            
            if worst_rate < 0.5:
                suggestions.append(f"考虑优化 {worst_model} 模型，成功率仅为 {worst_rate*100:.0f}%")
                suggestions.append(f"建议检查该模型的适用条件和阶段配置")
        
        # 基于总体成功率
        overall = analysis.get("overall_success_rate", 0)
        if overall < 0.7:
            suggestions.append("总体成功率较低，建议增加记忆库中的成功案例")
            suggestions.append("考虑使用更简单的问题分解方法")
        elif overall > 0.9:
            suggestions.append("成功率很高！可以考虑尝试更复杂的问题")
        
        # 基于会话量
        total = analysis.get("total_sessions", 0)
        if total < 10:
            suggestions.append("样本量较小，建议继续积累更多会话数据")
        elif total > 100:
            suggestions.append("已有足够的样本数据，可以进行深度模式分析")
        
        return suggestions
    
    def predict_outcome(self, 
                        problem: str, 
                        model: str,
                        historical_data: List[Dict]) -> Dict:
        """
        预测结果
        
        Args:
            problem: 问题描述
            model: 选择的模型
            historical_data: 历史数据
            
        Returns:
            预测结果
        """
        # 查找相似历史案例
        similar = []
        for session in historical_data:
            if model == session.get("selected_model"):
                similar.append(session)
        
        if not similar:
            return {
                "predicted_success": 0.5,
                "confidence": "low",
                "reason": "缺乏历史数据",
                "suggestion": "建议从简单问题开始积累数据"
            }
        
        # 计算成功率
        successful = [s for s in similar if s.get("success", False)]
        success_rate = len(successful) / len(similar) if similar else 0.5
        
        # 评估置信度
        confidence = "high" if len(similar) >= 20 else "medium" if len(similar) >= 5 else "low"
        
        # 生成建议
        suggestions = []
        if success_rate < 0.5:
            suggestions.append("建议更换其他模型")
        elif success_rate > 0.8:
            suggestions.append("该模型对此类问题表现良好")
        
        return {
            "predicted_success": success_rate,
            "confidence": confidence,
            "similar_cases": len(similar),
            "historical_success_rate": success_rate,
            "suggestions": suggestions
        }


def get_continuous_learning() -> ContinuousLearning:
    """获取持续学习系统实例"""
    return ContinuousLearning()


def get_model_collaborator() -> ModelCollaborator:
    """获取多模型协同管理器实例"""
    return ModelCollaborator()


def get_advanced_analyzer() -> AdvancedAnalyzer:
    """获取高级分析器实例"""
    return AdvancedAnalyzer()