Trim LangChain prompts with tiktoken to save on tokens

Trimming prompts with tiktoken is the most direct way to achieve langchain trim prompts tiktoken token savings in production. LangChain’s trim_messages utility combined with tiktoken’s exact token counting lets you fit conversation history into context windows without guessing. This guide walks through the complete implementation, from basic setup to production-ready strategies you can verify at each step.

Step 1: understand what needs trimming and why

LLM context windows are fixed. GPT-4o offers 128k tokens, but most models sit between 4k and 32k. Every message in your conversation history — system prompts, few-shot examples, user turns, assistant responses — consumes tokens. When you exceed the limit, the request fails or gets silently truncated by the provider.

LangChain’s trim_messages solves this by removing older messages until the total fits within a token budget. But it needs an accurate token counter. That’s where tiktoken comes in: it replicates OpenAI’s exact byte-pair encoding, so what you count locally matches what the API bills.

Install the dependencies:

pip install langchain-core tiktoken

Step 2: set up a token counter with tiktoken

LangChain expects a callable that takes a list of messages and returns an integer token count. Tiktoken provides encoding_for_model for OpenAI models and get_encoding for generic cl100k_base (used by GPT-4, GPT-3.5-turbo, and many others).

import tiktoken
from langchain_core.messages import BaseMessage
from typing import List

def count_tokens(messages: List[BaseMessage], model: str = "gpt-4o") -> int:
    """Count tokens for a list of LangChain messages using tiktoken."""
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        # Fallback for models not in tiktoken's registry
        encoding = tiktoken.get_encoding("cl100k_base")
    
    # Convert messages to the format tiktoken expects
    # Each message: role + content + overhead (~4 tokens per message)
    token_count = 0
    for msg in messages:
        # Role tokens (system/user/assistant) + content
        token_count += len(encoding.encode(msg.type))  # "system", "user", "assistant"
        token_count += len(encoding.encode(msg.content))
        token_count += 4  # Approximate overhead per message (role formatting, etc.)
    
    # Add priming tokens for the assistant reply
    token_count += 3
    return token_count

Verify it works:

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

messages = [
    SystemMessage(content="You are a helpful assistant."),
    HumanMessage(content="What is the capital of France?"),
    AIMessage(content="The capital of France is Paris."),
    HumanMessage(content="Tell me more about Paris."),
]

print(f"Token count: {count_tokens(messages, 'gpt-4o')}")
# Expected: ~60-80 tokens depending on exact content

Run this. If the output looks reasonable (tens of tokens for short messages, hundreds for longer ones), the counter is working.

Step 3: basic trimming with trim_messages

LangChain’s trim_messages (in langchain_core.messages) handles the removal logic. You give it a token counter, a max token limit, and a strategy.

from langchain_core.messages import trim_messages

# Keep the system message, drop oldest human/ai pairs first
trimmer = trim_messages(
    max_tokens=1000,
    token_counter=count_tokens,
    strategy="last",           # Keep most recent messages
    include_system=True,       # Always preserve system prompt
    allow_partial=False,       # Don't split messages
    start_on="human",          # Conversation should start with human turn
)

Test the trimmer:

# Build a conversation that exceeds 1000 tokens
long_messages = [
    SystemMessage(content="You are a helpful assistant."),
]

# Add many turns to push past the limit
for i in range(50):
    long_messages.append(HumanMessage(content=f"Question {i}: " + "x" * 200))
    long_messages.append(AIMessage(content=f"Answer {i}: " + "x" * 200))

print(f"Before trim: {len(long_messages)} messages, {count_tokens(long_messages)} tokens")

trimmed = trimmer.invoke(long_messages)

print(f"After trim: {len(trimmed)} messages, {count_tokens(trimmed)} tokens")
print(f"First message type: {trimmed[0].type}")  # Should be 'system'
print(f"Last message type: {trimmed[-1].type}")  # Should be 'human' or 'ai'

Verify success: The trimmed list should have fewer messages, token count under 1000, system message preserved, and the conversation ending on a human or AI turn (not cut mid-pair).

Step 4: token-aware trimming strategies

The strategy parameter controls which messages get dropped. Choose based on your use case:

Strategy	Behavior	Best for
`"last"`	Keep most recent messages, drop oldest	Chat history, general conversation
`"first"`	Keep oldest messages, drop newest	Preserving few-shot examples, instructions

Strategy “last” (default) — keep recent context:

trimmer_last = trim_messages(
    max_tokens=2000,
    token_counter=lambda msgs: count_tokens(msgs, "gpt-4o"),
    strategy="last",
    include_system=True,
    start_on="human",
)

Strategy “first” — preserve system + few-shot examples:

