Agent Cache — AI Agent-Aware Caching

AgentCache extends async-cache into an AI-agent-aware caching layer with tool execution caching, resource-based invalidation, session scoping, and loop detection.

from agent_cache import AgentCache, AgentCacheInvalidator, AgentCacheSession

Read Tool Caching

Cache results of read-only agent tools with resource tagging and TTL:

@AgentCache(resource="cart", scope="global", ttl=60)
async def get_cart(user_id):
    return await db.fetch_cart(user_id)

cart = await get_cart("user_1")   # fetches from DB
cart = await get_cart("user_1")   # cache hit

Write/Mutation Invalidation

Automatically invalidate related cached reads when mutations occur. Invalidation fires before the mutation executes, so stale data is never served even if the mutation raises:

@AgentCacheInvalidator(resource="cart", scope="global")
async def add_to_cart(user_id, item):
    await db.add_item(user_id, item)

await get_cart("user_1")                  # cached
await add_to_cart("user_1", "laptop")     # invalidates "cart", then runs mutation
await get_cart("user_1")                  # re-fetched

Session Scoping

Isolate cache state per agent session:

@AgentCache(resource="ticket", scope="session")
async def get_ticket(ticket_id):
    return await api.get_ticket(ticket_id)

async with AgentCacheSession(session_id="agent-A") as session:
    await get_ticket("T-100")   # fetched and cached
    await get_ticket("T-100")   # hit
# Session ended — cache cleared

Loop Detection

Detect and halt infinite agent tool loops:

async with AgentCacheSession(
    loop_detection=True,
    max_tool_repeats=5,
    max_execution_depth=50,
    on_loop="raise",       # "raise", "warn", or "short_circuit"
) as session:
    # Detects oscillations (A->B->A->B), retry storms, depth overflow

Loop detection modes:

Mode	Behavior
`raise`	Raises `AgentLoopDetectedError` (default)
`warn`	Logs a warning, continues execution
`short_circuit`	Returns cached value or `None`, skips execution

Disk Persistence

Agent caches support disk persistence via the backend parameter:

from cache import DiskBackend

@AgentCache(
    resource="embeddings",
    scope="global",
    backend=DiskBackend("/tmp/embeddings.pkl"),
)
async def get_embedding(text):
    return await ml_model.embed(text)

Observability

from agent_cache import get_metrics

async with AgentCacheSession() as session:
    # ... agent workflow ...
    print(session.get_metrics())
    # {'hits': 12, 'misses': 5, 'invalidations': 2, 'loop_detections': 0, 'hit_rate': 0.706}

print(get_metrics())  # global aggregate