hackaprompt/score_submission.py

import logging
import os
from typing import Dict

from fastapi.encoders import jsonable_encoder

from hackaprompt.completers import get_completer
from hackaprompt.evaluator import Response, get_evaluator
from hackaprompt.utils import init_db

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

database = init_db()

level_names = [
    "level_1",
    "level_2",
    "level_3",
    "level_4",
    "level_5",
    "level_6",
    "level_7",
    "level_8",
    "level_9",
    "level_10",
]

def log_to_db(evaluation):
    # save response to mongodb database
    try:
        submission_json = jsonable_encoder(evaluation)
        database["evaluations"].insert_one(submission_json)
        logger.info("response logged to mondogb")
    except Exception as err:
        logger.exception("Something went wrong logging to mongodb")


def evaluate_submission(submission: dict) -> Dict[str, Response]:
    evaluation: dict = {level: None for level in level_names}

    openai_api_key = os.getenv("OPENAI_API_KEY")
    assert openai_api_key is not None, "Set a valid openAI key."

    for level_name in level_names:
        logger.info(f"Evaluating {level_name}...")
        level = int(level_name[-1])

        user_input = submission.get(level_name, "")
        if user_input:
            prompt = user_input.get("prompt", "")
            model = user_input.get("model", "")

        submitted = user_input and prompt and model
        
        if submitted:

            completer = get_completer(model=model, openai_api_key=openai_api_key)

            evaluator = get_evaluator(level=level, completer=completer)
            response: Response = evaluator.evaluate(prompt)
            # TODO: Do we want to add auto-retries?
        else:
            # return an empty response instead, no need to waste GPT credits.
            response = Response(
                level=level,
                score=0,
                error=False,
            )
        evaluation[level_name] = response

    logger.info(f"evaluation done. Summary: {evaluation=}")
    return evaluation


def check_evaluation_errors(evaluation: Dict[str, Response]) -> int:
    # check for any errors during evaluation...
    evaluation_errors = False
    for level_name, response in evaluation.items():
        if response.error:
            logger.warning(f"Error encounted in {level_name} evaluation")
            evaluation_errors = True

    return evaluation_errors


def get_evaluation_total_score(evaluation: Dict[str, Response]) -> int:
    # Assumes no errors present during evaluation
    total_score = 0
    for level_name in level_names:
        if level_name in evaluation:
            score = evaluation[level_name].score
            if score != None:
                total_score += score

    return total_score