# RLVR核心原理示例
class RLVRTraining:
    """
    可验证奖励的强化学习
    核心思想：用自动化验证替代人工标注
    """
    
    def __init__(self, model, verifier, task_type="code"):
        self.model = model
        self.verifier = verifier  # 验证器：代码执行、数学判卷等
        self.task_type = task_type
    
    def generate_and_evaluate(self, prompt):
        """生成响应并获取可验证奖励"""
        response = self.model.generate(prompt)
        
        if self.task_type == "code":
            # 代码任务：运行测试用例
            reward = self.run_code_tests(response, prompt)
        elif self.task_type == "math":
            # 数学任务：与标准答案比对
            reward = self.check_math_answer(response, prompt)
        else:
            # 其他可验证任务
            reward = self.verifier.verify(response, prompt)
        
        return response, reward
    
    def run_code_tests(self, code, test_cases):
        """执行代码并验证测试用例"""
        try:
            # 动态执行生成的代码
            result = execute_sandbox(code)
            
            # 对每个测试用例进行验证
            passed = 0
            for test_input, expected_output in test_cases:
                actual = result.run(test_input)
                if actual == expected_output:
                    passed += 1
            
            # 返回通过率作为奖励
            return passed / len(test_cases)
        except Exception:
            return 0.0
    
    def check_math_answer(self, solution, problem):
        """验证数学解答"""
        try:
            # 解析模型生成的解答
            answer = extract_answer(solution)
            # 与标准答案比对
            return 1.0 if answer == problem.answer else 0.0
        except:
            return 0.0

# 从"刷题选手"到"职场打工人"的进化示例
class AIRoleEvolution:
    """AI角色从竞赛选手到职场员工的进化"""
    
    # 旧范式：竞赛选手
    @staticmethod
    def competition_mode(prompt: str) -> str:
        """
        竞赛模式特点：
        - 单一正确答案
        - 有限上下文
        - 瞬时响应
        """
        # 直接返回最佳答案
        return "42"  # Life, the Universe, and Everything
    
    # 新范式：职场员工
    @staticmethod
    def work_mode(project: "Project") -> "WorkResult":
        """
        职场模式特点：
        - 多目标优化
        - 长期上下文
        - 持续迭代
        - 团队协作
        """
        # 需要理解项目背景
        context = project.load_context()
        
        # 需要与相关方沟通
        stakeholders = project.get_stakeholders()
        requirements = []
        for stakeholder in stakeholders:
            requirements.append(stakeholder.gather_requirements())
        
        # 需要处理模糊性
        ambiguous_points = project.identify_ambiguities()
        clarifications = project.request_clarifications(ambiguous_points)
        
        # 需要持续迭代优化
        iterations = 0
        max_iterations = 10
        while not project.meets_criteria() and iterations < max_iterations:
            solution = project.implement_solution(requirements)
            feedback = project.get_feedback(solution)
            project.refine(solution, feedback)
            iterations += 1
        
        # 需要考虑非功能性需求
        result = project.finalize_solution()
        return WorkResult(
            deliverables=result,
            documentation=project.generate_docs(),
            tests=project.generate_tests(),
            deployment_plan=project.create_deployment_plan()
        )

#!/usr/bin/env python3
"""
Multi-Agent Code Generation System
递归AI自我改进系统的核心组件

功能：
1. Planning Agent - 任务规划与分解
2. Code Agent - 代码编写与优化
3. Review Agent - 代码审查与测试
4. Orchestrator - 智能体协调器

作者：AI Research Team
日期：2026-06-07
"""

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


# ==================== 核心数据模型 ====================

class TaskStatus(Enum):
    """任务状态枚举"""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"


class Priority(Enum):
    """优先级枚举"""
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4


@dataclass
class Task:
    """任务数据结构"""
    id: str
    description: str
    status: TaskStatus = TaskStatus.PENDING
    priority: Priority = Priority.MEDIUM
    dependencies: List[str] = field(default_factory=list)
    assigned_agent: Optional[str] = None
    result: Optional[Any] = None
    created_at: float = field(default_factory=time.time)
    updated_at: float = field(default_factory=time.time)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict:
        return {
            "id": self.id,
            "description": self.description,
            "status": self.status.value,
            "priority": self.priority.name,
            "dependencies": self.dependencies,
            "assigned_agent": self.assigned_agent,
            "result": self.result,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "metadata": self.metadata
        }


