Claude Opus 4.8 发布：Dynamic Workflows 驱动的"工程协作系统"范式跃迁

# Claude Opus 4.8 关键能力参数
capabilities = {
    "max_context_window": 1_000_000,  # 100万 Token 上下文
    "max_concurrent_agents": 500,       # 最多500个并行子 Agent
    "codebase_scale": "100K+ lines",    # 支持10万+行代码迁移
    "honesty_improvement": 0.25,        # 缺陷隐瞒率降至1/4
    "fast_mode_speedup": 2.5,           # Fast Mode 提速2.5倍
    "cost_reduction": 0.33,             # 成本降至1/3
}

# 任务分解示例
original_task = "将整个 Django 项目从 3.2 迁移到 5.0 版本"
decomposed_tasks = decompose_task(original_task)
# 输出：
# [
#     Task(id=1, type="dependency_analysis", deps=[]),
#     Task(id=2, type="breaking_changes_scan", deps=[1]),
#     Task(id=3, type="model_migration", deps=[2]),
#     Task(id=4, type="view_migration", deps=[2]),
#     Task(id=5, type="test_update", deps=[3, 4]),
#     ...
# ]

"""
Claude Opus 4.8 Dynamic Workflows - 并行任务调度器核心实现
Parallel Task Scheduler Core Implementation

核心功能：
1. 任务依赖图管理（DAG）
2. 并行执行调度
3. 动态优先级调整
4. 失败重试与容错
"""

import asyncio
import heapq
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Dict, List, Optional, Set
from collections import defaultdict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class TaskStatus(Enum):
    """任务状态枚举"""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
    WAITING_DEPS = "waiting_dependencies"


class TaskPriority(Enum):
    """任务优先级"""
    CRITICAL = 0   # 关键路径任务
    HIGH = 1       # 高优先级
    MEDIUM = 2     # 中优先级
    LOW = 3        # 低优先级


