Evaluating Google ADK Agents with W&B Weave for reliable insurance workflows

[
  {
    "eval_id": "case001",
    "conversation": [
      {
        "role": "user",
        "user_content": {
          "parts": [
            {"text": "I'd like to file a claim for my lost luggage."}
          ]
        }
      },
      {
        "role": "agent",
        "final_response": {
          "parts": [
            {"text": "Sure, I can help you file a claim for your lost luggage. Can you provide more details about your trip?"}
          ]
        },
        "intermediate_data": {
          "tool_uses": [
            {
              "name": "claim_intake",
              "args": {
                "type": "lost_luggage"
              }
            }
          ]
        }
      }
    ],
    "session_input": {
      "user_id": "U123",
      "app_name": "insurance_demo"
    }
  },
  {
    "eval_id": "case002",
    "conversation": [
      {
        "role": "user",
        "user_content": {
          "parts": [
            {"text": "When is my next premium payment due?"}
          ]
        }
      },
      {
        "role": "agent",
        "final_response": {
          "parts": [
            {"text": "Your next payment is $85.35, due July 15th."}
          ]
        },
        "intermediate_data": {
          "tool_uses": [
            {
              "name": "fetch_payment_due",
              "args": {
                "policy_id": "P10005"
              }
            }
          ]
        }
      }
    ]
  }
]

{
  "policy_lookup": [
    {"policy_id":"P100001","customer_id":"C20001","type":"Auto","start":"2022-05-01","end":"2023-05-01","coverage":{"liability":50000,"collision":1000,"deductible":500},"exclusions":["rental cars"],"status":"active"},
    ...
  ],
  "claims_status_checker": [
    {"claim_id":"CL5001","policy_id":"P100001","date_filed":"2023-04-15","status":"pending","type":"collision","adjuster":"Alex Wu","next_step":"awaiting police report"},
    ...
  ],
  "coverage_calculator": [
    {"policy_id":"P100001","option":"raise liability","change":25000,"new_premium":103.50,"new_deductible":500},
    ...
  ],
  "premium_payment_system": [
    {"invoice_id":"INV3001","policy_id":"P100001","due_date":"2023-05-05","amount_due":120.00,"status":"unpaid"},
    ...
  ],
  "appointment_schedule_checker": [
    {"appointment_id":"A4101","customer_id":"C20001","type":"callback","date":"2023-05-06T10:00:00","status":"scheduled"},
    ...
  ],
  "faq_search": [
    {"faq_id":"F001","question":"How can I file a claim?","answer":"You can file a claim online, by phone, or with your agent. Have your policy and event details ready."},
    ...
  ],
  "find_nearby_repair_shop": [
    {"shop_id":"R01","name":"AutoFix Pros","zip":"60636","approved":true,"phone":"773-555-1010"},
    ...
  ],
  "insurance_quote_data": [
    {"quote_id":"Q3001","customer_id":"C20001","coverage":"auto-basic","premium":109.50,"deductible":500,"valid_until":"2023-06-01"},
    ...
  ],
  "customer_profile_lookup": [
    {"customer_id":"C20001","name":"Jane Smith","email":"jsmith@email.com","phone":"555-4410","dob":"1982-03-01"},
    ...
  ]
}

import json
from typing import Any, Dict, List, Optional

# ---- Load data ----
with open("insurancedata.json", "r", encoding="utf-8") as f:
    INSURANCE_DATA = json.load(f)

def policy_lookup(policy_id: str) -> Optional[Dict[str, Any]]:
    return next((p for p in INSURANCE_DATA['policy_lookup'] if p['policy_id'] == policy_id), None)

def customer_policies(customer_id: str, status: Optional[str] = None) -> List[Dict[str, Any]]:
    return [
        p for p in INSURANCE_DATA['policy_lookup']
        if p['customer_id'] == customer_id and (status is None or p['status'] == status)
    ]

def claims_status_checker(policy_id: str, status: Optional[str] = None) -> List[Dict[str, Any]]:
    return [
        c for c in INSURANCE_DATA['claims_status_checker']
        if c['policy_id'] == policy_id and (status is None or c['status'] == status)
    ]