@dataclass
class CodeGenerationRequest:
    """代码生成请求"""
    project_name: str
    description: str
    language: str
    requirements: List[str]
    constraints: Dict[str, Any] = field(default_factory=dict)
    quality_bar: float = 0.8  # 质量门槛


@dataclass
class GeneratedCode:
    """生成的代码"""
    task_id: str
    code: str
    language: str
    tests: List[str]
    documentation: str
    quality_score: float
    generation_time: float
    iterations: int


# ==================== AI智能体基类 ====================

class BaseAgent(ABC):
    """AI智能体基类"""
    
    def __init__(self, name: str, model_name: str = "claude-3-5-sonnet"):
        self.name = name
        self.model_name = model_name
        self.conversation_history: List[Dict] = []
        
    @abstractmethod
    async def process(self, input_data: Any) -> Any:
        """处理输入数据"""
        pass
    
    def add_to_history(self, role: str, content: str):
        """添加对话历史"""
        self.conversation_history.append({
            "role": role,
            "content": content,
            "timestamp": time.time()
        })
    
    def clear_history(self):
        """清空对话历史"""
        self.conversation_history = []


# ==================== Planning Agent ====================

class PlanningAgent(BaseAgent):
    """
    任务规划智能体
    负责：
    - 理解需求
    - 分解任务
    - 制定执行计划
    - 评估风险
    """
    
    def __init__(self, model_name: str = "claude-3-5-sonnet"):
        super().__init__("Planning Agent", model_name)
        self.planning_template = """分析以下需求，制定详细执行计划：

需求：{description}
语言：{language}
要求：{requirements}
约束：{constraints}

请输出：
1. 任务分解（子任务列表）
2. 执行顺序
3. 关键里程碑
4. 潜在风险
"""
    
    async def process(self, request: CodeGenerationRequest) -> List[Task]:
        """将需求分解为可执行的任务"""
        self.add_to_history("system", "You are a senior software architect.")
        
        # 模拟AI分析过程
        prompt = self.planning_template.format(
            description=request.description,
            language=request.language,
            requirements="\n".join(f"- {r}" for r in request.requirements),
            constraints=json.dumps(request.constraints, indent=2)
        )
        
        self.add_to_history("user", prompt)
        
        # 模拟AI生成的规划
        plan = self._generate_plan(request)
        
        self.add_to_history("assistant", json.dumps(plan, indent=2))
        
        # 转换为任务列表
        tasks = []
        for i, step in enumerate(plan["steps"]):
            task = Task(
                id=f"task_{request.project_name}_{i+1}",
                description=step["description"],
                priority=Priority[step.get("priority", "MEDIUM")],
                dependencies=step.get("depends_on", []),
                metadata={
                    "project": request.project_name,
                    "step_number": i + 1,
                    "estimated_time": step.get("estimated_minutes", 30)
                }
            )
            tasks.append(task)
        
        return tasks
    
    def _generate_plan(self, request: CodeGenerationRequest) -> Dict:
        """生成执行计划"""
        steps = []
        
        # 分析语言和需求类型
        if request.language in ["python", "go", "rust"]:
            steps.extend([
                {"description": "设计项目结构和模块划分", "priority": "HIGH"},
                {"description": "实现核心数据结构和接口", "priority": "HIGH", "depends_on": []},
                {"description": "编写核心业务逻辑", "priority": "HIGH", "depends_on": [1]},
                {"description": "实现错误处理和日志", "priority": "MEDIUM", "depends_on": [2]},
                {"description": "编写单元测试", "priority": "HIGH", "depends_on": [2]},
                {"description": "编写集成测试", "priority": "MEDIUM", "depends_on": [3, 4]},
                {"description": "生成API文档和使用说明", "priority": "LOW", "depends_on": [3]},
                {"description": "性能优化和代码审查", "priority": "MEDIUM", "depends_on": [5]},
            ])
        else:
            steps.append({"description": "通用代码生成", "priority": "HIGH"})
        
        return {
            "project": request.project_name,
            "total_steps": len(steps),
            "estimated_minutes": sum(s.get("estimated_minutes", 30) for s in steps),
            "steps": steps
        }


# ==================== Code Agent ====================

