Token Optimization in RAG

A correct answer that costs too much and arrives too late

A steel manufacturer has built a RAG chatbot for its dealers and employees. The knowledge base holds product specifications, steel price policies, technical manuals, dealer FAQs, warranty documents, delivery policies, and thousands of PDF and Word files collected over many years.

A dealer asks:

“What is the minimum order quantity for TMT bars, and what documents are required to become an authorized dealer?”

The application retrieves several large document chunks and sends them to an LLM. The answer is correct. The system still has problems:

• Too many input tokens per question, so every answer is expensive.
• High monthly LLM API bills that grow linearly with traffic.
• Slow responses, because the model must read thousands of tokens before it writes the first word.
• Irrelevant retrieved content, such as warranty clauses and southern-region delivery rules.
• Repeated information: the same eligibility rule appears in three documents and in overlapping chunks.
• Large conversation histories resent on every turn.
• Poor performance when many dealers ask questions at once, because long prompts consume rate limits faster.

Why this is not a prompt-writing problem

Most of the tokens in this request were not written by a prompt engineer. They came from upstream decisions: how documents were extracted and cleaned, how large the chunks are, how many chunks the retriever returns, whether filters remove irrelevant regions, whether duplicates are removed, how much chat history is attached, which model is called, and how long the answer is allowed to be. Rewording the system prompt touches only a few hundred of the eight thousand input tokens. Token optimization is therefore a pipeline problem spanning retrieval, chunking, context selection, conversation memory, model choice, and generation.

What a token actually is

Language models do not read characters or words. They read tokens: integer IDs that each represent a piece of text from a fixed vocabulary. A token can be a whole common word (“steel”), part of a word (“author” + “ized”), a single punctuation mark, a space followed by a word, or a single digit.

A tokenizer converts text into these IDs. Most modern tokenizers use a variant of byte-pair encoding (BPE): training starts from individual bytes and repeatedly merges the most frequent adjacent pairs into new vocabulary entries. Frequent English words become single tokens; rare words, product codes and unfamiliar scripts are split into several smaller pieces.

An illustrative split — not verified against any specific tokenizer — might look like this. Note how the leading space is usually attached to the following word, and how “5” and “%” become separate tokens:

Illustrative only. “·” marks a leading space. Run the code below to see the real split for your tokenizer.

Why counts differ across text and models

• Different vocabularies. Each model family trains its own tokenizer. The same sentence can produce different counts on different models, so a budget tuned for one model does not transfer exactly to another.
• Numbers and codes. “12,500.75 MT” or a grade such as “Fe 550D” often splits into many small tokens. Spec sheets and price tables are token-dense.
• Punctuation and special characters. Bullets, pipes from table extraction, repeated dashes and OCR garbage each cost tokens and carry no meaning.
• Code and JSON. Braces, quotes and indentation all count. Pretty-printed JSON uses more tokens than compact JSON.
• Multilingual text. Many tokenizers are trained mostly on English, so Hindi, Tamil or Marathi text commonly needs more tokens per word. A bilingual dealer base can make average prompts noticeably longer.

A rule of thumb often quoted for English is about four characters per token. It is useful for back-of-envelope sizing and wrong for budgets, billing, and non-English text.

Input, output and cached tokens

• Input tokens (prompt tokens) are everything you send: system instructions, the question, history, retrieved context, tool definitions, and formatting the provider adds around messages.
• Output tokens (completion tokens) are what the model generates. They are produced one at a time, so they dominate latency, and providers commonly price them higher than input tokens.
• Cached input tokens are input tokens that a provider has recognized as an exact repeat of a recently processed prompt prefix. Where supported, they may be billed at a lower rate and processed faster. They are still input tokens; caching changes their price, not their count.

Embedding models and generation models are separate. The model that turns chunks into vectors may use a different tokenizer, a different maximum input length and a different price than the model that writes answers. Count each with its own tokenizer and read each provider’s usage report separately.

Counting tokens in Python

For OpenAI models supported by tiktoken, you can count locally. Verify that the encoding matches your model; other providers publish their own tokenizers or token-counting endpoints.

# pip install tiktoken
import tiktoken

# o200k_base is the encoding used by several recent OpenAI models.
# tiktoken.encoding_for_model("model-name") raises KeyError for models it does not know.
enc = tiktoken.get_encoding("o200k_base")

text = "Steel prices increased by 5% in Delhi."
ids = enc.encode(text)

print(len(ids))                          # token count for this encoding
print([enc.decode([i]) for i in ids])    # the actual split

