memory-reuse

An execution cache layer for AI agents that cuts LLM and tool call costs by avoiding redundant computation. Drop it into any Python agent or LangGraph workflow with a single decorator.

• Framework-agnostic — LangGraph, LiteLLM, or any plain Python function.
• Zero required dependencies — the core runs on the standard library alone.
• Safe by default — per-user / per-session scoping prevents cross-user cache leaks.
• Typed — ships with py.typed, fully type-hinted.

How it works

When your agent calls an LLM or a tool, memory-reuse checks the cache first. On a hit it returns the stored result instantly — no tokens spent, no API call made. On a miss it runs the real call and stores the result for next time:

request ──► cache lookup
              ├── HIT  ──► return stored result   (0 tokens, ~ms)
              └── MISS ──► run LLM / tool ──► store ──► return

The lookup always tries the fastest, cheapest path first. An exact hash match is attempted before anything else; only on an exact miss — and only if the semantic cache is enabled — is the query embedded and matched by similarity. So identical repeats never pay for an embedding.

The three cache types

memory-reuse supports three kinds of lookup. Use one, or combine them.

Exact:     "What is order 123 status?"  ==  "What is order 123 status?"
           same string  →  instant hash match

Tool:      fetch_order(order_id="123")  called 5 min ago
           same function + args  →  cached result returned (until TTL expires)

Semantic:  "What is order 123 status?"  ≈  "Where is my order 123?"
           different words, same intent  →  embedding-similarity match

Exact/tool are on by default; semantic is opt-in. Enabling the semantic cache is what lets reworded-but-equivalent requests reuse a stored answer. See the Architecture guide for the full component and data-flow diagrams.

Node-level and graph-level caching are here. Beyond the three cache types above, memory-reuse can skip whole LangGraph nodes or replay an entire agent run from cache. See Graph-level execution cache.

Install

Works with both pip and uv — pick whichever you use.

pip

pip install memory-reuse

uv

uv add memory-reuse

Optional extras

Note: uv is a fast Python package manager. If you don't have it yet: pip install uv or see docs.astral.sh/uv

Quick start

from memory_reuse import MemoryCache, CacheConfig
from memory_reuse.integrations import cached_tool

cache = MemoryCache()                          # in-memory backend, 1-hour TTL

@cached_tool(cache, scope="global", ttl=300)   # cache for 5 minutes
async def search_web(query: str) -> list[str]:
    return await my_search_api(query)          # only called on cache miss

Usage patterns

1 — Basic exact cache (LLM responses)

from memory_reuse import MemoryCache

cache = MemoryCache()

# Manual get/set
result = await cache.exact.get(["gpt-4", prompt], scope="global", scope_id=None)
if result is None:
    result = await llm.ainvoke(prompt)
    await cache.exact.set(["gpt-4", prompt], result, scope="global",
                          scope_id=None, ttl=3600)

2 — LangGraph node caching

from memory_reuse.integrations import cached_node

@cached_node(cache, scope="user", key_fields=["messages"])
async def summarise(state: dict) -> dict:
    summary = await llm.ainvoke(state["messages"])
    return {"summary": summary}

The decorator reads user_id from the state dict automatically, or from cache.set_context(user_id=...).

3 — LangGraph tool caching

from memory_reuse.integrations import cached_tool

@cached_tool(cache, scope="session", ttl=120)
async def fetch_user_profile(user_id: str) -> dict:
    return await db.get_user(user_id)

4 — LiteLLM (works with OpenAI, Claude, Bedrock, Groq, Ollama, and 100+ more)

from memory_reuse.integrations import cached_litellm_completion, cached_litellm_embedding

# Completion — same prompt + model = cache hit, 0 tokens used
response = await cached_litellm_completion(
    cache,
    model="gpt-4o-mini",                  # swap for any LiteLLM model string
    messages=[{"role": "user", "content": "What is the capital of France?"}],
    ttl=3600,
    scope="global",
)

# Embeddings — deterministic, safe to cache for 24 hours
embeddings = await cached_litellm_embedding(
    cache,
    model="text-embedding-3-small",
    input=["What is machine learning?"],
)

