Prompt Caching for Cost Reduction with LangGraph: Complete Guide

Prompt Caching for Cost Reduction with LangGraph: Complete Guide

As LLM-powered applications grow in complexity, the cost of repeated API calls becomes a significant concern. LangGraph, with its stateful graph-based orchestration, often reuses large context blocks across nodes and iterations. Prompt caching—particularly Anthropic's prompt caching feature—offers a powerful way to reduce both latency and cost when working with large, repeated prompt prefixes. In this guide, you'll learn what prompt caching is, why it matters in LangGraph workflows, and how to implement it effectively with practical code examples.

What Is Prompt Caching?

Prompt caching is a feature offered by LLM providers (most notably Anthropic Claude) that allows the model to cache the processing of a prompt prefix. When subsequent requests share that same prefix, the cached computation is reused instead of being recomputed from scratch. This results in two major benefits:

• Reduced cost: Cached tokens are billed at a significantly lower rate (often 90% cheaper than standard input tokens on Anthropic).
• Reduced latency: The model skips reprocessing the cached prefix, leading to faster time-to-first-token.

The cache has a time-to-live (TTL), typically 5 minutes by default, which can be extended to 1 hour on some providers. Each cache hit refreshes the TTL. The key constraint is that the prefix must match exactly—any change in a single token invalidates the cache from that point forward.

Why Prompt Caching Matters in LangGraph

LangGraph applications frequently involve multi-step workflows where the same context is passed between nodes. Common patterns include:

• Large system prompts with tool definitions, instructions, and few-shot examples.
• Long conversation histories maintained in graph state.
• Retrieved documents that remain stable across multiple reasoning steps.
• Agent loops where the same context is re-sent on each iteration.

Without caching, every node invocation that includes this large context pays full price for input tokens. In an agent loop that runs 10 iterations with a 10,000-token system prompt and document set, you'd pay for 100,000 input tokens just for the repeated context. With caching, you pay full price once and a fraction of the cost for the remaining 9 iterations. This can reduce your bill by 80% or more in such scenarios.

How Prompt Caching Works

Anthropic's API allows you to mark specific blocks in a message as cacheable using a cache_control parameter. You can place up to 4 cache breakpoints in a single request. The cache is keyed on the exact content preceding each breakpoint. When a new request shares that prefix, the cached portion is reused.

The structure typically looks like this:

[
  {
    "role": "system",
    "content": [
      { "type": "text", "text": "You are a helpful assistant..." },
      { "type": "text", "text": "[Large document context]", "cache_control": { "type": "ephemeral" } }
    ]
  },
  {
    "role": "user",
    "content": "What does the document say about X?"
  }
]

The cache_control marker tells the API to cache everything up to and including that block. Subsequent requests that include the same system message content will hit the cache.

Setting Up Your Environment

Before diving into code, install the required packages and set up your API keys:

pip install langgraph langchain-anthropic langchain-core

Set your Anthropic API key as an environment variable:

export ANTHROPIC_API_KEY="your-api-key-here"

Basic Prompt Caching with LangChain and Anthropic

LangChain's Anthropic integration supports prompt caching through the cache_control parameter on message content blocks. Let's start with a basic example before integrating it into LangGraph:

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage

llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
)

# Large context that we want to cache
large_context = "You are an expert legal assistant. " * 500  # Simulating a large prompt

system_message = SystemMessage(
    content=[
        {
            "type": "text",
            "text": large_context,
            "cache_control": {"type": "ephemeral"},
        }
    ]
)

# First call - cache is written
response1 = llm.invoke([system_message, HumanMessage(content="What is your role?")])
print("First call (cache write):", response1.usage_metadata)

# Second call - cache is hit
response2 = llm.invoke([system_message, HumanMessage(content="Summarize your capabilities.")])
print("Second call (cache hit):", response2.usage_metadata)

In the usage metadata, you'll see fields like cache_input_tokens and cache_read_input_tokens. The first call shows cache write tokens, while the second shows cache read tokens—confirming the cache was used.

Integrating Prompt Caching into LangGraph

Now let's build a LangGraph workflow that leverages prompt caching. We'll create a research agent that maintains a large document context across multiple reasoning steps.

from langgraph.graph import StateGraph, START, END
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages
import os

# Initialize the model
llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    max_tokens=2048,
)

# Define the graph state
class AgentState(TypedDict):
    messages: Annotated[List, add_messages]
    document_context: str
    question: str
    answer: str
    needs_more_info: bool

# A large document context that we want to cache
DOCUMENT_CONTEXT = """
[Large document content - e.g., a 50-page research paper, legal contract,
or technical specification. This content remains stable across all graph
iterations, making it an ideal candidate for prompt caching.]
""" * 100  # Simulating a large document

