app.py

import json
import re
from openai import OpenAI
from typing import List, Dict
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
import gradio as gr
from dotenv import load_dotenv
import os
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
client = OpenAI(api_key=api_key)


# Initialize Sentence Transformer model
model = SentenceTransformer('all-MiniLM-L6-v2')  # Lightweight and efficient


def load_courses(file_path: str) -> List[Dict]:
    """
    Load course data from a JSON file.

    :param file_path: Path to the JSON file containing course data.
    :return: List of courses as dictionaries.
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            courses = json.load(f)
        print(f"Loaded {len(courses)} courses from {file_path}.")
        return courses
    except Exception as e:
        print(f"Error loading courses: {e}")
        return []


def preprocess_text(text: str) -> str:
    """
    Clean and preprocess text for better understanding.

    :param text: Original text.
    :return: Cleaned text.
    """
    # Remove excessive whitespace and special characters
    text = re.sub(r'\s+', ' ', text)
    text = text.strip()
    return text


def prepare_course_embeddings(courses: List[Dict]) -> np.ndarray:
    """
    Generate embeddings for each course based on title and description.

    :param courses: List of course dictionaries.
    :return: Numpy array of embeddings.
    """
    texts = []
    for course in courses:
        title = preprocess_text(course.get('title', ''))
        description = preprocess_text(course.get('description', ''))
        combined_text = f"{title}. {description}"
        texts.append(combined_text)

    embeddings = model.encode(texts, convert_to_numpy=True, show_progress_bar=True)
    return embeddings


def generate_llm_prompt(user_query: str, courses_subset: List[Dict]) -> str:
    """
    Create a prompt for the LLM to process the user query against the courses.

    :param user_query: The user's search query.
    :param courses_subset: A subset of courses to consider for recommendations.
    :return: A string prompt for the LLM.
    """
    formatted_courses = prepare_course_data(courses_subset)
    prompt = f"""
You are an intelligent assistant specialized in recommending educational courses.

**User Query:** "{user_query}"

**Available Courses:**
{formatted_courses}

**Task:**
Based on the user's query, provide a list of the top 5 most relevant courses from the above list. For each recommended course, include:
- **Title**
- **Reason for Recommendation:** A brief explanation of why this course is relevant to the user's query.

**Response Format:**
Please respond in JSON format as a list of objects, each containing "title" and "reason".

