Generative AI Custom Evaluation - Quickstart

# Loading the credentials from the env file
from gen_ai_hub.proxy.gen_ai_hub_proxy import GenAIHubProxyClient
from dotenv import load_dotenv
import os

load_dotenv(override=True)

# Fetching environment variables
AICORE_BASE_URL = os.getenv("AICORE_BASE_URL")
AICORE_RESOURCE_GROUP = os.getenv("AICORE_RESOURCE_GROUP")
AICORE_AUTH_URL = os.getenv("AICORE_AUTH_URL")
AICORE_CLIENT_ID = os.getenv("AICORE_CLIENT_ID")
AICORE_CLIENT_SECRET = os.getenv("AICORE_CLIENT_SECRET")

AWS_ACCESS_KEY = os.getenv("AWS_ACCESS_KEY")
AWS_BUCKET_ID = os.getenv("AWS_BUCKET_ID")
AWS_REGION = os.getenv("AWS_REGION")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
DEPLOYMENT_URL = os.getenv("DEPLOYMENT_URL")

# Initializing the GenAIHubProxyClient
client = GenAIHubProxyClient(
    base_url=AICORE_BASE_URL,
    auth_url=AICORE_AUTH_URL,
    client_id=AICORE_CLIENT_ID,
    client_secret=AICORE_CLIENT_SECRET,
    resource_group=AICORE_RESOURCE_GROUP
)

import os
import json

def get_dataset_file_name(folder_path):
    """
    Retrieves the name of the first file in the specified folder.
    """
    if not os.path.isdir(folder_path):
        print(f"The folder path '{folder_path}' does not exist.")
        return None

    items_in_folder = os.listdir(folder_path)

    for item in items_in_folder:
        item_path = os.path.join(folder_path, item)
        if os.path.isfile(item_path):
            return item

    print(f"No files were found in the folder '{folder_path}'.")
    return None


# --- MAIN EXECUTION ---
DATASET_FOLDER = "./DATASET"

DATASET_NAME = get_dataset_file_name(DATASET_FOLDER)

if  DATASET_NAME:
    print(f"Dataset name: {DATASET_NAME}")
else:
    print("Missing run or dataset file.")
    raise SystemExit("Exiting due to missing run/dataset file.")

# Register S3 secret with AI Core which will be used an input source 
import requests
import json
import logging

def delete_oss_secret(oss_name=""):
    headers = _get_headers()
    
    DELETE_SECRETS_ENDPOINT = f'/v2/admin/objectStoreSecrets/{oss_name}'
    request_url = f"{AICORE_BASE_URL}{DELETE_SECRETS_ENDPOINT}"
    
    try:
        response = requests.delete(request_url, headers=headers, timeout=120)
        if response.status_code == 202:
            print(f"Successfully deleted object store secret: {oss_name}")
        elif response.status_code == 404:
            print(f"Object store secret not found: {oss_name}. It may not exist.")
        else:
            logging.error(f"Failed to delete object store secret: {oss_name}, Status Code: {response.status_code}")
    except Exception as e:
        logging.error(f"Error occurred while attempting to delete object store secret: {e}")
        raise

def register_oss_secret(oss_name="", path_prefix=""):
    headers = _get_headers()
    
    POST_SECRETS_ENDPOINT = '/v2/admin/objectStoreSecrets'
    request_url = f"{AICORE_BASE_URL}{POST_SECRETS_ENDPOINT}"
    
    request_body = {
        "name": oss_name,
        "data": {
            "AWS_ACCESS_KEY_ID": AWS_ACCESS_KEY,
            "AWS_SECRET_ACCESS_KEY": AWS_SECRET_ACCESS_KEY
        },
        "type": "S3",
        "bucket": AWS_BUCKET_ID,
        "endpoint": "s3-eu-central-1.amazonaws.com",
        "region": AWS_REGION,
        "pathPrefix": path_prefix,
        "verifyssl": "0",
        "usehttps": "1",
    }
    try:
        response = requests.post(
            request_url, headers=headers, data=json.dumps(request_body), timeout=120
        )
        result = response.json()
        return result
    except:
        logging.error("Error occurred while attempting to create object store secret")
        raise
        