5 — Strands / CrewAI tool caching

Strands and CrewAI each get a cached_tool decorator that mirrors the LangGraph one — same signature cached_tool(cache, *, scope="global", ttl=300, semantic=False, exact_only=False), same keying, scope resolution, TTL expiry, and exact-vs-semantic routing. They delegate to the shared caching machinery, so behaviour is identical across frameworks.

Strands — requires pip install "memory-reuse[strands]":

from memory_reuse.integrations.strands import cached_tool

@cached_tool(cache, scope="global", ttl=600)
async def fetch_weather(city: str) -> dict:
    return await weather_api.get(city)

CrewAI — requires pip install "memory-reuse[crewai]":

from memory_reuse.integrations.crewai import cached_tool

@cached_tool(cache, scope="session", ttl=300)
async def fetch_user_profile(user_id: str) -> dict:
    return await db.get_user(user_id)

CrewAI adds one decoration-time guard: passing both exact_only=True and semantic=True is contradictory, so it raises ConfigurationError rather than silently ignoring one argument. Using either integration without its dependency installed raises BackendNotAvailableError naming the extra to install.

Backend options

Configure via code or environment variables:

export MEMORY_REUSE_BACKEND=redis
export MEMORY_REUSE_REDIS_URL=redis://localhost:6379/0
export MEMORY_REUSE_DEFAULT_TTL=600
export MEMORY_REUSE_DEFAULT_SCOPE=user

cache = MemoryCache.from_env()

AgentCore backend (AWS)

The AgentCore backend targets the managed AWS AgentCore store, so a value cached in one AgentCore microVM is served to requests handled by another — solving the cross-microVM isolation problem where each VM starts with an empty in-memory cache. Requires pip install "memory-reuse[agentcore]".

Select it in code:

from memory_reuse import MemoryCache, CacheConfig

cache = MemoryCache(CacheConfig(
    backend="agentcore",
    agentcore_region="us-east-1",
    agentcore_memory_id="mem-123",
))

Or via environment variables read by MemoryCache.from_env():

export MEMORY_REUSE_BACKEND=agentcore
export MEMORY_REUSE_AGENTCORE_REGION=us-east-1
export MEMORY_REUSE_AGENTCORE_MEMORY_ID=mem-123

cache = MemoryCache.from_env()

Selecting backend="agentcore" without the dependency installed raises BackendNotAvailableError naming the extra; selecting it via from_env() without MEMORY_REUSE_AGENTCORE_REGION or MEMORY_REUSE_AGENTCORE_MEMORY_ID raises ConfigurationError naming the missing setting.

Multi-scope support

cache.set_context(user_id="alice", session_id="sess-001")

# User-scoped: alice cannot see bob's cache
await cache.exact.get(["key"], scope="user", scope_id="alice")

# Session-scoped: isolated per conversation
await cache.tool.get("search", args, scope="session", scope_id="sess-001")

# Global: shared across all users — safe for public, stateless data
await cache.exact.get(["key"], scope="global", scope_id=None)

Using scope="user" without a user_id raises ScopeViolationError to prevent accidental cross-user data leaks.

Semantic cache

Exact caching only hits when inputs are identical. The semantic cache also serves a cached result when a new query is meaningfully similar to a stored one — so "What is 128 multiplied by 47?" can reuse the answer to "What is 128 times 47?". This lifts hit rates for natural-language workloads (chatbots, FAQ agents, docs Q&A) where the same intent is phrased many ways.

Choosing an embedding provider

The semantic cache turns text into vectors using one of three interchangeable providers, selected by embedding_provider:

There are just two install commands to remember. The semantic extra covers both API providers (it bundles the small openai and litellm clients, plus numpy as a cosine-similarity speedup) and installs no torch:

pip install "memory-reuse[semantic]"     # OpenAI + LiteLLM embeddings, lightweight

Local embeddings need sentence-transformers, which depends on PyTorch. On a CPU-only machine (no NVIDIA GPU) install the CPU torch wheel first to avoid a ~2 GB GPU/CUDA download — the CPU build is ~200 MB:

