wehub-resource-sync
1044 lines
41 KiB
Python
1044 lines
41 KiB
Python
# Copyright 2025 Google LLC
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import datetime
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
|
|
import pandas as pd
|
|
import streamlit as st
|
|
import vertexai
|
|
from dotenv import load_dotenv
|
|
from google import genai
|
|
from google.cloud import storage
|
|
from pydantic import BaseModel
|
|
from src.gcp_prompt import GcpPrompt as gcp_prompt
|
|
from vertexai.evaluation import (
|
|
EvalTask,
|
|
MetricPromptTemplateExamples,
|
|
PairwiseMetricPromptTemplate,
|
|
PointwiseMetricPromptTemplate,
|
|
)
|
|
from vertexai.preview import prompts
|
|
|
|
load_dotenv("src/.env")
|
|
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_metric_object_by_name(metric_name: str):
|
|
"""Retrieves a metric template object from its string name."""
|
|
try:
|
|
return MetricPromptTemplateExamples._PROMPT_TEMPLATE_MAP[metric_name]
|
|
except Exception as e:
|
|
logger.exception(f"Failed to get metric object for {metric_name}: {e}")
|
|
raise
|
|
|
|
|
|
def refresh_bucket() -> list[str]:
|
|
"""Refreshes the list of available dataset URIs from the GCS bucket.
|
|
|
|
This function lists all blobs in the configured GCS bucket, filters for
|
|
CSV and JSONL files located within the 'datasets/' prefix, and constructs a list
|
|
of their full gs:// URI paths.
|
|
|
|
Returns:
|
|
A list of strings, where each string is a GCS URI to a dataset file.
|
|
"""
|
|
logger.info("Bucket: %s", os.getenv("BUCKET"))
|
|
bucket = st.session_state.storage_client.bucket(os.getenv("BUCKET"))
|
|
blobs = bucket.list_blobs()
|
|
data_uris = []
|
|
for i in blobs:
|
|
if i.name.split("/")[0] == "datasets" and (
|
|
i.name.endswith(".csv") or i.name.endswith(".jsonl")
|
|
):
|
|
data_uris.append(f"gs://{i.bucket.name}/{i.name}")
|
|
logger.info("Data URIs: %s", data_uris)
|
|
return data_uris
|
|
|
|
|
|
def get_autorater_pairwise_response(metric_prompt: str, model: str) -> dict:
|
|
"""Gets a response from the autorater model for pairwise evaluation.
|
|
|
|
Args:
|
|
metric_prompt: The prompt to send to the autorater model.
|
|
model: The name of the evaluation model to use.
|
|
|
|
Returns:
|
|
A dictionary containing the autorater's response.
|
|
"""
|
|
|
|
class AutoraterResponse(BaseModel):
|
|
pairwise_choice: str
|
|
explanation: str
|
|
|
|
client = genai.Client(
|
|
vertexai=True,
|
|
project=os.getenv("PROJECT_ID"),
|
|
location=os.getenv("LOCATION"),
|
|
)
|
|
|
|
response = client.models.generate_content(
|
|
model=model,
|
|
contents=metric_prompt,
|
|
config={
|
|
"response_mime_type": "application/json",
|
|
"response_schema": AutoraterResponse,
|
|
"safety_settings": [
|
|
{"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_NONE"},
|
|
{
|
|
"category": "HARM_CATEGORY_DANGEROUS_CONTENT",
|
|
"threshold": "BLOCK_NONE",
|
|
},
|
|
{
|
|
"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
|
|
"threshold": "BLOCK_NONE",
|
|
},
|
|
{"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"},
|
|
],
|
|
},
|
|
)
|
|
|
|
if response.parsed:
|
|
return response.parsed.model_dump()
|
|
|
|
try:
|
|
return json.loads(response.text)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to parse response: {e}")
|
|
return {}
|
|
|
|
|
|
def format_display_text(text):
|
|
"""Formats text for display in st.text_area, handling dicts/lists as JSON."""
|
|
if isinstance(text, (dict, list)):
|
|
try:
|
|
return json.dumps(text, indent=2)
|
|
except (TypeError, ValueError):
|
|
return str(text)
|
|
return str(text)
|
|
|
|
|
|
def main() -> None:
|
|
"""Initializes and runs the Streamlit evaluation application.
|
|
|
|
This function sets up the Streamlit page configuration, initializes session state
|
|
variables, and orchestrates the user interface for the evaluation workflow.
|
|
It handles dataset and prompt selection, response generation or loading,
|
|
human-in-the-loop rating, and the launching of automated evaluations.
|
|
"""
|
|
st.set_page_config(
|
|
layout="wide",
|
|
page_title="Prompt Engineering App",
|
|
page_icon="assets/favicon.ico",
|
|
)
|
|
|
|
st.header("Evaluation Set-Up")
|
|
|
|
if "storage_client" not in st.session_state:
|
|
st.session_state["storage_client"] = storage.Client()
|
|
|
|
if "data_uris" not in st.session_state:
|
|
st.session_state["data_uris"] = refresh_bucket()
|
|
|
|
if "current_index" not in st.session_state:
|
|
st.session_state.current_index = 0
|
|
|
|
if "eval_result" not in st.session_state:
|
|
st.session_state.eval_result = None
|
|
|
|
if "custom_eval_result" not in st.session_state:
|
|
st.session_state.custom_eval_result = None
|
|
|
|
if "df_data" not in st.session_state:
|
|
st.session_state.df_data = None
|
|
|
|
if "df_dataset_eval" not in st.session_state:
|
|
st.session_state.df_dataset_eval = None
|
|
|
|
if "all_metrics_eval" not in st.session_state:
|
|
st.session_state.all_metrics_eval = None
|
|
|
|
if "metrics" not in st.session_state:
|
|
st.session_state.metrics = None
|
|
|
|
if "local_prompt" not in st.session_state:
|
|
st.session_state.local_prompt = gcp_prompt()
|
|
|
|
if "cached_data_files" not in st.session_state:
|
|
st.session_state.cached_data_files = {}
|
|
if "last_selected_dataset_for_cache" not in st.session_state:
|
|
st.session_state.last_selected_dataset_for_cache = None
|
|
|
|
if "cached_prompt_versions" not in st.session_state:
|
|
st.session_state.cached_prompt_versions = {}
|
|
|
|
if "last_selected_prompt_for_versions_cache" not in st.session_state:
|
|
st.session_state.last_selected_prompt_for_versions_cache = None
|
|
|
|
if "human_rated_dict" not in st.session_state:
|
|
st.session_state.human_rated_dict = {}
|
|
|
|
if "metric_preview_index" not in st.session_state:
|
|
st.session_state.metric_preview_index = 0
|
|
|
|
if "vertex_session_init" not in st.session_state:
|
|
vertexai.init(
|
|
project=os.getenv("PROJECT_ID"),
|
|
location=os.getenv("LOCATION"),
|
|
staging_bucket=os.getenv("BUCKET"),
|
|
experiment=os.getenv("EXPERIMENT_NAME"),
|
|
)
|
|
st.session_state.vertex_session_init = True
|
|
|
|
data_sets = list({i.split("/")[4] for i in st.session_state.data_uris})
|
|
logger.info(f"Data Sets: {data_sets}")
|
|
st.selectbox(
|
|
"Select an Existing Dataset", options=[None, *data_sets], key="selected_dataset"
|
|
)
|
|
|
|
files_to_display_in_selectbox = []
|
|
if st.session_state.selected_dataset:
|
|
if (
|
|
st.session_state.selected_dataset
|
|
!= st.session_state.last_selected_dataset_for_cache
|
|
or st.session_state.selected_dataset
|
|
not in st.session_state.cached_data_files
|
|
):
|
|
logger.info(
|
|
"Cache miss or dataset changed for files. Fetching for: %s",
|
|
st.session_state.selected_dataset,
|
|
)
|
|
bucket = st.session_state.storage_client.bucket(os.getenv("BUCKET"))
|
|
prefix = f"datasets/{st.session_state.selected_dataset}/"
|
|
blobs_iterator = bucket.list_blobs(prefix=prefix)
|
|
|
|
current_dataset_files = []
|
|
for blob in blobs_iterator:
|
|
if (
|
|
blob.name.endswith(".csv") or blob.name.endswith(".jsonl")
|
|
) and not blob.name.endswith("/"):
|
|
filename = blob.name[len(prefix) :]
|
|
if filename:
|
|
current_dataset_files.append(filename)
|
|
|
|
st.session_state.cached_data_files[st.session_state.selected_dataset] = (
|
|
sorted(set(current_dataset_files))
|
|
)
|
|
st.session_state.last_selected_dataset_for_cache = (
|
|
st.session_state.selected_dataset
|
|
)
|
|
logger.info(
|
|
"Cached files for %s: %s",
|
|
st.session_state.selected_dataset,
|
|
st.session_state.cached_data_files[st.session_state.selected_dataset],
|
|
)
|
|
|
|
if "selected_file_from_dataset" in st.session_state:
|
|
st.session_state.selected_file_from_dataset = None
|
|
logger.info("Reset selected_file_from_dataset due to dataset change.")
|
|
|
|
files_to_display_in_selectbox = st.session_state.cached_data_files.get(
|
|
st.session_state.selected_dataset, []
|
|
)
|
|
|
|
st.selectbox(
|
|
"Select a file from this dataset:",
|
|
options=[None, *files_to_display_in_selectbox],
|
|
key="selected_file_from_dataset",
|
|
)
|
|
|
|
st.text_input("Number of Samples", key="n_samples")
|
|
|
|
st.text_input(
|
|
"Ground Truth Column Name",
|
|
key="ground_truth_column_name",
|
|
value="target",
|
|
help="The name of the column in your dataset that contains the ground truth or target response.",
|
|
)
|
|
|
|
st.selectbox(
|
|
"Select Existing Prompt",
|
|
options=[None, *list(st.session_state.local_prompt.existing_prompts.keys())],
|
|
placeholder="Select Prompt...",
|
|
key="selected_prompt",
|
|
)
|
|
|
|
versions_to_display_in_selectbox = []
|
|
if st.session_state.selected_prompt:
|
|
st.session_state.local_prompt.prompt_meta["name"] = (
|
|
st.session_state.selected_prompt
|
|
)
|
|
|
|
selected_prompt_obj = st.session_state.local_prompt.existing_prompts[
|
|
st.session_state.selected_prompt
|
|
]
|
|
prompt_resource_name_for_cache = str(selected_prompt_obj)
|
|
|
|
if (
|
|
st.session_state.selected_prompt
|
|
!= st.session_state.last_selected_prompt_for_versions_cache
|
|
or prompt_resource_name_for_cache
|
|
not in st.session_state.cached_prompt_versions
|
|
):
|
|
logger.info(
|
|
"Cache miss or prompt changed for versions. Fetching for: %s",
|
|
st.session_state.selected_prompt,
|
|
)
|
|
fetched_versions = [
|
|
v.version_id for v in prompts.list_versions(selected_prompt_obj)
|
|
]
|
|
|
|
st.session_state.cached_prompt_versions[prompt_resource_name_for_cache] = (
|
|
fetched_versions
|
|
)
|
|
st.session_state.last_selected_prompt_for_versions_cache = (
|
|
st.session_state.selected_prompt
|
|
)
|
|
logger.info(
|
|
"Cached versions for %s: %s",
|
|
st.session_state.selected_prompt,
|
|
fetched_versions,
|
|
)
|
|
|
|
if "selected_version" in st.session_state:
|
|
st.session_state.selected_version = None
|
|
logger.info("Reset selected_version due to prompt change.")
|
|
versions_to_display_in_selectbox = st.session_state.cached_prompt_versions.get(
|
|
prompt_resource_name_for_cache, []
|
|
)
|
|
|
|
st.selectbox(
|
|
"Select Version",
|
|
options=versions_to_display_in_selectbox,
|
|
placeholder="Select Version...",
|
|
key="selected_version",
|
|
)
|
|
|
|
st.button("Load Prompt", key="load_prompt_button")
|
|
if st.session_state.load_prompt_button:
|
|
if not st.session_state.get("selected_prompt"):
|
|
st.warning("Please select a prompt before loading.")
|
|
else:
|
|
logger.info(
|
|
f"Selected Prompt ID: {st.session_state.local_prompt.existing_prompts[st.session_state.selected_prompt]}"
|
|
)
|
|
logger.info(f"Version: {st.session_state.selected_version}")
|
|
st.session_state.local_prompt.load_prompt(
|
|
st.session_state.local_prompt.existing_prompts[
|
|
st.session_state.selected_prompt
|
|
],
|
|
st.session_state.selected_prompt,
|
|
st.session_state.selected_version,
|
|
)
|
|
logger.info(
|
|
f"Local Prompt Meta: {st.session_state.local_prompt.prompt_meta}"
|
|
)
|
|
logger.info(
|
|
f"Local Prompt Meta Dict Keys: {st.session_state.local_prompt.prompt_meta.keys()}"
|
|
)
|
|
|
|
st.session_state.prompt_name = (
|
|
st.session_state.local_prompt.prompt_to_run.prompt_name
|
|
)
|
|
st.session_state.prompt_data = (
|
|
st.session_state.local_prompt.prompt_to_run.prompt_data
|
|
)
|
|
st.session_state.model_name = (
|
|
st.session_state.local_prompt.prompt_to_run.model_name.split("/")[-1]
|
|
)
|
|
st.session_state.system_instructions = (
|
|
st.session_state.local_prompt.prompt_to_run.system_instruction
|
|
)
|
|
st.session_state.response_schema = json.dumps(
|
|
st.session_state.local_prompt.prompt_meta.get("response_schema", {})
|
|
)
|
|
st.session_state.generation_config = json.dumps(
|
|
st.session_state.local_prompt.prompt_meta.get("generation_config", {})
|
|
)
|
|
st.session_state.meta_tags = st.session_state.local_prompt.prompt_meta[
|
|
"meta_tags"
|
|
]
|
|
|
|
st.button("Upload Data and Get Responses", key="upload_data_get_responses_button")
|
|
|
|
if (
|
|
st.session_state.upload_data_get_responses_button
|
|
and st.session_state.n_samples
|
|
and st.session_state.selected_dataset
|
|
):
|
|
if not st.session_state.n_samples:
|
|
st.warning("Please enter the Number of Samples.")
|
|
return
|
|
if not st.session_state.selected_dataset:
|
|
st.warning("Please select an Existing Dataset.")
|
|
return
|
|
if not st.session_state.selected_file_from_dataset:
|
|
st.warning("Please select a file from the dataset.")
|
|
return
|
|
|
|
try:
|
|
num_samples = int(st.session_state.n_samples)
|
|
if num_samples <= 0:
|
|
st.warning("Number of Samples must be a positive integer.")
|
|
return
|
|
except ValueError:
|
|
st.warning("Number of Samples must be a valid integer.")
|
|
return
|
|
|
|
gcs_path = f"gs://{os.getenv('BUCKET')}/datasets/{st.session_state.selected_dataset}/{st.session_state.selected_file_from_dataset}"
|
|
st.session_state["input_data_uri"] = gcs_path
|
|
try:
|
|
if gcs_path.endswith(".csv"):
|
|
df_full = pd.read_csv(gcs_path)
|
|
elif gcs_path.endswith(".jsonl"):
|
|
df_full = pd.read_json(gcs_path, lines=True)
|
|
else:
|
|
st.error(
|
|
f"Unsupported file type: {gcs_path.rsplit('.', maxsplit=1)[-1]}"
|
|
)
|
|
return
|
|
except Exception as e:
|
|
st.error(f"Error reading data from {gcs_path}: {e}")
|
|
return
|
|
|
|
df = df_full.iloc[:num_samples]
|
|
if df.empty:
|
|
st.warning(
|
|
"No data found for the first %s samples in %s, or the file is smaller than requested.",
|
|
num_samples,
|
|
st.session_state.selected_file_from_dataset,
|
|
)
|
|
st.session_state.human_rated_dict = {}
|
|
st.session_state.ratings = []
|
|
st.session_state.include_in_evaluations = []
|
|
st.session_state.current_index = 0
|
|
return
|
|
|
|
user_input_list = []
|
|
expected_result_list = []
|
|
assistant_response_list = []
|
|
baseline_model_response_list = []
|
|
|
|
generate = False
|
|
ground_truth_col = st.session_state.ground_truth_column_name
|
|
required_cols_for_loading_existing = [
|
|
"user_input",
|
|
ground_truth_col,
|
|
"assistant_response",
|
|
]
|
|
|
|
if "assistant_response" in df.columns:
|
|
missing_loading_cols = [
|
|
col
|
|
for col in required_cols_for_loading_existing
|
|
if col not in df.columns
|
|
]
|
|
if not missing_loading_cols:
|
|
logger.info("Sufficient columns found to load existing responses.")
|
|
generate = False
|
|
else:
|
|
st.error(
|
|
f"The file has 'assistant_response' column, but is missing other essential columns for loading: {missing_loading_cols}. Required for loading: {required_cols_for_loading_existing}. Found columns: {df.columns.tolist()}",
|
|
)
|
|
st.session_state.human_rated_dict = {}
|
|
st.session_state.ratings = []
|
|
st.session_state.include_in_evaluations = []
|
|
st.session_state.current_index = 0
|
|
return
|
|
else:
|
|
if not st.session_state.get("prompt_data"):
|
|
st.error(
|
|
"To generate new responses, please load a prompt first using the 'Load Prompt' button."
|
|
)
|
|
st.session_state.human_rated_dict = {}
|
|
st.session_state.ratings = []
|
|
st.session_state.include_in_evaluations = []
|
|
st.session_state.current_index = 0
|
|
return
|
|
|
|
template_vars = re.findall(r"{(\w+)}", st.session_state.prompt_data)
|
|
required_cols_for_generating_new = list(set(template_vars))
|
|
|
|
all_required_cols = [*required_cols_for_generating_new, ground_truth_col]
|
|
missing_generating_cols = [
|
|
col for col in all_required_cols if col not in df.columns
|
|
]
|
|
|
|
if not missing_generating_cols:
|
|
logger.info(
|
|
"'assistant_response' column not found. Required columns for generating new responses are present. Will generate."
|
|
)
|
|
generate = True
|
|
else:
|
|
st.error(
|
|
f"The file does not have 'assistant_response' column, and is also missing columns required for generating new responses based on the loaded prompt: {missing_generating_cols}. Required for generation: {all_required_cols}. Found columns: {df.columns.tolist()}",
|
|
)
|
|
st.session_state.human_rated_dict = {}
|
|
st.session_state.ratings = []
|
|
st.session_state.include_in_evaluations = []
|
|
st.session_state.current_index = 0
|
|
return
|
|
|
|
logger.info("Generate flag set to: %s for %s samples.", generate, len(df))
|
|
|
|
if generate:
|
|
logger.info("Proceeding with generating new assistant responses.")
|
|
if (
|
|
not st.session_state.selected_prompt
|
|
or not st.session_state.selected_version
|
|
):
|
|
st.error(
|
|
"A prompt and version must be loaded to generate new responses. Please use the 'Load Prompt' button."
|
|
)
|
|
st.session_state.human_rated_dict = {}
|
|
st.session_state.ratings = []
|
|
st.session_state.include_in_evaluations = []
|
|
st.session_state.current_index = 0
|
|
return
|
|
if not st.session_state.local_prompt.prompt_to_run.prompt_data:
|
|
st.error(
|
|
"Prompt data is missing from the loaded prompt. Cannot generate. Please re-load the prompt using 'Load Prompt'."
|
|
)
|
|
st.session_state.human_rated_dict = {}
|
|
st.session_state.ratings = []
|
|
st.session_state.include_in_evaluations = []
|
|
st.session_state.current_index = 0
|
|
return
|
|
|
|
import concurrent.futures
|
|
|
|
template_vars = re.findall(r"{(\w+)}", st.session_state.prompt_data)
|
|
required_cols_for_generating_new = list(set(template_vars))
|
|
|
|
tasks = []
|
|
for idx, r in df.iterrows():
|
|
current_user_input_item = {
|
|
col: r[col] for col in required_cols_for_generating_new
|
|
}
|
|
expected_res = r[st.session_state.ground_truth_column_name]
|
|
tasks.append((idx, current_user_input_item, expected_res))
|
|
|
|
# Bind the function reference to avoid accessing st.session_state directly in threads
|
|
generate_function = st.session_state.local_prompt.generate_response
|
|
|
|
def process_task(task_args):
|
|
t_idx, item, expected = task_args
|
|
try:
|
|
res = generate_function(item)
|
|
return t_idx, item, expected, res, None
|
|
except Exception as e:
|
|
return t_idx, item, expected, None, e
|
|
|
|
completed = 0
|
|
total = len(tasks)
|
|
results = []
|
|
|
|
if total > 0:
|
|
progress_bar = st.progress(
|
|
0, text="Generating responses in parallel..."
|
|
)
|
|
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
|
|
future_to_idx = {
|
|
executor.submit(process_task, t): t[0] for t in tasks
|
|
}
|
|
for future in concurrent.futures.as_completed(future_to_idx):
|
|
t_idx, item, expected, res, error = future.result()
|
|
completed += 1
|
|
|
|
progress_text = f"Generating response {completed} of {total}..."
|
|
progress_bar.progress(completed / total, text=progress_text)
|
|
|
|
if error:
|
|
logger.error(
|
|
"Error generating response for row index %s: %s",
|
|
t_idx,
|
|
error,
|
|
)
|
|
else:
|
|
results.append((t_idx, item, expected, res))
|
|
|
|
progress_bar.empty()
|
|
if len(results) < total:
|
|
st.warning(
|
|
f"Generated {len(results)} responses. {total - len(results)} failed."
|
|
)
|
|
|
|
results.sort(key=lambda x: x[0])
|
|
for t_idx, item, expected, res in results:
|
|
user_input_list.append(item)
|
|
expected_result_list.append(expected)
|
|
assistant_response_list.append(res)
|
|
|
|
if len(user_input_list) < len(df):
|
|
st.info(
|
|
f"Successfully generated responses for {len(user_input_list)} out of {len(df)} requested samples due to errors during generation."
|
|
)
|
|
else:
|
|
logger.info(
|
|
"Proceeding with loading existing assistant responses from file."
|
|
)
|
|
parsed_user_inputs_temp = []
|
|
for item_str in df["user_input"].astype(str).tolist():
|
|
try:
|
|
parsed_user_inputs_temp.append(json.loads(item_str))
|
|
except json.JSONDecodeError:
|
|
logger.debug(
|
|
"User input item is not valid JSON, using as raw string: %s",
|
|
item_str[:100],
|
|
)
|
|
parsed_user_inputs_temp.append(item_str)
|
|
user_input_list = parsed_user_inputs_temp
|
|
expected_result_list = df[
|
|
st.session_state.ground_truth_column_name
|
|
].tolist()
|
|
assistant_response_list = df.assistant_response.tolist()
|
|
baseline_model_response_list = []
|
|
if "baseline_model_response" in df.columns:
|
|
baseline_model_response_list = df.baseline_model_response.tolist()
|
|
|
|
st.session_state.human_rated_dict = {
|
|
"user_input": user_input_list,
|
|
"ground_truth": expected_result_list,
|
|
"assistant_response": assistant_response_list,
|
|
}
|
|
if baseline_model_response_list:
|
|
st.session_state.human_rated_dict["baseline_model_response"] = (
|
|
baseline_model_response_list
|
|
)
|
|
num_processed_items = len(user_input_list)
|
|
|
|
if num_processed_items > 0:
|
|
st.session_state.include_in_evaluations = [True] * num_processed_items
|
|
st.session_state.current_index = 0
|
|
st.success(f"Successfully processed {num_processed_items} samples.")
|
|
else:
|
|
st.warning(
|
|
"No data items were processed successfully. Check logs for errors or review file structure."
|
|
)
|
|
st.session_state.human_rated_dict = {}
|
|
st.session_state.include_in_evaluations = []
|
|
st.session_state.current_index = 0
|
|
|
|
st.divider()
|
|
|
|
if st.session_state.human_rated_dict:
|
|
st.title("Review Responses")
|
|
|
|
col1, col2, col3 = st.columns(3)
|
|
|
|
with col1:
|
|
st.subheader("User Input")
|
|
st.text_area(
|
|
label="User's original query/text",
|
|
value=format_display_text(
|
|
st.session_state.human_rated_dict["user_input"][
|
|
st.session_state.current_index
|
|
]
|
|
),
|
|
height=200,
|
|
disabled=True,
|
|
)
|
|
|
|
with col2:
|
|
st.subheader("Ground Truth")
|
|
st.text_area(
|
|
label="The ideal/target response",
|
|
value=format_display_text(
|
|
st.session_state.human_rated_dict["ground_truth"][
|
|
st.session_state.current_index
|
|
]
|
|
),
|
|
height=200,
|
|
disabled=True,
|
|
)
|
|
|
|
with col3:
|
|
st.subheader("Assistant Response")
|
|
st.text_area(
|
|
label="The assistant's generated response",
|
|
value=format_display_text(
|
|
st.session_state.human_rated_dict["assistant_response"][
|
|
st.session_state.current_index
|
|
]
|
|
),
|
|
height=200,
|
|
disabled=True,
|
|
)
|
|
|
|
eval_include = st.checkbox(
|
|
"Include in Evaluation",
|
|
value=st.session_state.include_in_evaluations[
|
|
st.session_state.current_index
|
|
],
|
|
)
|
|
|
|
if (
|
|
st.session_state.include_in_evaluations
|
|
and eval_include
|
|
!= st.session_state.include_in_evaluations[st.session_state.current_index]
|
|
):
|
|
st.session_state.include_in_evaluations[st.session_state.current_index] = (
|
|
eval_include
|
|
)
|
|
|
|
st.markdown("---")
|
|
col_prev, col_spacer, col_next = st.columns([1, 3, 1])
|
|
|
|
with col_prev:
|
|
if st.button("Previous", disabled=(st.session_state.current_index == 0)):
|
|
st.session_state.current_index -= 1
|
|
st.rerun()
|
|
|
|
with col_next:
|
|
if st.button(
|
|
"Next",
|
|
disabled=(
|
|
st.session_state.current_index
|
|
== len(st.session_state.human_rated_dict["user_input"]) - 1
|
|
),
|
|
):
|
|
st.session_state.current_index += 1
|
|
st.rerun()
|
|
|
|
st.markdown(
|
|
f"<p style='text-align: center; font-size: 1.2em;'>Case {st.session_state.current_index + 1} of {len(st.session_state.human_rated_dict['user_input'])}</p>",
|
|
unsafe_allow_html=True,
|
|
)
|
|
|
|
st.markdown("---")
|
|
st.subheader("Launch Eval")
|
|
|
|
st.subheader("Metrics Selection")
|
|
|
|
col1, col2 = st.columns(2)
|
|
|
|
with col1:
|
|
st.write("**Model-Based**")
|
|
metric_names = MetricPromptTemplateExamples.list_example_metric_names()
|
|
selected_model_based_metrics = st.multiselect(
|
|
"Select from model-based metrics",
|
|
metric_names,
|
|
key="selected_model_based_metrics",
|
|
label_visibility="collapsed",
|
|
)
|
|
if selected_model_based_metrics:
|
|
num_metrics = len(selected_model_based_metrics)
|
|
if st.session_state.metric_preview_index >= num_metrics:
|
|
st.session_state.metric_preview_index = 0
|
|
|
|
current_metric_name = selected_model_based_metrics[
|
|
st.session_state.metric_preview_index
|
|
]
|
|
|
|
st.markdown(
|
|
f"**Previewing Template: {current_metric_name} ({st.session_state.metric_preview_index + 1}/{num_metrics})**"
|
|
)
|
|
|
|
try:
|
|
metric_object = get_metric_object_by_name(current_metric_name)
|
|
if isinstance(
|
|
metric_object,
|
|
PointwiseMetricPromptTemplate | PairwiseMetricPromptTemplate,
|
|
):
|
|
st.text_area(
|
|
"Template Preview",
|
|
metric_object.metric_prompt_template,
|
|
height=200,
|
|
)
|
|
except Exception as e:
|
|
st.error(
|
|
f"Could not retrieve template for {current_metric_name}: {e}"
|
|
)
|
|
|
|
if num_metrics > 1:
|
|
prev_col, next_col = st.columns(2)
|
|
with prev_col:
|
|
if st.button(
|
|
"Previous Template",
|
|
disabled=st.session_state.metric_preview_index <= 0,
|
|
):
|
|
st.session_state.metric_preview_index -= 1
|
|
st.rerun()
|
|
with next_col:
|
|
if st.button(
|
|
"Next Template",
|
|
disabled=st.session_state.metric_preview_index
|
|
>= num_metrics - 1,
|
|
):
|
|
st.session_state.metric_preview_index += 1
|
|
st.rerun()
|
|
|
|
with col2:
|
|
st.write("**Computation-Based Pointwise**")
|
|
computation_based_pointwise = [
|
|
"bleu",
|
|
"rouge_1",
|
|
"rouge_2",
|
|
"rouge_l",
|
|
"rouge_l_sum",
|
|
"exact_match",
|
|
]
|
|
st.multiselect(
|
|
"Select from computation-based pointwise metrics",
|
|
computation_based_pointwise,
|
|
key="selected_cbp",
|
|
label_visibility="collapsed",
|
|
)
|
|
|
|
st.selectbox(
|
|
"Select Evaluation Model",
|
|
options=[
|
|
"gemini-2.0-flash-lite",
|
|
"gemini-2.5-flash",
|
|
"gemini-2.5-pro",
|
|
],
|
|
key="selected_evaluation_model",
|
|
)
|
|
|
|
st.button("Launch Eval", key="launch_eval_button")
|
|
|
|
if st.session_state.launch_eval_button:
|
|
selected_mbp_names = st.session_state.get(
|
|
"selected_model_based_metrics", []
|
|
)
|
|
selected_cbp_metrics = st.session_state.get("selected_cbp", [])
|
|
|
|
all_metrics = selected_mbp_names + selected_cbp_metrics
|
|
|
|
if not all_metrics:
|
|
st.warning("Please select at least one evaluation metric.")
|
|
return
|
|
|
|
evaluation_data_list = []
|
|
for idx, include_item in enumerate(st.session_state.include_in_evaluations):
|
|
if include_item:
|
|
user_input_values = st.session_state.human_rated_dict["user_input"][
|
|
idx
|
|
]
|
|
prompt_template = (
|
|
st.session_state.local_prompt.prompt_to_run.prompt_data
|
|
)
|
|
logger.info(f"Prompt template: {prompt_template}")
|
|
system_instruction = (
|
|
st.session_state.local_prompt.prompt_to_run.system_instruction
|
|
)
|
|
prediction = str(
|
|
st.session_state.human_rated_dict["assistant_response"][idx]
|
|
)
|
|
|
|
# Process reference value like in the old code
|
|
reference_val = st.session_state.human_rated_dict["ground_truth"][
|
|
idx
|
|
]
|
|
final_reference_str = ""
|
|
if isinstance(reference_val, int | float | bool):
|
|
final_reference_str = json.dumps({"value": reference_val})
|
|
elif isinstance(reference_val, str):
|
|
try:
|
|
parsed_json = json.loads(reference_val)
|
|
if isinstance(parsed_json, int | float | bool):
|
|
final_reference_str = json.dumps({"value": parsed_json})
|
|
else:
|
|
final_reference_str = reference_val
|
|
except json.JSONDecodeError:
|
|
final_reference_str = reference_val
|
|
elif isinstance(reference_val, dict | list):
|
|
final_reference_str = json.dumps(reference_val)
|
|
else:
|
|
final_reference_str = str(reference_val)
|
|
|
|
# Extract all variables from the template
|
|
template_vars = re.findall(r"\{(\w+)\}", prompt_template)
|
|
|
|
if template_vars:
|
|
# Template expects variables
|
|
if isinstance(user_input_values, dict):
|
|
# Ensure all expected vars are present, default to empty string if missing
|
|
formatted_input = {
|
|
v: user_input_values.get(v, "") for v in template_vars
|
|
}
|
|
instruction = prompt_template.format(**formatted_input)
|
|
# Input is not a dict but template expects vars
|
|
# Try to use the input as the first variable if there's only one, otherwise empty
|
|
elif len(template_vars) == 1:
|
|
instruction = prompt_template.format(
|
|
**{template_vars[0]: user_input_values}
|
|
)
|
|
else:
|
|
instruction = prompt_template # Fallback
|
|
else:
|
|
# No variables in template
|
|
instruction = prompt_template
|
|
context = system_instruction if system_instruction else ""
|
|
prompt_str = (
|
|
json.dumps(user_input_values)
|
|
if isinstance(user_input_values, dict)
|
|
else str(user_input_values)
|
|
)
|
|
|
|
eval_item = {
|
|
"context": context,
|
|
"instruction": instruction,
|
|
"prompt": prompt_str,
|
|
"prediction": prediction,
|
|
"reference": final_reference_str,
|
|
}
|
|
evaluation_data_list.append(eval_item)
|
|
|
|
if not evaluation_data_list:
|
|
st.warning(
|
|
"No items were selected for evaluation. Please check the 'Include in Evaluation' checkboxes."
|
|
)
|
|
return
|
|
|
|
df_dataset = pd.DataFrame(evaluation_data_list)
|
|
st.session_state.df_dataset_eval = df_dataset
|
|
st.session_state.all_metrics_eval = all_metrics
|
|
logger.info(f"Evaluation DataFrame columns: {df_dataset.columns.tolist()}")
|
|
logger.info(f"Evaluation DataFrame head:\n{df_dataset.head()}")
|
|
|
|
task = EvalTask(
|
|
dataset=df_dataset,
|
|
metrics=all_metrics,
|
|
experiment=os.getenv("EXPERIMENT_NAME"),
|
|
)
|
|
st.session_state.eval_result = task.evaluate(
|
|
response_column_name="prediction",
|
|
baseline_model_response_column_name="reference",
|
|
)
|
|
|
|
st.markdown("---")
|
|
st.subheader("View Eval")
|
|
|
|
st.button("View Evaluation Results", key="eval_results_button")
|
|
|
|
if st.session_state.eval_result and st.session_state.eval_results_button:
|
|
print(st.session_state.eval_result.metrics_table)
|
|
st.dataframe(st.session_state.eval_result.metrics_table)
|
|
if st.session_state.eval_result:
|
|
st.markdown("---")
|
|
st.subheader("Summary Scores")
|
|
|
|
mean_scores = {}
|
|
if (
|
|
st.session_state.eval_result
|
|
and hasattr(st.session_state.eval_result, "metrics_table")
|
|
and not st.session_state.eval_result.metrics_table.empty
|
|
):
|
|
for col in st.session_state.eval_result.metrics_table.columns:
|
|
if col.endswith("/score"):
|
|
scores = pd.to_numeric(
|
|
st.session_state.eval_result.metrics_table[col],
|
|
errors="coerce",
|
|
)
|
|
if not scores.dropna().empty:
|
|
mean_scores[col] = scores.dropna().mean()
|
|
if mean_scores:
|
|
for metric, score in mean_scores.items():
|
|
st.metric(label=f"Mean {metric}", value=f"{score:.2f}")
|
|
else:
|
|
st.metric(
|
|
label="Mean Automated Score",
|
|
value="N/A",
|
|
)
|
|
|
|
st.markdown("---")
|
|
st.subheader("Save to Prompt Records")
|
|
save_to_records = st.checkbox(
|
|
"I want to save the results of this evaluation to the prompt records.",
|
|
key="save_to_records_checkbox",
|
|
)
|
|
if st.button("Save to Prompt Records", key="save_to_records_button"):
|
|
if save_to_records:
|
|
prompt_name = st.session_state.selected_prompt
|
|
prompt_version = st.session_state.selected_version
|
|
data_file = st.session_state.input_data_uri
|
|
|
|
if "df_dataset_eval" in st.session_state:
|
|
data = st.session_state.df_dataset_eval.to_dict(
|
|
orient="records"
|
|
)
|
|
else:
|
|
st.error(
|
|
"Evaluation data not found in session state. Please re-run evaluation."
|
|
)
|
|
return
|
|
|
|
if "all_metrics_eval" in st.session_state:
|
|
metrics = st.session_state.all_metrics_eval
|
|
else:
|
|
st.error(
|
|
"Metrics not found in session state. Please re-run evaluation."
|
|
)
|
|
return
|
|
|
|
scores_df = st.session_state.eval_result.metrics_table
|
|
scores = scores_df.to_dict(orient="records")
|
|
|
|
mean_scores_to_save = {}
|
|
for col in scores_df.columns:
|
|
if col.endswith("/score"):
|
|
s = pd.to_numeric(scores_df[col], errors="coerce")
|
|
if not s.dropna().empty:
|
|
mean_scores_to_save[col] = s.dropna().mean()
|
|
|
|
record_data = {
|
|
"prompt_name": prompt_name,
|
|
"prompt_version": prompt_version,
|
|
"data_file": data_file,
|
|
"metrics": metrics,
|
|
"mean_scores": mean_scores_to_save,
|
|
"scores": scores,
|
|
"evaluation_data": data,
|
|
"timestamp": datetime.datetime.now().isoformat(),
|
|
}
|
|
|
|
try:
|
|
filename = f"record_{prompt_name}_v{prompt_version}_{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}.json"
|
|
bucket = st.session_state.storage_client.bucket(
|
|
os.getenv("BUCKET")
|
|
)
|
|
blob = bucket.blob(f"records/{filename}")
|
|
|
|
json_data = json.dumps(record_data, indent=4)
|
|
blob.upload_from_string(
|
|
json_data, content_type="application/json"
|
|
)
|
|
|
|
gcs_path = f"gs://{os.getenv('BUCKET')}/records/{filename}"
|
|
|
|
st.success(
|
|
f"Successfully saved to prompt records at: {gcs_path}"
|
|
)
|
|
st.json(json_data)
|
|
except Exception as e:
|
|
st.error(f"Failed to save to GCS: {e}")
|
|
else:
|
|
st.warning(
|
|
"Please check the box to confirm you want to save to prompt records."
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|