← Back to DevBytes

Prompt Caching for Cost Reduction with AutoGen: Complete Guide

Introduction to Prompt Caching with AutoGen

Prompt caching is a powerful optimization technique that can dramatically reduce the cost and latency of LLM-based applications. When working with multi-agent frameworks like AutoGen, where agents often send large system prompts, conversation histories, and tool definitions back and forth, caching becomes an essential strategy for production-grade systems. In this guide, we'll explore what prompt caching is, why it matters in the context of AutoGen, and how to implement it effectively.

What Is Prompt Caching?

Prompt caching is a mechanism offered by several LLM providers (notably Anthropic's Claude and OpenAI) that stores the processed token representations of frequently used prompt prefixes. When a subsequent request shares the same prefix, the provider reuses the cached computation instead of reprocessing those tokens from scratch. The result is faster response times and significantly reduced costs—often a 50-90% reduction on cached tokens.

The key insight is that many LLM calls in agentic workflows share a common prefix: system instructions, tool schemas, few-shot examples, and earlier conversation turns. By structuring prompts so that the stable portion appears first, you maximize cache hits.

How Caching Works Under the Hood

When the model provider receives a request, it checks whether the leading tokens match a previously cached sequence. If they do, the provider skips the expensive prefill computation for those tokens and charges a reduced rate. Cache entries have a time-to-live (TTL)—typically 5 minutes for ephemeral caches, extendable with longer-lived options on some providers. Each cache lookup is based on an exact token match, so even a single character change in the prefix invalidates the cache from that point forward.

Why Prompt Caching Matters for AutoGen

AutoGen is a multi-agent conversation framework where agents exchange messages, often over many rounds. Each round typically resends the entire conversation history plus the agent's system prompt and tool definitions. This creates several cost amplifiers:

Without caching, a 20-round conversation between two agents can easily consume 10x the tokens of a single call, because each round reprocesses all prior rounds. With caching, only the genuinely new tokens incur full pricing.

Prerequisites and Setup

Before diving into code, make sure you have the necessary dependencies installed. This guide uses AutoGen's Python package along with the Anthropic SDK, which has first-class support for prompt caching.

pip install "autogen-agentchat" "autogen-ext[anthropic]" anthropic python-dotenv

You'll also need API keys. Store them in a .env file:

ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx

Understanding the Cost Model

Anthropic's pricing for Claude models with prompt caching typically looks like this:

This means the first time you send a prompt prefix, you pay a small premium to write it to cache. On every subsequent call that reuses it, you pay only a tenth of the normal price. The break-even point is usually reached after just two calls sharing the same prefix.

Implementing Prompt Caching with AutoGen and Anthropic

AutoGen's Anthropic integration exposes cache control parameters through the model client configuration. Let's start with a basic example that caches a large system prompt.

Basic Example: Caching a System Prompt

import asyncio
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.anthropic import AnthropicChatCompletionClient
from dotenv import load_dotenv

load_dotenv()

# A large, stable system prompt that benefits from caching
SYSTEM_PROMPT = """You are an expert software architect with deep knowledge of:
- Distributed systems design
- Database optimization
- API design and versioning
- Security best practices
- Cloud infrastructure (AWS, GCP, Azure)

When answering questions:
1. Always consider scalability implications
2. Provide concrete code examples when relevant
3. Discuss trade-offs explicitly
4. Cite relevant patterns (e.g., from the Gang of Four, SRE, DDIA)

[... many more lines of detailed instructions ...]
"""

async def main():
    model_client = AnthropicChatCompletionClient(
        model="claude-3-5-sonnet-20241022",
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        # Enable caching for system messages
        cache_control={"type": "ephemeral"},
    )

    architect = AssistantAgent(
        name="Architect",
        model_client=model_client,
        system_message=SYSTEM_PROMPT,
    )

    # First call: cache write (slightly more expensive)
    result = await architect.run(task="How should I design a rate limiter?")
    print(result.messages[-1].content)

    # Second call: cache read (90% cheaper on the system prompt portion)
    result = await architect.run(task="What about a distributed rate limiter?")
    print(result.messages[-1].content)

asyncio.run(main())

In this example, the cache_control parameter tells the Anthropic client to mark the system prompt as cacheable. The first run call writes the prompt to cache; the second call reads it back at the discounted rate.

Advanced Example: Multi-Agent Workflow with Caching

The real power of prompt caching emerges in multi-agent workflows. Here's a two-agent setup where a coder and a reviewer iterate on a solution, sharing a large common context.

import asyncio
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination, TextMentionTermination
from autogen_ext.models.anthropic import AnthropicChatCompletionClient
from dotenv import load_dotenv

load_dotenv()

SHARED_CONTEXT = """
Project Context:
- We are building a Python CLI tool for log analysis.
- The tool must parse JSON logs, filter by severity, and export to CSV.
- Target users are DevOps engineers comfortable with the terminal.
- Performance target: handle 1GB log files in under 30 seconds.

Codebase Conventions:
- Use type hints everywhere.
- Follow PEP 8 strictly.
- Prefer argparse over click for dependencies.
- All public functions need docstrings.
- Tests use pytest and live in tests/ directory.
"""

CODER_PROMPT = SHARED_CONTEXT + """
You are the Coder agent. Write clean, tested Python code based on the
project context above. Always include type hints and docstrings.
When you are done, say 'CODE COMPLETE'.
"""

REVIEWER_PROMPT = SHARED_CONTEXT + """
You are the Code Reviewer agent. Review code for correctness, style,
and adherence to the conventions above. Be specific about issues.
If the code is acceptable, say 'APPROVED'. Otherwise, list issues.
"""

async def main():
    model_client = AnthropicChatCompletionClient(
        model="claude-3-5-sonnet-20241022",
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        cache_control={"type": "ephemeral"},
    )

    coder = AssistantAgent(
        name="Coder",
        model_client=model_client,
        system_message=CODER_PROMPT,
    )

    reviewer = AssistantAgent(
        name="Reviewer",
        model_client=model_client,
        system_message=REVIEWER_PROMPT,
    )

    team = RoundRobinGroupChat(
        participants=[coder, reviewer],
        termination_condition=TextMentionTermination("APPROVED") | MaxMessageTermination(10),
    )

    result = await team.run(task="Implement the log parser module.")
    for msg in result.messages:
        print(f"[{msg.source}]: {msg.content[:200]}...")

asyncio.run(main())

Because both agents share the SHARED_CONTEXT prefix, and because AutoGen resends the full conversation history on each turn, the cache hits compound rapidly. By the third or fourth message, the vast majority of input tokens are cache reads rather than full-price inputs.

Cache Control Strategies

Anthropic supports up to four cache breakpoints per request. This lets you cache different segments independently—for example, the system prompt, the tool definitions, and the conversation history can each be cached separately. Here's how to configure multiple breakpoints.

Manual Cache Breakpoints

import anthropic

client = anthropic.Anthropic()

# Define a large tool schema that rarely changes
TOOLS = [
    {
        "name": "execute_code",
        "description": "Execute Python code and return stdout/stderr.",
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {"type": "string", "description": "Python code to execute"},
                "timeout": {"type": "integer", "default": 30},
            },
            "required": ["code"],
        },
    },
    # ... many more tool definitions ...
]

