Your AI backfill passed the quality check. More than a third of it was wrong.

"""
From hand fixes to an audited AI backfill, on a laptop.

Synthetic data, seeded, so every number below is reproducible.
Runs on DuckDB so the SQL reads the same as it would on Snowflake or BigQuery.
pip install duckdb jsonschema
"""
import hashlib
import random

import duckdb
import jsonschema

random.seed(7)
con = duckdb.connect()

# ---------------------------------------------------------------------------
# AS-IS: 1,000 orders, the city column the way a hand-maintained table ends up.
# ---------------------------------------------------------------------------
CITIES = ["São Paulo", "Campinas", "Santos", "Rio de Janeiro", "Belo Horizonte"]
SP_SPELLINGS = ["São Paulo", "Sao Paulo", "SP", "sao paulo"]   # same city, four ways


def make_order(i):
    true_city = random.choices(CITIES, weights=[50, 15, 10, 15, 10])[0]
    kind = "clean"
    note = f"Entregar na portaria, {true_city}"
    city = random.choice(SP_SPELLINGS) if true_city == "São Paulo" else true_city

    r = random.random()
    if r < 0.06:                                # blank, but the note names the city
        city, kind = None, "blank_clear"
    elif r < 0.09:                              # blank, and the note names two cities
        city, kind = None, "blank_ambiguous"
        other = random.choice([c for c in CITIES if c != true_city])
        note = f"Cliente de {other}, entregar no escritório em {true_city}"
    elif r < 0.10:                              # blank, and the note names no city at all
        city, kind = None, "blank_empty"
        note = "Entregar na portaria"
    return (i, city, note, true_city, kind)


rows = [make_order(i) for i in range(1, 1001)]
con.execute("""CREATE TABLE raw_orders (order_id INT, city VARCHAR, delivery_note VARCHAR,
                                        true_city VARCHAR, kind VARCHAR)""")
con.executemany("INSERT INTO raw_orders VALUES (?, ?, ?, ?, ?)", rows)

print("== AS-IS: the usual checks ==")
print(con.execute("""
    SELECT COUNT(*)                 AS row_count,
           COUNT(DISTINCT order_id) AS distinct_ids,
           COUNT(*) - COUNT(order_id) AS null_ids
    FROM raw_orders
""").fetchall())
print(con.execute("""
    SELECT city, COUNT(*) AS orders FROM raw_orders
    GROUP BY city ORDER BY orders DESC
""").fetchall())

== AS-IS: the usual checks ==
[(1000, 1000, 0)]
[('Campinas', 131), ('sao paulo', 129), ('São Paulo', 123), ('Rio de Janeiro', 123), ('Sao Paulo', 117), ('SP', 115), (None, 111), ('Belo Horizonte', 85), ('Santos', 66)]

-- the rebuild-from-raw logic lives in a view (same SQL as the Python below);
-- the task only materialises it
CREATE OR REPLACE TASK load_orders
  WAREHOUSE = transform_wh
  SCHEDULE = '60 MINUTE'
AS
  CREATE OR REPLACE TABLE orders AS SELECT * FROM orders_clean_v;

ALTER TASK load_orders RESUME;   -- without this line, the task exists and never runs

bq mk \
  --transfer_config \
  --target_dataset=mydataset \
  --display_name='load_orders' \
  --schedule='every 24 hours' \
  --params='{"query":"SELECT * FROM mydataset.orders_clean_v","destination_table_name_template":"orders","write_disposition":"WRITE_TRUNCATE"}' \
  --data_source=scheduled_query \
  --service_account_name=loader@myproject.iam.gserviceaccount.com

# ---------------------------------------------------------------------------
# STEP 1: Automate the load. One code path from raw to clean, no hand edits.
# Corrections arrive as data (a feed table), not as edits to the target.
# ---------------------------------------------------------------------------
con.execute("""CREATE TABLE city_aliases AS SELECT * FROM (VALUES
    ('são paulo','São Paulo'), ('sao paulo','São Paulo'), ('sp','São Paulo'),
    ('campinas','Campinas'), ('santos','Santos'),
    ('rio de janeiro','Rio de Janeiro'), ('belo horizonte','Belo Horizonte')
) t(alias, city)""")
con.execute("CREATE TABLE corrections (order_id INT, city VARCHAR, reason VARCHAR)")
con.execute("INSERT INTO corrections VALUES (42, 'Santos', 'customer called support')")


