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. .. code-block:: python from agent_cache import AgentCache, AgentCacheInvalidator, AgentCacheSession Read Tool Caching ----------------- Cache results of read-only agent tools with resource tagging and TTL: .. code-block:: python @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: .. code-block:: python @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: .. code-block:: python @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: .. code-block:: python 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: .. list-table:: :header-rows: 1 :widths: 20 80 * - 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: .. code-block:: python 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 ------------- .. code-block:: python 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