Lab 13 · In-SQL AI Functions

DuckDB Python UDFs stand in for functions like Snowflake Cortex AI_CLASSIFY and BigQuery AI.GENERATE, an in-table FLOAT[384] embedding column plus array_cosine_similarity stands in for warehouse VECTOR_SEARCH, and a cost-model table you compute in SQL compares warehouse-native versus external-API processing at a million-row scale.

You will classify 300 support tickets with a UDF, sample-audit the misclassification rate before trusting a backfill, run semantic search over the same table, and export a cost model whose punchline every data engineer should carry: at millions of rows, model size and token count dominate cost, not vendor branding — and the audit you run on 20 rows is what makes a million-row backfill defensible.

Time & tier

~90 minutes hands-on · Core tier (maps to module M8).

Setup

Laptop-only, $0 in dry-run mode. The classifier UDF has two implementations: a keyword stub (default, offline) and a real Claude call (set LAB_DRY_RUN=0 with an API key; 300 short calls cost a few cents). The point of the stub is not realism — it is that the SQL surface (UDF in a SELECT, enriched table, audit query, cost model) is identical whichever backend answers.

mkdir -p ~/labs/de-lab-13 && cd ~/labs/de-lab-13
python3 -m venv .venv && source .venv/bin/activate
pip install duckdb faker sentence-transformers pandas anthropic

Create make_tickets.py — 300 deterministic support tickets with true labels:

import duckdb, random
from faker import Faker

random.seed(13); Faker.seed(13)
fake = Faker()

CATEGORIES = ["billing", "shipping", "product_defect", "account_access", "other"]
TEMPLATES = {
    "billing": ["I was charged twice for order {o}", "refund still not processed for {o}",
                "invoice amount looks wrong on {o}", "why did my subscription price increase"],
    "shipping": ["package {o} stuck in transit for two weeks", "wrong delivery address on {o}",
                 "courier marked {o} delivered but nothing arrived", "need to expedite shipping for {o}"],
    "product_defect": ["the unit arrived with a cracked housing", "device stops working after an hour",
                       "screen flickers constantly since unboxing", "battery drains in thirty minutes"],
    "account_access": ["cannot log in after password reset", "two-factor codes never arrive",
                       "account locked after too many attempts", "email change did not take effect"],
    "other": ["do you have a student discount", "how do I download my data",
              "feature request: dark mode please", "is there a reseller program in my region"],
}

con = duckdb.connect("support.duckdb")
con.execute("CREATE OR REPLACE TABLE tickets (ticket_id VARCHAR, body VARCHAR, true_category VARCHAR)")
rows = []
for i in range(300):
    cat = random.choice(CATEGORIES)
    body = random.choice(TEMPLATES[cat]).format(o=f"ORD-{random.randint(10000,99999)}")
    body += " " + fake.sentence(nb_words=8)
    rows.append((f"T-{i:04d}", body, cat))
con.executemany("INSERT INTO tickets VALUES (?, ?, ?)", rows)
print("tickets:", con.execute("SELECT count(*) FROM tickets").fetchone()[0])

Run it: python make_tickets.py → tickets: 300.

Guided steps

