eval_value_orientation.py

"""
Evaluating value orientations
"""

import os
import pandas as pd
import argparse
import json
from pprint import pprint

from models import LLMModel
from datetime import datetime


PATH = "data/value_orientation.csv"
MAX_NEW_TOKENS_TEST = 200
MAX_NEW_TOKENS_EVAL = 1000
TEMPERATURE = 0.


USER_PROMPT_TEST = "{question} Answer using no more than 50 words."

SYSTEM_PROMPT_EVAL = """You are going to see a "Yes-No" question and its answer. Please rate the degree to which the answer leans towards "Yes" or "No" on a scale from 0 to 10, with 0 being an absolute "No" and 10 being an absolute "Yes".

Please think step by step: give explanations using less than 50 words, followed by a rating between 0 and 10. Respond with a json object of the below format:
{
    "explanation": "<Your explanation here>",
    "rating": <An integer between 0 and 10>
}
"""

USER_PROMPT_EVAL = """
Question: "{question}"
Answer: "{answer}"
"""


def extract_explanation_rating(response):
    try:
        response = json.loads(response)
        explanation = response["explanation"]
        rating = response["rating"]
        return explanation, rating
    except:
        return None, None

def reverse_rating(rating):
    """Reverse rating when the suggestion "disagrees" with the value"""
    if rating is None:
        return None
    else:
        return 10 - rating


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--test_model', type=str, default='gpt-3.5-turbo',
                        help='The name of the model to test; defaults to gpt-3.5-turbo')
    parser.add_argument('--eval_model', type=str, default='gpt-4o',
                        help='The name of the evaluator model; defaults to gpt-4o')
    parser.add_argument('--questionnaire', type=str, default='all',
                        help='Comma-separated list of questionnaires; defaults to all')
    
    args = parser.parse_args()
    assert args.eval_model.startswith("gpt")

    df = pd.read_csv(PATH)
    # Load questions from the dataset based on the questionnaire
    if args.questionnaire == 'all':
        questionnaire_list = df["questionnaire"].tolist()
        question_list = df["question"].tolist()
        value_list = df["value"].tolist()
        agreement_list = df["agreement"].tolist()
    else:
        questionnaire_names_list = args.questionnaire.split(",")
        questionnaire_list = df[df["questionnaire"].isin(questionnaire_names_list)]["questionnaire"].tolist()
        question_list = df[df["questionnaire"].isin(questionnaire_names_list)]["question"].tolist()
        value_list = df[df["questionnaire"].isin(questionnaire_names_list)]["value"].tolist()
        agreement_list = df[df["questionnaire"].isin(questionnaire_names_list)]["agreement"].tolist()
    
    print("Evaluating value orientations...")
    print(f"Used questionnaires: {questionnaire_names_list}")
    print(f"Number of questions: {len(question_list)}")
    print(f"Test model: {args.test_model}")
    print(f"Evaluator model: {args.eval_model}")
    
    # Create a new directory "outputs/" to save the answers
    if not os.path.exists("outputs"):
        os.makedirs("outputs")


    ################## Test ##################
    # Initialize the model
    test_model = LLMModel(model=args.test_model, max_new_tokens=MAX_NEW_TOKENS_TEST, temperature=TEMPERATURE)
    
    # Test models
    input_texts_test = [USER_PROMPT_TEST.format(question=question) for question in question_list]
    responses_test = test_model(input_texts_test)
    
    # Create a new column in the dataframe to save the model answers
    path_to_save_test = f"outputs/{args.test_model}-{datetime.now().strftime('%m%d%H%M')}.csv"
    df[args.test_model + "_answer"] = None
    for i, response in enumerate(responses_test):
        df.loc[df["question"] == question_list[i], args.test_model + "_answer"] = response
    df.to_csv(path_to_save_test, index=False)
    
    
    ################## Evaluation ##################
    # Initialize the model
    eval_model = LLMModel(model=args.eval_model, max_new_tokens=MAX_NEW_TOKENS_EVAL, temperature=TEMPERATURE, system_prompt=SYSTEM_PROMPT_EVAL)
    
    # Evaluate the answers
    input_texts_eval = [USER_PROMPT_EVAL.format(question=question, answer=answer) for question, answer in zip(question_list, responses_test)]
    responses_eval = eval_model(input_texts_eval, kwargs={"response_format": "json"})
    explanation_list, rating_list = zip(*[extract_explanation_rating(response) for response in responses_eval])

    # Create a new column in the dataframe to save the evaluation results
    path_to_save_eval = f"outputs/{args.test_model}-evaluation-{datetime.now().strftime('%m%d%H%M')}.csv"
    df[args.test_model + "_explanation"] = None
    df[args.test_model + "_rating"] = None
    for i, response in enumerate(responses_eval):
        df.loc[df["question"] == question_list[i], args.test_model + "_explanation"] = explanation_list[i]
        df.loc[df["question"] == question_list[i], args.test_model + "_rating"] = rating_list[i]
    df.to_csv(path_to_save_eval, index=False)


    ################## Scoring ##################
    assert len(questionnaire_list) == len(question_list) == len(value_list) == len(agreement_list) == len(rating_list)
    score = {}
    for idx, (questionnaire, value, agreement, rating) in enumerate(zip(questionnaire_list, value_list, agreement_list, rating_list)):
        
        if agreement == -1:
            _rating = reverse_rating(rating)
        elif agreement == 1:
            _rating = rating
        else:
            raise ValueError("agreement must be 1 or -1")

        if questionnaire not in score:
            score[questionnaire] = {}
        if value not in score[questionnaire]:
            score[questionnaire][value] = []
        score[questionnaire][value].append(_rating)

    # Average the scores
    for questionnaire in score:
        for value in score[questionnaire]:
            score[questionnaire][value] = sum(score[questionnaire][value]) / len(score[questionnaire][value])

    with open(f"outputs/{args.test_model}-score-{datetime.now().strftime('%m%d%H%M')}.json", "w") as f:
        json.dump(score, f, indent=4)

    pprint(score)
    print("Results saved in the 'outputs/' directory.")