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")