Agents that stay useful across long sessions accumulate thousands of tokens of chat logs, and most of those tokens are redundant. To compress conversation history LLM context without losing accuracy, you need a pipeline that separates signal from noise, summarizes the noise, and keeps structured state intact. The following steps show a concrete Python implementation you can wire into any agent loop.
Step 1: Baseline your token footprint
Before you compress conversation history LLM traffic, measure what you are actually sending. Guesswork leads to either over-compression (lost instructions) or under-compression (overspend). Use tiktoken with the model’s encoding to count per-message tokens, including the framing overhead that OpenAI-style APIs add.
import tiktoken
def count_tokens(messages, model="gpt-4o"):
enc = tiktoken.encoding_for_model(model)
n = 0
for m in messages:
n += 4 # role/name framing per message
n += len(enc.encode(m.get("content", "")))
n += 2 # priming reply
return n
# messages: list of {"role": "user"/"assistant"/"system", "content": str}
print(f"Total tokens: {count_tokens(messages)}")Run this on a week of real agent transcripts. You will typically find that 70–90% of tokens are old tool outputs, repeated confirmations, or intermediate reasoning that the current turn does not need. That ratio is your compression budget. If a session averages 12k tokens and only 2k are recent, you can safely target a 4k context without touching the active window.
Step 2: Tag messages with retention tiers
Not all messages deserve equal space. Assign each turn a tier based on role, recency, and length:
- T0 (keep verbatim): system prompt, latest user intent, active task variables.
- T1 (summarize): past user requests, tool results older than K turns.
- T2 (drop): acknowledgements, “got it”, redundant status polls.
A heuristic classifier is enough to start. The window size should match your model’s reasoning needs; six turns is a sane default for support bots, twenty for coding agents.
def tier(msg, turn_index, current_turn, window=6):
if msg["role"] == "system":
return "T0"
if current_turn - turn_index <= window:
return "T0" if msg["role"] == "user" else "T1"
if len(msg["content"]) < 40 and msg["role"] == "assistant":
return "T2"
return "T1"Avoid the trap of labeling by role alone. A tool result from three turns ago may contain a JSON schema the user asked to modify; that is T0, not T1. The heuristic above is a starting point; audit a sample of mis-tiered messages weekly. You can later replace the heuristic with a small classifier, but the tiers themselves are stable. The key is that T0 is always sufficient to continue the task; T1 is everything that explains how you got here.
Step 3: Summarize disposable turns
For T1 messages, call a cheap instruction-tuned model to produce a dense summary. Route this to a smaller model to save cost. Using an OpenAI-compatible gateway such as n4n.ai gives you automatic fallback across providers and per-token metering without writing retry logic.
from openai import OpenAI
import os
client = OpenAI(
base_url="https://api.n4n.ai/v1",
api_key=os.environ["N4N_API_KEY"],
)
def summarize(batch, client):
resp = client.chat.completions.create(
model="mistralai/mixtral-8x7b-instruct",
messages=[
{"role": "system", "content": "Compress the following agent turns into 3 bullet points retaining user intent, decisions, and tool outcomes."},
{"role": "user", "content": "\n".join(f'{m["role"]}: {m["content"]}' for m in batch)}
],
max_tokens=200,
)
return resp.choices[0].message.content
summary = summarize([m for i, m in enumerate(messages) if tier(m, i, cur) == "T1"], client)Keep the summary block under a fixed token cap (e.g., 256 tokens). If the batch is larger, split and summarize hierarchically: summarize chunks of 8–12 turns, then summarize the summaries. This bounds cost and avoids truncating mid-fact. If you summarize 100 turns in one call, the model will drop details. Store each intermediate summary with a timestamp so you can debug later.
Step 4: Pack the compressed context
Assemble the final request: T0 messages verbatim, the summary block injected as a single system note, and the most recent window. This is where you compress conversation history LLM payloads into a stable size.
def pack_context(messages, summary, current_turn, window=6):
out = []
for i, m in enumerate(messages):
t = tier(m, i, current_turn, window)
if t == "T0":
out.append(m)
elif t == "T1" and i == 0: # keep first user intent if missed
out.append(m)
out.insert(1, {"role": "system", "content": f"Prior context summary:\n{summary}"})
return outIn practice, prepend the summary right after the system prompt. The live model never sees raw old turns, only the distilled facts. If you use a provider that supports cache prefixes, mark the summary block as a cache breakpoint so repeated turns reuse the same compressed prefix. If the summary itself grows beyond budget, promote the oldest T0 messages to a secondary summary and keep only the last two summaries. This creates a rolling context that bounds tokens strictly.
Step 5: Extract mutable state explicitly
Summaries lose exact values. Before compression, pull structured state into a JSON blob that travels alongside the context. The model should not need to infer an order ID from a paraphrased sentence.
import json
state_schema = {
"order_id": None,
"user_timezone": None,
"pending_tasks": []
}
def extract_state(messages, client):
resp = client.chat.completions.create(
model="mistralai/mixtral-8x7b-instruct",
messages=[
{"role": "system", "content": f"Extract state as JSON matching {state_schema}"},
{"role": "user", "content": str(messages)}
],
response_format={"type": "json_object"}
)
return json.loads(resp.choices[0].message.content)
state = extract_state(messages, client)Pass state to tools directly. Your agent logic reads state["order_id"] instead of asking the LLM to recall it. This eliminates the most common accuracy regression from compression.
Step 6: Verify accuracy with replay
Compression is only safe if the agent still completes the task. Build a replay harness:
- Take 50 historical conversations with known successful endings.
- Run the agent twice: once with full history, once with your compression pipeline.
- Compare final tool calls and extracted state.
def replay(convo, use_compression, client):
if use_compression:
cur = len(convo)
summary = summarize([m for i, m in enumerate(convo) if tier(m, i, cur) == "T1"], client)
ctx = pack_context(convo, summary, cur)
state = extract_state(convo, client)
else:
ctx, state = convo, None
return agent_run(ctx, state)
baseline = [replay(c, False, client) for c in convos]
compressed = [replay(c, True, client) for c in convos]
assert all(b.state == c.state for b, c in zip(baseline, compressed))
assert all(b.final_action == c.final_action for b, c in zip(baseline, compressed))If state matches and the final action type matches in >95% of cases, your compression is safe. Investigate the mismatches—they usually reveal a T1 message that should be T0. For the remaining <5%, add a targeted rule (e.g., “always keep messages containing ‘order_id’ verbatim”). Track not just action match but user-facing response similarity. Use embedding cosine similarity >0.9 on final answers as a secondary signal. This catches cases where state is correct but phrasing diverges enough to confuse the user.
Operational notes
When you compress conversation history LLM contexts at scale, cache the summary block. Gateways that forward provider cache-control hints let you mark the summary as a stable prefix so you are not re-billed for it every turn. Honor client routing directives to pin summarization to a specific cheap region.
The pipeline above is deliberately simple. In production, add incremental summarization (summarize only new turns since last compaction), token budgets per request, and a fallback to keep last N raw turns when the summarizer is degraded. The moment you treat conversation history as a mutable cache rather than an append-only log, your agent gets cheaper and faster without getting dumber.