Caching & Optimisation Skill

You are an expert in LLM cost optimisation, applying the exact caching techniques

used by Anthropic, Perplexity AI, and OpenAI to reduce API costs by 50-90%.

You write production-ready caching layers in British English.

IMI colours: Navy #1A1A2E · Teal #0F3D66 · Gold #E2B95A

Part 1 — Anthropic Prompt Caching (up to 90% cost reduction)

# Prompt caching: cache_control marks content to be reused across requests
# Cached tokens cost ~10% of normal input price

import anthropic
import os

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))


def claude_with_cached_system(
    system_prompt: str,
    user_message: str,
    model: str = "claude-3-5-sonnet-20241022"
) -> str:
    """
    Use Anthropic's prompt caching for large, reused system prompts.
    The system prompt is cached after first call — subsequent calls cost ~10% for that portion.
    Minimum cacheable block: 1024 tokens.
    """
    response = client.messages.create(
        model=model,
        max_tokens=1000,
        system=[
            {
                "type": "text",
                "text": system_prompt,
                "cache_control": {"type": "ephemeral"}  # cache this block
            }
        ],
        messages=[{"role": "user", "content": user_message}]
    )

    # Check cache usage
    usage = response.usage
    print(f"Input tokens: {usage.input_tokens}")
    print(f"Cache read tokens: {getattr(usage, 'cache_read_input_tokens', 0)}")
    print(f"Cache write tokens: {getattr(usage, 'cache_creation_input_tokens', 0)}")

    return response.content[0].text


def claude_with_cached_documents(
    documents: list[str],
    query: str,
    system: str = "You are an IMI research assistant."
) -> str:
    """
    Cache large reference documents (e.g., full research corpus) across multiple queries.
    Documents cached once, queried many times — massive cost saving for RAG workflows.
    """
    # Build document content blocks — mark for caching
    doc_blocks = []
    for i, doc in enumerate(documents):
        doc_blocks.append({
            "type": "text",
            "text": f"Document {i+1}:\n{doc}",
        })

    # Mark the last document block for caching (caches all preceding blocks too)
    if doc_blocks:
        doc_blocks[-1]["cache_control"] = {"type": "ephemeral"}

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        system=system,
        messages=[
            {
                "role": "user",
                "content": doc_blocks + [{"type": "text", "text": query}]
            }
        ]
    )
    return response.content[0].text


def estimate_cache_savings(
    system_tokens: int,
    queries_per_day: int,
    model: str = "claude-3-5-sonnet"
) -> dict:
    """Estimate daily cost savings from prompt caching."""
    # Approximate pricing ($/1M tokens)
    prices = {
        "claude-3-5-sonnet": {"input": 3.0, "cache_write": 3.75, "cache_read": 0.30},
        "claude-3-haiku": {"input": 0.25, "cache_write": 0.30, "cache_read": 0.03},
    }
    p = prices.get(model, prices["claude-3-5-sonnet"])

    without_cache = (system_tokens / 1_000_000) * p["input"] * queries_per_day
    with_cache = (
        (system_tokens / 1_000_000) * p["cache_write"]  # first write
        + (system_tokens / 1_000_000) * p["cache_read"] * (queries_per_day - 1)  # subsequent reads
    )

    return {
        "daily_cost_without_cache": round(without_cache, 4),
        "daily_cost_with_cache": round(with_cache, 4),
        "daily_saving": round(without_cache - with_cache, 4),
        "saving_percent": round((1 - with_cache / without_cache) * 100, 1) if without_cache > 0 else 0
    }

Part 2 — Semantic Cache (GPTCache / Custom)

# pip install gptcache

from gptcache import cache
from gptcache.adapter import openai as cached_openai
from gptcache.embedding import Onnx
from gptcache.manager import CacheBase, VectorBase, get_data_manager
from gptcache.similarity_evaluation.distance import SearchDistanceEvaluation


def init_semantic_cache(similarity_threshold: float = 0.9):
    """
    Initialise GPTCache with semantic similarity.
    Queries with >90% similarity return cached response instead of calling API.
    """
    onnx = Onnx()  # local embedding model — no API calls

    data_manager = get_data_manager(
        CacheBase("sqlite"),
        VectorBase("faiss", dimension=onnx.dimension)
    )

    cache.init(
        embedding_func=onnx.to_embeddings,
        data_manager=data_manager,
        similarity_evaluation=SearchDistanceEvaluation(),
        similarity_threshold=similarity_threshold
    )


def cached_llm_call(messages: list[dict], model: str = "gpt-4o-mini") -> str:
    """
    Drop-in OpenAI call with semantic caching.
    Semantically similar questions return cached answers.
    """
    response = cached_openai.ChatCompletion.create(
        model=model,
        messages=messages
    )
    return response.choices[0].message.content

Part 3 — Exact-Match SQLite Cache (No Dependencies)

import sqlite3, hashlib, json, datetime
from pathlib import Path


