spb/wp20_uqo

1#!/usr/bin/env python32"""Smoke test: a few direct (non-batch) calls per tier to validate schemas and measure token usage.3Usage: WP20_ENV_FILE=/path/to/.env .venv/bin/python scripts/03_smoke_test.py [n_listings] [n_images]4"""5import json6import sys7from pathlib import Path89import pandas as pd1011sys.path.insert(0, str(Path(__file__).resolve().parents[1]))12from src.extraction import client as C  # noqa: E40213from src.extraction.textprep import build_text  # noqa: E40214from src.feature_schema.schema import IMAGE_SCHEMA, TEXT_SCHEMA  # noqa: E4021516ROOT = Path(__file__).resolve().parents[1]17n_l = int(sys.argv[1]) if len(sys.argv) > 1 else 218n_i = int(sys.argv[2]) if len(sys.argv) > 2 else 31920smp = pd.read_parquet(ROOT / "data" / "processed" / "sample.parquet").head(n_l)21usage = []22for r in smp.itertuples():23    print(f"\n=== {r.uid} | {r.source} | {r.roll_municipalite} | built {r.roll_year_built} | {r.roll_floor_area_m2} m² | ${r.price:,.0f}")24    for tier in ["haiku", "sonnet"]:25        for i, url in enumerate(list(r.image_urls_used)[:n_i]):26            out = C.direct(C.image_request(tier, url, IMAGE_SCHEMA), tag="smoke")27            if out["ok"]:28                d = out["data"]29                print(f"  [{tier}] img{i} {d['scene']:24s} q{d['photo_quality']} ext{d['exterior_condition']} int{d['interior_condition']} "30                      f"kit={d['kitchen_grade']} clad={d['cladding']} conf={d['confidence']:.2f} | {d['evidence'][:70]} | tok {out['usage']}")31                usage.append({"tier": tier, "kind": "image", **out["usage"]})32            else:33                print(f"  [{tier}] img{i} FAILED {out.get('error')} {out.get('stop_reason')}")34    text, stats = build_text(r._asdict())35    for tier in ["haiku", "sonnet"]:36        out = C.direct(C.text_request(tier, text, TEXT_SCHEMA), tag="smoke")37        if out["ok"]:38            d = out["data"]39            print(f"  [{tier}] TEXT clad={d['cladding']} roof={d['roof_covering']} kit={d['kitchen_grade']} cond={d['condition_claim']} "40                  f"reno={d['renovation_claim']} yb={d['year_built_claim']} area={d['living_area_claim_sqft']} price_leak={d['price_or_assessment_mentioned']} | tok {out['usage']} | mask {stats}")41            usage.append({"tier": tier, "kind": "text", **out["usage"]})42        else:43            print(f"  [{tier}] TEXT FAILED {out.get('error')}")44u = pd.DataFrame(usage)45print("\nmean tokens per call:\n", u.groupby(["tier", "kind"])[["input_tokens", "output_tokens"]].mean().round(0))46