SYSTEM = "You are a coding assistant. " * 200  # Simulate a large prompt

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": SYSTEM,
            "cache_control": {"type": "ephemeral"},  # Breakpoint 1
        }
    ],
    tools=[
        {**tool, "cache_control": {"type": "ephemeral"}}  # Breakpoint 2
        for tool in TOOLS[-1:]  # Cache the last tool as a breakpoint
    ],
    messages=[
        {"role": "user", "content": "Write a function to reverse a linked list."},
    ],
)

# Inspect cache usage
print(f"Input tokens: {response.usage.input_tokens}")
print(f"Cache creation tokens: {response.usage.cache_creation_input_tokens}")
print(f"Cache read tokens: {response.usage.cache_read_input_tokens}")

The usage object in the response tells you exactly how many tokens were cache writes versus cache reads. Monitoring these numbers is critical for validating that your caching strategy is working.

Integrating Cache Monitoring into AutoGen

To measure the real savings, you can wrap AutoGen's model client to log cache usage on every call. This is invaluable for debugging and for reporting cost savings to stakeholders.

import asyncio
import os
from autogen_agentchat.agents import AssistantAgent
from autogen_core.models import CreateResult
from autogen_ext.models.anthropic import AnthropicChatCompletionClient
from dotenv import load_dotenv

load_dotenv()

class CachingModelClient(AnthropicChatCompletionClient):
    """Wraps Anthropic client to track cache statistics."""

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.total_input = 0
        self.total_cache_write = 0
        self.total_cache_read = 0

    async def create(self, messages, **kwargs):
        result = await super().create(messages, **kwargs)
        usage = getattr(result, "usage", None)
        if usage:
            self.total_input += getattr(usage, "input_tokens", 0) or 0
            self.total_cache_write += getattr(usage, "cache_creation_input_tokens", 0) or 0
            self.total_cache_read += getattr(usage, "cache_read_input_tokens", 0) or 0
        return result

    def report(self):
        total = self.total_input + self.total_cache_write + self.total_cache_read
        if total == 0:
            print("No usage recorded yet.")
            return
        print("=== Cache Usage Report ===")
        print(f"Full-price input tokens: {self.total_input:,}")
        print(f"Cache write tokens:      {self.total_cache_write:,}")
        print(f"Cache read tokens:       {self.total_cache_read:,}")
        print(f"Total tokens processed:  {total:,}")
        if self.total_cache_read > 0:
            savings_pct = (self.total_cache_read / total) * 90  # 90% discount on reads
            print(f"Estimated savings:       ~{savings_pct:.1f}% of input cost")

