eval_bittensor.py

import collections
import lm_eval.tasks
import random
import time
from datetime import datetime as dt
import bittensor as bt
from tqdm import tqdm
import json

import http.client
import os
from argparse import ArgumentParser

parser = ArgumentParser()

parser.add_argument("--validator", required=True, type=str, help="validator name", choices=["opentensor_foundation", "taostats"], default="float16")
args = parser.parse_args()

default_prompt = '''
You are Chattensor.
Chattensor is a research project by Opentensor Cortex.
Chattensor is designed to be able to assist with a wide range of tasks, from answering simple questions to providing in-depth explanations and discussions on a wide range of topics. As a language model, Chattensor is able to generate human-like text based on the input it receives, allowing it to engage in natural-sounding conversations and provide responses that are coherent and relevant to the topic at hand.
'''

if args.validator == "taostats":
    print("TAOOOOSATATS")
    try:
        bitapai_key = os.environ["BITAPAI_KEY"]
        conn = http.client.HTTPSConnection("dashboard.bitapai.io")
        headers = {
        'Content-Type': 'application/json',
        'X-API-KEY': bitapai_key
        }
    except KeyError:
        raise RuntimeError(f"BITAPAI_KEY does not exist and chosen validator is taostats. Please set your bitapai key using export BITAPAI_KEY=x.")
        

def get_response(prompt):
    if args.validator == "taostats":
        payload = json.dumps({
        "system": default_prompt,
        "user": prompt
        })
        conn.request("POST", "/api/v1/prompt", payload, headers)
        res = conn.getresponse()
        data = res.read()
        # print('test')
        print(data)
        time.sleep(1)
        return data.decode("utf-8")
    else:
        return bt.prompt(prompt)

# Load all the LMEH tasks
tasks = ["hellaswag", "arc_challenge", "truthfulqa_mc", "hendrycksTest-abstract_algebra", "hendrycksTest-anatomy", "hendrycksTest-astronomy", "hendrycksTest-business_ethics", "hendrycksTest-clinical_knowledge", "hendrycksTest-college_biology", "hendrycksTest-college_chemistry", "hendrycksTest-college_computer_science", "hendrycksTest-college_mathematics", "hendrycksTest-college_medicine", "hendrycksTest-college_physics", "hendrycksTest-computer_security", "hendrycksTest-conceptual_physics", "hendrycksTest-econometrics", "hendrycksTest-electrical_engineering", "hendrycksTest-elementary_mathematics", "hendrycksTest-formal_logic", "hendrycksTest-global_facts", "hendrycksTest-high_school_biology", "hendrycksTest-high_school_chemistry", "hendrycksTest-high_school_computer_science", "hendrycksTest-high_school_european_history", "hendrycksTest-high_school_geography", "hendrycksTest-high_school_government_and_politics", "hendrycksTest-high_school_macroeconomics", "hendrycksTest-high_school_mathematics", "hendrycksTest-high_school_microeconomics", "hendrycksTest-high_school_physics", "hendrycksTest-high_school_psychology", "hendrycksTest-high_school_statistics", "hendrycksTest-high_school_us_history", "hendrycksTest-high_school_world_history", "hendrycksTest-human_aging", "hendrycksTest-human_sexuality", "hendrycksTest-international_law", "hendrycksTest-jurisprudence", "hendrycksTest-logical_fallacies", "hendrycksTest-machine_learning", "hendrycksTest-management", "hendrycksTest-marketing", "hendrycksTest-medical_genetics", "hendrycksTest-miscellaneous", "hendrycksTest-moral_disputes", "hendrycksTest-moral_scenarios", "hendrycksTest-nutrition", "hendrycksTest-philosophy", "hendrycksTest-prehistory", "hendrycksTest-professional_accounting", "hendrycksTest-professional_law", "hendrycksTest-professional_medicine", "hendrycksTest-professional_psychology", "hendrycksTest-public_relations", "hendrycksTest-security_studies", "hendrycksTest-sociology", "hendrycksTest-us_foreign_policy", "hendrycksTest-virology", "hendrycksTest-world_religions"]
task_dict = lm_eval.tasks.get_task_dict(tasks)
task_dict_items = [
    (name, task)
    for name, task in task_dict.items()
    if (task.has_validation_docs() or task.has_test_docs())
]
versions = collections.defaultdict(dict)

