Create a Real Life Example of Creating AI Courses using Multiple Agents

from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
from agents import Agent, Runner, Session
import json
import os

class DifficultyLevel(Enum):
    BEGINNER = "beginner"
    INTERMEDIATE = "intermediate"
    ADVANCED = "advanced"

class ContentType(Enum):
    LESSON = "lesson"
    EXERCISE = "exercise"
    QUIZ = "quiz"
    PROJECT = "project"

@dataclass
class LearningObjective:
    description: str
    difficulty: DifficultyLevel
    estimated_time: int  # in minutes

@dataclass
class CourseModule:
    title: str
    description: str
    objectives: List[LearningObjective]
    content: Optional[str] = None
    assessments: Optional[List[Dict]] = None
    order: int = 0

@dataclass
class Course:
    title: str
    description: str
    target_audience: str
    prerequisites: List[str]
    modules: List[CourseModule]
    total_duration: int = 0
    difficulty: DifficultyLevel = DifficultyLevel.BEGINNER

def create_learning_objectives(topic: str, audience: str, count: int = 5) -> List[Dict]:
    """Create learning objectives for a course topic."""
    objectives = []
    # This would typically involve more sophisticated logic
    base_objectives = [
        f"Understand the fundamentals of {topic}",
        f"Apply {topic} concepts to real-world scenarios",
        f"Analyze different approaches in {topic}",
        f"Create projects using {topic}",
        f"Evaluate and optimize {topic} solutions"
    ]

    for i, obj in enumerate(base_objectives[:count]):
        objectives.append({
            "description": obj,
            "difficulty": ["beginner", "intermediate", "advanced"][i % 3],
            "estimated_time": 30 + (i * 15)
        })

    return objectives

def structure_course_modules(topic: str, objectives: List[Dict]) -> List[Dict]:
    """Structure course content into logical modules."""
    modules = []

    # Group objectives into modules (simplified logic)
    module_themes = [
        f"Introduction to {topic}",
        f"Core Concepts of {topic}",
        f"Advanced {topic} Techniques",
        f"Practical Applications",
        f"Best Practices and Optimization"
    ]

    for i, theme in enumerate(module_themes):
        module_objectives = objectives[i:i+2] if i < len(objectives) else []
        modules.append({
            "title": theme,
            "description": f"Learn about {theme.lower()}",
            "objectives": module_objectives,
            "order": i + 1
        })

    return modules

course_planner = Agent(
    name="CoursePlanner",
    instructions="""You are an expert instructional designer specializing in AI and technology courses.

    Your role:
    - Analyze course requirements and target audience
    - Create comprehensive learning objectives
    - Structure content into logical modules
    - Ensure proper learning progression
    - Consider different learning styles and pacing

    Always create courses that are:
    - Well-structured with clear progression
    - Appropriate for the target audience
    - Practical with hands-on components
    - Comprehensive but not overwhelming""",
    tools=[create_learning_objectives, structure_course_modules]
)

def research_topic_content(topic: str, depth: str = "comprehensive") -> str:
    """Research and compile information about a topic."""
    # In a real implementation, this might call external APIs
    return f"Comprehensive research data about {topic} compiled from multiple sources."

def create_lesson_content(module_title: str, objectives: List[Dict], audience: str) -> str:
    """Create detailed lesson content for a module."""
    content_template = f"""
# {module_title}

## Learning Objectives
After completing this module, you will be able to:
{chr(10).join([f"- {obj['description']}" for obj in objectives])}

## Introduction
[Detailed introduction content would be generated here]

## Core Concepts
[Main content sections would be generated here]

## Practical Examples
[Code examples and demonstrations would be included]

## Summary
[Key takeaways and next steps]
"""
    return content_template

def generate_code_examples(topic: str, difficulty: str, count: int = 3) -> List[Dict]:
    """Generate relevant code examples for the topic."""
    examples = []
    for i in range(count):
        examples.append({
            "title": f"{topic} Example {i+1}",
            "description": f"Demonstrates {topic} concept #{i+1}",
            "code": f"# {topic} example code would go here\nprint('Example {i+1}')",
            "explanation": f"This example shows how to implement {topic} in practice."
        })
    return examples