@dataclass
class Task:
    """任务定义"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = ""
    task_type: str = ""           # 任务类型：code_generation, review, test, etc.
    payload: Dict[str, Any] = field(default_factory=dict)  # 任务输入数据
    dependencies: Set[str] = field(default_factory=set)    # 依赖的任务ID集合
    priority: TaskPriority = TaskPriority.MEDIUM
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None            # 任务执行结果
    error: Optional[str] = None   # 错误信息
    created_at: float = field(default_factory=lambda: asyncio.get_event_loop().time())
    started_at: Optional[float] = None
    completed_at: Optional[float] = None
    retry_count: int = 0
    max_retries: int = 3
    
    def __lt__(self, other):
        """支持优先级队列排序"""
        return self.priority.value < other.priority.value


class AgentExecutor:
    """
    子 Agent 执行器
    每个执行器负责一种特定类型的任务
    """
    
    def __init__(self, agent_type: str, max_concurrent: int = 5):
        self.agent_type = agent_type
        self.max_concurrent = max_concurrent
        self._semaphore = asyncio.Semaphore(max_concurrent)
        self._active_tasks: Set[str] = set()
        
    async def execute(self, task: Task) -> Any:
        """
        执行任务的异步方法
        
        Args:
            task: 要执行的任务
            
        Returns:
            任务执行结果
        """
        async with self._semaphore:
            self._active_tasks.add(task.id)
            logger.info(f"[{self.agent_type}] 开始执行任务 {task.id}: {task.name}")
            
            try:
                # 模拟不同类型 Agent 的执行逻辑
                if task.task_type == "code_generation":
                    result = await self._execute_code_generation(task)
                elif task.task_type == "code_review":
                    result = await self._execute_code_review(task)
                elif task.task_type == "test_generation":
                    result = await self._execute_test_generation(task)
                elif task.task_type == "code_search":
                    result = await self._execute_code_search(task)
                elif task.task_type == "documentation":
                    result = await self._execute_documentation(task)
                else:
                    result = await self._execute_generic(task)
                    
                logger.info(f"[{self.agent_type}] 任务 {task.id} 执行成功")
                return result
                
            except Exception as e:
                logger.error(f"[{self.agent_type}] 任务 {task.id} 执行失败: {e}")
                raise
            finally:
                self._active_tasks.discard(task.id)
    
    async def _execute_code_generation(self, task: Task) -> Dict[str, Any]:
        """代码生成 Agent 执行逻辑"""
        # 模拟 AI 代码生成过程
        await asyncio.sleep(0.1)  # 模拟处理时间
        
        code_template = task.payload.get("template", "")
        context = task.payload.get("context", {})
        
        return {
            "agent_type": "code_generation",
            "task_id": task.id,
            "generated_code": f"# Generated code for {context.get('target', 'unknown')}\n"
                             f"# Template: {code_template}\n"
                             f"class GeneratedClass:\n"
                             f"    def __init__(self):\n"
                             f"        pass\n",
            "files_created": [f"{context.get('name', 'module')}.py"],
            "lines_of_code": 50,
        }
    
    async def _execute_code_review(self, task: Task) -> Dict[str, Any]:
        """代码审查 Agent 执行逻辑"""
        await asyncio.sleep(0.15)
        
        code_content = task.payload.get("code", "")
        
        # 模拟静态分析
        issues = []
        if "TODO" in code_content:
            issues.append({"severity": "info", "message": "Contains TODO comments"})
        if "password" in code_content.lower() and "env" not in code_content.lower():
            issues.append({"severity": "warning", "message": "Potential hardcoded credential"})
            
        return {
            "agent_type": "code_review",
            "task_id": task.id,
            "issues_found": len(issues),
            "issues": issues,
            "score": 85 if len(issues) == 0 else 75,
            "recommendations": ["Consider adding type hints", "Add docstrings"],
        }
    
    async def _execute_test_generation(self, task: Task) -> Dict[str, Any]:
        """测试生成 Agent 执行逻辑"""
        await asyncio.sleep(0.12)
        
        target_code = task.payload.get("target_code", "")
        
        return {
            "agent_type": "test_generation",
            "task_id": task.id,
            "test_cases": [
                {"name": "test_basic_case", "status": "passed"},
                {"name": "test_edge_case", "status": "passed"},
                {"name": "test_error_case", "status": "passed"},
            ],
            "coverage": 92.5,
            "test_file": "test_generated.py",
        }
    
    async def _execute_code_search(self, task: Task) -> Dict[str, Any]:
        """代码检索 Agent 执行逻辑"""
        await asyncio.sleep(0.08)
        
        query = task.payload.get("query", "")
        
        return {
            "agent_type": "code_search",
            "task_id": task.id,
            "query": query,
            "results": [
                {"file": "utils.py", "line": 42, "match": f"...{query}..."},
                {"file": "main.py", "line": 128, "match": f"...{query}..."},
            ],
            "total_matches": 2,
        }
    
    async def _execute_documentation(self, task: Task) -> Dict[str, Any]:
        """文档生成 Agent 执行逻辑"""
        await asyncio.sleep(0.1)
        
        target = task.payload.get("target", "module")
        
        return {
            "agent_type": "documentation",
            "task_id": task.id,
            "doc_file": f"{target}_docs.md",
            "sections": ["Overview", "API Reference", "Examples", "FAQ"],
            "generated_lines": 150,
        }
    
    async def _execute_generic(self, task: Task) -> Dict[str, Any]:
        """通用任务执行"""
        await asyncio.sleep(0.1)
        return {"status": "completed", "task_id": task.id}


class ParallelTaskScheduler:
    """
    并行任务调度器 - Dynamic Workflows 的核心引擎
    
    特性：
    1. 基于 DAG 的任务依赖管理
    2. 支持任务并行执行（受限于依赖关系和并发限制）
    3. 动态优先级调度
    4. 自动重试与容错
    5. 执行进度监控
    """
    
    def __init__(self, max_concurrent_agents: int = 100):
        self.max_concurrent_agents = max_concurrent_agents
        self.tasks: Dict[str, Task] = {}
        self.dependency_graph: Dict[str, Set[str]] = defaultdict(set)  # task_id -> dependencies
        self.reverse_graph: Dict[str, Set[str]] = defaultdict(set)     # task_id -> dependents
        
        # Agent 执行器池
        self.agent_executors: Dict[str, AgentExecutor] = {}
        self._initialize_default_agents()
        
        # 执行状态
        self._running_tasks: Set[str] = set()
        self._completed_tasks: Set[str] = set()
        self._failed_tasks: Set[str] = set()
        
        # 锁和信号量
        self._lock = asyncio.Lock()
        self._task_available = asyncio.Condition(self._lock)
        
        # 统计信息
        self.stats = {
            "total_tasks": 0,
            "completed_tasks": 0,
            "failed_tasks": 0,
            "total_execution_time": 0.0,
        }
    
    def _initialize_default_agents(self):
        """初始化默认的 Agent 执行器"""
        agent_configs = [
            ("code_generation", 20),
            ("code_review", 15),
            ("test_generation", 15),
            ("code_search", 10),
            ("documentation", 10),
            ("migration", 10),
            ("refactoring", 10),
        ]
        
        for agent_type, max_concurrent in agent_configs:
            self.agent_executors[agent_type] = AgentExecutor(
                agent_type, 
                max_concurrent
            )
    
    def add_task(self, task: Task) -> str:
        """
        添加任务到调度器
        
        Args:
            task: 要添加的任务
            
        Returns:
            任务ID
        """
        self.tasks[task.id] = task
        self.dependency_graph[task.id] = task.dependencies.copy()
        
        # 构建反向依赖图
        for dep_id in task.dependencies:
            self.reverse_graph[dep_id].add(task.id)
            
        self.stats["total_tasks"] += 1
        logger.info(f"添加任务 {task.id}: {task.name}, 类型: {task.task_type}, 依赖: {task.dependencies}")
        
        return task.id
    
    def add_tasks_batch(self, tasks: List[Task]) -> List[str]:
        """批量添加任务"""
        return [self.add_task(task) for task in tasks]
    
    def _get_ready_tasks(self) -> List[Task]:
        """获取所有准备就绪的任务（依赖已满足且未执行）"""
        ready = []
        
        for task_id, task in self.tasks.items():
            if task.status != TaskStatus.PENDING:
                continue
                
            # 检查依赖是否都已完成
            deps_completed = all(
                self.tasks[dep_id].status == TaskStatus.COMPLETED
                for dep_id in task.dependencies
            )
            
            if deps_completed:
                ready.append(task)
        
        return ready
    
    def _can_run_task(self, task: Task) -> bool:
        """检查任务是否可以执行"""
        if task.status != TaskStatus.PENDING:
            return False
            
        # 检查依赖
        if not all(
            self.tasks[dep_id].status == TaskStatus.COMPLETED
            for dep_id in task.dependencies
        ):
            return False
            
        # 检查并发限制
        if len(self._running_tasks) >= self.max_concurrent_agents:
            return False
            
        return True
    
    async def _execute_single_task(self, task: Task) -> None:
        """执行单个任务"""
        async with self._lock:
            if not self._can_run_task(task):
                return
                
            task.status = TaskStatus.RUNNING
            task.started_at = asyncio.get_event_loop().time()
            self._running_tasks.add(task.id)
        
        try:
            # 获取对应的 Agent 执行器
            executor = self.agent_executors.get(
                task.task_type, 
                AgentExecutor(task.task_type)
            )
            
            # 执行任务
            result = await executor.execute(task)
            
            # 更新任务状态
            async with self._lock:
                task.status = TaskStatus.COMPLETED
                task.result = result
                task.completed_at = asyncio.get_event_loop().time()
                self._running_tasks.discard(task.id)
                self._completed_tasks.add(task.id)
                self.stats["completed_tasks"] += 1
                
                # 通知等待的任务
                self._task_available.notify_all()
                
        except Exception as e:
            async with self._lock:
                task.status = TaskStatus.FAILED
                task.error = str(e)
                task.completed_at = asyncio.get_event_loop().time()
                self._running_tasks.discard(task.id)
                self._failed_tasks.add(task.id)
                self.stats["failed_tasks"] += 1
                
                logger.error(f"任务 {task.id} 执行失败: {e}")
                
                # 尝试重试
                if task.retry_count < task.max_retries:
                    task.retry_count += 1
                    task.status = TaskStatus.PENDING
                    logger.info(f"任务 {task.id} 将在重试 ({task.retry_count}/{task.max_retries})")
    
    async def run(self, timeout: Optional[float] = None) -> Dict[str, Task]:
        """
        运行调度器，执行所有任务
        
        Args:
            timeout: 可选的执行超时时间（秒）
            
        Returns:
            所有任务的状态字典
        """
        start_time = asyncio.get_event_loop().time()
        
        async def worker():
            """工作协程：持续获取并执行就绪任务"""
            while True:
                async with self._task_available:
                    # 等待就绪任务
                    while True:
                        ready_tasks = self._get_ready_tasks()
                        if ready_tasks:
                            break
                        
                        # 检查是否还有运行中的任务
                        if not self._running_tasks:
                            # 检查是否所有任务都已完成
                            pending = sum(1 for t in self.tasks.values() 
                                        if t.status == TaskStatus.PENDING)
                            if pending == 0:
                                return  # 所有任务完成
                        
                        # 等待信号
                        await asyncio.wait_for(
                            self._task_available.wait(),
                            timeout=1.0
                        )
                        
                        # 检查超时
                        if timeout and (asyncio.get_event_loop().time() - start_time) > timeout:
                            logger.warning("调度器执行超时")
                            return
                
                # 执行所有就绪任务
                for task in ready_tasks:
                    if self._can_run_task(task):
                        asyncio.create_task(self._execute_single_task(task))
        
        await worker()
        
        self.stats["total_execution_time"] = asyncio.get_event_loop().time() - start_time
        
        return {task_id: task for task_id, task in self.tasks.items()}
    
    def get_progress(self) -> Dict[str, Any]:
        """获取执行进度"""
        total = len(self.tasks)
        completed = len(self._completed_tasks)
        failed = len(self._failed_tasks)
        running = len(self._running_tasks)
        pending = total - completed - failed - running
        
        return {
            "total": total,
            "completed": completed,
            "failed": failed,
            "running": running,
            "pending": pending,
            "progress_percent": (completed / total * 100) if total > 0 else 0,
            "stats": self.stats.copy(),
        }
    
    def visualize_dag(self) -> str:
        """生成 DAG 的可视化表示"""
        lines = ["digraph tasks {"]
        lines.append("    rankdir=TB;")
        lines.append("    node [shape=box];")
        
        for task_id, task in self.tasks.items():
            color = {
                TaskStatus.COMPLETED: "green",
                TaskStatus.FAILED: "red",
                TaskStatus.RUNNING: "yellow",
                TaskStatus.PENDING: "gray",
            }.get(task.status, "gray")
            
            lines.append(f'    "{task_id}" [label="{task.name}", fillcolor={color}, style=filled];')
        
        for task_id, deps in self.dependency_graph.items():
            for dep_id in deps:
                lines.append(f'    "{dep_id}" -> "{task_id}";')
        
        lines.append("}")
        return "\n".join(lines)


# 使用示例
async def demo_codebase_migration():
    """演示：跨10万行代码的代码库迁移任务"""
    
    print("=" * 60)
    print("演示：使用 Dynamic Workflows 进行代码库迁移")
    print("=" * 60)
    
    # 创建调度器（最多100个并发Agent）
    scheduler = ParallelTaskScheduler(max_concurrent_agents=100)
    
    # 第一阶段：依赖分析与准备工作
    print("\n📊 阶段1：依赖分析与准备工作")
    
    # 分析现有依赖
    dep_analysis = Task(
        name="分析项目依赖",
        task_type="code_search",
        payload={"query": "import.*django", "scope": "all_files"}
    )
    scheduler.add_task(dep_analysis)
    
    # 扫描 breaking changes
    breaking_changes = Task(
        name="扫描 Breaking Changes",
        task_type="code_review",
        dependencies={dep_analysis.id},
        payload={"check": "django_breaking_changes", "target_version": "5.0"}
    )
    scheduler.add_task(breaking_changes)
    
    # 识别需要迁移的模型文件
    model_files = Task(
        name="识别模型文件",
        task_type="code_search",
        dependencies={dep_analysis.id},
        payload={"query": "class.*models\\.Model", "scope": "models.py"}
    )
    scheduler.add_task(model_files)
    
    # 第二阶段：模型迁移
    print("\n📦 阶段2：模型迁移")
    
    model_migration = Task(
        name="迁移 Django 模型",
        task_type="code_generation",
        dependencies={breaking_changes.id, model_files.id},
        priority=TaskPriority.CRITICAL,
        payload={
            "template": "django_model_v5",
            "context": {"name": "models", "target": "Django 5.0"}
        }
    )
    scheduler.add_task(model_migration)
    
    # 第三阶段：视图迁移
    print("\n🎯 阶段3：视图迁移")
    
    view_migration = Task(
        name="迁移视图函数",
        task_type="code_generation",
        dependencies={breaking_changes.id},
        payload={
            "template": "django_view_v5",
            "context": {"name": "views", "target": "Django 5.0"}
        }
    )
    scheduler.add_task(view_migration)
    
    # 第四阶段：测试更新
    print("\n🧪 阶段4：测试更新")
    
    test_update = Task(
        name="更新单元测试",
        task_type="test_generation",
        dependencies={model_migration.id, view_migration.id},
        payload={"target_code": "migrated_code", "coverage_target": 95}
    )
    scheduler.add_task(test_update)
    
    # 第五阶段：文档更新
    print("\n📝 阶段5：文档更新")
    
    docs_update = Task(
        name="更新 API 文档",
        task_type="documentation",
        dependencies={model_migration.id, view_migration.id},
        payload={"target": "api_documentation"}
    )
    scheduler.add_task(docs_update)
    
    # 第六阶段：最终审查
    print("\n🔍 阶段6：最终审查")
    
    final_review = Task(
        name="最终代码审查",
        task_type="code_review",
        dependencies={test_update.id, docs_update.id},
        priority=TaskPriority.HIGH,
        payload={"code": "final_migrated_codebase", "strict_mode": True}
    )
    scheduler.add_task(final_review)
    
    # 执行调度
    print("\n🚀 开始执行调度...")
    results = await scheduler.run(timeout=60.0)
    
    # 打印结果
    print("\n" + "=" * 60)
    print("执行结果")
    print("=" * 60)
    
    progress = scheduler.get_progress()
    print(f"总任务数: {progress['total']}")
    print(f"已完成: {progress['completed']}")
    print(f"失败: {progress['failed']}")
    print(f"进行中: {progress['running']}")
    print(f"耗时: {progress['stats']['total_execution_time']:.2f}秒")
    
    # 打印 DAG
    print("\n📊 任务依赖图:")
    print(scheduler.visualize_dag())
    
    return results


# 运行演示
if __name__ == "__main__":
    asyncio.run(demo_codebase_migration())

"""
Claude Opus 4.8 Dynamic Workflows - 子 Agent 架构实现
Multi-Agent Architecture with MCP Protocol Support