def coverage_calculator(policy_id: str, option: Optional[str] = None) -> List[Dict[str, Any]]:
    return [
        cc for cc in INSURANCE_DATA['coverage_calculator']
        if cc['policy_id'] == policy_id and (option is None or cc['option'] == option)
    ]

def premium_payment_system(policy_id: str) -> List[Dict[str, Any]]:
    return [
        i for i in INSURANCE_DATA['premium_payment_system']
        if i['policy_id'] == policy_id
    ]

def appointment_schedule_checker(customer_id: str) -> List[Dict[str, Any]]:
    return [
        a for a in INSURANCE_DATA['appointment_schedule_checker']
        if a['customer_id'] == customer_id
    ]

def faq_search(query: str, topk: int = 1) -> List[Dict[str, Any]]:
    matches = [f for f in INSURANCE_DATA['faq_search'] if query.lower() in f['question'].lower()]
    return matches[:topk]

def find_nearby_repair_shop(zip_code: str, approved_only: bool = True, topk: int = 3) -> List[Dict[str, Any]]:
    shops = [s for s in INSURANCE_DATA['find_nearby_repair_shop'] if s['zip'] == zip_code]
    if approved_only:
        shops = [s for s in shops if s['approved']]
    return shops[:topk]

def insurance_quote_data(customer_id: str) -> List[Dict[str, Any]]:
    return [
        q for q in INSURANCE_DATA['insurance_quote_data']
        if q['customer_id'] == customer_id
    ]


agent_prompt = """
policy_lookup(policy_id)
# Use when you need all details for a specific policy using its policy_id (such as viewing coverages, start/end, type, etc).
print(policy_lookup("P100001"))
# Returns:
# {'policy_id': 'P100001', 'customer_id': 'C20001', ..., 'status': 'active'}

customer_policies(customer_id, status=None)
# Use when you want all policies (optionally filtered by status, e.g. 'active') for a given customer_id.
print(customer_policies("C20001", status="active"))
# Returns:
# [{'policy_id': 'P100001', ...}]

claims_status_checker(policy_id, status=None)
# Use when you need all claims for a given policy. Optionally filter by claim status ('pending', 'approved', etc).
print(claims_status_checker("P100001", status="pending"))
# Returns:
# [{'claim_id': 'CL5001', 'policy_id': 'P100001', 'status': 'pending', ...}]

coverage_calculator(policy_id, option=None)
# Use when you want to see alternate coverage/premium options for a policy, like how much premium changes if you change something.
print(coverage_calculator("P100001"))
# Returns:
# [{'policy_id': 'P100001', 'option': 'raise liability', 'change': 25000, 'new_premium': 103.50, 'new_deductible': 500}]

premium_payment_system(policy_id)
# Use when you need to see billing or payment info for a policy (invoices, due dates, amounts).
print(premium_payment_system("P100001"))
# Returns:
# [{'invoice_id': 'INV3001', 'policy_id': 'P100001', 'due_date': '2023-05-05', 'amount_due': 120.00, 'status': 'unpaid'}]

appointment_schedule_checker(customer_id)
# Use to list all scheduled or completed appointments (like adjuster visits or callbacks) for a customer.
print(appointment_schedule_checker("C20001"))
# Returns:
# [{'appointment_id': 'A4101', 'customer_id': 'C20001', 'type': 'callback', 'date': '2023-05-06T10:00:00', 'status': 'scheduled'}]

faq_search(query, topk=1)
# Use to look up answers to common insurance questions by keyword.
print(faq_search("deductible", topk=2))
# Returns:
# [{'faq_id': 'F002', 'question': 'What is a deductible?', 'answer': ...},
#  {'faq_id': 'F007', ...}]

find_nearby_repair_shop(zip_code, approved_only=True, topk=3)
# Use when you want to find repair shops near a ZIP code, optionally only approved ones. Useful for auto claims.
print(find_nearby_repair_shop("60636"))
# Returns:
# [{'shop_id': 'R01', 'name': 'AutoFix Pros', ...},
#  {'shop_id': 'R02', ...}]

insurance_quote_data(customer_id)
# Use when you wish to view all (recent/past) insurance quotes for a customer.
print(insurance_quote_data("C20001"))
# Returns:
# [{'quote_id': 'Q3001', 'customer_id': 'C20001', 'coverage': 'auto-basic', ...}]
"""