# get lists of each type of request
for task_name, task in task_dict_items:
    versions[task_name] = task.VERSION
    # default to test doc, fall back to val doc if validation unavailable
    # TODO: the test-fallback-to-val system isn't final, we should revisit it at some point
    if task.has_test_docs():
        task_doc_func = task.test_docs
        task_set = "test"  # Required for caching in the decontamination
    elif task.has_validation_docs():
        task_set = "val"  # Required for caching in the decontamination
        task_doc_func = task.validation_docs
    else:
        raise RuntimeError("Task has neither test_docs nor validation_docs")
        # deterministically shuffle docs and chop off the first `limit` because sometimes docs are in some kind of order
    task_docs = list(task_doc_func())
    rnd = random.Random()
    rnd.seed(42)
    rnd.shuffle(task_docs)

    i=0
    for task_doc in tqdm(task_docs):
        print(task_name)
        print(task_doc)
        if ("result" in task_doc) and ("inference_time" in task_doc) and ("prompt" in task_doc) and ("result" in task_doc) and (task_doc['result'] != ""):
            continue

        query = task_doc["query"] if "query" in task_doc else ""
        choices_list = "\n".join([str(number+1) + ". " + choice for number, choice in enumerate(task_doc["choices"])]) if "choices" in task_doc else ""
        number_list = ",".join([str(number) for number in range(1,len(task_doc["choices"])+1)]) if "choices" in task_doc else ""

        if (task_name == "hellaswag") :
            prompt = ""
            prompt_list = list(task.training_docs())[:10]
            for prompt_item in prompt_list:
                prompt_item_query = prompt_item["query"]
                prompt_item_choices_list = "\n".join([str(number+1) + ". " + choice for number, choice in enumerate(prompt_item["choices"])])
                prompt_item_number_list = ",".join([str(number) for number in range(1,len(prompt_item["choices"])+1)])
                prompt_item_gold = prompt_item["gold"]+1

                prompt += f"""{prompt_item_query}...\n{prompt_item_choices_list}\nRespond with just one number only: {prompt_item_number_list}.\n{prompt_item_gold}\n\n"""

            prompt += f"""{query}...\n{choices_list}\nRespond with just one number only: {number_list}. """
        
        elif (task_name == "arc_challenge"):
            prompt = ""
            prompt_list = list(task.training_docs())[:25]
            for prompt_item in prompt_list:
                prompt_item_query = prompt_item["query"]
                prompt_item_choices_list = "\n".join([str(number+1) + ". " + choice for number, choice in enumerate(prompt_item["choices"])])
                prompt_item_number_list = ",".join([str(number) for number in range(1,len(prompt_item["choices"])+1)])
                prompt_item_gold = prompt_item["gold"]+1

                prompt += f"""{prompt_item_query}...\n{prompt_item_choices_list}\nRespond with just one number only: {prompt_item_number_list}.\n{prompt_item_gold}\n\n"""

            prompt += f"""{query}...\n{choices_list}\nRespond with just one number only: {number_list}. """


        elif (task_name == "truthfulqa_mc"):
            continue
            prompt = ""
            
        elif ("hendrycksTest" in task_name):
            prompt = ""
            prompt_list = list(task.test_docs())[:5]
            for prompt_item in prompt_list:
                prompt_item_query = prompt_item["query"]

                prompt += f"""{prompt_item_query.replace("Answer:", "Respond with just one letter only: A, B, C, D:")}\n{["A", "B", "C", "D"][prompt_item["gold"]]}\n\n"""

            prompt += query.replace("Answer:", "Respond with just one letter only: A, B, C, D:")

        # print(prompt)

        start = time.time()
        task_doc["result"] = get_response(prompt)
        end = time.time()
        task_doc["inference_time"] = end - start
        task_doc["prompt"] = prompt
        task_doc["datetime"] = dt.now().strftime(format = "%Y-%m-%d %H:%M:%S")
        print(task_doc["result"])

        i = i + 1
        if ((i % 100) / 1000 == 0):
            with open(f"""_results/few-shot/{args.validator}/{task_name}_results.json""", "w") as final:
                json.dump(task_docs, final)

    with open(f"""_results/few-shot/{args.validator}/{task_name}_results.json""", "w") as final:
        json.dump(task_docs, final)