核心功能：
1. 专业化 Agent 实现
2. Agent 间通信协议
3. MCP (Model Context Protocol) 兼容
4. Agent 协作与状态共享
"""

import asyncio
import json
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional, Callable
from collections import defaultdict


class AgentCapability(Enum):
    """Agent 能力枚举"""
    CODE_GENERATION = "code_generation"
    CODE_REVIEW = "code_review"
    CODE_SEARCH = "code_search"
    TEST_GENERATION = "test_generation"
    DOCUMENTATION = "documentation"
    REFACTORING = "refactoring"
    DEBUGGING = "debugging"
    OPTIMIZATION = "optimization"
    MIGRATION = "migration"


@dataclass
class AgentMessage:
    """Agent 间通信消息"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    sender: str = ""
    receiver: Optional[str] = None  # None 表示广播
    content: Dict[str, Any] = field(default_factory=dict)
    message_type: str = "request"  # request, response, event, error
    timestamp: float = field(default_factory=lambda: asyncio.get_event_loop().time())
    correlation_id: Optional[str] = None  # 用于关联请求和响应


class AgentState:
    """Agent 状态管理"""
    
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.memory: Dict[str, Any] = {}
        self.context: List[Dict[str, Any]] = []
        self.capabilities: List[AgentCapability] = []
        self.status: str = "idle"
        self.current_task: Optional[str] = None
        
    def remember(self, key: str, value: Any) -> None:
        """记忆存储"""
        self.memory[key] = value
        self.context.append({
            "type": "memory",
            "key": key,
            "timestamp": asyncio.get_event_loop().time()
        })
        
    def recall(self, key: str) -> Optional[Any]:
        """记忆召回"""
        return self.memory.get(key)
    
    def get_context_summary(self, max_items: int = 100) -> str:
        """获取上下文摘要"""
        recent = self.context[-max_items:]
        return json.dumps(recent, indent=2)


