# 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. # """Visualization utilities for GenAI Evaluation SDK.""" import base64 import json import logging import textwrap from typing import Any, Optional import pandas as pd from pydantic import errors from . import types logger = logging.getLogger(__name__) def _is_ipython_env() -> bool: """Checks if the code is running in an IPython environment.""" try: from IPython import get_ipython return get_ipython() is not None except ImportError: return False def _pydantic_serializer(obj: Any) -> Any: """Custom serializer for Pydantic models.""" if hasattr(obj, "model_dump"): return obj.model_dump(mode="json") raise TypeError(f"Object of type {obj.__class__.__name__} is not JSON serializable") def _preprocess_df_for_json(df: Optional[pd.DataFrame]) -> Optional[pd.DataFrame]: """Prepares a DataFrame for JSON serialization by converting complex objects to strings.""" if df is None: return None df_copy = df.copy() for col in df_copy.columns: if ( df_copy[col].dtype == "object" or df_copy[col].apply(lambda x: isinstance(x, (dict, list))).any() ): def stringify_cell(cell: Any) -> Optional[str]: if isinstance(cell, (dict, list)): try: return json.dumps( cell, ensure_ascii=False, default=_pydantic_serializer ) except TypeError: return str(cell) elif pd.isna(cell): return None elif not isinstance(cell, (str, int, float, bool)): if hasattr(cell, "model_dump"): return json.dumps( cell.model_dump(mode="json"), ensure_ascii=False ) return str(cell) return str(cell) df_copy[col] = df_copy[col].apply(stringify_cell) return df_copy def _encode_to_base64(data: str) -> str: """Encodes a string to a web-safe Base64 string.""" return base64.b64encode(data.encode("utf-8")).decode("utf-8") def _get_evaluation_html(eval_result_json: str) -> str: """Returns a self-contained HTML for single evaluation visualization.""" payload_b64 = _encode_to_base64(eval_result_json) return textwrap.dedent( f""" Evaluation Report

Evaluation Report

<
""" ) def _get_comparison_html(eval_result_json: str) -> str: """Returns a self-contained HTML for a side-by-side eval comparison.""" payload_b64 = _encode_to_base64(eval_result_json) return textwrap.dedent( f""" Eval Comparison Report

Eval Comparison Report

<
""" ) def _get_inference_html(dataframe_json: str) -> str: """Returns a self-contained HTML for displaying inference results.""" payload_b64 = _encode_to_base64(dataframe_json) return textwrap.dedent( f""" Evaluation Dataset <

Evaluation Dataset