class CodeAgent(BaseAgent):
    """
    代码生成智能体
    负责：
    - 生成代码
    - 实现功能
    - 代码优化
    """
    
    def __init__(self, model_name: str = "claude-3-5-sonnet"):
        super().__init__("Code Agent", model_name)
        self.code_templates = self._load_code_templates()
    
    def _load_code_templates(self) -> Dict[str, str]:
        """加载代码模板"""
        return {
            "python_class": '''class {class_name}:
    """Auto-generated class"""
    
    def __init__(self{params}):
{init_body}
    
    def __repr__(self):
        return f"{class_name}({self._repr_attrs})"
''',
            "go_struct": '''type {struct_name} struct {{
{fields}
}}

func New{struct_name}({params}) *{struct_name} {{
    return &{struct_name}{{
{assignments}
    }}
}}
''',
        }
    
    async def process(self, task: Task) -> str:
        """根据任务生成代码"""
        self.add_to_history("system", 
            "You are an expert programmer. Write clean, efficient, well-documented code.")
        
        prompt = f"""Task: {task.description}
Project: {task.metadata.get('project', 'Unknown')}
Language: {task.metadata.get('language', 'python')}

Generate the code for this task following best practices:
1. Clean, readable code
2. Proper error handling
3. Type hints (if applicable)
4. Docstrings/comments
5. Unit tests

Output only the code block."""
        
        self.add_to_history("user", prompt)
        
        # 模拟代码生成
        code = self._generate_code(task)
        
        self.add_to_history("assistant", f"```\n{code}\n```")
        
        return code
    
    def _generate_code(self, task: Task) -> str:
        """根据任务类型生成代码"""
        project = task.metadata.get("project", "sample")
        language = task.metadata.get("language", "python")
        
        # 基于任务描述生成代码
        if "data" in task.description.lower():
            return self._generate_data_model(project, language)
        elif "api" in task.description.lower() or "endpoint" in task.description.lower():
            return self._generate_api_endpoint(project, language)
        elif "test" in task.description.lower():
            return self._generate_tests(project, language)
        else:
            return self._generate_utility_code(project, language)
    
    def _generate_data_model(self, project: str, language: str) -> str:
        """生成数据模型"""
        if language == "python":
            return f'''"""
Data Models for {project}
Auto-generated by AI Code Agent
"""

from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from datetime import datetime
from enum import Enum


class TaskStatus(Enum):
    """任务状态枚举"""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class Task:
    """任务数据模型"""
    id: str
    title: str
    description: str
    status: TaskStatus = TaskStatus.PENDING
    priority: int = 1
    assignee: Optional[str] = None
    tags: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)
    
    def is_complete(self) -> bool:
        """检查任务是否完成"""
        return self.status == TaskStatus.COMPLETED
    
    def mark_complete(self) -> None:
        """标记为完成"""
        self.status = TaskStatus.COMPLETED
        self.updated_at = datetime.now()
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return {{
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "status": self.status.value,
            "priority": self.priority,
            "assignee": self.assignee,
            "tags": self.tags,
            "metadata": self.metadata,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat()
        }}


@dataclass
class Project:
    """项目数据模型"""
    id: str
    name: str
    description: str
    tasks: List[Task] = field(default_factory=list)
    owner: str = ""
    
    def add_task(self, task: Task) -> None:
        """添加任务"""
        self.tasks.append(task)
    
    def get_completion_rate(self) -> float:
        """计算完成率"""
        if not self.tasks:
            return 0.0
        completed = sum(1 for t in self.tasks if t.is_complete())
        return completed / len(self.tasks)
    
    def get_pending_tasks(self) -> List[Task]:
        """获取待处理任务"""
        return [t for t in self.tasks if t.status == TaskStatus.PENDING]
'''
        elif language == "go":
            return f'''package models

import "time"

// TaskStatus represents the status of a task
type TaskStatus string

const (
    StatusPending    TaskStatus = "pending"
    StatusInProgress TaskStatus = "in_progress"
    StatusCompleted TaskStatus = "completed"
    StatusFailed    TaskStatus = "failed"
)

// Task represents a task in the system
type Task struct {{
    ID          string            `json:"id"`
    Title       string            `json:"title"`
    Description string            `json:"description"`
    Status      TaskStatus        `json:"status"`
    Priority    int               `json:"priority"`
    Assignee    string            `json:"assignee,omitempty"`
    Tags        []string          `json:"tags,omitempty"`
    Metadata    map[string]interface{{}} `json:"metadata,omitempty"`
    CreatedAt   time.Time         `json:"created_at"`
    UpdatedAt   time.Time         `json:"updated_at"`
}}

// IsComplete checks if the task is completed
func (t *Task) IsComplete() bool {{
    return t.Status == StatusCompleted
}}

// MarkComplete marks the task as completed
func (t *Task) MarkComplete() {{
    t.Status = StatusCompleted
    t.UpdatedAt = time.Now()
}}

// Project represents a project containing tasks
type Project struct {{
    ID          string  `json:"id"`
    Name        string  `json:"name"`
    Description string  `json:"description"`
    Owner       string  `json:"owner"`
    Tasks       []*Task `json:"tasks,omitempty"`
}}

// AddTask adds a task to the project
func (p *Project) AddTask(task *Task) {{
    p.Tasks = append(p.Tasks, task)
}}

// GetCompletionRate calculates the completion rate
func (p *Project) GetCompletionRate() float64 {{
    if len(p.Tasks) == 0 {{
        return 0.0
    }}
    completed := 0
    for _, t := range p.Tasks {{
        if t.IsComplete() {{
            completed++
        }}
    }}
    return float64(completed) / float64(len(p.Tasks))
}}
'''
        return "# Generated code placeholder"
    
    def _generate_api_endpoint(self, project: str, language: str) -> str:
        """生成API端点"""
        if language == "python":
            return f'''"""
API Endpoints for {project}
FastAPI implementation
"""
from fastapi import APIRouter, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
import uuid

router = APIRouter(prefix="/api/v1/{project}", tags=["{project}"])


class TaskCreate(BaseModel):
    """创建任务请求"""
    title: str = Field(..., min_length=1, max_length=200)
    description: str = ""
    priority: int = Field(default=1, ge=1, le=5)
    tags: List[str] = []


class TaskResponse(BaseModel):
    """任务响应"""
    id: str
    title: str
    description: str
    status: str
    priority: int
    tags: List[str]
    created_at: datetime
    updated_at: datetime


class TaskUpdate(BaseModel):
    """更新任务请求"""
    title: Optional[str] = None
    description: Optional[str] = None
    status: Optional[str] = None
    priority: Optional[int] = None


# 模拟数据库
tasks_db: dict = {{}}


@router.post("/tasks", response_model=TaskResponse, status_code=201)
async def create_task(task_data: TaskCreate):
    """创建新任务"""
    task_id = str(uuid.uuid4())
    now = datetime.now()
    
    task = TaskResponse(
        id=task_id,
        title=task_data.title,
        description=task_data.description,
        status="pending",
        priority=task_data.priority,
        tags=task_data.tags,
        created_at=now,
        updated_at=now
    )
    
    tasks_db[task_id] = task
    return task


@router.get("/tasks", response_model=List[TaskResponse])
async def list_tasks(status: Optional[str] = None):
    """获取任务列表"""
    tasks = list(tasks_db.values())
    
    if status:
        tasks = [t for t in tasks if t.status == status]
    
    return sorted(tasks, key=lambda x: x.created_at, reverse=True)


@router.get("/tasks/{{task_id}}", response_model=TaskResponse)
async def get_task(task_id: str):
    """获取单个任务"""
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="Task not found")
    return tasks_db[task_id]


@router.patch("/tasks/{{task_id}}", response_model=TaskResponse)
async def update_task(task_id: str, update_data: TaskUpdate):
    """更新任务"""
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="Task not found")
    
    task = tasks_db[task_id]
    update_dict = update_data.dict(exclude_unset=True)
    
    for key, value in update_dict.items():
        if hasattr(task, key):
            setattr(task, key, value)
    
    task.updated_at = datetime.now()
    tasks_db[task_id] = task
    
    return task


@router.delete("/tasks/{{task_id}}", status_code=204)
async def delete_task(task_id: str):
    """删除任务"""
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="Task not found")
    
    del tasks_db[task_id]
    return None
'''
        return "# API endpoint placeholder"
    
    def _generate_tests(self, project: str, language: str) -> str:
        """生成测试代码"""
        if language == "python":
            return f'''"""
Tests for {project}
Generated by AI Code Agent
"""
import pytest
from datetime import datetime
from {project}.models import Task, Project, TaskStatus


class TestTask:
    """Task模型测试"""
    
    def test_task_creation(self):
        """测试任务创建"""
        task = Task(
            id="test-1",
            title="Test Task",
            description="A test task"
        )
        assert task.id == "test-1"
        assert task.title == "Test Task"
        assert task.status == TaskStatus.PENDING
    
    def test_task_completion(self):
        """测试任务完成"""
        task = Task(id="test-2", title="Task", description="")
        assert not task.is_complete()
        
        task.mark_complete()
        assert task.is_complete()
        assert task.status == TaskStatus.COMPLETED
    
    def test_task_to_dict(self):
        """测试转换为字典"""
        task = Task(id="test-3", title="Task", description="Desc")
        data = task.to_dict()
        
        assert data["id"] == "test-3"
        assert data["status"] == "pending"
        assert "created_at" in data


class TestProject:
    """Project模型测试"""
    
    def test_project_creation(self):
        """测试项目创建"""
        project = Project(
            id="proj-1",
            name="Test Project",
            description="A test project"
        )
        assert project.name == "Test Project"
        assert len(project.tasks) == 0
    
    def test_add_task(self):
        """测试添加任务"""
        project = Project(id="proj-2", name="P", description="")
        task = Task(id="t1", title="T", description="")
        
        project.add_task(task)
        assert len(project.tasks) == 1
    
    def test_completion_rate(self):
        """测试完成率计算"""
        project = Project(id="proj-3", name="P", description="")
        
        # 无任务时完成率为0
        assert project.get_completion_rate() == 0.0
        
        # 添加任务
        for i in range(4):
            task = Task(id=f"t{{i}}", title=f"T{{i}}", description="")
            if i < 2:  # 前两个完成
                task.mark_complete()
            project.add_task(task)
        
        assert project.get_completion_rate() == 0.5


if __name__ == "__main__":
    pytest.main([__file__, "-v"])
'''
        return "# Test placeholder"
    
    def _generate_utility_code(self, project: str, language: str) -> str:
        """生成工具代码"""
        if language == "python":
            return f'''"""
Utility functions for {project}
"""
from typing import List, Dict, Any, Optional
import hashlib
import json
from datetime import datetime


def generate_id(prefix: str = "") -> str:
    """生成唯一ID"""
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
    hash_obj = hashlib.md5(str(timestamp).encode())
    return f"{{prefix}}_{{timestamp}}_{{hash_obj.hexdigest()[:8]}}"


def serialize_to_json(obj: Any, indent: int = 2) -> str:
    """序列化为JSON"""
    return json.dumps(obj, indent=indent, default=str, ensure_ascii=False)


def deserialize_from_json(json_str: str) -> Any:
    """从JSON反序列化"""
    return json.loads(json_str)


def deep_merge(base: Dict, update: Dict) -> Dict:
    """深度合并字典"""
    result = base.copy()
    for key, value in update.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = deep_merge(result[key], value)
        else:
            result[key] = value
    return result


def batch_process(items: List[Any], processor: callable, batch_size: int = 100) -> List[Any]:
    """批量处理"""
    results = []
    for i in range(0, len(items), batch_size):
        batch = items[i:i + batch_size]
        results.extend([processor(item) for item in batch])
    return results


def retry_on_failure(func: callable, max_retries: int = 3, delay: float = 1.0):
    """失败重试装饰器"""
    def wrapper(*args, **kwargs):
        import time
        last_exception = None
        for attempt in range(max_retries):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                last_exception = e
                if attempt < max_retries - 1:
                    time.sleep(delay * (attempt + 1))
        raise last_exception
    return wrapper


__all__ = [
    "generate_id",
    "serialize_to_json",
    "deserialize_from_json",
    "deep_merge",
    "batch_process",
    "retry_on_failure"
]
'''
        return "# Utility placeholder"


# ==================== Review Agent ====================

class ReviewAgent(BaseAgent):
    """
    代码审查智能体
    负责：
    - 代码质量检查
    - 安全漏洞扫描
    - 风格一致性检查
    - 测试覆盖率验证
    """
    
    def __init__(self, model_name: str = "claude-3-5-sonnet"):
        super().__init__("Review Agent", model_name)
        self.quality_criteria = [
            "code_quality",
            "security",
            "performance",
            "test_coverage",
            "documentation"
        ]
    
    async def process(self, code: str, task: Task) -> Dict[str, Any]:
        """审查代码质量"""
        self.add_to_history("system", 
            "You are a senior code reviewer. Be thorough and constructive.")
        
        prompt = f"""Review the following code for task: {task.description}

Code:

from {request.project_name} import *

# Your code here