|
import tiktoken |
|
|
|
enc = tiktoken.get_encoding("o200k_base") |
|
|
|
|
|
def n(s): |
|
return len(enc.encode(s)) |
|
|
|
|
|
# --------------------------------------------------------------- |
|
# Shared incompressible payloads (must survive abbreviation intact) |
|
# --------------------------------------------------------------- |
|
|
|
PY_CODE = '''```python |
|
def merge_intervals(intervals: list[tuple[int, int]]) -> list[tuple[int, int]]: |
|
if not intervals: |
|
return [] |
|
intervals.sort(key=lambda iv: iv[0]) |
|
merged = [intervals[0]] |
|
for start, end in intervals[1:]: |
|
last_start, last_end = merged[-1] |
|
if start <= last_end: |
|
merged[-1] = (last_start, max(last_end, end)) |
|
else: |
|
merged.append((start, end)) |
|
return merged |
|
|
|
|
|
def busiest_overlap(intervals): |
|
events = [] |
|
for s, e in intervals: |
|
events.append((s, 1)) |
|
events.append((e, -1)) |
|
events.sort() |
|
best = cur = 0 |
|
for _, delta in events: |
|
cur += delta |
|
best = max(best, cur) |
|
return best |
|
```''' |
|
|
|
JSON_SCHEMA = '''```json |
|
{ |
|
"type": "object", |
|
"required": ["invoice_id", "issued_at", "line_items", "total"], |
|
"properties": { |
|
"invoice_id": {"type": "string", "pattern": "^INV-[0-9]{6}$"}, |
|
"issued_at": {"type": "string", "format": "date-time"}, |
|
"currency": {"type": "string", "enum": ["USD", "EUR", "GBP"]}, |
|
"line_items": { |
|
"type": "array", |
|
"minItems": 1, |
|
"items": { |
|
"type": "object", |
|
"required": ["sku", "quantity", "unit_price"], |
|
"properties": { |
|
"sku": {"type": "string"}, |
|
"quantity": {"type": "integer", "minimum": 1}, |
|
"unit_price": {"type": "number", "exclusiveMinimum": 0}, |
|
"discount_pct": {"type": "number", "minimum": 0, "maximum": 100} |
|
} |
|
} |
|
}, |
|
"total": {"type": "number"} |
|
} |
|
} |
|
```''' |
|
|
|
STACK_TRACE = '''``` |
|
Traceback (most recent call last): |
|
File "/app/services/billing/worker.py", line 214, in process_batch |
|
invoice = render_invoice(order, customer, tax_table) |
|
File "/app/services/billing/render.py", line 88, in render_invoice |
|
line_total = item.unit_price * item.quantity * (1 - item.discount_pct / 100) |
|
TypeError: unsupported operand type(s) for /: 'NoneType' and 'int' |
|
```''' |
|
|
|
MD_TABLE = '''| Endpoint | Method | Auth | Rate limit | Idempotent | |
|
|---|---|---|---|---| |
|
| /v2/invoices | POST | Bearer | 10/min | via Idempotency-Key | |
|
| /v2/invoices/{id} | GET | Bearer | 120/min | yes | |
|
| /v2/invoices/{id}/void | POST | Bearer + admin scope | 5/min | yes | |
|
| /v2/webhooks | POST | HMAC signature | n/a | no |''' |
|
|
|
SQL_QUERY = '''```sql |
|
SELECT c.customer_id, |
|
c.region, |
|
DATE_TRUNC('month', o.created_at) AS month, |
|
SUM(oi.quantity * oi.unit_price) AS revenue, |
|
COUNT(DISTINCT o.order_id) AS orders |
|
FROM customers c |
|
JOIN orders o ON o.customer_id = c.customer_id |
|
JOIN order_items oi ON oi.order_id = o.order_id |
|
WHERE o.status NOT IN ('cancelled', 'refunded') |
|
AND o.created_at >= NOW() - INTERVAL '12 months' |
|
GROUP BY 1, 2, 3 |
|
HAVING SUM(oi.quantity * oi.unit_price) > 1000 |
|
ORDER BY revenue DESC; |
|
```''' |
|
|
|
# --------------------------------------------------------------- |
|
# Experiments: (category, variants dict, incompressible payloads) |
|
# Each variants dict maps variant name -> full text. |
|
# "Full" is always the baseline. |
|
# --------------------------------------------------------------- |
|
|
|
experiments = [] |
|
|
|
# 1. Pure prose, long-form (best case for abbreviation) |
|
experiments.append(("Pure prose: meeting summary brief", { |
|
"Full": ( |
|
"Please summarize the following meeting transcript. Focus on action items, " |
|
"decisions that were made, unresolved questions, and any deadlines that were " |
|
"mentioned. For each action item, identify the owner if one was assigned and " |
|
"note whether a due date was stated explicitly or merely implied. Where the " |
|
"discussion was inconclusive, briefly characterize the disagreement and list " |
|
"the participants on each side. Distinguish clearly between decisions that " |
|
"were finalized during the meeting and proposals that still require sign-off " |
|
"from someone who was not present. Finally, produce a short section of " |
|
"suggested agenda items for the follow-up meeting, ordered by urgency. " |
|
"Present all of the results as nested bulleted lists under clear headings, " |
|
"and keep the total length under four hundred words." |
|
), |
|
"Abbreviated": ( |
|
"pls sum mtg transcript. focus: actions (owner? due date explicit/implied), " |
|
"decisions, open qs, deadlines. inconclusive topics: note disagreement + who " |
|
"on each side. separate finalized decisions vs proposals needing absent " |
|
"sign-off. end w/ follow-up agenda ordered by urgency. nested bullets under " |
|
"headings, <400 words." |
|
), |
|
"Custom shorthand": ( |
|
"S mtg transcript: AI(owner,DD expl/impl) DEC OQ DL; inconcl->disagr+sides; " |
|
"final DEC vs pending-signoff; FU agenda by urgency; nested bullets+hdgs; <400w." |
|
), |
|
}, [])) |
|
|
|
# 2. Prose wrapping a code sample (code must stay intact) |
|
experiments.append(("Code review request (Python)", { |
|
"Full": ( |
|
"Please review the following Python module for correctness, efficiency, and " |
|
"readability. Pay particular attention to edge cases such as empty input, " |
|
"intervals that share a boundary point, and intervals that are fully " |
|
"contained inside another interval. Comment on whether the sorting strategy " |
|
"is optimal and whether the sweep-line approach in the second function " |
|
"handles touching intervals the way a caller would expect. Suggest concrete " |
|
"improvements with code where appropriate.\n\n" + PY_CODE |
|
), |
|
"Abbreviated": ( |
|
"review py module: correctness, perf, readability. edge cases: empty input, " |
|
"shared boundaries, fully-contained intervals. is sort optimal? does " |
|
"sweep-line handle touching intervals as expected? suggest fixes w/ code.\n\n" |
|
+ PY_CODE |
|
), |
|
}, [PY_CODE])) |
|
|
|
# 3. Structured-output prompt with a JSON Schema (schema untouchable) |
|
experiments.append(("Extraction prompt with JSON Schema", { |
|
"Full": ( |
|
"You will be given the raw text of a scanned invoice. Extract the fields " |
|
"described by the JSON Schema below and return a single JSON object that " |
|
"validates against it. If a field is missing from the document, omit it " |
|
"rather than guessing, unless it is listed as required, in which case use " |
|
"null and add the field name to a top-level list called \"missing\". Do not " |
|
"include any commentary outside the JSON object.\n\n" + JSON_SCHEMA |
|
), |
|
"Abbreviated": ( |
|
"extract fields per schema below from scanned invoice text -> single valid " |
|
"JSON obj. missing optional field: omit. missing required: null + add name " |
|
"to top-level \"missing\" list. JSON only, no commentary.\n\n" + JSON_SCHEMA |
|
), |
|
}, [JSON_SCHEMA])) |
|
|
|
# 4. Debugging prompt with a stack trace (trace untouchable) |
|
experiments.append(("Debugging prompt with stack trace", { |
|
"Full": ( |
|
"Our nightly billing worker started failing after yesterday's deploy. The " |
|
"traceback below is representative of all the failures. Please explain the " |
|
"most likely root cause, identify which recent schema or code change could " |
|
"have introduced it, and propose both an immediate hotfix and a longer-term " |
|
"defensive fix, including where validation should live so this class of " |
|
"error cannot reach the worker again.\n\n" + STACK_TRACE |
|
), |
|
"Abbreviated": ( |
|
"nightly billing worker failing since yesterday's deploy, all failures like " |
|
"trace below. likely root cause? which schema/code change? give hotfix + " |
|
"long-term defensive fix + where validation should live.\n\n" + STACK_TRACE |
|
), |
|
}, [STACK_TRACE])) |
|
|
|
# 5. API docs prompt with a markdown table (table untouchable) |
|
experiments.append(("Docs rewrite with markdown table", { |
|
"Full": ( |
|
"Rewrite the API reference section below for external developers. Keep the " |
|
"table exactly as it is, but add an introductory paragraph explaining the " |
|
"authentication model, a paragraph on how rate limits are enforced and what " |
|
"the client should do on a 429 response, and a short note on idempotency " |
|
"keys with a concrete example header. The tone should be friendly but " |
|
"precise, and every claim must be consistent with the table.\n\n" + MD_TABLE |
|
), |
|
"Abbreviated": ( |
|
"rewrite API ref for external devs. keep table verbatim. add: intro para on " |
|
"auth model, para on rate limits + client behavior on 429, note on " |
|
"idempotency keys w/ example header. friendly+precise, consistent w/ " |
|
"table.\n\n" + MD_TABLE |
|
), |
|
}, [MD_TABLE])) |
|
|
|
# 6. SQL optimization prompt (query untouchable) |
|
experiments.append(("SQL optimization request", { |
|
"Full": ( |
|
"The analytics query below has become slow as the orders table has grown to " |
|
"roughly two hundred million rows. Please suggest indexing strategies, " |
|
"rewrite opportunities, and any partitioning scheme that would help, " |
|
"assuming PostgreSQL 16. Explain the reasoning behind each suggestion and " |
|
"estimate the relative impact. The query must continue to return exactly " |
|
"the same results.\n\n" + SQL_QUERY |
|
), |
|
"Abbreviated": ( |
|
"query below slow, orders ~200M rows, PG16. suggest indexes, rewrites, " |
|
"partitioning. explain reasoning + est. impact. results must stay " |
|
"identical.\n\n" + SQL_QUERY |
|
), |
|
}, [SQL_QUERY])) |
|
|
|
# --------------------------------------------------------------- |
|
# Report |
|
# --------------------------------------------------------------- |
|
|
|
grand = [] |
|
print(f"{'':62s} {'tokens':>7s} {'saved':>7s}") |
|
for title, variants, payloads in experiments: |
|
base = n(variants["Full"]) |
|
payload_tokens = sum(n(p) for p in payloads) |
|
print(f"\n== {title} ==") |
|
if payloads: |
|
prose = base - payload_tokens |
|
ceiling = prose / base |
|
print(f" incompressible payload: {payload_tokens} tok " |
|
f"({payload_tokens/base:.0%} of prompt) -> max possible saving {ceiling:.0%}") |
|
for name, text in variants.items(): |
|
t = n(text) |
|
saved = (base - t) / base |
|
print(f" {name:58s} {t:7d} {saved:7.1%}") |
|
if name != "Full": |
|
grand.append((title, name, base, t, saved, payload_tokens)) |
|
# verify payloads survived abbreviation intact |
|
for name, text in variants.items(): |
|
for p in payloads: |
|
assert p in text, f"payload mangled in {title} / {name}" |
|
|
|
# --------------------------------------------------------------- |
|
# Aggregate: savings on prose portion only (isolating the effect) |
|
# --------------------------------------------------------------- |
|
print("\n== Savings on the compressible (prose) portion only ==") |
|
for title, name, base, t, saved, payload in grand: |
|
if payload: |
|
prose_full = base - payload |
|
prose_abbr = t - payload |
|
print(f" {title:45s} prose {prose_full:4d} -> {prose_abbr:4d} " |
|
f"({(prose_full-prose_abbr)/prose_full:.1%} of prose, " |
|
f"but only {saved:.1%} of whole prompt)") |