def create_cached_system_message(document_context: str) -> SystemMessage:
    """Create a system message with cache_control enabled."""
    return SystemMessage(
        content=[
            {
                "type": "text",
                "text": "You are a research assistant. Use the provided document context to answer questions accurately.",
            },
            {
                "type": "text",
                "text": f"Document context:\n{document_context}",
                "cache_control": {"type": "ephemeral"},
            },
        ]
    )

def research_node(state: AgentState) -> dict:
    """Process the question using the cached document context."""
    system_msg = create_cached_system_message(state["document_context"])
    
    messages = [system_msg] + state["messages"]
    response = llm.invoke(messages)
    
    return {
        "messages": [response],
        "answer": response.content,
        "needs_more_info": "need more information" in response.content.lower(),
    }

def review_node(state: AgentState) -> dict:
    """Review the answer and decide if refinement is needed."""
    system_msg = create_cached_system_message(state["document_context"])
    
    review_prompt = HumanMessage(
        content=f"Review this answer for completeness: {state['answer']}. If it's complete, say 'COMPLETE'. Otherwise, say what's missing."
    )
    
    messages = [system_msg] + state["messages"] + [review_prompt]
    response = llm.invoke(messages)
    
    needs_refinement = "COMPLETE" not in response.content.upper()
    
    return {
        "messages": [response],
        "needs_more_info": needs_refinement,
    }

def should_refine(state: AgentState) -> str:
    """Conditional edge to determine next step."""
    if state.get("needs_more_info", False):
        return "research"
    return END

# Build the graph
workflow = StateGraph(AgentState)

workflow.add_node("research", research_node)
workflow.add_node("review", review_node)

workflow.add_edge(START, "research")
workflow.add_edge("research", "review")
workflow.add_conditional_edges("review", should_refine, {
    "research": "research",
    END: END,
})

graph = workflow.compile()

In this workflow, both the research_node and review_node use the same cached system message containing the large document context. The first call writes the cache, and all subsequent calls—whether from the review node or from looping back to research—read from it.

Running the Graph and Observing Cache Hits

Let's run the graph and track token usage to verify caching is working:

initial_state = {
    "messages": [HumanMessage(content="What are the key findings in the document?")],
    "document_context": DOCUMENT_CONTEXT,
    "question": "What are the key findings?",
    "answer": "",
    "needs_more_info": False,
}

# Track total cost
total_input_tokens = 0
total_cache_read_tokens = 0
total_cache_creation_tokens = 0

# Wrap the LLM to track usage
original_invoke = llm.invoke

def tracking_invoke(messages, **kwargs):
    response = original_invoke(messages, **kwargs)
    usage = response.usage_metadata
    global total_input_tokens, total_cache_read_tokens, total_cache_creation_tokens
    total_input_tokens += usage.get("input_tokens", 0)
    total_cache_read_tokens += usage.get("cache_read_input_tokens", 0)
    total_cache_creation_tokens += usage.get("cache_creation_input_tokens", 0)
    print(f"  Input: {usage.get('input_tokens', 0)}, "
          f"Cache read: {usage.get('cache_read_input_tokens', 0)}, "
          f"Cache write: {usage.get('cache_creation_input_tokens', 0)}")
    return response

llm.invoke = tracking_invoke

result = graph.invoke(initial_state)

print("\n=== Final Answer ===")
print(result["answer"])
print("\n=== Token Usage Summary ===")
print(f"Total regular input tokens: {total_input_tokens}")
print(f"Total cache read tokens: {total_cache_read_tokens}")
print(f"Total cache creation tokens: {total_cache_creation_tokens}")

# Restore original
llm.invoke = original_invoke

You should observe that the first invocation has high cache_creation_input_tokens (the cache write), while subsequent invocations show high cache_read_input_tokens and very low regular input_tokens. This confirms the cache is working.

Advanced Pattern: Caching Tool Definitions

In agentic workflows, tool definitions can be large and are sent with every request. You can cache them too:

from langchain_core.tools import tool

@tool
def search_database(query: str) -> str:
    """Search the internal database for relevant records.
    
    Args:
        query: Natural language search query
        
    Returns:
        Matching records as formatted text
    """
    # Simulated database search
    return f"Results for: {query}"

@tool
def analyze_sentiment(text: str) -> str:
    """Analyze the sentiment of given text.
    
    Args:
        text: Text to analyze
        
    Returns:
        Sentiment analysis result
    """
    return "Positive sentiment detected"

# Bind tools with caching
llm_with_tools = llm.bind_tools(
    [search_database, analyze_sentiment],
    # The tool definitions are included in the system prompt
    # and will be cached along with other system content
)