pip install torch --index-url https://download.pytorch.org/whl/cpu
pip install "memory-reuse[semantic-local]"

With a GPU you can skip the first line and just pip install "memory-reuse[semantic-local]". Either way, the model weights themselves (e.g. all-MiniLM-L6-v2, ~90 MB) download from Hugging Face on first use and are then cached on disk for offline reuse.

Quieter / fully offline runs. memory-reuse already silences the Hugging Face log chatter and the per-embedding progress bar. Once the model is cached, you can additionally skip Hugging Face's cache-validation HTTP checks by exporting HF_HUB_OFFLINE=1 (and TRANSFORMERS_OFFLINE=1) in your own process. Do this only in your application — set after the first (downloading) run, never inside a shared library.

Enabling it

Semantic caching is off by default — existing exact-only code is unchanged. Turn it on via CacheConfig by setting semantic_enabled=True and choosing an embedding provider (local, openai, or litellm):

from memory_reuse import MemoryCache, CacheConfig

cache = MemoryCache(CacheConfig(
    backend="memory",
    semantic_enabled=True,
    embedding_provider="local",              # local sentence-transformers model
    embedding_model="all-MiniLM-L6-v2",
    similarity_threshold=0.95,               # how close is "close enough"
))

# Exact-first, then semantic. An exact hit never computes an embedding.
result = await cache.lookup(
    ["qa", "What is 128 multiplied by 47?"],
    query_text="What is 128 multiplied by 47?",
    scope="global", scope_id=None,
)
if result is None:
    result = await run_llm(...)              # only on a miss
    await cache.store(
        ["qa", "What is 128 times 47?"],
        query_text="What is 128 times 47?",
        value=result, scope="global", scope_id=None,
    )

Use cache.lookup(...) / cache.store(...) for the combined exact-then-semantic flow. The exact cache is always tried first, so a semantic embedding is only computed on an exact miss (no extra cost when an exact hit is available).

Three ways to set the threshold

The similarity_threshold is a float in [0.0, 1.0]; a higher value demands a closer match. You can set it three ways, from lowest to highest precedence:

1. Config field — the instance-wide default:

   CacheConfig(semantic_enabled=True, embedding_provider="local",
               similarity_threshold=0.92)
   

2. Environment variable — read by MemoryCache.from_env():

   export MEMORY_REUSE_SEMANTIC_ENABLED=true
   export MEMORY_REUSE_EMBEDDING_PROVIDER=local
   export MEMORY_REUSE_SIMILARITY_THRESHOLD=0.90
   

   cache = MemoryCache.from_env()
   

3. Per-call override — passed to a single lookup, taking precedence over the config/env value for that call only:

   await cache.lookup(key_parts, query_text="...", scope="global",
                      scope_id=None, threshold=0.98)
   

Returning just the relevant answer (extract_answer)

A semantic hit returns the whole stored answer by default. If you asked "Tell me about Python" and later ask "Who created Python?", the second query matches the first and returns the entire paragraph — even though only one sentence answers it.

Set extract_answer=True to have the cache return only the sentence(s) that best match the new question:

cache = MemoryCache(CacheConfig(
    semantic_enabled=True,
    embedding_provider="local",
    extract_answer=True,          # narrow the stored answer to the best sentence
    extract_min_similarity=0.5,   # confidence a sentence needs to be picked
))

Now "Who created Python?" returns just "Python is a high-level, interpreted programming language created by Guido van Rossum and first released in 1991." instead of the full paragraph.

How it works and its limits:

• Purely extractive, no LLM. It splits the stored answer into sentences, embeds each with the same model, and returns the sentence closest to the query. It never calls an LLM and never fabricates — it can only return text already present in the stored answer.
• Falls back to the full answer when no sentence clears extract_min_similarity, so you never get an empty result.
• String answers only. Non-string values (dicts, numbers) and single-sentence answers are returned unchanged.
• Best-effort, not QA. It returns a whole real sentence, so it can't reshape text into a crisp answer the way a model would. It is off by default.

Choosing where to use semantic matching

Semantic matching compares meaning, so it shines for natural-language questions where the same intent is phrased many ways — chatbots, FAQ agents, docs Q&A, search. That is exactly where it saves the most.