delete_oss_secret(oss_name="default")
delete_oss_secret(oss_name="genai-quick-data-notebook")
        
register_oss_secret(oss_name="default", path_prefix="")
register_oss_secret(oss_name="genai-quick-data-notebook", path_prefix="")

# uploading these files to Object store to register as an artifact inside ai core

import boto3
import os
import uuid

def upload_folder_to_s3(folder_path, bucket_name, s3_prefix=""):
    """
    Upload a folder to an S3 bucket recursively.

    :param folder_path: The local folder path to upload.
    :param bucket_name: The name of the S3 bucket.
    :param s3_prefix: Optional prefix to use for the S3 keys (e.g., subfolder in the bucket).
    """
    s3_client = boto3.client(
            's3',
            aws_access_key_id=AWS_ACCESS_KEY,
            aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
            region_name=AWS_REGION
            )

    for root, dirs, files in os.walk(folder_path):
        for file_name in files:
            print("val of root is ", file_name)
            local_path = os.path.join(root, file_name)
            # Compute the relative path for the S3 key
            relative_path = os.path.relpath(local_path, folder_path)
            s3_key = os.path.join(s3_prefix, relative_path).replace("\\", "/")  # Ensure S3-compatible paths
            print("val of s3 key is ", s3_key)
            print(f"Uploading {local_path} to s3://{bucket_name}/{s3_key}")
            
            # Upload the file
            s3_client.upload_file(local_path, bucket_name, s3_key)

# Example usage
folder_to_upload_testdata = "../DATASET"
user_directory_prefix = "" # replace with your i-number as string here
prefix_guid = user_directory_prefix if user_directory_prefix is not None else str(uuid.uuid4().hex)
s3_testdata_prefix = f"genaiEvaluation/{prefix_guid}/testdata" # Leave empty for root of the bucket


upload_folder_to_s3(folder_to_upload_testdata, AWS_BUCKET_ID, s3_testdata_prefix)
input_artifact_path = f"ai://genai-quick-data-notebook/genaiEvaluation/{prefix_guid}"

import requests
import logging
# Registering the uploaded files from AWS as artifacts to use inside configuration.

def register_artifact():
    headers = _get_headers()
    
    GET_ARTIFACTS_ENDPOINT = '/v2/lm/artifacts'
    request_url = f"{AICORE_BASE_URL}{GET_ARTIFACTS_ENDPOINT}"
    
    request_body = {
        "labels": [
            {
            "key": "ext.ai.sap.com/prompt-evaluation",
            "value": "true"
            }
        ],
        "name": "genai-eval-simplified-test-data",
        "kind": "other",
        "url": input_artifact_path, # input artifact path
        "description": "demo artifacts for evaluation flow.",
        "scenarioId": "genai-evaluations"
    }
    try:
        response = requests.post(
            request_url, headers=headers, data=json.dumps(request_body), timeout=120
        )
        result = response.json()
        print(result)
        return result['id']
    except:
        print("Error occurred while attempting to register artifact")
        raise
        
artifact_id = register_artifact()

def create_prompt_template():
    headers = _get_headers()
    GET_PROMPT_TEMPLATES_ENDPOINT = '/v2/lm/promptTemplates'
    request_url = f"{AICORE_BASE_URL}{GET_PROMPT_TEMPLATES_ENDPOINT}"
    
    
    prompt_template = {
    "template": [
        {
        "role": "user",
        "content": "List the benefits and side effects of the drug in the following consumer health question: {{?question}}."
        }
    ]
    }

    request_body = {
    "name": "prompt-registry-eval-demo",
    "version": "1.0.0",
    "scenario": "genai-evaluations",
    "spec": prompt_template
    }
    try:
        response = requests.post(
            request_url, headers=headers, data=json.dumps(request_body), timeout=120
        )
        if(response.status_code != 200):
            raise
        result = response.json()
        print(result)
        return result['id']
    except:
        logging.error("Error occurred while attempting to create a prompt template")
        raise