trimmer_first = trim_messages(
    max_tokens=2000,
    token_counter=lambda msgs: count_tokens(msgs, "gpt-4o"),
    strategy="first",
    include_system=True,
    start_on="human",
)

Verify the difference:

# Conversation: system + 3 few-shot pairs + 10 recent turns
messages = [
    SystemMessage(content="System prompt with instructions."),
    HumanMessage(content="Example 1 input"), AIMessage(content="Example 1 output"),
    HumanMessage(content="Example 2 input"), AIMessage(content="Example 2 output"),
    HumanMessage(content="Example 3 input"), AIMessage(content="Example 3 output"),
]

# Add 10 recent turns
for i in range(10):
    messages.append(HumanMessage(content=f"Recent Q{i}"))
    messages.append(AIMessage(content=f"Recent A{i}"))

print("Original:", len(messages), "messages")

trimmed_last = trimmer_last.invoke(messages)
trimmed_first = trimmer_first.invoke(messages)

print("Strategy 'last' keeps:", [m.content[:30] for m in trimmed_last])
print("Strategy 'first' keeps:", [m.content[:30] for m in trimmed_first])

With "last", you keep the 10 recent turns and drop few-shots. With "first", you keep the few-shots and drop recent turns. Pick based on whether your task needs examples or recency.

Step 5: handle conversation history in chains

In real applications, you trim inside a chain or runnable. The cleanest pattern: wrap the trimmer in a RunnableLambda and place it before the model call.

from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o", temperature=0)

# Trimmer configured for your context window minus output buffer
# GPT-4o: 128k context, leave ~4k for output + safety margin
trimmer = trim_messages(
    max_tokens=120_000,
    token_counter=lambda msgs: count_tokens(msgs, "gpt-4o"),
    strategy="last",
    include_system=True,
    start_on="human",
)

# Chain: trim -> model
chain = (
    RunnablePassthrough.assign(messages=RunnableLambda(trimmer))
    | model
)

# Invoke with full history
response = chain.invoke({
    "messages": long_conversation_history  # Your full message list
})

Alternative: trim in a prompt template

If you use ChatPromptTemplate with a MessagesPlaceholder, trim the variable before it reaches the template:

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

def prepare_inputs(inputs: dict) -> dict:
    """Trim history to fit context window."""
    trimmed = trimmer.invoke(inputs["history"])
    return {"history": trimmed, "input": inputs["input"]}

chain = RunnableLambda(prepare_inputs) | prompt | model

Verify the chain works:

# Test with a history that would exceed a small limit for demo
test_trimmer = trim_messages(
    max_tokens=500,
    token_counter=lambda msgs: count_tokens(msgs, "gpt-4o"),
    strategy="last",
    include_system=True,
    start_on="human",
)

test_chain = RunnableLambda(lambda x: {"messages": test_trimmer.invoke(x["messages"])}) | model

# This would fail without trimming
huge_history = [
    SystemMessage(content="System prompt."),
] + [
    HumanMessage(content="Q" * 100), AIMessage(content="A" * 100)
    for _ in range(20)
]

result = test_chain.invoke({"messages": huge_history})
print(f"Response: {result.content[:100]}...")

If you get a response instead of a context-length error, the trimming is working in-chain.

Step 6: verify token savings with real metrics

Don’t guess — measure. Log token counts before and after trimming, and compare against provider-reported usage.

import json
from datetime import datetime

def log_token_savings(original_messages, trimmed_messages, model="gpt-4o"):
    """Log token reduction for observability."""
    original = count_tokens(original_messages, model)
    trimmed = count_tokens(trimmed_messages, model)
    savings = original - trimmed
    pct = (savings / original * 100) if original > 0 else 0
    
    log_entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "model": model,
        "original_tokens": original,
        "trimmed_tokens": trimmed,
        "tokens_saved": savings,
        "savings_pct": round(pct, 1),
        "original_msg_count": len(original_messages),
        "trimmed_msg_count": len(trimmed_messages),
    }
    print(json.dumps(log_entry))
    return log_entry

# Usage in your chain
def trim_and_log(messages):
    trimmed = trimmer.invoke(messages)
    log_token_savings(messages, trimmed)
    return trimmed

chain_with_logging = (
    RunnableLambda(lambda x: {"messages": trim_and_log(x["messages"])})
    | model
)

Sample output:

{
  "timestamp": "2025-01-15T14:32:11.456Z",
  "model": "gpt-4o",
  "original_tokens": 45230,
  "trimmed_tokens": 119847,
  "tokens_saved": 33243,
  "savings_pct": 73.5,
  "original_msg_count": 142,
  "trimmed_msg_count": 48
}

This gives you concrete data for cost reporting and capacity planning.

Step 7: production considerations