content_creator = Agent(
    name="ContentCreator",
    instructions="""You are an expert technical writer and educator specializing in AI and programming.

    Your responsibilities:
    - Create engaging, clear, and comprehensive course content
    - Write practical code examples with detailed explanations
    - Ensure content matches the target audience level
    - Include real-world applications and use cases
    - Structure content for optimal learning

    Writing style:
    - Clear and concise explanations
    - Progressive complexity
    - Practical examples
    - Engaging and conversational tone
    - Include visual descriptions where helpful""",
    tools=[research_topic_content, create_lesson_content, generate_code_examples]
)

def create_quiz_questions(topic: str, objectives: List[Dict], count: int = 5) -> List[Dict]:
    """Create quiz questions based on learning objectives."""
    questions = []
    question_types = ["multiple_choice", "true_false", "short_answer"]

    for i in range(count):
        obj = objectives[i % len(objectives)] if objectives else {"description": topic}
        questions.append({
            "id": i + 1,
            "type": question_types[i % len(question_types)],
            "question": f"Question about {obj['description']}?",
            "options": ["Option A", "Option B", "Option C", "Option D"] if question_types[i % len(question_types)] == "multiple_choice" else None,
            "correct_answer": "Option A" if question_types[i % len(question_types)] == "multiple_choice" else "True",
            "explanation": f"Explanation for the correct answer about {obj['description']}"
        })

    return questions

def design_practical_exercises(topic: str, difficulty: str, count: int = 3) -> List[Dict]:
    """Design hands-on exercises for the topic."""
    exercises = []
    for i in range(count):
        exercises.append({
            "title": f"{topic} Exercise {i+1}",
            "description": f"Practical exercise to reinforce {topic} concepts",
            "difficulty": difficulty,
            "estimated_time": 30 + (i * 15),
            "instructions": f"Step-by-step instructions for {topic} exercise {i+1}",
            "solution": f"Sample solution for exercise {i+1}",
            "learning_outcomes": [f"Outcome {j+1}" for j in range(2)]
        })
    return exercises

def create_capstone_project(topic: str, modules: List[Dict]) -> Dict:
    """Create a comprehensive capstone project."""
    return {
        "title": f"{topic} Capstone Project",
        "description": f"Comprehensive project integrating all {topic} concepts",
        "requirements": [
            f"Implement core {topic} functionality",
            "Include proper documentation",
            "Add error handling and testing",
            "Create a user-friendly interface"
        ],
        "deliverables": [
            "Source code repository",
            "Documentation",
            "Demo video",
            "Project report"
        ],
        "evaluation_criteria": [
            "Code quality and organization",
            "Functionality and features",
            "Documentation quality",
            "Creativity and innovation"
        ],
        "estimated_time": "2-3 weeks"
    }

assessment_designer = Agent(
    name="AssessmentDesigner",
    instructions="""You are an expert in educational assessment and evaluation.

    Your role:
    - Create diverse and effective assessment methods
    - Design quizzes that test understanding, not just memorization
    - Develop practical exercises that reinforce learning
    - Create comprehensive projects that integrate multiple concepts
    - Ensure assessments match learning objectives

    Assessment principles:
    - Multiple assessment types for different learning styles
    - Progressive difficulty that builds confidence
    - Clear rubrics and evaluation criteria
    - Immediate feedback and explanations
    - Real-world relevance""",
    tools=[create_quiz_questions, design_practical_exercises, create_capstone_project]
)

def review_content_quality(content: str, criteria: List[str]) -> Dict:
    """Review content against quality criteria."""
    review_results = {
        "overall_score": 85,  # Would be calculated based on actual analysis
        "strengths": [
            "Clear explanations",
            "Good structure",
            "Relevant examples"
        ],
        "areas_for_improvement": [
            "Add more visual elements",
            "Include additional practice exercises",
            "Expand on advanced concepts"
        ],
        "specific_feedback": {
            "clarity": "Content is generally clear but could benefit from more examples",
            "accuracy": "Technical information appears accurate",
            "engagement": "Content is informative but could be more engaging"
        }
    }
    return review_results

def check_learning_alignment(content: str, objectives: List[Dict]) -> Dict:
    """Check if content aligns with learning objectives."""
    alignment_score = 90  # Would be calculated based on actual analysis
    return {
        "alignment_score": alignment_score,
        "covered_objectives": len(objectives),
        "missing_coverage": [],
        "recommendations": [
            "All objectives are adequately covered",
            "Consider adding more practical applications"
        ]
    }

def suggest_improvements(content_type: str, content: str, target_audience: str) -> List[str]:
    """Suggest specific improvements for content."""
    improvements = [
        f"Add more {content_type}-specific examples for {target_audience}",
        "Include interactive elements to increase engagement",
        "Provide additional resources for further learning",
        "Add visual aids to support different learning styles"
    ]
    return improvements