prompt_template_id = create_prompt_template()

#  creating an AICORE Configuration.
import requests

request_body = {
    "name": "genai-eval-conf",
    "scenarioId": "genai-evaluations",
    "executableId": "genai-evaluations-simplified",
    "inputArtifactBindings": [
        {
            "key": "datasetFolder",
            "artifactId": artifact_id
        }
    ],
    "parameterBindings": [
        {
            "key": "repetitions",
            "value": repetitions
        },
        {
            "key": "orchestrationDeploymentURL",
            "value": orchestration_deployment_url
        },
        {
            "key": "metrics",
            "value": metrics_list
        },
        {
            "key": "testDataset",
            "value": test_datasets
        },
        {
            "key": "promptTemplate",
            "value": prompt_template_id
        },
        {
            "key": "models",
            "value": models_list
        }
    ]
}

def create_aicore_configuration():
    headers = _get_headers()
    GET_CONFIGURATIONS_ENDPOINT = '/v2/lm/configurations'
    request_url = f"{AICORE_BASE_URL}{GET_CONFIGURATIONS_ENDPOINT}"
    try:
        print(request_body)
        response = requests.post(
            request_url, headers=headers, data=json.dumps(request_body), timeout=120
        )
        print(response)
        if(response.status_code != 201):
            raise Exception(f"Failed to create configuration: {response.status_code} - {response.text}")
        result = response.json()
        print(result)
        print(request_body)
        return result['id']
    except:
        logging.error("Error occurred while attempting to create a Configuration")
        raise
        
configuration_id = create_aicore_configuration()

{
  "name": "genai-eval-conf",
  "scenarioId": "genai-evaluations",
  "executableId": "genai-evaluations-simplified",
  "inputArtifactBindings": [
    {
      "key": "datasetFolder",
      "artifactId": "{{artifactId}}"
    }
  ],
  "parameterBindings": [
    {
      "key": "repetitions",
      "value": "1"
    },
    {
      "key": "orchestrationDeploymentURL",
      "value": "{{deployment_url}}"
    },
    {
      "key": "metrics",
      "value": "language_match"
    },
    {
      "key": "testDataset",
      "value": "{\"path\": \"testdata/{{dataset_file}}\", \"type\": \"csv\"}"
    },
    {
      "key": "promptTemplate",
      "value": "{{prompt_template_id}}"
    },
    {
      "key": "models",
      "value": "{{model_name}}:{{model_version}}"
    }
  ]
}

import pandas as pd

def get_model_from_run(run):
    for tag in run.get("tags", []):
        if tag.get("name") == "evaluation.ai.sap.com/model":
            return tag.get("value")

def aggregate_metrics_by_model(runs_list):
    transformed_data = []
    for run in runs_list:
        model = get_model_from_run(run)
        for metric in run["metrics"]:
            metric_value = metric.get("value")

            # Override only for /mode
            if metric.get("name").endswith("/mode"):
                for label in metric.get("labels", []):
                    if label.get("name") == "evaluation.ai.sap.com/mode_category":
                        metric_value = label.get("value")
                        break
            output_json = {
                "model": model,
                "metrics_name": metric.get("name"),
                "metric_value": metric_value
            }
            transformed_data.append(output_json)
    return transformed_data


def create_metrics_pivot_table(transformed_data):
    """
    Creates a pivot table where rows are models and columns are metrics.
    
    Args:
        transformed_data: List of dictionaries with 'model', 'metrics_name', 'metric_value'
    
    Returns:
        DataFrame with models as rows and metrics as columns
    """
    # Convert list of dictionaries to DataFrame
    df = pd.DataFrame(transformed_data)
    
    # Create pivot table
    pivot_table = df.pivot_table(
        index='model',
        columns='metrics_name',
        values='metric_value',
        aggfunc='first'  # Use 'first' to get the single value, or 'mean' if there are duplicates
    )
    
    return pivot_table

