initial commit
This commit is contained in:
extio-de
256
Validator.py
Normal file
256
Validator.py
Normal file
@@ -0,0 +1,256 @@
|
||||
import torch
|
||||
import gc
|
||||
import json
|
||||
|
||||
from Context import Context
|
||||
from ModelLoader import ModelLoader
|
||||
from Dataset import Dataset
|
||||
from transformers import AutoTokenizer
|
||||
from peft import PeftModel
|
||||
from dataclasses import dataclass
|
||||
|
||||
class Validator:
|
||||
peftTokenizer = None
|
||||
peftModel = None
|
||||
extPeftTokenizer = None
|
||||
extPeftModel = None
|
||||
graderTokenizer = None
|
||||
graderModel = None
|
||||
statistics = []
|
||||
|
||||
def validate(self, context: Context):
|
||||
if not context.validate or context.vInplace:
|
||||
return True
|
||||
if not context.storeAdapter:
|
||||
raise Exception("Configuration problem: validate requires storeAdapter")
|
||||
|
||||
print("Validation")
|
||||
|
||||
validations = None
|
||||
self.__loadPeftModel(context)
|
||||
try:
|
||||
validations = self.__askPeftModel(context)
|
||||
finally:
|
||||
self.unload(context)
|
||||
|
||||
self.__loadGraderModel(context)
|
||||
try:
|
||||
return self.__grade(validations, context)
|
||||
finally:
|
||||
self.unload(context)
|
||||
|
||||
def validateInPlace(self, peftTokenizer, peftModel, context: Context):
|
||||
if not context.validate or not context.vInplace:
|
||||
return True
|
||||
|
||||
print("In-Place validation")
|
||||
|
||||
self.extPeftTokenizer = peftTokenizer
|
||||
self.extPeftModel = peftModel
|
||||
validations = self.__askPeftModel(context)
|
||||
|
||||
if self.graderModel == None:
|
||||
self.__loadGraderModel(context)
|
||||
return self.__grade(validations, context)
|
||||
|
||||
def __loadPeftModel(self, context: Context):
|
||||
self.peftTokenizer = AutoTokenizer.from_pretrained(context.locBaseModel, trust_remote_code = True)
|
||||
if self.peftTokenizer.pad_token == None:
|
||||
self.peftTokenizer.pad_token = self.peftTokenizer.eos_token
|
||||
|
||||
baseModel = ModelLoader().load(context.locBaseModel, False, context.vQuantModel, context)
|
||||
self.peftModel = PeftModel.from_pretrained(baseModel, context.locAdapter)
|
||||
|
||||
def __askPeftModel(self, context: Context):
|
||||
tokenizer = self.extPeftTokenizer or self.peftTokenizer
|
||||
model = self.extPeftModel or self.peftModel
|
||||
|
||||
result = Validations([], 0)
|
||||
|
||||
dataset = Dataset().loadDataDir("json", context.locValidation, context)
|
||||
|
||||
print("Asking " + str(dataset["validation"].num_rows * context.vPasses) + " questions...")
|
||||
|
||||
noQuestion = 0
|
||||
result.total = 0
|
||||
for record in dataset["validation"]:
|
||||
answers = []
|
||||
for _ in range(0, context.vPasses):
|
||||
prompt = ""
|
||||
userPromptType = ""
|
||||
trimPrompt = True
|
||||
|
||||
userPromptType = "chat"
|
||||
userPrompt = record[userPromptType]
|
||||
if userPrompt:
|
||||
chat = [
|
||||
{"role": "user", "content": f"{userPrompt}"}
|
||||
]
|
||||
prompt = tokenizer.apply_chat_template(chat, add_generation_prompt = True, tokenize = False)
|
||||
else:
|
||||
userPromptType = "chatCompletion"
|
||||
userPrompt = record[userPromptType]
|
||||
if userPrompt:
|
||||
chat = [
|
||||
{"role": "user", "content": f"{userPrompt}"}
|
||||
]
|
||||
prompt = tokenizer.apply_chat_template(chat, continue_final_message = True, tokenize = False)
|
||||
else:
|
||||
userPromptType = "completion"
|
||||
userPrompt = prompt = record[userPromptType]
|
||||
trimPrompt = False
|
||||
if not userPrompt:
|
||||
continue
|
||||
|
||||
tokenized = tokenizer(prompt, return_tensors="pt", return_attention_mask = True, add_special_tokens = False).to(context.device)
|
||||
|
||||
outputs = model.generate(**tokenized, max_new_tokens = context.vGenMaxTokens, pad_token_id = tokenizer.eos_token_id)
|
||||
|
||||
output = str(tokenizer.decode(outputs[0]))
|
||||
if trimPrompt:
|
||||
output = output.replace(prompt, "")
|
||||
if tokenizer.eos_token:
|
||||
output = output.replace(tokenizer.eos_token, "")
|
||||
if tokenizer.pad_token:
|
||||
output = output.replace(tokenizer.pad_token, "").rstrip()
|
||||
|
||||
answers.append(output)
|
||||
|
||||
noQuestion += 1
|
||||
print(noQuestion, end = " " if noQuestion%10>0 else "\n", flush = True)
|
||||
|
||||
result.total += len(answers) * (len(record["oneOf"] or []) + len(record["string"] or []))
|
||||
result.validations.append(Validation(userPrompt, userPromptType, answers, record["oneOf"], record["string"]))
|
||||
|
||||
print("")
|
||||
|
||||
return result
|
||||
|
||||
def __loadGraderModel(self, context: Context):
|
||||
self.graderTokenizer = AutoTokenizer.from_pretrained(context.locGraderModel, trust_remote_code = True)
|
||||
if self.graderTokenizer.pad_token == None:
|
||||
self.graderTokenizer.pad_token = self.graderTokenizer.eos_token
|
||||
self.graderModel = ModelLoader().load(context.locGraderModel, context.vGraderOnCpu, context.vQuantGrader, context)
|
||||
|
||||
def __grade(self, validations, context: Context):
|
||||
print("Grading...")
|
||||
|
||||
cntEvaluations = 0
|
||||
cntPassed = 0
|
||||
cntValidations = 0
|
||||
cntCompletion = 0
|
||||
|
||||
for validation in validations.validations:
|
||||
cntCompletion += 1
|
||||
print(f"{validation.userPromptType} {cntCompletion}/{len(validations.validations)}: {validation.userPrompt}")
|
||||
|
||||
for answer in validation.answers:
|
||||
print(f" * {answer}")
|
||||
|
||||
if len(validation.evalOneOfs) > 0:
|
||||
cntEvaluations += 1
|
||||
passed = False
|
||||
for gradingQuestion in validation.evalOneOfs:
|
||||
cntValidations += 1
|
||||
print(f" > ({cntValidations}/{validations.total}) {gradingQuestion}")
|
||||
|
||||
chat = [
|
||||
{"role": "system", "content": """You are a grader that evaluates the relevance of a given text to a user question.
|
||||
Please provide a binary response 'true' or 'false' for the following text.
|
||||
'true' means that the text provides a truthful answer to the question, while 'false' means that it does not.
|
||||
Provide no preamble and a short explanation. Return the response in JSON format with the following field: "passed", "explanation\""""},
|
||||
{"role": "user", "content": f"""The text is: '{answer}'
|
||||
The question is: '{gradingQuestion}'"""}
|
||||
]
|
||||
prompt = self.graderTokenizer.apply_chat_template(chat, add_generation_prompt = True, tokenize = False)
|
||||
tokenized = self.graderTokenizer(prompt, return_tensors="pt", return_attention_mask = True, add_special_tokens = False)
|
||||
if not context.vGraderOnCpu:
|
||||
tokenized = tokenized.to(context.device)
|
||||
|
||||
outputs = self.graderModel.generate(**tokenized, max_new_tokens = context.vGenMaxTokens, pad_token_id = self.graderTokenizer.eos_token_id)
|
||||
|
||||
output = self.graderTokenizer.decode(outputs[0])
|
||||
output = output.replace(prompt, "")
|
||||
if self.graderTokenizer.eos_token:
|
||||
output = output.replace(self.graderTokenizer.eos_token, "")
|
||||
if self.graderTokenizer.pad_token:
|
||||
output = output.replace(self.graderTokenizer.pad_token, "").rstrip()
|
||||
|
||||
questionPassed = False
|
||||
try:
|
||||
processed = output[output.index("{") : output.index("}") + 1]
|
||||
print(processed)
|
||||
parsed = json.loads(processed)
|
||||
questionPassed = "true" in str(parsed["passed"]).lower()
|
||||
|
||||
except:
|
||||
if "\"passed\"" in output.lower():
|
||||
print(output)
|
||||
print(f" -> Warning: Json malformed but parseable")
|
||||
questionPassed = any(field in output.lower() for field in ("\"passed\": true", "\"passed\":true", "\"passed\": \"true\"", "\"passed\":\"true\""))
|
||||
|
||||
else:
|
||||
print(f" -> ERROR: Cannot parse: {output}")
|
||||
|
||||
passed |= questionPassed
|
||||
print(" -> " + ("PASSED" if questionPassed else "FAILED"))
|
||||
|
||||
print(" => " + ("PASSED" if passed else "FAILED"))
|
||||
if passed:
|
||||
cntPassed += 1
|
||||
|
||||
if len(validation.evalStrings) > 0:
|
||||
cntEvaluations += 1
|
||||
passed = False
|
||||
for evalString in validation.evalStrings:
|
||||
cntValidations += 1
|
||||
print(f" > ({cntValidations}/{validations.total}) Contains string {evalString}")
|
||||
passed |= evalString.lower() in answer.lower()
|
||||
print(" => " + ("PASSED" if passed else "FAILED"))
|
||||
if passed:
|
||||
cntPassed += 1
|
||||
|
||||
passedPerc = int(float(cntPassed) / float(cntEvaluations) * 100.0)
|
||||
result = passedPerc >= context.vExpected
|
||||
self.statistics.append((result, passedPerc))
|
||||
|
||||
print("")
|
||||
print("#############")
|
||||
print("Final result: " + str(cntPassed) + "/" + str(cntEvaluations) + " -> " + str(passedPerc) + "%")
|
||||
print("--> " + ("PASSED" if result else "FAILED"))
|
||||
|
||||
return result
|
||||
|
||||
def unload(self, context: Context):
|
||||
if self.peftTokenizer != None:
|
||||
del self.peftTokenizer
|
||||
self.peftTokenizer = None
|
||||
if self.peftModel != None:
|
||||
self.peftModel.unload()
|
||||
del self.peftModel
|
||||
self.peftModel = None
|
||||
if self.graderTokenizer != None:
|
||||
del self.graderTokenizer
|
||||
self.graderTokenizer = None
|
||||
if self.graderModel != None:
|
||||
del self.graderModel
|
||||
self.graderModel = None
|
||||
self.statistics = []
|
||||
|
||||
gc.collect()
|
||||
|
||||
if context.accel:
|
||||
torch.cuda.empty_cache()
|
||||
|
||||
@dataclass
|
||||
class Validations:
|
||||
validations: list
|
||||
total: int
|
||||
|
||||
@dataclass
|
||||
class Validation:
|
||||
userPrompt: str
|
||||
userPromptType: str
|
||||
answers: list
|
||||
evalOneOfs: list
|
||||
evalStrings: list
|
||||
Reference in New Issue
Block a user