**Example:**
[
  {{
    "title": "Introduction to Data Science",
    "reason": "Provides foundational knowledge in data science, aligning with the user's interest in data analysis."
  }},
  ...
]
"""
    return prompt


def prepare_course_data(courses_subset: List[Dict]) -> str:
    """
    Format course data into a structured string for the LLM prompt.

    :param courses_subset: List of course dictionaries.
    :return: Formatted string representing the subset of courses.
    """
    formatted_courses = ""
    for idx, course in enumerate(courses_subset, 1):
        title = preprocess_text(course.get('title', 'No Title'))
        description = preprocess_text(course.get('description', 'No Description'))
        formatted_courses += f"{idx}. **{title}**\n{description}\n\n"
    return formatted_courses


def call_openai_api(prompt: str, max_tokens: int = 500, temperature: float = 0.3) -> str:
    """
    Call the OpenAI API with the given prompt.

    :param prompt: The prompt to send to the LLM.
    :param max_tokens: Maximum number of tokens in the response.
    :param temperature: Sampling temperature.
    :return: The assistant's response as a string.
    """
    try:
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system",
                 "content": "You are a helpful assistant that recommends relevant courses based on user queries."},
                {"role": "user", "content": prompt}
            ],
            max_tokens=max_tokens,
            temperature=temperature,
            n=1,
            stop=None
        )
        assistant_reply = response.choices[0].message.content.strip()
        return assistant_reply
    except Exception as e:
        print(f"Error communicating with OpenAI API: {e}")
        return ""


def parse_llm_response(response: str) -> List[Dict]:
    """
    Parse the LLM's JSON response into a Python list.

    :param response: The raw response string from the LLM.
    :return: List of recommended courses with titles and reasons.
    """
    try:
        # Extract JSON content from the response
        json_start = response.find('[')
        json_end = response.rfind(']') + 1
        json_str = response[json_start:json_end]
        recommendations = json.loads(json_str)
        return recommendations
    except json.JSONDecodeError:
        print("Failed to parse JSON from the LLM response.")
        print("LLM Response:")
        print(response)
        return []
    except Exception as e:
        print(f"Unexpected error during parsing: {e}")
        return []


def smart_search(user_query: str, courses: List[Dict], course_embeddings: np.ndarray, shortlist_limit: int = 50) -> \
List[Dict]:
    """
    Perform a smart search to find relevant courses based on the user's query using semantic similarity.

    :param user_query: The user's search input.
    :param courses: List of all available courses.
    :param course_embeddings: Precomputed embeddings for all courses.
    :param shortlist_limit: Number of courses to shortlist for the LLM.
    :return: List of recommended courses with titles and reasons.
    """
    if not user_query:
        print("Empty query provided.")
        return []

    # Preprocessing the user query
    clean_query = preprocess_text(user_query)

    # Generate embedding for the user query
    query_embedding = model.encode([clean_query], convert_to_numpy=True)

    # Compute cosine similarity between query and all courses
    similarities = cosine_similarity(query_embedding, course_embeddings)[0]

    # Get indices of courses sorted by similarity (highest first)
    sorted_indices = np.argsort(similarities)[::-1]

    # Shortlist top N courses based on similarity
    shortlisted_indices = sorted_indices[:shortlist_limit]
    keyword_matched_courses = [courses[idx] for idx in shortlisted_indices]

    # Generate the LLM prompt
    prompt = generate_llm_prompt(clean_query, keyword_matched_courses)

    # Call the OpenAI API
    llm_response = call_openai_api(prompt)

    if not llm_response:
        print("No response from LLM.")
        return []

    # Parse the LLM's JSON response
    recommendations = parse_llm_response(llm_response)

    return recommendations


def main_interface(user_query: str, courses: List[Dict], course_embeddings: np.ndarray) -> List[Dict]:
    """
    The main function that ties everything together for the Gradio interface.

    :param user_query: User's input query.
    :param courses: List of all courses.
    :param course_embeddings: Precomputed embeddings for all courses.
    :return: List of recommended courses with titles and reasons.
    """
    recommendations = smart_search(user_query, courses, course_embeddings)
    return recommendations


def load_data():
    """
    Load course data and generate embeddings.

    :return: Tuple of (courses list, course embeddings)
    """
    courses = load_courses('all_courses.json')
    if not courses:
        raise ValueError("No courses loaded. Please check the all_courses.json file.")
    print("Generating course embeddings...")
    course_embeddings = prepare_course_embeddings(courses)
    print("Course embeddings generated.")
    return courses, course_embeddings


# Load data once at the start
courses, course_embeddings = load_data()

# Define the Gradio interface
with gr.Blocks() as demo:
    gr.Markdown("# Analytics Vidhya Free Courses Smart Search")
    gr.Markdown("Enter your search query below to find the most relevant free courses.")

    with gr.Row():
        user_input = gr.Textbox(label="Search Query", placeholder="e.g., best llm course")

    search_button = gr.Button("Search")

    output = gr.JSON(label="**Recommended Courses:**")


    def search(query):
        recommendations = smart_search(query, courses, course_embeddings)
        if recommendations:
            return recommendations
        else:
            return {"message": "No relevant courses found for your query."}


    search_button.click(fn=search, inputs=user_input, outputs=output)

    gr.Markdown("**Powered by OpenAI's GPT-4 and Sentence Transformers**")

# Launch the app
demo.launch()