Local counts are estimates. Chat APIs wrap messages in role markers and may add tokens for tool schemas or images, so the provider-reported usage field on each response is the authoritative number for billing.

Where a RAG system spends tokens

Retrieval-Augmented Generation has three phases. Ingestion runs once per document version and prepares a searchable index. Retrieval runs for every question and selects evidence. Generation runs for every question and produces the answer. Each phase has different costs.

Two kinds of token usage must be kept apart:

Storing vectors is not token consumption. A vector database charges for storage, memory and queries. Keeping a million embedding vectors does not consume a single generation token; only the chunk text you later paste into a prompt does.

Token limits, pricing and accounting rules depend on the model, the service tier and the API configuration. Treat every number in this guide that is not derived from a formula as an illustration and replace it with the values from your provider’s documentation.

Token optimization and its neighbours

These terms overlap heavily, and teams often mean different things by them. The distinctions below are the ones used in this guide.

A shorter prompt can cost more

Suppose prompt A uses 8,040 input tokens and succeeds on the first call. Prompt B is trimmed to 5,000 tokens, but in 40% of cases the answer fails validation and the system retries with the full 8,040-token prompt. Expected input per question for B is:

B also doubles latency for 40% of users and generates two answers for them. The same trap appears when an LLM-based compression step saves 3,000 tokens in the answer call but itself consumes 6,000 tokens. Optimize the end-to-end workflow per resolved question, not a single API request.

Measure before you optimize

Without a baseline you cannot tell whether a change saved money, lost quality, or did both. Instrument the current system first: log the token count of each prompt component, the provider-reported usage, latency, and whether the answer was correct.

An illustrative baseline for one dealer request (these are not industry benchmarks):

Retrieved context is three quarters of the input. That is where optimization should start.

Formulas

P_in, P_cached and P_out are prices per million tokens. Check how your provider reports cached tokens — some include them inside the input total, so uncached_input = input − cached_input.

Worked example. At hypothetical prices of $1.00 per million input tokens and $4.00 per million output tokens, with no caching:

Quality metrics to record alongside tokens

• Answer correctness — share of answers judged correct against a reference answer by a reviewer or a validated grader.
• Retrieval relevance — precision@K (share of retrieved chunks that are relevant) and recall@K (share of required evidence chunks that were retrieved).
• Groundedness — share of answer claims supported by the supplied context.
• Citation accuracy — share of citations that point to a chunk that actually supports the cited statement.

Reconcile your local estimates with the provider’s usage fields and invoices weekly. Differences come from message wrappers, tool schemas, tokenizer mismatches and retries you did not log; the bill is computed from provider usage, not from your estimate.

Sixteen techniques, from ingestion to the last output token

Each technique below explains what it is, why waste occurs, how it works on the steel dealer chatbot, and how to tell whether it helped. All before-and-after token counts are illustrative; your numbers depend on your documents, models and questions. The query-time path after applying the retrieval techniques looks like this:

# Raw extraction (noisy)
Home | Products | Dealers | Contact Us
NorthBar Steel Ltd. | Confidential
Page 4 of 38


3.2  Minimum Order Quan-
tity
TMT bars (Fe 500D, Fe 550D): 5 MT
per delivery, North region.
Excep-tion: orders below 5 MT
for project supply require RM
approval.   ¦¦ ~~ ¦
NorthBar Steel Ltd. | Confidential
Page 4 of 38

# Cleaned
## 3.2 Minimum Order Quantity
TMT bars (Fe 500D, Fe 550D): 5 MT
per delivery, North region.
Exception: orders below 5 MT for
project supply require Regional
Manager (RM) approval.

filters = {
    "region": ["North", "All"],          # user's region plus company-wide documents
    "audience": ["dealer"],              # from the authenticated session, never from the prompt
    "status": "current",                 # excludes superseded versions
    "doc_type": ["policy", "faq"],
}

