Measuring an AI feature from its logs

Write the metric down before you compute it

"Resolution rate" sounds obvious until you compute it. Out of which sessions: all of them, or only those where the assistant answered at least once? Does a test account count? A handoff the next day? Two analysts with the same logs will publish different numbers unless the definition is written down first, as a short metric spec:

All the timestamps are in UTC, so "a day" is a UTC day; a spec says so, because a team in California and one in Berlin would otherwise split the days differently.

Notice the last row. Whether to leave something out depends on the question: test sessions aren't customers, so they don't belong in a resolution rate, but the bill includes them.

Tokens become money through a price table, not a cost the logger computed: prices change, and a table keeps them in one place. Give it valid_from and valid_to dates, joined on each call's date, and every past day is priced at the rate that applied then. (The practice table below has one price per model, to keep it short.)

The daily numbers in one query

With typed views in place, each metric is a short, readable query. A common table expression (WITH name AS (...)) names each step, so the query reads like the spec. This cell rebuilds June 2's events with the three views (the ratings view is the drill's) and the price table, then computes the day:

import json
import sqlite3

db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE events (event_id TEXT PRIMARY KEY, ts TEXT, session_id TEXT, user_id TEXT, type TEXT, payload TEXT);
CREATE TABLE prices (model TEXT PRIMARY KEY, input_per_mtok REAL, output_per_mtok REAL);
INSERT INTO prices VALUES ('small', 1.0, 5.0), ('large', 3.0, 15.0);   -- practice prices, dollars per million tokens
CREATE VIEW sessions AS SELECT session_id, user_id, date(ts) AS day FROM events WHERE type = 'session_start';
CREATE VIEW model_calls AS
  SELECT event_id, session_id, json_extract(payload, '$.model') AS model,
         json_extract(payload, '$.input_tokens') AS input_tokens, json_extract(payload, '$.output_tokens') AS output_tokens
  FROM events WHERE type = 'model_call';
CREATE VIEW ratings AS
  SELECT reply_id, rating FROM (
    SELECT json_extract(payload, '$.reply_id') AS reply_id, json_extract(payload, '$.rating') AS rating,
           ROW_NUMBER() OVER (PARTITION BY json_extract(payload, '$.reply_id') ORDER BY ts DESC) AS newest
    FROM events WHERE type = 'feedback')
  WHERE newest = 1;
""")
day = [("e1", "09:00:05", "s1", "ana", "session_start", {}),
       ("e2", "09:00:09", "s1", "ana", "model_call", {"model": "small", "input_tokens": 900, "output_tokens": 120}),
       ("e3", "09:01:30", "s1", "ana", "feedback", {"reply_id": "e2", "rating": "down"}),
       ("e4", "09:02:10", "s1", "ana", "handoff", {}),
       ("e5", "14:15:00", "s2", "bo", "session_start", {}),
       ("e6", "14:15:04", "s2", "bo", "model_call", {"model": "large", "input_tokens": 1500, "output_tokens": 300}),
       ("e7", "14:16:00", "s2", "bo", "feedback", {"reply_id": "e6", "rating": "down"}),
       ("e7b", "14:17:30", "s2", "bo", "feedback", {"reply_id": "e6", "rating": "up"}),
       ("e8", "15:00:00", "s3", "test-load", "session_start", {}),
       ("e9", "15:00:02", "s3", "test-load", "model_call", {"model": "small", "input_tokens": 50, "output_tokens": 10}),
       ("e10", "16:40:00", "s4", "cy", "session_start", {})]
db.executemany("INSERT INTO events VALUES (?, ?, ?, ?, ?, ?)",
               [(i, "2025-06-02 " + t, s, u, k, json.dumps(p)) for i, t, s, u, k, p in day])

DAILY = """
WITH real AS (SELECT session_id FROM sessions WHERE day = :day AND user_id NOT LIKE 'test-%'),
     calls AS (SELECT m.event_id, m.session_id FROM model_calls m JOIN real r ON r.session_id = m.session_id),
     answered AS (SELECT DISTINCT session_id FROM calls),
     handed AS (SELECT DISTINCT session_id FROM events WHERE type = 'handoff')
SELECT (SELECT count(*) FROM real) AS sessions,
       (SELECT count(*) FROM answered) AS answered,
       (SELECT count(*) FROM answered a WHERE NOT EXISTS (SELECT 1 FROM handed h WHERE h.session_id = a.session_id)) AS resolved,
       (SELECT count(*) FROM calls) AS replies,
       (SELECT count(*) FROM calls c JOIN ratings g ON g.reply_id = c.event_id WHERE g.rating = 'down') AS downs