def load_orders():
    """The whole load. Safe to run twice: it rebuilds from raw every time."""
    con.execute("""
        CREATE OR REPLACE TABLE orders AS
        SELECT r.order_id,
               COALESCE(c.city, a.city)                          AS city,
               CASE WHEN c.city IS NOT NULL THEN 'correction'
                    WHEN a.city IS NOT NULL THEN 'source'
                    ELSE NULL END                                AS city_source,
               r.delivery_note
        FROM raw_orders r
        LEFT JOIN city_aliases a ON lower(trim(r.city)) = a.alias
        LEFT JOIN corrections  c ON c.order_id = r.order_id
    """)
    # a fingerprint of the whole table, so "same result" is checked, not assumed
    return con.execute("""SELECT COUNT(*), md5(string_agg(order_id || ':' || coalesce(city,'∅'),
                                  ',' ORDER BY order_id)) FROM orders""").fetchone()


print("\n== STEP 1: run the load twice ==")
print("run 1:", load_orders())
print("run 2:", load_orders())
print(con.execute("""SELECT city, COUNT(*) FROM orders GROUP BY city ORDER BY 2 DESC""").fetchall())

== STEP 1: run the load twice ==
run 1: (1000, '9fb6899493deff927b480901368c59d5')
run 2: (1000, '9fb6899493deff927b480901368c59d5')
[('São Paulo', 484), ('Campinas', 130), ('Rio de Janeiro', 123), (None, 111), ('Belo Horizonte', 85), ('Santos', 67)]

ALTER TABLE orders SET DATA_METRIC_SCHEDULE = '60 MINUTE';

-- built-in: nulls in the column
ALTER TABLE orders ADD DATA METRIC FUNCTION SNOWFLAKE.CORE.NULL_COUNT ON (city);

-- custom: values outside the list you accept
CREATE OR REPLACE DATA METRIC FUNCTION governance.dmfs.unknown_city_count(
  arg_t TABLE(arg_city VARCHAR)
)
RETURNS NUMBER
AS
$$
  SELECT COUNT(*) FROM arg_t
  WHERE arg_city NOT IN ('São Paulo','Campinas','Santos','Rio de Janeiro','Belo Horizonte')
$$;

ALTER TABLE orders ADD DATA METRIC FUNCTION governance.dmfs.unknown_city_count ON (city);

-- results land here; measurement_time is when the check actually ran
SELECT measurement_time, metric_name, argument_names, value
FROM SNOWFLAKE.LOCAL.DATA_QUALITY_MONITORING_RESULTS
WHERE table_name = 'ORDERS'
ORDER BY measurement_time DESC;

# spec.yaml
rules:
- column: city
  dimension: VALIDITY
  setExpectation:
    values: ['São Paulo', 'Campinas', 'Santos', 'Rio de Janeiro', 'Belo Horizonte']
  ignoreNull: false
  threshold: 1.0

gcloud dataplex datascans create data-quality orders-city-check \
  --location=us-central1 \
  --data-quality-spec-file=spec.yaml \
  --data-source-resource=//bigquery.googleapis.com/projects/myproject/datasets/mydataset/tables/orders

# ---------------------------------------------------------------------------
# STEP 2: Check quality where it lands.
# An accepted-values rule, the same shape as a BigQuery setExpectation rule.
# ---------------------------------------------------------------------------
def accepted_values_rule(column, values, ignore_null, threshold=1.0):
    """Pass ratio the way the rule defines it: passing rows / total rows.
    ignore_null=False mirrors the documented default: a null row fails."""
    in_list = ", ".join(f"'{v}'" for v in values)
    null_outcome = "1" if ignore_null else "0"
    passing, total = con.execute(f"""
        SELECT SUM(CASE WHEN {column} IS NULL THEN {null_outcome}
                        WHEN {column} IN ({in_list}) THEN 1 ELSE 0 END),
               COUNT(*)
        FROM orders""").fetchone()
    ratio = passing / total
    return round(ratio, 3), "PASS" if ratio >= threshold else "FAIL"


