learning_objective_generator/base_generation.py

from typing import List
from openai import OpenAI
import re
import os
from models import BaseLearningObjective, BaseLearningObjectiveWithoutCorrectAnswer, BaseLearningObjectivesWithoutCorrectAnswerResponse, TEMPERATURE_UNAVAILABLE
from prompts.learning_objectives import BASE_LEARNING_OBJECTIVES_PROMPT, BLOOMS_TAXONOMY_LEVELS, LEARNING_OBJECTIVE_EXAMPLES, LEARNING_OBJECTIVE_EXAMPLES_WITHOUT_ANSWERS


def _get_run_manager():
    """Get run manager if available, otherwise return None."""
    try:
        from ui.run_manager import get_run_manager
        return get_run_manager()
    except:
        return None

def generate_base_learning_objectives(client: OpenAI, model: str, temperature: float, file_contents: List[str], num_objectives: int) -> List[BaseLearningObjective]:
    """
    Generate learning objectives with correct answers by first generating the objectives and then adding correct answers.
    This is a wrapper function that calls the two separate functions for better separation of concerns.
    """
    print(f"Generating {num_objectives} learning objectives from {len(file_contents)} files")
    
    # First, generate the learning objectives without correct answers
    objectives_without_answers = generate_base_learning_objectives_without_correct_answers(
        client, model, temperature, file_contents, num_objectives
    )
    
    # Then, generate correct answers for those objectives
    objectives_with_answers = generate_correct_answers_for_objectives(
        client, model, temperature, file_contents, objectives_without_answers
    )
    
    return objectives_with_answers

def generate_base_learning_objectives_without_correct_answers(client: OpenAI, model: str, temperature: float, file_contents: List[str], num_objectives: int) -> List[BaseLearningObjectiveWithoutCorrectAnswer]:
    """Generate learning objectives without correct answers from course content."""
    # Extract the source filenames for reference
    sources = set()
    for file_content in file_contents:
        source_match = re.search(r"<source file='([^']+)'>", file_content)
        if source_match:
            source = source_match.group(1)
            sources.add(source)
            print(f"DEBUG - Found source file: {source}")
    print(f"DEBUG - Found {len(sources)} source files: {sources}")
    print(f"DEBUG - Using {len(file_contents)} files for learning objectives")
    combined_content = "\n\n".join(file_contents)
    prompt = f"""
    You are an expert educational content creator specializing in creating precise, relevant learning objectives from course materials. Based on the following course content, generate {num_objectives} clear and concise learning objectives.

    
    {BASE_LEARNING_OBJECTIVES_PROMPT}

    Consider Bloom's taxonomy in the context of the learning objective you are writing and choose the appropriate framing for the question
    and answer options in the context of Bloom's taxonomy.
    <BloomsTaxonomyLevels>
    {BLOOMS_TAXONOMY_LEVELS}
    </BloomsTaxonomyLevels>

    Format your response like this, according to the data model provided for each objective:
    ```json
    {{
    id: int = Unique identifier for the learning objective,
    learning_objective: str = the learning objective text,
    source_reference: Union[List[str], str] = List of paths to the files from which this learning objective was extracted,
    }}
    ```

    Here is an example of high quality learning objectives:
    <learning objectives>
    {LEARNING_OBJECTIVE_EXAMPLES_WITHOUT_ANSWERS}
    </learning objectives>

    Below is the course content. The source references are embedded in xml tags within the context.
    <course content>
    {combined_content}
    </course content>
    """
    try:
        # Use OpenAI beta API for structured output

        params = {
            "model": model,
            "messages": [
                {"role": "system", "content": "You are an expert educational content creator specializing in creating precise, relevant learning objectives from course materials."},
                {"role": "user", "content": prompt}
            ],
            "response_format": BaseLearningObjectivesWithoutCorrectAnswerResponse
        }
        if not TEMPERATURE_UNAVAILABLE.get(model, True):
            params["temperature"] = temperature

        completion = client.beta.chat.completions.parse(**params)
        response = completion.choices[0].message.parsed.objectives
        # Assign IDs and format source_reference
        for i, objective in enumerate(response):
            objective.id = i + 1
            if isinstance(objective.source_reference, str):
                if "," in objective.source_reference:
                    source_refs = [os.path.basename(src.strip()) for src in objective.source_reference.split(",")]
                    objective.source_reference = source_refs
                else:
                    objective.source_reference = os.path.basename(objective.source_reference)
            elif isinstance(objective.source_reference, list):
                objective.source_reference = [os.path.basename(src) for src in objective.source_reference]
        print(f"Successfully generated {len(response)} learning objectives without correct answers")
        return response
    except Exception as e:
        print(f"Error generating learning objectives without correct answers: {e}")
        # Re-raise the exception instead of generating fallbacks
        raise