1. 1.

   Register the classifier as a SQL function. Create classify.py. con.create_function registers a Python callable as a scalar SQL function — after that line, ai_classify(body) is just SQL.

   Copy

   import duckdb, os
   
   CATEGORIES = ["billing", "shipping", "product_defect", "account_access", "other"]
   DRY_RUN = os.environ.get("LAB_DRY_RUN", "1") == "1"
   
   def ai_classify_llm(body: str) -> str:
       import anthropic
       client = anthropic.Anthropic()
       msg = client.messages.create(
           model="claude-opus-4-8", max_tokens=16, thinking={"type": "adaptive"},
           messages=[{"role": "user", "content":
                      f"Classify this support ticket into exactly one of {CATEGORIES}. "
                      f"Reply with the label only.\n\nTicket: {body}"}])
       label = "".join(b.text for b in msg.content if b.type == "text").strip().lower()
       return label if label in CATEGORIES else "other"
   
   KEYWORDS = {
       "billing": ["charged", "refund", "invoice", "price", "subscription"],
       "shipping": ["package", "transit", "delivery", "courier", "shipping", "address"],
       "product_defect": ["cracked", "stops working", "flickers", "battery", "defect", "broken"],
       "account_access": ["log in", "password", "two-factor", "locked", "email change"],
   }
   def ai_classify_dry(body: str) -> str:
       b = body.lower()
       for cat, kws in KEYWORDS.items():
           if any(k in b for k in kws):
               return cat
       return "other"
   
   con = duckdb.connect("support.duckdb")
   con.create_function("ai_classify", ai_classify_dry if DRY_RUN else ai_classify_llm,
                       ["VARCHAR"], "VARCHAR")
   
   con.execute("""CREATE OR REPLACE TABLE tickets_enriched AS
     SELECT ticket_id, body, true_category, ai_classify(body) AS predicted_category
     FROM tickets""")
   
   acc = con.execute("""SELECT round(100.0 * count(*) FILTER (WHERE predicted_category = true_category)
                               / count(*), 1) FROM tickets_enriched""").fetchone()[0]
   print("accuracy vs labels:", acc, "%")
   print(con.execute("""SELECT predicted_category, count(*) FROM tickets_enriched
                        GROUP BY 1 ORDER BY 2 DESC""").fetchall())
   

   Run python classify.py. Dry-run accuracy lands around 99% (the stub's keywords were written against the same templates — an honest label for that is train-test contamination, and it is worth one sentence in your notes about why production accuracy numbers need held-out data).

2. 2.

   Sample-audit before you would backfill. The single most important habit with in-SQL AI functions: before running AI_CLASSIFY over 40 million rows, you classify a sample and read it. Do that here:

   Copy

   SELECT ticket_id, body[:60] AS snippet, true_category, predicted_category
   FROM tickets_enriched USING SAMPLE 20 ROWS;
   

   (Run via duckdb support.duckdb CLI or con.execute(...).) For each disagreement, decide: model wrong, label wrong, or genuinely ambiguous? Record the observed misclassification rate and the dominant confusion pair in your notes. In a warehouse this audit is the difference between "we enriched the table" and "we quietly wrote 8% wrong labels into every downstream dashboard."

3. 3.

   Embed the tickets into a SQL column. Create embed_search.py. The embedding lives in the table like any other column — that locality is the entire argument for warehouse-native vector search.

   Copy

   import duckdb
   from sentence_transformers import SentenceTransformer
   
   con = duckdb.connect("support.duckdb")
   model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
   rows = con.execute("SELECT ticket_id, body FROM tickets").fetchall()
   emb = model.encode([r[1] for r in rows], normalize_embeddings=True)
   
   con.execute("CREATE OR REPLACE TABLE ticket_embeddings (ticket_id VARCHAR, embedding FLOAT[384])")
   con.executemany("INSERT INTO ticket_embeddings VALUES (?, ?)",
                   [(r[0], e.tolist()) for r, e in zip(rows, emb)])
   
   q = model.encode(["customer says the parcel never showed up"], normalize_embeddings=True)[0].tolist()
   res = con.execute("""
     SELECT t.ticket_id, t.body[:60] AS snippet,
            round(array_cosine_similarity(e.embedding, ?::FLOAT[384]), 3) AS sim
     FROM ticket_embeddings e JOIN tickets t USING (ticket_id)
     ORDER BY sim DESC LIMIT 5""", [q]).fetchall()
   for r in res: print(r)
   

   Expected: five "courier marked … delivered but nothing arrived" tickets — a query that shares almost no keywords with the results. That is VECTOR_SEARCH semantics in plain SQL.

4. 4.

   Add an ANN index. Exact scan is fine at 300 rows and hopeless at 300 million. DuckDB's vss extension provides HNSW (one-time download to install):

   Copy

   con.execute("INSTALL vss; LOAD vss;")
   con.execute("SET hnsw_enable_experimental_persistence = true")
   con.execute("""CREATE INDEX IF NOT EXISTS emb_idx ON ticket_embeddings
                  USING HNSW (embedding) WITH (metric = 'cosine')""")
   

   Note the parallel in your notes: this index corresponds to the managed ANN infrastructure behind Cortex Search / BigQuery vector indexes / Databricks Vector Search — in the warehouse you get it as a service; here you see the moving part.

5. 5.

   Build the cost model in SQL. Create cost_model.py. Assumptions live in a table so changing a price is an UPDATE, not a code edit — hand this exact pattern to a finance partner.

   Copy

   import duckdb
   con = duckdb.connect("support.duckdb")
   con.execute("""CREATE OR REPLACE TABLE cost_assumptions (
     approach VARCHAR, price_per_m_input DOUBLE, price_per_m_output DOUBLE,
     tokens_in_per_row INTEGER, tokens_out_per_row INTEGER, batch_discount DOUBLE)""")
   con.execute("""INSERT INTO cost_assumptions VALUES
     ('warehouse_native_small', 0.20, 0.80, 120, 8, 1.00),
     ('warehouse_native_large', 3.00, 15.00, 120, 8, 1.00),
     ('external_api_frontier', 5.00, 25.00, 150, 8, 1.00),
     ('external_api_frontier_batch', 5.00, 25.00, 150, 8, 0.50),
     ('external_api_small', 0.25, 1.25, 150, 8, 1.00)""")
   df = con.execute("""
     SELECT approach,
            round((price_per_m_input * tokens_in_per_row + price_per_m_output * tokens_out_per_row)
                  * batch_discount, 2) AS usd_per_million_rows
     FROM cost_assumptions ORDER BY usd_per_million_rows""").df()
   print(df.to_string(index=False))
   df.to_csv("cost_model.csv", index=False)
   

   With these illustrative rates (verify current vendor pricing before quoting them — they move), the output is:

   approach	USD per 1M rows
   warehouse_native_small	30.40
   external_api_small	47.50
   external_api_frontier_batch	475.00
   warehouse_native_large	480.00
   external_api_frontier	950.00

   Read the ranking, then write the three conclusions it forces: (a) model size dominates — small-vs-frontier is a 15–30x spread that dwarfs the warehouse-vs-external distinction; (b) external tokens also count the ~30 prompt-scaffolding tokens per row that warehouse functions absorb — trim your per-row prompt; (c) batch tiers halve external cost for exactly this kind of never-urgent workload. Then add the costs SQL cannot see: external APIs add egress, secrets management, and a data-residency conversation; warehouse-native adds vendor lock-in and a narrower model menu. For regulated data, "the rows never leave the platform" often decides it before price does.

6. 6.

   Close the loop on audit economics. Append to your notes: at external_api_small rates, classifying 40M rows costs ~$1,900. A 500-row human audit at 30 seconds a row costs about four hours of one analyst. The audit is two orders of magnitude cheaper than the backfill — there is never a budget reason to skip it.

Verification

• •
  tickets_enriched has 300 rows; dry-run accuracy prints ≥ 95%.
• •
  Your 20-row sample audit is written up with a misclassification estimate and at least one examined disagreement (or an explicit "0 disagreements in sample").
• •
  The semantic search returns delivery-failure tickets for the "parcel never showed up" query with similarity ≈ 0.5, none of which contain the word "parcel".
• •
  The HNSW index creates without error (vss loaded).
• •
  cost_model.csv exists and reorders correctly when you UPDATE an assumption (test by doubling one price).

Completion artifact

A de-lab-13/ folder containing make_tickets.py, classify.py, embed_search.py, cost_model.py, cost_model.csv, and sql-ai-notes.md — the accuracy number with its contamination caveat, the sample-audit findings, the top-5 semantic search results, the cost table with your three written conclusions, and one paragraph mapping each local component to its Snowflake/BigQuery/Databricks equivalent. Self-grading rubric (pass = 8/10):

• UDF classification working (0–2): enriched table built by a SQL-registered function; accuracy measured in SQL = 2.
• Audit discipline (0–2): sample audit performed and written up before any talk of backfilling; audit-vs-backfill economics noted = 2.
• Vector search in SQL (0–2): embedding column + cosine top-5 correct; ANN index created = 2.
• Cost model (0–3): assumptions in a table, per-million-row output ranked, three conclusions drawn, prices flagged as needing verification = 3.
• Warehouse mapping (0–1): each component mapped to named vendor equivalents = 1.

Stuck?

• •
  create_function TypeError: signature is (name, fn, [input_types], return_type) — types are strings like "VARCHAR".
• •
  Accuracy near 20%: your UDF is returning None or a category with different casing — the stub must return exact members of CATEGORIES.
• •
  INSTALL vss fails offline: the extension download needs network once; the lab is complete without the index — note the skip.
• •
  FILTER (WHERE ...) syntax error: you are on an old DuckDB — pip install -U duckdb.
• •
  LLM mode is slow: 300 sequential calls take a few minutes; that is the point of the cost/latency discussion — do not parallelize past your rate limits.

Client-safe variant

Dry-run mode never calls out, so the full lab runs inside any client boundary. The client conversation this lab prepares you for is the architecture recommendation: for regulated Indian clients (BFSI under RBI/SEBI localization, anyone under DPDP), warehouse-native AI functions keep customer text inside the platform tenancy the client already governs — no new data processor, no cross-border transfer analysis for a US-hosted API. Bring the cost-model CSV with the client's actual vendor prices filled in and the sample-audit protocol as the quality gate; the recommendation writes itself, and it is backed by numbers you generated rather than vendor slides.