print("\n== STEP 2: accepted-values rule on city ==")
print("ignoreNull = true :", accepted_values_rule("city", CITIES, ignore_null=True))
print("ignoreNull = false:", accepted_values_rule("city", CITIES, ignore_null=False))

== STEP 2: accepted-values rule on city ==
ignoreNull = true : (1.0, 'PASS')
ignoreNull = false: (0.889, 'FAIL')

SELECT order_id,
       AI_COMPLETE(
         model => 'mistral-large2',
         prompt => 'Return the delivery city named in this note: ' || delivery_note,
         response_format => {
           'type': 'json',
           'schema': {'type': 'object',
                      'properties': {'city': {'type': 'string'}},
                      'required': ['city']}
         }
       ) AS fill
FROM orders
WHERE city IS NULL;

SELECT *
FROM AI.GENERATE_TABLE(
  MODEL `mydataset.gemini_model`,
  (
    SELECT order_id,
           CONCAT('Return the delivery city named in this note: ', delivery_note) AS prompt
    FROM mydataset.orders
    WHERE city IS NULL
  ),
  STRUCT('city STRING' AS output_schema)
);

# ---------------------------------------------------------------------------
# STEP 3: Backfill with AI, and label it as AI.
# stand_in_model is NOT an LLM. It behaves like a cheap one on this task:
# it returns the first city it can find in the note, or its best guess.
# Swap in AI_COMPLETE or AI.GENERATE_TABLE and the shape of the problem is the same.
# ---------------------------------------------------------------------------
FILL_SCHEMA = {"type": "object",
               "properties": {"city": {"type": "string"}},
               "required": ["city"]}
PROMPT = "Return the delivery city named in this note: {note}"
MODEL = "stand-in-v1"


def stand_in_model(note):
    found = [c for c in CITIES if c in note]
    found.sort(key=note.index)                 # first mention wins
    return {"city": found[0] if found else "São Paulo"}   # no city? guess the big one


blanks = con.execute("SELECT order_id, delivery_note FROM orders WHERE city IS NULL").fetchall()
fills, schema_failures = [], 0
for order_id, note in blanks:
    answer = stand_in_model(note)
    try:
        jsonschema.validate(answer, FILL_SCHEMA)       # the check the platforms give you
    except jsonschema.ValidationError:
        schema_failures += 1
        continue
    prompt_hash = hashlib.sha256(PROMPT.format(note=note).encode()).hexdigest()[:12]
    fills.append((order_id, answer["city"], MODEL, prompt_hash))

con.execute("CREATE TABLE fills (order_id INT, city VARCHAR, fill_model VARCHAR, prompt_hash VARCHAR)")
con.executemany("INSERT INTO fills VALUES (?, ?, ?, ?)", fills)
con.execute("""
    CREATE OR REPLACE TABLE orders AS
    SELECT o.order_id,
           COALESCE(o.city, f.city)                                     AS city,
           CASE WHEN f.city IS NOT NULL THEN 'ai_fill' ELSE o.city_source END AS city_source,
           f.fill_model, f.prompt_hash, o.delivery_note,
           CAST(NULL AS DOUBLE)                                         AS judge_score
    FROM orders o LEFT JOIN fills f USING (order_id)
""")

print("\n== STEP 3: the backfill ==")
print("blanks:", len(blanks), "| filled:", len(fills), "| schema failures:", schema_failures)
print("accepted-values rule now:", accepted_values_rule("city", CITIES, ignore_null=False))
# the answer key only a synthetic dataset has
print(con.execute("""
    SELECT r.kind, COUNT(*) AS fills, SUM(CASE WHEN o.city <> r.true_city THEN 1 ELSE 0 END) AS wrong
    FROM orders o JOIN raw_orders r USING (order_id)
    WHERE o.city_source = 'ai_fill' GROUP BY r.kind ORDER BY r.kind
""").fetchall())

== STEP 3: the backfill ==
blanks: 111 | filled: 111 | schema failures: 0
accepted-values rule now: (1.0, 'PASS')
[('blank_ambiguous', 36, 36), ('blank_clear', 67, 0), ('blank_empty', 8, 6)]

