learning_platform_old.py

from typing import List, Dict, Any
from dataclasses import dataclass, field
from datetime import datetime
import json
import os
from langchain_community.chat_models import ChatOpenAI
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
import streamlit as st
import logging

# Set up logging for debugging purposes
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')

@dataclass
class Section:
    title: str
    content: str = ""
    key_points: List[str] = field(default_factory=list)
    examples: List[str] = field(default_factory=list)
    quiz_questions: List[Dict[str, Any]] = field(default_factory=list)
    is_complete: bool = False

@dataclass
class CourseModule:
    title: str
    objectives: List[str]
    prerequisites: List[str] = field(default_factory=list)
    sections: List[Section] = field(default_factory=list)
    is_complete: bool = False

@dataclass
class LearningPath:
    topic: str
    description: str
    modules: List[CourseModule]
    difficulty_level: str
    created_at: datetime
    is_generating: bool = True

class CoursePrompts:
    @staticmethod
    def course_planning_prompt() -> str:
        return """As a course planning expert, design a structured learning path for {topic} at {difficulty} level.

Requirements:
1. 5-7 progressive modules
2. Clear prerequisites and objectives
3. Practical applications
4. Real-world examples
5. A compelling description that excites the learner about the journey ahead
6. Each module should contain content, quiz questions, and be designed to progressively enhance understanding.
Return a structured JSON with detailed content for each module.
"""

    @staticmethod
    def module_content_prompt() -> str:
        return """Create engaging module content for the module titled '{title}' with the following objectives:

Objectives: {objectives}

Include:
1. Clear explanations with practical examples to deepen understanding
2. Real-world applications relevant to the topic
3. Key points that summarize each section concisely
4. A set of 3-5 quiz questions for each section, with answers and explanations
5. Encourage learners to think critically and ask questions related to '{title}'
Return a structured JSON with detailed content for each section.
"""

    @staticmethod
    def user_question_prompt() -> str:
        return """As an AI assistant, provide a clear and informative answer to the user's question based on the course topic '{topic}', the module '{module_title}', and the related content.

User Question: {question}

Ensure your response is helpful, easy to understand, and relevant to the course content.
"""

class CourseBuilder:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.llm = ChatOpenAI(
            temperature=0.7,
            model="gpt-4",
            openai_api_key=api_key
        )
        self.embeddings = OpenAIEmbeddings(openai_api_key=api_key)
        self.vector_store = FAISS.from_texts(
            ["Initial course content"],
            embedding=self.embeddings
        )
        self.prompts = CoursePrompts()

    async def plan_course(self, topic: str, difficulty: str) -> LearningPath:
        prompt = self.prompts.course_planning_prompt().format(topic=topic, difficulty=difficulty)
        logging.debug(f"Sending course planning prompt: {prompt}")
        response = await self.llm.apredict(prompt)
        logging.debug(f"Received response for course planning: {response}")
        
        # Debug: Log the raw response for troubleshooting
        if not response.strip():
            logging.error("Empty response from API")
            raise ValueError("Empty response from API")
        
        try:
            course_plan = json.loads(response)
        except json.JSONDecodeError as e:
            logging.error(f"Invalid JSON response from API: {str(e)}\nResponse: {response}")
            raise ValueError(f"Invalid JSON response from API: {str(e)}\nResponse: {response}")

        modules = [
            CourseModule(
                title=module["title"],
                objectives=module.get("objectives", []),
                prerequisites=module.get("prerequisites", []),
                sections=[
                    Section(
                        title=section["title"],
                        content=section.get("content", ""),
                        key_points=section.get("key_points", []),
                        examples=section.get("examples", []),
                        quiz_questions=section.get("quiz_questions", [])
                    ) for section in module.get("sections", [])
                ]
            ) for module in course_plan.get("modules", [])
        ]

        learning_path = LearningPath(
            topic=topic,
            description=course_plan.get("description", ""),
            modules=modules,
            difficulty_level=difficulty,
            created_at=datetime.now(),
            is_generating=False
        )

        # Store embeddings for course plan for future reference
        self.vector_store.add_texts(
            [json.dumps(course_plan)],
            metadatas=[{"type": "course_plan", "topic": topic}]
        )

        logging.info(f"Created learning path for topic '{topic}' with difficulty '{difficulty}'")
        return learning_path

    async def create_module_content(self, module: CourseModule) -> List[Section]:
        prompt = self.prompts.module_content_prompt().format(
            title=module.title,
            objectives=", ".join(module.objectives)
        )
        logging.debug(f"Sending module content prompt: {prompt}")
        response = await self.llm.apredict(prompt)
        logging.debug(f"Received response for module content: {response}")

        if not response.strip():
            logging.error("Empty response from API")
            raise ValueError("Empty response from API")
        
        try:
            content_json = json.loads(response)
        except json.JSONDecodeError as e:
            logging.error(f"Invalid JSON response from API: {str(e)}\nResponse: {response}")
            raise ValueError(f"Invalid JSON response from API: {str(e)}\nResponse: {response}")
        
        sections = [Section(**section) for section in content_json.get("sections", [])]

        # Store embeddings for module content
        for section in sections:
            self.vector_store.add_texts(
                [section.content],
                metadatas=[{"type": "module_content", "module": module.title}]
            )

        logging.info(f"Created {len(sections)} sections for module: {module.title}")
        return sections

    async def answer_user_question(self, topic: str, module_title: str, question: str) -> str:
        prompt = self.prompts.user_question_prompt().format(
            topic=topic,
            module_title=module_title,
            question=question
        )
        logging.debug(f"Sending user question prompt: {prompt}")
        response = await self.llm.apredict(prompt)
        logging.debug(f"Received response for user question: {response}")

        if not response.strip():
            logging.error("Empty response from API")
            raise ValueError("Empty response from API")

        return response

class LearningPlatform:
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv("OPENAI_API_KEY")
        self.course_builder = CourseBuilder(self.api_key)

    async def create_course(self, topic: str, difficulty: str) -> LearningPath:
        try:
            learning_path = await self.course_builder.plan_course(topic, difficulty)
            return learning_path
        except Exception as e:
            logging.error(f"Course creation error: {str(e)}")
            raise Exception(f"Course creation error: {str(e)}")

    async def generate_next_module(self, path: LearningPath, module_index: int):
        if module_index < len(path.modules):
            module = path.modules[module_index]
            if not module.is_complete:
                logging.info(f"Generating content for module: {module.title}")
                sections = await self.course_builder.create_module_content(module)
                module.sections = sections
                module.is_complete = True
                logging.info(f"Module '{module.title}' is now complete.")

    async def handle_user_question(self, path: LearningPath, module_index: int, question: str) -> str:
        if module_index < len(path.modules):
            module = path.modules[module_index]
            logging.info(f"Answering user question for module: {module.title}")
            answer = await self.course_builder.answer_user_question(path.topic, module.title, question)
            return answer
        else:
            logging.error(f"Invalid module index: {module_index}")
            raise ValueError("Invalid module index")