def agent_node(state: AgentState) -> dict:
    system_msg = SystemMessage(
        content=[
            {
                "type": "text",
                "text": "You are a helpful agent with access to tools. Use them when needed.",
            },
            {
                "type": "text",
                "text": f"Context: {state['document_context']}",
                "cache_control": {"type": "ephemeral"},
            },
        ]
    )
    
    messages = [system_msg] + state["messages"]
    response = llm_with_tools.invoke(messages)
    
    return {"messages": [response]}

Advanced Pattern: Caching Conversation History

For chat applications with long histories, you can cache the conversation prefix while keeping the latest messages uncached. This is especially useful in LangGraph's MessagesState pattern:

from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, AnyMessage
from langgraph.graph import StateGraph, START, END, MessagesState

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", max_tokens=1024)

SYSTEM_PROMPT = "You are a knowledgeable assistant with expertise in many fields."

def build_cached_messages(messages: List[AnyMessage]) -> List[AnyMessage]:
    """Build message list with cache_control on the system prompt
    and on the conversation history prefix."""
    
    result = []
    
    # System prompt - always cached
    result.append(SystemMessage(
        content=[
            {
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"},
            }
        ]
    ))
    
    # If we have enough conversation history, cache the older portion
    if len(messages) > 4:
        # Cache everything except the last 2 messages
        history_to_cache = messages[:-2]
        recent_messages = messages[-2:]
        
        # Convert older messages to cached content blocks
        cached_content = []
        for msg in history_to_cache:
            role = "user" if isinstance(msg, HumanMessage) else "assistant"
            cached_content.append({
                "type": "text",
                "text": f"{role}: {msg.content}",
            })
        
        # Mark the last block of the cached history
        if cached_content:
            cached_content[-1]["cache_control"] = {"type": "ephemeral"}
            result.append({
                "role": "user",
                "content": cached_content,
            })
        
        result.extend(recent_messages)
    else:
        result.extend(messages)
    
    return result

def chat_node(state: MessagesState) -> dict:
    cached_messages = build_cached_messages(state["messages"])
    response = llm.invoke(cached_messages)
    return {"messages": [response]}

# Simple graph
workflow = StateGraph(MessagesState)
workflow.add_node("chat", chat_node)
workflow.add_edge(START, "chat")
workflow.add_edge("chat", END)
chat_graph = workflow.compile()

# Simulate a multi-turn conversation
messages = [HumanMessage(content="Tell me about quantum computing.")]
result = chat_graph.invoke({"messages": messages})

messages.append(result["messages"][-1])
messages.append(HumanMessage(content="How does it compare to classical computing?"))
result = chat_graph.invoke({"messages": messages})

messages.append(result["messages"][-1])
messages.append(HumanMessage(content="What are the practical applications?"))
result = chat_graph.invoke({"messages": messages})

print("Final response:", result["messages"][-1].content)

This pattern ensures that as the conversation grows, the older portion is cached and only the newest exchanges are processed at full cost.

Best Practices for Prompt Caching in LangGraph

To get the most out of prompt caching, follow these guidelines:

• Place stable content first: Always put your largest, most stable content (system prompts, documents, tool definitions) at the beginning of the message list. The cache is prefix-based, so any change before a cache breakpoint invalidates it.
• Use cache breakpoints strategically: You get up to 4 cache breakpoints per request. Place them at natural boundaries: after the system prompt, after tool definitions, after retrieved documents, and after conversation history.
• Keep the cache warm: The default TTL is 5 minutes. If your LangGraph workflow has long pauses between steps, consider using the 1-hour TTL ("ttl": "1h") to keep the cache alive longer.
• Minimize cache invalidation: Avoid including timestamps, random IDs, or other dynamic content in cached portions. Put dynamic content after the cache breakpoint.
• Monitor cache hit rates: Track cache_read_input_tokens vs cache_creation_input_tokens in your usage metadata. A healthy caching setup should show significantly more reads than writes.
• Consider minimum cache size: Anthropic requires a minimum of 1024 tokens for Sonnet models and 2048 tokens for Haiku to be cached. Don't bother caching small prompts.
• Cache at the right granularity: In LangGraph, if multiple nodes share the same context, use a helper function (like create_cached_system_message above) to ensure identical message construction across nodes.

Calculating Cost Savings

Let's put together a utility to calculate and compare costs with and without caching:

