Compaction: Compressing Long Conversations
On this page
The Problem with Long Conversations
Every message added to a conversation increases the number of tokens sent to the API. A 50-turn conversation can easily reach 30,000 input tokens, driving up both latency and cost. The issue is insidious: the first turn of a support assistant costs a few hundred tokens, while the thirtieth turn costs several thousand even though the user is asking the exact same type of question. You pay more and more for an identical service, and the user waits longer and longer for a response.
Compaction solves this by compressing history without losing essential information. It involves summarizing older exchanges to reduce token count while preserving the necessary context. Instead of sending the entire history, you send a summary of past exchanges plus the recent messages. The model still knows the customer’s name is Dupont, that they have a disputed order, and that you already offered a refund, without having to reread the forty messages that established those facts.
The Mechanism
The function below applies this logic in its simplest form: it estimates the volume, does nothing until the threshold is crossed, then splits the history into two parts. Older messages go to the summary; the most recent ones remain as-is.
import openai
client = openai.OpenAI()
def compact_history(
messages: list[dict],
token_threshold: int = 10_000,
keep_recent: int = 6,
) -> list[dict]:
"""Compacts history when it exceeds the threshold."""
# Estimate token count (approximation)
estimated_tokens = sum(len(m["content"]) // 4 for m in messages)
if estimated_tokens < token_threshold:
return messages # No need to compact
# Split: old messages vs recent ones
old_messages = messages[:-keep_recent]
recent_messages = messages[-keep_recent:]
# Summarize old messages
old_text = "\n".join(
f"{m['role']}: {m['content']}" for m in old_messages
)
response = client.responses.create(
model="gpt-5.6-terra",
input=(
"Summarize this conversation concisely. "
"Keep key factual information, decisions made, "
"and the context needed to understand what follows.\n\n"
f"{old_text}"
),
max_output_tokens=500,
)
# Rebuild compacted history
summary_message = {
"role": "system",
"content": f"Summary of previous conversation:\n{response.output_text}",
}
return [summary_message] + recent_messages
Two details deserve your attention. The summary prompt explicitly requests decisions made because a summary that only retains themes causes the model to lose memory of its own commitments. And max_output_tokens=500 caps the summary: without this limit, the summary grows with every compaction, and you end up with the very problem you were trying to eliminate.
Two Strategies to Know
Sliding Window with Summary
The following class industrializes the principle. It maintains a single summary that it rewrites at each compaction by integrating new exchanges, while keeping a window of raw messages. The context sent to the API therefore always consists of three blocks: the system prompt, the cumulative summary, and the recent window.
class CompactedConversation:
"""Manages a conversation with automatic compaction."""
def __init__(
self,
system_prompt: str,
compaction_threshold: int = 8_000,
window_size: int = 8,
):
self.system_prompt = system_prompt
self.threshold = compaction_threshold
self.window_size = window_size
self.summary: str | None = None
self.messages: list[dict] = []
def build_context(self) -> list[dict]:
"""Builds the context to send to the API."""
context = [{"role": "system", "content": self.system_prompt}]
if self.summary:
context.append({
"role": "system",
"content": f"Previous context: {self.summary}",
})
context.extend(self.messages[-self.window_size:])
return context
def add_message(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
self._check_compaction()
def _check_compaction(self):
tokens = sum(len(m["content"]) // 4 for m in self.messages)
if tokens > self.threshold:
self._compact()
def _compact(self):
old_messages = self.messages[:-self.window_size]
if not old_messages:
return
text = "\n".join(
f"{m['role']}: {m['content']}" for m in old_messages
)
existing_context = f"Existing summary: {self.summary}\n\n" if self.summary else ""
response = client.responses.create(
model="gpt-5.6-terra",
input=(
f"{existing_context}"
f"New exchanges to integrate into the summary:\n{text}\n\n"
"Produce an updated, concise, and factual summary."
),
max_output_tokens=400,
)
self.summary = response.output_text
self.messages = self.messages[-self.window_size:]
Note that _compact reinjects the existing summary into the prompt before the new exchanges. This prevents progressive amnesia: without this reinjection, each compaction would erase everything the previous one had retained.
Selective Compaction
Summarization is destructive, and some content does not survive reformulation. A code snippet, a JSON API response, or a table of figures loses its utility as soon as it is paraphrased. Selective compaction therefore sorts old messages before summarizing: those containing code or structured data pass through intact, while only conversational text is compressed.
def selective_compaction(
messages: list[dict],
keep_recent: int = 6,
) -> list[dict]:
"""Compacts while keeping structured messages intact."""
old_messages = messages[:-keep_recent]
recent_messages = messages[-keep_recent:]
to_summarize = []
to_keep = []
for msg in old_messages:
content = msg["content"]
# Keep messages containing code or data
if "```" in content or "{" in content:
to_keep.append(msg)
else:
to_summarize.append(msg)
# Summarize only conversational text
summary = ""
if to_summarize:
text = "\n".join(f"{m['role']}: {m['content']}" for m in to_summarize)
response = client.responses.create(
model="gpt-5.6-terra",
input=f"Summarize these exchanges in 3 sentences:\n{text}",
max_output_tokens=200,
)
summary = response.output_text
result = []
if summary:
result.append({
"role": "system",
"content": f"Summary of exchanges: {summary}",
})
result.extend(to_keep)
result.extend(recent_messages)
return result
The sorting criterion used here—presence of triple backticks or a curly brace—is intentionally crude. Adapt it to what your tools actually produce: for an assistant that handles case references, you might want to preserve any message containing a customer ID.
Verifying That Compaction Is Worth It
Poorly calibrated compaction costs an extra API call with no noticeable gain, or compresses so aggressively that the model loses the thread and asks questions already resolved. Therefore, measure the ratio before fixing your final thresholds, using a sample of real conversations.
def compaction_ratio(
before: list[dict],
after: list[dict],
) -> dict:
tokens_before = sum(len(m["content"]) // 4 for m in before)
tokens_after = sum(len(m["content"]) // 4 for m in after)
reduction = 1 - (tokens_after / tokens_before) if tokens_before else 0
return {
"tokens_before": tokens_before,
"tokens_after": tokens_after,
"reduction": f"{reduction:.0%}",
"messages_before": len(before),
"messages_after": len(after),
}
Take a dozen transcripts from your production environment, run them through compaction, and read the produced summaries: if you cannot resume the conversation based on the summary alone, your threshold is too low or your summary prompt too vague.
Key Takeaways
- Compaction reduces tokens by summarizing older messages
- Always keep the most recent messages intact (sliding window)
- Preserve messages containing code or structured data
- Reinject the existing summary at every new compaction to avoid progressive amnesia
- Measure the compression ratio to calibrate your thresholds