class BaseAgent(ABC):
    """
    基础 Agent 抽象类
    
    所有专业化 Agent 都应继承此类
    """
    
    def __init__(
        self,
        agent_id: str,
        name: str,
        capabilities: List[AgentCapability],
        max_retries: int = 3
    ):
        self.agent_id = agent_id
        self.name = name
        self.capabilities = capabilities
        self.max_retries = max_retries
        
        self.state = AgentState(agent_id)
        self.message_queue: asyncio.Queue[AgentMessage] = asyncio.Queue()
        self._running = False
        
        # 消息处理函数映射
        self.handlers: Dict[str, Callable] = {}
        self._register_default_handlers()
    
    def _register_default_handlers(self):
        """注册默认消息处理器"""
        self.handlers["execute"] = self.handle_execute
        self.handlers["query"] = self.handle_query
        self.handlers["status"] = self.handle_status
    
    @abstractmethod
    async def initialize(self) -> None:
        """Agent 初始化"""
        pass
    
    @abstractmethod
    async def execute_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """
        执行任务（子类必须实现）
        
        Args:
            task: 任务定义
            
        Returns:
            执行结果
        """
        pass
    
    async def process_message(self, message: AgentMessage) -> Optional[AgentMessage]:
        """处理接收到的消息"""
        handler = self.handlers.get(message.content.get("action"))
        
        if handler:
            try:
                result = await handler(message)
                return AgentMessage(
                    sender=self.agent_id,
                    receiver=message.sender,
                    content={"status": "success", "result": result},
                    message_type="response",
                    correlation_id=message.id
                )
            except Exception as e:
                return AgentMessage(
                    sender=self.agent_id,
                    receiver=message.sender,
                    content={"status": "error", "error": str(e)},
                    message_type="error",
                    correlation_id=message.id
                )
        
        return None
    
    async def handle_execute(self, message: AgentMessage) -> Dict[str, Any]:
        """处理执行请求"""
        task = message.content.get("task", {})
        self.state.status = "working"
        self.state.current_task = task.get("id")
        
        try:
            result = await self.execute_task(task)
            self.state.status = "idle"
            self.state.current_task = None
            return result
        except Exception as e:
            self.state.status = "error"
            raise
    
    async def handle_query(self, message: AgentMessage) -> Dict[str, Any]:
        """处理查询请求"""
        query_type = message.content.get("query_type")
        
        if query_type == "status":
            return {"status": self.state.status, "task": self.state.current_task}
        elif query_type == "capabilities":
            return {"capabilities": [c.value for c in self.capabilities]}
        elif query_type == "memory":
            return {"memory": self.state.memory}
        
        return {"error": "Unknown query type"}
    
    async def handle_status(self, message: AgentMessage) -> Dict[str, Any]:
        """处理状态查询"""
        return {
            "agent_id": self.agent_id,
            "name": self.name,
            "status": self.state.status,
            "current_task": self.state.current_task,
            "capabilities": [c.value for c in self.capabilities]
        }
    
    async def send_message(
        self,
        to: str,
        content: Dict[str, Any],
        message_type: str = "request"
    ) -> AgentMessage:
        """发送消息（需要通过 AgentNetwork）"""
        return AgentMessage(
            sender=self.agent_id,
            receiver=to,
            content=content,
            message_type=message_type
        )


class CodeGenerationAgent(BaseAgent):
    """
    代码生成 Agent
    
    负责根据需求生成代码，支持多种编程语言和框架
    """
    
    def __init__(self, agent_id: str = "code_gen_001"):
        super().__init__(
            agent_id=agent_id,
            name="Code Generation Agent",
            capabilities=[
                AgentCapability.CODE_GENERATION,
                AgentCapability.REFACTORING
            ]
        )
        self.supported_languages = ["python", "go", "typescript", "rust", "java"]
    
    async def initialize(self) -> None:
        """初始化代码生成 Agent"""
        self.state.remember("initialized", True)
        self.state.remember("supported_languages", self.supported_languages)
        print(f"[{self.name}] 初始化完成，支持语言: {self.supported_languages}")
    
    async def execute_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """执行代码生成任务"""
        language = task.get("language", "python")
        spec = task.get("spec", {})
        context = task.get("context", {})
        
        if language not in self.supported_languages:
            raise ValueError(f"不支持的语言: {language}")
        
        # 根据语言生成代码
        if language == "python":
            code = self._generate_python_code(spec, context)
        elif language == "go":
            code = self._generate_go_code(spec, context)
        elif language == "typescript":
            code = self._generate_typescript_code(spec, context)
        else:
            code = self._generate_generic_code(language, spec, context)
        
        # 记忆生成的代码
        self.state.remember(f"code_{task.get('id')}", code)
        
        return {
            "language": language,
            "code": code,
            "files": [f"generated.{self._get_extension(language)}"],
            "lines": code.count('\n') + 1
        }
    
    def _generate_python_code(self, spec: Dict, context: Dict) -> str:
        """生成 Python 代码"""
        class_name = context.get("class_name", "GeneratedClass")
        
        return f'''"""
Auto-generated Python module
Generated by {self.name}
"""

from typing import Optional, List, Dict, Any
from dataclasses import dataclass, field
import asyncio


@dataclass
class {class_name}:
    """
    Auto-generated class based on specification.
    
    Attributes:
        name: The name of the instance
        data: Internal data storage
        config: Configuration options
    """
    
    name: str
    data: Dict[str, Any] = field(default_factory=dict)
    config: Optional[Dict[str, Any]] = None
    
    def __post_init__(self):
        """Post-initialization setup"""
        if self.config is None:
            self.config = {{}}
        self._initialized = True
    
    def process(self, input_data: Any) -> Any:
        """
        Process input data.
        
        Args:
            input_data: Input to process
            
        Returns:
            Processed result
        """
        # TODO: Implement processing logic
        return {{
            "status": "processed",
            "input": input_data,
            "source": self.name
        }}
    
    async def async_process(self, input_data: Any) -> Any:
        """
        Asynchronous processing.
        
        Args:
            input_data: Input to process
            
        Returns:
            Processed result
        """
        await asyncio.sleep(0.1)  # Simulate async operation
        return self.process(input_data)
    
    @classmethod
    def create_default(cls) -> "{class_name}":
        """Factory method to create default instance."""
        return cls(name="default", data={{}})
    
    def __repr__(self) -> str:
        return f"{class_name}(name='{self.name}', data={self.data})"


# Example usage
if __name__ == "__main__":
    instance = {class_name}(
        name="example",
        data={{"key": "value"}}
    )
    result = instance.process({{"test": "data"}})
    print(f"Result: {{result}}")
'''
    
    def _generate_go_code(self, spec: Dict, context: Dict) -> str:
        """生成 Go 代码"""
        struct_name = context.get("struct_name", "GeneratedStruct")
        
        return f'''// Auto-generated Go module
// Generated by {self.name}

package main

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

// {struct_name} represents an auto-generated struct
type {struct_name} struct {{
    mu    sync.RWMutex
    name  string
    data  map[string]interface{{}}
    ready bool
}}

// New{struct_name} creates a new instance
func New{struct_name}(name string) *{struct_name} {{
    return &{struct_name}{{
        name: name,
        data: make(map[string]interface{{}}),
    }}
}}

// Process handles input processing
func (s *{struct_name}) Process(ctx context.Context, input interface{{}}) (interface{{}}, error) {{
    s.mu.Lock()
    defer s.mu.Unlock()
    
    s.ready = true
    s.data["last_input"] = input
    s.data["processed_at"] = time.Now().UTC()
    
    return map[string]interface{{}}{{
        "status": "processed",
        "input": input,
        "source": s.name,
    }}, nil
}}

// GetStatus returns current status
func (s *{struct_name}) GetStatus() map[string]interface{{}} {{
    s.mu.RLock()
    defer s.mu.RUnlock()
    
    return map[string]interface{{}}{{
        "name": s.name,
        "ready": s.ready,
        "data_count": len(s.data),
    }}
}}

// MarshalJSON custom JSON marshaling
func (s *{struct_name}) MarshalJSON() ([]byte, error) {{
    type Alias {struct_name}
    return json.Marshal(&struct {{
        Alias
        Status map[string]interface{{}} `json:"status"`
    }}{{
        Alias:   Alias(*s),
        Status:  s.GetStatus(),
    }})
}}

func main() {{
    ctx := context.Background()
    instance := New{struct_name}("example")
    
    result, err := instance.Process(ctx, map[string]string{{"test": "data"}})
    if err != nil {{
        fmt.Printf("Error: %v\\n", err)
        return
    }}
    
    fmt.Printf("Result: %+v\\n", result)
}}
'''
    
    def _generate_typescript_code(self, spec: Dict, context: Dict) -> str:
        """生成 TypeScript 代码"""
        class_name = context.get("class_name", "GeneratedClass")
        
        return f'''/**
 * Auto-generated TypeScript module
 * Generated by {self.name}
 */

interface ProcessedResult<T = unknown> {{
  status: 'success' | 'error';
  data: T;
  timestamp: number;
}}

class {class_name}<T = unknown> {{
  private readonly name: string;
  private data: Map<string, unknown>;
  private initialized: boolean;

  constructor(name: string) {{
    this.name = name;
    this.data = new Map();
    this.initialized = false;
  }}

  async initialize(): Promise<void> {{
    this.initialized = true;
    this.data.set('initialized_at', Date.now());
  }}

  async process(input: T): Promise<ProcessedResult<T>> {{
    if (!this.initialized) {{
      await this.initialize();
    }}

    const result: ProcessedResult<T> = {{
      status: 'success',
      data: input,
      timestamp: Date.now(),
    }};

    this.data.set('last_result', result);
    return result;
  }}

  getStatus(): {{ name: string; initialized: boolean; dataSize: number }} {{
    return {{
      name: this.name,
      initialized: this.initialized,
      dataSize: this.data.size,
    }};
  }}
}}

// Example usage
async function main() {{
  const instance = new {class_name}<{{ test: string }}>('example');
  const result = await instance.process({{ test: 'data' }});
  console.log('Result:', result);
}}

main().catch(console.error);

export {{ {class_name}, ProcessedResult }};
'''
    
    def _generate_generic_code(self, language: str, spec: Dict, context: Dict) -> str:
        """生成通用代码"""
        return f'''// Auto-generated code for {language}
// This is a placeholder for languages without specific generators
console.log("Generated for language: {language}");
'''
    
    def _get_extension(self, language: str) -> str:
        """获取文件扩展名"""
        extensions = {
            "python": "py",
            "go": "go",
            "typescript": "ts",
            "rust": "rs",
            "java": "java",
        }
        return extensions.get(language, "txt")