def calculate_cost_savings(
    num_calls: int,
    cached_tokens: int,
    uncached_tokens: int,
    model: str = "claude-3-5-sonnet"
):
    """
    Calculate cost savings from prompt caching.
    
    Args:
        num_calls: Number of API calls made
        cached_tokens: Number of tokens in the cached prefix
        uncached_tokens: Number of tokens NOT cached (per call)
        model: Model name for pricing
    """
    # Pricing per 1M tokens (example rates, verify current pricing)
    pricing = {
        "claude-3-5-sonnet": {
            "input": 3.00,
            "cache_write": 3.75,
            "cache_read": 0.30,
        },
        "claude-3-5-haiku": {
            "input": 0.80,
            "cache_write": 1.00,
            "cache_read": 0.08,
        },
    }
    
    rates = pricing.get(model, pricing["claude-3-5-sonnet"])
    
    # Without caching: every call pays full input price
    cost_without_cache = num_calls * (cached_tokens + uncached_tokens) * rates["input"] / 1_000_000
    
    # With caching: first call writes cache, rest read from it
    cost_with_cache = (
        # First call: cache write (cached tokens) + regular input (uncached)
        (cached_tokens * rates["cache_write"] + uncached_tokens * rates["input"]) / 1_000_000
        # Subsequent calls: cache read (cached tokens) + regular input (uncached)
        + (num_calls - 1) * (cached_tokens * rates["cache_read"] + uncached_tokens * rates["input"]) / 1_000_000
    )
    
    savings = cost_without_cache - cost_with_cache
    savings_pct = (savings / cost_without_cache) * 100 if cost_without_cache > 0 else 0
    
    print(f"Model: {model}")
    print(f"Calls: {num_calls}")
    print(f"Cached tokens per call: {cached_tokens:,}")
    print(f"Uncached tokens per call: {uncached_tokens:,}")
    print(f"Cost without caching: ${cost_without_cache:.4f}")
    print(f"Cost with caching:    ${cost_with_cache:.4f}")
    print(f"Savings:              ${savings:.4f} ({savings_pct:.1f}%)")
    
    return savings

# Example: 20 calls with 10,000 cached tokens and 500 uncached tokens
calculate_cost_savings(
    num_calls=20,
    cached_tokens=10_000,
    uncached_tokens=500,
    model="claude-3-5-sonnet"
)

Running this example shows that with 20 calls and 10,000 cached tokens, you can save over 80% on input token costs. The savings increase as the number of calls grows and as the ratio of cached-to-uncached tokens increases.

Handling Cache Misses Gracefully

Cache misses happen when the TTL expires or when the prefix changes. Your code should handle both scenarios transparently. Here's a robust wrapper pattern:

import time
from functools import wraps

class CachingLLMWrapper:
    """Wrapper that tracks cache performance and handles misses."""
    
    def __init__(self, llm):
        self.llm = llm
        self.cache_hits = 0
        self.cache_misses = 0
        self.last_cache_write_time = None
    
    def invoke(self, messages, **kwargs):
        response = self.llm.invoke(messages, **kwargs)
        usage = response.usage_metadata or {}
        
        cache_read = usage.get("cache_read_input_tokens", 0)
        cache_write = usage.get("cache_creation_input_tokens", 0)
        
        if cache_read > 0:
            self.cache_hits += 1
        elif cache_write > 0:
            self.cache_misses += 1
            self.last_cache_write_time = time.time()
        
        return response
    
    def get_stats(self):
        total = self.cache_hits + self.cache_misses
        hit_rate = (self.cache_hits / total * 100) if total > 0 else 0
        return {
            "cache_hits": self.cache_hits,
            "cache_misses": self.cache_misses,
            "hit_rate": f"{hit_rate:.1f}%",
        }
    
    def reset_stats(self):
        self.cache_hits = 0
        self.cache_misses = 0

# Usage in a LangGraph workflow
caching_llm = CachingLLMWrapper(
    ChatAnthropic(model="claude-3-5-sonnet-20241022", max_tokens=1024)
)

# Use caching_llm.invoke() in your graph nodes
# After running, check stats:
# print(caching_llm.get_stats())

Common Pitfalls to Avoid

• Changing message order: If you reorder messages between calls, the prefix changes and the cache is invalidated. Always construct messages in the same order.
• Including dynamic metadata in cached blocks: Things like request IDs, timestamps, or user-specific data in the system prompt will break caching. Move these to uncached portions.
• Over-caching small prompts: If your cached prefix is below the minimum threshold, caching provides no benefit and may even cost more due to the higher cache write price.
• Ignoring cache write costs: Cache writes cost more than regular input tokens (typically 25% more). You only save money if the cache is read at least once. Don't cache content that changes every call.
• Not accounting for TTL in long workflows: If a LangGraph workflow takes more than 5 minutes between steps, the cache may expire. Use the 1-hour TTL or restructure your workflow to keep calls within the TTL window.

Conclusion

Prompt caching is one of the most impactful cost optimization techniques available for LLM applications, and LangGraph's stateful, iterative workflows are an ideal use case. By strategically placing cache breakpoints on large, stable content like system prompts, document contexts, and tool definitions, you can reduce input token costs by up to 90% while also improving response latency. The key is to identify which parts of your prompts remain constant across graph iterations, place them at the beginning of your message lists, and use the cache_control parameter to mark cache boundaries. Combined with the monitoring and best practices outlined in this guide, you can build LangGraph applications that are both powerful and cost-efficient at scale.