in evaluations/genai_evaluation.py [0:0]
def main():
"""
Main function to execute the evaluation process.
"""
args = parse_arguments()
data_file_to_use = args.test_data
# Check if the specified data file exists
if not os.path.exists(data_file_to_use):
logger.error(f"The specified data file '{data_file_to_use}' does not exist.")
sys.exit(1)
print(f"Using data file: {data_file_to_use}")
load_dotenv()
current_time = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
use_rest_api = os.getenv('USE_REST_API', "False").lower() == "true"
if use_rest_api:
orchestrator_endpoint, function_key = get_rest_api_config()
print("Configured to use REST API for processing questions.")
else:
orchestrator_endpoint = None
function_key = None
print("Configured to use local execution for processing questions.")
# Azure configuration (used if needed)
azure_config = {
"aoai_endpoint": os.environ.get('AZURE_OPENAI_ENDPOINT', ''),
"aoai_api_version": os.environ.get('AZURE_OPENAI_API_VERSION', '2024-02-01'),
"aoai_api_key": os.environ.get('AZURE_OPENAI_API_KEY', ''),
"subscription_id": os.environ.get('AZURE_SUBSCRIPTION_ID', ''),
"resource_group": os.environ.get('AZURE_RESOURCE_GROUP', ''),
"project_name": os.environ.get('AZUREAI_PROJECT_NAME', '')
}
print("Azure configuration loaded.")
# Ensure 'evaluations' directory exists
os.makedirs('evaluations', exist_ok=True)
print("Ensured 'evaluations' directory exists.")
output_jsonl_file = f"evaluations/responses_{current_time}.jsonl"
output_excel_file = f"evaluations/responses_{current_time}.xlsx" # Excel output file
conversation_id = ""
last_response_data = None
# Initialize a list to collect all output data for Excel
excel_data = []
# Process each question in the test dataset
with open(data_file_to_use, 'r', encoding='utf-8') as f_in, open(output_jsonl_file, 'w', encoding='utf-8') as f_out:
print("Opened data file and output JSONL file.")
for line_number, line in enumerate(f_in, start=1):
line = line.strip()
if not line:
continue
try:
data = json.loads(line)
question = data.get('question', '')
ground_truth = data.get('ground_truth', '')
print(f"Processing question {line_number}: {question}")
start_time = time.time()
# Process the question
response_data = process_question(
question,
use_rest_api,
orchestrator_endpoint if use_rest_api else None,
function_key if use_rest_api else None,
conversation_id
)
duration = time.time() - start_time
# Prepare the output data
output_data = {
"Question": question,
"Ground ground_truth": ground_truth,
"Answer": response_data.get('answer', 'No answer provided.'),
"Context": response_data.get('data_points', 'No data points provided.'),
"Reasoning": response_data.get('reasoning', 'No reasoning provided.'),
"Processing Time (seconds)": duration
}
f_out.write(json.dumps(output_data) + '\n')
# Append to excel_data list
excel_data.append(output_data)
except Exception as e:
logger.exception(f"Error processing line {line_number}: {e}")
print("Finished processing all questions.")
# Optionally prettify the JSONL file
prettify_jsonl_file(output_jsonl_file)
# Save results to Excel
try:
df = pd.DataFrame(excel_data)
df.to_excel(output_excel_file, index=False)
print(f"Results have been saved to Excel file: '{output_excel_file}'")
except Exception as e:
logger.exception(f"Failed to save results to Excel: {e}")
print(f"Error: Could not save results to Excel file. {e}")