Create learning_platform.py

learning_platform.py

@@ -0,0 +1,212 @@
+# learning_platform.py
+from typing import List, Dict, Any, Optional
+from dataclasses import dataclass
+from datetime import datetime
+import json
+import logging
+from langchain.chat_models import ChatOpenAI
+from langchain.embeddings import OpenAIEmbeddings
+from langchain.vectorstores import FAISS
+from langchain.memory import ConversationBufferMemory
+from langchain.chains import ConversationalRetrievalQA
+from prompts import CoursePromptTemplates
+from models import *
+from sqlalchemy.orm import Session
+import tiktoken
+
+# Enhanced logging configuration
+logging.basicConfig(
+    level=logging.DEBUG,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler('learning_platform.log'),
+        logging.StreamHandler()
+    ]
+)
+
+logger = logging.getLogger(__name__)
+
+class EnhancedCourseBuilder:
+    def __init__(self, api_key: str, db_session: Session):
+        self.api_key = api_key
+        self.db_session = db_session
+        self.prompt_templates = CoursePromptTemplates()
+        
+        # Initialize LLM with increased tokens and temperature
+        self.llm = ChatOpenAI(
+            temperature=0.7,
+            model="gpt-4-turbo-preview",  # Using the latest model with higher token limit
+            max_tokens=4096,
+            openai_api_key=api_key
+        )
+        
+        # Initialize embeddings and vector store
+        self.embeddings = OpenAIEmbeddings(openai_api_key=api_key)
+        self.vector_store = FAISS.from_texts(
+            ["Initial course content"],
+            embedding=self.embeddings
+        )
+        
+        # Initialize conversation memory
+        self.memory = ConversationBufferMemory(
+            memory_key="chat_history",
+            return_messages=True
+        )
+        
+        # Initialize retrieval QA chain
+        self.qa_chain = ConversationalRetrievalQA.from_llm(
+            llm=self.llm,
+            retriever=self.vector_store.as_retriever(),
+            memory=self.memory,
+            verbose=True
+        )
+
+    async def create_course(self, topic: str, difficulty: str, user_id: int) -> Course:
+        """Create a new course with enhanced content generation"""
+        logger.info(f"Creating course for topic: {topic}, difficulty: {difficulty}")
+        
+        try:
+            # Generate course content using enhanced prompt
+            prompt = self.prompt_templates.COURSE_PLANNING.substitute(
+                topic=topic,
+                difficulty=difficulty,
+                audience_level=difficulty,
+                duration="8 weeks",
+                learning_style="interactive",
+                industry_focus="general"
+            )
+            
+            logger.debug(f"Sending course planning prompt: {prompt}")
+            response = await self.llm.agenerate([prompt])
+            course_plan = json.loads(response.generations[0].text)
+            
+            # Create course in database
+            new_course = Course(
+                title=topic,
+                description=course_plan.get("description"),
+                difficulty_level=difficulty,
+                content=course_plan,
+                metadata={
+                    "generator_version": "2.0",
+                    "model": "gpt-4-turbo-preview",
+                    "creation_parameters": {
+                        "difficulty": difficulty,
+                        "topic": topic
+                    }
+                }
+            )
+            
+            self.db_session.add(new_course)
+            self.db_session.commit()
+            
+            # Create user course association
+            user_course = UserCourse(
+                user_id=user_id,
+                course_id=new_course.id,
+                status="enrolled"
+            )
+            
+            self.db_session.add(user_course)
+            self.db_session.commit()
+            
+            # Store course content in vector store
+            self.vector_store.add_texts(
+                [json.dumps(course_plan)],
+                metadatas=[{"type": "course_plan", "course_id": new_course.id}]
+            )
+            
+            logger.info(f"Successfully created course: {new_course.id}")
+            return new_course
+            
+        except Exception as e:
+            logger.error(f"Error creating course: {str(e)}", exc_info=True)
+            raise
+
+    async def generate_module_content(self, module_id: int) -> Dict[str, Any]:
+        """Generate detailed content for a specific module"""
+        logger.info(f"Generating content for module: {module_id}")
+        
+        try:
+            module = self.db_session.query(CourseModule).get(module_id)
+            if not module:
+                raise ValueError(f"Module not found: {module_id}")
+            
+            prompt = self.prompt_templates.MODULE_CONTENT.substitute(
+                title=module.title,
+                objectives=json.dumps(module.content.get("objectives", [])),
+                prerequisites=json.dumps(module.prerequisites),
+                competency_level=module.course.difficulty_level,
+                industry_context="general"
+            )
+            
+            logger.debug(f"Sending module content prompt: {prompt}")
+            response = await self.llm.agenerate([prompt])
+            content = json.loads(response.generations[0].text)
+            
+            # Update module content
+            module.content.update(content)
+            self.db_session.commit()
+            
+            # Store content in vector store
+            self.vector_store.add_texts(
+                [json.dumps(content)],
+                metadatas=[{"type": "module_content", "module_id": module_id}]
+            )
+            
+            logger.info(f"Successfully generated content for module: {module_id}")
+            return content
+            
+        except Exception as e:
+            logger.error(f"Error generating module content: {str(e)}", exc_info=True)
+            raise
+
+    async def answer_user_question(
+        self, 
+        user_id: int,
+        course_id: int,
+        module_id: int,
+        question: str
+    ) -> str:
+        """Answer user questions with context awareness"""
+        logger.info(f"Answering question for user {user_id} in course {course_id}")
+        
+        try:
+            # Get context
+            module = self.db_session.query(CourseModule).get(module_id)
+            course = module.course
+            user_course = self.db_session.query(UserCourse).filter_by(
+                user_id=user_id,
+                course_id=course_id
+            ).first()
+            
+            # Generate prompt with context
+            prompt = self.prompt_templates.USER_QUESTION.substitute(
+                topic=course.title,
+                module_title=module.title,
+                completed_modules=self.get_completed_modules(user_course.id),
+                user_level=course.difficulty_level,
+                question=question
+            )
+            
+            # Use QA chain for answer
+            response = await self.qa_chain.arun(prompt)
+            
+            # Log interaction
+            interaction = UserInteraction(
+                user_id=user_id,
+                interaction_type="question_asked",
+                content_reference=f"module_{module_id}",
+                metadata={
+                    "question": question,
+                    "response": response
+                }
+            )
+            self.db_session.add(interaction)
+            self.db_session.commit()
+            
+            logger.info(f"Successfully answered question for user {user_id}")
+            return response
+            
+        except Exception as e:
+            logger.error(f"Error answering question: {str(e)}", exc_info=True)
+            raise