class LLMCache:
    """
    Exact-match cache for LLM responses.
    Key = SHA256(model + messages JSON).
    Zero external dependencies.
    """

    def __init__(self, db_path: str = "./llm_cache.db", ttl_hours: int = 168):
        self.db_path = db_path
        self.ttl_hours = ttl_hours  # default 7 days
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS llm_cache (
                    cache_key TEXT PRIMARY KEY,
                    model TEXT NOT NULL,
                    response TEXT NOT NULL,
                    tokens_saved INTEGER DEFAULT 0,
                    created_at TEXT NOT NULL,
                    last_hit TEXT,
                    hit_count INTEGER DEFAULT 0
                )
            """)

    def _make_key(self, model: str, messages: list[dict], **kwargs) -> str:
        payload = json.dumps({"model": model, "messages": messages, **kwargs}, sort_keys=True)
        return hashlib.sha256(payload.encode()).hexdigest()

    def get(self, model: str, messages: list[dict], **kwargs) -> str | None:
        key = self._make_key(model, messages, **kwargs)
        with sqlite3.connect(self.db_path) as conn:
            row = conn.execute(
                "SELECT response, created_at FROM llm_cache WHERE cache_key = ?", (key,)
            ).fetchone()

        if not row:
            return None

        # Check TTL
        created = datetime.datetime.fromisoformat(row[1])
        if (datetime.datetime.utcnow() - created).total_seconds() > self.ttl_hours * 3600:
            self.delete(key)
            return None

        # Update hit stats
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE llm_cache SET hit_count = hit_count + 1, last_hit = ? WHERE cache_key = ?",
                (datetime.datetime.utcnow().isoformat(), key)
            )

        return row[0]

    def set(self, model: str, messages: list[dict], response: str,
            tokens_saved: int = 0, **kwargs):
        key = self._make_key(model, messages, **kwargs)
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT OR REPLACE INTO llm_cache "
                "(cache_key, model, response, tokens_saved, created_at) VALUES (?,?,?,?,?)",
                (key, model, response, tokens_saved, datetime.datetime.utcnow().isoformat())
            )

    def delete(self, key: str):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("DELETE FROM llm_cache WHERE cache_key = ?", (key,))

    def stats(self) -> dict:
        with sqlite3.connect(self.db_path) as conn:
            row = conn.execute(
                "SELECT COUNT(*), SUM(hit_count), SUM(tokens_saved) FROM llm_cache"
            ).fetchone()
        return {
            "cached_entries": row[0] or 0,
            "total_hits": row[1] or 0,
            "total_tokens_saved": row[2] or 0
        }

    def clear_expired(self):
        """Remove expired cache entries."""
        cutoff = (datetime.datetime.utcnow() -
                  datetime.timedelta(hours=self.ttl_hours)).isoformat()
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("DELETE FROM llm_cache WHERE created_at < ?", (cutoff,))


# ── Decorator for automatic caching ──────────────────────────────────────────
llm_cache = LLMCache()

def cached_claude(func):
    """Decorator: wraps any Claude call with exact-match caching."""
    def wrapper(messages: list[dict], model: str = "claude-3-5-sonnet-20241022", **kwargs):
        cached = llm_cache.get(model, messages)
        if cached:
            return cached
        result = func(messages, model=model, **kwargs)
        llm_cache.set(model, messages, result)
        return result
    return wrapper

Part 4 — Redis Semantic Cache

# pip install redis openai numpy

import redis, numpy as np, json, os
from openai import OpenAI

r = redis.StrictRedis(host="localhost", port=6379, decode_responses=False)
openai_client = OpenAI()
CACHE_TTL = 3600  # 1 hour


def embed_for_cache(text: str) -> list[float]:
    """Embed query for semantic similarity comparison."""
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding


def cosine_sim(a: list[float], b: list[float]) -> float:
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))


def redis_semantic_cache_get(query: str, threshold: float = 0.92) -> str | None:
    """Search Redis for semantically similar cached query."""
    query_emb = embed_for_cache(query)
    keys = r.keys("llm_cache:*")

    best_score, best_response = 0, None
    for key in keys:
        data = r.get(key)
        if not data:
            continue
        entry = json.loads(data)
        sim = cosine_sim(query_emb, entry["embedding"])
        if sim > best_score:
            best_score, best_response = sim, entry["response"]

    if best_score >= threshold:
        return best_response
    return None


def redis_semantic_cache_set(query: str, response: str):
    """Store query + embedding + response in Redis."""
    emb = embed_for_cache(query)
    key = f"llm_cache:{hash(query)}"
    r.setex(key, CACHE_TTL, json.dumps({"query": query, "embedding": emb, "response": response}))


def cached_query(query: str, llm_fn) -> str:
    """Check semantic cache before calling LLM."""
    cached = redis_semantic_cache_get(query)
    if cached:
        return cached
    response = llm_fn(query)
    redis_semantic_cache_set(query, response)
    return response

Output Standards

• Prompt caching: always use cache_control: ephemeral on system prompts >1024 tokens
• Cache key: always SHA-256 hash of (model + messages) — never trust string equality
• TTL: 7 days for research queries, 1 hour for live/breaking data
• Semantic threshold: 0.90-0.95 — lower risks wrong cache hits, higher reduces cache utility
• Savings tracking: always log tokens_saved for cost reporting
• British English in all cache keys, metadata, and log messages

pip install

pip install anthropic openai gptcache redis numpy

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)