#!/usr/bin/env python3 """ evaluation.py A script to evaluate questions either by sending them to a REST API or processing them locally using the Orchestrator class. Additionally, results are saved to an Excel spreadsheet. Usage: bash: export PYTHONPATH=./:$PYTHONPATH python evaluation.py --test-data path/to/test_data.jsonl Powershell: $env:PYTHONPATH = "./;$env:PYTHONPATH" python evaluation.py --test-data path/to/test_data.jsonl Environment Variables: - USE_REST_API: Set to "True" to use the REST API for processing questions. Otherwise, local execution is used. - ORCHESTRATOR_ENDPOINT: The API endpoint URI (required if USE_REST_API is "True"). - FUNCTION_KEY: The API access key (required if USE_REST_API is "True"). - AZURE_OPENAI_ENDPOINT: Azure OpenAI endpoint. - AZURE_OPENAI_API_VERSION: Azure OpenAI API version. - AZURE_OPENAI_API_KEY: Azure OpenAI API key. - AZURE_SUBSCRIPTION_ID: Azure subscription ID. - AZURE_RESOURCE_GROUP: Azure resource group. - AZUREAI_PROJECT_NAME: Azure AI project name. Requirements: - Python 3.x - requests library (`pip install requests`) - python-dotenv library (`pip install python-dotenv`) - promptflow library (`pip install promptflow`) - pandas library (`pip install pandas`) - openpyxl library (`pip install openpyxl`) Security Note: Ensure that your `.env` file is not committed to version control systems as it contains sensitive information. Add `.env` to your `.gitignore` file. """ import os import sys import json import requests import datetime import logging import logging.config from dotenv import load_dotenv import asyncio import argparse import pandas as pd # Import pandas import time # Import Orchestrator for local execution try: from orchestration import Orchestrator except ImportError: print("Error: Could not import Orchestrator from 'orchestration' module.") sys.exit(1) # Configure logging LOGGING_CONFIG = { 'version': 1, 'disable_existing_loggers': False, # Allow existing loggers to propagate 'formatters': { 'standard': { 'format': '%(asctime)s - %(levelname)s - %(name)s - %(message)s', }, }, 'handlers': { 'file_handler': { 'class': 'logging.FileHandler', 'filename': 'evaluation.log', 'mode': 'a', 'formatter': 'standard', 'level': 'INFO', }, 'console_handler': { 'class': 'logging.StreamHandler', 'stream': sys.stdout, 'formatter': 'standard', 'level': 'ERROR', }, }, 'root': { 'handlers': ['file_handler', 'console_handler'], 'level': 'DEBUG', }, } logging.config.dictConfig(LOGGING_CONFIG) logger = logging.getLogger(__name__) # Use a module-specific logger def get_rest_api_config(): """ Load environment variables required for REST API configuration. Returns: tuple: (orchestrator_endpoint, function_key) """ orchestrator_endpoint = os.getenv('ORCHESTRATOR_ENDPOINT') function_key = os.getenv('FUNCTION_KEY') if not orchestrator_endpoint: logger.error("ORCHESTRATOR_ENDPOINT not found in environment variables.") sys.exit(1) if not function_key: logger.error("FUNCTION_KEY not found in environment variables.") sys.exit(1) return orchestrator_endpoint, function_key def send_question_to_rest_api(uri, x_functions_key, question, conversation_id): """ Send the question to the orchestrator API and return the response. Args: uri (str): The API endpoint URI. x_functions_key (str): The API access key. question (str): The question to send. conversation_id (str): The conversation ID. Returns: dict: The API response parsed as a JSON object. """ headers = { 'x-functions-key': x_functions_key, 'Content-Type': 'application/json' } body = { 'conversation_id': conversation_id, 'question': question } try: response = requests.post(uri, headers=headers, json=body) response.raise_for_status() # Raises HTTPError for bad responses try: response_data = response.json() if not isinstance(response_data, dict): logger.error("Response JSON is not a dictionary.") return {"error": "Invalid response format from orchestrator API."} return response_data except json.JSONDecodeError: logger.error("Response is not valid JSON.") return {"error": "Response is not valid JSON."} except requests.exceptions.RequestException as e: logger.exception(f"HTTP Request failed: {e}") return {"error": f"HTTP Request failed: {e}"} def send_question_to_python(question, conversation_id): """ Process the question using the Orchestrator locally. Args: question (str): The user's question. conversation_id (str): The conversation ID. Returns: dict: The response from the Orchestrator. """ client_principal = { 'id': '00000000-0000-0000-0000-000000000000', 'name': 'anonymous' } if question: try: orchestrator = Orchestrator(conversation_id, client_principal) result = asyncio.run(orchestrator.answer(question)) if not isinstance(result, dict): logger.error("Expected result to be a dictionary.") return {"error": "Invalid response format from orchestrator."} return result except Exception as e: logger.exception(f"An error occurred while orchestrating the question: {e}") return {"error": "An error occurred while processing your question."} else: logger.warning("No question provided to orchestrate.") return {"error": "No question provided."} def process_question(question, use_rest_api, orchestrator_endpoint, function_key, conversation_id): """ Process a single question either via REST API or locally. Args: question (str): The question to process. use_rest_api (bool): Flag to determine the method of processing. orchestrator_endpoint (str): The API endpoint URI. function_key (str): The API access key. conversation_id (str): The conversation ID. Returns: dict: The response from the chosen processing method. """ if use_rest_api: response_data = send_question_to_rest_api( orchestrator_endpoint, function_key, question, conversation_id) else: response_data = send_question_to_python(question, conversation_id) return response_data def prettify_jsonl_file(input_file): # Check if the input file exists if not os.path.isfile(input_file): print(f"Error: The file '{input_file}' does not exist.") return try: # Read the entire content from the input file first with open(input_file, 'r', encoding='utf-8') as infile: lines = infile.readlines() # Prettify the content and write back to the same file with open(input_file, 'w', encoding='utf-8') as outfile: for line in lines: # Load the JSON object from the line json_obj = json.loads(line.strip()) # Write the pretty-printed JSON back to the file json.dump(json_obj, outfile, indent=4, ensure_ascii=False) outfile.write('\n') # Add a newline after each JSON object print(f"Prettified JSONL content has been written to '{input_file}'") except Exception as e: print(f"Error occurred while processing the file: {e}") def parse_arguments(): """ Parse command-line arguments. Returns: Namespace: Parsed arguments. """ parser = argparse.ArgumentParser( description="Evaluate questions either by sending them to a REST API or processing them locally." ) parser.add_argument( "--test-data", type=str, required=True, help="Path to the test dataset file in JSONL format.", ) return parser.parse_args() 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}") if __name__ == '__main__': try: main() except Exception as e: logger.exception(f"An unhandled exception occurred: {e}") sys.exit(1)