from google.adk.agents import Agent
import os

# Configure environment if needed
os.environ["GOOGLE_CLOUD_PROJECT"] = "dsports-6ab79"
os.environ["GOOGLE_CLOUD_LOCATION"] = "us-central1"
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"  # or set as needed

root_agent = Agent(
    model="gemini-2.5-pro",
    name="insurance_agent",
    instruction="""
You are an expert insurance assistant. You can answer customer questions and perform insurance-related tasks
using the available tools. For every question, carefully select and use the most relevant tool(s) from your toolkit.
Never guess—return only what is found in the database or tools. Here are your tool instructions and usage examples:

""" + agent_prompt,
    description="An insurance assistant that can answer any insurance query using the provided tools.",
    tools=[
        policy_lookup,
        customer_policies,
        claims_status_checker,
        coverage_calculator,
        premium_payment_system,
        appointment_schedule_checker,
        faq_search,
        find_nearby_repair_shop,
        insurance_quote_data
    ]
)

import os
import json

# HARDCODE your input file here:
infile = "/path_to/original_insurance_eval_set.evalset.json"

# Create output filename ending with _cleaned.evalset.json
dirname, basename = os.path.split(infile)
if basename.endswith('evalset.json'):
    core = basename[:-len('evalset.json')].rstrip('.-_')
    outbase = f"{core}_cleaned.evalset.json"
else:
    core = os.path.splitext(basename)[0]
    outbase = f"{core}_cleaned.evalset.json"
outfile = os.path.join(dirname, outbase)

with open(infile, "r", encoding="utf-8") as f:
    data = json.load(f)

for case in data.get("eval_cases", []):
    conv = case.get("conversation", [])
    if not conv:
        continue
    # Collect all tool_uses for this case
    merged_tool_uses = []
    for turn in conv:
        merged_tool_uses.extend(turn.get("intermediate_data", {}).get("tool_uses", []))
    # Extract first user message and last model response
    first = conv[0]
    last = conv[-1]
    # Build new single-turn conversation:
    # user_content from first, final_response from last, new intermediate_data
    new_turn = {
        "invocation_id": last.get("invocation_id", first.get("invocation_id")),
        "user_content": first.get("user_content", {}),
        "final_response": last.get("final_response", {}),
        "intermediate_data": {
            "tool_uses": merged_tool_uses,
            "intermediate_responses": []
        },
        "creation_timestamp": last.get("creation_timestamp", first.get("creation_timestamp"))
    }
    case["conversation"] = [new_turn]

with open(outfile, "w", encoding="utf-8") as f:
    json.dump(data, f, indent=2, ensure_ascii=False)

print(f'Pruned/cleaned eval set written as: {outfile}')

import dotenv
import pytest
from google.adk.evaluation.agent_evaluator import AgentEvaluator
from lite_custom_evaluator import LiteAgentEvaluator  # see the github repo for this script

pytest_plugins = ("pytest_asyncio",)

use_weave = True  # Set to True to use LiteAgentEvaluator (with weave), or False to use AgentEvaluator

@pytest.fixture(scope="session", autouse=True)
def load_env():
    dotenv.load_dotenv()

@pytest.mark.asyncio
async def test_all():
    """Test the agent's basic ability on a few examples."""
    agent_name = "seq_agent"
    eval_path = "/Users/brettyoung/Desktop/dev25/tutorials/adk/adk-samples/python/agents/seq_agent/seq_agent/insurance_eval_set_cleaned.evalset.json"

    if use_weave:
        await LiteAgentEvaluator.evaluate(
            agent_name,
            eval_path,
            num_runs=2,
            use_weave=True
        )
    else:
        await AgentEvaluator.evaluate(
            agent_name,
            eval_path,
            num_runs=2
        )

python -m pytest -s eval