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