It's a similarity match, though, so keep it to reads and questions rather than correctness-critical commands. Two prompts can look close yet mean opposite things — "cancel order 123" vs "confirm order 123" — so use the exact cache for anything whose result depends on precise wording, especially actions with side effects. memory-reuse gives you two simple levers to stay on the safe side:

• Tune the threshold. The default (0.95) favours precision — matches only fire when queries are very close. Raise it if you ever see a wrong match; lower it to trade some precision for a higher hit rate.

• Opt a call out with exact_only=True. For a sensitive call site, skip the semantic cache entirely regardless of the global config:

  await cache.lookup(key_parts, query_text="...", scope="global",
                     scope_id=None, exact_only=True)   # exact match only
  

Used this way — similarity for questions, exact for commands — semantic caching is both safe and a big hit-rate win.

Latency and cost tradeoff

Enabling semantic caching adds an embedding computation on every exact miss. That embedding costs time (local model inference or an API round-trip) and, for hosted providers, money. The win is fewer full LLM calls when reworded queries match; the cost is the embedding overhead on misses. Enable it when your workload has many differently-worded but equivalent requests, so the saved LLM calls outweigh the embedding cost. An exact hit short-circuits before any embedding, so identical repeats stay as cheap as Phase 1.

Graph-level execution cache (wrap_graph)

Wrap a compiled LangGraph graph so an entire run can be served from cache. On a hit the stored final result is replayed with zero nodes executed; on a miss the real graph runs and its final state is stored.

Requires the langgraph extra: pip install "memory-reuse[langgraph]".

from memory_reuse import MemoryCache

cache = MemoryCache()
graph = build_graph().compile()          # your compiled LangGraph graph

cached_graph = cache.wrap_graph(
    graph,
    scope="user",                         # global | user | session
    key_fields=["question"],              # ignore ephemeral state fields
    ttl=3600,
)

# Same signatures as the wrapped graph, plus per-call cache controls.
result = await cached_graph.ainvoke({"question": "How do I reset my password?",
                                     "user_id": "alice"})
result = cached_graph.invoke({"question": "...", "user_id": "alice"})  # sync too

Semantic matching. Enable semantic=True (with semantic_enabled=True on the config) so reworded but equivalent questions reuse a stored run. A per-wrapper similarity_threshold overrides the config default.

cached_graph = cache.wrap_graph(graph, semantic=True, similarity_threshold=0.92)

Per-call controls.

• bypass_cache=True — always run the graph, skip the lookup.
• no_store=True — run the graph but do not store the result.

Node-level invalidation. Invalidate a single cached node output when its upstream state is known to have changed. Safe and idempotent when no entry exists.

await cache.invalidate_node(summarise, {"messages": [...]}, scope="user",
                            scope_id="alice", key_fields=["messages"])

Side effects. Graph-level caching replays a full stored result. It is unsuitable for runs whose side effects must occur on every invocation (writes, emails, payments). Use bypass_cache / no_store for those, or leave the graph unwrapped.

See the graph-level cache guide for the full walkthrough.

Configuration reference

All options live on CacheConfig. Every field can also be set from an environment variable (read by MemoryCache.from_env()) where noted.

Invalid values raise at construction time: an out-of-range similarity_threshold or extract_min_similarity raises ConfigurationError; a non-positive default_ttl raises InvalidTTLError; enabling semantic_enabled without an embedding_provider raises ConfigurationError.

Cache statistics

stats = cache.stats
print(f"Hit rate: {stats.hit_rate:.1%}")
print(f"Hits: {stats.hits}  Misses: {stats.misses}")
print(f"Exact hits: {stats.exact_hits}  Semantic hits: {stats.semantic_hits}")
print(stats.to_dict())

hits always equals exact_hits + semantic_hits, so you can see how many of your hits came from the faster exact path versus semantic matching.

Cost analytics

Beyond hit rate, memory-reuse can quantify the work a cache eliminated — tokens saved, cost saved, and latency saved — so you can answer "how much did the cache actually save me?". It layers on the same counters as stats; nothing changes when you don't use it.