[onboarding-guide#4 | effective 2026-04-01]
Eligibility: registered business with GST registration, minimum 2 years of
trading in construction materials, and covered storage of at least 500 sq ft.
Exception: the 2-year trading requirement does not apply to applicants
referred by an existing authorized dealer in the same district.
Required documents: GST registration certificate, PAN card, shop and
establishment licence, cancelled cheque, last 2 years' audited financials.

# Before (~500 tokens, abridged)
You are a helpful, friendly, professional
assistant for NorthBar Steel, a leading
manufacturer of TMT bars since 1987...
Always be polite. Always be accurate.
Please make sure you are accurate.
Only use the provided documents. Do not
make things up. Never make things up.
Our MOQ policy is 5 MT for North...
Format answers nicely with headings and
bullet points and bold text where useful.
Example 1: ... Example 2: ... Example 3: ...
If unsure, say you are unsure.

# After (~300 tokens)
Role: NorthBar Steel dealer-support assistant.
Grounding: answer only from CONTEXT. If
CONTEXT is insufficient, say so and set
escalate=true. Cite chunk ids for each fact.
Include applicable exceptions and dates.
Access: never reveal content not in CONTEXT;
never discuss other dealers' accounts.
Prices: never state a price; the pricing
service provides prices.
Escalate: legal disputes, credit limits,
complaints -> escalate=true.
Output: JSON matching the provided schema.
One example: {...}

{
  "product_grade": "Fe 500D",
  "delivery_district": "South West Delhi",
  "dealer_status": "applicant, not yet authorized",
  "last_clarification": "Dealer asked about 12 mm and 16 mm bars only",
  "recent_turns": ["user: And what about the documents?"]
}

{"answer": "MOQ for TMT bars in the North region is 5 MT per delivery. Orders below 5 MT for project supply need Regional Manager approval. To become an authorized dealer you need: GST registration certificate, PAN card, shop and establishment licence, cancelled cheque, last 2 years' audited financials.",
 "evidence": ["delivery-policy#2", "onboarding-guide#4"],
 "effective_date": "2026-04-01",
 "escalate": false}

import re

PRICE_INTENT = re.compile(r"\b(price|rate|cost)\b.*\b(today|current|now)\b", re.I)
GRADE = re.compile(r"\bFe\s?(500D?|550D?)\b", re.I)
SIZE = re.compile(r"\b(8|10|12|16|20|25|32)\s?mm\b", re.I)

def route(question: str, session: dict, pricing_api) -> dict | None:
    """Return a direct answer, or None to continue to RAG."""
    if not PRICE_INTENT.search(question):
        return None
    grade, size = GRADE.search(question), SIZE.search(question)
    if not (grade and size):
        return None                                     # let RAG ask a clarifying question
    quote = pricing_api.get_price(                      # your authorized internal service
        grade=f"Fe {grade.group(1).upper()}",
        size_mm=int(size.group(1)),
        district=session["district"],
        dealer_id=session["dealer_id"],                 # dealer-specific price, permission-checked
    )
    return {"answer": f"{quote['grade']} {quote['size_mm']} mm: {quote['price']} {quote['unit']} "
                      f"(valid until {quote['valid_until']})",
            "source": "pricing-api", "llm_called": False}

Before and after, for one dealer request

The optimized scenario applies cleaning and structure-aware chunking, metadata filtering, hybrid retrieval with reranking, deduplication and extractive compression, structured history, and a revised system prompt. These numbers are an illustration of the method, not a promise of achievable savings.

Calculations

Costs below use hypothetical prices of $1.00 per million input tokens and $4.00 per million output tokens, no caching, and 300,000 requests per month. They are not any provider’s prices; adjust them in the Tweaks panel.

Why cost fell by 66.9% while tokens fell by 71.1%. Input and output are priced differently. Input fell by 72.8% but output only by 37.5% (400 → 250), and output carries the higher price, so the blended cost reduction is different from the token reduction. With prompt caching, cached input would be priced lower again and the percentages would shift once more. Always compute savings from priced token categories, not from total tokens.

The full picture includes added costs: the reranker, any query-rewrite calls, and extra embedding calls for expansion. Subtract them from the saving.

A token-aware RAG pipeline you can run locally

The file below runs end to end with only the Python standard library (Python 3.10+); tiktoken is optional. To keep it runnable without accounts or GPUs, three components are local stand-ins behind clearly defined interfaces: a lexical retriever in place of vector/hybrid search, an overlap scorer in place of a cross-encoder, and an extractive “LLM” that assembles an answer from the context. A provider adapter for a real chat API follows the listing.

"""token_aware_rag.py — minimal, runnable, token-aware RAG pipeline.

Python 3.10+. Optional: pip install tiktoken
Local token counts are estimates; provider-reported usage is authoritative for billing.
"""
from __future__ import annotations

import hashlib
import json
import logging
import math
import re
import time
from dataclasses import dataclass
from typing import Protocol

logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("rag")

# HYPOTHETICAL prices in USD per 1M tokens. Replace with your provider's current prices.
PRICES = {"input": 1.00, "cached_input": 0.25, "output": 4.00}

# ---------------------------------------------------------------- 1. Token counting
try:
    import tiktoken
    _ENC = tiktoken.get_encoding("o200k_base")   # verify this matches your generation model
except Exception:                                 # not installed, or encoding download failed
    _ENC = None


def count_tokens(text: str) -> int:
    """Estimate tokens. Falls back to ~4 chars/token (English-biased, rough)."""
    if _ENC is not None:
        return len(_ENC.encode(text))
    return math.ceil(len(text) / 4)


# ---------------------------------------------------------------- 2. Cleaning
BOILERPLATE = [
    re.compile(r"^page \d+ of \d+$", re.I),
    re.compile(r"^northbar steel ltd\. \| confidential$", re.I),
    re.compile(r"^(home|products|dealers|contact us)( \| (home|products|dealers|contact us))+$", re.I),
]


def clean_text(raw: str) -> str:
    """Remove boilerplate and noise; keep single blank lines as paragraph breaks."""
    lines: list[str] = []
    for line in raw.splitlines():
        line = re.sub(r"[ \t]+", " ", line.replace("\u00ad", "")).strip()
        line = re.sub(r"[¦~]{2,}", "", line).strip()      # OCR artifacts seen in this corpus
        if any(p.match(line) for p in BOILERPLATE):
            continue
        if not line:
            if lines and lines[-1] != "":
                lines.append("")
            continue
        lines.append(line)
    text = "\n".join(lines).strip()
    # Re-join words hyphenated across lines, lowercase only so codes like "Fe-500D" survive.
    return re.sub(r"([a-z])-\n([a-z])", r"\1\2", text)


# ---------------------------------------------------------------- 3. Chunking
@dataclass
class Chunk:
    chunk_id: str
    doc_id: str
    text: str
    metadata: dict
    tokens: int


def chunk_document(doc_id: str, text: str, metadata: dict,
                   max_tokens: int = 400, overlap_paragraphs: int = 0) -> list[Chunk]:
    """Paragraph-aware chunking bounded by token count.

    Limitation: a single paragraph longer than max_tokens becomes one oversized chunk.
    """
    paragraphs = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()]
    chunks: list[Chunk] = []
    current: list[str] = []

    def flush() -> None:
        if current:
            body = "\n".join(current)
            chunks.append(Chunk(f"{doc_id}#{len(chunks)}", doc_id, body,
                                dict(metadata), count_tokens(body)))

    for para in paragraphs:
        if current and count_tokens("\n".join(current + [para])) > max_tokens:
            flush()
            current = current[-overlap_paragraphs:] if overlap_paragraphs else []
        current.append(para)
    flush()
    return chunks