transformed_data = aggregate_metrics_by_model(runs_data['resources'])
metrics_pivot = create_metrics_pivot_table(transformed_data)

HTML(metrics_pivot.to_html())

# download the result artifacts from Object store.
import boto3

def download_all_objects(prefix, destination_folder):
    """
    Recursively download all objects from an S3 bucket starting with a specific prefix.

    :param bucket_name: Name of the S3 bucket.
    :param prefix: Prefix to filter objects in the bucket.
    :param destination_folder: Local folder to save the downloaded files.
    """
    s3_client = boto3.client(
            's3',
            aws_access_key_id=AWS_ACCESS_KEY,
            aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
            region_name=AWS_REGION
            )

    # Ensure the destination folder exists
    if not os.path.exists(destination_folder):
        os.makedirs(destination_folder)

    # Paginate through objects
    paginator = s3_client.get_paginator('list_objects_v2')
    pages = paginator.paginate(Bucket=AWS_BUCKET_ID, Prefix=prefix)

    for page in pages:
        if 'Contents' in page:
            for obj in page['Contents']:
                key = obj['Key']
                local_file_path = os.path.join(destination_folder, os.path.relpath(key, prefix))

                # Ensure the local directory structure exists
                local_directory = os.path.dirname(local_file_path)
                if not os.path.exists(local_directory):
                    os.makedirs(local_directory)

                # Download the object
                print(f"Downloading {key} to {local_file_path}")
                s3_client.download_file(AWS_BUCKET_ID, key, local_file_path)


# Download the evaluation results from the object store. Look at execution status under "outputArtifacts" key to see the 'url'
# which shows the data path of where your output results are stored
EXECUTION_ID = execution_id
sqlite_db_prefix = f'{EXECUTION_ID}/tmp/'  # change the prefix based on where your output artifact is stored in the bucket.
destination_folder = 'results-new'

download_all_objects(sqlite_db_prefix, destination_folder)

# viewing the results from sqlite db in tabular format..
import sqlite3
import pandas as pd
from IPython.display import display, HTML

# Path to your SQLite database file
db_file = 'results-new/results.db'

connection = sqlite3.connect(db_file)

# Specify the table names you want to display
table_names = ['run','configuration', 'submission', 'submission_result', 'evaluation_result'] 

# Create the CSS and HTML container
html_content = """
<style>
/* Container to hold all tables */
.fixed-container {
    max-height: 600px;  /* Total container height */
    overflow-y: auto;   /* Vertical scroll for container */
    border: 2px solid #ddd;
    padding: 10px;
}

/* Table styling */
.table-container table {
    border-collapse: collapse; /* Merge table borders */
    width: 100%; /* Full width for tables */
    margin-bottom: 20px; /* Space between tables */
}

.table-container th, .table-container td {
    border: 1px solid #ddd; /* Add gridlines */
    text-align: left;       /* Align text to the left */
    padding: 8px;           /* Cell padding */
    white-space: nowrap;    /* Prevent text wrapping */
}

.table-container th {
    font-weight: bold;         /* Bold header text */
}

/* Allow dynamic column widths */
.table-container td {
    white-space: nowrap;   /* Prevent wrapping for long text */
}

</style>
<div class="fixed-container">
"""

for table_name in table_names:
    query = f"SELECT * FROM {table_name};"
    df = pd.read_sql_query(query, connection)
    # If you want to see all the rows across all tables, remove/comment the next line
    df = df.head(5)  # Limiting the number of rows displayed
    table_html = df.to_html(classes='table-container', index=False)
    html_content += f"""
    <div class="table-container">
        <h3>Table: {table_name}</h3>
        {table_html}
    </div>
    """

html_content += "</div>"

display(HTML(html_content))

# Close the connection
connection.close()

import pandas as pd
import numpy as np
import sqlite3
import json
import os
from IPython.display import display, HTML