Attribution is caller-supplied: the cache can't know a call's token or latency cost, so you record it on a hit (the LiteLLM completion wrapper does this automatically from the response's usage block). Supply a PricingConfig to turn saved tokens into a saved amount:

from memory_reuse import CacheConfig, CacheHitEvent, MemoryCache, PricingConfig

cache = MemoryCache(CacheConfig(
    pricing=PricingConfig(
        input_token_price=0.0000005,     # $0.50 / 1M input tokens
        output_token_price=0.0000015,    # $1.50 / 1M output tokens
        currency="USD",
    ),
))

# On a cache hit, record what the hit avoided:
cache.record_hit_event(CacheHitEvent(
    tokens_in=1200, tokens_out=400, latency_saved=0.4,
    operation="search_confluence",
))

snap = cache.analytics
print(snap.tokens_saved, snap.cost_saved, snap.currency, snap.latency_saved)

Cost is accumulated at full precision and only rounded to the currency's minor unit when read, so many small per-hit savings still sum correctly. With no PricingConfig, cost_saved stays 0 while tokens and latency still track. When enable_stats=False, analytics is fully zeroed.

CLI

Dump a snapshot from your app and inspect it with the memory-reuse command (standard library only — no extra required):

from memory_reuse.cli import dump_snapshot
dump_snapshot(cache, "snapshot.json")

memory-reuse stats   --snapshot snapshot.json
memory-reuse savings --snapshot snapshot.json

Prometheus / OpenTelemetry export

Publish the analytics to your monitoring stack. Both exporters are opt-in and read the snapshot at scrape/collection time.

# pip install "memory-reuse[prometheus]"
from memory_reuse.analytics.exporters.prometheus import PrometheusExporter
PrometheusExporter(lambda: cache.analytics, lambda: cache.stats)

# pip install "memory-reuse[opentelemetry]"
from memory_reuse.analytics.exporters.opentelemetry import OpenTelemetryExporter
OpenTelemetryExporter(lambda: cache.analytics)

Examples

Runnable examples live in examples/:

• basic_exact_cache.py — the cache primitives with no framework.
• langgraph_agent_example.py — cached nodes and tools in a LangGraph-style flow.
• langgraph_math_agent.py — a real ReAct agent with a calculator and a web-search tool, calling an LLM via LiteLLM.
• semantic_cache_demo.py — a reworded query hitting the semantic cache via the combined lookup/store flow (offline, no model download).
• semantic_agent.py — a real ReAct agent (calculator + web search) whose LLM calls run through the semantic cache with a local embedding model, so reworded questions reuse cached answers.
• framework_tool_caching.py — the Strands and CrewAI cached_tool decorators: store-and-replay round trip plus the CrewAI exact_only+semantic guard (offline, no framework install needed).
• agentcore_backend.py — the AWS AgentCore shared backend: cross-microVM cache sharing, byte round-trip, and TTL/connectivity semantics against an in-process fake service (offline).
• cost_analytics_demo.py — the cost analytics layer: a miss + repeated hits accumulate tokens / cost / latency saved, printed as a snapshot and dumped for the memory-reuse CLI (offline, no API key).
• agentcore_analytics_otel.py — seeing savings on AWS AgentCore Runtime: the OpenTelemetryExporter + record_hit_event handler pattern, plus a runnable offline demo simulating per-microVM metrics aggregating into a fleet total (as CloudWatch would).

export API_KEY="your-groq-key"          # example uses Groq via LiteLLM
python examples/langgraph_math_agent.py

# The analytics demo needs no API key:
python examples/cost_analytics_demo.py

Roadmap

Documentation

Full documentation — guides plus an auto-generated API reference — is published at pranit-p.github.io/memory-reuse.

Build and preview it locally with the docs extra:

pip install -e ".[docs]"
mkdocs serve            # live preview at http://127.0.0.1:8000
mkdocs build --strict   # produce the static site in ./site

Contributing

Contributions are welcome. See CONTRIBUTING.md for setup, tests, and code-style guidelines, and CONTRIBUTORS.md for the list of people who have helped build this project.

License

MIT © Pranit Pawar