""" ) def _extract_text_and_raw_json(content: Any) -> dict[str, str]: """Extracts display text and raw JSON from a content object. This function handles raw strings, Gemini's `contents` format, and OpenAI's `messages` format. Args: content: The content from a 'prompt', 'request', or 'response' column. Returns: A dictionary with 'display_text' for direct rendering and 'raw_json' for an expandable view. """ if not isinstance(content, (str, dict)): return {"display_text": str(content or ""), "raw_json": ""} try: data = json.loads(content) if isinstance(content, str) else content if not isinstance(data, dict): return {"display_text": str(content), "raw_json": ""} pretty_json = json.dumps(data, indent=2, ensure_ascii=False) # Gemini format check. if ( "contents" in data and isinstance(data.get("contents"), list) and data["contents"] ): first_part = data["contents"][0].get("parts", [{}])[0] display_text = first_part.get("text", str(data)) return {"display_text": display_text, "raw_json": pretty_json} # OpenAI response format check. elif ( "choices" in data and isinstance(data.get("choices"), list) and data["choices"] ): message = data["choices"][0].get("message", {}) display_text = message.get("content", str(data)) return {"display_text": display_text, "raw_json": pretty_json} # OpenAI request format check. elif ( "messages" in data and isinstance(data.get("messages"), list) and data["messages"] ): user_messages = [ message.get("content", "") for message in data["messages"] if message.get("role") == "user" ] display_text = user_messages[-1] if user_messages else str(data) return {"display_text": display_text, "raw_json": pretty_json} else: # Not a recognized format. return {"display_text": str(content), "raw_json": pretty_json} except (json.JSONDecodeError, TypeError, IndexError): return {"display_text": str(content), "raw_json": ""} def display_evaluation_result( eval_result_obj: types.EvaluationResult, candidate_names: Optional[list[str]] = None, ) -> None: """Displays evaluation result in an IPython environment.""" if not _is_ipython_env(): logger.warning("Skipping display: not in an IPython environment.") return else: from IPython import display try: result_dump = eval_result_obj.model_dump( mode="json", exclude_none=True, exclude={"evaluation_dataset"} ) except errors.PydanticSerializationError as e: logger.error( "Serialization Error: %s\nCould not display the evaluation " "result due to a data serialization issue. Please check the " "content of the EvaluationResult object.", e, ) return except Exception as e: logger.error("Failed to serialize EvaluationResult: %s", e, exc_info=True) raise input_dataset_list = eval_result_obj.evaluation_dataset is_comparison = input_dataset_list and len(input_dataset_list) > 1 metadata_payload = result_dump.get("metadata", {}) metadata_payload["candidate_names"] = candidate_names or metadata_payload.get( "candidate_names" ) if is_comparison: if ( input_dataset_list and input_dataset_list[0] and input_dataset_list[0].eval_dataset_df is not None ): base_df = _preprocess_df_for_json(input_dataset_list[0].eval_dataset_df) processed_rows: list[dict[str, Any]] = [] if base_df is not None: for _, row in base_df.iterrows(): prompt_key = "request" if "request" in row else "prompt" prompt_info = _extract_text_and_raw_json(row.get(prompt_key)) processed_row = { "prompt_display_text": prompt_info["display_text"], "prompt_raw_json": prompt_info["raw_json"], "reference": row.get("reference", ""), } processed_rows.append(processed_row) metadata_payload["dataset"] = processed_rows if "eval_case_results" in result_dump: for case_res in result_dump["eval_case_results"]: for resp_idx, cand_res in enumerate( case_res.get("response_candidate_results", []) ): if ( resp_idx < len(input_dataset_list) and input_dataset_list is not None and input_dataset_list[resp_idx].eval_dataset_df is not None ): df = _preprocess_df_for_json( input_dataset_list[resp_idx].eval_dataset_df ) case_idx = case_res.get("eval_case_index") if ( df is not None and case_idx is not None and case_idx < len(df) ): response_content = df.iloc[case_idx].get("response") display_info = _extract_text_and_raw_json(response_content) cand_res["display_text"] = display_info["display_text"] cand_res["raw_json"] = display_info["raw_json"] win_rates = eval_result_obj.win_rates if eval_result_obj.win_rates else {} if "summary_metrics" in result_dump: for summary in result_dump["summary_metrics"]: if summary.get("metric_name") in win_rates: summary.update(win_rates[summary["metric_name"]]) result_dump["metadata"] = metadata_payload html_content = _get_comparison_html(json.dumps(result_dump)) else: single_dataset = input_dataset_list[0] if input_dataset_list else None processed_rows = [] if ( single_dataset is not None and isinstance(single_dataset, types.EvaluationDataset) and single_dataset.eval_dataset_df is not None ): processed_df = _preprocess_df_for_json(single_dataset.eval_dataset_df) if processed_df is not None: for _, row in processed_df.iterrows(): prompt_key = "request" if "request" in row else "prompt" prompt_info = _extract_text_and_raw_json(row.get(prompt_key)) response_info = _extract_text_and_raw_json(row.get("response")) processed_row = { "prompt_display_text": prompt_info["display_text"], "prompt_raw_json": prompt_info["raw_json"], "reference": row.get("reference", ""), "response_display_text": response_info["display_text"], "response_raw_json": response_info["raw_json"], "intermediate_events": row.get("intermediate_events", None), } processed_rows.append(processed_row) metadata_payload["dataset"] = processed_rows if "eval_case_results" in result_dump and processed_rows: for case_res in result_dump["eval_case_results"]: case_idx = case_res.get("eval_case_index") if ( case_idx is not None and case_idx < len(processed_rows) and case_res.get("response_candidate_results") ): original_case = processed_rows[case_idx] cand_res = case_res["response_candidate_results"][0] cand_res["display_text"] = original_case[ "response_display_text" ] cand_res["raw_json"] = original_case["response_raw_json"] result_dump["metadata"] = metadata_payload html_content = _get_evaluation_html(json.dumps(result_dump)) display.display(display.HTML(html_content)) def display_evaluation_dataset(eval_dataset_obj: types.EvaluationDataset) -> None: """Displays an evaluation dataset in an IPython environment.""" if not _is_ipython_env(): logger.warning("Skipping display: not in an IPython environment.") return else: from IPython import display if ( eval_dataset_obj.eval_dataset_df is None or eval_dataset_obj.eval_dataset_df.empty ): logger.warning("No inference data to display.") return processed_rows = [] df = eval_dataset_obj.eval_dataset_df for _, row in df.iterrows(): processed_row = {} for col_name, cell_value in row.items(): if col_name in ["prompt", "request", "response"]: processed_row[col_name] = _extract_text_and_raw_json(cell_value) elif col_name == "rubric_groups": # Special handling for rubric_groups to keep it as a dict if isinstance(cell_value, dict): processed_row[col_name] = { k: [ ( v_item.model_dump(mode="json") if hasattr(v_item, "model_dump") else v_item ) for v_item in v ] for k, v in cell_value.items() } else: processed_row[col_name] = cell_value else: if isinstance(cell_value, (dict, list)): processed_row[col_name] = json.dumps( cell_value, ensure_ascii=False, default=_pydantic_serializer ) else: processed_row[col_name] = cell_value processed_rows.append(processed_row) dataframe_json_string = json.dumps(processed_rows, ensure_ascii=False, default=str) html_content = _get_inference_html(dataframe_json_string) display.display(display.HTML(html_content)) def _get_status_html(status: str, error_message: Optional[str] = None) -> str: """Returns a simple HTML string for displaying a status and optional error.""" error_html = "" if error_message: error_html = f"""

Error: 

{error_message}

""" return textwrap.dedent( f"""

Status: {status}

{error_html}
""" ) def display_evaluation_run_status(eval_run_obj: "types.EvaluationRun") -> None: """Displays the status of an evaluation run in an IPython environment.""" if not _is_ipython_env(): logger.warning("Skipping display: not in an IPython environment.") return else: from IPython import display status = eval_run_obj.state.name if eval_run_obj.state else "UNKNOWN" error_message = str(eval_run_obj.error) if eval_run_obj.error else None html_content = _get_status_html(status, error_message) display.display(display.HTML(html_content))