AI Education & E-Learning Revolution: From AI Tutors to Adaptive Learning and Auto-Grading

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.memory import ConversationBufferMemory
from langchain.chains import LLMChain

SOCRATIC_SYSTEM_PROMPT = """
You are a Socratic AI tutor. You must follow these rules at all times:

1. Never give the student the direct answer to a question.
2. First ask questions to understand the student's current level of understanding.
3. Provide step-by-step hints so the student can discover the answer independently.
4. When you detect a misconception, use questions to guide their thinking rather than correcting directly.
5. Use positive reinforcement appropriately.

Current subject: {subject}
Student level: {level}
"""

def create_socratic_tutor(subject: str, level: str = "high school"):
    llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

    prompt = ChatPromptTemplate.from_messages([
        ("system", SOCRATIC_SYSTEM_PROMPT),
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}")
    ])

    memory = ConversationBufferMemory(
        memory_key="history",
        return_messages=True
    )

    chain = LLMChain(
        llm=llm,
        prompt=prompt,
        memory=memory,
        verbose=True
    )

    return chain, {"subject": subject, "level": level}


def tutor_session(chain, chain_inputs: dict, student_message: str) -> str:
    response = chain.invoke({
        **chain_inputs,
        "input": student_message
    })
    return response["text"]


# Example usage
tutor, inputs = create_socratic_tutor("Quadratic Equations", "Grade 10")
reply = tutor_session(tutor, inputs, "How do I solve x^2 - 5x + 6 = 0?")
print(reply)

