Running into context window errors mid-conversation wastes tokens and breaks user flows. To count tokens accurately context limit before you call the model, you need a tokenizer that matches the target model and an accounting routine that covers every field the API serializes. This guide walks through a concrete pre-flight check you can drop into any Python service.
Step 1: Identify the tokenizer for your target model
Tokenization is model-specific. GPT-4 class models use the cl100k or o200k BPE tokenizer; Llama 3 uses a SentencePiece variant; Mistral has its own. If you guess with the wrong one, your count drifts by 10–30% on mixed-language text, and that drift turns into silent truncations or hard 400 errors.
For OpenAI-compatible endpoints, tiktoken is the reference implementation. Install it:
pip install tiktokenLoad the encoding by model name:
import tiktoken
def get_encoder(model_name: str):
try:
return tiktoken.encoding_for_model(model_name)
except KeyError:
# fallback for unknown but compatible models
return tiktoken.get_encoding("cl100k_base")
enc = get_encoder("gpt-4o")Do not use len(text.split()) or len(text)/4. Those heuristics fail on code, CJK text, and emoji. For Llama or Mistral, use the official tokenizer from HuggingFace transformers or the tokenizers library. The counting principle stays identical: encode the exact string the gateway will send.
Step 2: Count a chat messages array, not just the text
The API does not send raw strings. It serializes each message with role tags, delimiters, and a trailing assistant prompt. OpenAI’s chat format adds approximately 4 tokens per message plus 2 per named role. Your pre-flight must replicate that overhead or your estimate will be systematically low.
def count_messages_tokens(messages: list[dict], encoder) -> int:
tokens_per_message = 4
tokens_per_name = 2
total = 0
for msg in messages:
total += tokens_per_message
for key, value in msg.items():
if isinstance(value, str):
total += len(encoder.encode(value))
if key == "name":
total += tokens_per_name
return total
messages = [
{"role": "system", "content": "You are a terse debugger."},
{"role": "user", "content": "Why does my token count mismatch?"}
]
print(count_messages_tokens(messages, enc)) # ~28 tokensThat function returns the input token cost of the conversation history alone. If you batch multiple users, count each thread separately.
Step 3: Add system prompts, tools, and response format overhead
Developers forget that tool schemas are serialized into the context. A single function definition with JSON schema can eat 80–200 tokens. Count them explicitly, because they are static across turns and easy to overlook.
def count_tools_tokens(tools: list[dict], encoder) -> int:
import json
total = 0
for tool in tools:
serialized = json.dumps(tool, separators=(",", ":"))
total += len(encoder.encode(serialized))
total += 4 # same per-item overhead as messages
return total
tools = [{
"type": "function",
"function": {
"name": "get_weather",
"description": "Fetch weather for a city",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"]
}
}
}]
print(count_tools_tokens(tools, enc))If you pass response_format or seed, those add a few tokens but are usually negligible. Include them if you run tight margins.
Step 4: Reserve space for output and provider headroom
The context limit is a sum of input and max output tokens. If your model has a 128k window but you request max_tokens=4096, your input budget is 128k minus 4096 minus a safety buffer. I reserve 1–2% for undocumented overhead like whitespace normalization.
CONTEXT_LIMIT = 128_000
MAX_OUTPUT = 4_096
SAFETY_BUFFER = int(CONTEXT_LIMIT * 0.02)
def available_input_budget() -> int:
return CONTEXT_LIMIT - MAX_OUTPUT - SAFETY_BUFFERCall this before building the request. If count_messages_tokens + count_tools_tokens exceeds available_input_budget(), trim history before sending.
Step 5: Build a pre-flight validation function
Wrap the pieces into one check that raises or returns a trimmed payload. This is the core of how you count tokens accurately context limit in production.
def preflight_check(messages, tools, encoder, model_limit=CONTEXT_LIMIT):
used = count_messages_tokens(messages, encoder)
used += count_tools_tokens(tools, encoder)
budget = model_limit - MAX_OUTPUT - SAFETY_BUFFER
if used > budget:
while used > budget and len(messages) > 1:
removed = messages.pop(1)
used -= count_messages_tokens([removed], encoder)
if used > budget:
raise ValueError("System prompt alone exceeds context budget")
return messages, toolsRun this synchronously before the HTTP call. It adds sub-millisecond latency for normal payloads and prevents expensive failed round-trips.
Step 6: Verify against real provider usage
Estimates are only as good as your overhead constants. After the call, inspect the usage field. For an OpenAI-compatible response:
{
"usage": {
"prompt_tokens": 42,
"completion_tokens": 128,
"total_tokens": 170
}
}Diff prompt_tokens against your pre-flight sum. If the delta is consistently positive, increase tokens_per_message. If you route through n4n.ai, its per-token usage metering returns exact counts from the upstream provider, so you can calibrate your local estimator without standing up separate provider accounts.
Log the delta in tests:
assert abs(estimated - usage.prompt_tokens) <= 5, "Tokenizer drift too high"Step 7: Recount on streaming appends and multi-turn loops
In agentic loops, you append tool results and re-call. Each append changes the count. Do not cache the initial number.
def agent_step(messages, tool_result, encoder):
messages.append({"role": "tool", "content": tool_result})
messages, _ = preflight_check(messages, tools, encoder)
return call_model(messages)If you stream the assistant message and then fold it back as history, encode the final text, not the streamed chunks, to avoid double counting control tokens.
Step 8: Handle provider-specific limits and fallbacks
Different models behind the same endpoint often have different context sizes. A request routed to llama-3-70b may have 8k limit while gpt-4o has 128k. Your code should branch on the resolved model, not the requested alias.
When a provider is degraded, some gateways auto-route to a fallback. That fallback may have a smaller window. If you count tokens accurately context limit for the primary but the gateway shifts to a smaller model, you still get a 400. Query the model metadata endpoint first if your gateway exposes it, or set a conservative global limit.
model_meta = client.models.retrieve("gpt-4o")
limit = model_meta.context_window # use real field if availableStep 9: Centralize the check in a single middleware function
If multiple services call the model, duplicate counters drift. Put preflight_check behind your HTTP client wrapper. In FastAPI or a LangChain callback, intercept the request body, count, trim, and forward. This guarantees every path counts tokens accurately context limit.
async def routed_chat_completion(request: dict):
encoder = get_encoder(request.get("model", "gpt-4o"))
request["messages"], request["tools"] = preflight_check(
request["messages"], request.get("tools", []), encoder
)
return await upstream_post(request)Now the counting logic is tested once and reused across your stack.
How to verify success
Success means zero context_length_exceeded errors in production and a stable estimator delta under 5 tokens across your test corpus. Implement a unit test with a captured real conversation:
def test_preflight_matches_usage():
enc = get_encoder("gpt-4o")
msgs = load_fixture("long_chat.json")
est = count_messages_tokens(msgs, enc)
real = call_and_get_prompt_tokens(msgs)
assert abs(est - real) <= 5Run this in CI against a mock that returns fixed usage. Then spot-check weekly against live traffic logs. When the estimator and the provider agree, you have solved the silent truncation problem.
Practical trimming strategies
Dropping the oldest message is naive. Prefer summarization: compress the first N messages with a cheap model, then keep the summary as a single system note. That preserves context while cutting tokens by 5–10x.
def summarize_history(old_msgs, encoder, client):
if len(old_msgs) < 4:
return old_msgs
to_compress = old_msgs[:-2]
summary = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role":"system","content":"Compress to 2 sentences"}] + to_compress
).choices[0].message.content
return [{"role":"system","content":f"Summary: {summary}"}] + old_msgs[-2:]This keeps recent turns verbatim for coherence.
Edge cases that break naive counters
- Unicode:
tiktokenhandles it; character division does not. - Base64 images in multimodal models: count by token cost per tile, not by string length.
- Function calling with parallel calls: each call result is a separate message with overhead.
- Cache control hints: some providers treat certain prefixes as cached, but the input tokens still count toward the limit on first send.
Address these by extending the counting functions with type checks.
Closing checklist
- Use model-specific tokenizer, not char/4.
- Count roles, tools, system prompt, and output reservation.
- Run pre-flight on every request, including loops.
- Diff against real
usage.prompt_tokensin logs. - Trim via summarization, not blind drops.
Following these steps lets you count tokens accurately context limit without surprise mid-flight failures. The code above is minimal but production-shaped; adapt the overhead constants to your gateway’s serialization.