mirror of
https://github.com/GoogleCloudPlatform/vertex-ai-samples.git
synced 2026-09-26 14:42:04 +00:00
Create Eval Quality Flywheel Skill for preview (#4505)
This commit is contained in:
Jason Dai
@@ -0,0 +1,94 @@
|
||||
suite_name: quality_flywheel
|
||||
timeout_seconds: 300
|
||||
max_tool_calls: 30
|
||||
|
||||
cases:
|
||||
- name: cold_start_dataset_creation
|
||||
prompt: >
|
||||
I'm building a RAG agent that answers questions about internal
|
||||
docs but I have no evaluation data yet. Help me get started
|
||||
with evaluations.
|
||||
skills: [quality-flywheel]
|
||||
difficulty: medium
|
||||
outcome_assertions:
|
||||
- "The agent identifies this as a cold-start scenario and recommends synthetic data generation"
|
||||
- "The agent explains the evaluation dataset format (EvalCase, EvaluationDataset)"
|
||||
- "The agent suggests relevant metrics for RAG evaluation (hallucination, grounding, or general quality)"
|
||||
- "The agent asks for or attempts to discover the GCP Project ID and Location"
|
||||
expect_keywords_any:
|
||||
- "synthetic"
|
||||
- "generate"
|
||||
- "dataset"
|
||||
- "EvalCase"
|
||||
|
||||
- name: metric_selection_for_tool_agent
|
||||
prompt: >
|
||||
My agent uses 5 different tools (search, calculator, calendar,
|
||||
email, file manager). It's working okay but I need to set up
|
||||
evals to measure quality. What metrics should I track?
|
||||
skills: [quality-flywheel]
|
||||
difficulty: easy
|
||||
outcome_assertions:
|
||||
- "The agent recommends tool-call-specific metrics like tool_use_quality or tool_call_valid"
|
||||
- "The agent recommends task completion metrics like multi_turn_task_success"
|
||||
- "The agent explains how to assess correct tool selection vs wrong tool usage"
|
||||
- "The agent does not recommend only generic text quality metrics"
|
||||
expect_keywords_any:
|
||||
- "tool_use"
|
||||
- "tool_call"
|
||||
- "task_success"
|
||||
- "trajectory"
|
||||
|
||||
- name: trace_ingestion_adk
|
||||
prompt: >
|
||||
I have ADK session logs from my local development in the default
|
||||
ADK format. How do I turn these into an evaluation dataset I can
|
||||
run through Vertex Eval?
|
||||
skills: [quality-flywheel]
|
||||
difficulty: medium
|
||||
outcome_assertions:
|
||||
- "The agent describes how to convert traces into AgentData format"
|
||||
- "The agent references the canonical types: AgentData, ConversationTurn, AgentEvent"
|
||||
- "The agent provides concrete code or a step-by-step conversion plan"
|
||||
expect_keywords_any:
|
||||
- "AgentData"
|
||||
- "ConversationTurn"
|
||||
- "AgentEvent"
|
||||
- "EvalCase"
|
||||
|
||||
- name: analyze_poor_results
|
||||
prompt: >
|
||||
I ran evals on my RAG agent and got 0.45 on the hallucination
|
||||
metric and 0.62 on grounding. The agent seems to make up answers
|
||||
instead of using the retrieved context. What should I fix?
|
||||
skills: [quality-flywheel]
|
||||
difficulty: easy
|
||||
outcome_assertions:
|
||||
- "The agent identifies low hallucination score as a critical problem"
|
||||
- "The agent connects the symptom (making up answers) to a grounding/prompting root cause"
|
||||
- "The agent suggests specific prompt or system instruction changes to improve grounding"
|
||||
- "The agent recommends re-running evals after applying the fix"
|
||||
expect_keywords_any:
|
||||
- "grounding"
|
||||
- "prompt"
|
||||
- "hallucin"
|
||||
- "context"
|
||||
|
||||
- name: custom_llm_metric_creation
|
||||
prompt: >
|
||||
I need to evaluate whether my customer support agent is always
|
||||
polite and empathetic, even when delivering bad news. The
|
||||
predefined metrics don't cover this. Can you help me create a
|
||||
custom metric for this?
|
||||
skills: [quality-flywheel]
|
||||
difficulty: medium
|
||||
outcome_assertions:
|
||||
- "The agent creates a custom LLMMetric with a prompt_template for politeness/empathy"
|
||||
- "The metric template evaluates tone specifically, not just general quality"
|
||||
- "The agent provides runnable Python code for the custom metric"
|
||||
- "The agent explains how to integrate it with client.evals.evaluate()"
|
||||
expect_keywords_any:
|
||||
- "LLMMetric"
|
||||
- "prompt_template"
|
||||
- "polite"
|
||||
- "empathy"
|
||||
@@ -0,0 +1,192 @@
|
||||
---
|
||||
name: quality-flywheel
|
||||
description: >-
|
||||
Evaluate and improve GenAI models and agents using the Google GenAI
|
||||
Evaluation SDK. Creates eval datasets (from session traces or synthetic
|
||||
generation), selects and configures metrics (RubricMetric, LLMMetric,
|
||||
CodeExecutionMetric), executes evals via client.evals.evaluate(), and
|
||||
analyzes results to suggest concrete fixes. Supports both single-turn
|
||||
model evaluation and multi-turn agent trajectory evaluation. Use when
|
||||
asked to "evaluate my agent", "evaluate my model", "create eval dataset",
|
||||
"run evals", "analyze eval results", "which metrics should I use",
|
||||
"generate test data", or "improve quality".
|
||||
---
|
||||
|
||||
# Quality Flywheel Skill
|
||||
|
||||
You are the **Quality Flywheel** — an expert in GenAI evaluation. Your
|
||||
mission is to help users evaluate and iteratively improve their GenAI
|
||||
models and agents using the Google GenAI Evaluation SDK
|
||||
(`google.genai` / `vertexai`).
|
||||
|
||||
## When to use this skill
|
||||
|
||||
- Evaluating GenAI agents or models using `client.evals.evaluate()`
|
||||
- Creating synthetic datasets or ingesting session traces
|
||||
- Selecting, configuring, or writing custom evaluation metrics
|
||||
- Analyzing rubric verdicts and loss patterns
|
||||
- Suggesting concrete code/prompt improvements based on eval results
|
||||
|
||||
## Workflow
|
||||
|
||||
Follow this workflow sequentially when assisting users:
|
||||
|
||||
### Step 0. Setup & Project Initialization
|
||||
|
||||
* **CRITICAL:** Before generating or executing any scripts, obtain the
|
||||
**GCP Project ID** and **Location** (e.g., `global`, `us-central1`).
|
||||
Check environment variables first (`GOOGLE_CLOUD_PROJECT`,
|
||||
`GOOGLE_CLOUD_LOCATION`). If not found, ask the user.
|
||||
* Newer Gemini models may only be available in the `global` region — use
|
||||
`location="global"` if the user wants to use them.
|
||||
|
||||
### Step 1. Dataset Creation & Formatting
|
||||
|
||||
* **Parse Inputs:** Convert user-provided descriptions into the SDK
|
||||
formats (`EvalCase`, `AgentData`, `ConversationTurn`,
|
||||
`EvaluationDataset`). See
|
||||
[references/dataset_schema.md](references/dataset_schema.md) for the
|
||||
full type hierarchy and examples.
|
||||
|
||||
* **Single-Turn (Model Eval):** Create `EvalCase` objects with `prompt`
|
||||
strings. Use `client.evals.run_inference(model=..., src=dataset)` to
|
||||
populate model responses if needed.
|
||||
|
||||
* **Multi-Turn (Agent Eval):** If the user wants to test a multi-turn
|
||||
agent but lacks data:
|
||||
1. **Generate Scenarios:** Use `client.evals.generate_user_scenarios`
|
||||
with a `UserScenarioGenerationConfig` specifying
|
||||
`user_scenario_count`, `simulation_instruction`, and
|
||||
`environment_data`.
|
||||
2. **Run Inference:** Use `client.evals.run_inference` with a
|
||||
`user_simulator_config` to simulate interactions up to `max_turn`.
|
||||
|
||||
### Step 2. Metric Selection & Customization
|
||||
|
||||
Use the quick-reference table to pick metrics. For the full catalog, see
|
||||
[references/metric_registry.md](references/metric_registry.md).
|
||||
|
||||
| Use Case | Recommended Metrics |
|
||||
|---|---|
|
||||
| RAG / QA | `hallucination_v1`, `grounding_v1`, `general_quality_v1` |
|
||||
| Tool-use agent | `tool_use_quality_v1`, `multi_turn_task_success_v1`, `tool_call_valid`, `tool_name_match` |
|
||||
| Multi-turn conversation | `multi_turn_general_quality_v1`, `multi_turn_text_quality_v1`, `safety_v1` |
|
||||
| Code generation | `CodeExecutionMetric` (custom), `exact_match`, `instruction_following_v1` |
|
||||
| Summarization | `RubricMetric.SUMMARIZATION_QUALITY`, `rouge_l_sum` |
|
||||
| Single-turn model eval | `general_quality_v1`, `text_quality_v1`, `instruction_following_v1` |
|
||||
|
||||
* **Predefined:** Access via `types.RubricMetric.<NAME>`. Server-side
|
||||
AutoRater — no judge model needed.
|
||||
* **Custom LLM-as-a-judge:** `types.LLMMetric` with `prompt_template` or
|
||||
`types.MetricPromptBuilder` for structured rubrics.
|
||||
* **Custom Code:** `types.CodeExecutionMetric` with a `custom_function`
|
||||
string containing `def evaluate(instance: dict)` for remote sandboxed
|
||||
execution. Or `types.Metric` with `custom_function=<callable>` for
|
||||
local execution.
|
||||
|
||||
### Step 3. Automated Execution
|
||||
|
||||
* Generate a complete Python evaluation script using
|
||||
`client.evals.evaluate(dataset=..., metrics=...)`.
|
||||
* Save the script to a file and execute it to get real results.
|
||||
* Ensure the script prints results in a parseable format (JSON).
|
||||
|
||||
### Step 4. Result Analysis & Auto-Optimization
|
||||
|
||||
* Read the stdout/stderr from the evaluation run.
|
||||
|
||||
* **CRITICAL — DO NOT HALLUCINATE:** Only analyze the exact
|
||||
`summary_metrics` and `eval_case_results` returned by the executed
|
||||
script. Never fabricate scores or results.
|
||||
|
||||
* Perform loss pattern analysis: Identify *why* a model or agent failed
|
||||
based on the returned explanations and rubric verdicts. See
|
||||
[references/failure_patterns.md](references/failure_patterns.md) for
|
||||
common failure modes and their fixes.
|
||||
|
||||
* Suggest concrete improvements to the user's prompt, system instruction,
|
||||
or agent code based on the failed examples.
|
||||
|
||||
### Step 5. Iterate (The Flywheel)
|
||||
|
||||
After applying fixes, re-run evaluation (Step 3) and compare results.
|
||||
Repeat until quality targets are met. Track progress across iterations:
|
||||
|
||||
| Iteration | Metric A | Metric B | Change Made |
|
||||
|---|---|---|---|
|
||||
| Baseline | 0.62 | 0.55 | — |
|
||||
| v2 | 0.78 | 0.68 | Added grounding prompt |
|
||||
| v3 | 0.81 | 0.72 | Fixed tool selection |
|
||||
|
||||
### Rules of Engagement
|
||||
|
||||
1. **Always Plan First:** Before writing a script, output a `<plan>`
|
||||
block detailing the steps you are about to take.
|
||||
2. **Step-by-Step Execution:** Write the script, execute it, wait for
|
||||
output, then analyze. Don't do everything in one response.
|
||||
3. **Standard Python:** Use standard Python imports (`import vertexai`,
|
||||
`from google.genai import types`). Don't use internal import paths.
|
||||
4. **Verify Before Guessing:** When unsure about SDK types or metrics,
|
||||
check the SDK source code rather than guessing or hallucinating.
|
||||
|
||||
### Error Handling
|
||||
|
||||
If execution returns a traceback:
|
||||
1. Analyze the error immediately.
|
||||
2. Fix the script.
|
||||
3. Run again.
|
||||
4. Keep iterating until success or user input is needed.
|
||||
|
||||
### SDK Quick Reference
|
||||
|
||||
```python
|
||||
import vertexai
|
||||
from vertexai import Client, types
|
||||
from google.genai import types as genai_types
|
||||
|
||||
# Initialize client
|
||||
client = vertexai.Client(project="PROJECT_ID", location="LOCATION")
|
||||
|
||||
# --- SINGLE-TURN EVAL ---
|
||||
dataset = types.EvaluationDataset(eval_cases=[
|
||||
types.EvalCase(prompt="Query here", response="Model response here"),
|
||||
])
|
||||
|
||||
# --- MULTI-TURN AGENT EVAL ---
|
||||
agent_data = types.evals.AgentData(
|
||||
agents={"my_agent": types.evals.AgentConfig(
|
||||
agent_id="my_agent", instruction="You are helpful.")},
|
||||
turns=[types.evals.ConversationTurn(turn_index=0, events=[
|
||||
types.evals.AgentEvent(author="user",
|
||||
content=genai_types.Content(role="user",
|
||||
parts=[genai_types.Part(text="Hello")])),
|
||||
types.evals.AgentEvent(author="my_agent",
|
||||
content=genai_types.Content(role="model",
|
||||
parts=[genai_types.Part(text="Hi! How can I help?")])),
|
||||
])],
|
||||
)
|
||||
dataset = types.EvaluationDataset(
|
||||
eval_cases=[types.EvalCase(agent_data=agent_data)])
|
||||
|
||||
# --- METRICS ---
|
||||
predefined = types.RubricMetric.MULTI_TURN_TRAJECTORY_QUALITY
|
||||
custom_llm = types.LLMMetric(name="tone",
|
||||
prompt_template="Is this polite? Response: {response}")
|
||||
custom_code = types.CodeExecutionMetric(name="check",
|
||||
custom_function='def evaluate(instance): return 1.0')
|
||||
|
||||
# --- EVALUATE ---
|
||||
result = client.evals.evaluate(dataset=dataset, metrics=[predefined])
|
||||
|
||||
# --- RESULTS ---
|
||||
for s in result.summary_metrics:
|
||||
print(f"{s.metric_name}: mean={s.mean_score}, pass_rate={s.pass_rate}")
|
||||
for case in result.eval_case_results:
|
||||
for cand in case.response_candidate_results:
|
||||
for name, r in cand.metric_results.items():
|
||||
print(f" {name}: score={r.score}, explanation={r.explanation}")
|
||||
```
|
||||
|
||||
See [references/sdk_patterns.md](references/sdk_patterns.md) for
|
||||
advanced patterns: synthetic data generation, pairwise comparison,
|
||||
MetricPromptBuilder, multi-agent evaluation.
|
||||
@@ -0,0 +1,83 @@
|
||||
# Quality Flywheel Skill - Manual Test Plan
|
||||
|
||||
## Prerequisites
|
||||
|
||||
- Read `SKILL.md` to understand the workflow
|
||||
- Have a GCP project with the GenAI Evaluation API enabled
|
||||
- Access to an environment with the skill installed
|
||||
|
||||
---
|
||||
|
||||
## Test 1: Cold Start — No Data
|
||||
|
||||
**Prompt:** "I want to evaluate my customer support agent but I don't
|
||||
have any test data. Can you help me set up evals from scratch?"
|
||||
|
||||
**Verify:**
|
||||
- Agent asks for GCP Project ID and Location (or attempts env vars)
|
||||
- Agent recommends synthetic data generation via `generate_user_scenarios`
|
||||
- Agent explains the EvalCase / EvaluationDataset format
|
||||
- Agent suggests appropriate metrics for customer support
|
||||
|
||||
---
|
||||
|
||||
## Test 2: Metric Selection
|
||||
|
||||
**Prompt:** "My agent is a multi-turn travel booking assistant that
|
||||
uses tools like search_flights and book_hotel. Which metrics should
|
||||
I use to evaluate it?"
|
||||
|
||||
**Verify:**
|
||||
- Agent recommends multi-turn agent metrics (trajectory quality, task success)
|
||||
- Agent recommends tool-specific computation metrics (tool_call_valid, tool_name_match)
|
||||
- Agent uses `types.RubricMetric.<NAME>` syntax
|
||||
- Agent does NOT recommend only single-turn text quality metrics
|
||||
|
||||
---
|
||||
|
||||
## Test 3: Custom Metric
|
||||
|
||||
**Prompt:** "I need a metric that checks if my agent always includes
|
||||
a disclaimer when giving financial advice. Can you write one?"
|
||||
|
||||
**Verify:**
|
||||
- Agent creates a `CodeExecutionMetric` or `LLMMetric`
|
||||
- The metric logic specifically checks for disclaimer presence
|
||||
- Code is syntactically valid Python
|
||||
- Agent shows how to use it with `client.evals.evaluate()`
|
||||
|
||||
---
|
||||
|
||||
## Test 4: End-to-End Execution
|
||||
|
||||
**Prompt:** "Here's my agent — it's a simple Q&A bot. I have 3 test
|
||||
questions. Can you write an eval script and run it?"
|
||||
|
||||
Provide sample questions when asked.
|
||||
|
||||
**Verify:**
|
||||
- Agent writes a complete Python script with proper imports
|
||||
- Script uses `vertexai.Client()` initialization
|
||||
- Agent saves the script to a file and executes it
|
||||
- Agent reads and analyzes the actual output (no hallucinated results)
|
||||
- Agent uses standard Python imports (not internal paths)
|
||||
|
||||
---
|
||||
|
||||
## Test 5: Result Analysis
|
||||
|
||||
**Prompt:** "My eval results show 0.3 on tool_use_quality and 0.8
|
||||
on general_quality. The agent keeps calling the wrong tool when
|
||||
users ask about order status. What should I fix?"
|
||||
|
||||
**Verify:**
|
||||
- Agent identifies tool_use_quality as the critical failure
|
||||
- Agent suggests specific tool description or system prompt improvements
|
||||
- Agent recommends checking `tool_name_match` for granular diagnosis
|
||||
- Agent references failure patterns (from references/failure_patterns.md)
|
||||
|
||||
---
|
||||
|
||||
## Cleanup
|
||||
|
||||
No persistent state created. Scripts written to /tmp/ can be deleted.
|
||||
@@ -0,0 +1,273 @@
|
||||
# Evaluation Dataset Schema
|
||||
|
||||
Canonical formats for evaluation datasets in the Google GenAI Evaluation SDK.
|
||||
Source of truth: `vertexai/_genai/types/evals.py` and
|
||||
`vertexai/_genai/types/common.py`.
|
||||
|
||||
## Core Types
|
||||
|
||||
```
|
||||
EvaluationDataset
|
||||
├── eval_cases: list[EvalCase] # Primary: list of cases
|
||||
└── eval_dataset_df: pd.DataFrame # Alternative: pandas DataFrame
|
||||
|
||||
EvalCase
|
||||
├── prompt: str # Single-turn: the user query
|
||||
├── response: str # Single-turn: the model response
|
||||
├── reference: str # Ground truth (for reference-based metrics)
|
||||
├── agent_data: AgentData # Multi-turn: full conversation trajectory
|
||||
└── (extra fields allowed) # Custom fields for custom metrics
|
||||
|
||||
AgentData
|
||||
├── agents: dict[str, AgentConfig] # Agent definitions
|
||||
└── turns: list[ConversationTurn] # Ordered conversation turns
|
||||
|
||||
ConversationTurn
|
||||
├── turn_index: int # 0-based turn number
|
||||
└── events: list[AgentEvent] # Events within this turn
|
||||
|
||||
AgentEvent
|
||||
├── author: str # "user", agent_id, or "tool"
|
||||
└── content: genai_types.Content # Content with role and parts
|
||||
```
|
||||
|
||||
## Single-Turn Dataset
|
||||
|
||||
For simple prompt-response evaluation (e.g., QA, summarization).
|
||||
|
||||
```python
|
||||
from vertexai import types
|
||||
|
||||
dataset = types.EvaluationDataset(eval_cases=[
|
||||
types.EvalCase(
|
||||
prompt="What is the capital of France?",
|
||||
response="The capital of France is Paris.",
|
||||
reference="Paris",
|
||||
),
|
||||
types.EvalCase(
|
||||
prompt="Summarize this article: ...",
|
||||
response="The article discusses...",
|
||||
),
|
||||
])
|
||||
```
|
||||
|
||||
### From pandas DataFrame
|
||||
|
||||
```python
|
||||
import pandas as pd
|
||||
from vertexai import types
|
||||
|
||||
df = pd.DataFrame({
|
||||
"prompt": ["What is 2+2?", "Name the planets"],
|
||||
"response": ["4", "Mercury, Venus, Earth, ..."],
|
||||
"reference": ["4", "Mercury, Venus, Earth, Mars, ..."],
|
||||
})
|
||||
dataset = types.EvaluationDataset(eval_dataset_df=df)
|
||||
```
|
||||
|
||||
### Required fields by metric type
|
||||
|
||||
| Metric category | Required fields |
|
||||
|---|---|
|
||||
| Predefined (single-turn) | `prompt`, `response` |
|
||||
| Computation-based | `response`, `reference` |
|
||||
| Translation | `prompt` (source), `response`, `reference` |
|
||||
| Custom LLM/code | Fields referenced in your template/function |
|
||||
|
||||
## Multi-Turn Dataset (AgentData)
|
||||
|
||||
For evaluating multi-turn agent conversations with tool calls.
|
||||
|
||||
```python
|
||||
from vertexai import types
|
||||
from google.genai import types as genai_types
|
||||
|
||||
agent_data = types.evals.AgentData(
|
||||
agents={
|
||||
"support_agent": types.evals.AgentConfig(
|
||||
agent_id="support_agent",
|
||||
instruction="You are a helpful support agent.",
|
||||
tools=[genai_types.Tool(function_declarations=[
|
||||
genai_types.FunctionDeclaration(
|
||||
name="lookup_order",
|
||||
description="Look up order status by ID",
|
||||
parameters=genai_types.Schema(
|
||||
type="OBJECT",
|
||||
properties={"order_id": genai_types.Schema(type="STRING")},
|
||||
),
|
||||
)
|
||||
])],
|
||||
)
|
||||
},
|
||||
turns=[
|
||||
types.evals.ConversationTurn(
|
||||
turn_index=0,
|
||||
events=[
|
||||
# User message
|
||||
types.evals.AgentEvent(
|
||||
author="user",
|
||||
content=genai_types.Content(
|
||||
role="user",
|
||||
parts=[genai_types.Part(text="Where is my order #12345?")]
|
||||
),
|
||||
),
|
||||
# Agent calls tool
|
||||
types.evals.AgentEvent(
|
||||
author="support_agent",
|
||||
content=genai_types.Content(
|
||||
role="model",
|
||||
parts=[genai_types.Part(
|
||||
function_call=genai_types.FunctionCall(
|
||||
name="lookup_order",
|
||||
args={"order_id": "12345"},
|
||||
)
|
||||
)]
|
||||
),
|
||||
),
|
||||
# Tool response
|
||||
types.evals.AgentEvent(
|
||||
author="support_agent",
|
||||
content=genai_types.Content(
|
||||
role="tool",
|
||||
parts=[genai_types.Part(
|
||||
function_response=genai_types.FunctionResponse(
|
||||
name="lookup_order",
|
||||
response={"status": "shipped", "eta": "tomorrow"},
|
||||
)
|
||||
)]
|
||||
),
|
||||
),
|
||||
# Agent final response
|
||||
types.evals.AgentEvent(
|
||||
author="support_agent",
|
||||
content=genai_types.Content(
|
||||
role="model",
|
||||
parts=[genai_types.Part(
|
||||
text="Your order #12345 has been shipped and should arrive tomorrow!"
|
||||
)]
|
||||
),
|
||||
),
|
||||
],
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
eval_case = types.EvalCase(agent_data=agent_data)
|
||||
dataset = types.EvaluationDataset(eval_cases=[eval_case])
|
||||
```
|
||||
|
||||
## Multi-Agent Dataset
|
||||
|
||||
For evaluating systems with multiple collaborating agents.
|
||||
|
||||
```python
|
||||
agent_data = types.evals.AgentData(
|
||||
agents={
|
||||
"router": types.evals.AgentConfig(
|
||||
agent_id="router",
|
||||
agent_type="RouterAgent",
|
||||
instruction="Route requests to the appropriate specialist.",
|
||||
),
|
||||
"flight_bot": types.evals.AgentConfig(
|
||||
agent_id="flight_bot",
|
||||
agent_type="SpecialistAgent",
|
||||
instruction="Search and book flights.",
|
||||
tools=[genai_types.Tool(function_declarations=[
|
||||
genai_types.FunctionDeclaration(name="search_flights")
|
||||
])],
|
||||
),
|
||||
},
|
||||
turns=[
|
||||
types.evals.ConversationTurn(
|
||||
turn_index=0,
|
||||
events=[
|
||||
types.evals.AgentEvent(
|
||||
author="user",
|
||||
content=genai_types.Content(
|
||||
role="user",
|
||||
parts=[genai_types.Part(text="Book a flight to NYC")]
|
||||
),
|
||||
),
|
||||
# Router delegates
|
||||
types.evals.AgentEvent(
|
||||
author="router",
|
||||
content=genai_types.Content(
|
||||
role="model",
|
||||
parts=[genai_types.Part(
|
||||
function_call=genai_types.FunctionCall(
|
||||
name="delegate_to_agent",
|
||||
args={"agent_name": "flight_bot"},
|
||||
)
|
||||
)]
|
||||
),
|
||||
),
|
||||
],
|
||||
),
|
||||
types.evals.ConversationTurn(
|
||||
turn_index=1,
|
||||
events=[
|
||||
# Specialist works
|
||||
types.evals.AgentEvent(
|
||||
author="flight_bot",
|
||||
content=genai_types.Content(
|
||||
role="model",
|
||||
parts=[genai_types.Part(
|
||||
function_call=genai_types.FunctionCall(
|
||||
name="search_flights",
|
||||
args={"destination": "NYC"},
|
||||
)
|
||||
)]
|
||||
),
|
||||
),
|
||||
],
|
||||
),
|
||||
],
|
||||
)
|
||||
```
|
||||
|
||||
## Synthetic Data Generation
|
||||
|
||||
### Generate User Scenarios (Cold Start)
|
||||
|
||||
```python
|
||||
scenarios = client.evals.generate_user_scenarios(
|
||||
agents={
|
||||
"my_agent": types.evals.AgentConfig(
|
||||
agent_id="my_agent",
|
||||
instruction="You are a helpful customer support agent.",
|
||||
)
|
||||
},
|
||||
root_agent_id="my_agent",
|
||||
user_scenario_generation_config=types.evals.UserScenarioGenerationConfig(
|
||||
user_scenario_count=10,
|
||||
simulation_instruction="Simulate a customer asking about order status.",
|
||||
environment_data="Orders can be: pending, shipped, delivered, cancelled.",
|
||||
model_name="gemini-2.5-flash",
|
||||
),
|
||||
)
|
||||
```
|
||||
|
||||
### Run Inference (Populate Responses)
|
||||
|
||||
```python
|
||||
dataset_with_responses = client.evals.run_inference(
|
||||
agent=my_agent_callable,
|
||||
src=scenarios,
|
||||
config={
|
||||
"user_simulator_config": {
|
||||
"model_name": "gemini-2.5-flash",
|
||||
"max_turn": 5,
|
||||
}
|
||||
},
|
||||
)
|
||||
```
|
||||
|
||||
## Common Mistakes
|
||||
|
||||
| Mistake | Fix |
|
||||
|---|---|
|
||||
| Using `role="assistant"` | Use `role="model"` (Vertex convention) |
|
||||
| Missing `turn_index` | Always set sequential 0-based indices |
|
||||
| Tool response without `function_response` | Wrap in `genai_types.FunctionResponse` |
|
||||
| Using `response` field for multi-turn | Use `agent_data` with full trajectory |
|
||||
| Mixing `prompt` and `agent_data` | Use one or the other per EvalCase |
|
||||
@@ -0,0 +1,153 @@
|
||||
# Evaluation Failure Patterns & Fixes
|
||||
|
||||
Common failure modes observed in GenAI agent evaluations, mapped to their
|
||||
root causes and concrete fixes.
|
||||
|
||||
## Metric-Specific Failures
|
||||
|
||||
### Low `hallucination_v1` or `grounding_v1` Score
|
||||
|
||||
**Symptom:** Agent generates plausible-sounding but factually incorrect
|
||||
information, or doesn't use the provided context.
|
||||
|
||||
**Root causes:**
|
||||
- System prompt lacks explicit grounding instructions
|
||||
- Retrieved context not passed into the prompt
|
||||
- Agent ignores context in favor of parametric knowledge
|
||||
|
||||
**Fixes:**
|
||||
1. Add to system prompt: "Base ALL answers strictly on the provided context.
|
||||
If the context doesn't contain the answer, say 'I don't have that
|
||||
information.'"
|
||||
2. Verify context is actually injected into the prompt (check tool responses)
|
||||
3. Add `temperature=0` or lower temperature to reduce creative generation
|
||||
|
||||
### Low `general_quality_v1` or `text_quality_v1`
|
||||
|
||||
**Symptom:** Agent responses are poorly structured, unclear, or unhelpful.
|
||||
|
||||
**Root causes:**
|
||||
- System prompt too vague
|
||||
- Agent over-explains or under-explains
|
||||
- Missing output format instructions
|
||||
|
||||
**Fixes:**
|
||||
1. Add explicit format instructions: "Respond concisely in 2-3 sentences."
|
||||
2. Add few-shot examples in the system prompt
|
||||
3. Review rubric verdicts for specific quality dimensions that scored low
|
||||
|
||||
### Low `tool_use_quality_v1` or `tool_call_valid`
|
||||
|
||||
**Symptom:** Agent calls the wrong tool, uses wrong parameters, or doesn't
|
||||
call tools when it should.
|
||||
|
||||
**Root causes:**
|
||||
- Tool descriptions are ambiguous
|
||||
- Multiple tools have overlapping functionality
|
||||
- Function declaration parameter schemas are incomplete
|
||||
|
||||
**Fixes:**
|
||||
1. Make tool `description` fields precise and mutually exclusive
|
||||
2. Add parameter descriptions and constraints to `FunctionDeclaration`
|
||||
3. Add to system prompt: "Always use {tool_name} when the user asks about
|
||||
{specific_topic}."
|
||||
4. Check `tool_name_match` and `tool_parameter_kv_match` for granular diagnosis
|
||||
|
||||
### Low `multi_turn_trajectory_quality_v1`
|
||||
|
||||
**Symptom:** Agent takes suboptimal paths through a conversation — unnecessary
|
||||
tool calls, redundant questions, or wrong delegation order.
|
||||
|
||||
**Root causes:**
|
||||
- Router agent lacks clear delegation rules
|
||||
- Agent retries failed operations without adaptation
|
||||
- Missing escalation logic
|
||||
|
||||
**Fixes:**
|
||||
1. Add explicit routing rules: "Route to {agent} when {condition}."
|
||||
2. Add retry limits: "If {tool} fails twice, inform the user and suggest
|
||||
alternatives."
|
||||
3. Review the trajectory events in `agent_data` to identify the specific
|
||||
turn where the agent deviated
|
||||
|
||||
### Low `multi_turn_task_success_v1`
|
||||
|
||||
**Symptom:** Agent engages in conversation but doesn't complete the user's
|
||||
actual goal.
|
||||
|
||||
**Root causes:**
|
||||
- Agent gets sidetracked by follow-up questions
|
||||
- Missing confirmation/completion step
|
||||
- Agent doesn't track task state across turns
|
||||
|
||||
**Fixes:**
|
||||
1. Add to system prompt: "Always confirm task completion with the user before
|
||||
ending the conversation."
|
||||
2. Implement explicit task tracking in agent logic
|
||||
3. Verify `max_turn` in user simulator is sufficient for the task complexity
|
||||
|
||||
### Low `safety_v1`
|
||||
|
||||
**Symptom:** Agent generates unsafe content or complies with harmful requests.
|
||||
|
||||
**Root causes:**
|
||||
- System prompt lacks safety constraints
|
||||
- Agent follows user instructions too literally
|
||||
- Missing refusal logic for out-of-scope requests
|
||||
|
||||
**Fixes:**
|
||||
1. Add safety guardrails: "Never provide medical/legal/financial advice.
|
||||
Redirect to appropriate professionals."
|
||||
2. Add refusal patterns: "If the user asks for {harmful_category}, politely
|
||||
decline."
|
||||
3. Use `safety_v1` alongside domain-specific `LLMMetric` safety checks
|
||||
|
||||
## Structural Failures
|
||||
|
||||
### `is_infra_error: true`
|
||||
|
||||
**Symptom:** Eval case fails with infrastructure error, not a quality issue.
|
||||
|
||||
**Root causes:**
|
||||
- API quota exceeded
|
||||
- Network timeout
|
||||
- Model endpoint temporarily unavailable
|
||||
|
||||
**Fix:** Re-run the evaluation. If persistent, check quota and endpoint health.
|
||||
|
||||
### Timeout
|
||||
|
||||
**Symptom:** Evaluation times out before completing.
|
||||
|
||||
**Root causes:**
|
||||
- Dataset too large for a single API call
|
||||
- Complex custom metric code takes too long
|
||||
- Judge model sampling count too high
|
||||
|
||||
**Fixes:**
|
||||
1. Reduce dataset size or batch into smaller chunks
|
||||
2. Optimize custom metric code (avoid network calls in `evaluate()`)
|
||||
3. Reduce `judge_model_sampling_count` (default 1, max 32)
|
||||
|
||||
### `KeyError` in Custom Metric
|
||||
|
||||
**Symptom:** Custom function crashes with missing field.
|
||||
|
||||
**Root cause:** Metric function expects a field not present in the eval case.
|
||||
|
||||
**Fix:** Check available fields in the `instance` dict. Common fields:
|
||||
`prompt`, `response`, `reference`, `agent_data`. Always use `.get()` with
|
||||
defaults.
|
||||
|
||||
## Analysis Workflow
|
||||
|
||||
When eval results show failures:
|
||||
|
||||
1. **Start with `summary_metrics`** — identify which metrics scored lowest
|
||||
2. **Drill into `eval_case_results`** — find specific failing cases
|
||||
3. **Read `rubric_verdicts`** — understand why the judge scored low
|
||||
4. **Cross-reference with `agent_data`** — find the exact turn/event
|
||||
that caused the failure
|
||||
5. **Identify the pattern** — is it a prompt issue, tool issue, or data issue?
|
||||
6. **Apply the targeted fix** — from the table above
|
||||
7. **Re-run and compare** — verify the fix improved the target metric
|
||||
@@ -0,0 +1,264 @@
|
||||
# GenAI Evaluation Metric Registry
|
||||
|
||||
Complete catalog of evaluation metrics available in the `vertexai._genai` SDK.
|
||||
Source of truth: `vertexai/_genai/_evals_metric_loaders.py` and
|
||||
`vertexai/_genai/_evals_metric_handlers.py`.
|
||||
|
||||
## Metric Type Hierarchy
|
||||
|
||||
```
|
||||
Metric (base)
|
||||
├── LLMMetric — LLM-as-a-judge with prompt_template
|
||||
├── CodeExecutionMetric — Sandboxed remote Python function
|
||||
└── (base Metric) — Local callable or predefined name
|
||||
```
|
||||
|
||||
Access predefined metrics via `types.RubricMetric.<NAME>` (preferred).
|
||||
`types.PrebuiltMetric` is an alias with identical behavior.
|
||||
|
||||
## Predefined API Metrics (AutoRater)
|
||||
|
||||
Server-side evaluation via Vertex AI AutoRater. No judge model needed.
|
||||
|
||||
### Single-Turn
|
||||
|
||||
| Metric name | What it measures | Required fields |
|
||||
|---|---|---|
|
||||
| `general_quality_v1` | Overall response quality | `prompt`, `response` |
|
||||
| `text_quality_v1` | Text quality (grammar, clarity) | `prompt`, `response` |
|
||||
| `instruction_following_v1` | How well response follows instructions | `prompt`, `response` |
|
||||
| `grounding_v1` | Factual grounding in provided context | `prompt`, `response`, context |
|
||||
| `safety_v1` | Safety assessment | `prompt`, `response` |
|
||||
| `hallucination_v1` | Hallucination detection | `prompt`, `response` |
|
||||
| `tool_use_quality_v1` | Quality of tool/function calling | `agent_data` with tool calls |
|
||||
|
||||
### Multi-Turn / Agent
|
||||
|
||||
| Metric name | What it measures | Required fields |
|
||||
|---|---|---|
|
||||
| `multi_turn_general_quality_v1` | Overall multi-turn quality | `agent_data` (1+ turns) |
|
||||
| `multi_turn_text_quality_v1` | Text quality across turns | `agent_data` (1+ turns) |
|
||||
| `multi_turn_tool_use_quality_v1` | Tool call quality across trajectory | `agent_data` with function calls |
|
||||
| `multi_turn_trajectory_quality_v1` | Quality of agent's action sequence | `agent_data` with full trajectory |
|
||||
| `multi_turn_task_success_v1` | Whether agent completed the task | `agent_data` with task context |
|
||||
|
||||
### Agent Final Response
|
||||
|
||||
| Metric name | What it measures | Required fields |
|
||||
|---|---|---|
|
||||
| `final_response_match_v2` | Reference-based final response matching | `agent_data`, `reference` |
|
||||
| `final_response_reference_free_v1` | Final response quality (no reference) | `agent_data` |
|
||||
| `final_response_quality_v1` | Final response quality | `agent_data` |
|
||||
|
||||
### Multimodal
|
||||
|
||||
| Metric name | What it measures | Required fields |
|
||||
|---|---|---|
|
||||
| `gecko_text2image_v1` | Text-to-image quality | image content |
|
||||
| `gecko_text2video_v1` | Text-to-video quality | video content |
|
||||
|
||||
### Accessing predefined metrics
|
||||
|
||||
```python
|
||||
from vertexai import types
|
||||
|
||||
# Via RubricMetric (preferred)
|
||||
metric = types.RubricMetric.MULTI_TURN_TRAJECTORY_QUALITY
|
||||
|
||||
# Via PrebuiltMetric (alias — identical behavior)
|
||||
metric = types.PrebuiltMetric.MULTI_TURN_TRAJECTORY_QUALITY
|
||||
|
||||
# With version override
|
||||
metric = types.RubricMetric.GENERAL_QUALITY(version="v2")
|
||||
```
|
||||
|
||||
## Computation-Based Metrics
|
||||
|
||||
No LLM judge. Deterministic comparison of `response` vs `reference`.
|
||||
|
||||
| Metric name | What it measures | Notes |
|
||||
|---|---|---|
|
||||
| `exact_match` | Exact string match | Case-sensitive |
|
||||
| `bleu` | BLEU score (translation/generation) | Standard BLEU |
|
||||
| `rouge_1` | ROUGE-1 (unigram overlap) | Summarization |
|
||||
| `rouge_l_sum` | ROUGE-L (longest common subsequence) | Summary-level |
|
||||
| `tool_call_valid` | Whether tool calls are syntactically valid | Agent evals |
|
||||
| `tool_name_match` | Whether tool names match reference | Agent evals |
|
||||
| `tool_parameter_key_match` | Whether tool parameter keys match | Agent evals |
|
||||
| `tool_parameter_kv_match` | Whether tool parameter key-value pairs match | Agent evals |
|
||||
|
||||
```python
|
||||
# Usage
|
||||
metric = types.Metric(name="exact_match")
|
||||
metric = types.Metric(name="tool_call_valid")
|
||||
```
|
||||
|
||||
## Translation Metrics
|
||||
|
||||
| Metric name | Default version | Notes |
|
||||
|---|---|---|
|
||||
| `comet` | `COMET_22_SRC_REF` | Requires `prompt` (source), `response`, `reference` |
|
||||
| `metricx` | `METRICX_24_SRC_REF` | Requires `prompt` (source), `response`, `reference` |
|
||||
|
||||
## RubricMetric / PrebuiltMetric (GCS-Loaded LLM Recipes)
|
||||
|
||||
These resolve first against the API predefined list, then fall back to
|
||||
GCS-hosted LLM metric YAML definitions.
|
||||
|
||||
| Property | Resolution |
|
||||
|---|---|
|
||||
| `GENERAL_QUALITY` | API predefined |
|
||||
| `TEXT_QUALITY` | API predefined |
|
||||
| `INSTRUCTION_FOLLOWING` | API predefined |
|
||||
| `SAFETY` | API predefined |
|
||||
| `HALLUCINATION` | API predefined |
|
||||
| `TOOL_USE_QUALITY` | API predefined |
|
||||
| `MULTI_TURN_GENERAL_QUALITY` | API predefined |
|
||||
| `MULTI_TURN_TEXT_QUALITY` | API predefined |
|
||||
| `MULTI_TURN_TOOL_USE_QUALITY` | API predefined |
|
||||
| `MULTI_TURN_TRAJECTORY_QUALITY` | API predefined |
|
||||
| `MULTI_TURN_TASK_SUCCESS` | API predefined |
|
||||
| `FINAL_RESPONSE_MATCH` | API predefined (v2) |
|
||||
| `FINAL_RESPONSE_REFERENCE_FREE` | API predefined |
|
||||
| `FINAL_RESPONSE_QUALITY` | API predefined |
|
||||
| `COHERENCE` | GCS-loaded LLM recipe |
|
||||
| `FLUENCY` | GCS-loaded LLM recipe |
|
||||
| `VERBOSITY` | GCS-loaded LLM recipe |
|
||||
| `SUMMARIZATION_QUALITY` | GCS-loaded LLM recipe |
|
||||
| `QUESTION_ANSWERING_QUALITY` | GCS-loaded LLM recipe |
|
||||
| `MULTI_TURN_CHAT_QUALITY` | GCS-loaded LLM recipe |
|
||||
| `MULTI_TURN_SAFETY` | GCS-loaded LLM recipe |
|
||||
|
||||
Any arbitrary name can be tried via `RubricMetric.<NAME>` — it will
|
||||
attempt resolution against the API list and then GCS.
|
||||
|
||||
## Custom Metrics
|
||||
|
||||
### Custom Local Function
|
||||
|
||||
Runs client-side. Fastest iteration, no API call.
|
||||
|
||||
```python
|
||||
def my_evaluator(instance: dict) -> float:
|
||||
response_text = instance.get("response", "")
|
||||
return 1.0 if "thank you" in response_text.lower() else 0.0
|
||||
|
||||
metric = types.Metric(
|
||||
name="politeness_check",
|
||||
custom_function=my_evaluator,
|
||||
)
|
||||
```
|
||||
|
||||
### CodeExecutionMetric (Remote Sandboxed)
|
||||
|
||||
Runs server-side in a secure sandbox. Must contain `def evaluate(instance)`.
|
||||
|
||||
```python
|
||||
metric = types.CodeExecutionMetric(
|
||||
name="link_validator",
|
||||
custom_function='''
|
||||
import re
|
||||
def evaluate(instance: dict) -> dict:
|
||||
text = instance.get("response", "")
|
||||
links = re.findall(r"https?://\\S+", text)
|
||||
valid = all(link.startswith("https://") for link in links)
|
||||
return {"score": 1.0 if valid else 0.0, "explanation": f"Found {len(links)} links"}
|
||||
''',
|
||||
)
|
||||
```
|
||||
|
||||
### LLMMetric (LLM-as-a-Judge)
|
||||
|
||||
Uses a judge model to evaluate with a custom prompt template.
|
||||
|
||||
```python
|
||||
metric = types.LLMMetric(
|
||||
name="helpfulness",
|
||||
prompt_template="""
|
||||
Evaluate whether the response is helpful for the given query.
|
||||
|
||||
Query: {prompt}
|
||||
Response: {response}
|
||||
|
||||
Score 1 if helpful, 0 if not. Explain your reasoning.
|
||||
""",
|
||||
judge_model="gemini-2.5-flash",
|
||||
judge_model_sampling_count=3,
|
||||
)
|
||||
|
||||
# Load from YAML/JSON file
|
||||
metric = types.LLMMetric.load("path/to/metric_config.yaml")
|
||||
```
|
||||
|
||||
### MetricPromptBuilder (Structured Judge Prompt)
|
||||
|
||||
Builds structured LLM judge prompts from criteria, rating scores, and
|
||||
evaluation steps. Preferred over raw `prompt_template` strings for complex
|
||||
rubrics.
|
||||
|
||||
```python
|
||||
metric = types.LLMMetric(
|
||||
name="structured_quality",
|
||||
prompt_template=types.MetricPromptBuilder(
|
||||
criteria={
|
||||
"Accuracy": "Response contains factually correct information",
|
||||
"Completeness": "Response addresses all aspects of the query",
|
||||
},
|
||||
rating_scores={
|
||||
"1": "Poor — fails on both criteria",
|
||||
"3": "Acceptable — meets one criterion",
|
||||
"5": "Excellent — meets both criteria",
|
||||
},
|
||||
),
|
||||
judge_model="gemini-2.5-flash",
|
||||
)
|
||||
```
|
||||
|
||||
### Registered Metric (Server-Side Resource)
|
||||
|
||||
For reusable metrics shared across teams.
|
||||
|
||||
```python
|
||||
# Create once
|
||||
resource = client.evals.create_evaluation_metric(metric_config)
|
||||
|
||||
# Use by resource name
|
||||
metric = types.Metric(
|
||||
name="team_quality",
|
||||
metric_resource_name="projects/.../evaluationMetrics/...",
|
||||
)
|
||||
```
|
||||
|
||||
## Metric Selection Guide
|
||||
|
||||
| Agent Type | Recommended Metrics |
|
||||
|---|---|
|
||||
| **RAG agent** | `hallucination_v1`, `grounding_v1`, `general_quality_v1` |
|
||||
| **Tool-use agent** | `tool_use_quality_v1`, `multi_turn_task_success_v1`, `tool_call_valid`, `tool_name_match` |
|
||||
| **Multi-turn conversational** | `multi_turn_general_quality_v1`, `multi_turn_text_quality_v1`, `safety_v1` |
|
||||
| **Code generation** | `CodeExecutionMetric` (custom), `exact_match`, `instruction_following_v1` |
|
||||
| **Summarization** | `RubricMetric.SUMMARIZATION_QUALITY`, `rouge_l_sum` |
|
||||
| **Translation** | `comet`, `metricx` |
|
||||
|
||||
## Pairwise Comparison
|
||||
|
||||
There is no `PairwiseMetric` class. For model comparison, provide multiple
|
||||
`EvaluationDataset` instances and use `calculate_win_rates()`:
|
||||
|
||||
```python
|
||||
result_a = client.evals.evaluate(dataset=dataset_a, metrics=[...])
|
||||
result_b = client.evals.evaluate(dataset=dataset_b, metrics=[...])
|
||||
win_rates = calculate_win_rates(result_a, result_b)
|
||||
```
|
||||
|
||||
## Handler Dispatch Order
|
||||
|
||||
When the SDK receives a metric, it checks in this order:
|
||||
|
||||
1. `CodeExecutionMetric` with `custom_function` (str) or `remote_custom_function`
|
||||
2. `Metric` with `custom_function` (local `Callable`)
|
||||
3. `Metric` with `metric_resource_name` (registered)
|
||||
4. Name in computation metrics (`exact_match`, `bleu`, etc.)
|
||||
5. Name in translation metrics (`comet`, `metricx`)
|
||||
6. Name in predefined API metrics (`general_quality_v1`, etc.)
|
||||
7. `LLMMetric` with `prompt_template` (custom LLM judge)
|
||||
@@ -0,0 +1,251 @@
|
||||
# Vertex Evaluation SDK Patterns
|
||||
|
||||
Code patterns for common evaluation scenarios using `vertexai._genai.evals`.
|
||||
|
||||
## Initialization
|
||||
|
||||
```python
|
||||
import vertexai
|
||||
from vertexai import types
|
||||
from google.genai import types as genai_types
|
||||
|
||||
client = vertexai.Client(project="{PROJECT_ID}", location="{LOCATION}")
|
||||
```
|
||||
|
||||
For Gemini 3+ models, use `location="global"`.
|
||||
|
||||
## Pattern 1: Single-Turn Evaluation
|
||||
|
||||
Simplest case — evaluate prompt/response pairs against predefined metrics.
|
||||
|
||||
```python
|
||||
dataset = types.EvaluationDataset(eval_cases=[
|
||||
types.EvalCase(
|
||||
prompt="What causes rain?",
|
||||
response="Rain is caused by water evaporating...",
|
||||
reference="Rain forms when water vapor condenses...",
|
||||
),
|
||||
])
|
||||
|
||||
result = client.evals.evaluate(
|
||||
dataset=dataset,
|
||||
metrics=[
|
||||
types.RubricMetric.GENERAL_QUALITY,
|
||||
types.Metric(name="rouge_l_sum"),
|
||||
],
|
||||
)
|
||||
```
|
||||
|
||||
## Pattern 2: Multi-Turn Agent Evaluation
|
||||
|
||||
Evaluate a full agent conversation trajectory with tool calls.
|
||||
|
||||
```python
|
||||
agent_data = types.evals.AgentData(
|
||||
agents={
|
||||
"my_agent": types.evals.AgentConfig(
|
||||
agent_id="my_agent",
|
||||
instruction="You are a helpful assistant.",
|
||||
tools=[genai_types.Tool(function_declarations=[
|
||||
genai_types.FunctionDeclaration(
|
||||
name="search",
|
||||
description="Search the web",
|
||||
parameters=genai_types.Schema(
|
||||
type="OBJECT",
|
||||
properties={"query": genai_types.Schema(type="STRING")},
|
||||
),
|
||||
),
|
||||
])],
|
||||
),
|
||||
},
|
||||
turns=[
|
||||
types.evals.ConversationTurn(turn_index=0, events=[
|
||||
types.evals.AgentEvent(
|
||||
author="user",
|
||||
content=genai_types.Content(role="user",
|
||||
parts=[genai_types.Part(text="Find me the weather in NYC")]),
|
||||
),
|
||||
types.evals.AgentEvent(
|
||||
author="my_agent",
|
||||
content=genai_types.Content(role="model",
|
||||
parts=[genai_types.Part(function_call=genai_types.FunctionCall(
|
||||
name="search", args={"query": "NYC weather"}))]),
|
||||
),
|
||||
types.evals.AgentEvent(
|
||||
author="my_agent",
|
||||
content=genai_types.Content(role="tool",
|
||||
parts=[genai_types.Part(function_response=genai_types.FunctionResponse(
|
||||
name="search", response={"result": "72F, sunny"}))]),
|
||||
),
|
||||
types.evals.AgentEvent(
|
||||
author="my_agent",
|
||||
content=genai_types.Content(role="model",
|
||||
parts=[genai_types.Part(text="It's 72F and sunny in NYC.")]),
|
||||
),
|
||||
]),
|
||||
],
|
||||
)
|
||||
|
||||
result = client.evals.evaluate(
|
||||
dataset=types.EvaluationDataset(eval_cases=[
|
||||
types.EvalCase(agent_data=agent_data),
|
||||
]),
|
||||
metrics=[
|
||||
types.RubricMetric.MULTI_TURN_TRAJECTORY_QUALITY,
|
||||
types.RubricMetric.MULTI_TURN_TASK_SUCCESS,
|
||||
],
|
||||
)
|
||||
```
|
||||
|
||||
## Pattern 3: Synthetic Data Generation (Cold Start)
|
||||
|
||||
Generate user scenarios when no eval data exists.
|
||||
|
||||
```python
|
||||
# Step 1: Generate scenarios
|
||||
scenarios = client.evals.generate_user_scenarios(
|
||||
agents={
|
||||
"agent": types.evals.AgentConfig(
|
||||
agent_id="agent",
|
||||
instruction="You are a customer support agent for an airline.",
|
||||
),
|
||||
},
|
||||
root_agent_id="agent",
|
||||
user_scenario_generation_config=types.evals.UserScenarioGenerationConfig(
|
||||
user_scenario_count=10,
|
||||
simulation_instruction="Simulate customers with flight booking issues.",
|
||||
environment_data="Flights available: NYC-LAX, NYC-SFO. Cancellation policy: free within 24h.",
|
||||
model_name="gemini-2.5-flash",
|
||||
),
|
||||
)
|
||||
|
||||
# Step 2: Run inference with user simulation
|
||||
dataset_with_responses = client.evals.run_inference(
|
||||
agent=my_agent, # Your callable agent
|
||||
src=scenarios,
|
||||
config={
|
||||
"user_simulator_config": {
|
||||
"model_name": "gemini-2.5-flash",
|
||||
"max_turn": 5,
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
# Step 3: Evaluate
|
||||
result = client.evals.evaluate(
|
||||
dataset=dataset_with_responses,
|
||||
metrics=[types.RubricMetric.MULTI_TURN_GENERAL_QUALITY, types.RubricMetric.SAFETY],
|
||||
)
|
||||
```
|
||||
|
||||
## Pattern 4: Custom LLM-as-a-Judge with MetricPromptBuilder
|
||||
|
||||
For domain-specific evaluation with structured rubrics.
|
||||
|
||||
```python
|
||||
metric = types.LLMMetric(
|
||||
name="domain_expertise",
|
||||
prompt_template=types.MetricPromptBuilder(
|
||||
metric_definition="Evaluates domain expertise in the response.",
|
||||
criteria={
|
||||
"Accuracy": "Claims are factually correct for the domain",
|
||||
"Depth": "Response shows understanding beyond surface level",
|
||||
"Actionability": "Advice is specific and actionable",
|
||||
},
|
||||
rating_scores={
|
||||
"1": "Incorrect or misleading information",
|
||||
"2": "Partially correct but superficial",
|
||||
"3": "Correct and shows reasonable understanding",
|
||||
"4": "Accurate with good depth",
|
||||
"5": "Expert-level accuracy, depth, and actionability",
|
||||
},
|
||||
),
|
||||
judge_model="gemini-2.5-flash",
|
||||
judge_model_sampling_count=3,
|
||||
)
|
||||
```
|
||||
|
||||
## Pattern 5: CodeExecutionMetric for Structured Validation
|
||||
|
||||
For programmatic checks that go beyond text comparison.
|
||||
|
||||
```python
|
||||
# Validate JSON output structure
|
||||
json_validator = types.CodeExecutionMetric(
|
||||
name="json_structure_check",
|
||||
custom_function='''
|
||||
import json
|
||||
def evaluate(instance: dict) -> dict:
|
||||
try:
|
||||
data = json.loads(instance.get("response", ""))
|
||||
required_keys = {"name", "status", "result"}
|
||||
missing = required_keys - set(data.keys())
|
||||
if missing:
|
||||
return {"score": 0.0, "explanation": f"Missing keys: {missing}"}
|
||||
return {"score": 1.0, "explanation": "All required keys present"}
|
||||
except json.JSONDecodeError as e:
|
||||
return {"score": 0.0, "explanation": f"Invalid JSON: {e}"}
|
||||
''',
|
||||
)
|
||||
```
|
||||
|
||||
## Pattern 6: Pairwise Model Comparison
|
||||
|
||||
Compare two models using `calculate_win_rates()`.
|
||||
|
||||
```python
|
||||
# Same dataset, two different model responses
|
||||
dataset_a = types.EvaluationDataset(eval_cases=[
|
||||
types.EvalCase(prompt="Explain quantum computing", response="Model A response..."),
|
||||
])
|
||||
dataset_b = types.EvaluationDataset(eval_cases=[
|
||||
types.EvalCase(prompt="Explain quantum computing", response="Model B response..."),
|
||||
])
|
||||
|
||||
result_a = client.evals.evaluate(dataset=dataset_a, metrics=[types.RubricMetric.GENERAL_QUALITY])
|
||||
result_b = client.evals.evaluate(dataset=dataset_b, metrics=[types.RubricMetric.GENERAL_QUALITY])
|
||||
|
||||
# Compare
|
||||
from vertexai._genai._evals_metric_handlers import calculate_win_rates
|
||||
win_rates = calculate_win_rates(result_a, result_b)
|
||||
```
|
||||
|
||||
## Pattern 7: Parsing Results
|
||||
|
||||
```python
|
||||
result = client.evals.evaluate(dataset=dataset, metrics=metrics)
|
||||
|
||||
# Summary level
|
||||
for summary in result.summary_metrics:
|
||||
print(f"{summary.metric_name}: mean={summary.mean_score}, pass_rate={summary.pass_rate}")
|
||||
|
||||
# Per-case level
|
||||
for case in result.eval_case_results:
|
||||
for candidate in case.response_candidate_results:
|
||||
for metric_name, metric_result in candidate.metric_results.items():
|
||||
print(f" {metric_name}: score={metric_result.score}")
|
||||
print(f" explanation: {metric_result.explanation}")
|
||||
|
||||
# Rubric verdicts (for rubric-based metrics)
|
||||
if metric_result.rubric_verdicts:
|
||||
for v in metric_result.rubric_verdicts:
|
||||
print(f" rubric {v.evaluated_rubric.rubric_id}: "
|
||||
f"{'PASS' if v.verdict else 'FAIL'} - {v.reasoning}")
|
||||
```
|
||||
|
||||
## Error Handling
|
||||
|
||||
```python
|
||||
try:
|
||||
result = client.evals.evaluate(dataset=dataset, metrics=metrics)
|
||||
except Exception as e:
|
||||
error_type = type(e).__name__
|
||||
if "PermissionDenied" in error_type:
|
||||
print("Check: GCP project permissions, API enabled, billing active")
|
||||
elif "InvalidArgument" in error_type:
|
||||
print("Check: dataset format, metric compatibility with data type")
|
||||
elif "ResourceExhausted" in error_type:
|
||||
print("Check: API quota, reduce dataset size or add delay")
|
||||
else:
|
||||
raise
|
||||
```
|
||||
@@ -0,0 +1,282 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate a runnable Vertex Evaluation SDK script from a dataset and metrics.
|
||||
|
||||
Reads an evaluation dataset JSON and generates a complete Python script
|
||||
that executes the evaluation and prints results.
|
||||
|
||||
Usage:
|
||||
python generate_eval_code.py --dataset dataset.json --metrics
|
||||
hallucination_v1,safety_v1
|
||||
python generate_eval_code.py --dataset dataset.json --metrics hallucination_v1
|
||||
--output eval_script.py
|
||||
python generate_eval_code.py --dataset dataset.json # defaults to
|
||||
general_quality_v1
|
||||
|
||||
The generated script can be run directly:
|
||||
python eval_script.py --project my-project --location us-central1
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
# Predefined metrics accessed via types.RubricMetric.<NAME>
|
||||
_RUBRIC_METRICS = frozenset({
|
||||
"general_quality_v1",
|
||||
"text_quality_v1",
|
||||
"instruction_following_v1",
|
||||
"grounding_v1",
|
||||
"safety_v1",
|
||||
"hallucination_v1",
|
||||
"tool_use_quality_v1",
|
||||
"multi_turn_general_quality_v1",
|
||||
"multi_turn_text_quality_v1",
|
||||
"multi_turn_tool_use_quality_v1",
|
||||
"multi_turn_trajectory_quality_v1",
|
||||
"multi_turn_task_success_v1",
|
||||
"final_response_match_v2",
|
||||
"final_response_reference_free_v1",
|
||||
"final_response_quality_v1",
|
||||
})
|
||||
|
||||
# Computation metrics accessed via types.Metric(name="...")
|
||||
_COMPUTATION_METRICS = frozenset({
|
||||
"exact_match",
|
||||
"bleu",
|
||||
"rouge_1",
|
||||
"rouge_l_sum",
|
||||
"tool_call_valid",
|
||||
"tool_name_match",
|
||||
"tool_parameter_key_match",
|
||||
"tool_parameter_kv_match",
|
||||
})
|
||||
|
||||
# Map from metric name to RubricMetric constant name
|
||||
_RUBRIC_CONSTANT_MAP = {
|
||||
"general_quality_v1": "GENERAL_QUALITY",
|
||||
"text_quality_v1": "TEXT_QUALITY",
|
||||
"instruction_following_v1": "INSTRUCTION_FOLLOWING",
|
||||
"grounding_v1": "GROUNDING",
|
||||
"safety_v1": "SAFETY",
|
||||
"hallucination_v1": "HALLUCINATION",
|
||||
"tool_use_quality_v1": "TOOL_USE_QUALITY",
|
||||
"multi_turn_general_quality_v1": "MULTI_TURN_GENERAL_QUALITY",
|
||||
"multi_turn_text_quality_v1": "MULTI_TURN_TEXT_QUALITY",
|
||||
"multi_turn_tool_use_quality_v1": "MULTI_TURN_TOOL_USE_QUALITY",
|
||||
"multi_turn_trajectory_quality_v1": "MULTI_TURN_TRAJECTORY_QUALITY",
|
||||
"multi_turn_task_success_v1": "MULTI_TURN_TASK_SUCCESS",
|
||||
"final_response_match_v2": "FINAL_RESPONSE_MATCH",
|
||||
"final_response_reference_free_v1": "FINAL_RESPONSE_REFERENCE_FREE",
|
||||
"final_response_quality_v1": "FINAL_RESPONSE_QUALITY",
|
||||
}
|
||||
|
||||
|
||||
def _metric_to_code(metric_name: str) -> str:
|
||||
"""Convert a metric name to its Python SDK expression."""
|
||||
if metric_name in _RUBRIC_CONSTANT_MAP:
|
||||
return f"types.RubricMetric.{_RUBRIC_CONSTANT_MAP[metric_name]}"
|
||||
if metric_name in _COMPUTATION_METRICS:
|
||||
return f'types.Metric(name="{metric_name}")'
|
||||
# Unknown metric — try as RubricMetric constant
|
||||
return f'types.Metric(name="{metric_name}")'
|
||||
|
||||
|
||||
def _detect_dataset_type(dataset: dict[str, Any]) -> str:
|
||||
"""Detect whether dataset is single-turn or multi-turn."""
|
||||
cases = dataset.get("eval_cases", [])
|
||||
if not cases:
|
||||
return "unknown"
|
||||
first = cases[0]
|
||||
if "agent_data" in first or "agentData" in first:
|
||||
return "multi_turn"
|
||||
if "prompt" in first:
|
||||
return "single_turn"
|
||||
return "unknown"
|
||||
|
||||
|
||||
def generate_script(
|
||||
dataset_path: str,
|
||||
metrics: list[str],
|
||||
dataset: dict[str, Any],
|
||||
) -> str:
|
||||
"""Generate a complete evaluation Python script."""
|
||||
dataset_type = _detect_dataset_type(dataset)
|
||||
n_cases = len(dataset.get("eval_cases", []))
|
||||
|
||||
metrics_code = ",\n ".join(_metric_to_code(m) for m in metrics)
|
||||
|
||||
script = f'''#!/usr/bin/env python3
|
||||
"""Auto-generated Vertex Evaluation script.
|
||||
|
||||
Dataset: {dataset_path} ({n_cases} case(s), {dataset_type})
|
||||
Metrics: {", ".join(metrics)}
|
||||
|
||||
Run: python <this_script>.py --project <PROJECT_ID> --location <LOCATION>
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
|
||||
import vertexai
|
||||
from vertexai import types
|
||||
from google.genai import types as genai_types
|
||||
|
||||
|
||||
def load_dataset(path: str) -> types.EvaluationDataset:
|
||||
"""Load evaluation dataset from JSON file."""
|
||||
with open(path) as f:
|
||||
data = json.load(f)
|
||||
return types.EvaluationDataset.model_validate(data)
|
||||
|
||||
|
||||
def run_evaluation(project: str, location: str, dataset_path: str):
|
||||
"""Run evaluation and print results."""
|
||||
print(f"Initializing Vertex AI client (project={{project}}, location={{location}})...")
|
||||
client = vertexai.Client(project=project, location=location)
|
||||
|
||||
print(f"Loading dataset from {{dataset_path}}...")
|
||||
dataset = load_dataset(dataset_path)
|
||||
|
||||
metrics = [
|
||||
{metrics_code}
|
||||
]
|
||||
|
||||
print(f"Running evaluation with {{len(metrics)}} metric(s)...")
|
||||
result = client.evals.evaluate(dataset=dataset, metrics=metrics)
|
||||
|
||||
# Print summary metrics
|
||||
print("\\n" + "=" * 60)
|
||||
print("SUMMARY METRICS")
|
||||
print("=" * 60)
|
||||
if result.summary_metrics:
|
||||
for summary in result.summary_metrics:
|
||||
print(f" {{summary.metric_name}}:")
|
||||
print(f" Mean Score: {{summary.mean_score}}")
|
||||
print(f" Pass Rate: {{summary.pass_rate}}")
|
||||
print(f" Std Dev: {{summary.stdev_score}}")
|
||||
print(f" Valid/Total: {{summary.num_cases_valid}}/{{summary.num_cases_total}}")
|
||||
print()
|
||||
|
||||
# Print per-case results
|
||||
print("=" * 60)
|
||||
print("PER-CASE RESULTS")
|
||||
print("=" * 60)
|
||||
if result.eval_case_results:
|
||||
for case_result in result.eval_case_results:
|
||||
print(f"\\n Case {{case_result.eval_case_index}}:")
|
||||
if case_result.response_candidate_results:
|
||||
for candidate in case_result.response_candidate_results:
|
||||
if candidate.metric_results:
|
||||
for metric_name, metric_result in candidate.metric_results.items():
|
||||
print(f" {{metric_name}}:")
|
||||
print(f" Score: {{metric_result.score}}")
|
||||
if metric_result.explanation:
|
||||
explanation = metric_result.explanation[:200]
|
||||
print(f" Explanation: {{explanation}}...")
|
||||
if metric_result.rubric_verdicts:
|
||||
for v in metric_result.rubric_verdicts:
|
||||
status = "PASS" if v.verdict else "FAIL"
|
||||
rubric_id = v.evaluated_rubric.rubric_id if v.evaluated_rubric else "?"
|
||||
print(f" Rubric {{rubric_id}}: {{status}}")
|
||||
if v.reasoning:
|
||||
print(f" Reasoning: {{v.reasoning[:150]}}...")
|
||||
|
||||
# Output as JSON for programmatic consumption
|
||||
print("\\n" + "=" * 60)
|
||||
print("JSON OUTPUT")
|
||||
print("=" * 60)
|
||||
json_output = {{
|
||||
"summary": [
|
||||
{{
|
||||
"metric": s.metric_name,
|
||||
"mean_score": s.mean_score,
|
||||
"pass_rate": s.pass_rate,
|
||||
"stdev": s.stdev_score,
|
||||
}}
|
||||
for s in (result.summary_metrics or [])
|
||||
],
|
||||
}}
|
||||
print(json.dumps(json_output, indent=2, default=str))
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Run Vertex Evaluation")
|
||||
parser.add_argument("--project", "-p", required=True, help="GCP Project ID")
|
||||
parser.add_argument("--location", "-l", default="us-central1", help="GCP Location")
|
||||
parser.add_argument("--dataset", "-d", default="{dataset_path}", help="Dataset JSON path")
|
||||
args = parser.parse_args()
|
||||
run_evaluation(args.project, args.location, args.dataset)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
'''
|
||||
return script
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Generate a runnable Vertex Eval SDK script."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dataset",
|
||||
"-d",
|
||||
required=True,
|
||||
help="Path to the evaluation dataset JSON file.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--metrics",
|
||||
"-m",
|
||||
default="general_quality_v1",
|
||||
help="Comma-separated metric names (default: general_quality_v1).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output",
|
||||
"-o",
|
||||
help="Output script path. Prints to stdout if not specified.",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
if not os.path.exists(args.dataset):
|
||||
print(f"ERROR: File not found: {args.dataset}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
with open(args.dataset) as f:
|
||||
dataset = json.load(f)
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"ERROR: Invalid JSON in {args.dataset}: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
metrics = [m.strip() for m in args.metrics.split(",")]
|
||||
|
||||
# Validate metric names
|
||||
all_known = _RUBRIC_METRICS | _COMPUTATION_METRICS
|
||||
unknown = [m for m in metrics if m not in all_known]
|
||||
if unknown:
|
||||
print(
|
||||
f"WARNING: Unknown metric(s): {', '.join(unknown)}. "
|
||||
"They will be passed as types.Metric(name=...).",
|
||||
file=sys.stderr,
|
||||
)
|
||||
|
||||
script = generate_script(args.dataset, metrics, dataset)
|
||||
|
||||
if args.output:
|
||||
with open(args.output, "w") as f:
|
||||
f.write(script)
|
||||
os.chmod(args.output, 0o755)
|
||||
print(f"Generated eval script: {args.output}", file=sys.stderr)
|
||||
print(
|
||||
f"Run: python {args.output} --project <PROJECT> --location <LOCATION>",
|
||||
file=sys.stderr,
|
||||
)
|
||||
else:
|
||||
print(script)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,277 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Parse ADK session traces into Vertex Evaluation SDK dataset format.
|
||||
|
||||
Reads serialized ADK session JSON (from Session.model_dump_json() or
|
||||
DatabaseSessionService exports) and converts to the canonical
|
||||
EvaluationDataset format for use with client.evals.evaluate().
|
||||
|
||||
Usage:
|
||||
python parse_adk_traces.py --input session.json --output dataset.json
|
||||
python parse_adk_traces.py --input_dir ./sessions/ --output dataset.json
|
||||
python parse_adk_traces.py --input session.json # prints to stdout
|
||||
|
||||
Input format: JSON file(s) with ADK Session structure:
|
||||
{
|
||||
"id": "...", "app_name": "...", "user_id": "...",
|
||||
"events": [{"author": "user"|"agent_name", "content": {...}}, ...]
|
||||
}
|
||||
|
||||
Output format: JSON with EvaluationDataset structure:
|
||||
{
|
||||
"eval_cases": [{"agent_data": {"agents": {...}, "turns": [...]}}]
|
||||
}
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
|
||||
def _is_user_event(event: dict[str, Any]) -> bool:
|
||||
"""Check if an event is from the user."""
|
||||
if event.get("author") == "user":
|
||||
return True
|
||||
content = event.get("content")
|
||||
if isinstance(content, dict) and content.get("role") == "user":
|
||||
return True
|
||||
if event.get("role") == "user":
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _extract_content(event: dict[str, Any]) -> dict[str, Any] | None:
|
||||
"""Extract genai Content from an ADK event dict."""
|
||||
if "content" in event:
|
||||
raw = event["content"]
|
||||
if isinstance(raw, dict) and "parts" in raw:
|
||||
return raw
|
||||
if isinstance(raw, str):
|
||||
return {
|
||||
"role": "user" if _is_user_event(event) else "model",
|
||||
"parts": [{"text": raw}],
|
||||
}
|
||||
if "parts" in event:
|
||||
return {"role": event.get("role", "model"), "parts": event["parts"]}
|
||||
return None
|
||||
|
||||
|
||||
def _extract_author(event: dict[str, Any], default_agent_id: str) -> str:
|
||||
"""Extract the author from an event, preserving sub-agent attribution."""
|
||||
author = event.get("author")
|
||||
if author:
|
||||
return author
|
||||
content = event.get("content")
|
||||
if isinstance(content, dict) and content.get("role") == "user":
|
||||
return "user"
|
||||
if event.get("role") == "user":
|
||||
return "user"
|
||||
return default_agent_id
|
||||
|
||||
|
||||
def _extract_agent_configs(
|
||||
session: dict[str, Any],
|
||||
) -> dict[str, dict[str, Any]]:
|
||||
"""Extract agent configs from session metadata if available."""
|
||||
configs = {}
|
||||
# Check for agent_config in session metadata
|
||||
agent_config = session.get("agent_config") or session.get("agentConfig")
|
||||
if agent_config:
|
||||
agent_id = agent_config.get("agent_id") or agent_config.get(
|
||||
"agentId", "agent"
|
||||
)
|
||||
configs[agent_id] = {
|
||||
"agent_id": agent_id,
|
||||
"agent_type": agent_config.get(
|
||||
"agent_type", agent_config.get("agentType")
|
||||
),
|
||||
"instruction": agent_config.get("instruction"),
|
||||
"description": agent_config.get("description"),
|
||||
}
|
||||
return configs
|
||||
|
||||
# Infer from events — collect unique non-user authors
|
||||
events = session.get("events", [])
|
||||
authors = set()
|
||||
for event in events:
|
||||
author = event.get("author", "")
|
||||
if author and author != "user":
|
||||
authors.add(author)
|
||||
|
||||
if not authors:
|
||||
authors.add(session.get("app_name", session.get("appName", "agent")))
|
||||
|
||||
for author in authors:
|
||||
configs[author] = {"agent_id": author}
|
||||
|
||||
return configs
|
||||
|
||||
|
||||
def _segment_into_turns(
|
||||
events: list[dict[str, Any]], default_agent_id: str
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Segment a flat event list into ConversationTurns.
|
||||
|
||||
A new turn starts with each user message (matching AgentData.from_session()
|
||||
behavior).
|
||||
"""
|
||||
turns = []
|
||||
current_events = []
|
||||
|
||||
for event in events:
|
||||
is_user = _is_user_event(event)
|
||||
|
||||
# Start new turn on user message (if we have accumulated events)
|
||||
if is_user and current_events:
|
||||
turns.append({
|
||||
"turn_index": len(turns),
|
||||
"turn_id": f"turn_{len(turns)}",
|
||||
"events": current_events,
|
||||
})
|
||||
current_events = []
|
||||
|
||||
content = _extract_content(event)
|
||||
if content is None:
|
||||
continue
|
||||
|
||||
author = _extract_author(event, default_agent_id)
|
||||
|
||||
agent_event = {"author": author, "content": content}
|
||||
|
||||
# Preserve state_delta if present (from EventActions)
|
||||
actions = event.get("actions", {})
|
||||
state_delta = actions.get("state_delta") or actions.get("stateDelta")
|
||||
if state_delta:
|
||||
agent_event["state_delta"] = state_delta
|
||||
|
||||
current_events.append(agent_event)
|
||||
|
||||
# Don't forget the last turn
|
||||
if current_events:
|
||||
turns.append({
|
||||
"turn_index": len(turns),
|
||||
"turn_id": f"turn_{len(turns)}",
|
||||
"events": current_events,
|
||||
})
|
||||
|
||||
return turns
|
||||
|
||||
|
||||
def parse_session(session: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Convert a single ADK session dict to an EvalCase dict."""
|
||||
events = session.get("events", [])
|
||||
if not events:
|
||||
raise ValueError(f"Session {session.get('id', 'unknown')} has no events.")
|
||||
|
||||
agent_configs = _extract_agent_configs(session)
|
||||
default_agent_id = next(iter(agent_configs))
|
||||
|
||||
turns = _segment_into_turns(events, default_agent_id)
|
||||
if not turns:
|
||||
raise ValueError(
|
||||
f"Session {session.get('id', 'unknown')} produced no turns."
|
||||
)
|
||||
|
||||
agent_data = {"agents": agent_configs, "turns": turns}
|
||||
return {"agent_data": agent_data}
|
||||
|
||||
|
||||
def parse_file(filepath: str) -> list[dict[str, Any]]:
|
||||
"""Parse a JSON file containing one or more ADK sessions."""
|
||||
with open(filepath) as f:
|
||||
data = json.load(f)
|
||||
|
||||
# Handle single session or list of sessions
|
||||
if isinstance(data, list):
|
||||
sessions = data
|
||||
elif isinstance(data, dict):
|
||||
if "events" in data:
|
||||
sessions = [data]
|
||||
elif "sessions" in data:
|
||||
sessions = data["sessions"]
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Unrecognized format in {filepath}. Expected a session object "
|
||||
"with 'events' field, a list of sessions, or an object with "
|
||||
"'sessions' field."
|
||||
)
|
||||
else:
|
||||
raise ValueError(f"Unexpected JSON type in {filepath}: {type(data)}")
|
||||
|
||||
eval_cases = []
|
||||
for i, session in enumerate(sessions):
|
||||
try:
|
||||
eval_cases.append(parse_session(session))
|
||||
except ValueError as e:
|
||||
print(f"WARNING: Skipping session {i}: {e}", file=sys.stderr)
|
||||
|
||||
return eval_cases
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Parse ADK session traces into Vertex Eval dataset format."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--input",
|
||||
"-i",
|
||||
help="Path to a single ADK session JSON file.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--input_dir",
|
||||
"-d",
|
||||
help="Path to a directory of ADK session JSON files.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output",
|
||||
"-o",
|
||||
help="Output file path. Prints to stdout if not specified.",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
if not args.input and not args.input_dir:
|
||||
parser.error("Specify --input or --input_dir")
|
||||
|
||||
eval_cases = []
|
||||
|
||||
if args.input:
|
||||
if not os.path.exists(args.input):
|
||||
print(f"ERROR: File not found: {args.input}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
eval_cases.extend(parse_file(args.input))
|
||||
|
||||
if args.input_dir:
|
||||
if not os.path.isdir(args.input_dir):
|
||||
print(f"ERROR: Directory not found: {args.input_dir}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
json_files = sorted(
|
||||
f for f in os.listdir(args.input_dir) if f.endswith(".json")
|
||||
)
|
||||
if not json_files:
|
||||
print(f"ERROR: No .json files in {args.input_dir}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
for filename in json_files:
|
||||
filepath = os.path.join(args.input_dir, filename)
|
||||
eval_cases.extend(parse_file(filepath))
|
||||
|
||||
if not eval_cases:
|
||||
print("ERROR: No eval cases produced from input.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
dataset = {"eval_cases": eval_cases}
|
||||
output_json = json.dumps(dataset, indent=2, default=str)
|
||||
|
||||
if args.output:
|
||||
with open(args.output, "w") as f:
|
||||
f.write(output_json)
|
||||
print(
|
||||
f"Wrote {len(eval_cases)} eval case(s) to {args.output}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
else:
|
||||
print(output_json)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,395 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Validate an evaluation dataset for Vertex Eval SDK compatibility.
|
||||
|
||||
Checks structural compliance with the EvaluationDataset schema,
|
||||
required fields per metric type, and common formatting mistakes.
|
||||
|
||||
Usage:
|
||||
python validate_dataset.py --dataset dataset.json
|
||||
python validate_dataset.py --dataset dataset.json --metrics
|
||||
hallucination_v1,tool_call_valid
|
||||
|
||||
Exit codes: 0 = valid, 1 = invalid (with specific errors).
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
|
||||
# Metrics that require specific fields
|
||||
_SINGLE_TURN_METRICS = frozenset({
|
||||
"general_quality_v1",
|
||||
"text_quality_v1",
|
||||
"instruction_following_v1",
|
||||
"grounding_v1",
|
||||
"safety_v1",
|
||||
"hallucination_v1",
|
||||
})
|
||||
_MULTI_TURN_METRICS = frozenset({
|
||||
"multi_turn_general_quality_v1",
|
||||
"multi_turn_text_quality_v1",
|
||||
"multi_turn_tool_use_quality_v1",
|
||||
"multi_turn_trajectory_quality_v1",
|
||||
"multi_turn_task_success_v1",
|
||||
})
|
||||
_FINAL_RESPONSE_METRICS = frozenset({
|
||||
"final_response_match_v2",
|
||||
"final_response_reference_free_v1",
|
||||
"final_response_quality_v1",
|
||||
})
|
||||
_COMPUTATION_METRICS = frozenset({
|
||||
"exact_match",
|
||||
"bleu",
|
||||
"rouge_1",
|
||||
"rouge_l_sum",
|
||||
"tool_call_valid",
|
||||
"tool_name_match",
|
||||
"tool_parameter_key_match",
|
||||
"tool_parameter_kv_match",
|
||||
})
|
||||
_VALID_ROLES = frozenset({"user", "model", "tool"})
|
||||
|
||||
|
||||
class ValidationError:
|
||||
"""A single validation error with location context."""
|
||||
|
||||
def __init__(self, path: str, message: str, severity: str = "ERROR"):
|
||||
self.path = path
|
||||
self.message = message
|
||||
self.severity = severity
|
||||
|
||||
def __str__(self):
|
||||
return f"[{self.severity}] {self.path}: {self.message}"
|
||||
|
||||
|
||||
def _validate_content(content: Any, path: str) -> list[ValidationError]:
|
||||
"""Validate a genai Content object."""
|
||||
errors = []
|
||||
if not isinstance(content, dict):
|
||||
errors.append(
|
||||
ValidationError(path, f"Expected dict, got {type(content).__name__}")
|
||||
)
|
||||
return errors
|
||||
|
||||
role = content.get("role")
|
||||
if role and role not in _VALID_ROLES:
|
||||
errors.append(
|
||||
ValidationError(
|
||||
f"{path}.role",
|
||||
f"Invalid role '{role}'. Must be one of:"
|
||||
f" {', '.join(sorted(_VALID_ROLES))}",
|
||||
)
|
||||
)
|
||||
if role == "assistant":
|
||||
errors.append(
|
||||
ValidationError(
|
||||
f"{path}.role",
|
||||
"Use 'model' instead of 'assistant' (Vertex convention).",
|
||||
)
|
||||
)
|
||||
|
||||
parts = content.get("parts")
|
||||
if parts is None:
|
||||
errors.append(ValidationError(f"{path}.parts", "Missing 'parts' field."))
|
||||
elif not isinstance(parts, list):
|
||||
errors.append(ValidationError(f"{path}.parts", "Must be a list."))
|
||||
elif not parts:
|
||||
errors.append(
|
||||
ValidationError(
|
||||
f"{path}.parts",
|
||||
"Empty parts list.",
|
||||
severity="WARNING",
|
||||
)
|
||||
)
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def _validate_agent_event(event: Any, path: str) -> list[ValidationError]:
|
||||
"""Validate an AgentEvent object."""
|
||||
errors = []
|
||||
if not isinstance(event, dict):
|
||||
errors.append(
|
||||
ValidationError(path, f"Expected dict, got {type(event).__name__}")
|
||||
)
|
||||
return errors
|
||||
|
||||
if "author" not in event:
|
||||
errors.append(ValidationError(f"{path}", "Missing 'author' field."))
|
||||
|
||||
content = event.get("content")
|
||||
if content is None:
|
||||
errors.append(ValidationError(f"{path}", "Missing 'content' field."))
|
||||
else:
|
||||
errors.extend(_validate_content(content, f"{path}.content"))
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def _validate_turn(turn: Any, path: str) -> list[ValidationError]:
|
||||
"""Validate a ConversationTurn object."""
|
||||
errors = []
|
||||
if not isinstance(turn, dict):
|
||||
errors.append(
|
||||
ValidationError(path, f"Expected dict, got {type(turn).__name__}")
|
||||
)
|
||||
return errors
|
||||
|
||||
if "turn_index" not in turn and "turnIndex" not in turn:
|
||||
errors.append(ValidationError(f"{path}", "Missing 'turn_index' field."))
|
||||
|
||||
events = turn.get("events")
|
||||
if events is None:
|
||||
errors.append(ValidationError(f"{path}", "Missing 'events' field."))
|
||||
elif not isinstance(events, list):
|
||||
errors.append(ValidationError(f"{path}.events", "Must be a list."))
|
||||
elif not events:
|
||||
errors.append(ValidationError(f"{path}.events", "Empty events list."))
|
||||
else:
|
||||
for i, event in enumerate(events):
|
||||
errors.extend(_validate_agent_event(event, f"{path}.events[{i}]"))
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def _validate_agent_data(agent_data: Any, path: str) -> list[ValidationError]:
|
||||
"""Validate an AgentData object."""
|
||||
errors = []
|
||||
if not isinstance(agent_data, dict):
|
||||
errors.append(
|
||||
ValidationError(path, f"Expected dict, got {type(agent_data).__name__}")
|
||||
)
|
||||
return errors
|
||||
|
||||
agents = agent_data.get("agents")
|
||||
if agents is None:
|
||||
errors.append(ValidationError(f"{path}", "Missing 'agents' field."))
|
||||
elif not isinstance(agents, dict):
|
||||
errors.append(ValidationError(f"{path}.agents", "Must be a dict."))
|
||||
elif not agents:
|
||||
errors.append(ValidationError(f"{path}.agents", "Empty agents map."))
|
||||
else:
|
||||
for agent_id, config in agents.items():
|
||||
if not isinstance(config, dict):
|
||||
errors.append(
|
||||
ValidationError(
|
||||
f"{path}.agents.{agent_id}",
|
||||
f"Expected dict, got {type(config).__name__}",
|
||||
)
|
||||
)
|
||||
elif "agent_id" not in config and "agentId" not in config:
|
||||
errors.append(
|
||||
ValidationError(
|
||||
f"{path}.agents.{agent_id}",
|
||||
"Missing 'agent_id' field.",
|
||||
severity="WARNING",
|
||||
)
|
||||
)
|
||||
|
||||
turns = agent_data.get("turns")
|
||||
if turns is None:
|
||||
errors.append(ValidationError(f"{path}", "Missing 'turns' field."))
|
||||
elif not isinstance(turns, list):
|
||||
errors.append(ValidationError(f"{path}.turns", "Must be a list."))
|
||||
elif not turns:
|
||||
errors.append(ValidationError(f"{path}.turns", "Empty turns list."))
|
||||
else:
|
||||
for i, turn in enumerate(turns):
|
||||
errors.extend(_validate_turn(turn, f"{path}.turns[{i}]"))
|
||||
|
||||
# Check turn_index ordering
|
||||
indices = []
|
||||
for turn in turns:
|
||||
idx = turn.get("turn_index", turn.get("turnIndex"))
|
||||
if idx is not None:
|
||||
indices.append(idx)
|
||||
if indices and indices != list(range(len(indices))):
|
||||
errors.append(
|
||||
ValidationError(
|
||||
f"{path}.turns",
|
||||
f"turn_index values are not sequential 0-based: {indices}",
|
||||
severity="WARNING",
|
||||
)
|
||||
)
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def _validate_eval_case(
|
||||
case: Any, index: int, metrics: list[str] | None
|
||||
) -> list[ValidationError]:
|
||||
"""Validate a single EvalCase."""
|
||||
path = f"eval_cases[{index}]"
|
||||
errors = []
|
||||
|
||||
if not isinstance(case, dict):
|
||||
errors.append(
|
||||
ValidationError(path, f"Expected dict, got {type(case).__name__}")
|
||||
)
|
||||
return errors
|
||||
|
||||
has_prompt = "prompt" in case
|
||||
has_agent_data = "agent_data" in case or "agentData" in case
|
||||
has_response = "response" in case
|
||||
has_reference = "reference" in case
|
||||
|
||||
if not has_prompt and not has_agent_data:
|
||||
errors.append(
|
||||
ValidationError(
|
||||
path,
|
||||
"Must have either 'prompt' (single-turn) or 'agent_data'"
|
||||
" (multi-turn).",
|
||||
)
|
||||
)
|
||||
|
||||
if has_prompt and has_agent_data:
|
||||
errors.append(
|
||||
ValidationError(
|
||||
path,
|
||||
"Has both 'prompt' and 'agent_data'. Use one or the other.",
|
||||
severity="WARNING",
|
||||
)
|
||||
)
|
||||
|
||||
# Validate agent_data structure
|
||||
if has_agent_data:
|
||||
ad = case.get("agent_data") or case.get("agentData")
|
||||
errors.extend(_validate_agent_data(ad, f"{path}.agent_data"))
|
||||
|
||||
# Check metric-specific requirements
|
||||
if metrics:
|
||||
for metric in metrics:
|
||||
if (
|
||||
metric in _SINGLE_TURN_METRICS
|
||||
and not has_prompt
|
||||
and not has_agent_data
|
||||
):
|
||||
errors.append(
|
||||
ValidationError(
|
||||
path,
|
||||
f"Metric '{metric}' requires 'prompt' and 'response' fields.",
|
||||
)
|
||||
)
|
||||
if metric in _MULTI_TURN_METRICS and not has_agent_data:
|
||||
errors.append(
|
||||
ValidationError(
|
||||
path,
|
||||
f"Metric '{metric}' requires 'agent_data' with conversation"
|
||||
" turns.",
|
||||
)
|
||||
)
|
||||
if metric in _FINAL_RESPONSE_METRICS and not has_agent_data:
|
||||
errors.append(
|
||||
ValidationError(
|
||||
path,
|
||||
f"Metric '{metric}' requires 'agent_data'.",
|
||||
)
|
||||
)
|
||||
if metric in _COMPUTATION_METRICS and not has_reference:
|
||||
errors.append(
|
||||
ValidationError(
|
||||
path,
|
||||
f"Metric '{metric}' requires a 'reference' field.",
|
||||
severity="WARNING",
|
||||
)
|
||||
)
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def validate_dataset(
|
||||
dataset: dict[str, Any], metrics: list[str] | None = None
|
||||
) -> list[ValidationError]:
|
||||
"""Validate an entire EvaluationDataset."""
|
||||
errors = []
|
||||
|
||||
if not isinstance(dataset, dict):
|
||||
errors.append(
|
||||
ValidationError("root", f"Expected dict, got {type(dataset).__name__}")
|
||||
)
|
||||
return errors
|
||||
|
||||
eval_cases = dataset.get("eval_cases")
|
||||
if eval_cases is None:
|
||||
errors.append(ValidationError("root", "Missing 'eval_cases' field."))
|
||||
return errors
|
||||
|
||||
if not isinstance(eval_cases, list):
|
||||
errors.append(ValidationError("eval_cases", "Must be a list."))
|
||||
return errors
|
||||
|
||||
if not eval_cases:
|
||||
errors.append(ValidationError("eval_cases", "Empty eval_cases list."))
|
||||
return errors
|
||||
|
||||
for i, case in enumerate(eval_cases):
|
||||
errors.extend(_validate_eval_case(case, i, metrics))
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Validate an evaluation dataset for Vertex Eval SDK."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dataset",
|
||||
"-d",
|
||||
required=True,
|
||||
help="Path to the evaluation dataset JSON file.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--metrics",
|
||||
"-m",
|
||||
help="Comma-separated list of metrics to validate against.",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
with open(args.dataset) as f:
|
||||
dataset = json.load(f)
|
||||
except FileNotFoundError:
|
||||
print(f"ERROR: File not found: {args.dataset}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"ERROR: Invalid JSON in {args.dataset}: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
metrics = args.metrics.split(",") if args.metrics else None
|
||||
|
||||
errors = validate_dataset(dataset, metrics)
|
||||
|
||||
# Report
|
||||
n_cases = len(dataset.get("eval_cases", []))
|
||||
real_errors = [e for e in errors if e.severity == "ERROR"]
|
||||
warnings = [e for e in errors if e.severity == "WARNING"]
|
||||
|
||||
print(f"Dataset: {args.dataset}")
|
||||
print(f"Eval cases: {n_cases}")
|
||||
if metrics:
|
||||
print(f"Validating against metrics: {', '.join(metrics)}")
|
||||
print()
|
||||
|
||||
if real_errors:
|
||||
print(f"ERRORS ({len(real_errors)}):")
|
||||
for e in real_errors:
|
||||
print(f" {e}")
|
||||
print()
|
||||
|
||||
if warnings:
|
||||
print(f"WARNINGS ({len(warnings)}):")
|
||||
for w in warnings:
|
||||
print(f" {w}")
|
||||
print()
|
||||
|
||||
if not real_errors and not warnings:
|
||||
print("VALID: No issues found.")
|
||||
|
||||
if real_errors:
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user