# ==========================================
# 1. CONFIGURATION (Separated Groups)
# ==========================================
METRIC_GROUPS = {
    "Categorical": {
        "type": "categorical",
        "description": "Weighted Average (1-5 scale)",
        "metrics": [
            "Pointwise Conciseness", 
            "Pointwise Instruction Following", 
            "Pointwise Correctness", 
            "Pointwise Answer Relevance"
        ]
    },
    "Boolean": {
        "type": "categorical", # Uses same weighted avg logic (0 or 1)
        "description": "Pass Rate (0-1 scale)",
        "metrics": [
            "Exact Match",
            "Content Filter on Input",
            "Content Filter on Output",
            "Language Match",
            "JSON Schema Match"
        ]
    },
    "Numerical": {
        "type": "numerical",
        "description": "Mean Value",
        "metrics": [
            "BLEU", 
            "ROUGE", 
            "BERT Score",
            "test-metric"
        ]
    }
}

# ==========================================
# 2. DATA EXTRACTION
# ==========================================
def extract_db_metadata(db_path):
    if not os.path.exists(db_path): return pd.DataFrame()
    conn = sqlite3.connect(db_path)
    df_runs = pd.read_sql_query("SELECT id, name, tags, config FROM run", conn)
    conn.close()
    
    meta_data = []
    for _, row in df_runs.iterrows():
        run_id = str(row["id"])
        run_name = str(row["name"])
        tags = {}
        config = {}
        try: tags = json.loads(row["tags"]) if isinstance(row["tags"], str) else row["tags"]
        except: pass
        try: config = json.loads(row["config"]) if isinstance(row["config"], str) else row["config"]
        except: pass

        model = "Unknown"
        try: model = config["modules"]["prompt_templating"]["model"]["name"]
        except:
            if isinstance(tags, dict): model = tags.get("evaluation.ai.sap.com/model", "Unknown")
            elif isinstance(tags, list):
                for t in tags: 
                    if t.get("key") == "evaluation.ai.sap.com/model": model = t.get("value")

        meta_data.append({"run_id": run_id, "run_name": run_name, "model": model})
    return pd.DataFrame(meta_data)

def extract_api_metrics(runs_data_resource):
    flat_data = []
    for run in runs_data_resource:
        model = "Unknown"
        for t in run.get("tags", []):
            if t.get("name") == "evaluation.ai.sap.com/model":
                model = t.get("value")
                break
        for m in run.get("metrics", []):
            clean_name = m.get("name", "").replace('"', '').strip()
            flat_data.append({
                "model": model,
                "metrics_name_clean": clean_name,
                "metric_value": m.get("value")
            })
    df = pd.DataFrame(flat_data)
    df['metric_value'] = pd.to_numeric(df['metric_value'], errors='coerce')
    return df

# ==========================================
# 3. SCORING & HELM LOGIC
# ==========================================
def calculate_weighted_avg_score(row, cols):
    """ Returns a score based on counts. 
        Categorical: 1-5 scale. 
        Boolean: 0-1 scale (Pass Rate). 
    """
    total_score = 0
    total_count = 0
    # Check counts 0-5 (covers Boolean 0/1 and Categorical 1-5)
    for rating in range(0, 6):
        col_name = next((c for c in cols if f"/{rating}/count" in c), None)
        if col_name and not pd.isna(row[col_name]):
            count = row[col_name]
            total_score += count * rating
            total_count += count
    return total_score / total_count if total_count > 0 else 0.0

def get_metric_score_series(df_metrics, metric_name, group_type):
    """ Returns a Series of SCORES (Scalar) for each model for a specific metric """
    subset = df_metrics[df_metrics['metrics_name_clean'].str.startswith(metric_name)]
    if subset.empty: return None

    # Pivot to get columns for this metric
    pivot = subset.pivot_table(index='model', columns='metrics_name_clean', values='metric_value', aggfunc='first')
    cols = pivot.columns.tolist()
    
    if group_type == "categorical":
        # Calculate Weighted Average (or Pass Rate for Boolean)
        return pivot.apply(lambda row: calculate_weighted_avg_score(row, cols), axis=1)
    else:
        # Calculate Mean (Numerical)
        c_mean = next((c for c in cols if "mean" in c), None)
        if c_mean: return pivot[c_mean]
        return None

