Send results to BigQuery

Recipe — stream evaluation results into BigQuery.

from datetime import datetime, timedelta
from google.cloud import bigquery
from layerlens import Stratix

client = Stratix()
bq = bigquery.Client()

since = datetime.utcnow() - timedelta(hours=24)

# Page evaluations and emit one BigQuery row per result
page = 1
rows = []
while True:
 response = client.evaluations.get_many(page=page, page_size=100, sort_by="submitted_at", order="desc")
 evaluations = [e for e in response.evaluations if e.submitted_at >= since]
 if not evaluations:
 break
 for evaluation in evaluations:
 results = client.results.get_all(evaluation=evaluation)
 for r in results:
 rows.append({
 "evaluation_id": evaluation.id,
 "model_id": evaluation.model_id,
 "benchmark_id": evaluation.benchmark_id,
 "prompt": r.prompt,
 "result": r.result,
 "truth": r.truth,
 "score": r.score,
 "subset": r.subset,
 })
 if page >= response.pagination.total_pages:
 break
 page += 1

if rows:
 errors = bq.insert_rows_json("project.dataset.eval_rows", rows)
 if errors:
 print(f"BigQuery insert errors: {errors}")

See also

• SDK reference: results
• SDK reference: evaluations