Implement Anthropic prompt caching with cache_control markers

Loading diagram...

• Line 1: Declares a Mermaid flowchart with left-to-right (LR) directional layout.
• Lines 2-7: Defines a subgraph labeled "Context Window (Stability Order)" containing five nodes (A through E) connected by arrows, representing the sequential components of an LLM context window — from System Prompt (~2K tokens, most stable) → Tool Definitions (~5K tokens, stable) → Mem0 Memories (~1K tokens, session-stable) → Conversation History (~30K tokens, growing) → Current User Message (~200 tokens, volatile).
• Line 9: Draws a dotted arrow labeled "cache_control ①" from node A to node B, indicating the first cache breakpoint is placed after the system prompt and tool definitions block.
• Line 10: Draws a dotted arrow labeled "cache_control ②" from node C to node D, indicating the second cache breakpoint is placed between Mem0 memories and the conversation history.
• Lines 12-13: Styles nodes A (System Prompt) and B (Tool Definitions) with a dark green background (#2d5016) and white text, visually grouping them as the most stable, cached portion of the context.
• Line 14: Styles node C (Mem0 Memories) with a dark goldenrod background (#8b6914) and white text, indicating moderate stability between the cached and volatile sections.
• Line 15: Styles node D (Conversation History) with a saddle brown background (#8b4513) and white text, signaling this is a growing, less stable region of the context.
• Line 16: Styles node E (Current User Message) with a dark red background (#8b0000) and white text, marking it as the most volatile, frequently changing part of the context window.

1from dataclasses import dataclass, field
2from typing import Any
3
4@dataclass
5class CacheStats:
6    cache_creation_input_tokens: int = 0
7    cache_read_input_tokens: int = 0
8    input_tokens: int = 0
9
10    @property
11    def hit_rate(self) -> float:
12        total = self.cache_read_input_tokens + self.cache_creation_input_tokens + self.input_tokens
13        return self.cache_read_input_tokens / total if total > 0 else 0.0
14
15    @property
16    def estimated_savings_pct(self) -> float:
17        if self.cache_read_input_tokens == 0:
18            return 0.0
19        read_savings = self.cache_read_input_tokens * 0.9
20        write_penalty = self.cache_creation_input_tokens * 0.25
21        total = self.cache_read_input_tokens + self.cache_creation_input_tokens + self.input_tokens
22        return ((read_savings - write_penalty) / total) * 100 if total > 0 else 0.0
23
24@dataclass
25class ComposedContext:
26    system_prompt: str
27    tool_definitions: list[dict[str, Any]]
28    mem0_memories: list[str]
29    conversation_history: list[dict[str, Any]]
30    current_message: str
31    user_id: str
32    session_id: str
33    token_budget: int = 100_000
34
35class CachedContextManager:
36    CACHE_MARKER = {"type": "ephemeral"}
37
38    def __init__(self, model: str = "claude-sonnet-4-20250514"):
39        self.model = model
40        self.stats = CacheStats()
41
42    def prepare_cached_messages(
43        self, context: ComposedContext
44    ) -> dict[str, Any]:
45        system_blocks = self._build_system_blocks(
46            context.system_prompt, context.tool_definitions
47        )
48        messages = self._build_message_list(
49            context.mem0_memories,
50            context.conversation_history,
51            context.current_message,
52        )
53        return {
54            "model": self.model,
55            "max_tokens": 4096,
56            "system": system_blocks,
57            "messages": messages,
58        }

• Lines 1-2: Imports the dataclass decorator and field factory for defining structured data containers, plus Any for flexible type annotations across the composed context fields.
• Lines 4-8: The CacheStats dataclass tracks three token counters returned by the Anthropic API response's usage object — creation tokens (cache writes), read tokens (cache hits), and standard uncached input tokens.
• Lines 10-12: The hit_rate property computes what fraction of total input tokens were served from cache. A value above 0.7 indicates effective cache placement.
• Lines 14-20: The estimated_savings_pct property calculates net cost savings accounting for the 0.9× discount on cache reads minus the 0.25× surcharge on cache writes. A sustained rate above 60% confirms your prefix ordering is correct.
• Lines 23-30: ComposedContext holds all sections produced by the context engineering pipeline. The mem0_memories field contains strings returned by MemoryClient.search(), while conversation_history holds message dicts loaded from PostgreSQL.
• Lines 31-32: The token_budget field defaults to 100,000 tokens, matching Claude Sonnet's practical context limit. Your Redis session cache uses this budget to decide when to evict older messages.
• Lines 35-36: The class-level CACHE_MARKER constant defines the ephemeral cache control block. Anthropic currently supports only the "ephemeral" type, which provides a five-minute TTL.
• Lines 38-40: The constructor accepts a model identifier and initializes a fresh CacheStats instance that accumulates metrics across multiple calls within a session.
• Lines 42-55: The prepare_cached_messages method orchestrates the full payload construction. It delegates system block assembly and message list construction to private methods, returning a dictionary ready to pass directly to anthropic.AsyncAnthropic().messages.create(**payload).

1import json
2
3class CachedContextManager:
4    # ... (continued from above)
5
6    def _build_system_blocks(
7        self, system_prompt: str, tools: list[dict]
8    ) -> list[dict]:
9        tool_text = json.dumps(tools, indent=2) if tools else ""
10        combined = f"{system_prompt}\n\n## Available Tools\n{tool_text}" if tool_text else system_prompt
11
12        return [
13            {
14                "type": "text",
15                "text": combined,
16                "cache_control": self.CACHE_MARKER,
17            }
18        ]
19
20    def _build_message_list(
21        self, memories: list[str], history: list[dict], current_msg: str
22    ) -> list[dict]:
23        messages = []
24
25        if memories:
26            memory_text = "## What I Remember About You\n" + "\n".join(
27                f"- {m}" for m in memories
28            )
29            messages.append({
30                "role": "user",
31                "content": [
32                    {
33                        "type": "text",
34                        "text": memory_text,
35                        "cache_control": self.CACHE_MARKER,
36                    }
37                ],
38            })
39            messages.append({
40                "role": "assistant",
41                "content": "I'll keep these memories in mind throughout our conversation.",
42            })
43
44        for msg in history:
45            messages.append({
46                "role": msg["role"],
47                "content": msg["content"],
48            })
49
50        messages.append({
51            "role": "user",
52            "content": current_msg,
53        })
54
55        return messages
56
57    def update_stats(self, usage: dict) -> None:
58        self.stats.cache_creation_input_tokens += usage.get(
59            "cache_creation_input_tokens", 0
60        )
61        self.stats.cache_read_input_tokens += usage.get(
62            "cache_read_input_tokens", 0
63        )
64        self.stats.input_tokens += usage.get("input_tokens", 0)

• Lines 1-1: Imports json for serializing tool definitions into the system prompt block. Tool schemas from your context pipeline are Python dicts that must become text for the system content block.
• Lines 6-11: The _build_system_blocks method combines the system prompt and tool definitions into a single text block. Merging them ensures they share one cache entry rather than requiring two separate cache writes.
• Lines 12-19: Returns a list containing one system content block with cache_control applied. This is breakpoint ① from the architecture diagram — the most stable prefix segment.
• Lines 21-24: The _build_message_list method initializes an empty message list that will be built up section by section in stability order.
• Lines 26-29: When Mem0 memories exist, they are formatted as a bulleted list under a clear header. Each memory string comes from MemoryClient.search(query=current_msg, user_id=user_id), which performs semantic retrieval against the user's stored memory graph.
• Lines 30-38: The memories are wrapped in a user message content block with cache_control applied — this is breakpoint ②. The content block format (a list containing a dict with type, text, and cache_control) is required when mixing cached and uncached content within a single message role.
• Lines 39-42: An assistant acknowledgment message follows the memories injection. This priming pattern ensures the model treats the memories as established context rather than a new user query requiring a response.
• Lines 44-48: Conversation history messages are appended without any cache markers. Since history grows on every turn, caching it would cause constant cache misses and waste the 25% write surcharge. The history is loaded from PostgreSQL using your SQLAlchemy async session with cursor-based pagination to avoid loading unbounded message counts.
• Lines 50-53: The current user message is appended last as a plain user message — no cache marker, since it is unique to every request.
• Lines 55-62: The update_stats method extracts cache metrics from the Anthropic API response's usage field. Call this after every API response to maintain running statistics. These stats can be persisted to your Redis session cache alongside per-section token counts for observability dashboards.