# ---------------------------------------------------------------- 4. Metadata filtering
def matches(meta: dict, filters: dict) -> bool:
    for key, allowed in filters.items():
        value = meta.get(key)
        if isinstance(allowed, (list, tuple, set)):
            if value not in allowed:
                return False
        elif value != allowed:
            return False
    return True


# ---------------------------------------------------------------- 5. Candidate retrieval
def words(text: str) -> list[str]:
    return re.findall(r"[a-z0-9]+", text.lower())


class Retriever(Protocol):
    def search(self, query: str, filters: dict, k: int) -> list[tuple[float, Chunk]]: ...


class LexicalRetriever:
    """LOCAL STAND-IN for vector or hybrid search: IDF-weighted term overlap."""

    def __init__(self, chunks: list[Chunk]):
        self.chunks = chunks
        self.df: dict[str, int] = {}
        for c in chunks:
            for w in set(words(c.text)):
                self.df[w] = self.df.get(w, 0) + 1

    def search(self, query: str, filters: dict, k: int = 20) -> list[tuple[float, Chunk]]:
        q, n = set(words(query)), len(self.chunks)
        scored = []
        for c in self.chunks:
            if not matches(c.metadata, filters):        # filter BEFORE ranking
                continue
            shared = q & set(words(c.text))
            score = sum(math.log(1 + n / self.df[w]) for w in shared)
            if score > 0:
                scored.append((score, c))
        scored.sort(key=lambda s: s[0], reverse=True)
        return scored[:k]


# ---------------------------------------------------------------- 6. Reranking
class Reranker(Protocol):
    def score(self, query: str, texts: list[str]) -> list[float]: ...


class OverlapReranker:
    """LOCAL STAND-IN for a cross-encoder: share of query terms present in the text."""

    def score(self, query: str, texts: list[str]) -> list[float]:
        q = set(words(query))
        return [len(q & set(words(t))) / (len(q) or 1) for t in texts]