async def main():
    client = CachingModelClient(
        model="claude-3-5-sonnet-20241022",
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        cache_control={"type": "ephemeral"},
    )

    agent = AssistantAgent(
        name="Helper",
        model_client=client,
        system_message="You are a helpful assistant. " * 300,
    )

    questions = [
        "What is the capital of France?",
        "What is the capital of Germany?",
        "What is the capital of Japan?",
        "What is the capital of Brazil?",
    ]

    for q in questions:
        result = await agent.run(task=q)
        print(f"Q: {q}")
        print(f"A: {result.messages[-1].content[:100]}\n")

    client.report()

asyncio.run(main())

Running this script, you should see the cache read tokens grow with each subsequent call while full-price input tokens stay low, confirming that the system prompt is being reused from cache.

Best Practices for Prompt Caching in AutoGen

1. Structure Prompts with Stable Content First

Cache matches are prefix-based. Place all stable content—system instructions, tool definitions, few-shot examples, and project context—at the beginning of the prompt. Put variable content like the user's latest question at the end. Any dynamic content inserted before the stable block will break the cache.

2. Keep the Cache Warm

Ephemeral caches expire after about 5 minutes of inactivity. If your AutoGen workflow has long pauses between agent calls (for example, waiting for human input), consider sending a lightweight "keepalive" request or restructuring your workflow to batch calls within the TTL window. For longer-lived caches, investigate Anthropic's 1-hour cache option.

3. Minimize Prompt Variability

Even small changes—like injecting a timestamp or a random ID into the system prompt—will invalidate the cache. If you need dynamic metadata, append it to the user message rather than the system prompt. The following anti-pattern breaks caching:

# BAD: Timestamp in system prompt invalidates cache every call
import datetime
system_prompt = f"You are a helpful assistant. Current time: {datetime.datetime.now()}"
# GOOD: Keep system prompt static, pass dynamic info in the user message
system_prompt = "You are a helpful assistant."
user_message = f"[Context: current time is {datetime.datetime.now()}]\n\nWhat's the weather?"

4. Use Cache Breakpoints Strategically

With a maximum of four breakpoints, prioritize caching the largest and most stable segments. A typical ordering is: system prompt, tool definitions, few-shot examples, conversation history. Each breakpoint adds a small overhead, so don't waste them on short, frequently-changing segments.

5. Monitor and Validate Cache Hit Rates

Always log the cache_read_input_tokens and cache_creation_input_tokens from the usage object. If you see high cache creation but low cache reads, your prompts are changing between calls and you're paying the write premium without reaping the read discount. Adjust your prompt structure accordingly.

6. Consider Caching for Long Conversations

In multi-turn AutoGen conversations, the conversation history itself becomes a large stable prefix. By placing a cache breakpoint at the end of the history, each new turn only pays full price for the latest message while reusing the cached history. This is where the savings are most dramatic for agentic loops.

7. Be Aware of Provider Differences

While this guide focuses on Anthropic, OpenAI also offers automatic prompt caching for prompts over 1024 tokens. OpenAI's caching is automatic and requires no code changes, but it only caches prefixes of 1024+ tokens. Anthropic gives you explicit control with breakpoints and works with shorter prefixes (minimum 1024 tokens for Sonnet, 2048 for Haiku). Choose the approach that fits your provider and use case.

Common Pitfalls and How to Avoid Them

Conclusion

Prompt caching is one of the highest-impact, lowest-effort optimizations available for AutoGen applications. By structuring your prompts to place stable content first, configuring cache breakpoints strategically, and monitoring your cache hit rates, you can reduce input token costs by up to 90% while simultaneously improving response latency. As multi-agent workflows scale to dozens of turns and multiple participants, these savings compound dramatically. Start by caching your system prompts and tool definitions, measure the results with the monitoring wrapper shown above, and iteratively refine your caching strategy as your application grows. The combination of AutoGen's flexible agent architecture and Anthropic's granular cache control gives you everything you need to build sophisticated, cost-efficient agentic systems at scale.

— Ad —

Google AdSense will appear here after approval

← Back to all articles