quality_assurance = Agent(
    name="QualityAssurance",
    instructions="""You are a senior educational quality assurance specialist.

    Your responsibilities:
    - Review all course content for quality, accuracy, and effectiveness
    - Ensure alignment between content and learning objectives
    - Check for consistency in tone, style, and difficulty progression
    - Identify gaps or areas needing improvement
    - Suggest specific enhancements

    Quality standards:
    - Content accuracy and up-to-date information
    - Clear and engaging presentation
    - Appropriate difficulty progression
    - Inclusive and accessible language
    - Practical relevance and applicability""",
    tools=[review_content_quality, check_learning_alignment, suggest_improvements]
)

def coordinate_agent_handoff(from_agent: str, to_agent: str, data: Dict) -> str:
    """Coordinate handoff between agents."""
    return f"Handoff from {from_agent} to {to_agent} with data: {list(data.keys())}"

def compile_course_materials(modules: List[Dict], assessments: List[Dict]) -> Dict:
    """Compile all course materials into final structure."""
    return {
        "course_structure": {
            "modules": modules,
            "assessments": assessments,
            "total_modules": len(modules),
            "total_assessments": len(assessments)
        },
        "metadata": {
            "created_date": "2025-01-15",
            "version": "1.0",
            "status": "ready_for_review"
        }
    }

def generate_course_summary(course_data: Dict) -> str:
    """Generate a comprehensive course summary."""
    summary = f"""
Course Creation Summary:
- Total Modules: {course_data.get('total_modules', 0)}
- Total Assessments: {course_data.get('total_assessments', 0)}
- Status: {course_data.get('status', 'unknown')}
- Creation Date: {course_data.get('created_date', 'unknown')}
"""
    return summary

course_coordinator = Agent(
    name="CourseCoordinator",
    instructions="""You are the master coordinator for the AI course creation system.

    Your role:
    - Orchestrate the entire course creation workflow
    - Coordinate handoffs between specialized agents
    - Ensure all components work together cohesively
    - Monitor progress and quality throughout the process
    - Compile final course materials

    Workflow management:
    - Plan → Create → Assess → Review → Finalize
    - Ensure each agent completes their tasks properly
    - Handle any issues or conflicts between agents
    - Maintain overall project timeline and quality""",
    tools=[coordinate_agent_handoff, compile_course_materials, generate_course_summary]
)

