# SkillPatch skill: agent-platform-eval-flywheel

This skill guides agents through evaluating and iteratively improving GenAI models and agents on Google Cloud using the Agent Platform Eval Quality Flywheel methodology. It covers building eval datasets, selecting or writing custom metrics, analyzing failures, comparing results, and using the GenAI Evaluation SDK. It also includes safety tiers for cost-incurring operations and a structured five-stage quality flywheel workflow.

You (the agent) were given this URL and asked to use this skill. This is a **public** skill — no SkillPatch account, API key, or prior setup is required. Two ways to use it:

**1. Use it right now** — the skill's complete file tree (SKILL.md + all reference files) is inlined below; read `SKILL.md` first, then follow it, consulting the other files as it directs.

**2. Install the exact package onto disk** (recommended if you can run a shell — this reproduces the skill byte-for-byte, including any binary assets that can't be inlined):

```bash
mkdir -p .claude/skills/agent-platform-eval-flywheel
curl -sSL https://skillpatch.dev/install_skill/agent-platform-eval-flywheel | tar -xz -C .claude/skills/
```

(`.claude/skills/` is Claude Code's convention; use whatever directory your agent loads skills from.)


---

## Skill files (13)

- `SKILL.md`
- `references/dataset_schema.md`
- `references/deployment.md`
- `references/failure_patterns.md`
- `references/metric_registry.md`
- `references/sdk_patterns.md`
- `scripts/compare_results.py`
- `scripts/endpoint_evaluation.py`
- `scripts/inspect_results.py`
- `scripts/maas_evaluation.py`
- `scripts/parse_adk_traces.py`
- `scripts/render_html_report.py`
- `scripts/validate_dataset.py`


### `SKILL.md`

````markdown
---
name: agent-platform-eval-flywheel
metadata:
  category: AiAndMachineLearning
description: >-
  Measures and improves the quality of AI models and agents on Google Cloud
  using the Eval Quality Flywheel methodology. Use when evaluating an agent or
  model, building an eval dataset, picking or writing evaluation metrics,
  analyzing failures, comparing results before and after a fix, or when
  guidance is needed on Agent Platform eval methodology — including
  dataset schema, LLM-as-judge scoring, and common failure causes. For
  fine-tuning, use agent-platform-tuning. For general production deployment,
  use agent-platform-deploy.
---

# Agent Platform Eval Flywheel Skill

Help users evaluate and iteratively improve GenAI models and agents using
the Agent Platform GenAI Evaluation SDK (`google.genai` / `agentplatform`).

## When to use this skill

-   Evaluating GenAI agents or models with the Agent Platform GenAI
    Evaluation SDK (`client.evals.evaluate()`).
-   Creating evaluation datasets from session traces, pandas DataFrames, or
    synthetic generation.
-   Selecting, configuring, or writing custom evaluation metrics.
-   Analyzing rubric verdicts, loss patterns, and clustering failures.
-   Suggesting concrete code/prompt improvements based on eval results.
-   Evaluating a model served on an Agent Platform **endpoint** (BYOM) or a
    **Model-as-a-Service (MaaS)** model by ID — including deploying the model
    first if needed. For this case, follow
    [references/deployment.md](references/deployment.md) and use the
    `endpoint_evaluation.py` / `maas_evaluation.py` scripts.

## Safety & Confirmation Tiers (CRITICAL)

Before executing any commands or scripts on behalf of the user, you MUST adhere
to the following safety tiers based on the action requested:

1.  **Tier R**: Read-only (`inspect_results.py`, `compare_results.py`, `validate_dataset.py`, `parse_adk_traces.py`, `render_html_report.py`)
    *   **Rule**: No confirmation needed. You may execute these helper scripts immediately to inspect data, validate schemas, parse traces, or compare evaluation results.
2.  **Tier M: Read-only with Compute Costs (`client.evals.run_inference`, `client.evals.evaluate`, `client.evals.generate_user_scenarios`, `client.evals.generate_loss_clusters`)**
    *   **Rule**: These operations invoke LLMs or remote evaluation services
    that consume compute resources and incur costs. This requires
    **interactive confirmation** with 'Yes'/'No' options. Once granted once,
    you do not have to prompt for future evaluation.

## Setup

Install the SDK:

```bash
pip install google-cloud-aiplatform[evaluation]>=1.154.0 google-genai>=1.0.0
```

Need `GOOGLE_CLOUD_PROJECT` and `GOOGLE_CLOUD_LOCATION`. Check env vars
first; if missing, ask the user. Newer Gemini models often need
`location="global"`.

## The Quality Flywheel

Five stages, run in order on the first pass, then loop 2 → 5 until quality
targets are met.

### Shortcuts that waste time

| Shortcut                                                 | Why it fails                                                                                              |
| -------------------------------------------------------- | --------------------------------------------------------------------------------------------------------- |
| "I'll tune the metric threshold down so it passes."      | Hides real failures. Fix the agent, not the bar.                                                          |
| "This case is flaky, I'll skip it."                      | Flakiness reveals non-determinism in the agent. Fix with `temperature=0` or stricter instructions.        |
| "I just need to fix the eval dataset, not the agent."    | If expected outputs keep moving, the agent has a behavior problem.                                        |
| "I can tell from the trace it works — skip Stage 3."     | Self-grading doesn't generalize. Always run `evaluate()` and read scores.                                 |
| "One iteration is enough."                               | Expect 5–10+ iterations. Stopping early leaves regressions on other metrics undetected.                   |

### 1. Prepare Data

Produce an `EvaluationDataset`. There are three input shapes, pick the one
that matches the data the user already has:

-   **`EvalCase` list (single-turn or multi-turn):**

    ```python
    from agentplatform import types
    dataset = types.EvaluationDataset(eval_cases=[
        types.EvalCase(prompt="What is 2+2?", response="4", reference="4"),
        # For multi-turn agent traces, set agent_data instead of prompt/response.
    ])
    ```

    Multi-turn agent traces wrap each conversation in `AgentData` →
    `ConversationTurn` → `AgentEvent`. See
    [references/dataset_schema.md](references/dataset_schema.md) for the
    full type hierarchy.

-   **Pandas DataFrame (tabular sources — CSV, BigQuery, Sheets):**

    ```python
    import pandas as pd
    from agentplatform import types

    df = pd.DataFrame({
        "prompt":    ["What is 2+2?", "Capital of France?"],
        "response":  ["4",            "Paris"],
        "reference": ["4",            "Paris"],
    })
    dataset = types.EvaluationDataset(eval_dataset_df=df)
    ```

    Column names must match the fields the chosen metrics expect (see
    [references/dataset_schema.md](references/dataset_schema.md) for the
    per-metric requirements table).

-   **Cold start (no data at all):** synthesize scenarios server-side with
    `client.evals.generate_user_scenarios(...)` and a
    `UserScenarioGenerationConfig` (`user_scenario_count`,
    `simulation_instruction`, `environment_data`). Stage 2 plays them out.

For ADK session dumps, use `scripts/parse_adk_traces.py` instead of writing
the conversion by hand.

### 2. Run Inference

Populate responses/traces on the dataset. **Skip this stage** if traces are
already complete (e.g., production logs or replay).

```python
# Agent eval — pass a callable wrapping the user's ADK Agent/App.
client.evals.run_inference(model=agent_callable, src=dataset)

# Model eval — pass a model ID directly.
client.evals.run_inference(model="gemini-2.5-flash", src=dataset)

# Synthesized scenarios — let the simulator drive.
client.evals.run_inference(
    model=agent_callable,
    src=dataset,
    user_simulator_config=UserSimulatorConfig(max_turn=10),
)

# DataFrame also works as src= — no EvalCase wrapping needed.
client.evals.run_inference(model="gemini-2.5-flash", src=df)
```

### 3. Grade (always run)

```python
result = client.evals.evaluate(dataset=dataset, metrics=[...])
```

**Pick metrics by what you want to measure.** Full catalog in
[references/metric_registry.md](references/metric_registry.md).

**Agent metrics (multi-turn, adaptive rubrics)** — start here for agent eval.

| Goal                                          | Metric                          |
| --------------------------------------------- | ------------------------------- |
| Did the agent achieve the user's goal?        | `multi_turn_task_success`       |
| Was the reasoning path logical and efficient? | `multi_turn_trajectory_quality` |
| Tool/function calling quality across turns    | `multi_turn_tool_use_quality`   |
| Overall conversational quality                | `multi_turn_general_quality`    |
| Final response quality (no reference needed)  | `final_response_quality`        |
| Final response vs. a golden reference         | `final_response_match`          |
| Single-turn tool use                          | `tool_use_quality`              |

**General quality metrics (single-turn, adaptive rubrics)** — for model eval.

| Goal                                                  | Metric                  |
| ----------------------------------------------------- | ----------------------- |
| Overall response quality (recommended starting point) | `general_quality`       |
| Linguistic quality (fluency, coherence, grammar)      | `text_quality`          |
| Adherence to specific constraints / instructions      | `instruction_following` |

**Static rubric metrics (fixed criteria)** — apply alongside the above.

| Goal                                              | Metric          |
| ------------------------------------------------- | --------------- |
| Catch hallucinated claims (RAG, factual answers)  | `hallucination` |
| Factuality / consistency against provided context | `grounding`     |
| Safety policy compliance                          | `safety`        |

**Domain-specific check no built-in covers:** write a custom metric.

-   **Predefined:** `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.

**Always persist the result** so Stage 4 and 5 can read it. Save both JSON
(machine-readable, diffable) and HTML (human-readable, linkable):

```python
import datetime
from pathlib import Path

from agentplatform._genai import _evals_visualization

out_dir = Path("artifacts/grade_results")
out_dir.mkdir(parents=True, exist_ok=True)
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")

result_json = result.model_dump_json()
(out_dir / f"results_{ts}.json").write_text(result_json)

html = _evals_visualization.get_evaluation_html(result_json)
(out_dir / f"results_{ts}.html").write_text(str(html))
```

Or after the fact: `scripts/render_html_report.py --type evaluation` or
`scripts/inspect_results.py --save-html`.

### 4. Analyze Failures

Read `summary_metrics` and `eval_case_results` — never fabricate scores.
Use `scripts/inspect_results.py --failing-only` to filter to failures.

For each failed metric, see
[references/failure_patterns.md](references/failure_patterns.md) for deeper
diagnoses. The compact mapping:

| Failing metric                       | What to change                                                                                                            |
| ------------------------------------ | ------------------------------------------------------------------------------------------------------------------------- |
| `multi_turn_task_success` low        | The agent isn't completing the goal — fix orchestration, missing tool calls, premature termination, wrong tool selection. |
| `multi_turn_trajectory_quality` low  | The agent reaches the goal inefficiently — refine planning prompts, remove redundant tool calls.                          |
| `multi_turn_tool_use_quality` low    | Fix tool descriptions, parameter docstrings, or agent instructions for tool selection.                                    |
| `final_response_quality` low         | Read auto-generated rubric verdicts; refine instructions to address the worst-scoring criterion.                          |
| `final_response_match` low           | The agent's final answer doesn't match the golden reference — adjust response format or update the reference.             |
| `hallucination` low                  | Tighten instructions to stay grounded in tool output; verify the tool actually returned the claimed data.                 |
| `grounding` low                      | The response contradicts the provided context — add explicit "cite only from context" instructions.                       |
| `safety` low                         | Add safety guardrails; review the violating content category in the rubric verdict.                                       |
| `general_quality` / `text_quality` low | Adjust system instruction wording; the model's default phrasing is too generic for the task.                            |
| `instruction_following` low          | The agent is ignoring constraints — restate them in the system instruction or use stricter wording.                       |
| Agent calls wrong tools              | Fix tool descriptions, agent instructions, or `tool_config`.                                                              |
| Agent calls extra tools              | Add explicit stop instructions, or switch to `multi_turn_tool_use_quality` to surface the extra calls in the rubric.      |

**For 10+ failures on the same metric**, use the **Error Analysis service**
to cluster failures into themes (L1/L2 taxonomy categories) instead of
reading every trace:

```python
# Only supports multi_turn_task_success and multi_turn_tool_use_quality.
# Service runs in the global region.
analysis_client = agentplatform.Client(project="PROJECT_ID", location="global")
response = analysis_client.evals.generate_loss_clusters(
    eval_result=result,
    metric="multi_turn_task_success",
    config={"max_top_cluster_count": 5},
)
for r in response.results:
    for cluster in r.clusters:
        print(
            f"[{cluster.taxonomy_entry.l1_category}/"
            f"{cluster.taxonomy_entry.l2_category}] "
            f"{cluster.item_count} cases — {cluster.taxonomy_entry.description}"
        )
```

Save `response.model_dump_json()` and render with
`scripts/render_html_report.py --type loss-analysis`.

### 5. Optimize & Iterate

Apply a fix targeting the failing metric. Re-run Stage 3. Compare with
`scripts/compare_results.py --baseline <prev> --candidate <new>` to confirm
the target improved AND no other metric regressed.

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    |

Expect 5–10+ iterations per failing case. Only after a case passes should
you expand coverage with more eval cases.

## Proving your work

Never claim eval results you didn't read from an actual `result` object.

-   After running eval, print the `summary_metrics` table
    (`scripts/inspect_results.py`).
-   After a fix, show before/after via `scripts/compare_results.py`.
-   Before declaring success, confirm ALL cases pass — not just the one you
    were working on.

If you can't produce the evidence (SDK call failed, result truncated,
metric unsupported), say so explicitly. Don't paper over gaps.

## 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
    agentplatform`, `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.

## SDK Quick Reference

```python
import agentplatform
from agentplatform import types
from google.genai import types as genai_types
import pandas as pd

# Initialize client
client = agentplatform.Client(project="PROJECT_ID", location="LOCATION")

# --- SINGLE-TURN EVAL (EvalCase list) ---
dataset = types.EvaluationDataset(eval_cases=[
    types.EvalCase(prompt="Query here", response="Model response here"),
])

# --- SINGLE-TURN EVAL (pandas DataFrame) ---
df = pd.DataFrame({
    "prompt":   ["Q1", "Q2"],
    "response": ["A1", "A2"],
})
dataset = types.EvaluationDataset(eval_dataset_df=df)

# --- 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 {"score": 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.

## Bundled scripts

Script                   | When to use
------------------------ | -----------
`validate_dataset.py`    | Before Stage 3 — catch malformed `EvaluationDataset` JSON.
`parse_adk_traces.py`    | Stage 1 — convert ADK session dumps to the canonical dataset shape.
`inspect_results.py`     | Stages 3/4 — render summary + per-case scores. `--save-html` for a browsable report.
`compare_results.py`     | Stage 5 — diff baseline vs. candidate, detect regressions.
`render_html_report.py`  | Render HTML from a saved result JSON or loss-clusters JSON.
`endpoint_evaluation.py` | Stages 2/3 against a deployed Agent Platform endpoint (BYOM). See [references/deployment.md](references/deployment.md).
`maas_evaluation.py`     | Stages 2/3 against a Model-as-a-Service model by ID. See [references/deployment.md](references/deployment.md).

````


### `references/dataset_schema.md`

````markdown
# Evaluation Dataset Schema

Canonical formats for evaluation datasets in the Google GenAI Evaluation SDK.
Source of truth: `agentplatform/_genai/types/evals.py` and
`agentplatform/_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 agentplatform 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 agentplatform 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 agentplatform 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"` (Agent Platform convention) |
| Missing `turn_index`             | Always set sequential 0-based indices  |
| Tool response without            | Wrap in `genai_types.FunctionResponse` |
: `function_response`              :                                        :
| Using `response` field for       | Use `agent_data` with full trajectory  |
: multi-turn                       :                                        :
| Mixing `prompt` and `agent_data` | Use one or the other per EvalCase      |

````


### `references/deployment.md`

````markdown
# Deploying and Evaluating Models on Agent Platform Endpoints

Reference for the deployment-and-endpoint-evaluation subset of the Quality
Flywheel. Use this guide when the user needs to evaluate a model that is **not
yet served** (must be deployed first) or is served on an Agent Platform
**endpoint** rather than called through the `agentplatform` Python client
directly.

The two scripts referenced here — `scripts/endpoint_evaluation.py` and
`scripts/maas_evaluation.py` — wrap Stage 2 (inference) and Stage 3 (grading) of
the Flywheel against deployed endpoints. They produce the same
`EvaluationResult` shape consumed by `scripts/inspect_results.py` and
`scripts/compare_results.py`, so the rest of the Flywheel (Stages 4 and 5)
applies unchanged.

For the public-docs version of this workflow, see the
[Agent Platform model evaluation docs](https://docs.cloud.google.com/gemini-enterprise-agent-platform/models/eval-python-sdk/run-evaluation).

## When to use this guide

-   The user wants to evaluate a **custom-weights / Bring-Your-Own-Model
    (BYOM)** model deployed to an Agent Platform endpoint.
-   The user wants to evaluate a **Model-as-a-Service (MaaS)** model (e.g.
    `meta/llama3-8b`, `gemini-1.5-pro`) by model ID.
-   The user is at the deploy-then-eval stage: weights exist in GCS or in Model
    Garden but no endpoint has been provisioned yet.

For agent (multi-turn) evaluation, dataset preparation, metric selection,
failure clustering, or iteration, stay in the main SKILL.md flow. This guide
only covers the deployment + endpoint-inference subset.

## Setup

Install the deployment-eval dependencies (in addition to the base SDK install in
the main SKILL.md):

```bash
python3 -m venv .model-eval
source .model-eval/bin/activate
pip install google-cloud-aiplatform[evaluation]>=1.154.0 google-genai>=1.0.0 requests
```

The scripts also shell out to `gcloud` and `gsutil`, so the user must have the
Google Cloud SDK installed and `gcloud auth application-default login`
completed.

Confirm project / region / quota project:

```bash
gcloud config get project
gcloud config get compute/region
gcloud config get billing/quota_project
```

## 1. Deploy a Model

> [!TIP]
>
> If the `agent-platform-deploy` skill is available, prefer it — it carries the
> full deployment surface. This section is the minimum needed to get an endpoint
> to evaluate against.

```bash
gcloud ai model-garden models deploy \
    --project=$PROJECT_ID \
    --region=$LOCATION_ID \
    --model=$MODEL \
    --machine-type=$MACHINE_TYPE \
    --accelerator-type=$ACCELERATOR_TYPE \
    --accelerator-count=$ACCELERATOR_COUNT \
    --endpoint-display-name=$ENDPOINT_NAME \
    --asynchronous
```

`MODEL` is either a Model Garden model ID (`meta/llama3-8b`) or a GCS URL to
custom weights. If any of `MACHINE_TYPE`, `ACCELERATOR_TYPE`,
`ACCELERATOR_COUNT` are missing, **stop and ask the user** — do not pick
defaults silently. Confirm the full command with the user before running it.

### Determining hardware requirements

For custom weights stored in GCS, check for a tuning marker first:

```bash
gcloud storage cat $GCS_URL/managed_oss_fine_tuning_marker.json
gcloud storage cat $GCS_URL/config.json
```

If neither is present, ask the user for the base model ID, then query
recommended configs:

```bash
gcloud ai model-garden models list-deployment-config --model=$BASE_MODEL_ID
```

Summarize the recommended machine + accelerator combinations and confirm with
the user before deploying.

## 2. Check Deploy Status

Deploys are asynchronous. Poll the operation:

```bash
gcloud ai operations describe $OPERATION_ID --region=$LOCATION_ID
```

Report the status to the user — including the "not found" case — without asking
for confirmation first.

## 3. Run Inference and Evaluation

> [!WARNING]
>
> Both scripts use LLM-as-a-judge metrics. Tell the user up-front: **this can
> take ~30 minutes per 50 samples**.

If the scripts' imports (`google.cloud.aiplatform.agentplatform`) fail in the
local environment, do **not** loop on `pip install` — fail fast and tell the
user to install the deps from the Setup section above.

Dataset format: JSONL in GCS with a `prompt` field on each line. See
[dataset_schema.md](dataset_schema.md) for the per-metric column requirements.
If the user's dataset doesn't have a `prompt` column, reformat it before
running.

Recommended starter metrics (confirm with the user first):

-   `coherence` — does the response hold together logically?
-   `fluency` — grammatical correctness and natural flow.
-   `text_quality` — overall text quality.

For the full metric catalog, see [metric_registry.md](metric_registry.md).

Suggest an experiment name based on `${MODEL_ID}_${DATASET_NAME}` if the user
doesn't provide one.

### 3.1. Bring-Your-Own-Model (BYOM) Endpoint

Inspect the endpoint first to confirm it exists and to capture the
`dedicatedEndpointDns` (if dedicated) for the `--dedicated_endpoint_dns` flag:

```bash
gcloud ai endpoints describe $ENDPOINT_ID --region=$LOCATION_ID --project=$PROJECT_ID
```

If the endpoint ID or the dataset bucket doesn't exist, **stop and ask the
user** for a valid value — don't propose a confirmation prompt with broken
inputs.

```bash
python3 scripts/endpoint_evaluation.py \
    --project_id=$PROJECT_ID \
    --location=$LOCATION_ID \
    --endpoint_id=$ENDPOINT_ID \
    --dataset=gs://your-bucket/eval.jsonl \
    --metrics "GENERAL_QUALITY"
```

With a dedicated DNS:

```bash
python3 scripts/endpoint_evaluation.py \
    --project_id=$PROJECT_ID \
    --location=$LOCATION_ID \
    --endpoint_id=$ENDPOINT_ID \
    --dataset=gs://your-bucket/eval.jsonl \
    --dedicated_endpoint_dns=$DEDICATED_ENDPOINT_DNS \
    --metrics "GENERAL_QUALITY"
```

### 3.2. Model-as-a-Service (MaaS)

For MaaS models, no endpoint deploy is needed — call the model by ID:

```bash
python3 scripts/maas_evaluation.py \
    --project_id=$PROJECT_ID \
    --location=$LOCATION_ID \
    --model_id=$MAAS_MODEL_ID \
    --dataset=gs://your-bucket/eval.jsonl \
    --metrics "GENERAL_QUALITY"
```

Same input-validation rule as 3.1: if the bucket doesn't resolve, stop and ask
the user.

## 4. Feed Results Back Into the Flywheel

Both scripts print `summary_metrics` and the first 5 rows of `metrics_table`. To
persist the full result for Stage 4/5 analysis, capture the returned
`eval_result` and serialize it the same way as the main SKILL.md "Always persist
the result" block. Then:

-   `scripts/inspect_results.py --failing-only` for per-case triage.
-   `scripts/compare_results.py --baseline <prev> --candidate <new>` after a fix
    to confirm the target metric improved and nothing regressed.

The deployment workflow is only Stages 1–3 of the Flywheel. Stages 4 (Analyze
Failures) and 5 (Optimize & Iterate) work identically whether the result came
from a deployed endpoint, a MaaS model, or a direct `client.evals.evaluate(...)`
call.

````


### `references/failure_patterns.md`

```markdown
# Evaluation Failure Patterns & Fixes

Common failure modes observed in GenAI agent evaluations, mapped to their root
causes and concrete fixes. Metric IDs below are the **unversioned** form (the
SDK auto-resolves them to the latest version); see
[metric_registry.md](metric_registry.md) for the full catalog.

For the compact failure → fix mapping, see the *What to fix when scores fail*
table in SKILL.md.

## Metric-Specific Failures

### Low `hallucination` or `grounding` 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` or `text_quality`

**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` (single-turn) or `multi_turn_tool_use_quality`

**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.  For granular diagnosis of parameter mismatches, add a computation metric
    like `tool_parameter_kv_match` alongside the rubric metric.

### Low `multi_turn_trajectory_quality`

**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`

**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`

**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` 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. For agent-trace
datasets, the standard top-level field is `agent_data` (a structured
turns/events object) — the agent's final response lives inside it. Flat
placeholders like `{response}` only resolve in the DataFrame eval path. Always
use `.get()` with defaults.

## Analysis Workflow

When eval results show failures:

1.  **Start with `summary_metrics`** — identify which metrics scored lowest.
    Use `scripts/inspect_results.py` to render the table.
2.  **Drill into `eval_case_results`** — find specific failing cases. Use
    `scripts/inspect_results.py --failing-only` to filter.
3.  **For 10+ failures on the same metric** — run the Error Analysis service
    (`client.evals.generate_loss_clusters`) for `multi_turn_task_success` or
    `multi_turn_tool_use_quality` to cluster failures into L1/L2 themes
    instead of skimming case-by-case.
4.  **Read `rubric_verdicts`** — understand why the judge scored low.
5.  **Cross-reference with `agent_data`** — find the exact turn/event that
    caused the failure.
6.  **Identify the pattern** — is it a prompt issue, tool issue, or data issue?
7.  **Apply the targeted fix** — from the table above or the SKILL.md
    *What to fix when scores fail* table.
8.  **Re-run and compare** — use `scripts/compare_results.py` to verify the
    fix improved the target metric without regressing others.

```


### `references/metric_registry.md`

````markdown
# GenAI Evaluation Metric Registry

Complete catalog of evaluation metrics available in the `agentplatform` SDK.
Source of truth: `agentplatform/_genai/_evals_metric_loaders.py` and
`agentplatform/_genai/_evals_metric_handlers.py`.

Metric IDs in this file are the **unversioned** form (e.g.,
`multi_turn_task_success`, not `multi_turn_task_success_v1`). The SDK resolves
unversioned names to the latest version; pin to a specific version only when
required for reproducibility (e.g., `RubricMetric.GENERAL_QUALITY(version="v2")`).

## 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 the Agent Platform AutoRater. No judge model needed.

### Agent metrics (multi-turn, adaptive rubrics)

Start here for agent evaluation. Adaptive rubrics generate criteria from the
trace at runtime.

| Metric ID                       | What it measures                                                                       | Required fields                  |
| ------------------------------- | -------------------------------------------------------------------------------------- | -------------------------------- |
| `multi_turn_task_success`       | Whether the user's goal was fulfilled across the full multi-turn conversation.         | `agent_data` with task context   |
| `multi_turn_trajectory_quality` | Sequential logic, efficiency, and error-recovery robustness across turns.              | `agent_data` with full trajectory |
| `multi_turn_tool_use_quality`   | Technical and semantic correctness of tool calls across the multi-turn conversation.   | `agent_data` with function calls |
| `multi_turn_general_quality`    | Overall response quality within a multi-turn dialogue.                                 | `agent_data` (1+ turns)          |
| `multi_turn_text_quality`       | Linguistic text quality within a multi-turn dialogue.                                  | `agent_data` (1+ turns)          |
| `final_response_quality`        | Comprehensive evaluation of the final response and intermediate tool usage.            | `agent_data`                     |
| `final_response_match`          | Compares the agent's final response to a provided golden reference answer.             | `agent_data`, `reference`        |
| `final_response_reference_free` | Final-response quality without a reference answer (requires custom rubrics).           | `agent_data` + rubric_groups     |
| `tool_use_quality`              | Tool selection, parameter accuracy, and step sequence correctness (single-turn).       | `agent_data` with tool calls     |

### General quality metrics (single-turn, adaptive rubrics)

For model evaluation (no agent orchestration).

| Metric ID               | What it measures                                                  | Required fields      |
| ----------------------- | ----------------------------------------------------------------- | -------------------- |
| `general_quality`       | Overall response quality with auto-generated content-based criteria. Recommended starting point for non-agent eval. | `prompt`, `response` |
| `text_quality`          | Linguistic aspects: fluency, coherence, grammar.                  | `prompt`, `response` |
| `instruction_following` | How well the response adheres to specific constraints / instructions. | `prompt`, `response` |

### Static rubric metrics (fixed criteria)

Apply alongside the agent or general-quality metrics above. Fixed rubrics — no
adaptive generation.

| Metric ID       | What it measures                                                                        | Required fields                 |
| --------------- | --------------------------------------------------------------------------------------- | ------------------------------- |
| `hallucination` | Segments the response into atomic claims; verifies each against tool-returned context.  | `response`, intermediate context |
| `grounding`     | Factuality and consistency against provided context.                                    | `prompt`, `response`, `context`  |
| `safety`        | Policy compliance (PII, hate speech, dangerous content, harassment, sexual).            | `prompt`, `response`            |

### Accessing predefined metrics

```python
from agentplatform import types

# Via RubricMetric (preferred) — uppercase enum, unversioned
metric = types.RubricMetric.MULTI_TURN_TRAJECTORY_QUALITY

# Via PrebuiltMetric (alias — identical behavior)
metric = types.PrebuiltMetric.MULTI_TURN_TRAJECTORY_QUALITY

# Pin to a specific version (only when needed for reproducibility)
metric = types.RubricMetric.GENERAL_QUALITY(version="v2")
```

## Computation-Based Metrics

No LLM judge. Deterministic comparison of `response` vs `reference`. Use only
when you have exact reference text to compare against; for agent eval prefer
the rubric-based metrics above.

| Metric ID                  | What it measures                            | Notes          |
| -------------------------- | ------------------------------------------- | -------------- |
| `bleu`                     | BLEU score (translation/generation).        | Standard BLEU  |
| `rouge_1`                  | ROUGE-1 (unigram overlap).                  | Summarization  |
| `tool_parameter_key_match` | Whether tool parameter keys match reference. | Agent evals    |
| `tool_parameter_kv_match`  | Whether tool parameter key-value pairs match reference. | Agent evals    |

```python
metric = types.Metric(name="bleu")
metric = types.Metric(name="tool_parameter_kv_match")
```

## Translation Metrics

| Metric ID | Default version      | Notes                                   |
| --------- | -------------------- | --------------------------------------- |
| `comet`   | `COMET_22_SRC_REF`   | Requires `prompt` (source), `response`, `reference` |
| `metricx` | `METRICX_24_SRC_REF` | Requires `prompt` (source), `response`, `reference` |

## Multimodal Metrics

| Metric ID            | What it measures      | Required fields |
| -------------------- | --------------------- | --------------- |
| `gecko_text2image_v1` | Text-to-image quality | image content   |
| `gecko_text2video_v1` | Text-to-video quality | video content   |

## RubricMetric / PrebuiltMetric (Enum Access)

`RubricMetric.<NAME>` resolves first against the API predefined list, then
falls back to GCS-hosted LLM metric recipes. The names below are the
**uppercase enum form** of the unversioned IDs above; pass them as
`types.RubricMetric.<NAME>`.

| Property                        | Resolution           |
| ------------------------------- | -------------------- |
| `MULTI_TURN_TASK_SUCCESS`       | API predefined       |
| `MULTI_TURN_TRAJECTORY_QUALITY` | API predefined       |
| `MULTI_TURN_TOOL_USE_QUALITY`   | API predefined       |
| `MULTI_TURN_GENERAL_QUALITY`    | API predefined       |
| `MULTI_TURN_TEXT_QUALITY`       | API predefined       |
| `FINAL_RESPONSE_QUALITY`        | API predefined       |
| `FINAL_RESPONSE_MATCH`          | API predefined (v2)  |
| `FINAL_RESPONSE_REFERENCE_FREE` | API predefined       |
| `TOOL_USE_QUALITY`              | API predefined       |
| `GENERAL_QUALITY`               | API predefined       |
| `TEXT_QUALITY`                  | API predefined       |
| `INSTRUCTION_FOLLOWING`         | API predefined       |
| `HALLUCINATION`                 | API predefined       |
| `SAFETY`                        | API predefined       |

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. Runs with the calling
process's privileges, so only use trusted code.

```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 an Agent Platform 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`, `grounding`, `final_response_quality`, `safety`       |
| **Tool-use agent (multi-turn)** | `multi_turn_task_success`, `multi_turn_tool_use_quality`, `multi_turn_trajectory_quality` |
| **Tool-use agent (single-turn)** | `tool_use_quality`, `final_response_quality`                        |
| **Multi-turn conversational** | `multi_turn_general_quality`, `multi_turn_text_quality`, `safety`      |
| **Goal-oriented agent**       | `multi_turn_task_success`, `final_response_quality`                    |
| **Single-turn model eval**    | `general_quality`, `text_quality`, `instruction_following`             |
| **Translation**               | `comet`, `metricx`                                                     |
| **Code generation**           | Custom `CodeExecutionMetric` + `instruction_following`                 |

## 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 (`bleu`, `rouge_1`, etc.)
5.  Name in translation metrics (`comet`, `metricx`)
6.  Name in predefined API metrics (`general_quality`, `multi_turn_task_success`, etc.)
7.  `LLMMetric` with `prompt_template` (custom LLM judge)

````


### `references/sdk_patterns.md`

````markdown
# Agent Platform Evaluation SDK Patterns

Code patterns for common evaluation scenarios using `agentplatform._genai.evals`.

## Initialization

```python
import agentplatform
from agentplatform import types
from google.genai import types as genai_types

client = agentplatform.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 agentplatform._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
```

````


### `scripts/compare_results.py`

```
#!/usr/bin/env python3
"""Diff two Agent Platform Eval result JSON files side by side.

Use in Stage 5 of the Quality Flywheel (Optimize & Iterate) to confirm a fix
improved the target metric without regressing others.

Reads two JSON files produced by ``result.model_dump_json()``, joins summary
metrics by metric_name, and prints baseline vs. candidate scores with a delta.

Usage:
  python compare_results.py --baseline baseline.json --candidate candidate.json
  python compare_results.py -b baseline.json -c candidate.json --threshold 0.05
  python compare_results.py -b baseline.json -c candidate.json --json

Exit codes: 0 = candidate >= baseline on every metric (within threshold),
            1 = at least one metric regressed beyond the threshold.
"""

import argparse
import json
import sys
from typing import Any


def _summary_by_name(result: dict[str, Any]) -> dict[str, dict[str, Any]]:
    """Index summary metrics by metric_name."""
    out = {}
    for s in result.get("summary_metrics") or []:
        name = s.get("metric_name") or s.get("metricName")
        if name:
            out[name] = {
                "mean_score": s.get("mean_score") or s.get("meanScore"),
                "pass_rate": s.get("pass_rate") or s.get("passRate"),
                "num_cases": s.get("num_cases") or s.get("numCases"),
            }
    return out


def _delta(a: float | None, b: float | None) -> float | None:
    """Compute b - a, or None if either input is missing."""
    if a is None or b is None:
        return None
    return b - a


def _format_signed(x: float | None) -> str:
    """Render a delta with explicit sign and 4 decimal places."""
    if x is None:
        return "—"
    return f"{x:+.4f}"


def compare(
    baseline: dict[str, Any],
    candidate: dict[str, Any],
    threshold: float = 0.0,
) -> tuple[list[dict[str, Any]], bool]:
    """Return per-metric diff rows and whether the candidate is safe to ship.

    A row regresses when (candidate - baseline) < -threshold on either
    mean_score or pass_rate. Threshold defaults to 0 (any drop is a regression).
    """
    b = _summary_by_name(baseline)
    c = _summary_by_name(candidate)
    metric_names = sorted(set(b) | set(c))

    rows = []
    regressed = False
    for name in metric_names:
        bv = b.get(name, {})
        cv = c.get(name, {})
        d_mean = _delta(bv.get("mean_score"), cv.get("mean_score"))
        d_pass = _delta(bv.get("pass_rate"), cv.get("pass_rate"))
        row = {
            "metric_name": name,
            "baseline_mean": bv.get("mean_score"),
            "candidate_mean": cv.get("mean_score"),
            "delta_mean": d_mean,
            "baseline_pass": bv.get("pass_rate"),
            "candidate_pass": cv.get("pass_rate"),
            "delta_pass": d_pass,
        }
        is_regression = (d_mean is not None and d_mean < -threshold) or (
            d_pass is not None and d_pass < -threshold
        )
        row["regressed"] = is_regression
        if is_regression:
            regressed = True
        rows.append(row)
    return rows, not regressed


def _format_table(rows: list[dict[str, Any]]) -> str:
    """Render the diff as a fixed-width text table."""
    if not rows:
        return "(no metrics in either result)\n"
    header = [
        "metric_name",
        "base_mean",
        "cand_mean",
        "Δmean",
        "base_pass",
        "cand_pass",
        "Δpass",
        "regressed",
    ]

    def _cell(r, key):
        v = r.get(key)
        if isinstance(v, bool):
            return "YES" if v else ""
        if isinstance(v, float):
            return (
                f"{v:.4f}"
                if key not in ("delta_mean", "delta_pass")
                else _format_signed(v)
            )
        if v is None:
            return "—"
        return str(v)

    name_map = {
        "metric_name": "metric_name",
        "base_mean": "baseline_mean",
        "cand_mean": "candidate_mean",
        "Δmean": "delta_mean",
        "base_pass": "baseline_pass",
        "cand_pass": "candidate_pass",
        "Δpass": "delta_pass",
        "regressed": "regressed",
    }
    widths = {}
    for col in header:
        cells = [_cell(r, name_map[col]) for r in rows]
        widths[col] = max(len(col), *(len(c) for c in cells))

    line1 = "  ".join(c.ljust(widths[c]) for c in header)
    line2 = "  ".join("-" * widths[c] for c in header)
    body = "\n".join(
        "  ".join(_cell(r, name_map[c]).ljust(widths[c]) for c in header) for r in rows
    )
    return f"{line1}\n{line2}\n{body}\n"


def main():
    parser = argparse.ArgumentParser(
        description="Diff two Agent Platform Eval result JSON files side by side."
    )
    parser.add_argument(
        "--baseline",
        "-b",
        required=True,
        help="Path to the baseline result JSON.",
    )
    parser.add_argument(
        "--candidate",
        "-c",
        required=True,
        help="Path to the candidate result JSON (after the fix).",
    )
    parser.add_argument(
        "--threshold",
        "-t",
        type=float,
        default=0.0,
        help="A candidate metric counts as regressed only if it drops more"
        " than this much vs. baseline. Default 0.0 (any drop regresses).",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Emit the diff as JSON instead of a text table.",
    )
    args = parser.parse_args()

    try:
        with open(args.baseline) as f:
            baseline = json.load(f)
        with open(args.candidate) as f:
            candidate = json.load(f)
    except FileNotFoundError as e:
        print(f"ERROR: {e}", file=sys.stderr)
        sys.exit(1)
    except json.JSONDecodeError as e:
        print(f"ERROR: Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)

    rows, ok = compare(baseline, candidate, threshold=args.threshold)

    if args.json:
        print(json.dumps({"rows": rows, "no_regression": ok}, indent=2))
    else:
        print(_format_table(rows))
        if not ok:
            print(
                f"REGRESSION: at least one metric dropped by more than"
                f" {args.threshold:+.4f} vs. baseline.",
                file=sys.stderr,
            )

    sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()

```


### `scripts/endpoint_evaluation.py`

```
"""Evaluate an Agent Platform endpoint using Agent Platform Evaluation."""

import argparse
import json
import subprocess

from agentplatform import Client
import pandas as pd
import requests


def run_inference(prompt: str, endpoint_url: str, token: str) -> str:
  """Runs inference on a prompt.

  Args:
      prompt: The prompt to send to the model.
      endpoint_url: The URL of the Agent Platform endpoint.
      token: The gcloud access token.

  Returns:
      The model's response.
  """
  headers = {
      "Authorization": f"Bearer {token}",
      "Content-Type": "application/json",
  }

  payload = {
      "instances": [{
          "@requestFormat": "chatCompletions",
          "messages": [{"role": "user", "content": prompt}],
          "max_tokens": 100,
      }]
  }

  try:
    with requests.post(
        endpoint_url,
        headers=headers,
        data=json.dumps(payload),
        timeout=60,
    ) as response:
      response.raise_for_status()
      json_response = response.json()

      try:
        preds = json_response.get("predictions")
        if not preds:
          raise ValueError(f"No predictions in response: {json_response}")

        if isinstance(preds, dict):
          return str(preds["choices"][0]["message"]["content"])

        if not isinstance(preds, list):
          raise ValueError(f"Unexpected preds type: {type(preds)}")

        first_pred = preds[0]
        if isinstance(first_pred, list):
          if not first_pred:
            raise ValueError(f"Empty inner prediction list: {json_response}")
          return str(first_pred[0]["message"]["content"])
        elif isinstance(first_pred, dict):
          return str(first_pred["choices"][0]["message"]["content"])
        raise ValueError(
            f"Unable to parse prediction response: {json_response}"
        )
      except (KeyError, IndexError, TypeError) as e:
        raise ValueError(
            f"Unable to parse prediction response: {json_response}"
        ) from e

  except requests.exceptions.RequestException as e:
    raise RuntimeError("Error calling Agent Platform endpoint.") from e


def run_evaluation(
    *,
    project_id: str,
    location: str,
    endpoint_id: str,
    dataset: str,
    dedicated_endpoint_dns: str | None = None,
    metrics: list[str],
) -> pd.DataFrame:
  """Main function to run evaluation on an Agent Platform endpoint.

  Args:
      project_id: The project ID.
      location: The location ID.
      endpoint_id: The endpoint ID.
      dataset: GCS path to the evaluation dataset (.jsonl).
      dedicated_endpoint_dns: Dedicated DNS for the endpoint.
      metrics: List of metrics to use for evaluation.

  Returns:
      Evaluation result from Agent Platform Evaluation.
  """
  client = Client(project=project_id, location=location)

  print(f"--- Running Evaluation for Endpoint: {endpoint_id} in {location} ---")

  if dedicated_endpoint_dns is not None:
    endpoint_url = f"https://{dedicated_endpoint_dns}/v1/projects/{project_id}/locations/{location}/endpoints/{endpoint_id}:predict"
  else:
    endpoint_url = f"https://{location}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{location}/endpoints/{endpoint_id}:predict"
  try:
    token = subprocess.run(
        ["gcloud", "auth", "print-access-token"],
        capture_output=True,
        text=True,
        check=True,
    ).stdout.strip()
  except subprocess.CalledProcessError as e:
    raise RuntimeError("Error getting gcloud access token.") from e

  try:
    jsonl_content = subprocess.run(
        ["gsutil", "cat", dataset],
        capture_output=True,
        text=True,
        check=True,
    ).stdout
  except subprocess.CalledProcessError as e:
    raise RuntimeError(f"Error reading dataset from {dataset}.") from e

  dataset_list = []
  for line in jsonl_content.strip().split("\n"):
    if line:
      dataset_list.append(json.loads(line))

  df = pd.DataFrame(dataset_list)
  if "prompt" not in df.columns:
    raise ValueError("Dataset missing 'prompt' column in JSONL file.")

  df["response"] = df["prompt"].apply(
      lambda prompt: run_inference(prompt, endpoint_url, token)
  )

  eval_result = client.evals.evaluate(
      dataset=df,
      prompt_column="prompt",
      response_column="response",
      metrics=metrics,
  )

  return eval_result


if __name__ == "__main__":
  parser = argparse.ArgumentParser(
      description=(
          "Evaluate an Agent Platform endpoint using Agent Platform Evaluation."
      )
  )
  parser.add_argument(
      "--project_id", type=str, required=True, help="Project ID."
  )
  parser.add_argument("--location", type=str, required=True, help="Location.")
  parser.add_argument(
      "--endpoint_id", type=str, required=True, help="Endpoint ID."
  )
  parser.add_argument(
      "--dedicated_endpoint_dns",
      type=str,
      help="Dedicated endpoint DNS.",
  )
  parser.add_argument(
      "--dataset",
      type=str,
      required=True,
      help="GCS path to evaluation dataset.",
  )
  parser.add_argument(
      "--metrics",
      type=str,
      nargs="+",
      default=["GENERAL_QUALITY"],
      help="List of metrics to use for evaluation.",
  )
  args = parser.parse_args()

  result = run_evaluation(**vars(args))

  print(f"""{"-" * 20}
Evaluation Summary Metrics:
{result.summary_metrics}
{"-" * 20}
Evaluation Details (first 5 rows):
{result.metrics_table.head(5)}""")

```


### `scripts/inspect_results.py`

```
#!/usr/bin/env python3
"""Print summary metrics and per-case scores from an Agent Platform Eval result.

Use after ``client.evals.evaluate(...)`` returns to read scores. The field
paths (``summary_metrics``, ``eval_case_results[].response_candidate_results``)
are deep and easy to get wrong; this helper renders the standard view.

Two input shapes are accepted:

1.  A JSON file produced by ``result.model_dump_json()`` (preferred — save
    the result to disk so it can be diffed later with ``compare_results.py``).
2.  An in-process result object, when imported as a library:

      from inspect_results import render
      render(result)

Usage:
  python inspect_results.py --result result.json
  python inspect_results.py --result result.json --failing-only
  python inspect_results.py --result result.json --metric multi_turn_task_success
  python inspect_results.py --result result.json --save-html report.html


Exit codes: 0 = at least one case present, 1 = empty / malformed result.
"""

import argparse
import json
import sys
from typing import Any


def _summary_rows(result: dict[str, Any]) -> list[dict[str, Any]]:
    """Extract summary metric rows from a serialized result."""
    rows = []
    for s in result.get("summary_metrics") or []:
        rows.append({
            "metric_name": s.get("metric_name") or s.get("metricName"),
            "mean_score": s.get("mean_score") or s.get("meanScore"),
            "pass_rate": s.get("pass_rate") or s.get("passRate"),
            "num_cases": s.get("num_cases") or s.get("numCases"),
        })
    return rows


def _case_rows(
    result: dict[str, Any],
    metric_filter: str | None,
    failing_only: bool,
) -> list[dict[str, Any]]:
    """Extract per-case metric rows from a serialized result."""
    rows = []
    for case in result.get("eval_case_results") or []:
        case_id = case.get("eval_case_id") or case.get("evalCaseId") or "?"
        for cand in case.get("response_candidate_results") or []:
            metric_results = cand.get("metric_results") or {}
            for name, r in metric_results.items():
                if metric_filter and name != metric_filter:
                    continue
                score = r.get("score") if isinstance(r, dict) else None
                if failing_only and score is not None and score >= 1.0:
                    continue
                rows.append({
                    "case_id": case_id,
                    "metric_name": name,
                    "score": score,
                    "explanation": (
                        (r.get("explanation") or "")[:200]
                        if isinstance(r, dict)
                        else ""
                    ),
                })
    return rows


def _format_table(rows: list[dict[str, Any]], cols: list[str]) -> str:
    """Render rows as a fixed-width text table."""
    if not rows:
        return "(no rows)\n"
    widths = {
        c: max(len(c), *(len(str(r.get(c, ""))) for r in rows)) for c in cols
    }
    header = "  ".join(c.ljust(widths[c]) for c in cols)
    sep = "  ".join("-" * widths[c] for c in cols)
    body = "\n".join(
        "  ".join(str(r.get(c, "")).ljust(widths[c]) for c in cols) for r in rows
    )
    return f"{header}\n{sep}\n{body}\n"


def render(
    result: dict[str, Any] | Any,
    metric_filter: str | None = None,
    failing_only: bool = False,
) -> str:
    """Render summary and per-case tables. Accepts dict or SDK result object."""
    if not isinstance(result, dict):
        if hasattr(result, "model_dump"):
            result = result.model_dump()
        else:
            result = json.loads(json.dumps(result, default=lambda o: o.__dict__))

    out = []
    out.append("=== Summary Metrics ===")
    out.append(
        _format_table(
            _summary_rows(result),
            ["metric_name", "mean_score", "pass_rate", "num_cases"],
        )
    )
    out.append("=== Per-Case Scores ===")
    if failing_only:
        out.append("(failing cases only, score < 1.0)")
    if metric_filter:
        out.append(f"(filtered to metric: {metric_filter})")
    out.append(
        _format_table(
            _case_rows(result, metric_filter, failing_only),
            ["case_id", "metric_name", "score", "explanation"],
        )
    )
    return "\n".join(out)


def main():
    parser = argparse.ArgumentParser(
        description=(
            "Print summary metrics and per-case scores from an Agent"
            " Platform Eval result JSON file."
        )
    )
    parser.add_argument(
        "--result",
        "-r",
        required=True,
        help="Path to a JSON file produced by result.model_dump_json().",
    )
    parser.add_argument(
        "--metric",
        "-m",
        help="Only show per-case scores for this metric name.",
    )
    parser.add_argument(
        "--failing-only",
        action="store_true",
        help="Only show per-case scores with score < 1.0.",
    )
    parser.add_argument(
        "--save-html",
        "-o",
        dest="html_path",
        help="Also render the result as a browsable HTML report to this path.",
    )
    args = parser.parse_args()

    try:
        with open(args.result) as f:
            result = json.load(f)
    except FileNotFoundError:
        print(f"ERROR: File not found: {args.result}", file=sys.stderr)
        sys.exit(1)
    except json.JSONDecodeError as e:
        print(f"ERROR: Invalid JSON in {args.result}: {e}", file=sys.stderr)
        sys.exit(1)

    if not result.get("summary_metrics") and not result.get("eval_case_results"):
        print(
            "ERROR: result has no summary_metrics or eval_case_results."
            " Did you save result.model_dump_json() to this file?",
            file=sys.stderr,
        )
        sys.exit(1)

    print(
        render(result, metric_filter=args.metric, failing_only=args.failing_only)
    )

    if args.html_path:
        save_html(result, args.html_path)


def save_html(result: dict[str, Any], html_path: str) -> None:
    """Render the result as a standalone HTML report and write to ``html_path``."""
    try:
        from agentplatform._genai import _evals_visualization
    except ImportError as e:
        print(
            "ERROR: --save-html requires the agentplatform SDK to be installed"
            f" (`pip install google-cloud-aiplatform`): {e}",
            file=sys.stderr,
        )
        sys.exit(1)

    try:
        html_content = _evals_visualization.get_evaluation_html(json.dumps(result))
    except Exception as e:
        print(f"ERROR: Failed to render HTML report: {e}", file=sys.stderr)
        sys.exit(1)

    if not html_content:
        print(
            "ERROR: Agent Platform Eval SDK returned empty HTML — the result"
            " may be missing fields the visualizer requires.",
            file=sys.stderr,
        )
        sys.exit(1)

    with open(html_path, "w", encoding="utf-8") as f:
        f.write(str(html_content))
    print(f"Saved HTML report to {html_path}")


if __name__ == "__main__":
    main()

```


### `scripts/maas_evaluation.py`

```
"""Evaluate a MaaS model using Agent Platform Evaluation."""

import argparse
from agentplatform import Client
import pandas as pd


def run_evaluation(
    *,
    project_id: str,
    location: str,
    model_id: str,
    dataset: str,
    metrics: list[str],
) -> pd.DataFrame:
  """Main function to run evaluation on a MaaS model.

  Args:
      project_id: The project ID.
      location: The location ID.
      model_id: The MaaS model ID.
      dataset: GCS path to the evaluation dataset (.jsonl).
      metrics: List of metrics to use for evaluation.

  Returns:
      Evaluation result from Agent Platform Evaluation.
  """
  client = Client(project=project_id, location=location)

  print(f"--- Running Inference for MaaS Model: {model_id} in {location} ---")

  try:
    maas_responses = client.evals.run_inference(
        model=model_id,
        src=dataset,
    )
  except Exception as e:
    raise RuntimeError(
        f"Error running inference for MaaS model: {model_id}"
    ) from e

  print(f"\n--- Running Evaluation for MaaS Model: {model_id} ---")
  try:
    maas_eval_result = client.evals.evaluate(
        dataset=maas_responses,
        metrics=metrics,
    )
  except Exception as e:
    raise RuntimeError(
        f"Error running evaluation for MaaS model: {model_id}"
    ) from e

  return maas_eval_result


if __name__ == "__main__":
  parser = argparse.ArgumentParser(
      description="Evaluate a MaaS model using Agent Platform Evaluation."
  )
  parser.add_argument(
      "--project_id", type=str, required=True, help="Project ID."
  )
  parser.add_argument("--location", type=str, required=True, help="Location.")
  parser.add_argument(
      "--model_id", type=str, required=True, help="MaaS Model ID."
  )
  parser.add_argument(
      "--dataset",
      type=str,
      required=True,
      help="GCS path to evaluation dataset.",
  )
  parser.add_argument(
      "--metrics",
      type=str,
      nargs="+",
      default=["GENERAL_QUALITY"],
      help="List of metrics to use for evaluation.",
  )
  args = parser.parse_args()

  results = run_evaluation(**vars(args))

  print(f"""{"-" * 20}
Evaluation Summary Metrics:
{results.summary_metrics}
{"-" * 20}
Evaluation Details (first 5 rows):
{results.metrics_table.head(5)}
{"-" * 20}
Evaluation complete. Displaying report:""")
  results.show()

```


### `scripts/parse_adk_traces.py`

```
#!/usr/bin/env python3
"""Parse ADK session traces into Agent Platform 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 Agent Platform 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()

```


### `scripts/render_html_report.py`

```
#!/usr/bin/env python3
"""Render Agent Platform Eval HTML reports from saved result / loss-cluster JSON.

Two report types are supported:

  * ``evaluation`` — rendered from a result JSON produced by
    ``result.model_dump_json()``
  * ``loss-analysis`` — rendered from a loss-clusters JSON produced by
    ``response.model_dump_json()``

Use when you have a saved JSON artifact and want a browsable HTML report
to share with the user, link in a PR description, or attach to a bug.

Usage:
  python render_html_report.py --input result.json --type evaluation --output report.html
  python render_html_report.py --input clusters.json --type loss-analysis --output clusters.html

Exit codes: 0 = HTML written, 1 = SDK import / render failure.
"""

import argparse
import json
import sys


def render_evaluation_html(result_json_str: str) -> str:
    """Render an EvaluationResult JSON string as standalone HTML."""
    from agentplatform._genai import _evals_visualization

    html = _evals_visualization.get_evaluation_html(result_json_str)
    if not html:
        raise RuntimeError(
            "Agent Platform Eval SDK returned empty HTML — the input may be"
            " missing fields the evaluation visualizer requires."
        )
    return str(html)


def render_loss_analysis_html(response_json_str: str) -> str:
    """Render a loss-clusters response JSON string as standalone HTML."""
    from agentplatform._genai import _evals_visualization

    html = _evals_visualization.get_loss_analysis_html(response_json_str)
    if not html:
        raise RuntimeError(
            "Agent Platform Eval SDK returned empty HTML — the input may be"
            " missing fields the loss-analysis visualizer requires."
        )
    return str(html)


def main():
    parser = argparse.ArgumentParser(
        description="Render Agent Platform Eval HTML reports from saved JSON artifacts."
    )
    parser.add_argument(
        "--input",
        "-i",
        required=True,
        help="Path to the input JSON file (result or loss-clusters response).",
    )
    parser.add_argument(
        "--type",
        "-t",
        choices=["evaluation", "loss-analysis"],
        required=True,
        help=(
            "evaluation = client.evals.evaluate result;"
            " loss-analysis = client.evals.generate_loss_clusters response."
        ),
    )
    parser.add_argument(
        "--output",
        "-o",
        required=True,
        help="Path to write the HTML report to.",
    )
    args = parser.parse_args()

    try:
        with open(args.input) as f:
            content = f.read()
            json.loads(content)
    except FileNotFoundError:
        print(f"ERROR: File not found: {args.input}", file=sys.stderr)
        sys.exit(1)
    except json.JSONDecodeError as e:
        print(f"ERROR: Invalid JSON in {args.input}: {e}", file=sys.stderr)
        sys.exit(1)

    try:
        if args.type == "evaluation":
            html = render_evaluation_html(content)
        else:
            html = render_loss_analysis_html(content)
    except ImportError as e:
        print(
            "ERROR: requires the agentplatform SDK to be installed"
            f" (`pip install google-cloud-aiplatform`): {e}",
            file=sys.stderr,
        )
        sys.exit(1)
    except Exception as e:
        print(f"ERROR: Failed to render HTML: {e}", file=sys.stderr)
        sys.exit(1)

    with open(args.output, "w", encoding="utf-8") as f:
        f.write(html)
    print(f"Saved {args.type} HTML report to {args.output}")


if __name__ == "__main__":
    main()

```


### `scripts/validate_dataset.py`

```
#!/usr/bin/env python3
"""Validate an evaluation dataset for Agent Platform 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,multi_turn_task_success

Metric names are the unversioned form (the SDK resolves them to the latest
version). Pinned versions (e.g., general_quality_v1) are also accepted.

Exit codes: 0 = valid, 1 = invalid (with specific errors).
"""

import argparse
import json
import sys
from typing import Any

_SINGLE_TURN_METRICS = frozenset({
    "general_quality",
    "text_quality",
    "instruction_following",
    "grounding",
    "safety",
    "hallucination",
    "tool_use_quality",
    "final_response_reference_free",
    "final_response_quality",
})
_MULTI_TURN_METRICS = frozenset({
    "multi_turn_tool_use_quality",
    "multi_turn_trajectory_quality",
    "multi_turn_task_success",
    "multi_turn_general_quality",
    "multi_turn_text_quality",
})
_COMPUTATION_METRICS = frozenset({
    "bleu",
    "rouge_1",
    "rouge_l_sum",
})


_VALID_ROLES = frozenset({"user", "model", "tool"})


def _canonical_metric(name: str) -> str:
  """Strip a trailing _v<N> version suffix so versioned IDs match the sets."""
  parts = name.rsplit("_v", 1)
  if len(parts) == 2 and parts[1].isdigit():
    return parts[0]
  return name


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' (Agent Platform 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}]"))

    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_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",
        )
    )

  if has_agent_data:
    ad = case.get("agent_data") or case.get("agentData")
    errors.extend(_validate_agent_data(ad, f"{path}.agent_data"))

  if metrics:
    for raw_metric in metrics:
      metric = _canonical_metric(raw_metric)
      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 _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 Agent Platform 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)

  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()

```