"""
COST = """SELECT round(sum(m.input_tokens * p.input_per_mtok + m.output_tokens * p.output_per_mtok) / 1e6, 4)
          FROM model_calls m JOIN prices p ON p.model = m.model
          WHERE m.session_id IN (SELECT session_id FROM sessions WHERE day = :day)"""
cursor = db.execute(DAILY, {"day": "2025-06-02"})
print(dict(zip([c[0] for c in cursor.description], cursor.fetchone())))
print("cost", db.execute(COST, {"day": "2025-06-02"}).fetchone()[0])

Three real sessions started (the test one is left out), two got an answer, and one of those ended in a handoff: a resolution rate of 1 in 2. The query says NOT EXISTS rather than NOT IN on purpose: if a single handoff event was logged with a NULL session, x NOT IN (...) is never true for any session (SQL can't say x differs from an unknown), and every day would report zero resolved sessions. Messy logs make that trap real. So do missing prices: the join to prices silently drops calls from a model with no price row, so a trust check before the report counts calls with no price, and calls with no session_start. The cost counts every call in the sessions that started that day, including the test session's, as the spec says. (Attributing everything to the day the session started keeps a conversation that runs past midnight in one day's numbers.)

Python's built-in SQLite has no percentile function, so for latency percentiles pull the typed latency_ms column into Python and use np.percentile, as in the observability lesson. The view has already put every row in milliseconds.

Rollups you can run twice

A dashboard shouldn't scan every event each time someone opens it, so a nightly job writes each day's numbers into a rollup table, one row per day. Two things will happen to that job: it will run twice (a retry, a person rerunning it after a fix), and events will arrive late (a phone that was offline sends its rating the next morning). As the queues lesson put it, the job must be idempotent: running it twice has the same effect as running it once.

Three habits get you there:

1. Ingest with a unique event id. event_id is the primary key, and new events go in with INSERT OR IGNORE, so a batch delivered twice adds nothing the second time. (OR IGNORE also skips rows that break other rules, such as a NOT NULL column, without a word; INSERT ... ON CONFLICT(event_id) DO NOTHING skips only duplicates, so prefer it when a table has other rules.)
2. Write rollups as upserts. INSERT ... ON CONFLICT(day) DO UPDATE replaces the day's row instead of adding a second one. (Deleting the day's row and inserting it again inside one transaction works too.)
3. Recompute a window, not just yesterday. Each night, rebuild the last few days, so late events land in the right day. The price is that a recent day's numbers can still change, so say so on the dashboard.

import sqlite3

db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE events (event_id TEXT PRIMARY KEY, session_day TEXT, type TEXT);
CREATE TABLE daily_metrics (day TEXT PRIMARY KEY, sessions INTEGER, ratings INTEGER);
""")

def ingest(batch):
    db.executemany("INSERT OR IGNORE INTO events VALUES (?, ?, ?)", batch)

def rollup(day):
    sessions, ratings = db.execute("""SELECT sum(type = 'session_start'), sum(type = 'feedback')
                                      FROM events WHERE session_day = ?""", (day,)).fetchone()
    db.execute("""INSERT INTO daily_metrics VALUES (?, ?, ?)
                  ON CONFLICT(day) DO UPDATE SET sessions = excluded.sessions, ratings = excluded.ratings""",
               (day, sessions, ratings))

batch = [("e1", "2025-06-03", "session_start"), ("e2", "2025-06-03", "feedback"), ("e3", "2025-06-03", "session_start")]
ingest(batch); ingest(batch)                  # the same batch delivered twice
rollup("2025-06-03"); rollup("2025-06-03")    # the job run twice
print(db.execute("SELECT * FROM daily_metrics").fetchall())

ingest([("e9", "2025-06-03", "feedback")])    # a rating for a June 3 session, arriving on June 4
rollup("2025-06-03")                          # the next night's window rerun includes June 3
print(db.execute("SELECT * FROM daily_metrics").fetchall())

Delivered twice and rolled up twice, June 3 still has two sessions and one rating, in one row. The late rating arrived on June 4, but it belongs to a June 3 session, so rerunning June 3 counted it. (In the real views, a late rating finds its day by joining to its reply's session.)

import sqlite3


def ingest(db, events):
    # a plain INSERT fails on a repeated event_id: store only new events, and count them
    return 0


db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE events (event_id TEXT PRIMARY KEY, ts TEXT, session_id TEXT, user_id TEXT, type TEXT, payload TEXT)")
batch = [("e1", "2025-06-02 09:00", "s1", "ana", "session_start", "{}"),
         ("e2", "2025-06-02 09:01", "s1", "ana", "model_call", "{}")]
print(ingest(db, batch), ingest(db, batch))