class AICourseCreator:
    def __init__(self):
        self.agents = {
            'planner': course_planner,
            'creator': content_creator,
            'assessor': assessment_designer,
            'qa': quality_assurance,
            'coordinator': course_coordinator
        }
        self.session = Session()
        self.course_data = {}

    def create_course(self, topic: str, target_audience: str, difficulty: str) -> Dict:
        """Complete course creation workflow."""
        print(f"🚀 Starting course creation for: {topic}")

        # Step 1: Course Planning
        print("📋 Step 1: Course Planning")
        planning_result = self._plan_course(topic, target_audience, difficulty)

        # Step 2: Content Creation
        print("✍️ Step 2: Content Creation")
        content_result = self._create_content(planning_result)

        # Step 3: Assessment Design
        print("📝 Step 3: Assessment Design")
        assessment_result = self._design_assessments(planning_result, content_result)

        # Step 4: Quality Assurance
        print("🔍 Step 4: Quality Assurance")
        qa_result = self._quality_review(content_result, assessment_result)

        # Step 5: Final Compilation
        print("📦 Step 5: Final Compilation")
        final_course = self._compile_final_course(
            planning_result, content_result, assessment_result, qa_result
        )

        print("✅ Course creation completed!")
        return final_course

    def _plan_course(self, topic: str, audience: str, difficulty: str) -> Dict:
        """Step 1: Create course plan and structure."""
        prompt = f"""Create a comprehensive course plan for "{topic}" targeting {audience} at {difficulty} level.

        Requirements:
        - 5-7 learning objectives
        - 4-5 course modules
        - Logical progression from basics to advanced
        - Estimated time for each component
        """

        result = Runner.run_sync(self.agents['planner'], prompt, session=self.session)

        # Extract structured data from result (simplified)
        planning_data = {
            'topic': topic,
            'audience': audience,
            'difficulty': difficulty,
            'modules': [
                {'title': f'Module {i+1}', 'objectives': []}
                for i in range(5)
            ]
        }

        return planning_data

    def _create_content(self, planning_data: Dict) -> Dict:
        """Step 2: Create detailed content for each module."""
        content_data = {'modules': []}

        for module in planning_data['modules']:
            prompt = f"""Create comprehensive content for module: {module['title']}

            Target audience: {planning_data['audience']}
            Difficulty: {planning_data['difficulty']}

            Include:
            - Detailed explanations
            - Code examples
            - Practical applications
            - Visual descriptions
            """

            result = Runner.run_sync(self.agents['creator'], prompt, session=self.session)

            module_content = {
                'title': module['title'],
                'content': result.final_output,
                'examples': []  # Would be populated from tool calls
            }
            content_data['modules'].append(module_content)

        return content_data

    def _design_assessments(self, planning_data: Dict, content_data: Dict) -> Dict:
        """Step 3: Design assessments for the course."""
        prompt = f"""Design comprehensive assessments for the course on {planning_data['topic']}.

        Create:
        - Quiz questions for each module
        - Practical exercises
        - A capstone project

        Target audience: {planning_data['audience']}
        Difficulty: {planning_data['difficulty']}
        """

        result = Runner.run_sync(self.agents['assessor'], prompt, session=self.session)

        assessment_data = {
            'quizzes': [],  # Would be populated from tool calls
            'exercises': [],
            'capstone_project': {}
        }

        return assessment_data

    def _quality_review(self, content_data: Dict, assessment_data: Dict) -> Dict:
        """Step 4: Quality assurance review."""
        prompt = f"""Review the course content and assessments for quality and effectiveness.

        Content modules: {len(content_data['modules'])}
        Assessments: {len(assessment_data.get('quizzes', []))} quizzes, {len(assessment_data.get('exercises', []))} exercises

        Provide:
        - Overall quality assessment
        - Specific improvement suggestions
        - Alignment with learning objectives
        """

        result = Runner.run_sync(self.agents['qa'], prompt, session=self.session)

        qa_data = {
            'overall_score': 85,
            'recommendations': result.final_output,
            'approved': True
        }

        return qa_data

    def _compile_final_course(self, planning: Dict, content: Dict, assessments: Dict, qa: Dict) -> Dict:
        """Step 5: Compile final course package."""
        prompt = f"""Compile the final course package with all components.

        Components:
        - Course plan: {planning['topic']}
        - Content modules: {len(content['modules'])}
        - Assessments created
        - QA score: {qa['overall_score']}

        Create final course structure and summary.
        """

        result = Runner.run_sync(self.agents['coordinator'], prompt, session=self.session)

        final_course = {
            'metadata': {
                'title': f"Complete Course: {planning['topic']}",
                'target_audience': planning['audience'],
                'difficulty': planning['difficulty'],
                'total_modules': len(content['modules']),
                'qa_score': qa['overall_score'],
                'status': 'completed'
            },
            'content': content,
            'assessments': assessments,
            'qa_report': qa,
            'summary': result.final_output
        }

        return final_course

# Usage Example
def main():
    """Demonstrate the complete AI course creation system."""
    creator = AICourseCreator()

    # Create a course on machine learning
    course = creator.create_course(
        topic="Machine Learning Fundamentals",
        target_audience="Software developers with Python experience",
        difficulty="intermediate"
    )

    # Display results
    print("\n" + "="*50)
    print("COURSE CREATION RESULTS")
    print("="*50)
    print(f"Title: {course['metadata']['title']}")
    print(f"Target Audience: {course['metadata']['target_audience']}")
    print(f"Difficulty: {course['metadata']['difficulty']}")
    print(f"Total Modules: {course['metadata']['total_modules']}")
    print(f"QA Score: {course['metadata']['qa_score']}/100")
    print(f"Status: {course['metadata']['status']}")
    print("\nCourse Summary:")
    print(course['summary'])

if __name__ == "__main__":
    main()

class AgentMessage:
    def __init__(self, sender: str, recipient: str, message_type: str, data: Dict):
        self.sender = sender
        self.recipient = recipient
        self.message_type = message_type
        self.data = data
        self.timestamp = datetime.now()

class AgentCommunicationHub:
    def __init__(self):
        self.message_queue = []
        self.agent_status = {}

    def send_message(self, message: AgentMessage):
        """Send message between agents."""
        self.message_queue.append(message)
        print(f"📨 {message.sender} → {message.recipient}: {message.message_type}")

    def get_messages_for_agent(self, agent_name: str) -> List[AgentMessage]:
        """Get pending messages for an agent."""
        messages = [msg for msg in self.message_queue if msg.recipient == agent_name]
        # Remove retrieved messages
        self.message_queue = [msg for msg in self.message_queue if msg.recipient != agent_name]
        return messages

