A DevOps Domain Knowledge Evaluation Benchmark for Large Language Models
from loguru import logger
from transformers import PreTrainedTokenizer


class ContextBuilder:
    '''
    Parent class
    '''
    def __init__(self):
        pass

    def make_context(
        self, 
        tokenizer, 
        query: str,
        system: str = ""
    ):
        '''
        Make context for query, default is do nothing
        '''
        raw_text = query
        context_tokens = tokenizer.encode(raw_text)
        return raw_text, context_tokens

if __name__ == '__main__':
    pass


src/context_builder/context_builder.py
luban-agi/DevOps-Eval
import json

from loguru import logger
from src.context_builder.context_builder import ContextBuilder

def get_context_builder(eval_args):
    '''
    Load context_builder by model_name
    '''
    with open(eval_args.model_conf_path, 'r') as f:
        model_conf = json.load(f)
    
    context_builder = globals()[model_conf[eval_args.model_name]['context_builder']]()
    return context_builder

class QwenChatContextBuilder(ContextBuilder):
    def __init__(self):
        super().__init__()
    
    def make_context(
        self,
        tokenizer, 
        query: str,
        system: str = "you are a helpful assistant"
    ):
        im_start, im_end = "<|im_start|>", "<|im_end|>"
        im_start_tokens = [tokenizer.im_start_id]
        im_end_tokens = [tokenizer.im_end_id]
        nl_tokens = tokenizer.encode("\n")

        def _tokenize_str(role, content):
            return f"{role}\n{content}", tokenizer.encode(
                role, allowed_special=set()
            ) + nl_tokens + tokenizer.encode(content, allowed_special=set())

        system_text, system_tokens_part = _tokenize_str("system", system)
        system_tokens = im_start_tokens + system_tokens_part + im_end_tokens

        raw_text = ""
        context_tokens = []

        context_tokens = system_tokens + context_tokens
        raw_text = f"{im_start}{system_text}{im_end}" + raw_text
        context_tokens += (
            nl_tokens
            + im_start_tokens
            + _tokenize_str("user", query)[1]
            + im_end_tokens
            + nl_tokens
            + im_start_tokens
            + tokenizer.encode("assistant")
            + nl_tokens
        )
        raw_text += f"\n{im_start}user\n{query}{im_end}\n{im_start}assistant\n"
        return raw_text, context_tokens


if __name__ == '__main__':
    query = '你好'
    system = '请帮助我'
    tokenizer = '/mnt/llm/devopspal/model/Qwen-7B'


src/context_builder/context_builder_family.py
luban-agi/DevOps-Eval
import torch
import copy

from loguru import logger


def evaluate(model, tokenizer, context_builder, all_dataset):
    '''
    Evaluate on all_dataset
    '''
    all_dataset_pred = copy.deepcopy(all_dataset)
    for dataset_name, dataset in all_dataset_pred.items():
        for question in dataset:
            question['pred'] = get_pred(model, tokenizer, context_builder, question)
    return all_dataset_pred

def get_pred(model, tokenizer, context_builder, question: dict):
    '''
    Get the prediction for single question
    '''
    options = question['options']
    query = question['query']

    option_dict = {}
    for option in options:
        option_dict[option] = tokenizer.encode(option)
    # logger.debug('option_dict={}'.format(option_dict))

    # build context
    raw_text, context_tokens = context_builder.make_context(tokenizer, query)
    input_ids = torch.tensor([context_tokens]).to(model.device)
    # logger.debug('raw_text={}\ncontext_tokens={}'.format(raw_text, context_tokens))

    # feed to the model
    output = model(input_ids)
    logits = output.logits

    # get pred option
    score_dict = {}
    for option in option_dict:
        score = logits[0][0][option_dict[option]]
        score_dict[option] = float(score)
    # logger.debug('score_dict={}'.format(score_dict))

    max_score = float('-inf')
    best_option = None
    for option, score in score_dict.items():
        if score > max_score:
            max_score = score
            best_option = option
    # logger.debug('max_score={}, best_option={}'.format(max_score, best_option))
    return best_option


src/evaluate/evaluate.py
luban-agi/DevOps-Eval
import os
cur_dir = os.path.split(os.path.realpath(__file__))[0]
root_dir = os.path.abspath(os.path.dirname(cur_dir) + os.path.sep + "..")

import json
from typing import List, Literal, Optional
from dataclasses import dataclass, field

from loguru import logger


@dataclass
class EvaluateArguments:
    r"""
    Arguments
    """
    model_path: str = field(
        metadata={"help": "Path of the model and tokenizer"}
    )
    model_name: str = field(
        metadata={"help": "Name of the model, now support qwen, baichuan, ..."}
    )
    model_conf_path: str = field(
        default = root_dir + os.path.sep + 'conf' + os.path.sep + 'model_conf.json',
        metadata={"help": "Path of model's loader and context_builder class"}
    )
    eval_dataset_list: str = field(
        default = 'all',
        metadata={"help": "Which datasets to evaluate on. default is all datasets, if want to test on multiple datasets, use # as seperator"}
    )
    eval_dataset_type: str = field(
        default = 'test',
        metadata={"help": "Which type of datasets to evaluate on. default is test, must be one of (test, valid, dev)"}
    )
    eval_dataset_fp_conf_path: str = field(
        default = root_dir + os.path.sep + 'conf' + os.path.sep + 'dataset_fp.json',
        metadata={"help": "Path of dataset_name and filepath config file"}
    )
    k_shot: int = field(
        default = 0,
        metadata={"help": "k-shot test, k should be in (0, 1,2,3,4,5)]"}
    )

    def init_for_training(self):
        self.eval_dataset_list = self.eval_dataset_list.split('#')
        if 'all' in self.eval_dataset_list:
            logger.info('Detecting all in eval_dataset_list, evaluating on all dataset')
            self.eval_dataset_list = ['all']
        
        assert self.eval_dataset_type in ('dev', 'test', 'val')
        assert self.k_shot in (0,1,2,3,4,5)
