|
|
import json
|
|
|
import tqdm
|
|
|
import logging
|
|
|
from scripts.get_factual_evaluation import get_factual_evaluation
|
|
|
from scripts.groq_client import GroqClient
|
|
|
from scripts.helper import adaptive_delay, ensure_directory_exists, load_used_data
|
|
|
from scripts.prompt import get_factual_prompt
|
|
|
|
|
|
def evaluate_factual_robustness(config):
|
|
|
"""Evaluates negative rejection for a given model by processing predictions and computing scores."""
|
|
|
config['noise_rate'] = 0.4
|
|
|
model_name = config['model_name']
|
|
|
|
|
|
if model_name in config['models']:
|
|
|
model = GroqClient(plm=config['model_name'])
|
|
|
else:
|
|
|
logging.warning(f"Skipping unknown model: {model_name}")
|
|
|
return
|
|
|
|
|
|
|
|
|
base_path = "results/Counterfactual Robustness"
|
|
|
evalue_file = get_factual_evaluation(config)
|
|
|
print(f"Factual pred file {evalue_file}")
|
|
|
output_file = f"{base_path}/output_{config['output_file_extension']}.json"
|
|
|
result_file = f"{base_path}/scores_{config['output_file_extension']}.json"
|
|
|
ensure_directory_exists(output_file)
|
|
|
|
|
|
def process_query(model, data, used_data, output_file):
|
|
|
"""Processes a single query, generates evaluation, and writes the result."""
|
|
|
if data['id'] in used_data and data['query'] == used_data[data['id']]['query'] and data['ans'] == used_data[data['id']]['ans']:
|
|
|
output_file.write(json.dumps(used_data[data['id']], ensure_ascii=False) + '\n')
|
|
|
return used_data[data['id']]
|
|
|
|
|
|
try:
|
|
|
instruction = get_factual_prompt(data['query'], data['prediction'])
|
|
|
|
|
|
|
|
|
for attempt in range(1, 4):
|
|
|
evaluation = model.generate(instruction)
|
|
|
if evaluation:
|
|
|
break
|
|
|
adaptive_delay(attempt)
|
|
|
|
|
|
data['evaluation'] = evaluation
|
|
|
print(f"Model Response for Factual robustness: {evaluation}")
|
|
|
output_file.write(json.dumps(data, ensure_ascii=False) + '\n')
|
|
|
return data
|
|
|
|
|
|
except Exception as e:
|
|
|
print(f"Error processing query: {e}")
|
|
|
return None
|
|
|
|
|
|
def calculate_scores(results, config):
|
|
|
"""Calculates and returns rejection rates and other metrics."""
|
|
|
rejecttt = 0
|
|
|
tt = 0
|
|
|
correct_tt = 0
|
|
|
for i in results:
|
|
|
if "has identified" in i['evaluation'] or "Yes" in i['evaluation']:
|
|
|
rejecttt += 1
|
|
|
if 0 not in i['label'] and 1 in i['label']:
|
|
|
correct_tt += 1
|
|
|
if 0 not in i['label'] and 1 in i['label']:
|
|
|
tt += 1
|
|
|
|
|
|
scores = {
|
|
|
'reject_rate': rejecttt/len(results),
|
|
|
'all_rate': (tt)/len(results),
|
|
|
'correct_rate': correct_tt/rejecttt if rejecttt > 0 else 0,
|
|
|
'tt':tt,
|
|
|
'rejecttt':rejecttt,
|
|
|
'correct_tt':correct_tt,
|
|
|
'nums': len(results),
|
|
|
'noise_rate': config['noise_rate'],
|
|
|
}
|
|
|
return scores
|
|
|
|
|
|
used_data = []
|
|
|
results = []
|
|
|
if config['UsePreCalculatedValue']:
|
|
|
logging.info(f"Trying to use pre calculated values for Counterfactual report generation")
|
|
|
used_data = load_used_data(output_file)
|
|
|
else:
|
|
|
logging.info(f"Recalculating the metrics...")
|
|
|
|
|
|
with open(output_file, 'w', encoding='utf-8') as f_out, open(evalue_file, 'r', encoding='utf-8') as f_eval:
|
|
|
for line in tqdm.tqdm(f_eval):
|
|
|
data = json.loads(line)
|
|
|
processed_data = process_query(model, data, used_data, f_out)
|
|
|
if processed_data:
|
|
|
results.append(processed_data)
|
|
|
|
|
|
|
|
|
scores = calculate_scores(results, config)
|
|
|
logging.info(f"Counterfactual Robustness Score: {scores}")
|
|
|
|
|
|
with open(result_file, 'w', encoding='utf-8') as f_result:
|
|
|
json.dump(scores, f_result, ensure_ascii=False, indent=4)
|
|
|
|