class CourseCreationError(Exception):
    pass

class RobustCourseCreator(AICourseCreator):
    def __init__(self, max_retries: int = 3):
        super().__init__()
        self.max_retries = max_retries

    def _execute_with_retry(self, agent_name: str, prompt: str, step_name: str) -> Dict:
        """Execute agent task with retry logic."""
        for attempt in range(self.max_retries):
            try:
                print(f"🔄 Attempt {attempt + 1}/{self.max_retries} for {step_name}")
                result = Runner.run_sync(self.agents[agent_name], prompt, session=self.session)
                return {'success': True, 'result': result, 'attempt': attempt + 1}

            except Exception as e:
                print(f"❌ Attempt {attempt + 1} failed: {str(e)}")
                if attempt == self.max_retries - 1:
                    raise CourseCreationError(f"Failed to complete {step_name} after {self.max_retries} attempts")

                # Wait before retry
                time.sleep(2 ** attempt)  # Exponential backoff

        return {'success': False, 'error': 'Max retries exceeded'}

import time
from dataclasses import dataclass
from typing import Dict, List

@dataclass
class PerformanceMetrics:
    step_name: str
    duration: float
    token_usage: int
    success: bool
    error_message: Optional[str] = None

class PerformanceMonitor:
    def __init__(self):
        self.metrics: List[PerformanceMetrics] = []

    def track_step(self, step_name: str):
        """Context manager for tracking step performance."""
        return StepTracker(self, step_name)

    def add_metric(self, metric: PerformanceMetrics):
        self.metrics.append(metric)

    def get_summary(self) -> Dict:
        total_time = sum(m.duration for m in self.metrics)
        total_tokens = sum(m.token_usage for m in self.metrics)
        success_rate = sum(1 for m in self.metrics if m.success) / len(self.metrics)

        return {
            'total_steps': len(self.metrics),
            'total_time': total_time,
            'total_tokens': total_tokens,
            'success_rate': success_rate,
            'average_step_time': total_time / len(self.metrics) if self.metrics else 0
        }

class StepTracker:
    def __init__(self, monitor: PerformanceMonitor, step_name: str):
        self.monitor = monitor
        self.step_name = step_name
        self.start_time = None

    def __enter__(self):
        self.start_time = time.time()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        duration = time.time() - self.start_time
        success = exc_type is None
        error_message = str(exc_val) if exc_val else None

        metric = PerformanceMetrics(
            step_name=self.step_name,
            duration=duration,
            token_usage=0,  # Would be extracted from agent result
            success=success,
            error_message=error_message
        )

        self.monitor.add_metric(metric)

import asyncio
from concurrent.futures import ThreadPoolExecutor

class ScalableCourseCreator:
    def __init__(self, max_workers: int = 4):
        self.max_workers = max_workers
        self.executor = ThreadPoolExecutor(max_workers=max_workers)

    async def create_multiple_courses(self, course_requests: List[Dict]) -> List[Dict]:
        """Create multiple courses concurrently."""
        tasks = []

        for request in course_requests:
            task = asyncio.create_task(
                self._create_course_async(request)
            )
            tasks.append(task)

        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

    async def _create_course_async(self, request: Dict) -> Dict:
        """Create a single course asynchronously."""
        loop = asyncio.get_event_loop()
        creator = AICourseCreator()

        result = await loop.run_in_executor(
            self.executor,
            creator.create_course,
            request['topic'],
            request['audience'],
            request['difficulty']
        )

        return result

# Usage example for batch processing
async def batch_course_creation():
    creator = ScalableCourseCreator(max_workers=3)

    requests = [
        {
            'topic': 'Python for Data Science',
            'audience': 'Data analysts',
            'difficulty': 'intermediate'
        },
        {
            'topic': 'React Fundamentals',
            'audience': 'Web developers',
            'difficulty': 'beginner'
        },
        {
            'topic': 'Advanced Machine Learning',
            'audience': 'ML engineers',
            'difficulty': 'advanced'
        }
    ]

    results = await creator.create_multiple_courses(requests)

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"Course {i+1} failed: {result}")
        else:
            print(f"Course {i+1} completed: {result['metadata']['title']}")

# Run batch creation
# asyncio.run(batch_course_creation())