# ---------------------------------------------------------------- 7. Deduplication
def shingles(text: str, n: int = 5) -> set[str]:
    w = words(text)
    return {" ".join(w[i:i + n]) for i in range(max(1, len(w) - n + 1))}


def deduplicate(chunks: list[Chunk], threshold: float = 0.85) -> list[Chunk]:
    """Drop exact copies (hash) and near copies (Jaccard), keeping input order.

    Near-duplicates are compared only within the same region and effective date,
    so similar-looking but distinct policies are both kept.
    """
    kept: list[Chunk] = []
    hashes: set[str] = set()
    sigs: list[tuple[tuple, set[str]]] = []
    for c in chunks:
        h = hashlib.sha256(" ".join(words(c.text)).encode()).hexdigest()
        if h in hashes:
            continue
        key = (c.metadata.get("region"), c.metadata.get("effective_date"))
        s = shingles(c.text)
        if any(k == key and len(s & t) / len(s | t) >= threshold for k, t in sigs):
            continue
        hashes.add(h)
        sigs.append((key, s))
        kept.append(c)
    return kept


# ---------------------------------------------------------------- 8. Context budgeting
def context_budget(window: int, max_output: int, fixed_text: str,
                   margin_ratio: float = 0.05, task_cap: int = 3000) -> int:
    """Tokens available for retrieved context = window - output - margin - fixed parts."""
    available = window - max_output - int(window * margin_ratio) - count_tokens(fixed_text)
    return max(0, min(available, task_cap))


def format_chunk(c: Chunk) -> str:
    return f"[{c.chunk_id}] effective {c.metadata.get('effective_date', 'n/a')}\n{c.text}"


def pack_context(ranked: list[Chunk], budget: int) -> tuple[list[Chunk], int]:
    """Greedy packing in rank order. Whole chunks only: never cut a clause mid-way."""
    selected, used = [], 0
    for c in ranked:
        cost = count_tokens(format_chunk(c)) + 2      # +2 for the blank-line separator
        if used + cost > budget:
            continue                                   # a smaller, lower-ranked chunk may still fit
        selected.append(c)
        used += cost
    return selected, used


# ---------------------------------------------------------------- 9. Prompt construction
SYSTEM_PROMPT = (
    "Role: NorthBar Steel dealer-support assistant.\n"
    "Grounding: answer only from CONTEXT. If CONTEXT is insufficient, say so and set escalate=true.\n"
    "Cite chunk ids for each fact. Include applicable exceptions and effective dates.\n"
    "Prices: never state a price; the pricing service provides prices.\n"
    'Output JSON: {"answer": str, "sources": [str], "effective_date": str or null, "escalate": bool}'
)