PROFILING_PROMPT = """
Analyze the following tutoring conversation and return a learning profile as JSON.

Conversation:
{conversation}

Return format:
{{
  "strengths": ["list of strengths"],
  "weaknesses": ["list of weaknesses"],
  "misconceptions": ["identified misconceptions"],
  "recommended_topics": ["next recommended topics"],
  "difficulty_level": "easy|medium|hard",
  "engagement_score": 0.0 to 1.0
}}
"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import subprocess
import ast
import time
from typing import Optional

app = FastAPI()

class CodeSubmission(BaseModel):
    student_id: str
    problem_id: str
    code: str
    language: str = "python"

class GradingResult(BaseModel):
    passed: int
    total: int
    score: float
    feedback: str
    execution_time_ms: float
    style_score: Optional[float] = None

# Test case store (in production, load from DB)
TEST_CASES = {
    "fibonacci": [
        {"input": "0", "expected": "0"},
        {"input": "1", "expected": "1"},
        {"input": "10", "expected": "55"},
        {"input": "20", "expected": "6765"},
    ]
}

def run_python_code(code: str, input_data: str, timeout: float = 5.0) -> tuple[str, float]:
    """Execute code in a sandboxed subprocess."""
    start = time.time()
    try:
        result = subprocess.run(
            ["python3", "-c", code],
            input=input_data,
            capture_output=True,
            text=True,
            timeout=timeout
        )
        elapsed = (time.time() - start) * 1000
        return result.stdout.strip(), elapsed
    except subprocess.TimeoutExpired:
        return "TIMEOUT", timeout * 1000

def analyze_code_style(code: str) -> float:
    """Return a code style score between 0 and 1."""
    score = 1.0
    try:
        tree = ast.parse(code)
        has_function = any(isinstance(n, ast.FunctionDef) for n in ast.walk(tree))
        if not has_function:
            score -= 0.2
        for node in ast.walk(tree):
            if isinstance(node, ast.Name) and len(node.id) == 1 and node.id not in ["i", "j", "k", "n", "x", "y"]:
                score -= 0.05
    except SyntaxError:
        return 0.0
    return max(0.0, score)

@app.post("/grade", response_model=GradingResult)
async def grade_submission(submission: CodeSubmission):
    test_cases = TEST_CASES.get(submission.problem_id)
    if not test_cases:
        raise HTTPException(status_code=404, detail="Problem not found.")

    passed = 0
    total_time = 0.0
    feedback_lines = []

    for i, tc in enumerate(test_cases):
        output, elapsed = run_python_code(submission.code, tc["input"])
        total_time += elapsed
        if output == tc["expected"]:
            passed += 1
        else:
            feedback_lines.append(
                f"Test case {i+1} failed: input={tc['input']}, "
                f"expected={tc['expected']}, got={output}"
            )

    style_score = analyze_code_style(submission.code)
    score = (passed / len(test_cases)) * 0.8 + style_score * 0.2

    feedback = f"{passed}/{len(test_cases)} test cases passed."
    if feedback_lines:
        feedback += " " + " | ".join(feedback_lines[:3])

    return GradingResult(
        passed=passed,
        total=len(test_cases),
        score=round(score, 3),
        feedback=feedback,
        execution_time_ms=round(total_time / len(test_cases), 2),
        style_score=round(style_score, 3)
    )

from sentence_transformers import SentenceTransformer, util
import language_tool_python

class AutoEssayScorer:
    def __init__(self):
        self.embedder = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
        self.lang_tool = language_tool_python.LanguageTool("en-US")

    def score_content(self, essay: str, reference_topics: list[str]) -> float:
        """Topic relevance and content score (0-1)."""
        essay_emb = self.embedder.encode(essay, convert_to_tensor=True)
        topic_embs = self.embedder.encode(reference_topics, convert_to_tensor=True)
        similarities = util.cos_sim(essay_emb, topic_embs)
        return float(similarities.max().item())

    def score_language(self, essay: str) -> dict:
        """Grammar errors, vocabulary diversity, and sentence count."""
        words = essay.split()
        unique_ratio = len(set(words)) / len(words) if words else 0
        sentences = [s.strip() for s in essay.split(".") if s.strip()]
        matches = self.lang_tool.check(essay)
        grammar_error_rate = len(matches) / len(sentences) if sentences else 0
        return {
            "vocabulary_diversity": round(unique_ratio, 3),
            "grammar_errors": len(matches),
            "grammar_error_rate": round(grammar_error_rate, 3),
            "sentence_count": len(sentences),
            "language_score": round(max(0, 1 - grammar_error_rate * 0.5) * unique_ratio, 3)
        }

    def generate_feedback(self, content_score: float, lang_stats: dict) -> str:
        feedback = []
        if content_score < 0.5:
            feedback.append("Try to stay more on topic.")
        if lang_stats["vocabulary_diversity"] < 0.4:
            feedback.append("Try to use a wider range of vocabulary.")
        if lang_stats["grammar_errors"] > 5:
            feedback.append(f"Please fix {lang_stats['grammar_errors']} grammar errors.")
        return " ".join(feedback) if feedback else "Excellent essay!"

import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
import numpy as np

class DKTModel(nn.Module):
    """Deep Knowledge Tracing: LSTM-based learner state estimation."""

    def __init__(self, num_skills: int, hidden_size: int = 128, num_layers: int = 2):
        super().__init__()
        self.num_skills = num_skills
        # Input: one-hot encoding of (problem_id, correct) pairs -> 2 * num_skills dims
        self.input_size = 2 * num_skills

        self.lstm = nn.LSTM(
            input_size=self.input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=0.2
        )
        self.output_layer = nn.Linear(hidden_size, num_skills)
        self.sigmoid = nn.Sigmoid()

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        """
        x: (batch, seq_len, 2*num_skills) - one-hot encoded (problem, correct) pairs
        returns: (batch, seq_len, num_skills) - predicted correctness probability per skill
        """
        lstm_out, _ = self.lstm(x)
        logits = self.output_layer(lstm_out)
        return self.sigmoid(logits)

class StudentInteractionDataset(Dataset):
    def __init__(self, interactions: list, num_skills: int, max_seq_len: int = 200):
        self.data = interactions
        self.num_skills = num_skills
        self.max_seq_len = max_seq_len

    def __len__(self):
        return len(self.data)

    def encode_interaction(self, skill_id: int, correct: int) -> np.ndarray:
        """(skill_id, correct) -> 2*num_skills one-hot vector."""
        vec = np.zeros(2 * self.num_skills)
        if correct == 1:
            vec[skill_id] = 1
        else:
            vec[self.num_skills + skill_id] = 1
        return vec

    def __getitem__(self, idx):
        seq = self.data[idx][:self.max_seq_len]
        x = np.array([self.encode_interaction(s, c) for s, c in seq[:-1]])
        y_skill = np.array([s for s, c in seq[1:]])
        y_correct = np.array([c for s, c in seq[1:]])
        return (
            torch.FloatTensor(x),
            torch.LongTensor(y_skill),
            torch.FloatTensor(y_correct)
        )

def train_dkt(model: DKTModel, dataloader: DataLoader, epochs: int = 10):
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
    criterion = nn.BCELoss()
    model.train()

    for epoch in range(epochs):
        total_loss = 0.0
        for x, y_skill, y_correct in dataloader:
            optimizer.zero_grad()
            pred = model(x)  # (batch, seq, num_skills)
            idx = y_skill.unsqueeze(-1)
            skill_pred = pred.gather(2, idx).squeeze(-1)
            loss = criterion(skill_pred, y_correct)
            loss.backward()
            optimizer.step()
            total_loss += loss.item()
        print(f"Epoch {epoch+1}: Loss = {total_loss/len(dataloader):.4f}")

from openai import OpenAI
import json

client = OpenAI()

BLOOM_LEVELS = {
    "remember": "recall and recognition of facts",
    "understand": "explaining concepts and finding examples",
    "apply": "using formulas or procedures in new situations",
    "analyze": "breaking down components and understanding relationships",
    "evaluate": "making judgments and critical assessments",
    "create": "designing or creating something new"
}

def generate_questions(
    topic: str,
    bloom_level: str,
    num_questions: int = 3,
    student_level: str = "High School"
) -> list[dict]:
    """Auto-generate educational questions based on Bloom's Taxonomy."""
    level_desc = BLOOM_LEVELS.get(bloom_level, BLOOM_LEVELS["understand"])

    prompt = f"""
You are an expert educational content developer.
Generate {num_questions} multiple-choice questions in JSON format.

Requirements:
- Topic: {topic}
- Student level: {student_level}
- Bloom's Taxonomy level: {bloom_level} ({level_desc})
- Each question must include 4 options, the correct answer, and a detailed explanation.

Return JSON format:
[
  {{
    "question": "Question text",
    "options": ["A. ...", "B. ...", "C. ...", "D. ..."],
    "answer": "A",
    "explanation": "Detailed explanation",
    "bloom_level": "{bloom_level}"
  }}
]

Return only JSON, no other text.
"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0.8
    )

    result = json.loads(response.choices[0].message.content)
    return result if isinstance(result, list) else result.get("questions", [])

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional

@dataclass
class FlashCard:
    card_id: str
    front: str
    back: str
    # SM-2 parameters
    ease_factor: float = 2.5      # Difficulty multiplier (minimum 1.3)
    interval: int = 1              # Current review interval in days
    repetitions: int = 0           # Consecutive successful reviews
    next_review: datetime = field(default_factory=datetime.now)
    last_reviewed: Optional[datetime] = None

def sm2_update(card: FlashCard, quality: int) -> FlashCard:
    """
    Update card parameters using the SM-2 algorithm.

    quality: 0-5
      0 = complete blackout
      1 = incorrect, but remembered on seeing answer
      2 = incorrect, but easy to recall
      3 = correct with serious difficulty
      4 = correct after a hesitation
      5 = perfect, immediate recall
    """
    assert 0 <= quality <= 5, "quality must be between 0 and 5."

    if quality < 3:
        # Failure: reset to beginning
        card.repetitions = 0
        card.interval = 1
    else:
        # Success: calculate new interval
        if card.repetitions == 0:
            card.interval = 1
        elif card.repetitions == 1:
            card.interval = 6
        else:
            card.interval = round(card.interval * card.ease_factor)
        card.repetitions += 1

    # Update ease factor
    card.ease_factor = max(
        1.3,
        card.ease_factor + 0.1 - (5 - quality) * (0.08 + (5 - quality) * 0.02)
    )

    card.last_reviewed = datetime.now()
    card.next_review = datetime.now() + timedelta(days=card.interval)
    return card


class SpacedRepetitionSystem:
    """A simple spaced repetition learning system."""

    def __init__(self):
        self.cards: dict[str, FlashCard] = {}

    def add_card(self, card_id: str, front: str, back: str):
        self.cards[card_id] = FlashCard(card_id=card_id, front=front, back=back)

    def get_due_cards(self) -> list[FlashCard]:
        """Return all cards due for review today."""
        now = datetime.now()
        return [c for c in self.cards.values() if c.next_review <= now]

    def review_card(self, card_id: str, quality: int) -> FlashCard:
        card = self.cards[card_id]
        updated = sm2_update(card, quality)
        self.cards[card_id] = updated
        return updated

    def get_stats(self) -> dict:
        cards = list(self.cards.values())
        return {
            "total": len(cards),
            "due_today": len(self.get_due_cards()),
            "avg_ease": round(sum(c.ease_factor for c in cards) / len(cards), 3) if cards else 0,
            "avg_interval": round(sum(c.interval for c in cards) / len(cards), 1) if cards else 0
        }

class MasteryLearningTracker:
    """Track mastery thresholds per concept for mastery-based learning."""

    MASTERY_THRESHOLD = 0.80  # 80% correct to advance

    def __init__(self):
        self.concept_scores: dict[str, list[int]] = {}

    def record_attempt(self, concept: str, correct: bool):
        self.concept_scores.setdefault(concept, []).append(1 if correct else 0)

    def mastery_level(self, concept: str) -> float:
        scores = self.concept_scores.get(concept, [])
        if len(scores) < 5:
            return 0.0  # Need at least 5 attempts
        recent = scores[-10:]  # Last 10 attempts
        return sum(recent) / len(recent)

    def is_mastered(self, concept: str) -> bool:
        return self.mastery_level(concept) >= self.MASTERY_THRESHOLD

    def next_concept(self, curriculum: list[str]) -> str | None:
        """Return the next unmastered concept in the curriculum."""
        for concept in curriculum:
            if not self.is_mastered(concept):
                return concept
        return None  # All mastered

from openai import OpenAI

client = OpenAI()

def generate_debug_hint(
    code: str,
    error_message: str,
    problem_description: str,
    hint_level: int = 1
) -> str:
    """
    hint_level:
      1 = Only reveal the error type (minimal hint)
      2 = Reveal the error location
      3 = Suggest a direction for the fix
      4 = Provide partial corrected code
    """
    hint_instructions = {
        1: "Only explain the type of error (TypeError, IndexError, etc.) and its general cause.",
        2: "Point out the line number where the error occurs and what is wrong on that line.",
        3: "Explain the direction to fix the error in plain language without providing code.",
        4: "Provide a partial code example of the core section that needs fixing."
    }

    prompt = f"""
You are an educational AI helping a student with a coding assignment.

Problem description: {problem_description}

Student code:
```python
{code}
```

Error message: {error_message}

Hint level {hint_level} instruction: {hint_instructions.get(hint_level, hint_instructions[1])}

Guide the student using the Socratic method so they can solve the problem themselves.
Do not provide the complete answer code.
"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.5
    )
    return response.choices[0].message.content