# The trap query. Every AI-filled value that nothing has checked beyond its shape.
TRAP_QUERY = """
    SELECT order_id, city, fill_model, delivery_note
    FROM orders
    WHERE city_source = 'ai_fill'
      AND (judge_score IS NULL                                   -- never judged
           OR strpos(lower(delivery_note), lower(city)) = 0)     -- value not in its own source
"""
print("unchecked fills:", len(con.execute(TRAP_QUERY).fetchall()))
print("of those, value not in the note:",
      con.execute(f"SELECT COUNT(*) FROM ({TRAP_QUERY}) WHERE strpos(lower(delivery_note), lower(city)) = 0").fetchone()[0])

unchecked fills: 111
of those, value not in the note: 8

from trulens.core.run import RunConfig

run_config = RunConfig(
    run_name="city_backfill_v1",
    source_type="TABLE",
    dataset_name="ORDERS_FILLED",
    dataset_spec={"RECORD_ROOT.INPUT": "DELIVERY_NOTE"},
    llm_judge_name="llama3.1-70b",      # the default; set it anyway, so the run says what judged it
)
run = tru_app.add_run(run_config=run_config)   # tru_app: your app, registered with TruLens
run.start()
run.compute_metrics(metrics=["groundedness"])

from trulens.providers.google import Google

judge = Google()     # model_engine defaults to "gemini-2.5-flash"; vertexai=True for Vertex AI
score, reasons = judge.groundedness_measure_with_cot_reasons(
    source="Cliente de Campinas, entregar no escritório em São Paulo",
    statement="The delivery city is Campinas.",
)

# ---------------------------------------------------------------------------
# STEP 4: Judge the fill with a second model, then grade the judge.
# Real call, per row (TruLens, Google provider; Snowflake runs the same metric server-side):
#     from trulens.providers.google import Google
#     score, reasons = Google().groundedness_measure_with_cot_reasons(
#         source=delivery_note, statement=f"The delivery city is {city}.")
# Here the scores are SYNTHETIC stand-ins so the calibration runs without an API key.
# The assumptions are below. Change them and re-run.
# ---------------------------------------------------------------------------
JUDGE_ASSUMPTIONS = {               # (alpha, beta) of a Beta distribution per case
    "value in note, one city":   (9, 1.5),   # clearly supported, scores high
    "value in note, two cities": (6, 2.5),   # the city IS in the source, so it looks supported
    "value not in note":         (1.5, 6),   # nothing to ground on, scores low
}


def stand_in_judge(city, note):
    in_note = city in note
    n_cities = sum(c in note for c in CITIES)
    case = ("value not in note" if not in_note
            else "value in note, two cities" if n_cities > 1
            else "value in note, one city")
    return round(random.betavariate(*JUDGE_ASSUMPTIONS[case]), 3)


for order_id, city, note in con.execute(
        "SELECT order_id, city, delivery_note FROM orders WHERE city_source = 'ai_fill'").fetchall():
    con.execute("UPDATE orders SET judge_score = ? WHERE order_id = ?",
                [stand_in_judge(city, note), order_id])

print("\n== STEP 4: judge every fill ==")
print("unchecked fills after judging:", len(con.execute(TRAP_QUERY).fetchall()))

# Grade the judge: a hand-checked sample. Here "hand-checked" = the synthetic answer key.
sample = con.execute("""
    SELECT o.order_id, o.judge_score, (o.city = r.true_city) AS human_says_correct
    FROM orders o JOIN raw_orders r USING (order_id)
    WHERE o.city_source = 'ai_fill'
    ORDER BY hash(o.order_id) LIMIT 40
""").fetchall()

print(f"\nhand-checked sample: {len(sample)} fills, "
      f"{sum(1 for s in sample if not s[2])} of them wrong")
print(f"{'threshold':>9} | {'agree':>5} | {'wrong but passed':>16} | {'right but rejected':>18}")
for t in (0.5, 0.7, 0.9):
    agree = sum((score >= t) == ok for _, score, ok in sample)
    wrong_passed = sum(score >= t and not ok for _, score, ok in sample)
    right_rejected = sum(score < t and ok for _, score, ok in sample)
    print(f"{t:>9} | {agree:>2}/{len(sample)} | {wrong_passed:>16} | {right_rejected:>18}")