class CodeReviewAgent(BaseAgent):
    """
    代码审查 Agent
    
    负责静态代码分析、漏洞检测、性能分析等
    """
    
    def __init__(self, agent_id: str = "code_review_001"):
        super().__init__(
            agent_id=agent_id,
            name="Code Review Agent",
            capabilities=[
                AgentCapability.CODE_REVIEW,
                AgentCapability.DEBUGGING
            ]
        )
        self.review_rules = self._load_review_rules()
    
    def _load_review_rules(self) -> Dict[str, Any]:
        """加载审查规则"""
        return {
            "security": [
                {"pattern": r"password\s*=", "severity": "high", "message": "Potential hardcoded password"},
                {"pattern": r"eval\s*\(", "severity": "high", "message": "Use of eval() is dangerous"},
                {"pattern": r"exec\s*\(", "severity": "medium", "message": "Use of exec() should be avoided"},
            ],
            "style": [
                {"pattern": r"TODO", "severity": "info", "message": "TODO comment found"},
                {"pattern": r"FIXME", "severity": "warning", "message": "FIXME comment found"},
            ],
            "performance": [
                {"pattern": r"for.*in.*range\(len\(", "severity": "info", "message": "Consider using enumerate()"},
            ]
        }
    
    async def initialize(self) -> None:
        """初始化代码审查 Agent"""
        self.state.remember("initialized", True)
        self.state.remember("review_rules_count", len(self.review_rules))
        print(f"[{self.name}] 初始化完成，规则数: {len(self.review_rules)}")
    
    async def execute_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
        """执行代码审查任务"""
        code = task.get("code", "")
        language = task.get("language", "python")
        strict_mode = task.get("strict_mode", False)
        
        issues = self._analyze_code(code, strict_mode)
        
        # 计算评分
        score = self._calculate_score(issues, strict_mode)
        
        return {
            "issues_found": len(issues),
            "issues": issues,
            "score": score,
            "language": language,
            "recommendations": self._generate_recommendations(issues)
        }
    
    def _analyze_code(self, code: str, strict_mode: bool) -> List[Dict[str, Any]]:
        """分析代码问题"""
        import re
        issues = []
        
        for category, rules in self.review_rules.items():
            for rule in rules:
                matches = re.finditer(rule["pattern"], code, re.IGNORECASE)
                for match in matches:
                    issues.append({
                        "category": category,
                        "severity": rule["severity"],
                        "message": rule["message"],
                        "line": code[:match.start()].count('\n') + 1,
                        "match": match.group()
                    })
        
        return issues
    
    def _calculate_score(self, issues: List[Dict], strict_mode: bool) -> int:
        """计算代码评分"""
        base_score = 100
        
        severity_weights = {
            "high": 20 if strict_mode else 15,
            "medium": 10 if strict_mode else 5,
            "warning": 5 if strict_mode else 3,
            "info": 2 if strict_mode else 1
        }
        
        for issue in issues:
            weight = severity_weights.get(issue["severity"], 1)
            base_score -= weight
        
        return max(0, min(100, base_score))
    
    def _generate_recommendations(self, issues: List[Dict]) -> List[str]:
        """生成改进建议"""
        recommendations = []
        
        categories_with_issues = set(issue["category"] for issue in issues)
        
        if "security" in categories_with_issues:
            recommendations.append("Review security concerns identified above")
            recommendations.append("Consider using secure alternatives (e.g., getpass instead of hardcoded passwords)")
        
        if "performance" in categories_with_issues:
            recommendations.append("Optimize identified performance bottlenecks")
            recommendations.append("Consider using list comprehensions instead of loops where applicable")
        
        if not recommendations:
            recommendations.append("Code quality looks good!")
            recommendations.append("Continue following best practices")
        
        return recommendations