def find_sources_for_user_objectives(
    client: OpenAI,
    model: str,
    temperature: float,
    file_contents: List[str],
    objective_texts: List[str]
) -> List[BaseLearningObjectiveWithoutCorrectAnswer]:
    """
    Find relevant source files in course materials for user-provided learning objectives.

    For each objective text, asks the LLM to identify which source file(s) contain the
    most relevant information, using the same source-tag convention as the rest of the pipeline.

    Returns:
        List of BaseLearningObjectiveWithoutCorrectAnswer objects with source_reference populated.
    """
    combined_content = "\n\n".join(file_contents)

    # Collect available source filenames from XML tags
    available_sources = []
    for content in file_contents:
        match = re.search(r"<source file='([^']+)'>", content)
        if match:
            available_sources.append(os.path.basename(match.group(1)))

    print(f"Finding sources for {len(objective_texts)} user-provided objectives across {len(available_sources)} source files: {available_sources}")

    result = []
    for i, obj_text in enumerate(objective_texts):
        print(f"Finding source for objective {i+1}/{len(objective_texts)}: {obj_text[:60]}...")

        prompt = f"""You are an expert educational content analyst.

Given the following learning objective and course content, identify which source file(s) contain the most relevant information for this learning objective.

Learning Objective: {obj_text}

Available source files: {available_sources}

Course Content:
{combined_content}

Return ONLY the filename(s) most relevant to this learning objective. If multiple files are relevant, separate them with commas. Return just the filename(s), nothing else. Use exact filenames from the available source files list."""

        source_refs = available_sources if available_sources else []

        try:
            params = {
                "model": model,
                "messages": [
                    {"role": "system", "content": "You are an expert educational content analyst. Identify which source files contain information relevant to a given learning objective. Return only filename(s)."},
                    {"role": "user", "content": prompt}
                ]
            }
            if not TEMPERATURE_UNAVAILABLE.get(model, True):
                params["temperature"] = temperature

            completion = client.chat.completions.create(**params)
            source_text = completion.choices[0].message.content.strip()

            if "," in source_text:
                source_refs = [os.path.basename(s.strip()) for s in source_text.split(",")]
            else:
                source_refs = os.path.basename(source_text)

        except Exception as e:
            print(f"Error finding sources for objective '{obj_text[:50]}': {e}")

        result.append(BaseLearningObjectiveWithoutCorrectAnswer(
            id=i + 1,
            learning_objective=obj_text,
            source_reference=source_refs
        ))

    return result


def generate_correct_answers_for_objectives(client: OpenAI, model: str, temperature: float, file_contents: List[str], objectives_without_answers: List[BaseLearningObjectiveWithoutCorrectAnswer]) -> List[BaseLearningObjective]:
    """Generate correct answers for the given learning objectives."""
    combined_content = "\n\n".join(file_contents)
    run_manager = _get_run_manager()

    # Create a list to store the objectives with answers
    objectives_with_answers = []

    # Process each objective to generate a correct answer
    for objective in objectives_without_answers:
        prompt = f"""
        You are an expert educational content creator specializing in creating precise, relevant, and concise correct answers for learning objectives. 
        

        Use the below learning objective to generate the correct answer:

        <learning_objective>
        "id": {objective.id},
        "learning_objective": "{objective.learning_objective}",
        "source_reference": "{objective.source_reference}"
        </learning_objective>


        Use the below course content to generate the correct answer:
        <course_content>
        {combined_content}
        </course_content>




        Please provide a clear, concise, and accurate correct answer for this learning objective. The answer should be:
        1. Directly answering the learning objective
        2. Concise (preferably under 20 words). Avoids unnecessary length. See example below on avoiding unnecessary length.
        3. Focused on the core concept without unnecessary elaboration
        4. Based on the course content provided

        Format your response as a plain text answer only, without any additional explanation or formatting.

        Here are examples of high quality learning objective examples with correct answers:

        <learning_objective_examples>
        {LEARNING_OBJECTIVE_EXAMPLES}
        </learning_objective_examples>

        """
        
        try:
            params = {
                "model": model,
                "messages": [
                    {"role": "system", "content": "You are an expert educational content creator specializing in creating precise, relevant correct answers for learning objectives."},
                    {"role": "user", "content": prompt}
                ]
            }
            if not TEMPERATURE_UNAVAILABLE.get(model, True):
                params["temperature"] = temperature
                
            completion = client.chat.completions.create(**params)
            correct_answer = completion.choices[0].message.content.strip()
            
            # Create a new BaseLearningObjective with the correct answer
            objective_with_answer = BaseLearningObjective(
                id=objective.id,
                learning_objective=objective.learning_objective,
                source_reference=objective.source_reference,
                correct_answer=correct_answer
            )
            
            objectives_with_answers.append(objective_with_answer)
            if run_manager:
                run_manager.log(f"Generated correct answer for objective {objective.id}", level="INFO")

        except Exception as e:
            if run_manager:
                run_manager.log(f"Error generating correct answer for objective {objective.id}: {e}", level="ERROR")
            # Create an objective with an error message as the correct answer
            objective_with_answer = BaseLearningObjective(
                id=objective.id,
                learning_objective=objective.learning_objective,
                source_reference=objective.source_reference,
                correct_answer="[Error generating correct answer]"
            )
            objectives_with_answers.append(objective_with_answer)
    
    print(f"Successfully generated correct answers for {len(objectives_with_answers)} learning objectives")
    return objectives_with_answers