# 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_community.embeddings import OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain.memory import ConversationBufferMemory from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.chains import create_stuff_documents_chain, create_history_aware_retriever, create_retrieval_chain 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 ) # Create the history-aware retriever contextualize_q_prompt = ChatPromptTemplate.from_messages([ ["system", """ Given a chat history and the latest user question which might reference context in the chat history, formulate a standalone question which can be understood without the chat history. Do NOT answer the question, just reformulate it if needed and otherwise return it as is."""], MessagesPlaceholder("chat_history"), ["human", "{input}"] ]) self.history_aware_retriever = create_history_aware_retriever( llm=self.llm, retriever=self.vector_store.as_retriever(), rephrase_prompt=contextualize_q_prompt ) # Create the QA system prompt qa_system_prompt = ChatPromptTemplate.from_messages([ ["system", """ You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Use three sentences maximum and keep the answer concise. {context}"""], MessagesPlaceholder("chat_history"), ["human", "{input}"] ]) # Create the question-answer chain self.question_answer_chain = create_stuff_documents_chain( llm=self.llm, prompt=qa_system_prompt ) # Create the retrieval chain self.rag_chain = create_retrieval_chain( retriever=self.history_aware_retriever, combine_docs_chain=self.question_answer_chain ) 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() # Use retrieval chain for answer response = await self.rag_chain.arun({ "chat_history": self.memory.load_memory_variables({}), "input": question }) # 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