class AgentNetwork:
    """
    Agent 网络 - 管理多个 Agent 的通信和协作
    
    实现 MCP (Model Context Protocol) 兼容的 Agent 间通信
    """
    
    def __init__(self):
        self.agents: Dict[str, BaseAgent] = {}
        self.message_log: List[AgentMessage] = []
        self._lock = asyncio.Lock()
    
    def register_agent(self, agent: BaseAgent) -> None:
        """注册 Agent"""
        self.agents[agent.agent_id] = agent
        print(f"[AgentNetwork] 注册 Agent: {agent.name} ({agent.agent_id})")
    
    async def send_message(self, message: AgentMessage) -> Optional[AgentMessage]:
        """发送消息（Agent 间通信）"""
        async with self._lock:
            self.message_log.append(message)
        
        if message.receiver:
            # 点对点消息
            receiver = self.agents.get(message.receiver)
            if receiver:
                return await receiver.process_message(message)
        
        return None
    
    async def broadcast(
        self, 
        sender: str, 
        content: Dict[str, Any]
    ) -> List[AgentMessage]:
        """广播消息"""
        responses = []
        
        for agent_id, agent in self.agents.items():
            if agent_id != sender:
                message = AgentMessage(
                    sender=sender,
                    receiver=agent_id,
                    content=content
                )
                response = await self.send_message(message)
                if response:
                    responses.append(response)
        
        return responses
    
    async def orchestrate(
        self, 
        task: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        编排多个 Agent 协作完成任务
        
        这是 Dynamic Workflows 的核心编排逻辑
        """
        results = {}
        
        # 根据任务类型选择 Agent
        task_type = task.get("type", "code_generation")
        
        if task_type == "code_generation":
            # 代码生成任务：生成 -> 审查 -> 测试 -> 文档
            gen_agent = self.agents.get("code_gen_001")
            review_agent = self.agents.get("code_review_001")
            
            # 1. 代码生成
            gen_task = {
                "id": "task_1",
                "language": task.get("language", "python"),
                "spec": task.get("spec", {}),
                "context": task.get("context", {})
            }
            
            if gen_agent:
                gen_result = await gen_agent.execute_task(gen_task)
                results["generated_code"] = gen_result
                
                # 2. 代码审查
                if review_agent:
                    review_task = {
                        "code": gen_result.get("code", ""),
                        "language": task.get("language", "python"),
                        "strict_mode": task.get("strict_mode", False)
                    }
                    review_result = await review_agent.execute_task(review_task)
                    results["review"] = review_result
        
        return results
    
    def get_network_status(self) -> Dict[str, Any]:
        """获取网络状态"""
        return {
            "total_agents": len(self.agents),
            "agents": [
                {
                    "id": agent.agent_id,
                    "name": agent.name,
                    "status": agent.state.status,
                    "capabilities": [c.value for c in agent.capabilities]
                }
                for agent in self.agents.values()
            ],
            "messages_logged": len(self.message_log)
        }


# 使用示例
async def demo_multi_agent_system():
    """演示多 Agent 协作系统"""
    
    print("=" * 60)
    print("演示：多 Agent 协作系统")
    print("=" * 60)
    
    # 创建 Agent 网络
    network = AgentNetwork()
    
    # 创建专业 Agent
    code_gen_agent = CodeGenerationAgent("code_gen_001")
    code_review_agent = CodeReviewAgent("code_review_001")
    
    # 初始化 Agent
    await code_gen_agent.initialize()
    await code_review_agent.initialize()
    
    # 注册到网络
    network.register_agent(code_gen_agent)
    network.register_agent(code_review_agent)
    
    # 演示任务编排
    print("\n📋 执行任务编排...")
    
    orchestrated_result = await network.orchestrate({
        "type": "code_generation",
        "language": "python",
        "spec": {"class": "DataProcessor"},
        "context": {
            "class_name": "DataProcessor",
            "features": ["async", "type_hints", "dataclass"]
        },
        "strict_mode": True
    })
    
    print("\n📊 编排结果:")
    print(json.dumps(orchestrated_result, indent=2, ensure_ascii=False))
    
    # 网络状态
    print("\n🌐 网络状态:")
    status = network.get_network_status()
    print(json.dumps(status, indent=2, ensure_ascii=False))


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

"""
Claude Opus 4.8 Dynamic Workflows - 任务分解器实现
Task Decomposer for Complex Multi-Step Tasks

核心功能：
1. 复杂任务的智能分解
2. 依赖关系分析
3. 任务优先级计算
4. 增量分解（支持大任务分批处理）
"""

import asyncio
import hashlib
import json
import re
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional, Set, Tuple
from collections import defaultdict


class TaskType(Enum):
    """任务类型枚举"""
    DEPENDENCY_ANALYSIS = "dependency_analysis"
    BREAKING_CHANGES_SCAN = "breaking_changes_scan"
    CODE_GENERATION = "code_generation"
    CODE_REVIEW = "code_review"
    TEST_GENERATION = "test_generation"
    MIGRATION = "migration"
    REFACTORING = "refactoring"
    DOCUMENTATION = "documentation"
    DEPLOYMENT = "deployment"


class TaskGranularity(Enum):
    """任务粒度枚举"""
    FILE = "file"           # 文件级别
    CLASS = "class"         # 类级别
    FUNCTION = "function"   # 函数级别
    STATEMENT = "statement" # 语句级别


@dataclass
class DecomposedTask:
    """分解后的子任务"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    name: str = ""
    description: str = ""
    task_type: TaskType = TaskType.CODE_GENERATION
    granularity: TaskGranularity = TaskGranularity.FILE
    target: str = ""  # 目标文件/类/函数
    payload: Dict[str, Any] = field(default_factory=dict)
    dependencies: Set[str] = field(default_factory=set)
    estimated_tokens: int = 0
    priority: int = 5  # 1-10, 1 is highest
    tags: Set[str] = field(default_factory=set)
    parent_id: Optional[str] = None
    
    def to_task_dict(self) -> Dict[str, Any]:
        """转换为任务字典"""
        return {
            "id": self.id,
            "name": self.name,
            "description": self.description,
            "task_type": self.task_type.value,
            "granularity": self.granularity.value,
            "target": self.target,
            "payload": self.payload,
            "dependencies": list(self.dependencies),
            "estimated_tokens": self.estimated_tokens,
            "priority": self.priority,
            "tags": list(self.tags),
            "parent_id": self.parent_id
        }


class CodebaseAnalyzer:
    """
    代码库分析器 - 辅助任务分解的代码库理解
    """
    
    def __init__(self):
        self.file_structure: Dict[str, Dict] = {}
        self.dependencies: Dict[str, Set[str]] = defaultdict(set)
        self.symbol_map: Dict[str, Dict] = {}  # symbol -> location
    
    def parse_file_structure(self, files: List[str]) -> Dict[str, Any]:
        """解析项目文件结构"""
        structure = {
            "modules": [],
            "models": [],
            "views": [],
            "services": [],
            "utils": []
        }
        
        for file_path in files:
            if "models" in file_path:
                structure["models"].append(file_path)
            elif "views" in file_path or "controllers" in file_path:
                structure["views"].append(file_path)
            elif "services" in file_path:
                structure["services"].append(file_path)
            elif "utils" in file_path or "helpers" in file_path:
                structure["utils"].append(file_path)
            else:
                structure["modules"].append(file_path)
        
        self.file_structure = structure
        return structure
    
    def extract_imports(self, content: str) -> Set[str]:
        """提取 import 语句"""
        # Python imports
        python_imports = set(re.findall(
            r'^(?:from\s+([\w.]+)\s+import|import\s+([\w.]+))',
            content,
            re.MULTILINE
        ))
        
        # Go imports
        go_imports = set(re.findall(
            r'"([^"]+)"',
            content.split('package')[1].split('func')[0] if 'func' in content else ''
        ))
        
        return python_imports | go_imports
    
    def identify_symbols(self, content: str, file_path: str) -> List[Dict]:
        """识别代码中的符号（类、函数、变量）"""
        symbols = []
        
        # Python class definitions
        for match in re.finditer(r'^class\s+(\w+)', content, re.MULTILINE):
            symbols.append({
                "name": match.group(1),
                "type": "class",
                "line": content[:match.start()].count('\n') + 1,
                "file": file_path
            })
        
        # Python function definitions
        for match in re.finditer(r'^def\s+(\w+)', content, re.MULTILINE):
            symbols.append({
                "name": match.group(1),
                "type": "function",
                "line": content[:match.start()].count('\n') + 1,
                "file": file_path
            })
        
        return symbols


class TaskDecomposer:
    """
    任务分解器 - 将复杂任务分解为可执行的子任务
    
    支持多种分解策略：
    1. 基于代码结构的分解
    2. 基于依赖关系的分解
    3. 基于功能模块的分解
    4. 增量分解（分批处理大任务）
    """
    
    def __init__(self, max_tokens_per_task: int = 50000):
        self.max_tokens_per_task = max_tokens_per_task
        self.analyzer = CodebaseAnalyzer()
        self.decomposition_cache: Dict[str, List[DecomposedTask]] = {}
    
    def decompose(
        self,
        task_description: str,
        context: Dict[str, Any],
        strategy: str = "auto"
    ) -> List[DecomposedTask]:
        """
        分解复杂任务
        
        Args:
            task_description: 任务描述
            context: 任务上下文（包含代码库信息等）
            strategy: 分解策略 ("auto", "by_structure", "by_dependencies", "incremental")
            
        Returns:
            分解后的子任务列表
        """
        # 检查缓存
        cache_key = self._generate_cache_key(task_description, context)
        if cache_key in self.decomposition_cache:
            return self.decomposition_cache[cache_key]
        
        # 根据任务类型选择分解策略
        task_type = self._infer_task_type(task_description)
        
        if strategy == "auto":
            strategy = self._select_strategy(task_type, context)
        
        if strategy == "by_structure":
            tasks = self._decompose_by_structure(task_description, context)
        elif strategy == "by_dependencies":
            tasks = self._decompose_by_dependencies(task_description, context)
        elif strategy == "incremental":
            tasks = self._decompose_incrementally(task_description, context)
        else:
            tasks = self._decompose_auto(task_description, context)
        
        # 排序和优化
        tasks = self._optimize_task_order(tasks)
        
        # 缓存结果
        self.decomposition_cache[cache_key] = tasks
        
        return tasks
    
    def _infer_task_type(self, description: str) -> TaskType:
        """推断任务类型"""
        description_lower = description.lower()
        
        if "migration" in description_lower or "migrate" in description_lower:
            return TaskType.MIGRATION
        elif "refactor" in description_lower or "restructure" in description_lower:
            return TaskType.REFACTORING
        elif "test" in description_lower:
            return TaskType.TEST_GENERATION
        elif "review" in description_lower or "audit" in description_lower:
            return TaskType.CODE_REVIEW
        elif "document" in description_lower or "doc" in description_lower:
            return TaskType.DOCUMENTATION
        elif "deploy" in description_lower:
            return TaskType.DEPLOYMENT
        else:
            return TaskType.CODE_GENERATION
    
    def _select_strategy(self, task_type: TaskType, context: Dict) -> str:
        """选择分解策略"""
        codebase_size = context.get("codebase_size", "small")
        
        if task_type == TaskType.MIGRATION:
            return "by_dependencies"
        elif codebase_size == "large":
            return "incremental"
        else:
            return "auto"
    
    def _decompose_auto(
        self,
        task_description: str,
        context: Dict[str, Any]
    ) -> List[DecomposedTask]:
        """自动分解策略"""
        tasks = []
        task_type = self._infer_task_type(task_description)
        
        # 1. 添加依赖分析任务
        tasks.append(DecomposedTask(
            name="分析项目依赖",
            description="扫描项目依赖关系和代码结构",
            task_type=TaskType.DEPENDENCY_ANALYSIS,
            granularity=TaskGranularity.FILE,
            priority=1,
            tags={"setup", "analysis"}
        ))
        
        # 2. 添加 breaking changes 扫描（如果是迁移任务）
        if task_type == TaskType.MIGRATION:
            tasks.append(DecomposedTask(
                name="扫描 Breaking Changes",
                description="识别迁移过程中可能出现的兼容性问题",
                task_type=TaskType.BREAKING_CHANGES_SCAN,
                dependencies={tasks[0].id},
                priority=2,
                tags={"migration", "analysis"}
            ))
        
        # 3. 基于文件结构分解主任务
        files = context.get("files", [])
        if files:
            file_tasks = self._decompose_by_files(
                files, 
                task_description, 
                task_type,
                deps=[tasks[0].id]
            )
            tasks.extend(file_tasks)
        
        # 4. 添加测试任务
        test_task = DecomposedTask(
            name="生成测试用例",
            description="为修改的代码生成单元测试和集成测试",
            task_type=TaskType.TEST_GENERATION,
            dependencies={tasks[-1].id} if tasks else set(),
            priority=8,
            tags={"testing", "quality"}
        )
        tasks.append(test_task)
        
        # 5. 添加文档更新任务
        doc_task = DecomposedTask(
            name="更新文档",
            description="更新相关 API 文档和 README",
            task_type=TaskType.DOCUMENTATION,
            dependencies={tasks[-2].id},
            priority=9,
            tags={"documentation"}
        )
        tasks.append(doc_task)
        
        # 6. 添加最终审查任务
        review_task = DecomposedTask(
            name="最终代码审查",
            description="审查所有修改，确保代码质量和一致性",
            task_type=TaskType.CODE_REVIEW,
            dependencies={tasks[-1].id, tasks[-2].id},
            priority=10,
            tags={"review", "quality"}
        )
        tasks.append(review_task)
        
        return tasks
    
    def _decompose_by_structure(
        self,
        task_description: str,
        context: Dict[str, Any]
    ) -> List[DecomposedTask]:
        """基于代码结构的分解"""
        tasks = []
        structure = context.get("structure", {})
        
        # 按模块分解
        for module_name, module_data in structure.items():
            module_task = DecomposedTask(
                name=f"处理模块: {module_name}",
                description=f"处理 {module_name} 模块的所有文件",
                task_type=TaskType.MIGRATION if "migration" in task_description.lower() else TaskType.CODE_GENERATION,
                granularity=TaskGranularity.FILE,
                target=module_name,
                payload={"module": module_data},
                priority=5,
                tags={"module", module_name}
            )
            tasks.append(module_task)
        
        return tasks
    
    def _decompose_by_dependencies(
        self,
        task_description: str,
        context: Dict[str, Any]
    ) -> List[DecomposedTask]:
        """基于依赖关系的分解"""
        tasks = []
        dependencies = context.get("dependencies", {})
        
        # 使用拓扑排序确定处理顺序
        processed = set()
        
        def process_dependencies(dep_name: str, level: int = 0):
            if dep_name in processed:
                return
            
            dep_info = dependencies.get(dep_name, {})
            dep_deps = dep_info.get("depends_on", [])
            
            # 先处理依赖
            for sub_dep in dep_deps:
                process_dependencies(sub_dep, level + 1)
            
            # 创建任务
            task = DecomposedTask(
                name=f"迁移依赖: {dep_name}",
                description=f"处理 {dep_name} 及其依赖项",
                task_type=TaskType.MIGRATION,
                target=dep_name,
                payload={"dependency": dep_info},
                priority=max(1, 10 - level),
                tags={"dependency", dep_name, f"level_{level}"}
            )
            
            if dep_deps:
                task.dependencies = {f"dep_{d}" for d in dep_deps}
            
            tasks.append(task)
            processed.add(dep_name)
        
        # 处理所有顶层依赖
        for dep_name in dependencies:
            process_dependencies(dep_name)
        
        return tasks
    
    def _decompose_incrementally(
        self,
        task_description: str,
        context: Dict[str, Any]
    ) -> List[DecomposedTask]:
        """增量分解 - 分批处理大任务"""
        tasks = []
        files = context.get("files", [])
        batch_size = context.get("batch_size", 10)
        
        # 计算总批次数
        total_batches = (len(files) + batch_size - 1) // batch_size
        
        for i in range(0, len(files), batch_size):
            batch_files = files[i:i + batch_size]
            batch_num = i // batch_size + 1
            
            batch_task = DecomposedTask(
                name=f"批处理 {batch_num}/{total_batches}",
                description=f"处理 {len(batch_files)} 个文件 ({batch_num}/{total_batches})",
                task_type=TaskType.MIGRATION,
                granularity=TaskGranularity.FILE,
                payload={"files": batch_files, "batch_num": batch_num},
                priority=5,
                tags={"batch", f"batch_{batch_num}", f"of_{total_batches}"}
            )
            
            # 批次间有依赖关系
            if tasks:
                batch_task.dependencies = {tasks[-1].id}
            
            tasks.append(batch_task)
        
        return tasks
    
    def _decompose_by_files(
        self,
        files: List[str],
        task_description: str,
        task_type: TaskType,
        deps: List[str] = None
    ) -> List[DecomposedTask]:
        """基于文件列表分解"""
        tasks = []
        deps = deps or []
        
        for idx, file_path in enumerate(files):
            file_task = DecomposedTask(
                name=f"处理文件: {file_path}",
                description=f"处理 {file_path} 文件的迁移/修改",
                task_type=task_type,
                granularity=TaskGranularity.FILE,
                target=file_path,
                payload={"file_path": file_path},
                priority=3 + idx * 0.1,  # 保持大致相同的优先级
                tags={"file", "source"}
            )
            
            # 第一个文件依赖于分析任务
            if idx == 0 and deps:
                file_task.dependencies = set(deps)
            
            tasks.append(file_task)
        
        return tasks
    
    def _optimize_task_order(
        self, 
        tasks: List[DecomposedTask]
    ) -> List[DecomposedTask]:
        """优化任务顺序"""
        # 构建依赖图
        task_map = {t.id: t for t in tasks}
        in_degree = {t.id: 0 for t in tasks}
        
        for task in tasks:
            for dep_id in task.dependencies:
                if dep_id in in_degree:
                    in_degree[task.id] += 1
        
        # 拓扑排序
        result = []
        queue = [(t.priority, t.id) for t in tasks if in_degree[t.id] == 0]
        heapq.heapify(queue)
        
        while queue:
            _, task_id = heapq.heappop(queue)
            task = task_map[task_id]
            result.append(task)
            
            # 更新依赖计数
            for other_task in tasks:
                if task_id in other_task.dependencies:
                    in_degree[other_task.id] -= 1
                    if in_degree[other_task.id] == 0:
                        heapq.heappush(queue, (other_task.priority, other_task.id))
        
        return result
    
    def _generate_cache_key(
        self, 
        description: str, 
        context: Dict
    ) -> str:
        """生成缓存键"""
        content = json.dumps({"desc": description, "ctx": context}, sort_keys=True)
        return hashlib.md5(content.encode()).hexdigest()
    
    def estimate_total_tokens(self, tasks: List[DecomposedTask]) -> int:
        """估算总 Token 数量"""
        return sum(task.estimated_tokens for task in tasks)
    
    def split_large_task(
        self, 
        task: DecomposedTask, 
        max_tokens: int = 50000
    ) -> List[DecomposedTask]:
        """拆分过大的任务"""
        if task.estimated_tokens <= max_tokens:
            return [task]
        
        subtasks = []
        subtask_count = (task.estimated_tokens + max_tokens - 1) // max_tokens
        
        for i in range(subtask_count):
            subtask = DecomposedTask(
                name=f"{task.name} (Part {i+1}/{subtask_count})",
                description=f"{task.description} - 第 {i+1} 部分",
                task_type=task.task_type,
                granularity=task.granularity,
                target=task.target,
                payload={**task.payload, "part": i + 1, "total_parts": subtask_count},
                priority=task.priority,
                tags=task.tags | {f"part_{i+1}"},
                parent_id=task.id
            )
            
            if i > 0:
                subtask.dependencies = {subtasks[-1].id}
            
            subtasks.append(subtask)
        
        return subtasks


# 使用示例
def demo_task_decomposer():
    """演示任务分解器"""
    
    print("=" * 60)
    print("演示：智能任务分解器")
    print("=" * 60)
    
    # 创建分解器
    decomposer = TaskDecomposer(max_tokens_per_task=50000)
    
    # 模拟代码库上下文
    context = {
        "codebase_size": "large",
        "project_type": "django",
        "target_version": "5.0",
        "files": [
            "models/user.py",
            "models/post.py",
            "models/comment.py",
            "views/api.py",
            "views/admin.py",
            "services/auth.py",
            "services/notification.py",
            "utils/helpers.py",
        ],
        "dependencies": {
            "django": {"version": "3.2", "target": "5.0", "depends_on": []},
            "djangorestframework": {"version": "3.12", "target": "3.15", "depends_on": ["django"]},
            "celery": {"version": "4.4", "target": "5.3", "depends_on": []},
        },
        "batch_size": 3
    }
    
    # 演示1：自动分解
    print("\n📋 策略1：自动分解")
    print("-" * 40)
    
    task_description = "将 Django 项目从 3.2 迁移到 5.0"
    tasks = decomposer.decompose(task_description, context, strategy="auto")
    
    print(f"分解为 {len(tasks)} 个子任务:")
    for i, task in enumerate(tasks, 1):
        deps_str = f" (依赖: {', '.join(list(task.dependencies)[:2])})" if task.dependencies else ""
        print(f"  {i}. [{task.task_type.value}] {task.name}{deps_str}")
        print(f"     描述: {task.description}")
    
    # 演示2：基于依赖分解
    print("\n\n📋 策略2：基于依赖分解")
    print("-" * 40)
    
    tasks_by_deps = decomposer.decompose(task_description, context, strategy="by_dependencies")
    
    print(f"分解为 {len(tasks_by_deps)} 个子任务:")
    for i, task in enumerate(tasks_by_deps, 1):
        deps_str = f" (依赖: {', '.join(list(task.dependencies))})" if task.dependencies else ""
        print(f"  {i}. {task.name}{deps_str}")
    
    # 演示3：增量分解
    print("\n\n📋 策略3：增量分解（大批次）")
    print("-" * 40)
    
    tasks_incremental = decomposer.decompose(task_description, context, strategy="incremental")
    
    print(f"分解为 {len(tasks_incremental)} 个批次:")
    for i, task in enumerate(tasks_incremental, 1):
        batch_info = task.payload.get("batch_num", "N/A")
        deps_str = f" <- 依赖于上一批次" if task.dependencies else ""
        print(f"  {i}. {task.name} (批次 {batch_info}){deps_str}")
    
    # 演示4：优化后的任务顺序
    print("\n\n📋 优化后的执行顺序")
    print("-" * 40)
    
    optimized_tasks = decomposer._optimize_task_order(tasks)
    
    for i, task in enumerate(optimized_tasks, 1):
        deps_str = f" (依赖: {list(task.dependencies)})" if task.dependencies else ""
        print(f"  Step {i}: {task.name}{deps_str}")
    
    # 演示5：生成执行计划
    print("\n\n📊 完整执行计划")
    print("-" * 40)
    
    for task in optimized_tasks:
        task_dict = task.to_task_dict()
        print(f"""
任务ID: {task_dict['id']}
名称: {task_dict['name']}
类型: {task_dict['task_type']}
目标: {task_dict['target']}
依赖: {task_dict['dependencies']}
优先级: {task_dict['priority']}
标签: {', '.join(task_dict['tags'])}
        """)
    
    return tasks


if __name__ == "__main__":
    import heapq  # 确保可用
    demo_task_decomposer()