def build_messages(question: str, memory_json: str, chunks: list[Chunk]) -> list[dict]:
    context = "\n\n".join(format_chunk(c) for c in chunks)
    user = f"MEMORY: {memory_json}\n\nCONTEXT:\n{context}\n\nQUESTION: {question}"
    # Stable system prompt first (cache-friendly); variable content last.
    return [{"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user}]


# ---------------------------------------------------------------- 10. Generation interface
@dataclass
class LLMResult:
    text: str
    input_tokens: int
    output_tokens: int
    cached_input_tokens: int = 0
    finish_reason: str = "stop"


class LLMClient(Protocol):
    def generate(self, messages: list[dict], max_output_tokens: int) -> LLMResult: ...


class ExtractiveLocalLLM:
    """LOCAL STAND-IN, not a language model: answers with the first sentence of each
    context block so the pipeline runs offline. Token counts are local estimates."""

    BLOCK = re.compile(r"^\[([^\]]+)\] effective (\S+)\n(.+?)(?=\n\n\[|\n\nQUESTION|\Z)", re.S | re.M)

    def generate(self, messages: list[dict], max_output_tokens: int) -> LLMResult:
        prompt = "\n".join(m["content"] for m in messages)
        blocks = self.BLOCK.findall(prompt)
        answer = " ".join(body.split(". ")[0].rstrip(".") + "." for _, _, body in blocks)
        dates = sorted(d for _, d, _ in blocks if d != "n/a")
        payload = {"answer": answer or "The available documents do not cover this question.",
                   "sources": [cid for cid, _, _ in blocks],
                   "effective_date": dates[-1] if dates else None,
                   "escalate": not blocks}
        text = json.dumps(payload)
        return LLMResult(text, count_tokens(prompt), count_tokens(text))


# ---------------------------------------------------------------- 11. Cost, usage, latency
def estimate_cost(r: LLMResult, prices: dict = PRICES) -> float:
    uncached = r.input_tokens - r.cached_input_tokens
    return (uncached * prices["input"] + r.cached_input_tokens * prices["cached_input"]
            + r.output_tokens * prices["output"]) / 1_000_000


# ---------------------------------------------------------------- 12. Evaluation
def evaluate(result_text: str, required_facts: list[str], allowed_ids: set[str]) -> dict:
    try:
        data = json.loads(result_text)
    except json.JSONDecodeError:
        return {"valid_json": False}
    answer = data.get("answer", "").lower()
    found = [f for f in required_facts if f.lower() in answer]
    cited = set(data.get("sources", []))
    return {"valid_json": True,
            "fact_recall": len(found) / len(required_facts) if required_facts else 1.0,
            "missing_facts": [f for f in required_facts if f not in found],
            "citations_valid": bool(cited) and cited <= allowed_ids,
            "escalate": data.get("escalate")}


# ---------------------------------------------------------------- Orchestration
def answer_question(question: str, filters: dict, memory: dict, retriever: Retriever,
                    reranker: Reranker, llm: LLMClient, *, window: int = 16_000,
                    max_output: int = 600, task_cap: int = 1_500,
                    required_facts: tuple[str, ...] = ()) -> dict:
    t0 = time.perf_counter()
    candidates = retriever.search(question, filters, k=20)
    if not candidates:                                         # early exit: no LLM call
        return {"answer": "No authorized document covers this question.",
                "escalate": True, "llm_called": False}

    chunks = [c for _, c in candidates]
    try:
        scores = reranker.score(question, [c.text for c in chunks])
    except Exception as exc:                                   # degrade, don't fail
        log.warning("reranker failed, using retrieval order: %s", exc)
        scores = [s for s, _ in candidates]
    ranked = [c for _, c in sorted(zip(scores, chunks), key=lambda p: p[0], reverse=True)]
    ranked = deduplicate(ranked)

    memory_json = json.dumps(memory, separators=(",", ":"))   # compact JSON
    budget = context_budget(window, max_output, SYSTEM_PROMPT + memory_json + question,
                            task_cap=task_cap)
    selected, context_tokens = pack_context(ranked, budget)
    messages = build_messages(question, memory_json, selected)
    t1 = time.perf_counter()

    try:
        result = llm.generate(messages, max_output)
    except Exception as exc:
        log.error("generation failed: %s", exc)
        return {"answer": None, "escalate": True, "error": str(exc), "llm_called": True}
    t2 = time.perf_counter()

    if result.finish_reason == "length":
        log.warning("answer truncated at max_output=%d", max_output)

    record = {
        "candidates": len(candidates), "after_dedup": len(ranked), "sent_chunks": len(selected),
        "context_budget": budget, "context_tokens_est": context_tokens,
        "input_tokens": result.input_tokens, "cached_input_tokens": result.cached_input_tokens,
        "output_tokens": result.output_tokens, "cost_usd_est": round(estimate_cost(result), 6),
        "retrieval_ms": round((t1 - t0) * 1000, 1), "generation_ms": round((t2 - t1) * 1000, 1),
    }
    log.info(json.dumps(record))
    evaluation = evaluate(result.text, list(required_facts), {c.chunk_id for c in selected})
    return {"response": result.text, "usage": record, "evaluation": evaluation, "llm_called": True}


# ---------------------------------------------------------------- Demo (fictional data)
RAW_DOCS = [
    ("delivery-policy", """Home | Products | Dealers | Contact Us
NorthBar Steel Ltd. | Confidential
Page 4 of 38

3.2 Minimum Order Quantity
TMT bars minimum order quantity is 5 metric tonnes per delivery for dealers in the North region. Orders below 5 metric tonnes for project supply require Regional Manager approval.   ¦¦ ~~

3.3 Delivery Window
Deliveries within Delhi NCR are scheduled within 3 working days of order confirmation.""",
     {"region": "North", "audience": "dealer", "doc_type": "policy", "effective_date": "2026-04-01"}),

    ("delivery-policy-south", """3.2 Minimum Order Quantity
TMT bars minimum order quantity is 8 metric tonnes per delivery for dealers in the South region.""",
     {"region": "South", "audience": "dealer", "doc_type": "policy", "effective_date": "2026-04-01"}),

    ("onboarding-guide", """4 Becoming an Authorized Dealer
Required documents to become an authorized dealer: GST registration certificate, PAN card, shop and establishment licence, cancelled cheque, and the last 2 years of audited financial statements. Applicants referred by an existing authorized dealer in the same district are exempt from the 2-year trading requirement.""",
     {"region": "All", "audience": "dealer", "doc_type": "policy", "effective_date": "2026-04-01"}),

    ("dealer-faq-copy", """4 Becoming an Authorized Dealer
Required documents to become an authorized dealer: GST registration certificate, PAN card, shop and establishment licence, cancelled cheque, and the last 2 years of audited financial statements. Applicants referred by an existing authorized dealer in the same district are exempt from the 2-year trading requirement.""",
     {"region": "All", "audience": "dealer", "doc_type": "faq", "effective_date": "2026-04-01"}),

    ("credit-terms-internal", """Internal: dealer credit limits are reviewed quarterly by finance.""",
     {"region": "All", "audience": "internal", "doc_type": "policy", "effective_date": "2026-01-01"}),
]

if __name__ == "__main__":
    all_chunks: list[Chunk] = []
    for doc_id, raw, meta in RAW_DOCS:
        all_chunks += chunk_document(doc_id, clean_text(raw), meta, max_tokens=120)

    out = answer_question(
        "What is the minimum order quantity for TMT bars, and what documents are "
        "required to become an authorized dealer?",
        filters={"region": ["North", "All"], "audience": ["dealer"]},  # from the session
        memory={"product_grade": "Fe 500D", "delivery_district": "South West Delhi",
                "dealer_status": "applicant"},
        retriever=LexicalRetriever(all_chunks),
        reranker=OverlapReranker(),
        llm=ExtractiveLocalLLM(),
        required_facts=("5 metric tonnes", "GST registration certificate"),
    )
    print(json.dumps(out, indent=2))

Running it shows the southern MOQ and the internal credit document removed by filters, the FAQ copy of the onboarding text removed by deduplication, and an evaluation confirming both required facts and valid citations. Change task_cap, max_tokens or the filters to see the effect on context tokens.

Provider adapter for real generation

Provider-specific sketch. This adapter targets the OpenAI Python SDK’s Chat Completions interface. Parameter and field names change between SDK versions and models; verify them against the current documentation before use. Note that the SDK reports cached tokens inside the prompt-token total, which matches estimate_cost above.

# PROVIDER-SPECIFIC — verify against your SDK version. pip install openai
from openai import OpenAI

class OpenAIChatAdapter:
    def __init__(self, model: str, client: OpenAI | None = None):
        self.model, self.client = model, client or OpenAI()   # reads OPENAI_API_KEY

    def generate(self, messages: list[dict], max_output_tokens: int) -> LLMResult:
        resp = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            max_completion_tokens=max_output_tokens,
            response_format={"type": "json_object"},
        )
        usage = resp.usage                                       # authoritative for billing
        details = getattr(usage, "prompt_tokens_details", None)
        cached = getattr(details, "cached_tokens", 0) or 0
        choice = resp.choices[0]
        return LLMResult(choice.message.content or "", usage.prompt_tokens,
                         usage.completion_tokens, cached, choice.finish_reason)

A production vector or hybrid retriever implements the same search(query, filters, k) method; a cross-encoder implements score(query, texts). The orchestration does not change.

Limitations and production controls

• The lexical retriever and overlap reranker are teaching stand-ins; they do not understand synonyms.
• The fallback token estimate is crude; the tiktoken encoding may not match a non-OpenAI model at all.
• The extractive evaluator checks fact strings, not meaning. Use human review or a validated grader for correctness and groundedness.
• Add before production: timeouts and retries with backoff, rate-limit handling, request IDs, permission checks in the retriever (not just filters passed by the caller), PII redaction in logs, schema validation with a retry on failure, and metrics export to your observability stack.

A worked budget for a 16,000-token window

Assume a hypothetical generation model with a 16,000-token context window that covers input and output together. Replace this with the documented limits of your model; some models state input and output limits separately.

The usable budget is smaller than the window because output shares the window, providers add formatting tokens around each message, tool-call payloads and tool results are counted as input, and a local tokenizer may disagree with the provider’s by a few percent. If the input reaches the window, the request fails or the output is cut short.

Then apply a task cap below the ceiling: about 800 tokens for a simple factual question, 1,500–2,500 for a multi-document policy question, 3,000–5,000 for a technical comparison, and whatever the applicable clauses require for a high-risk contractual question, up to the 12,000 ceiling. If even the ceiling cannot hold the required evidence, split the task (answer per document, then combine) or escalate rather than truncating clauses.

Fifteen ways token optimization goes wrong

Proving the optimization did not break anything

Build an evaluation set from real, anonymised dealer and employee questions, labelled with the reference answer and the chunk IDs that contain the required evidence. A few hundred questions spread across categories is more useful than thousands of easy ones. Cover each category deliberately:

Metrics, measured together

Running a controlled comparison

• Run baseline and candidate on the same questions, same index snapshot, same model version, and low or zero temperature.
• Change one technique at a time so you can attribute gains and losses.
• Repeat runs or use enough questions to separate real differences from noise; report differences per category, not only overall.
• Agree acceptance thresholds before running, for example: correctness within 1 point of baseline, recall@K ≥ 0.90, zero permission leaks, all exception cases pass, P95 latency not worse.
• After offline acceptance, shadow or A/B test on live traffic and compare billed usage, not just logged estimates.

There is a real trade-off curve between tokens and quality. Early cuts (boilerplate, duplicates, irrelevant regions) are usually free; later cuts start removing evidence. Plot tokens per request against correctness for each configuration and pick the point where further savings start costing accuracy. Token reduction is never the only success criterion.

Production techniques and what each one actually improves

These techniques address different goals. Some reduce the tokens processed, some reduce the price paid for them, some mainly reduce latency, and some mainly improve quality. Few applications need all of them.

Streaming does not reduce tokens or cost; it reduces how long the user waits to see something. Semantic caching reduces cost but is the riskiest item on the list: two similar questions from dealers in different regions may need different answers, so the cache key must include region, permissions and the versions of the source documents.

From symptom to technique

Prioritization

1. Measure. Log tokens per prompt component and build the evaluation set.
2. Remove waste with no evidence in it. Cleaning, deduplication, history control, prompt clean-up, output structure.
3. Route away from the LLM where a database or API is the correct source.
4. Improve retrieval with filters, hybrid search and reranking, then lower K.
5. Add caching and model routing once traffic patterns are known.
6. Introduce compression last, with exception tests, because it carries the highest quality risk.

Checklist

Fewer tokens, or better RAG?

Fewer tokens is a measurement. Better RAG is an outcome: the dealer gets the correct MOQ, the applicable exception, the complete document list and a citation, quickly and at a sustainable cost. The two usually move together when you remove waste, and diverge when you start removing evidence.

Principles

• Optimize the pipeline, not the prompt: most tokens come from retrieval and history.
• Measure cost per resolved question, not per request.
• Count with the right tokenizer and reconcile with provider-reported usage.
• Remove noise and duplicates before removing evidence.
• Use deterministic systems for prices, stock and calculations.
• Treat budgets as ceilings and context windows as limits, not targets.
• Protect exceptions, negations, units and permissions in every transformation.

Implementation order

Instrument → evaluation set → cleaning → deduplication → history control → system prompt and output structure → deterministic routing → metadata filters → hybrid search and reranking → tuned K and dynamic budgets → prompt caching and model routing → extractive compression.

Metrics to watch

Input tokens per request by component, output tokens, cost per resolved question, recall@K, correctness, groundedness, citation accuracy, exception coverage, P95 latency, escalation rate and cache hit rate.

For tomorrow

1. Log token counts per prompt component for one day of traffic.
2. Pull 50 real questions, label answers and evidence, and include exceptions.
3. Read ten logged prompts end to end and mark every duplicate and boilerplate line.
4. Replace full history with structured memory plus the last two turns.
5. Route current-price questions to the pricing API.
6. Re-run the 50 questions and compare tokens, cost and correctness against the baseline.

Provider features, limits and prices change frequently. Check these pages for current details before relying on any behaviour described above.

• openai/tiktoken — tokenizer library and supported encodings.
• OpenAI — Prompt caching guide
• OpenAI — API pricing
• Anthropic — Prompt caching
• Anthropic — Pricing
• LangChain — Text splitters
• Sentence Transformers — Cross-encoders
• Cormack, Clarke and Büttcher (2009). “Reciprocal Rank Fusion outperforms Condorcet and individual rank learning methods.” SIGIR.
• Robertson and Zaragoza (2009). “The Probabilistic Relevance Framework: BM25 and Beyond.” Foundations and Trends in Information Retrieval.

All companies, documents, token counts, prices and evaluation scores in this guide are fictional or hypothetical and are used only to illustrate methods.