# Copyright 2024 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
#
# https://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.
# pylint: disable=too-many-lines
"""Utility functions and classes for the VAPO notebook."""
import csv
import io
import json
import random
import re
import string
import subprocess
from collections.abc import Callable
from typing import Any, Union
import ipywidgets as widgets
import jinja2
import jinja2.meta
import pandas as pd
import plotly.graph_objects as go
from IPython.core.display import DisplayHandle
from IPython.display import HTML, display
from google.cloud import aiplatform, storage
from jsonschema import ValidationError, validate
from tenacity import retry, wait_random_exponential
from tensorflow.io import gfile
from vertexai import generative_models
from vertexai.evaluation import EvalTask
from vertexai.generative_models import (
Content,
GenerationConfig,
GenerativeModel,
Part,
SafetySetting,
Tool,
ToolConfig,
)
def is_target_required_metric(eval_metric: str) -> bool:
"""Check if the metric requires the target label."""
return eval_metric in [
"bleu",
"exact_match",
"question_answering_correctness",
"rouge_1",
"rouge_2",
"rouge_l",
"rouge_l_sum",
"tool_call_valid",
"tool_name_match",
"tool_parameter_key_match",
"tool_parameter_kv_match",
]
def is_run_target_required(eval_metric_types: list[str], source_model: str) -> bool:
"""Check if the run requires the target label."""
if source_model:
return False
label_required = False
for metric in eval_metric_types:
label_required = label_required or is_target_required_metric(metric)
return label_required
_TARGET_KEY = "target"
def load_file_from_gcs(dataset: str) -> str:
"""Loads the file from GCS and returns it as a string."""
if dataset.startswith("gs://"):
with gfile.GFile(dataset, "r") as f:
return f.read()
else:
raise ValueError(
"Unsupported file location. Only GCS paths starting with 'gs://' are"
" supported."
)
def parse_jsonl(data_str: str) -> list[dict[str, str]]:
"""Parses the content of a JSONL file and returns a list of dictionaries."""
data = []
lines = data_str.splitlines()
for line in lines:
if line:
try:
data.append(json.loads(line))
except json.JSONDecodeError as e:
raise ValueError(
f"Error decoding JSON on line: {line}. Error: {e}"
) from e
return data
def parse_and_validate_csv(data_str: str) -> list[dict[str, str]]:
"""Parses and validates the content of a CSV file and returns a list of dictionaries."""
data = []
csv_reader = csv.reader(io.StringIO(data_str))
# Extract and validate headers
try:
headers = next(csv_reader)
if not headers:
raise ValueError("The CSV file has an empty or invalid header row.")
except StopIteration as e:
raise ValueError("The CSV file is empty.") from e
# Validate and process rows
for row_number, row in enumerate(csv_reader, start=2):
if len(row) != len(headers):
raise ValueError(
f"Row {row_number} has an inconsistent number of fields. "
f"Expected {len(headers)} fields but found {len(row)}."
)
# Create dictionary for each row using headers as keys
item = dict(zip(headers, row))
data.append(item)
return data
def load_dataset(dataset: str) -> list[dict[str, str]]:
"""Loads and parses the dataset based on its file type ('.jsonl' or '.csv')."""
# Load the file from GCS
data_str = load_file_from_gcs(dataset)
# Parse based on file type
if dataset.endswith(".jsonl"):
return parse_jsonl(data_str)
if dataset.endswith(".csv"):
return parse_and_validate_csv(data_str)
raise ValueError(
"Unsupported file type. Please provide a file with '.jsonl' or '.csv'"
" extension."
)
def validate_prompt_and_data(
template: str,
dataset_path: str,
placeholder_to_content: str,
label_enforced: bool,
) -> None:
"""Validates the prompt template and the dataset."""
data = load_dataset(dataset_path)
placeholder_to_content_json = json.loads(placeholder_to_content)
template = re.sub(r"(? str:
"""A sample to create custom jobs."""
worker_pool_specs = [
{
"replica_count": 1,
"container_spec": {
"image_uri": container_uri,
"args": [f"--{k}={v}" for k, v in container_args.items()],
},
"machine_spec": {
"machine_type": "n1-standard-4",
},
}
]
custom_job = aiplatform.CustomJob(
display_name=display_name,
worker_pool_specs=worker_pool_specs,
)
custom_job.submit()
return custom_job
def run_apd(config: dict[str, str], bucket_uri: str, display_name: str) -> str:
"""A function to the vertex prompt optimizer."""
print(f"\n\nJob display name: {display_name}")
version = "preview_v1_0"
container_uri = "us-docker.pkg.dev/vertex-ai-restricted/builtin-algorithm/apd"
config_path = f"{bucket_uri}/{display_name}/input_config.json"
with gfile.GFile(config_path, "w") as f:
json.dump(config, f)
aiplatform.init(
project=config["project"],
location=config["target_model_location"],
staging_bucket=f"{bucket_uri}/{display_name}",
)
return run_custom_job(
display_name=display_name,
container_uri=f"{container_uri}:{version}",
container_args={"config": config_path},
)
def update_best_display(
df: pd.DataFrame,
textarea: widgets.Textarea,
best_score_label: widgets.Label,
eval_metric: str,
) -> None:
"""Update the best prompt display."""
df["score"] = df[f"metrics.{eval_metric}/mean"]
best_template = df.loc[df["score"].argmax(), "prompt"]
best_score = df.loc[df["score"].argmax(), "score"]
original_score = df.loc[0, "score"]
def placeholder_llm() -> str:
return "{{llm()}}"
env = jinja2.Environment(loader=jinja2.BaseLoader())
env.globals["llm"] = placeholder_llm
best_template = best_template.replace("store('answer', llm())", "llm()")
textarea.value = best_template
improvement = best_score - original_score
no_improvement_str = "\nNo better template is found yet." if not improvement else ""
best_score_label.value = (
f"Score: {best_score} Improvement: {improvement: .3f} {no_improvement_str}"
)
def generate_dataframe(filename: str) -> pd.DataFrame:
"""Generates a pandas dataframe from a json file."""
if not gfile.exists(filename):
return pd.DataFrame()
with gfile.GFile(filename, "r") as f:
try:
data = json.load(f)
except json.JSONDecodeError:
return pd.DataFrame()
return pd.json_normalize(data)
def left_aligned_df_html(df: pd.DataFrame) -> HTML:
"""Displays a Pandas DataFrame in Colab with left-aligned values."""
# Convert to HTML table, but keep the HTML in a variable
html_table = df.to_html(index=False, classes="left-aligned")
# Add CSS styling to left-align table data cells and override default styles
styled_html = f"""
{html_table}
"""
# Display the styled HTML table
return HTML(styled_html)
def extract_top_level_function_name(source_code: str) -> str | None:
"""Extract the top level function name from the source code."""
match = re.search(r"^def\s+([a-zA-Z_]\w*)\s*\(", source_code, re.MULTILINE)
if match:
return match.group(1)
return None
class ProgressForm:
"""A class to display the progress of the optimization job."""
# pylint: disable=too-many-instance-attributes
def __init__(self, params: dict[str, str]) -> None:
"""Initialize the progress form."""
self.instruction_progress_bar = None
self.instruction_display = None
self.instruction_best = None
self.instruction_score = None
self.demo_progress_bar = None
self.demo_display = None
self.demo_best = None
self.demo_score = None
self.job_state_display = display(
HTML("Job State: Not Started!"), display_id=True
)
self.status_display = display(HTML(""), display_id=True)
if params["optimization_mode"] in ["instruction", "instruction_and_demo"]:
(
self.instruction_progress_bar,
self.instruction_display,
self.instruction_best,
self.instruction_score,
) = self.create_progress_ui("Instruction", params["num_steps"])
if params["optimization_mode"] in ["demonstration", "instruction_and_demo"]:
(
self.demo_progress_bar,
self.demo_display,
self.demo_best,
self.demo_score,
) = self.create_progress_ui(
"Demonstration", params["num_demo_set_candidates"]
)
if len(params["eval_metrics_types"]) == 1:
self.eval_metric = params["eval_metrics_types"][0]
else:
self.eval_metric = "composite_metric"
self.output_path = params["output_path"]
self.instruction_df = None
self.demo_df = None
# pylint: disable=too-many-positional-arguments,too-many-arguments
def update_progress(
self,
progress_bar: widgets.IntProgress | None,
templates_file: str,
df: pd.DataFrame | None,
df_display: DisplayHandle,
best_textarea: widgets.Textarea,
best_score: widgets.Label,
eval_metric: str,
) -> pd.DataFrame:
"""Update the progress of the optimization job."""
def get_last_step(df: pd.DataFrame) -> int:
if df.empty:
return -1
return int(df["step"].max())
if progress_bar is None or df is None:
return pd.DataFrame()
new_df = generate_dataframe(templates_file)
last_step = get_last_step(df)
new_last_step = get_last_step(new_df)
if new_last_step > last_step:
df_display.update(left_aligned_df_html(new_df))
update_best_display(new_df, best_textarea, best_score, eval_metric)
progress_bar.value = progress_bar.value + new_last_step - last_step
return new_df
def create_progress_ui(
self, opt_mode: str, num_opt_steps: str
) -> tuple[widgets.IntProgress, DisplayHandle, widgets.Textarea, widgets.Label]:
"""Create the progress UI for a specific optimization mode."""
print(f"\n\n{opt_mode} Optimization")
progress_bar = widgets.IntProgress(
value=0, min=0, max=int(num_opt_steps), step=1, description="Progress"
)
display(progress_bar)
print("\nGenerated Templates:")
templates_display = display("No template is evaluated yet!", display_id=True)
print("\nBest Template so far:")
best_textarea = widgets.Textarea(
value="NA",
disabled=False,
layout=widgets.Layout(width="80%", height="150px"),
)
display(best_textarea)
best_score = widgets.Label(value="Score: NA Improvement: NA")
display(best_score)
return progress_bar, templates_display, best_textarea, best_score
def monitor_progress(self, job: aiplatform.CustomJob) -> bool:
"""Monitor the progress of the optimization job."""
self.job_state_display.update(HTML(f"Job State: {job.state.name}"))
# Initial display of the templates.
instruction_templates_file = f"{self.output_path}/instruction/templates.json"
demo_templates_file = f"{self.output_path}/demonstration/templates.json"
if not job.done():
self.instruction_df = self.update_progress(
self.instruction_progress_bar,
instruction_templates_file,
self.instruction_df,
self.instruction_display,
self.instruction_best,
self.instruction_score,
self.eval_metric,
)
self.demo_df = self.update_progress(
self.demo_progress_bar,
demo_templates_file,
self.demo_df,
self.demo_display,
self.demo_best,
self.demo_score,
self.eval_metric,
)
return True
if job.state.name != "JOB_STATE_SUCCEEDED":
errors = [f"Error: Job failed with error {job.error}."]
for err_file in [
f"{self.output_path}/instruction/error.json",
f"{self.output_path}/demonstration/error.json",
f"{self.output_path}/error.json",
]:
if gfile.exists(err_file):
with gfile.GFile(err_file, "r") as f:
error_json = json.load(f)
errors.append(f"Detailed error: {error_json['Error']}")
errors.append(
f"Please feel free to send {err_file} to the VAPO team to help"
" resolving the issue."
)
errors.append(
"All the templates found before failure can be found under"
f" {self.output_path}"
)
errors.append(
"Please consider rerunning to make sure the failure is intransient."
)
err = "\n".join(errors)
err = err.replace("\n", "
")
self.status_display.update(HTML(f'{err}'))
else:
self.status_display.update(
HTML(
'Job succeeded! All the'
f" artifacts can be found under {self.output_path}"
)
)
return False
def display_dataframe(df: pd.DataFrame) -> None:
"""Display a pandas dataframe in Colab."""
# Function to wrap text in a scrollable div
def wrap_in_scrollable_div(text: str) -> str:
return f'