== STEP 4: judge every fill ==
unchecked fills after judging: 8

hand-checked sample: 40 fills, 16 of them wrong
threshold | agree | wrong but passed | right but rejected
      0.5 | 28/40 |               11 |                  1
      0.7 | 28/40 |                7 |                  5
      0.9 | 27/40 |                2 |                 11

# No threshold fixes it, because the judge is being asked a question the data can't answer:
# a note that names two cities supports both. That's a data problem, so route it as one.
city_count = " + ".join(f"(strpos(delivery_note, '{c}') > 0)::INT" for c in CITIES)
con.execute(f"""
    CREATE TABLE fill_quarantine AS
    SELECT order_id, city, judge_score, delivery_note,
           CASE WHEN strpos(lower(delivery_note), lower(city)) = 0 THEN 'value not in source'
                ELSE 'source names more than one city' END AS reason
    FROM orders
    WHERE city_source = 'ai_fill'
      AND (strpos(lower(delivery_note), lower(city)) = 0 OR ({city_count}) > 1)
""")
con.execute("""UPDATE orders SET city = NULL, city_source = 'quarantined'
               WHERE order_id IN (SELECT order_id FROM fill_quarantine)""")

print("\n== STEP 4b: route what the source can't support ==")
print(con.execute("SELECT reason, COUNT(*) FROM fill_quarantine GROUP BY reason ORDER BY 1").fetchall())

sample = con.execute("""
    SELECT o.order_id, o.judge_score, (o.city = r.true_city) AS human_says_correct
    FROM orders o JOIN raw_orders r USING (order_id)
    WHERE o.city_source = 'ai_fill'
    ORDER BY hash(o.order_id) LIMIT 40
""").fetchall()
print(f"hand-checked sample of what's left: {len(sample)} fills, "
      f"{sum(1 for s in sample if not s[2])} wrong")
for t in (0.5, 0.7, 0.9):
    agree = sum((score >= t) == ok for _, score, ok in sample)
    print(f"  threshold {t}: agree {agree}/{len(sample)}")
print("accepted-values rule, nulls failing:", accepted_values_rule("city", CITIES, ignore_null=False))

== STEP 4b: route what the source can't support ==
[('source names more than one city', 36), ('value not in source', 8)]
hand-checked sample of what's left: 40 fills, 0 wrong
  threshold 0.5: agree 40/40
  threshold 0.7: agree 35/40
  threshold 0.9: agree 23/40
accepted-values rule, nulls failing: (0.956, 'FAIL')

def which_step(symptom):
    return {
        "someone reruns the load by hand":            "1. Automate the load",
        "corrections live in a spreadsheet":          "1. Automate the load",
        "one value, several spellings":               "1. Automate the load (alias table)",
        "blanks nobody can fill":                     "2. Check quality, then 3. Backfill",
        "a quality check that always passes":         "2. Check quality (how are nulls counted?)",
        "AI-filled values you can't tell apart":      "3. Backfill, label it as AI",
        "a judge score nobody has checked":           "4. Judge, then grade the judge",
        "all of the above":                           "1. Start at the load",
    }[symptom]


print("\n== Which step, when you inherit X ==")
for s in ["someone reruns the load by hand", "corrections live in a spreadsheet",
          "one value, several spellings", "blanks nobody can fill",
          "a quality check that always passes", "AI-filled values you can't tell apart",
          "a judge score nobody has checked", "all of the above"]:
    print(f"{s:<40} -> {which_step(s)}")

== Which step, when you inherit X ==
someone reruns the load by hand          -> 1. Automate the load
corrections live in a spreadsheet        -> 1. Automate the load
one value, several spellings             -> 1. Automate the load (alias table)
blanks nobody can fill                   -> 2. Check quality, then 3. Backfill
a quality check that always passes       -> 2. Check quality (how are nulls counted?)
AI-filled values you can't tell apart    -> 3. Backfill, label it as AI
a judge score nobody has checked         -> 4. Judge, then grade the judge
all of the above                         -> 1. Start at the load