def calculate_group_win_rate(score_table):
    """
    Calculates HELM Win Rate: % of times a model beats another model across all metrics in this group.
    """
    models = score_table.index.tolist()
    metrics = score_table.columns.tolist()
    win_rates = {}

    for model_a in models:
        wins = 0
        comparisons = 0
        
        for model_b in models:
            if model_a == model_b: continue
            
            # Compare across ALL metrics in this table
            for metric in metrics:
                score_a = score_table.at[model_a, metric]
                score_b = score_table.at[model_b, metric]
                
                # Only compare valid scores
                if pd.isna(score_a) or pd.isna(score_b): continue
                
                comparisons += 1
                if score_a > score_b:
                    wins += 1
        
        win_rates[model_a] = wins / comparisons if comparisons > 0 else 0.0
        
    return pd.Series(win_rates)

# ==========================================
# 4. EXECUTION
# ==========================================
db_file = 'results-new/results.db'

# A. Metadata
df_db_meta = extract_db_metadata(db_file)
df_db_unique = df_db_meta.drop_duplicates(subset=['model'], keep='last')

# B. CSS
html_content = """
<style>
.fixed-container {
    max-height: 600px;
    overflow-y: auto;
    border: 2px solid #ddd;
    padding: 10px;
}
.table-container table {
    border-collapse: collapse;
    width: 100%;
    margin-bottom: 20px;
}
.table-container th, .table-container td {
    border: 1px solid #ddd;
    text-align: left;
    padding: 8px;
    white-space: nowrap;
}
.table-container th {
    font-weight: bold;
}
.table-container td {
    white-space: nowrap;
}
</style>
<div class="fixed-container">
"""
if 'runs_data' in locals() and runs_data:
    df_metrics_all = extract_api_metrics(runs_data['resources'])
    
    for group_name, config in METRIC_GROUPS.items():
        
        # 1. Build Score Table
        score_table = pd.DataFrame(index=df_db_unique['model'].unique())
        score_table.index.name = 'model'
        
        valid_metrics = []
        
        # 2. Calculate Scores
        for metric in config["metrics"]:
            scores = get_metric_score_series(df_metrics_all, metric, config["type"])
            if scores is not None:
                score_table[metric] = scores
                valid_metrics.append(metric)
        
        if not valid_metrics:
            continue

        # 3. Calculate HELM Win Rate (Specific to this group)
        score_table['Win Rate'] = calculate_group_win_rate(score_table[valid_metrics])
        
        # 4. Calculate Final Rank
        score_table['Final Rank'] = score_table['Win Rate'].rank(ascending=False, method='min')
        
        # 5. Merge & Format
        df_final = pd.merge(df_db_unique, score_table, on='model', how='inner')
        df_final = df_final.sort_values('Final Rank')
        
        # Rounding
        for c in valid_metrics: df_final[c] = df_final[c].fillna(0.0).astype(float).round(4)
        df_final['Win Rate'] = df_final['Win Rate'].fillna(0.0).astype(float).round(4)
        df_final['Final Rank'] = df_final['Final Rank'].fillna(0).astype(int)
        
        # Columns
        meta_cols = ['run_id', 'run_name', 'model']
        final_cols = meta_cols + ['Win Rate', 'Final Rank'] + valid_metrics
        
        # 6. Generate HTML
        table_html = df_final[final_cols].to_html(classes='table-container', index=False)
        
        html_content += f"""
        <div class="table-container">
            <h3>{group_name} Comparison</h3>
            <p><i>Values: {config['description']}. Win Rate based on head-to-head performance.</i></p>
            {table_html}
        </div>
        """

    html_content += "</div>"
    display(HTML(html_content))
    
else:
    print("'runs_data' missing.")