src/hparams/evaluate_args.py
luban-agi/DevOps-Eval
import sys
sys.path.append('/mnt/llm/DevOpsEval')

from loguru import logger
from transformers import HfArgumentParser
from src.hparams.evaluate_args import EvaluateArguments


def get_all_args(
    args = None
) -> EvaluateArguments:
    parser = HfArgumentParser((
        EvaluateArguments
    ))
    eval_args = parser.parse_args_into_dataclasses(args)[0]
    eval_args.init_for_training()
    return eval_args


if __name__ == '__main__':
    a = get_all_args()
    logger.debug(a)
src/hparams/parser.py
luban-agi/DevOps-Eval
def get_acc_score(all_pred):
    '''
    Get accuracy score by dataset
    '''
    score_dict = {i:None for i in all_pred}
    total_corr = 0
    total_count = 0
    for dataset_name, dataset_pred in all_pred.items():
        corr = 0

        for pred_single in dataset_pred:
            if pred_single['answer'] == pred_single['pred']:
                corr += 1
        score_dict[dataset_name] = {
            'total': len(dataset_pred),
            'corr': corr,
            'score': corr / len(dataset_pred)
        }

        total_corr += corr
        total_count += len(dataset_pred)
    res = {
        'total': total_count,
        'corr': total_corr,
        'score': total_corr / total_count,
        'detail': score_dict
    }
    return res

src/metric/metric_score.py
luban-agi/DevOps-Eval
import torch
import transformers
from transformers import AutoTokenizer, AutoModelForCausalLM, GenerationConfig


class ModelAndTokenizerLoader:
    def __init__(self):
        pass

    def load_model_and_tokenizer(self, model_path: str):
        model = self.load_model(model_path)
        tokenizer = self.load_tokenizer(model_path)
        return model, tokenizer

    def load_model(self, model_path: str):
        model =  AutoModelForCausalLM.from_pretrained(model_path, device_map="auto", trust_remote_code=True).eval()
        return model

    def load_tokenizer(self, model_path: str):
        tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
        return tokenizer


src/model_and_tokenizer_loader/model_and_tokenizer_loader.py
luban-agi/DevOps-Eval
import json

from loguru import logger

import torch
import transformers
from transformers import AutoTokenizer, AutoModelForCausalLM, GenerationConfig

from src.model_and_tokenizer_loader.model_and_tokenizer_loader import ModelAndTokenizerLoader

def load_model_and_tokenizer(eval_args):
    '''
    Load model and tokenizer by model_path and model_name
    '''
    with open(eval_args.model_conf_path, 'r') as f:
        model_conf = json.load(f)
    
    loader = globals()[model_conf[eval_args.model_name]['loader']]()

    return loader.load_model_and_tokenizer(eval_args.model_path)

class QwenModelAndTokenizerLoader(ModelAndTokenizerLoader):
    def __init__(self):
        super().__init__()
        pass

    def load_model(self, model_path: str):
        model = super().load_model(model_path)

        return model

    def load_tokenizer(self, model_path: str):
        tokenizer = super().load_tokenizer(model_path)

        # read generation config
        with open(model_path + '/generation_config.json', 'r') as f:
            generation_config = json.load(f)
        
        tokenizer.pad_token_id = generation_config['pad_token_id']
        tokenizer.eos_token_id = generation_config['eos_token_id']
        return tokenizer


if __name__ == '__main__':
    model_path = '/mnt/llm/devopspal/model/Qwen-7B'
    qwen_model_loader = QwenModelAndTokenizerLoader()
    tokenizer = qwen_model_loader.load_tokenizer(model_path)

    logger.info(tokenizer)

src/model_and_tokenizer_loader/model_and_tokenizer_loader_family.py
luban-agi/DevOps-Eval
from loguru import logger

from src.hparams.parser import get_all_args
from src.data.data_load import load_all_dataset
from src.model_and_tokenizer_loader.model_and_tokenizer_loader_family import load_model_and_tokenizer
from src.context_builder.context_builder_family import get_context_builder
from src.evaluate.evaluate import evaluate
from src.metric.metric_score import get_acc_score

def run_eval(args=None):
    # Get eval args
    eval_args = get_all_args(args)
    logger.info('eval_args={}'.format(eval_args))

    # Get all dataset
    eval_datasets = load_all_dataset(eval_args)
    logger.info('Load all dataset success')

    # Load model and tokenizer
    model, tokenizer = load_model_and_tokenizer(eval_args)
    logger.info('Load model and tokenizer success')
    logger.info('tokenizer={}'.format(tokenizer))

    # load context_builder
    context_builder = get_context_builder(eval_args)
    logger.info('context_builder={}'.format(context_builder))

    # run model
    all_pred = evaluate(model, tokenizer, context_builder, eval_datasets)

    # get metric
    score_dict = get_acc_score(all_pred)
    logger.info('Evaluation result={}'.format(score_dict))


if __name__ == '__main__':
    run_eval()


src/run_eval.py
luban-agi/DevOps-Eval