Model-specific token counters

Different models use different encodings. Create a registry:

MODEL_ENCODINGS = {
    "gpt-4o": "o200k_base",
    "gpt-4o-mini": "o200k_base",
    "gpt-4-turbo": "cl100k_base",
    "gpt-3.5-turbo": "cl100k_base",
    "text-embedding-3-large": "cl100k_base",
    "text-embedding-3-small": "cl100k_base",
}

def get_counter(model: str):
    encoding_name = MODEL_ENCODINGS.get(model, "cl100k_base")
    encoding = tiktoken.get_encoding(encoding_name)
    
    def counter(messages: List[BaseMessage]) -> int:
        total = 0
        for msg in messages:
            total += len(encoding.encode(msg.type))
            total += len(encoding.encode(msg.content))
            total += 4
        total += 3
        return total
    return counter

Handle non-OpenAI models

For Anthropic, Google, or open models, you need their tokenizers. The principle is the same — swap the counter:

# Anthropic example (requires anthropic package)
try:
    from anthropic import Anthropic
    anthropic_client = Anthropic()
    
    def anthropic_counter(messages: List[BaseMessage]) -> int:
        # Convert to Anthropic format
        anthropic_msgs = [
            {"role": "user" if m.type == "human" else "assistant" if m.type == "ai" else "system", 
             "content": m.content}
            for m in messages
        ]
        return anthropic_client.count_tokens(anthropic_msgs)
except ImportError:
    anthropic_counter = None

Budget for output tokens

Always reserve space for the model’s response. A safe formula:

def calculate_max_input_tokens(context_window: int, max_output: int, safety_margin: int = 1000) -> int:
    """Calculate max input tokens given context window and expected output."""
    return context_window - max_output - safety_margin

# GPT-4o: 128k context, expect up to 4k output
MAX_INPUT = calculate_max_input_tokens(128_000, 4_000)  # 123,000

trimmer = trim_messages(
    max_tokens=MAX_INPUT,
    token_counter=get_counter("gpt-4o"),
    strategy="last",
    include_system=True,
    start_on="human",
)

Streaming and trim_messages

trim_messages works with streaming chains too — it runs before the model call, so streaming is unaffected:

async def stream_with_trim(messages):
    trimmed = trimmer.invoke(messages)
    async for chunk in model.astream(trimmed):
        yield chunk

Step 8: common pitfalls and fixes

Pitfall 1: Token counter mismatch

If your local count differs from the provider’s bill, check:

Are you using the right encoding for the model?
Are you accounting for message formatting overhead (the +4 per message)?
Does the provider count function/tool call tokens differently?

Fix: Compare your counter against the provider’s reported prompt_tokens for a few requests and calibrate the overhead constant.

Pitfall 2: System message dropped accidentally

include_system=True preserves the first system message. If you have multiple system messages (e.g., injected context), only the first survives. Fix: Consolidate system content into one message before trimming.

Pitfall 3: Conversation starts with AI message

start_on="human" ensures the trimmed conversation begins with a human turn. If your history starts with AI (rare), either prepend a dummy human message or use start_on="any".

Pitfall 4: Tool calls and structured output

Messages with tool_calls or tool_call_id have extra token overhead. Extend your counter:

def count_tokens_with_tools(messages: List[BaseMessage], model: str) -> int:
    encoding = tiktoken.encoding_for_model(model)
    total = 0
    for msg in messages:
        total += len(encoding.encode(msg.type))
        total += len(encoding.encode(msg.content or ""))
        # Tool calls add significant tokens
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            for tc in msg.tool_calls:
                total += len(encoding.encode(tc["name"]))
                total += len(encoding.encode(json.dumps(tc["args"])))
        total += 4
    total += 3
    return total

Verification checklist

Before deploying, confirm each item:

Token counter matches provider billing within ~5% for sample conversations
Trimmer preserves system message (include_system=True)
Trimmed conversation starts with human turn (start_on="human")
Token budget leaves room for max expected output + safety margin
Chain integrates trimmer before model call
Logging captures original vs. trimmed token counts
Works with your longest realistic conversation history
Handles tool calls / structured output if used

Summary

You now have a complete, verifiable pipeline for langchain trim prompts tiktoken token savings:

Count accurately with tiktoken’s model-specific encodings
Trim strategically using trim_messages with strategy="last" for chat or "first" for few-shot preservation
Integrate cleanly via RunnableLambda before your model call
Measure continuously by logging before/after token counts
Reserve budget for output tokens and safety margins

The code here runs as-is with LangChain 0.2+ and tiktoken 0.7+. Adjust the token budget for your specific model’s context window, and you’ll stop hitting context-length errors while keeping the most relevant conversation history.