mirror of
https://github.com/pydantic/pydantic-ai-harness.git
synced 2026-07-21 10:55:35 +00:00
fix(cache_stability): per-key timing + collapse re-baseline; graduate as released
Douwe's second review found two correctness bugs a prior pass missed. The expiry-gap hedge used a single global last-observation time, so after a model switch-away-and-back it measured the gap against the other model's request and withheld the expiry explanation exactly when expiry was the likely cause. And a 0/0 response after an established prefix re-tripped the collapse check on every remaining step, because the mark only ever grew. Track per-key observation state in a non-init _RunState (rebuilt fresh in for_run), time the gap from a monotonic clock seam instead of provider-populated response.timestamp (which can skew across providers), and re-baseline the mark to the collapsed prefix so a sustained collapse warns once. Reject collapse_ratio<=0 (never warns) and cache_ttl_seconds<=0. Note #6337 as the future source for the per-model TTL. Graduate the module properly as a released capability: add the docs page, nav, index, and parity meta rows, and drop the (preview) framing everywhere. It stays un-re-exported from the root, as every new capability is.
This commit is contained in:
David SF
@@ -157,7 +157,7 @@ We studied leading coding agents, agent frameworks, and Claw-style assistants to
|
||||
| | **Context compaction** | LLM-powered summarization of older messages | :white_check_mark: [Docs](pydantic_ai_harness/compaction/) | [summarization-pydantic-ai](https://github.com/vstorm-co/summarization-pydantic-ai) (vstorm‑co) |
|
||||
| | **Limit warnings** | Warn agent before hitting context/iteration limits | :white_check_mark: [Docs](pydantic_ai_harness/compaction/) | [summarization-pydantic-ai](https://github.com/vstorm-co/summarization-pydantic-ai) (vstorm‑co) |
|
||||
| | **Tool output management** | Truncate, summarize, or spill large tool outputs | :white_check_mark: [Docs](pydantic_ai_harness/overflowing_tool_output/) | |
|
||||
| | **Cache-bust monitoring** | Warn when a run's prompt-cache prefix collapses between model requests | :white_check_mark: [Docs](pydantic_ai_harness/cache_stability/) (preview) | |
|
||||
| | **Cache-bust monitoring** | Warn when a run's prompt-cache prefix collapses between model requests | :white_check_mark: [Docs](pydantic_ai_harness/cache_stability/) | |
|
||||
| | **System reminders** | Inject periodic reminders to counteract instruction drift | :construction: [PR #181](https://github.com/pydantic/pydantic-ai-harness/pull/181) | |
|
||||
| **Memory & persistence** | **Memory** | Persistent, namespaced notebook with bounded prompt injection, on-demand search, and concurrency-safe stores | :white_check_mark: [Docs](pydantic_ai_harness/memory/) | [pydantic-deep](https://github.com/vstorm-co/pydantic-deepagents) (vstorm‑co) |
|
||||
| | **Session persistence** | Save and restore full conversation state | :white_check_mark: [Docs](pydantic_ai_harness/step_persistence/) | |
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
---
|
||||
title: Cache Stability Monitor
|
||||
description: Warn when a run's prompt-cache hit collapses between model requests, so a moved prefix or an expired cache surfaces instead of silently re-charging tokens.
|
||||
---
|
||||
|
||||
# Cache Stability Monitor
|
||||
|
||||
Warn when a run's prompt cache hit collapses between model requests, so a moved cacheable prefix or an expired provider cache surfaces instead of quietly re-charging tokens it could have served from cache.
|
||||
|
||||
> [!NOTE]
|
||||
> Import this capability from its submodule. It is not re-exported from `pydantic_ai_harness`:
|
||||
>
|
||||
> ```python
|
||||
> from pydantic_ai_harness.cache_stability import CacheStabilityMonitor
|
||||
> ```
|
||||
|
||||
Cache Stability Monitor is a released, non-experimental capability. Pydantic AI Harness is still on 0.x releases, so the API may change between minor releases. See the [version policy](index.md#version-policy).
|
||||
|
||||
## What it watches
|
||||
|
||||
Prompt caching pays off only while the cacheable prefix (tools, then system instructions, then message history) stays byte-stable across a run's consecutive requests. When something moves that prefix -- reordered tools, a timestamp injected into instructions, a serialization-level block hop -- the provider re-charges tokens it could have served from cache.
|
||||
|
||||
This is the **observe** signal: it reads the provider's own verdict rather than guessing from the structured request. On each response it reads `usage.cache_read_tokens` and tracks the cacheable prefix the run has established (`cache_read_tokens + cache_write_tokens`), keyed by the response's `(provider_name, model_name)`. Because message history is append-only, a stable prefix means each request for that model reads back at least what the previous one cached; a large drop is the observable signature of a collapse.
|
||||
|
||||
When a request reads back less than `collapse_ratio` of the established prefix, the monitor emits a `CacheBustWarning` and re-baselines the tracked prefix to the collapsed value. A sustained collapse -- for example caching toggled off mid-run, which reports `read == 0, write == 0` on every remaining step -- therefore warns once, not on every request.
|
||||
|
||||
```python
|
||||
from pydantic_ai import Agent
|
||||
from pydantic_ai_harness.cache_stability import CacheStabilityMonitor
|
||||
|
||||
agent = Agent('anthropic:claude-sonnet-4-5', capabilities=[CacheStabilityMonitor()])
|
||||
await agent.run('...') # a CacheBustWarning fires if a cached prefix collapses mid-run
|
||||
```
|
||||
|
||||
The verdict is cross-provider for free -- pyai normalizes every provider into the `cache_read_tokens` / `cache_write_tokens` fields on `RequestUsage`.
|
||||
|
||||
## Model switches and expiry
|
||||
|
||||
Keying per provider and model means a mid-run model switch does not warn: a `FallbackModel` failover or a per-step model change uses a different cache key, so the monitor starts a fresh mark for it instead of comparing against the previous model's. Marks are kept per key rather than reset, so switching back to an earlier model within its cache TTL still compares against that model's prefix.
|
||||
|
||||
A collapse has two shapes the monitor cannot tell apart, so the warning names both: the cacheable prefix moved, or the provider's cache expired under an unchanged prefix (a gap between requests longer than the cache TTL -- Anthropic's default is 5 minutes, refreshed on each hit). When the gap since the same model's previous request exceeds `cache_ttl_seconds`, the message reports the gap so a long tool or approval pause isn't mistaken for a moved prefix. The gap is timed per model, so switching away and back measures the returning model's own idle time, not whatever ran in between.
|
||||
|
||||
## Options
|
||||
|
||||
- `collapse_ratio` (default `0.5`): warn when a request reads back less than this fraction of the established prefix. Conservative by default so ordinary rounding or a partial miss does not fire; raise toward `1.0` to warn on smaller regressions. It must be greater than `0.0` -- a ratio of `0.0` could never warn, so it is rejected rather than treated as a silent disable switch.
|
||||
- `min_prefix_tokens` (default `1024`): only judge collapse once the established prefix reaches this many tokens. Below a provider's minimum cacheable size (Anthropic's is 1024) `cache_read_tokens` is noisy or zero.
|
||||
- `cache_ttl_seconds` (default `300`): the assumed provider cache TTL. Message-only -- when the gap since the same model's previous request exceeds it, the warning notes the collapse may be a cache expiry rather than a moved prefix. It does not change whether a warning fires. Lower it for providers with a shorter cache lifetime.
|
||||
|
||||
## Silencing and escalation
|
||||
|
||||
There is no bespoke suppression API. Use the stdlib `warnings` machinery, exactly as you would manage any other `UserWarning`:
|
||||
|
||||
```python
|
||||
import warnings
|
||||
from pydantic_ai_harness.cache_stability import CacheBustWarning
|
||||
|
||||
# Silence the whole category:
|
||||
warnings.filterwarnings('ignore', category=CacheBustWarning)
|
||||
|
||||
# Silence one intentional bust, scoped to the operation that causes it:
|
||||
with warnings.catch_warnings():
|
||||
warnings.simplefilter('ignore', CacheBustWarning)
|
||||
result = agent.run_sync('...') # e.g. a step that switches models or adds a file
|
||||
|
||||
# Treat every bust as an error (dev/CI enforcement):
|
||||
warnings.filterwarnings('error', category=CacheBustWarning)
|
||||
```
|
||||
|
||||
In tests, assert an intentional bust with `pytest.warns(CacheBustWarning)`, or silence a legitimately-busting test with `@pytest.mark.filterwarnings('ignore::pydantic_ai_harness.cache_stability.CacheBustWarning')`.
|
||||
|
||||
## Logfire
|
||||
|
||||
Logfire bridges the stdlib `logging` module, not the `warnings` module, so a `CacheBustWarning` does not reach your traces on its own. To route busts into Logfire, redirect Python warnings to the `logging` system once at startup:
|
||||
|
||||
```python
|
||||
import logging
|
||||
|
||||
logging.captureWarnings(True) # warnings.warn(...) -> the 'py.warnings' logger -> Logfire
|
||||
```
|
||||
|
||||
For now the monitor emits a warning only; a dedicated Logfire event may be added as the capability matures.
|
||||
|
||||
## Composition
|
||||
|
||||
- The monitor only implements `for_run` and `after_model_request`; it adds no tools, instructions, or model settings, so it composes with any other capability, toolset, or `ToolSearch` setup without interference.
|
||||
- Per-run state (the per-key marks and timing) is materialized in `for_run`, so one `CacheStabilityMonitor` instance can be reused across many `Agent.run` calls -- each run is judged independently.
|
||||
|
||||
## Scope
|
||||
|
||||
- **Observational only.** It reports that a cached prefix collapsed, not why -- a moved prefix and a provider-side cache expiry look the same from the token counts, so the warning names both. The structural explanation ("what moved the prefix this turn") is a separate job.
|
||||
- **Fires only when caching is enabled and reported.** A run that never establishes a cache never warns.
|
||||
- **A mid-run model switch does not warn.** Marks are per `(provider_name, model_name)`, so a `FallbackModel` failover starts a fresh mark rather than collapsing the previous model's.
|
||||
|
||||
## API reference
|
||||
|
||||
- [`pydantic_ai_harness.cache_stability` source](https://github.com/pydantic/pydantic-ai-harness/tree/main/pydantic_ai_harness/cache_stability/)
|
||||
- [Pydantic AI capabilities](/ai/core-concepts/capabilities/)
|
||||
- [Pydantic AI hooks](/ai/core-concepts/hooks/)
|
||||
|
||||
The public module exports `CacheStabilityMonitor` and `CacheBustWarning`. Import them from `pydantic_ai_harness.cache_stability`.
|
||||
|
||||
::: pydantic_ai_harness.cache_stability.CacheStabilityMonitor
|
||||
|
||||
::: pydantic_ai_harness.cache_stability.CacheBustWarning
|
||||
@@ -118,6 +118,7 @@ Each capability is a self-contained battery you drop into an agent's `capabiliti
|
||||
| [Pydantic AI Docs](pydantic-ai-docs.md) | An on-demand `read_pyai_docs` tool that pulls Pydantic AI documentation into the run when the agent needs it, instead of preloading it. | -- |
|
||||
| [Compaction](compaction.md) | Keeps a run within token limits: sliding-window trimming, LLM-powered summarization of older messages, and warnings before the context or iteration ceiling is hit. | -- |
|
||||
| [Overflowing Tool Output](overflowing-tool-output.md) | Reduces an oversized tool return when it is produced -- truncate, spill to a queryable file, or summarize -- so a large payload does not persist in history and get re-sent every request. | -- |
|
||||
| [Cache Stability Monitor](cache-stability.md) | Warns when a run's prompt-cache hit collapses between model requests -- a moved cacheable prefix or an expired provider cache -- reading the provider's own `cache_read_tokens` verdict. | -- |
|
||||
| [Step Persistence](step-persistence.md) | Saves and restores full conversation state; snapshot, resume (`continue_run`), and fork (`fork_run`) a run. | -- |
|
||||
| [Media](media.md) | Offloads large `BinaryContent` to content-addressed stores (local or S3) so big media does not bloat message history. | -- |
|
||||
| [Subagents](subagents.md) | Delegates subtasks to specialized child agents through a delegate tool. | -- |
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
{ "label": "Pydantic AI Docs", "slug": "pydantic-ai-docs" },
|
||||
{ "label": "Compaction", "slug": "compaction" },
|
||||
{ "label": "Overflowing Tool Output", "slug": "overflowing-tool-output" },
|
||||
{ "label": "Cache Stability Monitor", "slug": "cache-stability" },
|
||||
{ "label": "Step Persistence", "slug": "step-persistence" },
|
||||
{ "label": "Media", "slug": "media" },
|
||||
{ "label": "Subagents", "slug": "subagents" },
|
||||
|
||||
@@ -13,25 +13,30 @@ that collapse visible.
|
||||
|
||||
This is the **observe** signal: it reads the provider's own verdict rather than
|
||||
guessing from the structured request. On each response it reads
|
||||
`usage.cache_read_tokens` and tracks the largest cacheable prefix the run has
|
||||
established (`cache_read_tokens + cache_write_tokens`, a high-water mark), keyed by
|
||||
the response's `(provider_name, model_name)`. Because message history is
|
||||
append-only, a stable prefix means each request for that model reads back at least
|
||||
what the previous one cached; a large drop is the observable signature of a
|
||||
collapse.
|
||||
`usage.cache_read_tokens` and tracks the cacheable prefix the run has established
|
||||
(`cache_read_tokens + cache_write_tokens`), keyed by the response's
|
||||
`(provider_name, model_name)`. Because message history is append-only, a stable
|
||||
prefix means each request for that model reads back at least what the previous one
|
||||
cached; a large drop is the observable signature of a collapse.
|
||||
|
||||
When a request reads back less than `collapse_ratio` of the established prefix, the
|
||||
monitor warns and re-baselines the tracked prefix to the collapsed value. A sustained
|
||||
collapse -- for example caching toggled off mid-run, which reports `read == 0,
|
||||
write == 0` on every remaining step -- therefore warns once, not on every request.
|
||||
|
||||
Keying per provider and model means a mid-run model switch does not warn: a
|
||||
`FallbackModel` failover or a per-step model change uses a different cache key, so
|
||||
the monitor starts a fresh high-water mark for it instead of comparing against the
|
||||
previous model's. Marks are kept per key rather than reset, so switching back to an
|
||||
earlier model within its cache TTL still compares against that model's prefix.
|
||||
the monitor starts a fresh mark for it instead of comparing against the previous
|
||||
model's. Marks are kept per key rather than reset, so switching back to an earlier
|
||||
model within its cache TTL still compares against that model's prefix.
|
||||
|
||||
A collapse has two shapes the monitor cannot tell apart, so the warning names both:
|
||||
the cacheable prefix moved, or the provider's cache expired under an unchanged prefix
|
||||
(a gap between requests longer than the cache TTL -- Anthropic's default is 5 minutes,
|
||||
refreshed on each hit). When the gap since the previous request exceeds
|
||||
refreshed on each hit). When the gap since the same model's previous request exceeds
|
||||
`cache_ttl_seconds`, the message reports the gap so a long tool or approval pause
|
||||
isn't mistaken for a moved prefix.
|
||||
isn't mistaken for a moved prefix. The gap is timed per model, so switching away and
|
||||
back measures the returning model's own idle time, not whatever ran in between.
|
||||
|
||||
The verdict is cross-provider for free -- pyai normalizes every provider into the
|
||||
`cache_read_tokens` / `cache_write_tokens` fields on `RequestUsage`.
|
||||
@@ -43,9 +48,10 @@ The verdict is cross-provider for free -- pyai normalizes every provider into th
|
||||
> from pydantic_ai_harness.cache_stability import CacheStabilityMonitor
|
||||
> ```
|
||||
|
||||
This is a preview capability: its surface is not a committed public API and may
|
||||
change or be removed in any release. It is the opt-in observe arm of the broader
|
||||
prompt-cache-prefix-stability work.
|
||||
Cache Stability Monitor is a released, non-experimental capability. Pydantic AI
|
||||
Harness is still on 0.x releases, so the API may change between minor releases. See
|
||||
the repository [version policy](https://github.com/pydantic/pydantic-ai-harness#version-policy).
|
||||
It is the opt-in observe arm of the broader prompt-cache-prefix-stability work.
|
||||
|
||||
## Minimal usage
|
||||
|
||||
@@ -67,14 +73,15 @@ belongs at the wire level in tests, not here.
|
||||
- `collapse_ratio` (default `0.5`): warn when a request reads back less than this
|
||||
fraction of the established prefix. Conservative by default so ordinary rounding
|
||||
or a partial miss does not fire; raise toward `1.0` to warn on smaller
|
||||
regressions.
|
||||
regressions. It must be greater than `0.0` -- a ratio of `0.0` could never warn,
|
||||
so it is rejected rather than treated as a silent disable switch.
|
||||
- `min_prefix_tokens` (default `1024`): only judge collapse once the established
|
||||
prefix reaches this many tokens. Below a provider's minimum cacheable size
|
||||
(Anthropic's is 1024) `cache_read_tokens` is noisy or zero.
|
||||
- `cache_ttl_seconds` (default `300`): the assumed provider cache TTL. Message-only
|
||||
-- when the gap since the previous request exceeds it, the warning notes the
|
||||
collapse may be a cache expiry rather than a moved prefix. It does not change
|
||||
whether a warning fires. Lower it for providers with a shorter cache lifetime.
|
||||
-- when the gap since the same model's previous request exceeds it, the warning
|
||||
notes the collapse may be a cache expiry rather than a moved prefix. It does not
|
||||
change whether a warning fires. Lower it for providers with a shorter cache lifetime.
|
||||
|
||||
## Silencing and escalation
|
||||
|
||||
|
||||
@@ -15,9 +15,9 @@ lives at the wire level in `tests/` (VCR cassette prefix assertion), not here.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
import warnings
|
||||
from dataclasses import dataclass, field, replace
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic_ai.capabilities import AbstractCapability
|
||||
from pydantic_ai.messages import ModelResponse
|
||||
@@ -27,6 +27,11 @@ from pydantic_ai.tools import AgentDepsT, RunContext
|
||||
# A response's (provider_name, model_name) -- identifies which provider cache the tokens came from.
|
||||
_CacheKey = tuple[str | None, str | None]
|
||||
|
||||
# Monotonic clock seam. Referenced as `_now()` so a test can monkeypatch it to drive
|
||||
# inter-request gaps deterministically; `response.timestamp` is unusable for this because
|
||||
# it is provider-populated and can skew or run backwards across a provider switch.
|
||||
_now = time.monotonic
|
||||
|
||||
_SILENCE_HINT = (
|
||||
' import warnings\n'
|
||||
' from pydantic_ai_harness.cache_stability import CacheBustWarning\n'
|
||||
@@ -35,6 +40,22 @@ _SILENCE_HINT = (
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _KeyState:
|
||||
"""Per-(provider, model) cache observation: the established prefix and when it was last seen."""
|
||||
|
||||
prefix: int
|
||||
seen_at: float
|
||||
|
||||
|
||||
@dataclass
|
||||
class _RunState:
|
||||
"""Per-run cache-observation state, rebuilt fresh for each run so runs are judged alone."""
|
||||
|
||||
step: int = 0
|
||||
keys: dict[_CacheKey, _KeyState] = field(default_factory=dict[_CacheKey, _KeyState])
|
||||
|
||||
|
||||
class CacheBustWarning(UserWarning):
|
||||
"""Warned when a previously-established prompt cache hit collapses on a later request.
|
||||
|
||||
@@ -72,16 +93,19 @@ class CacheStabilityMonitor(AbstractCapability[AgentDepsT]):
|
||||
"""Warn when a run's prompt cache hit collapses between requests.
|
||||
|
||||
Attach it to any agent whose model uses prompt caching. On each response the monitor
|
||||
reads `usage.cache_read_tokens` and tracks the largest cacheable prefix the run has
|
||||
established so far (`cache_read_tokens + cache_write_tokens`, a high-water mark), keyed by
|
||||
the response's `(provider_name, model_name)`. When a later request for the same key reads
|
||||
back fewer than `collapse_ratio` of that established prefix, it emits a `CacheBustWarning`.
|
||||
reads `usage.cache_read_tokens` and tracks the cacheable prefix the run has established
|
||||
(`cache_read_tokens + cache_write_tokens`), keyed by the response's
|
||||
`(provider_name, model_name)`. When a later request for the same key reads back fewer than
|
||||
`collapse_ratio` of that established prefix, it emits a `CacheBustWarning` and re-baselines
|
||||
the tracked prefix to the collapsed value, so a sustained collapse warns once rather than on
|
||||
every subsequent request.
|
||||
|
||||
Keying per provider and model means a mid-run model switch does not warn: a `FallbackModel`
|
||||
failover or a per-step model change uses a different cache key, so it starts a fresh high-water
|
||||
mark for that key instead of comparing against the previous model's. Marks are kept per key
|
||||
failover or a per-step model change uses a different cache key, so it starts a fresh mark
|
||||
for that key instead of comparing against the previous model's. Marks are kept per key
|
||||
rather than reset, so switching back to an earlier model within its cache TTL still compares
|
||||
against that model's established prefix.
|
||||
against that model's established prefix -- and the expiry hedge measures the gap against that
|
||||
same model's previous request, not whatever ran in between.
|
||||
|
||||
Because message history is append-only, a stable prefix means each request reads back at
|
||||
least what the previous one cached. A large drop is the observable signature of a collapse,
|
||||
@@ -107,7 +131,8 @@ class CacheStabilityMonitor(AbstractCapability[AgentDepsT]):
|
||||
|
||||
Conservative by default (0.5): only a drop below half the previously-cached prefix counts
|
||||
as a collapse, so ordinary provider rounding or a partial cache miss does not fire. Raise
|
||||
it toward 1.0 to warn on smaller regressions.
|
||||
it toward 1.0 to warn on smaller regressions. Must be greater than 0.0 (a ratio of 0.0
|
||||
could never warn, so it is rejected rather than treated as a silent disable switch).
|
||||
"""
|
||||
|
||||
min_prefix_tokens: int = 1024
|
||||
@@ -120,26 +145,27 @@ class CacheStabilityMonitor(AbstractCapability[AgentDepsT]):
|
||||
cache_ttl_seconds: float = 300.0
|
||||
"""Assumed provider cache TTL, in seconds (Anthropic's default is 300, refreshed on each hit).
|
||||
|
||||
Message-only: when the gap since the previous request in the run exceeds this, the warning
|
||||
notes that the collapse may be a provider-side cache expiry rather than a moved prefix. It
|
||||
does not change whether a warning fires. Lower it for providers with a shorter cache lifetime.
|
||||
Message-only: when the gap since the previous request for the same model exceeds this, the
|
||||
warning notes that the collapse may be a provider-side cache expiry rather than a moved
|
||||
prefix. It does not change whether a warning fires. Lower it for providers with a shorter
|
||||
cache lifetime.
|
||||
"""
|
||||
# TODO(#6337): once ModelProfile.prompt_cache_retention ships, prefer the per-model profile
|
||||
# value over this single default -- the monitor is already keyed per model.
|
||||
|
||||
_marks: dict[_CacheKey, int] = field(default_factory=dict[_CacheKey, int], compare=False)
|
||||
_step: int = field(default=0, compare=False)
|
||||
_last_time: datetime | None = field(default=None, compare=False)
|
||||
_state: _RunState = field(init=False, default_factory=_RunState, compare=False, repr=False)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not 0.0 <= self.collapse_ratio <= 1.0:
|
||||
raise ValueError('collapse_ratio must be between 0.0 and 1.0')
|
||||
if not 0.0 < self.collapse_ratio <= 1.0:
|
||||
raise ValueError('collapse_ratio must be greater than 0.0 and at most 1.0')
|
||||
if self.min_prefix_tokens < 0:
|
||||
raise ValueError('min_prefix_tokens must be non-negative')
|
||||
if self.cache_ttl_seconds < 0:
|
||||
raise ValueError('cache_ttl_seconds must be non-negative')
|
||||
if self.cache_ttl_seconds <= 0:
|
||||
raise ValueError('cache_ttl_seconds must be positive')
|
||||
|
||||
async def for_run(self, ctx: RunContext[AgentDepsT]) -> AbstractCapability[AgentDepsT]:
|
||||
"""Reset per-run state (per-key high-water marks, step, timing) so each run is judged alone."""
|
||||
return replace(self, _marks={}, _step=0, _last_time=None)
|
||||
"""Give this run a fresh per-key state (marks, timing, step) so each run is judged alone."""
|
||||
return replace(self)
|
||||
|
||||
async def after_model_request(
|
||||
self,
|
||||
@@ -149,30 +175,39 @@ class CacheStabilityMonitor(AbstractCapability[AgentDepsT]):
|
||||
response: ModelResponse,
|
||||
) -> ModelResponse:
|
||||
"""Compare this response's cache read against the established prefix for its model, then update it."""
|
||||
self._step += 1
|
||||
state = self._state
|
||||
state.step += 1
|
||||
usage = response.usage
|
||||
read = usage.cache_read_tokens
|
||||
key = (response.provider_name, response.model_name)
|
||||
established = self._marks.get(key, 0)
|
||||
# 0.0 on the run's first request, which has no predecessor to measure a gap against.
|
||||
gap = 0.0 if self._last_time is None else (response.timestamp - self._last_time).total_seconds()
|
||||
self._last_time = response.timestamp
|
||||
now = _now()
|
||||
entry = state.keys.get(key)
|
||||
if entry is None:
|
||||
established, prev_seen = 0, now
|
||||
else:
|
||||
established, prev_seen = entry.prefix, entry.seen_at
|
||||
new_prefix = read + usage.cache_write_tokens
|
||||
if established >= self.min_prefix_tokens and read < established * self.collapse_ratio:
|
||||
wasted = established - read
|
||||
gap = now - prev_seen
|
||||
if gap > self.cache_ttl_seconds:
|
||||
expiry = (
|
||||
f' -- the previous request was ~{gap:.0f}s earlier, '
|
||||
f' -- the previous request for this model was ~{gap:.0f}s earlier, '
|
||||
f'past the assumed ~{self.cache_ttl_seconds:.0f}s cache TTL'
|
||||
)
|
||||
else:
|
||||
expiry = ' (e.g. a gap longer than the cache TTL)'
|
||||
warnings.warn(
|
||||
f'Cache hit collapsed at model request {self._step}: read {read} cached tokens but '
|
||||
f'Cache hit collapsed at model request {state.step}: read {read} cached tokens but '
|
||||
f'a prior request established ~{established} (~{wasted} tokens re-sent uncached). '
|
||||
f"The cacheable prefix moved between requests, or the provider's cache expired{expiry}.\n\n"
|
||||
f'To silence or escalate:\n\n{_SILENCE_HINT}\n',
|
||||
CacheBustWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
self._marks[key] = max(established, read + usage.cache_write_tokens)
|
||||
# Re-baseline to the collapsed prefix so a sustained collapse warns once, not per step.
|
||||
prefix = new_prefix
|
||||
else:
|
||||
prefix = max(established, new_prefix)
|
||||
state.keys[key] = _KeyState(prefix, now)
|
||||
return response
|
||||
|
||||
@@ -11,7 +11,6 @@ that explicitly.
|
||||
from __future__ import annotations
|
||||
|
||||
import warnings
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
from pydantic_ai import Agent
|
||||
@@ -64,9 +63,9 @@ def _agent(usages: list[RequestUsage], monitor: CacheStabilityMonitor[None]) ->
|
||||
def _agent_from_responses(responses: list[ModelResponse], monitor: CacheStabilityMonitor[None]) -> Agent[None, str]:
|
||||
"""Agent whose model replays preset `ModelResponse`s, one per step.
|
||||
|
||||
Lets a test control `provider_name` and `timestamp` per response (a model switch, a wall-clock
|
||||
gap) -- fields `FunctionModel` leaves untouched -- which the simpler `_agent` helper can't.
|
||||
Every response but the last must carry a tool call so the run keeps stepping.
|
||||
Lets a test control `provider_name` per response (a mid-run model switch) -- a field
|
||||
`FunctionModel` leaves untouched -- which the simpler `_agent` helper can't. Every response
|
||||
but the last must carry a tool call so the run keeps stepping.
|
||||
"""
|
||||
state = {'i': 0}
|
||||
|
||||
@@ -81,6 +80,17 @@ def _agent_from_responses(responses: list[ModelResponse], monitor: CacheStabilit
|
||||
return Agent(FunctionModel(fn), deps_type=type(None), capabilities=[monitor], tools=[noop])
|
||||
|
||||
|
||||
def _install_clock(monkeypatch: pytest.MonkeyPatch, times: list[float]) -> None:
|
||||
"""Drive the monitor's monotonic clock with a preset sequence, one value per model request.
|
||||
|
||||
The monitor calls its `_now` seam exactly once per response, so `times` must have one entry
|
||||
per step. This controls the inter-request gap deterministically instead of relying on
|
||||
wall-clock timing.
|
||||
"""
|
||||
seq = iter(times)
|
||||
monkeypatch.setattr('pydantic_ai_harness.cache_stability._capability._now', lambda: next(seq))
|
||||
|
||||
|
||||
async def test_collapse_warns() -> None:
|
||||
"""A large drop in cache_read below the established prefix warns."""
|
||||
usages = [_usage(read=0, write=8000), _usage(read=8000, write=200), _usage(read=500)]
|
||||
@@ -190,12 +200,40 @@ async def test_switch_back_within_ttl_uses_preserved_mark() -> None:
|
||||
assert result.output == 'done'
|
||||
|
||||
|
||||
async def test_expiry_gap_named_when_beyond_ttl() -> None:
|
||||
async def test_expiry_gap_named_when_beyond_ttl(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A collapse after a gap longer than the assumed TTL names the gap, avoiding mis-attribution."""
|
||||
base = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
_install_clock(monkeypatch, [0.0, 400.0])
|
||||
usages = [_usage(read=0, write=8000), _usage(read=100)]
|
||||
agent = _agent(usages, CacheStabilityMonitor())
|
||||
with pytest.warns(CacheBustWarning, match='past the assumed') as record:
|
||||
await agent.run('hi')
|
||||
assert '400s earlier' in str(record[0].message)
|
||||
|
||||
|
||||
async def test_small_gap_keeps_generic_expiry_hedge(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A collapse with a short inter-request gap keeps the generic TTL hedge, not a concrete gap."""
|
||||
_install_clock(monkeypatch, [0.0, 5.0])
|
||||
usages = [_usage(read=0, write=8000), _usage(read=100)]
|
||||
agent = _agent(usages, CacheStabilityMonitor())
|
||||
with pytest.warns(CacheBustWarning) as record:
|
||||
await agent.run('hi')
|
||||
message = str(record[0].message)
|
||||
assert 'e.g. a gap longer than the cache TTL' in message
|
||||
assert 'past the assumed' not in message
|
||||
|
||||
|
||||
async def test_expiry_gap_measured_per_key_after_switch_away_and_back(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""After switching away and back, the expiry gap is measured against the same model's last request.
|
||||
|
||||
A global last-observation clock would time the gap from whatever ran in between (model B at
|
||||
250s), report ~150s, and withhold the expiry hedge exactly when expiry is the likely cause.
|
||||
Keying the clock per model measures A's own gap (400s) and names it.
|
||||
"""
|
||||
_install_clock(monkeypatch, [0.0, 250.0, 400.0])
|
||||
responses = [
|
||||
ModelResponse(parts=[ToolCallPart('noop', {})], usage=_usage(read=0, write=8000), timestamp=base),
|
||||
ModelResponse(parts=[TextPart('done')], usage=_usage(read=100), timestamp=base + timedelta(seconds=400)),
|
||||
ModelResponse(parts=[ToolCallPart('noop', {})], usage=_usage(read=0, write=8000), provider_name='anthropic'),
|
||||
ModelResponse(parts=[ToolCallPart('noop', {})], usage=_usage(read=0, write=8000), provider_name='openai'),
|
||||
ModelResponse(parts=[TextPart('done')], usage=_usage(read=100), provider_name='anthropic'),
|
||||
]
|
||||
agent = _agent_from_responses(responses, CacheStabilityMonitor())
|
||||
with pytest.warns(CacheBustWarning, match='past the assumed') as record:
|
||||
@@ -203,19 +241,20 @@ async def test_expiry_gap_named_when_beyond_ttl() -> None:
|
||||
assert '400s earlier' in str(record[0].message)
|
||||
|
||||
|
||||
async def test_small_gap_keeps_generic_expiry_hedge() -> None:
|
||||
"""A collapse with a short inter-request gap keeps the generic TTL hedge, not a concrete gap."""
|
||||
base = datetime(2026, 1, 1, tzinfo=timezone.utc)
|
||||
responses = [
|
||||
ModelResponse(parts=[ToolCallPart('noop', {})], usage=_usage(read=0, write=8000), timestamp=base),
|
||||
ModelResponse(parts=[TextPart('done')], usage=_usage(read=100), timestamp=base + timedelta(seconds=5)),
|
||||
]
|
||||
agent = _agent_from_responses(responses, CacheStabilityMonitor())
|
||||
async def test_caching_off_mid_run_warns_once_not_per_step() -> None:
|
||||
"""A `0/0` response after an established prefix warns once, not on every remaining request.
|
||||
|
||||
Caching toggled off mid-run reports read==0, write==0. Against a high-water mark that only
|
||||
grew, that tripped the collapse check on every subsequent step. Re-baselining the mark to the
|
||||
collapsed prefix surfaces the collapse once and then stays quiet.
|
||||
"""
|
||||
usages = [_usage(read=0, write=8000), _usage(read=0, write=0), _usage(read=0, write=0)]
|
||||
agent = _agent(usages, CacheStabilityMonitor())
|
||||
with pytest.warns(CacheBustWarning) as record:
|
||||
await agent.run('hi')
|
||||
message = str(record[0].message)
|
||||
assert 'e.g. a gap longer than the cache TTL' in message
|
||||
assert 'past the assumed' not in message
|
||||
busts = [w for w in record if issubclass(w.category, CacheBustWarning)]
|
||||
assert len(busts) == 1
|
||||
assert 'request 2' in str(busts[0].message)
|
||||
|
||||
|
||||
def test_invalid_config_rejected() -> None:
|
||||
@@ -228,3 +267,19 @@ def test_invalid_config_rejected() -> None:
|
||||
CacheStabilityMonitor[None](min_prefix_tokens=-1)
|
||||
with pytest.raises(ValueError, match='cache_ttl_seconds'):
|
||||
CacheStabilityMonitor[None](cache_ttl_seconds=-1.0)
|
||||
|
||||
|
||||
def test_config_boundaries() -> None:
|
||||
"""`collapse_ratio=0.0` (never warns) and `cache_ttl_seconds=0.0` are rejected; `1.0` is accepted."""
|
||||
with pytest.raises(ValueError, match='collapse_ratio'):
|
||||
CacheStabilityMonitor[None](collapse_ratio=0.0)
|
||||
with pytest.raises(ValueError, match='cache_ttl_seconds'):
|
||||
CacheStabilityMonitor[None](cache_ttl_seconds=0.0)
|
||||
# The upper bound is inclusive: 1.0 warns on any regression at all.
|
||||
CacheStabilityMonitor[None](collapse_ratio=1.0)
|
||||
|
||||
|
||||
def test_per_run_state_is_not_constructor_surface() -> None:
|
||||
"""Per-run marks/timing live in non-init state, so they can't be seeded through the constructor."""
|
||||
with pytest.raises(TypeError):
|
||||
CacheStabilityMonitor[None](_state=None) # pyright: ignore[reportCallIssue]
|
||||
|
||||
@@ -120,6 +120,7 @@ _CAPABILITY_PAGE_META = {
|
||||
'pydantic-ai-docs.md': ('docs', 'Pydantic AI Docs'),
|
||||
'compaction.md': ('compaction', 'Compaction'),
|
||||
'overflowing-tool-output.md': ('overflowing_tool_output', 'Overflowing Tool Output'),
|
||||
'cache-stability.md': ('cache_stability', 'Cache Stability Monitor'),
|
||||
'step-persistence.md': ('step_persistence', 'Step Persistence'),
|
||||
'media.md': ('media', 'Media Externalization'),
|
||||
'subagents.md': ('subagents', 'Subagents'),
|
||||
|
||||
Reference in New Issue
Block a user