Spaces:
Running
Running
import os | |
import time | |
from datetime import datetime | |
import logging | |
from pathlib import Path | |
import requests | |
import json | |
import numpy as np | |
import pandas as pd | |
import spacy | |
from sentence_transformers import CrossEncoder | |
import litellm | |
# from litellm import completion | |
from tqdm import tqdm | |
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig, AutoConfig, pipeline | |
# from accelerate import PartialState | |
# from accelerate.inference import prepare_pippy | |
import torch | |
import cohere | |
from openai import OpenAI | |
# import google | |
import google.generativeai as genai | |
import src.backend.util as util | |
import src.envs as envs | |
# import pandas as pd | |
import scipy | |
from scipy.spatial.distance import jensenshannon | |
# import numpy as np | |
# litellm.set_verbose=False | |
litellm.set_verbose=True | |
# Set up basic configuration for logging | |
logging.basicConfig(level=logging.INFO, | |
format='%(asctime)s - %(levelname)s - %(message)s') | |
# Load spacy model for word tokenization | |
nlp = spacy.load("en_core_web_sm") | |
nlp1 = spacy.load("en_core_web_trf") | |
# os.environ["HUGGINGFACE_API_KEY"] = envs.TOKEN | |
def load_evaluation_model(model_path): | |
"""Load the evaluation model from the given path | |
Args: | |
model_path (str): Path to the evaluation model | |
Returns: | |
CrossEncoder: The evaluation model | |
""" | |
# model = CrossEncoder(model_path) | |
model = "" | |
return model | |
class ModelLoadingException(Exception): | |
"""Exception raised for errors in loading a model. | |
Attributes: | |
model_id (str): The model identifier. | |
revision (str): The model revision. | |
""" | |
def __init__(self, model_id, revision, messages="Error initializing model"): | |
self.model_id = model_id | |
self.revision = revision | |
super().__init__(f"{messages} id={model_id} revision={revision}") | |
class SummaryGenerator: | |
"""A class to generate summaries using a causal language model. | |
Attributes: | |
model (str): huggingface/{model_id} | |
api_base (str): https://api-inference.huggingface.co/models/{model_id} | |
summaries_df (DataFrame): DataFrame to store generated summaries. | |
revision (str): Model revision. | |
avg_length (float): Average length of summaries. | |
answer_rate (float): Rate of non-empty summaries. | |
""" | |
def __init__(self, model_id, revision): | |
""" | |
Initializes the SummaryGenerator with a model. | |
Args: | |
model_id (str): Identifier for the model. | |
revision (str): Revision of the model. | |
""" | |
self.model_id = model_id | |
self.model = f"huggingface/{model_id}" | |
self.api_base = f"https://api-inference.huggingface.co/models/{model_id}" | |
self.summaries_df = pd.DataFrame() | |
self.revision = revision | |
self.avg_length = None | |
self.answer_rate = None | |
self.exceptions = None | |
self.local_model = None | |
def generate_summaries(self, dataset, df_prompt, save_path=None): | |
"""Generate summaries for a given DataFrame of source docs. | |
修改这里拉取模型生成结果 | |
Args: | |
df (DataFrame): DataFrame containing source docs. | |
Returns: | |
summaries_df (DataFrame): Generated summaries by the model. | |
""" | |
exceptions = [] | |
if (save_path is not None) and os.path.exists(save_path): | |
'''已存在文件,可以读取已经存在的测试文本''' | |
self.summaries_df = pd.read_csv(save_path) | |
# print(self.summaries_df['Experiment']) | |
print(f'Loaded generated summaries from {save_path}') | |
else: | |
'''测试文件不存在,则需要调用指定的模型来进行测试''' | |
# prompt = {} | |
# for index, row in tqdm(df_prompt.iterrows(), total=df_prompt.shape[0]): | |
# prompt['E' + row['Item']] = row['Prompt'] | |
xls = pd.ExcelFile(dataset) | |
sheet_names = xls.sheet_names | |
# sheet_names = df.sheetnames | |
print(f"Total: {len(sheet_names)}") | |
print(sheet_names) | |
Experiment_ID, Questions_ID, Item_ID, Condition, User_prompt, Response, Factor_2, Stimuli_1 = [], [], [], [], [] ,[], [], [] | |
for i, sheet_name in enumerate(sheet_names, start=1): | |
# 读取每个工作表 | |
# if i > 2 and i ==1: | |
# continue | |
print(i, sheet_name) | |
df_sheet = pd.read_excel(xls, sheet_name=sheet_name) | |
# 假设第一列是'Prompt0',但这里我们使用列名来避免硬编码 | |
if 'Prompt0' in df_sheet.columns: | |
prompt_column = df_sheet['Prompt0'] | |
else: | |
# 如果'Prompt0'列不存在,则跳过该工作表或进行其他处理 | |
continue | |
if i == 3 : | |
word1_list = df_sheet['Stimuli-2'] | |
word2_list = df_sheet['Stimuli-3'] | |
V2_column = [] | |
for jj in range(len(word1_list)): | |
V2_column.append(word1_list[jj] + '_' + word2_list[jj]) | |
# print(V2_column) | |
elif i == 9: | |
V2_column = df_sheet['V2'] #SL, LS | |
elif i == 4 or i == 6 : | |
V2_column = df_sheet['Stimuli-2'] #Stimuli-2 | |
else: | |
V2_column = [""] * len(prompt_column) | |
q_column = df_sheet["ID"] | |
Item_column = df_sheet["Item"] | |
Condition_column = df_sheet["Condition"] | |
Stimuli_1_column = df_sheet["Stimuli-1"] | |
if 'Stimuli-2' in df_sheet.columns: | |
Stimuli_2_column = df_sheet["Stimuli-2"] | |
# 遍历Prompt0列的值 | |
for j, prompt_value in enumerate(tqdm(prompt_column, desc=f"Processing {sheet_name}"), start=0): | |
ID = 'E' + str(i) | |
# q_ID = ID + '_' + str(j) | |
# print(ID, q_ID, prompt_value) | |
system_prompt = envs.SYSTEM_PROMPT | |
_user_prompt = prompt_value | |
for ii in range(1): | |
# user_prompt = f"{envs.USER_PROMPT}\nPassage:\n{_source}" | |
while True: | |
try: | |
'''调用''' | |
print('开始调用LLM-API') | |
_response = self.generate_summary(system_prompt, _user_prompt) | |
# print(f"Finish index {index}") | |
break | |
except Exception as e: | |
if 'Rate limit reached' in str(e): | |
wait_time = 3660 | |
current_time = datetime.now().strftime('%H:%M:%S') | |
print(f"Rate limit hit at {current_time}. Waiting for 1 hour before retrying...") | |
time.sleep(wait_time) | |
elif 'is currently loading' in str(e): | |
wait_time = 200 | |
print(f"Model is loading, wait for {wait_time}") | |
time.sleep(wait_time) | |
elif '429 Resource has been exhausted' in str(e): # for gemini models | |
wait_time = 60 | |
print(f"Quota has reached, wait for {wait_time}") | |
time.sleep(wait_time) | |
else: | |
print(f"Error at index {i}: {e}") | |
_response = "" | |
exceptions.append(i) | |
break | |
if i == 5: | |
print(_response) | |
if _response == None: | |
_response1, _response2 = "", "" | |
else: | |
try: | |
import re | |
_response1,_response2 = re.split(r'\n\s*\n', _response.strip()) | |
except: | |
_response1 = _response.split('\n\n') | |
if len(_response) == 2: | |
_response1, _response2 = _response[0], _response[1] | |
else: | |
_response1, _response2 = _response[0], "" | |
Experiment_ID.append(ID) | |
Questions_ID.append(q_column[j]) | |
User_prompt.append(_user_prompt) | |
Response.append(_response2) | |
Factor_2.append(V2_column[j]) | |
Stimuli_1.append(Stimuli_2_column[j]) | |
Item_ID.append(Item_column[j]) | |
Condition.append(Condition_column[j]) | |
# the first sentence in the response is saved as E51 | |
Experiment_ID.append(ID + '1') | |
Questions_ID.append(str(q_column[j]) + '1') | |
User_prompt.append(_user_prompt) | |
Response.append(_response1) | |
Factor_2.append(V2_column[j]) | |
Stimuli_1.append(Stimuli_1_column[j]) | |
Item_ID.append(Item_column[j]) | |
Condition.append(Condition_column[j]) | |
else: | |
Experiment_ID.append(ID) | |
Questions_ID.append(q_column[j]) | |
User_prompt.append(_user_prompt) | |
Response.append(_response) | |
if i == 6: | |
Factor_2.append(Condition_column[j]) | |
Stimuli_1.append(V2_column[j]) | |
else: | |
Factor_2.append(V2_column[j]) | |
Stimuli_1.append(Stimuli_1_column[j]) | |
Item_ID.append(Item_column[j]) | |
Condition.append(Condition_column[j]) | |
print(_response) | |
# exit() | |
# Sleep to prevent hitting rate limits too frequently | |
time.sleep(1) | |
self.summaries_df = pd.DataFrame(list(zip(Experiment_ID, Questions_ID, Item_ID, Condition, User_prompt, Response, Factor_2, Stimuli_1)), | |
columns=["Experiment", "Question_ID", "Item", "Condition", "User_prompt", "Response","Factor 2","Stimuli 1"]) | |
if save_path is not None: | |
print(f'Save summaries to {save_path}') | |
fpath = Path(save_path) | |
fpath.parent.mkdir(parents=True, exist_ok=True) | |
self.summaries_df.to_csv(fpath) | |
self.exceptions = exceptions | |
# self._compute_avg_length() | |
# self._compute_answer_rate() | |
return self.summaries_df | |
def generate_summary(self, system_prompt: str, user_prompt: str): | |
# Using Together AI API | |
using_together_api = False | |
together_ai_api_models = ['mixtral', 'dbrx', 'wizardlm'] | |
for together_ai_api_model in together_ai_api_models: | |
if together_ai_api_model in self.model_id.lower(): | |
using_together_api = True | |
break | |
# print('适用哪一种LLM',together_ai_api_model , using_together_api) | |
# print(self.model_id.lower()) #meta-llama/llama-2-7b-chat-hf | |
# print('local',self.local_model) $None | |
# exit() | |
# if 'mixtral' in self.model_id.lower() or 'dbrx' in self.model_id.lower() or 'wizardlm' in self.model_id.lower(): # For mixtral and dbrx models, use Together AI API | |
if using_together_api: | |
# suffix = "completions" if ('mixtral' in self.model_id.lower() or 'base' in self.model_id.lower()) else "chat/completions" | |
suffix = "chat/completions" | |
url = f"https://api.together.xyz/v1/{suffix}" | |
payload = { | |
"model": self.model_id, | |
# "max_tokens": 4096, | |
'max_new_tokens': 50, | |
# "temperature": 0.0, | |
# 'repetition_penalty': 1.1 if 'mixtral' in self.model_id.lower() else 1 | |
} | |
# if 'mixtral' in self.model_id.lower(): | |
# # payload['prompt'] = user_prompt | |
# # payload['prompt'] = "Write a summary of the following passage:\nPassage:\n" + user_prompt.split('Passage:\n')[-1] + '\n\nSummary:' | |
# payload['prompt'] = 'You must stick to the passage provided. Provide a concise summary of the following passage, covering the core pieces of information described:\nPassage:\n' + user_prompt.split('Passage:\n')[-1] + '\n\nSummary:' | |
# print(payload) | |
# else: | |
# payload['messages'] = [{"role": "system", "content": system_prompt}, | |
# {"role": "user", "content": user_prompt}] | |
payload['messages'] = [{"role": "system", "content": system_prompt}, | |
{"role": "user", "content": user_prompt}] | |
headers = { | |
"accept": "application/json", | |
"content-type": "application/json", | |
"Authorization": f"Bearer {os.environ['TOGETHER_API_KEY']}" | |
} | |
response = requests.post(url, json=payload, headers=headers) | |
try: | |
result = json.loads(response.text) | |
# print(result) | |
result = result["choices"][0] | |
if 'message' in result: | |
result = result["message"]["content"].strip() | |
else: | |
result = result["text"] | |
result_candidates = [result_cancdidate for result_cancdidate in result.split('\n\n') if len(result_cancdidate) > 0] | |
result = result_candidates[0] | |
print(result) | |
except: | |
print(response) | |
result = '' | |
print(result) | |
return result | |
# Using OpenAI API | |
elif 'gpt' in self.model_id.lower(): | |
response = litellm.completion( | |
model=self.model_id.replace('openai/',''), | |
messages=[{"role": "system", "content": system_prompt}, | |
{"role": "user", "content": user_prompt}], | |
# temperature=0.0, | |
max_tokens=50, | |
) | |
result = response['choices'][0]['message']['content'] | |
# print() | |
print(result) | |
return result | |
# Using Google AI API for Gemini models | |
elif 'gemini' in self.model_id.lower(): | |
genai.configure(api_key=os.getenv('GOOGLE_AI_API_KEY')) | |
generation_config = { | |
"temperature": 0, | |
"top_p": 0.95, # cannot change | |
"top_k": 0, | |
"max_output_tokens": 50, | |
# "response_mime_type": "application/json", | |
} | |
safety_settings = [ | |
{ | |
"category": "HARM_CATEGORY_HARASSMENT", | |
"threshold": "BLOCK_NONE" | |
}, | |
{ | |
"category": "HARM_CATEGORY_HATE_SPEECH", | |
"threshold": "BLOCK_NONE" | |
}, | |
{ | |
"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", | |
"threshold": "BLOCK_NONE" | |
}, | |
{ | |
"category": "HARM_CATEGORY_DANGEROUS_CONTENT", | |
"threshold": "BLOCK_NONE" | |
}, | |
] | |
model = genai.GenerativeModel(model_name="gemini-1.5-pro-latest" if "gemini-1.5-pro" in self.model_id.lower() else self.model_id.lower().split('google/')[-1], | |
generation_config=generation_config, | |
system_instruction=system_prompt, | |
safety_settings=safety_settings) | |
convo = model.start_chat(history=[]) | |
convo.send_message(user_prompt) | |
# print(convo.last) | |
result = convo.last.text | |
print(result) | |
return result | |
# Using HF API or download checkpoints | |
elif self.local_model is None: | |
# print(self.model_id) | |
# print(self.api_base) | |
# mistralai/Mistral-7B-Instruct-v0.1 | |
# https://api-inference.huggingface.co/models/mistralai/Mistral-7B-Instruct-v0.1 | |
try: # try use HuggingFace API | |
# response = litellm.completion( | |
# model="huggingface/"+'command-r-plus' if 'command' in self.model_id else self.model_id, | |
# messages=[{"role": "system", "content": system_prompt}, | |
# {"role": "user", "content": user_prompt}], | |
# temperature=0.0, | |
# max_tokens=1024, | |
# api_base= "https://api-inference.huggingface.co/models/" + self.model_id, | |
# ) | |
# self.model_id = 'command-r-plus' if 'command' in self.model_id else self.model_id | |
# response = litellm.completion( | |
# model="huggingface/" + self.model_id, | |
# # mistralai/Mistral-7B-Instruct-v0.1", | |
# messages=[{"role": "system", "content": system_prompt}, | |
# {"role": "user", "content": user_prompt}], | |
# #temperature=0.0, | |
# max_tokens=1024, | |
# api_base="https://api-inference.huggingface.co/models/" + self.model_id) | |
# print("模型返回结果",response) | |
# print("模型返回结果结束") | |
# # exit() | |
# result = response['choices'][0]['message']['content'] | |
# print(result) | |
from huggingface_hub import InferenceClient | |
client = InferenceClient(self.model_id,api_key=envs.TOKEN) | |
messages = [{"role": "system", "content": system_prompt},{"role": "user", "content": user_prompt}] | |
outputs = client.chat_completion(messages, max_tokens=50) | |
result = outputs['choices'][0]['message']['content'] | |
return result | |
# exit() | |
except: # fail to call api. run it locally. | |
self.tokenizer = AutoTokenizer.from_pretrained(self.model_id, trust_remote_code=True) | |
print("Tokenizer loaded") | |
self.local_model = AutoModelForCausalLM.from_pretrained(self.model_id, trust_remote_code=True, device_map="auto", torch_dtype="auto", cache_dir='/home/paperspace/cache') | |
print("Local model loaded") | |
# exit() | |
# Using local model | |
if self.local_model: # cannot call API. using local model | |
messages=[ | |
{"role": "system", "content": system_prompt}, # gemma-1.1 does not accept system role | |
{"role": "user", "content": user_prompt} | |
] | |
try: # some models support pipeline | |
pipe = pipeline( | |
"text-generation", | |
model=self.local_model, | |
tokenizer=self.tokenizer, | |
) | |
generation_args = { | |
"max_new_tokens": 50, | |
"return_full_text": False, | |
#"temperature": 0.0, | |
"do_sample": False, | |
} | |
output = pipe(messages, **generation_args) | |
result = output[0]['generated_text'] | |
print(result) | |
except: | |
prompt = self.tokenizer.apply_chat_template(messages,add_generation_prompt=True, tokenize=False) | |
print(prompt) | |
input_ids = self.tokenizer(prompt, return_tensors="pt").to('cuda') | |
with torch.no_grad(): | |
outputs = self.local_model.generate(**input_ids, max_new_tokens=50, do_sample=True, pad_token_id=self.tokenizer.eos_token_id) | |
result = self.tokenizer.decode(outputs[0], skip_special_tokens=True) | |
result = result.replace(prompt[0], '') | |
print(result) | |
return result | |
def _compute_avg_length(self): | |
""" | |
Compute the average length of non-empty summaries using SpaCy. | |
""" | |
total_word_count = 0 | |
total_count = 0 | |
for summary in self.summaries_df['summary']: | |
if util.is_summary_valid(summary): | |
doc = nlp(summary) | |
words = [token.text for token in doc if token.is_alpha] | |
total_word_count += len(words) | |
total_count += 1 | |
self.avg_length = 0 if total_count == 0 else total_word_count / total_count | |
def _compute_answer_rate(self): | |
""" | |
Compute the rate of non-empty summaries. | |
""" | |
valid_count = sum(1 for summary in self.summaries_df['summary'] | |
if util.is_summary_valid(summary)) | |
total_count = len(self.summaries_df) | |
self.answer_rate = 0 if total_count == 0 else valid_count / total_count | |
class EvaluationModel: | |
"""A class to evaluate generated summaries. | |
Attributes: | |
model (CrossEncoder): The evaluation model. | |
scores (list): List of evaluation scores. | |
accuracy (float): Accuracy of the summaries. | |
hallucination_rate (float): Rate of hallucination in summaries. | |
""" | |
def __init__(self, model_path): | |
""" | |
Initializes the EvaluationModel with a CrossEncoder model. | |
Args: | |
model_path (str): Path to the CrossEncoder model. | |
""" | |
self.model = load_evaluation_model(model_path) | |
self.scores = [] | |
self.factual_consistency_rate = None | |
self.hallucination_rate = None | |
self.humanlike_score = None | |
def code_results(self, summaries_df): | |
'''code results from LLM's response''' | |
output = [] | |
'''database for Exp4''' | |
item4 = pd.read_csv(envs.ITEM_4_DATA) | |
wordpair2code = {} | |
for j in range(len(item4['Coding'])): | |
wordpair2code[item4['Pair'][j]] = item4['Coding'][j] | |
'''verb for Exp5''' | |
item5 = pd.read_csv(envs.ITEM_5_DATA) | |
# item corresponding to verb, same item id corresponding to verb pair | |
item2verb2 = {} | |
item2verb1 = {} | |
Stimuli1, Stimuli2 = {}, {} | |
for j in range(len(item5['Item'])): | |
item2verb1[item5['Item'][j]] = item5['Verb1'][j] | |
item2verb2[item5['Item'][j]] = item5['Verb2'][j] | |
Stimuli1[item5['ID'][j]] = item5['Stimuli-1'][j] | |
Stimuli2[item5['ID'][j]] = item5['Stimuli-2'][j] | |
male_keyword = ["he", "his", "himself"] | |
female_keyword = ["she", "her", "herself"] | |
print(len(summaries_df["Experiment"])) | |
for i in range(len(summaries_df["Experiment"])): | |
# vote_1_1, vote_1_2, vote_1_3 = 0, 0, 0 | |
# print() | |
if pd.isna(summaries_df["Response"][i]): | |
output.append("Other") | |
continue | |
rs = summaries_df["Response"][i].strip().lower() | |
sentences = rs.split('\n') | |
sentences = [sentence.split(':', 1)[-1].strip() if ':' in sentence else sentence | |
for sentence in sentences] | |
rs = [sentence.strip() for sentence in sentences if sentence.strip()] | |
'''Exp1''' | |
if summaries_df["Experiment"][i] == "E1": | |
print("E1", rs) | |
rs = rs.replace('"','') | |
if rs == "round": | |
# vote_1_1 += 1 | |
output.append("Round") | |
elif rs == "spiky": | |
output.append("Spiky") | |
else: | |
output.append("Other") | |
'''Exp2''' | |
elif summaries_df["Experiment"][i] == "E2": | |
# rs = summaries_df["Response"][i].strip() | |
rs = rs.split(' ') | |
print("E2", rs) | |
male, female = 0, 0 | |
for word in rs: | |
if word in female_keyword and male == 0: | |
female = 1 | |
output.append("Female") | |
break | |
if word in male_keyword and female == 0: | |
male = 1 | |
output.append("Male") | |
break | |
if male == 0 and female == 0 : | |
output.append("Other") | |
'''Exp3''' | |
elif summaries_df["Experiment"][i] == "E3": | |
# rs = summaries_df["Response"][i].strip() | |
print("E3", rs) | |
if pd.isna(summaries_df["Factor 2"][i]): | |
output.append("Other") | |
else: | |
if summaries_df["Factor 2"][i].strip() == "LS": | |
if "2" in rs: | |
output.append("Long") | |
elif "3" in rs: | |
output.append("Short") | |
else: | |
output.append("Other") | |
if summaries_df["Factor 2"][i].strip() == "SL": | |
if "2" in rs: | |
output.append("Short") | |
elif "3" in rs: | |
output.append("Long") | |
else: | |
output.append("Other") | |
'''Exp4''' | |
elif summaries_df["Experiment"][i] == "E4": | |
# rs = summaries_df["Response"][i].strip() | |
target = summaries_df["Factor 2"][i].strip().lower() | |
pair = target + "_" + rs | |
print("E4:", pair) | |
if pair in wordpair2code.keys(): | |
output.append(wordpair2code[pair]) | |
else: | |
output.append("Other") | |
'''Exp5''' | |
elif summaries_df["Experiment"][i] == "E5" or summaries_df["Experiment"][i] == "E51": | |
# sentence = summaries_df["Response"][i].strip() | |
item_id = summaries_df["Item"][i] | |
question_id = summaries_df["Question_ID"][i] | |
sti1, sti2 = "", "" | |
if summaries_df["Experiment"][i] == "E51": | |
sti1 = Stimuli1[question_id[0:-1]].lower().replace("...", "") | |
sti2 = Stimuli2[question_id[0:-1]].lower().replace("...", "") | |
verb = item2verb1[item_id].lower() | |
sentence = sti1 + " " + rs.replace(sti1, "") | |
print("E5", verb, sentence) | |
if summaries_df["Experiment"][i] == "E5": | |
sti1 = Stimuli1[question_id].lower().replace("...", "") | |
# print(sti1) | |
sti2 = Stimuli2[question_id].lower().replace("...", "") | |
verb = item2verb2[item_id].lower() | |
sentence = sti2.replace("...","") + " " + rs.replace(sti2, "") | |
print("E5", verb, sentence) | |
doc = nlp1(sentence.replace(" "," ")) | |
# print(doc) | |
# print() | |
verb_token = None | |
for token in doc: | |
# print(token.lemma_) | |
if token.lemma_ == verb: | |
verb_token = token | |
break | |
# exit() | |
if verb_token is None: | |
output.append("Other") | |
print("E5 The target verb is missing from the sentence.") | |
else: | |
pobj, dative = None, None | |
# print(verb_token.children) | |
# exit() | |
for child in verb_token.children: | |
print(child) | |
if (child.dep_ == 'dative' and child.pos_ == "ADP") or (child.text == "to" and child.dep_ == 'prep' and child.pos_ == "ADP"): | |
pobj = child.text | |
if child.dep_ == 'dative': | |
dative = child.text | |
print("E5", pobj, dative) | |
# exit() | |
if pobj: | |
output.append("PO") | |
elif dative: | |
output.append("DO") | |
else: | |
print("Other", sentence, pobj, dative) | |
# exit() | |
output.append("Other") | |
'''Exp6''' | |
elif summaries_df["Experiment"][i] == "E6": | |
sentence = summaries_df["Stimuli 1"][i].strip().lower() | |
print("E6", sentence) | |
doc = nlp1(sentence) | |
subject = "None" | |
obj = "None" | |
# 遍历依存关系,寻找主语和宾语 | |
for token in doc: | |
if token.dep_ == "nsubj": | |
subject = token.text | |
elif token.dep_ == "dobj": | |
obj = token.text | |
print("E6", subject, obj) | |
if subject in rs and obj in rs: | |
print(rs, subject, obj, "Other") | |
output.append("Other") | |
elif subject in rs: | |
print(rs, subject, obj, "VP") | |
output.append("VP") | |
elif obj in rs: | |
print(rs, subject, obj, "NP") | |
output.append("NP") | |
else: | |
print(rs, subject, obj, "Other") | |
output.append("Other") | |
'''Exp7''' | |
elif summaries_df["Experiment"][i] == "E7": | |
# rs = summaries_df["Response"][i].strip().lower() | |
print("E7",rs) | |
if rs == "no": | |
output.append("0") | |
elif rs == "yes": | |
output.append("1") | |
else: | |
output.append("Other") | |
'''Exp8''' | |
elif summaries_df["Experiment"][i] == "E8": | |
# rs = summaries_df["Response"][i].strip() | |
if "something is wrong with the question" in rs: | |
output.append("1") | |
else: | |
output.append("0") | |
'''Exp9''' | |
elif summaries_df["Experiment"][i] == "E9": | |
male, female = 0, 0 | |
# rs = summaries_df["Response"][i].strip() | |
if "because" in rs: | |
rs = rs.replace("because because","because").split("because")[1] | |
else: | |
rs = rs | |
condition = summaries_df["Factor 2"][i].strip() | |
rs = rs.split(" ") | |
for w in rs: | |
if w in male_keyword and female != 1: | |
male = 1 | |
break | |
if w in female_keyword and male != 1: | |
female = 1 | |
break | |
print("E9", "condition", condition, "male", male, "female", female) | |
if male == 0 and female == 0: | |
output.append('Other') | |
else: | |
if male == 1 and female==0: | |
if condition == "MF": | |
output.append("Subject") | |
elif condition == "FM": | |
output.append("Object") | |
else: | |
output.append("Other") | |
elif female == 1 and male ==0: | |
if condition == "MF": | |
output.append("Object") | |
elif condition == "FM": | |
output.append("Subject") | |
else: | |
output.append("Other") | |
'''Exp10''' | |
elif summaries_df["Experiment"][i] == "E10": | |
# rs = summaries_df["Response"][i].strip() | |
if rs == "yes": | |
output.append("1") | |
else: | |
output.append("0") | |
else: | |
print("can;t find the Exp:", summaries_df["Experiment"][i]) | |
output.append("NA") | |
# print(output) | |
# exit() | |
'''human''' | |
self.data = pd.DataFrame(list(zip(summaries_df["Experiment"], summaries_df["Question_ID"], summaries_df["Item"], summaries_df["Response"], summaries_df["Factor 2"], summaries_df["Stimuli 1"], summaries_df["Coding"], output)), | |
columns=["Experiment", "Question_ID", "Item", "Response", "Factor 2", "Simulate 1","Original_Coding","Coding"]) | |
# '''LLM''' | |
# self.data = pd.DataFrame(list(zip(summaries_df["Experiment"], summaries_df["Question_ID"], summaries_df["Item"], summaries_df["Response"], summaries_df["Factor 2"], summaries_df["Stimuli 1"], output)), | |
# columns=["Experiment", "Question_ID", "Item", "Response", "Factor 2", "Simulate 1","Coding"]) | |
print(self.data.head()) | |
return self.data | |
def code_results_llm(self, summaries_df): | |
'''code results from LLM's response''' | |
output = [] | |
'''database for Exp4''' | |
item4 = pd.read_csv(envs.ITEM_4_DATA) | |
wordpair2code = {} | |
for j in range(len(item4['Coding'])): | |
wordpair2code[item4['Pair'][j]] = item4['Coding'][j] | |
'''verb for Exp5''' | |
item5 = pd.read_csv(envs.ITEM_5_DATA) | |
# item corresponding to verb, same item id corresponding to verb pair | |
item2verb2 = {} | |
item2verb1 = {} | |
Stimuli1, Stimuli2 = {}, {} | |
for j in range(len(item5['Item'])): | |
item2verb1[item5['Item'][j]] = item5['Verb1'][j] | |
item2verb2[item5['Item'][j]] = item5['Verb2'][j] | |
Stimuli1[item5['ID'][j]] = item5['Stimuli-1'][j] | |
Stimuli2[item5['ID'][j]] = item5['Stimuli-2'][j] | |
male_keyword = ["he", "his", "himself"] | |
female_keyword = ["she", "her", "herself"] | |
print(len(summaries_df["Experiment"])) | |
for i in range(len(summaries_df["Experiment"])): | |
# vote_1_1, vote_1_2, vote_1_3 = 0, 0, 0 | |
# print() | |
if pd.isna(summaries_df["Response"][i]): | |
output.append("Other") | |
continue | |
rs = summaries_df["Response"][i].strip().lower() | |
'''Exp1''' | |
if summaries_df["Experiment"][i] == "E1": | |
print("E1", rs) | |
rs = rs.replace('"','') | |
if rs == "round": | |
# vote_1_1 += 1 | |
output.append("Round") | |
elif rs == "spiky": | |
output.append("Spiky") | |
else: | |
output.append("Other") | |
'''Exp2''' | |
elif summaries_df["Experiment"][i] == "E2": | |
# rs = summaries_df["Response"][i].strip() | |
rs = rs.split(' ') | |
print("E2", rs) | |
male, female = 0, 0 | |
for word in rs: | |
if word in female_keyword and male == 0: | |
female = 1 | |
output.append("Female") | |
break | |
if word in male_keyword and female == 0: | |
male = 1 | |
output.append("Male") | |
break | |
if male == 0 and female == 0 : | |
output.append("Other") | |
'''Exp3''' | |
elif summaries_df["Experiment"][i] == "E3": | |
# rs = summaries_df["Response"][i].strip() | |
print("E3", rs) | |
rs = rs.replace('"', '') | |
pair = summaries_df["Factor 2"][i] | |
word1, word2 = pair.split('_') | |
if rs == word1: | |
if len(word1) > len(word2): | |
output.append("Long") | |
else: | |
output.append("Short") | |
elif rs == word2: | |
if len(word1) > len(word2): | |
output.append("Short") | |
else: | |
output.append("Long") | |
else: | |
output.append("Other") | |
'''Exp4''' | |
elif summaries_df["Experiment"][i] == "E4": | |
try: | |
meaning_word = rs.split(";")[4].replace(" ", '') | |
except IndexError: | |
output.append("Other") | |
continue | |
except Exception as e: | |
print(f"Unexpected error: {e}") | |
output.append("Other") | |
continue | |
target = summaries_df["Factor 2"][i].strip().lower() | |
pair = target + "_" + meaning_word | |
print("E4:", pair) | |
if pair in wordpair2code.keys(): | |
output.append(wordpair2code[pair]) | |
else: | |
output.append("Other") | |
'''Exp5''' | |
elif summaries_df["Experiment"][i] == "E5" or summaries_df["Experiment"][i] == "E51": | |
# sentence = summaries_df["Response"][i].strip() | |
item_id = summaries_df["Item"][i] | |
question_id = summaries_df["Question_ID"][i] | |
sti1, sti2 = "", "" | |
if summaries_df["Experiment"][i] == "E51": | |
sti1 = Stimuli1[question_id[0:-1]].lower().replace("...", "") | |
sti2 = Stimuli2[question_id[0:-1]].lower().replace("...", "") | |
verb = item2verb1[item_id].lower() | |
sentence = sti1 + " " + rs.replace(sti1, "") | |
print("E5", verb, sentence) | |
if summaries_df["Experiment"][i] == "E5": | |
sti1 = Stimuli1[question_id].lower().replace("...", "") | |
# print(sti1) | |
sti2 = Stimuli2[question_id].lower().replace("...", "") | |
verb = item2verb2[item_id].lower() | |
sentence = sti2.replace("...","") + " " + rs.replace(sti2, "") | |
print("E5", verb, sentence) | |
doc = nlp1(sentence.replace(" "," ")) | |
# print(doc) | |
# print() | |
verb_token = None | |
for token in doc: | |
# print(token.lemma_) | |
if token.lemma_ == verb: | |
verb_token = token | |
break | |
# exit() | |
if verb_token is None: | |
output.append("Other") | |
print("E5 The target verb is missing from the sentence.") | |
else: | |
pobj, dative = None, None | |
# print(verb_token.children) | |
# exit() | |
for child in verb_token.children: | |
print(child) | |
if (child.dep_ == 'dative' and child.pos_ == "ADP") or (child.text == "to" and child.dep_ == 'prep' and child.pos_ == "ADP"): | |
pobj = child.text | |
if child.dep_ == 'dative': | |
dative = child.text | |
print("E5", pobj, dative) | |
# exit() | |
if pobj: | |
output.append("PO") | |
elif dative: | |
output.append("DO") | |
else: | |
print("Other", sentence, pobj, dative) | |
# exit() | |
output.append("Other") | |
'''Exp6''' | |
elif summaries_df["Experiment"][i] == "E6": | |
sentence = summaries_df["Stimuli 1"][i].strip().lower() | |
print("E6", sentence) | |
doc = nlp1(sentence) | |
subject = "None" | |
obj = "None" | |
# 遍历依存关系,寻找主语和宾语 | |
for token in doc: | |
if token.dep_ == "nsubj": | |
subject = token.text | |
elif token.dep_ == "dobj": | |
obj = token.text | |
print("E6", subject, obj) | |
if subject in rs and obj in rs: | |
print(rs, subject, obj, "Other") | |
output.append("Other") | |
elif subject in rs: | |
print(rs, subject, obj, "VP") | |
output.append("VP") | |
elif obj in rs: | |
print(rs, subject, obj, "NP") | |
output.append("NP") | |
else: | |
print(rs, subject, obj, "Other") | |
output.append("Other") | |
'''Exp7''' | |
elif summaries_df["Experiment"][i] == "E7": | |
# rs = summaries_df["Response"][i].strip().lower() | |
rs = rs.replace(".", "").replace(",", "") | |
print("E7",rs) | |
if rs == "no": | |
output.append("0") | |
elif rs == "yes": | |
output.append("1") | |
else: | |
output.append("Other") | |
'''Exp8''' | |
elif summaries_df["Experiment"][i] == "E8": | |
# rs = summaries_df["Response"][i].strip() | |
print("E8",rs) | |
if "something is wrong with the question" in rs: | |
output.append("1") | |
else: | |
output.append("0") | |
'''Exp9''' | |
elif summaries_df["Experiment"][i] == "E9": | |
male, female = 0, 0 | |
# rs = summaries_df["Response"][i].strip() | |
if "because" in rs: | |
rs = rs.replace("because because","because").split("because")[1] | |
else: | |
rs = rs | |
condition = summaries_df["Factor 2"][i].strip() | |
rs = rs.split(" ") | |
for w in rs: | |
if w in male_keyword and female != 1: | |
male = 1 | |
break | |
if w in female_keyword and male != 1: | |
female = 1 | |
break | |
print("E9", "condition", condition, "male", male, "female", female) | |
if male == 0 and female == 0: | |
output.append('Other') | |
else: | |
if male == 1 and female==0: | |
if condition == "MF": | |
output.append("Subject") | |
elif condition == "FM": | |
output.append("Object") | |
else: | |
output.append("Other") | |
elif female == 1 and male ==0: | |
if condition == "MF": | |
output.append("Object") | |
elif condition == "FM": | |
output.append("Subject") | |
else: | |
output.append("Other") | |
'''Exp10''' | |
elif summaries_df["Experiment"][i] == "E10": | |
# rs = summaries_df["Response"][i].strip() | |
rs = rs.replace(".", "") | |
if rs == "yes": | |
output.append("1") | |
else: | |
output.append("0") | |
else: | |
print("can;t find the Exp:", summaries_df["Experiment"][i]) | |
output.append("NA") | |
# print(output) | |
# exit() | |
'''human''' | |
# self.data = pd.DataFrame(list(zip(summaries_df["Experiment"], summaries_df["Question_ID"], summaries_df["Item"], summaries_df["Response"], summaries_df["Factor 2"], summaries_df["Stimuli 1"], summaries_df["Coding"], output)), | |
# columns=["Experiment", "Question_ID", "Item", "Response", "Factor 2", "Simulate 1","Original_Coding","Coding"]) | |
'''LLM''' | |
self.data = pd.DataFrame(list(zip(summaries_df["Experiment"], summaries_df["Question_ID"], summaries_df["Item"], summaries_df["Response"], summaries_df["Factor 2"], summaries_df["Stimuli 1"], output)), | |
columns=["Experiment", "Question_ID", "Item", "Response", "Factor 2", "Simulate 1","Coding"]) | |
print(self.data.head()) | |
return self.data | |
def calculate_js_divergence(self, file_path_1, file_path_2): | |
""" | |
Calculate the Jensen-Shannon divergence for response distributions between two datasets. | |
- Extracts E5 and E51 pairs, creates new data based on comparison, | |
removes the original E5 and E51, and then calculates the JS divergence between the datasets. | |
Parameters: | |
file_path_1 (str): Path to the first dataset file (Excel format). | |
file_path_2 (str): Path to the second dataset file (CSV format). | |
Returns: | |
float: The average JS divergence across all common Question_IDs. | |
""" | |
# Load the datasets | |
human_df = pd.read_csv(file_path_1, encoding='ISO-8859-1') | |
llm_df = pd.read_csv(file_path_2) | |
def create_e5_entries(df): | |
new_entries = [] | |
for i in range(len(df) - 1): | |
if 'E51' in df.iloc[i]['Experiment']: | |
priming_id = df.iloc[i][0]-1 | |
priming_row_id = df[df.iloc[:, 0] == priming_id].index[0] | |
new_question_id = df.iloc[priming_row_id]['Question_ID'] | |
label = 1 if df.iloc[i]['Coding'] == df.iloc[priming_row_id]['Coding'] else 0 | |
new_entries.append({ | |
'Question_ID': new_question_id, | |
'Response': f'{df.iloc[i]["Coding"]}-{df.iloc[priming_row_id]["Coding"]}', | |
'Coding': label | |
}) | |
return pd.DataFrame(new_entries) | |
# Create new E5 entries for both datasets | |
human_e5 = create_e5_entries(human_df) | |
llm_e5 = create_e5_entries(llm_df) | |
# Remove E5 and E51 entries from both datasets | |
human_df = human_df[~human_df['Question_ID'].str.contains('E5')] | |
llm_df = llm_df[~llm_df['Question_ID'].str.contains('E5')] | |
# Append new E5 entries to the cleaned dataframes | |
human_df = pd.concat([human_df, human_e5], ignore_index=True) | |
llm_df = pd.concat([llm_df, llm_e5], ignore_index=True) | |
### Calculate Average JS Divergence ### | |
# Extract the relevant columns for JS divergence calculation | |
human_responses = human_df[['Question_ID', 'Coding']] | |
llm_responses = llm_df[['Question_ID', 'Coding']] | |
# Get unique Question_IDs present in both datasets | |
common_question_ids = set(human_responses['Question_ID']).intersection(set(llm_responses['Question_ID'])) | |
# Initialize a list to store JS divergence for each Question_ID | |
js_divergence_list = [] | |
js_divergence ={} | |
# Calculate JS divergence for each common Question_ID | |
for q_id in common_question_ids: | |
# Get response distributions for the current Question_ID in both datasets | |
human_dist = human_responses[human_responses['Question_ID'] == q_id]['Coding'].value_counts(normalize=True) | |
llm_dist = llm_responses[llm_responses['Question_ID'] == q_id]['Coding'].value_counts(normalize=True) | |
# Reindex the distributions to have the same index, filling missing values with 0 | |
all_responses = set(human_dist.index).union(set(llm_dist.index)) | |
human_dist = human_dist.reindex(all_responses, fill_value=0) | |
llm_dist = llm_dist.reindex(all_responses, fill_value=0) | |
# Calculate JS divergence and add to the list | |
js_div = jensenshannon(human_dist, llm_dist, base=2) | |
experiment_id = q_id.split('_')[1] | |
if experiment_id not in js_divergence: | |
js_divergence[experiment_id] = [] | |
js_divergence[experiment_id].append(js_div) | |
js_divergence_list.append(js_div) | |
#js_divergence[q_id] = js_div | |
# Calculate the average JS divergence | |
# JS per experiment | |
avg_js_divergence_per_experiment = {exp: 1- np.nanmean(divs) for exp, divs in js_divergence.items()} | |
print(avg_js_divergence_per_experiment) | |
# JS overall | |
avg_js_divergence = 1 - np.nanmean(js_divergence_list) | |
print("avg_js_divergence:", avg_js_divergence) | |
return avg_js_divergence | |
def evaluate_humanlike(self, summaries_df: object, human_data_path: object, result_save_path: object) -> object: | |
''' | |
evaluate humanlike score | |
1. code the result | |
2. comput the similaritirs between human and model | |
process model responses''' | |
'''coding human data''' | |
# self.huamn_df = pd.read_csv(human_data_path) | |
# self.data = self.code_results(self.huamn_df) | |
#save_path = human_data_path.replace('.csv','_coding.csv') | |
#human_save_path = "./src/datasets/coding_human.xlsx" | |
# if save_path is not None: | |
# print(f'Save human coding results to {save_path}') | |
# fpath = Path(save_path) | |
# fpath.parent.mkdir(parents=True, exist_ok=True) | |
# self.data.to_csv(fpath) | |
'''coding llm data''' | |
save_path = result_save_path.replace('.csv','_coding.csv') | |
self.llm_df = self.code_results_llm(summaries_df) | |
if save_path is not None: | |
print(f'Save LLM coding results to {save_path}') | |
fpath = Path(save_path) | |
fpath.parent.mkdir(parents=True, exist_ok=True) | |
self.llm_df.to_csv(fpath) | |
# file_path_1 = '/Users/simon/Downloads/coding_human.xlsx' | |
# file_path_2 = '/Users/simon/Downloads/Meta-Llama-3.1-70B-Instruct_coding.csv' | |
avg_js_divergence = self.calculate_js_divergence(human_data_path, save_path) | |
return avg_js_divergence | |
def evaluate_hallucination(self, summaries_df): | |
""" | |
Evaluate the hallucination rate in summaries. Updates the 'scores' attribute | |
of the instance with the computed scores. | |
Args: | |
summaries_df (DataFrame): DataFrame containing source docs and summaries. | |
Returns: | |
list: List of hallucination scores. Also updates the 'scores' attribute of the instance. | |
""" | |
hem_scores = [] | |
sources = [] | |
summaries = [] | |
source_summary_pairs = util.create_pairs(summaries_df) | |
'''评价模型结果''' | |
for doc, summary in tqdm(source_summary_pairs, desc="Evaluating Humanlikeness"): | |
if util.is_summary_valid(summary): | |
try: | |
summary = summary.replace('<bos>','').replace('<eos>','') | |
score = self.model.predict([doc, summary])# [0] | |
if not isinstance(score, float): | |
try: | |
score = score.item() | |
except: | |
logging.warning(f"Score type mismatch: Expected float, got {type(score)}.") | |
continue | |
hem_scores.append(score) | |
sources.append(doc) | |
summaries.append(summary) | |
except Exception as e: | |
logging.error(f"Error while running HEM: {e}") | |
raise | |
self.scores = hem_scores | |
eval_results = {'source': sources, 'summary': summaries, 'HEM scores': hem_scores} | |
return hem_scores, eval_results | |
# for doc, summary in tqdm(source_summary_pairs, desc="Evaluating hallucinations"): | |
# if util.is_summary_valid(summary): | |
# try: | |
# # summary_pieces = summary.split('\n') | |
# # summary = summary_pieces[0] if len(summary_pieces[0].strip()) > 0 else summary_pieces[1] | |
# summary = summary.replace('<bos>','').replace('<eos>','') | |
# # print([doc, summary]) | |
# # print(self.model.predict([doc, summary])) | |
# score = self.model.predict([doc, summary])# [0] | |
# if not isinstance(score, float): | |
# try: | |
# score = score.item() | |
# except: | |
# logging.warning(f"Score type mismatch: Expected float, got {type(score)}.") | |
# continue | |
# hem_scores.append(score) | |
# sources.append(doc) | |
# summaries.append(summary) | |
# except Exception as e: | |
# logging.error(f"Error while running HEM: {e}") | |
# raise | |
# self.scores = hem_scores | |
# eval_results = {'source': sources, 'summary': summaries, 'HEM scores': hem_scores} | |
# return hem_scores, eval_results | |
def compute_factual_consistency_rate(self, threshold=0.5): | |
""" | |
Compute the factual consistency rate of the evaluated summaries based on | |
the previously calculated scores. This method relies on the 'scores' | |
attribute being populated, typically via the 'evaluate_hallucination' method. | |
Returns: | |
float: Factual Consistency Rate. Also updates the 'factual_consistency_rate' | |
and 'hallucination_rate' attributes of the instance. | |
Raises: | |
ValueError: If scores have not been calculated prior to calling this method. | |
""" | |
if not self.scores: | |
error_msg = "Scores not calculated. Call evaluate_hallucination() first." | |
logging.error(error_msg) | |
raise ValueError(error_msg) | |
# Use threshold of 0.5 to compute factual_consistency_rate | |
num_above_threshold = sum(score >= threshold for score in self.scores) | |
num_total = len(self.scores) | |
if not num_total: | |
raise ValueError("No scores available to compute factual consistency rate.") | |
self.factual_consistency_rate = (num_above_threshold / num_total) * 100 | |
self.hallucination_rate = 100 - self.factual_consistency_rate | |
return self.factual_consistency_rate | |