From 6db5f46f68edc846249d7ff94349b22bbf809ddf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Fri, 3 Jul 2026 13:59:03 +0200 Subject: [PATCH] feat(reflect): step-by-step context caching for the Gemini tool loop Reflect's cost is dominated by the tool-result context re-sent on every turn, not the static system prefix. Caching only the fixed system+tools prefix (Gemini CachedContent) recovered ~9% of input on a short reflect and 0% via implicit caching. Instead, roll a cache forward one step at a time: after each turn extend the CachedContent to cover that turn's full input, so the next `auto` turn reuses the entire prior conversation at the cached-input rate and sends only its own new tool results as the delta. Each retrieved payload is billed fresh exactly once, then cached thereafter. Measured on gemini-2.5-flash-lite: short loops ~29%, deep loops ~74-81% cached input (deepest turns ~99%). - GeminiCacheManager: cache `contents` (growing conversation), session- tracked create/delete for ephemeral per-reflect caches. - call_with_tools gains `cached_prefix_message_count` so only the un-cached tail is sent; forced (tool_config) turns stay uncached as Gemini requires. - The next-turn cache create is scheduled to run concurrently with tool execution, hiding its latency; the reflect wrapper deletes the session (all caches) on every exit path, with a short TTL as backstop. - New HINDSIGHT_API_REFLECT_PROMPT_CACHE_ENABLED flag (default true) to disable reflect caching independently of the global prompt cache. Tests: deterministic step-by-step contract test, cache-manager units, and a real-Gemini cached-ratio measurement test. --- hindsight-api-slim/hindsight_api/config.py | 10 + .../hindsight_api/engine/llm_interface.py | 40 +++ .../hindsight_api/engine/llm_wrapper.py | 12 +- .../engine/providers/gemini_cache.py | 97 ++++++- .../engine/providers/gemini_llm.py | 253 ++++++++++++------ .../hindsight_api/engine/reflect/agent.py | 192 +++++++++++-- hindsight-api-slim/tests/test_gemini_cache.py | 69 +++++ .../tests/test_gemini_implicit_cache_ratio.py | 26 +- .../tests/test_reflect_agent.py | 145 ++++++++++ .../docs/developer/configuration.md | 1 + .../references/developer/configuration.md | 1 + 11 files changed, 726 insertions(+), 120 deletions(-) diff --git a/hindsight-api-slim/hindsight_api/config.py b/hindsight-api-slim/hindsight_api/config.py index 44d2e9e81e..0fffa93236 100644 --- a/hindsight-api-slim/hindsight_api/config.py +++ b/hindsight-api-slim/hindsight_api/config.py @@ -617,6 +617,7 @@ def _resolve_operation_temperature(operation_env: str, default: float) -> float # Reflect agent settings ENV_REFLECT_MAX_ITERATIONS = "HINDSIGHT_API_REFLECT_MAX_ITERATIONS" +ENV_REFLECT_PROMPT_CACHE_ENABLED = "HINDSIGHT_API_REFLECT_PROMPT_CACHE_ENABLED" ENV_REFLECT_MAX_CONTEXT_TOKENS = "HINDSIGHT_API_REFLECT_MAX_CONTEXT_TOKENS" ENV_REFLECT_WALL_TIMEOUT = "HINDSIGHT_API_REFLECT_WALL_TIMEOUT" ENV_REFLECT_MISSION = "HINDSIGHT_API_REFLECT_MISSION" @@ -1054,6 +1055,10 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]: # Reflect agent settings DEFAULT_REFLECT_MAX_ITERATIONS = 10 # Max tool call iterations before forcing response +# Step-by-step context caching for the reflect tool loop (Gemini). On by default; +# requires the global prompt cache (HINDSIGHT_API_LLM_PROMPT_CACHE_ENABLED) to also +# be on. Set false to force reflect to run uncached even when prompt caching is on. +DEFAULT_REFLECT_PROMPT_CACHE_ENABLED = True DEFAULT_REFLECT_MAX_CONTEXT_TOKENS = 100_000 # Max accumulated context tokens before forcing final prompt DEFAULT_REFLECT_WALL_TIMEOUT = 300 # Wall-clock timeout in seconds for the entire reflect operation (5 minutes) DEFAULT_REFLECT_SOURCE_FACTS_MAX_TOKENS = -1 # Token budget for source facts in search_observations (-1 = disabled) @@ -1914,6 +1919,7 @@ class HindsightConfig: reflect_max_iterations: int reflect_max_context_tokens: int reflect_wall_timeout: int + reflect_prompt_cache_enabled: bool # OpenTelemetry tracing configuration otel_traces_enabled: bool @@ -2927,6 +2933,10 @@ def from_env(cls) -> "HindsightConfig": retain_max_concurrent=int(os.getenv(ENV_RETAIN_MAX_CONCURRENT, str(DEFAULT_RETAIN_MAX_CONCURRENT))), # Reflect agent settings reflect_max_iterations=int(os.getenv(ENV_REFLECT_MAX_ITERATIONS, str(DEFAULT_REFLECT_MAX_ITERATIONS))), + reflect_prompt_cache_enabled=os.getenv( + ENV_REFLECT_PROMPT_CACHE_ENABLED, str(DEFAULT_REFLECT_PROMPT_CACHE_ENABLED) + ).lower() + in ("1", "true", "yes", "on"), reflect_max_context_tokens=int( os.getenv(ENV_REFLECT_MAX_CONTEXT_TOKENS, str(DEFAULT_REFLECT_MAX_CONTEXT_TOKENS)) ), diff --git a/hindsight-api-slim/hindsight_api/engine/llm_interface.py b/hindsight-api-slim/hindsight_api/engine/llm_interface.py index b1cf67c9c4..a5eeec50a1 100644 --- a/hindsight-api-slim/hindsight_api/engine/llm_interface.py +++ b/hindsight-api-slim/hindsight_api/engine/llm_interface.py @@ -116,6 +116,7 @@ async def call_with_tools( max_backoff: float = 30.0, tool_choice: str | dict[str, Any] = "auto", cached_prefix: str | None = None, + cached_prefix_message_count: int = 0, ) -> LLMToolCallResult: """ Make an LLM API call with tool/function calling support. @@ -185,6 +186,45 @@ async def get_or_create_cached_prefix( """ return None + # ── Step-by-step incremental prompt caching (optional) ───────────────────── + # + # For agentic loops (reflect) the dominant cost is the conversation prefix + # re-sent every turn, not the static system prefix. Providers that can cache + # a *growing* prefix implement these: the caller rolls one cache per step + # (each covering the previous step's full input), passes its handle plus the + # message count it covers to ``call_with_tools`` so only the new turns are + # sent fresh, and tears the caches down when the loop ends. Default no-ops so + # non-supporting providers transparently run uncached. + + def supports_incremental_prompt_cache(self) -> bool: + """Whether this provider can cache a growing multi-turn conversation prefix.""" + return False + + async def create_incremental_cache( + self, + *, + session_id: str, + messages: list[dict[str, Any]], + tools: list[dict[str, Any]] | None = None, + ) -> str | None: + """Cache ``system + tools + messages`` and return an opaque handle, or None. + + The handle is passed back to ``call_with_tools(cached_prefix=..., + cached_prefix_message_count=len(messages))``. Caches are grouped under + ``session_id`` for teardown via ``delete_cache_session``. Returns None + when caching is unavailable or the prefix is too small — caller falls + back to an uncached call. + """ + return None + + async def delete_cached_prefix(self, name: str) -> None: + """Best-effort delete of a single cache handle (a superseded step).""" + return None + + async def delete_cache_session(self, session_id: str) -> None: + """Best-effort teardown of every cache created under ``session_id``.""" + return None + async def submit_batch( self, requests: list[dict[str, Any]], diff --git a/hindsight-api-slim/hindsight_api/engine/llm_wrapper.py b/hindsight-api-slim/hindsight_api/engine/llm_wrapper.py index 620158f198..c2c62f02cd 100644 --- a/hindsight-api-slim/hindsight_api/engine/llm_wrapper.py +++ b/hindsight-api-slim/hindsight_api/engine/llm_wrapper.py @@ -967,6 +967,7 @@ async def call_with_tools( max_backoff: float | None = None, tool_choice: str | dict[str, Any] = "auto", cached_prefix: str | None = None, + cached_prefix_message_count: int = 0, ) -> "LLMToolCallResult": """ Make an LLM API call with tool/function calling support. @@ -1034,9 +1035,14 @@ async def call_with_tools( await stack.enter_async_context(sem) # cached_prefix is only set for providers that returned a handle - # from get_or_create_cached_prefix(); forward it only when present - # so non-caching providers keep their signature (same as call()). - cache_kwarg = {"cached_prefix": cached_prefix} if cached_prefix is not None else {} + # from get_or_create_cached_prefix() / create_incremental_cache(); + # forward it (plus how many leading messages it covers) only when + # present so non-caching providers keep their signature. + cache_kwarg = ( + {"cached_prefix": cached_prefix, "cached_prefix_message_count": cached_prefix_message_count} + if cached_prefix is not None + else {} + ) try: # Delegate to provider implementation result = await self._provider_impl.call_with_tools( diff --git a/hindsight-api-slim/hindsight_api/engine/providers/gemini_cache.py b/hindsight-api-slim/hindsight_api/engine/providers/gemini_cache.py index 12e5ca715d..0ab2875290 100644 --- a/hindsight-api-slim/hindsight_api/engine/providers/gemini_cache.py +++ b/hindsight-api-slim/hindsight_api/engine/providers/gemini_cache.py @@ -56,6 +56,14 @@ # to None and callers proceed uncached, rather than stalling the whole batch. _DEFAULT_CREATE_TIMEOUT_SECONDS = 30.0 +# TTL for the per-step reflect caches created by ``create_incremental``. These +# live only for the duration of one reflect (seconds), so the TTL is just a +# storage backstop in case the explicit ``delete_session`` at reflect end is +# missed (crash / event-loop teardown). Short so orphaned caches age out fast — +# storage is billed per token-hour, so a 5-minute cap keeps the cost of a leaked +# cache negligible. +_DEFAULT_INCREMENTAL_TTL_SECONDS = 5 * 60 + @dataclass class _CacheEntry: @@ -92,6 +100,10 @@ def __init__( self._create_timeout_seconds = create_timeout_seconds self._entries: dict[str, _CacheEntry] = {} self._lock = asyncio.Lock() + # session_id -> CachedContent names created via ``create_incremental``. + # A reflect creates a fresh rolling cache per step under one session id; + # ``delete_session`` tears them all down when the reflect finishes. + self._sessions: dict[str, list[str]] = {} @staticmethod def fingerprint( @@ -229,18 +241,99 @@ def invalidate(self, name: str) -> None: if entry.name == name: self._entries.pop(key, None) + async def create_incremental( + self, + *, + session_id: str, + model: str, + system_instruction: str, + contents: list[Any], + tools: list[dict[str, Any]] | None = None, + ) -> str | None: + """Create a fresh CachedContent holding ``system + tools + contents`` and + track it under ``session_id`` for later teardown. + + Unlike ``get_or_create``, this does NOT deduplicate by fingerprint: each + step of a reflect grows the conversation prefix, so every call is a + distinct, single-use cache. The reflect loop creates one per step (each + covering the previous step's full input) and reuses it for exactly the + next model turn, then supersedes it. All caches for the session are + deleted by ``delete_session`` when the reflect ends; the short TTL is + only a backstop. + + Returns the cache resource name, or ``None`` when caching is disabled, + the prefix is below the model minimum, or the create otherwise fails — + callers MUST fall back to an uncached call in that case. + """ + try: + name = await self._create_cache( + model=model, + system_instruction=system_instruction, + tools=tools, + contents=contents, + ttl_seconds=_DEFAULT_INCREMENTAL_TTL_SECONDS, + ) + except _CacheNotEligible as e: + logger.debug( + "GeminiCacheManager: incremental prefix not eligible (model=%s, reason=%s) — caller falls back", + model, + e, + ) + return None + except Exception: + logger.exception( + "GeminiCacheManager: failed to create incremental cache (model=%s); caller falls back", + model, + ) + return None + if name is not None: + self._sessions.setdefault(session_id, []).append(name) + return name + + async def delete(self, name: str) -> None: + """Best-effort server-side delete of a single CachedContent. + + Swallows all errors: a failed delete just means the cache ages out on + its TTL. Also drops any matching in-process entry. + """ + self.invalidate(name) + try: + await self._client.aio.caches.delete(name=name) + except Exception: + logger.debug("GeminiCacheManager: delete of cache %s failed (will age out on TTL)", name, exc_info=True) + + async def delete_session(self, session_id: str) -> None: + """Delete every CachedContent created for ``session_id`` (reflect teardown). + + Deletes concurrently and best-effort — a reflect must never fail because + a cache couldn't be torn down; the short TTL is the backstop. + """ + names = self._sessions.pop(session_id, []) + if not names: + return + await asyncio.gather(*(self.delete(n) for n in names), return_exceptions=True) + async def _create_cache( self, *, model: str, system_instruction: str, tools: list[dict[str, Any]] | None = None, + contents: list[Any] | None = None, + ttl_seconds: int | None = None, ) -> str | None: """Wrap ``client.aio.caches.create`` with the config we want. The SDK surface differs slightly across google-genai versions; this implementation targets the >=1.0.0 line where caches live under ``client.aio.caches``. + + ``contents`` (already-converted ``genai_types.Content`` turns) is + appended after the system_instruction/tools so the cache can hold a + growing multi-turn conversation prefix, not just the static prefix — + this is what the step-by-step reflect cache relies on. ``ttl_seconds`` + overrides the manager default (used to give per-step reflect caches a + short backstop TTL). """ # Lazy import so this module doesn't require the SDK at import time. from google.genai import types as genai_types @@ -254,8 +347,10 @@ async def _create_cache( # still part of the fingerprint so a schema change keys a fresh cache. config_kwargs: dict[str, Any] = { "system_instruction": system_instruction, - "ttl": f"{self._ttl_seconds}s", + "ttl": f"{ttl_seconds if ttl_seconds is not None else self._ttl_seconds}s", } + if contents: + config_kwargs["contents"] = contents if tools: # OpenAI-style {"function": {...}} entries must be converted to # Gemini's Tool/FunctionDeclaration shape before caching. diff --git a/hindsight-api-slim/hindsight_api/engine/providers/gemini_llm.py b/hindsight-api-slim/hindsight_api/engine/providers/gemini_llm.py index 93fad79d95..b8c3225638 100644 --- a/hindsight-api-slim/hindsight_api/engine/providers/gemini_llm.py +++ b/hindsight-api-slim/hindsight_api/engine/providers/gemini_llm.py @@ -13,6 +13,7 @@ import logging import time from contextvars import ContextVar +from dataclasses import dataclass from typing import Any from google import genai @@ -63,6 +64,77 @@ def _usage_from_gemini_response(response: Any) -> LLMResponseUsage: ) +@dataclass(frozen=True) +class _GeminiConversation: + """A message list converted to Gemini's request shape.""" + + system_instruction: str | None + contents: list["genai_types.Content"] + + +def _convert_messages_to_gemini(msg_list: list[dict[str, Any]]) -> _GeminiConversation: + """Convert OpenAI-style messages to a Gemini (system_instruction, contents) pair. + + Shared by ``call_with_tools`` (request body) and the incremental cache + builder so a cached prefix and the live request serialise turns identically — + any drift would fingerprint differently and defeat the cache. Consecutive + ``role="tool"`` messages are grouped into a single ``user`` Content with + multiple FunctionResponse parts, matching Gemini's multi-turn requirement. + """ + system_instruction: str | None = None + gemini_contents: list[genai_types.Content] = [] + i = 0 + while i < len(msg_list): + msg = msg_list[i] + role = msg.get("role", "user") + content = msg.get("content", "") + + if role == "system": + system_instruction = (system_instruction + "\n\n" + content) if system_instruction else content + i += 1 + elif role == "tool": + parts = [] + while i < len(msg_list) and msg_list[i].get("role") == "tool": + tool_msg = msg_list[i] + tool_content = tool_msg.get("content", "") + parts.append( + genai_types.Part( + function_response=genai_types.FunctionResponse( + name=tool_msg.get("name", ""), + response={"result": tool_content}, + ) + ) + ) + i += 1 + gemini_contents.append(genai_types.Content(role="user", parts=parts)) + elif role == "assistant": + tool_calls_in_msg = msg.get("tool_calls", []) + if tool_calls_in_msg: + parts = [] + if content: + parts.append(genai_types.Part(text=content)) + for tc in tool_calls_in_msg: + fn = tc.get("function", {}) + fn_name = fn.get("name", "") + fn_args_str = fn.get("arguments", "{}") + fn_args = parse_llm_json(fn_args_str) + thought_signature = tc.get("thought_signature") + fc_kwargs: dict[str, Any] = {"name": fn_name, "args": fn_args} + part_kwargs: dict[str, Any] = {"function_call": genai_types.FunctionCall(**fc_kwargs)} + if thought_signature: + part_kwargs["thought_signature"] = base64.b64decode(thought_signature) + parts.append(genai_types.Part(**part_kwargs)) + gemini_contents.append(genai_types.Content(role="model", parts=parts)) + else: + gemini_contents.append(genai_types.Content(role="model", parts=[genai_types.Part(text=content)])) + i += 1 + else: + gemini_contents.append(genai_types.Content(role="user", parts=[genai_types.Part(text=content)])) + i += 1 + + return _GeminiConversation(system_instruction=system_instruction, contents=gemini_contents) + + class GeminiLLM(LLMInterface): """ LLM provider for Google Gemini and Vertex AI. @@ -527,6 +599,7 @@ async def call_with_tools( max_backoff: float = 30.0, tool_choice: str | dict[str, Any] = "auto", cached_prefix: str | None = None, + cached_prefix_message_count: int = 0, ) -> LLMToolCallResult: """ Make a Gemini/VertexAI API call with tool/function calling support. @@ -542,13 +615,20 @@ async def call_with_tools( max_backoff: Maximum backoff time in seconds. tool_choice: How to choose tools (Gemini uses "auto" only). cached_prefix: Optional CachedContent resource name (from - ``GeminiCacheManager.get_or_create`` with ``tools=...``). When - set, the system_instruction and tool definitions are assumed + ``GeminiCacheManager.get_or_create`` or ``create_incremental``). + When set, the system_instruction and tool definitions are assumed to live in the cache; this call will skip resending them and the cached prefix is billed at the cached-input rate. The ``tools`` argument is still required (the caller may pass an empty list when the cache holds them) so existing call sites don't break. + cached_prefix_message_count: Number of leading ``messages`` already + baked into ``cached_prefix`` (the step-by-step reflect cache holds + a growing conversation prefix, not just system+tools). Only the + messages AFTER this index are sent as request contents — the rest + come from the cache and bill at the cached rate. 0 means the cache + holds only the static prefix (system+tools), so the full + conversation is still sent (legacy behaviour). Returns: LLMToolCallResult with content and/or tool_calls. @@ -556,86 +636,45 @@ async def call_with_tools( start_time = time.time() using_cache = cached_prefix is not None - # Convert tools to Gemini format. When the cache is in use, the - # tool definitions are baked into the CachedContent at create time - # and the SDK rejects re-sending them alongside ``cached_content``. + # Convert tools to Gemini format. While the cache is in use the tool + # definitions live in the CachedContent and the SDK rejects re-sending + # them alongside ``cached_content`` (see ``_build_tools_config``), but we + # still build them unconditionally so the cached-call-failed fallback — + # which drops the cache and re-sends prefix + tools inline — has real + # tools to send rather than an empty list. gemini_tools = [] - if not using_cache: - for tool in tools: - func = tool.get("function", {}) - gemini_tools.append( - genai_types.Tool( - function_declarations=[ - genai_types.FunctionDeclaration( - name=func.get("name", ""), - description=func.get("description", ""), - parameters=func.get("parameters"), - ) - ] - ) + for tool in tools: + func = tool.get("function", {}) + gemini_tools.append( + genai_types.Tool( + function_declarations=[ + genai_types.FunctionDeclaration( + name=func.get("name", ""), + description=func.get("description", ""), + parameters=func.get("parameters"), + ) + ] ) + ) - # Convert messages - system_instruction = None - gemini_contents = [] - msg_list = list(messages) - i = 0 - while i < len(msg_list): - msg = msg_list[i] - role = msg.get("role", "user") - content = msg.get("content", "") - - if role == "system": - # Always capture system_instruction. _build_tools_config omits it - # (and tools) from the request while the cache carries the prefix, - # but it must be available so the cached-call-failed safety net can - # re-send the prefix + tools inline. - system_instruction = (system_instruction + "\n\n" + content) if system_instruction else content - i += 1 - elif role == "tool": - # Gemini requires ALL tool responses for a given model turn to be grouped - # into a single Content with multiple FunctionResponse parts. - # Consecutive role="tool" messages correspond to one model turn's tool calls. - parts = [] - while i < len(msg_list) and msg_list[i].get("role") == "tool": - tool_msg = msg_list[i] - tool_content = tool_msg.get("content", "") - parts.append( - genai_types.Part( - function_response=genai_types.FunctionResponse( - name=tool_msg.get("name", ""), - response={"result": tool_content}, - ) - ) - ) - i += 1 - gemini_contents.append(genai_types.Content(role="user", parts=parts)) - elif role == "assistant": - tool_calls_in_msg = msg.get("tool_calls", []) - if tool_calls_in_msg: - # Convert OpenAI-style tool_calls to Gemini function_call parts - # This is required for proper multi-turn conversation history - parts = [] - if content: - parts.append(genai_types.Part(text=content)) - for tc in tool_calls_in_msg: - fn = tc.get("function", {}) - fn_name = fn.get("name", "") - fn_args_str = fn.get("arguments", "{}") - fn_args = parse_llm_json(fn_args_str) - thought_signature = tc.get("thought_signature") - fc_kwargs: dict[str, Any] = {"name": fn_name, "args": fn_args} - part_kwargs: dict[str, Any] = {"function_call": genai_types.FunctionCall(**fc_kwargs)} - if thought_signature: - part_kwargs["thought_signature"] = base64.b64decode(thought_signature) - parts.append(genai_types.Part(**part_kwargs)) - gemini_contents.append(genai_types.Content(role="model", parts=parts)) - else: - gemini_contents.append(genai_types.Content(role="model", parts=[genai_types.Part(text=content)])) - i += 1 - else: - gemini_contents.append(genai_types.Content(role="user", parts=[genai_types.Part(text=content)])) - i += 1 + # Convert messages. ``system_instruction`` and the FULL contents are always + # computed: _build_tools_config omits system/tools from the request while + # the cache carries the prefix, but the cached-call-failed safety net must + # be able to re-send the whole prefix + tools inline. + converted = _convert_messages_to_gemini(list(messages)) + system_instruction = converted.system_instruction + full_contents = converted.contents + + # Step-by-step reflect cache: when the cache already holds the first + # ``cached_prefix_message_count`` messages, send ONLY the newer turns as + # request contents — the cached prefix supplies the rest at the cached + # rate. The split is always at a whole-turn boundary (the reflect loop + # advances the cache one completed turn at a time), so slicing the raw + # messages before conversion never splits a grouped tool turn. + if using_cache and cached_prefix_message_count > 0: + delta_contents = _convert_messages_to_gemini(list(messages)[cached_prefix_message_count:]).contents + else: + delta_contents = full_contents # Apply safety settings: context var (per-request bank override) takes precedence over instance default effective_safety_settings = _safety_settings_ctx.get() @@ -701,10 +740,14 @@ def _build_tools_config(use_cache: bool) -> "genai_types.GenerateContentConfig": if attempt > 0: set_stage(f"llm.gemini.tools.attempt={attempt + 1}/{max_retries + 1}") try: + # With the cache active, send only the un-cached tail (delta); + # on the uncached fallback path send the full conversation so the + # re-inlined system+tools prefix has its whole context. + active_contents = delta_contents if cache_active else full_contents response = await asyncio.wait_for( self._client.aio.models.generate_content( model=self.model, - contents=gemini_contents, + contents=active_contents, config=config, ), timeout=90.0, # Safety net for network hangs; valid slow responses are <90s @@ -883,6 +926,56 @@ async def get_or_create_cached_prefix( tools=tools, ) + # ── Step-by-step incremental prompt caching (reflect tool loop) ────────── + + def supports_incremental_prompt_cache(self) -> bool: + """True when explicit caching is on — the reflect loop can then roll a + per-step CachedContent that grows with the conversation.""" + return self._prompt_cache_enabled + + def _ensure_cache_manager(self) -> Any: + if self._cache_manager is None: + from hindsight_api.engine.providers.gemini_cache import GeminiCacheManager + + self._cache_manager = GeminiCacheManager(self._client) + return self._cache_manager + + async def create_incremental_cache( + self, + *, + session_id: str, + messages: list[dict[str, Any]], + tools: list[dict[str, Any]] | None = None, + ) -> str | None: + """Cache ``system + tools + messages`` as a conversation prefix and return + its resource name (or ``None`` — caller falls back to an uncached call). + + The reflect loop calls this once per step with the growing message list so + each step's cache entirely contains the previous step's input; the next + model turn then references it and re-sends only its own delta. Caches are + tracked under ``session_id`` and torn down by ``delete_cache_session``. + """ + if not self._prompt_cache_enabled or self._client is None: + return None + converted = _convert_messages_to_gemini(list(messages)) + return await self._ensure_cache_manager().create_incremental( + session_id=session_id, + model=self.model, + system_instruction=converted.system_instruction or "", + contents=converted.contents, + tools=tools, + ) + + async def delete_cached_prefix(self, name: str) -> None: + """Best-effort delete of a single CachedContent (superseded reflect step).""" + if self._cache_manager is not None: + await self._cache_manager.delete(name) + + async def delete_cache_session(self, session_id: str) -> None: + """Tear down every CachedContent created for a reflect session.""" + if self._cache_manager is not None: + await self._cache_manager.delete_session(session_id) + # ── Batch API (Gemini API only — not Vertex AI) ───────────────────────── # # Google's Gemini Batch API gives a flat 50% discount on input + output diff --git a/hindsight-api-slim/hindsight_api/engine/reflect/agent.py b/hindsight-api-slim/hindsight_api/engine/reflect/agent.py index bded980467..b809231947 100644 --- a/hindsight-api-slim/hindsight_api/engine/reflect/agent.py +++ b/hindsight-api-slim/hindsight_api/engine/reflect/agent.py @@ -414,6 +414,69 @@ def _all_mental_models_are_usable_and_fresh(tool_output: dict[str, Any]) -> bool async def run_reflect_agent( + llm_config: "LLMProvider", + bank_id: str, + query: str, + bank_profile: dict[str, Any], + search_mental_models_fn: Callable[[str, int], Awaitable[dict[str, Any]]], + search_observations_fn: Callable[[str, int], Awaitable[dict[str, Any]]], + recall_fn: Callable[[str, int, int], Awaitable[dict[str, Any]]], + expand_fn: Callable[[list[str], str], Awaitable[dict[str, Any]]], + **kwargs: Any, +) -> ReflectAgentResult: + """Public entrypoint: runs the agent loop and guarantees teardown of any + per-step context caches created during it. + + The step-by-step caches (Gemini ``CachedContent``) are ephemeral — scoped to + exactly one reflect — so they're deleted the moment the loop returns, whatever + the exit path (answer, error, cancellation). The short cache TTL is only a + backstop if this teardown is somehow missed; the delete is best-effort and + never allowed to fail a reflect. + """ + reflect_id = f"{bank_id[:8]}-{int(time.time() * 1000) % 100000}" + provider_impl = getattr(llm_config, "_provider_impl", None) + # Reflect step-by-step caching needs the provider to support it AND the + # dedicated reflect flag (on by default; distinct from the global prompt-cache + # switch so it can be turned off for reflect alone). + incremental_caching = ( + provider_impl is not None + and provider_impl.supports_incremental_prompt_cache() + and get_config().reflect_prompt_cache_enabled + ) + cache_session_id = f"reflect:{reflect_id}" + # In-flight cache-create tasks (scheduled to overlap tool execution). Awaited + # before teardown so every created cache is tracked and deleted — no orphans. + cache_tasks: list[asyncio.Task] = [] + try: + return await _run_reflect_agent_inner( + llm_config, + bank_id, + query, + bank_profile, + search_mental_models_fn, + search_observations_fn, + recall_fn, + expand_fn, + reflect_id=reflect_id, + provider_impl=provider_impl, + incremental_caching=incremental_caching, + cache_session_id=cache_session_id, + cache_tasks=cache_tasks, + **kwargs, + ) + finally: + if incremental_caching and provider_impl is not None: + try: + # Let any overlapped create finish so its cache is registered, + # then delete the whole session. + if cache_tasks: + await asyncio.gather(*cache_tasks, return_exceptions=True) + await provider_impl.delete_cache_session(cache_session_id) + except Exception: + logger.debug("[REFLECT %s] cache session teardown failed (will age out on TTL)", reflect_id) + + +async def _run_reflect_agent_inner( llm_config: "LLMProvider", bank_id: str, query: str, @@ -434,6 +497,12 @@ async def run_reflect_agent( max_context_tokens: int = 100_000, llm_output_language: str | None = None, cancel_check: Callable[[], None] | None = None, + *, + reflect_id: str, + provider_impl: Any, + incremental_caching: bool, + cache_session_id: str, + cache_tasks: list[asyncio.Task], ) -> ReflectAgentResult: """ Execute the reflect agent loop using native tool calling. @@ -461,7 +530,6 @@ async def run_reflect_agent( Returns: ReflectAgentResult with final answer and metadata """ - reflect_id = f"{bank_id[:8]}-{int(time.time() * 1000) % 100000}" start_time = time.time() # Build directives_applied for the trace @@ -498,27 +566,68 @@ async def run_reflect_agent( {"role": "user", "content": query}, ] - # Opt into context caching for the agentic tool loop. The system - # prompt and tool definitions are stable for the duration of this - # reflect call (and across reflects against the same bank), so - # caching them once and reusing across every iteration of the loop - # collapses the dominant input cost — the prefix repeated on every - # turn. ``get_or_create_cached_prefix`` returns None when caching is - # disabled, unsupported, or the prefix is too small; the - # ``call_with_tools`` invocation below transparently falls back to - # the uncached path in that case. - cached_prefix_name: str | None = None - provider_impl = getattr(llm_config, "_provider_impl", None) - if provider_impl is not None and provider_impl.supports_prompt_caching(): + # Step-by-step context caching for the agentic tool loop. + # + # Caching only the static system+tools prefix wins little here: it's dwarfed + # by the tool results (recall/observations) that get re-sent on every turn. + # Instead we roll a cache forward one step at a time — after each turn the + # cache is extended to cover that turn's FULL input, so the next ``auto`` turn + # reuses the entire prior conversation at the cached rate and sends only its + # own new tool results as the delta. Each new tool payload is therefore billed + # at full price exactly once (the turn it's produced), then cached thereafter. + # + # The cache create for turn N+1 covers turn N's input, which is fully known the + # moment turn N's LLM call returns — so we kick it off as a background task that + # runs CONCURRENTLY with turn N's tool execution (``_schedule_cache``) and only + # await it (``_resolve_pending_cache``) right before the next ``auto`` call, + # hiding the create latency behind work we'd do anyway. + # + # ``rolling_cache_boundary`` is the number of leading ``messages`` baked into + # the adopted ``rolling_cache_name``. ``incremental_caching`` is False for + # providers/config without explicit caching, so every branch below is a no-op. + rolling_cache_name: str | None = None + rolling_cache_boundary = 0 + pending_cache_task: asyncio.Task | None = None + pending_cache_boundary = 0 + + async def _resolve_pending_cache() -> None: + """Adopt the overlapped next-cache once it's ready as the rolling cache. + + Best-effort: a failed/``None`` create just leaves the previous (smaller) + cache in place, so the next call sends a larger delta but stays correct. + """ + nonlocal rolling_cache_name, rolling_cache_boundary, pending_cache_task + if pending_cache_task is None: + return + task = pending_cache_task + pending_cache_task = None try: - cached_prefix_name = await provider_impl.get_or_create_cached_prefix( - system_instruction=system_prompt, - tools=tools, + new_name = await task + except Exception: + new_name = None + if new_name is not None: + rolling_cache_name = new_name + rolling_cache_boundary = pending_cache_boundary + + def _schedule_cache(upto: int) -> None: + """Start building the cache covering ``messages[:upto]`` in the background + so it overlaps the tool execution that follows this turn.""" + nonlocal pending_cache_task, pending_cache_boundary + # ``messages[:upto]`` is snapshotted now, so appends during tool execution + # can't change what gets cached. ``ensure_future`` raises if the provider + # didn't return a coroutine (e.g. a test double) — caching is a soft + # optimisation and must never break a reflect, so swallow and skip. + try: + task = asyncio.ensure_future( + provider_impl.create_incremental_cache( + session_id=cache_session_id, messages=messages[:upto], tools=tools + ) ) except Exception: - # Caching is a soft optimisation; never let a cache-side - # error block a reflect. - cached_prefix_name = None + return + pending_cache_boundary = upto + pending_cache_task = task + cache_tasks.append(task) # Tracking total_tools_called = 0 @@ -745,6 +854,29 @@ def _log_completion(answer: str, iterations: int, forced: bool = False): else: iter_tool_choice = "auto" + # Will the NEXT turn be an ``auto`` turn (the only kind that references a + # cache)? The cache we schedule this turn covers this turn's input and is + # used by the next turn, so we only bother building it when the next turn + # can use it — skipping the wasted creates between two forced turns. + next_iter = iteration + 1 + if stop_forcing_from_iteration is not None and next_iter >= stop_forcing_from_iteration: + next_is_auto = True + elif next_iter < len(forced_sequence): + next_is_auto = False + else: + next_is_auto = True + + # Before an ``auto`` turn, adopt the cache that was being built in the + # background during the previous turn's tool execution. It covers that + # turn's full input, so THIS call reuses the entire prior conversation at + # the cached rate and sends only the turns appended since. Forced turns + # can't use a cache (Gemini rejects ``cached_content`` + ``tool_config``), + # but the cache still advances underneath them, so the first ``auto`` turn + # inherits a cache covering all the forced results. + if incremental_caching and iter_tool_choice == "auto": + await _resolve_pending_cache() + + call_msg_count = len(messages) try: ct_kwargs: dict[str, Any] = dict( messages=messages, @@ -752,15 +884,9 @@ def _log_completion(answer: str, iterations: int, forced: bool = False): scope="reflect_tool_call", tool_choice=iter_tool_choice, ) - # Gemini rejects ``cached_content`` alongside a per-request - # ``tool_config`` (forced tool choice): "CachedContent can not be used - # with GenerateContent request setting system_instruction, tools or - # tool_config." The forced-sequence iterations set tool_config, so only - # the ``auto`` iterations can reference the cache; forced iterations send - # the prefix inline. The cache (tools + system prompt) is identical - # either way, so this just limits *which* iterations are billed cached. - if cached_prefix_name is not None and iter_tool_choice == "auto": - ct_kwargs["cached_prefix"] = cached_prefix_name + if incremental_caching and iter_tool_choice == "auto" and rolling_cache_name is not None: + ct_kwargs["cached_prefix"] = rolling_cache_name + ct_kwargs["cached_prefix_message_count"] = rolling_cache_boundary result = await llm_config.call_with_tools(**ct_kwargs) llm_duration = int((time.time() - llm_start) * 1000) consecutive_errors = 0 @@ -1078,6 +1204,16 @@ def _log_completion(answer: str, iterations: int, forced: bool = False): other_tools = allowed_tools + # Kick off the next-turn cache (covering THIS call's input) so it + # builds concurrently with the tool execution below — hiding the + # create latency. Only schedule when the next turn is ``auto`` (the + # only kind that references it); the next turn's pre-call resolve then + # adopts it. Resolve any prior in-flight create first so we don't drop + # its handle. + if incremental_caching and next_is_auto: + await _resolve_pending_cache() + _schedule_cache(call_msg_count) + # Execute tools in parallel tool_tasks = [ _execute_tool_with_timing( diff --git a/hindsight-api-slim/tests/test_gemini_cache.py b/hindsight-api-slim/tests/test_gemini_cache.py index 2a070c17c0..d803d6f21d 100644 --- a/hindsight-api-slim/tests/test_gemini_cache.py +++ b/hindsight-api-slim/tests/test_gemini_cache.py @@ -444,3 +444,72 @@ async def fake_create(*, model, config): cfg = captured["config_dict"] assert "tools" in cfg, f"tools should be in cache config; got keys: {list(cfg.keys())}" assert cfg["tools"], "tools list should be non-empty" + + +# ---- Step-by-step incremental caches (reflect) --------------------------- + + +def _make_client_with_delete(create_side_effect=None): + client, create_mock = _make_client(create_side_effect) + client.aio.caches.delete = AsyncMock() + return client, create_mock, client.aio.caches.delete + + +@pytest.mark.asyncio +async def test_create_incremental_tracks_names_per_session(): + """Each incremental create is tracked under its session id (no fingerprint + dedup — every step is a distinct, single-use cache).""" + names = iter(["cachedContents/c1", "cachedContents/c2"]) + client, create_mock = _make_client(lambda *a, **k: SimpleNamespace(name=next(names))) + mgr = GeminiCacheManager(client) + + n1 = await mgr.create_incremental(session_id="reflect:s1", model="m", system_instruction="SYS", contents=["turn1"]) + n2 = await mgr.create_incremental( + session_id="reflect:s1", model="m", system_instruction="SYS", contents=["turn1", "turn2"] + ) + assert (n1, n2) == ("cachedContents/c1", "cachedContents/c2") + assert mgr._sessions["reflect:s1"] == ["cachedContents/c1", "cachedContents/c2"] + # contents were forwarded to the SDK create (the growing prefix is cached, not just system+tools). + assert create_mock.await_args.kwargs["config"].contents == ["turn1", "turn2"] + + +@pytest.mark.asyncio +async def test_delete_session_tears_down_all_caches(): + client, _create, delete_mock = _make_client_with_delete( + lambda *a, **k: SimpleNamespace(name=f"cachedContents/{k['config'].contents[-1]}") + ) + mgr = GeminiCacheManager(client) + await mgr.create_incremental(session_id="reflect:s1", model="m", system_instruction="S", contents=["a"]) + await mgr.create_incremental(session_id="reflect:s1", model="m", system_instruction="S", contents=["a", "b"]) + + await mgr.delete_session("reflect:s1") + + deleted = {c.kwargs["name"] for c in delete_mock.await_args_list} + assert deleted == {"cachedContents/a", "cachedContents/b"} + assert "reflect:s1" not in mgr._sessions # session cleared + + +@pytest.mark.asyncio +async def test_create_incremental_soft_fallback_not_tracked(): + """A too-small prefix (Gemini 'minimum token count' 400) yields None and is + not tracked, so delete_session has nothing dangling to remove.""" + client, _create, delete_mock = _make_client_with_delete( + ValueError("Cached content is too small: minimum token count is 1024") + ) + mgr = GeminiCacheManager(client) + name = await mgr.create_incremental(session_id="reflect:s1", model="m", system_instruction="S", contents=["x"]) + assert name is None + assert "reflect:s1" not in mgr._sessions + await mgr.delete_session("reflect:s1") # no-op, must not raise + delete_mock.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_delete_session_swallows_delete_errors(): + """A failed server-side delete must never propagate (caches age out on TTL).""" + client, _create, delete_mock = _make_client_with_delete(lambda *a, **k: SimpleNamespace(name="cachedContents/c1")) + delete_mock.side_effect = RuntimeError("boom") + mgr = GeminiCacheManager(client) + await mgr.create_incremental(session_id="reflect:s1", model="m", system_instruction="S", contents=["a"]) + await mgr.delete_session("reflect:s1") # must not raise + assert "reflect:s1" not in mgr._sessions diff --git a/hindsight-api-slim/tests/test_gemini_implicit_cache_ratio.py b/hindsight-api-slim/tests/test_gemini_implicit_cache_ratio.py index 9de2c6ccf1..8cd8bbab35 100644 --- a/hindsight-api-slim/tests/test_gemini_implicit_cache_ratio.py +++ b/hindsight-api-slim/tests/test_gemini_implicit_cache_ratio.py @@ -190,11 +190,15 @@ def _assert(self, scope: str, rows: list[LLMRequestEntry], *, min_calls: int, mi ``min_ratio`` is per-operation because the achievable ratio differs by design: retain re-sends a pure fixed prefix (~90%); consolidation's prefix - is fixed but the facts/observations payload is large (~30%); reflect can - only cache its ``auto`` iterations — Gemini forbids ``cached_content`` with - a per-request ``tool_config`` — and the tool-result context grows, so the - ratio is modest (~10%). The universal guarantee in explicit mode is simply - that caching engaged at all (cached_tokens > 0). + is fixed but the facts/observations payload is large (~30%); reflect rolls + a step-by-step cache forward — each ``auto`` turn reuses the previous + turn's FULL input (system + tools + all prior tool results) at the cached + rate and re-sends only its own new results, so each retrieved payload is + billed fresh exactly once. The forced retrieval turns still can't cache + (Gemini forbids ``cached_content`` with a per-request ``tool_config``), so + on a short reflect the ratio is bounded by the forced:auto mix (~25-30%) + and climbs on longer loops. The universal guarantee in explicit mode is + that caching engaged (cached_tokens > 0). """ ratio = _report(scope, rows) cached_total = sum((r.cached_tokens or 0) for r in rows) @@ -229,8 +233,10 @@ async def test_retain_chunks_cached_ratio(self, memory_no_llm_verify, request_co await mem.delete_bank(bank_id, request_context=request_context) async def test_reflect_tool_loop_cached_ratio(self, memory_no_llm_verify, request_context): - """Reflect runs an agentic tool loop; the system_prompt + tools prefix is - cached once and reused across every iteration (scope ``reflect_tool_call``).""" + """Reflect runs an agentic tool loop with a step-by-step rolling cache: each + ``auto`` turn reuses the previous turn's full input (system + tools + all + prior tool results) and sends only its own new results (scope + ``reflect_tool_call``). The per-reflect caches are deleted when it ends.""" mem = await _gemini_engine(memory_no_llm_verify) bank_id = f"gemini-cache-reflect-{uuid.uuid4().hex[:8]}" await mem.get_bank_profile(bank_id, request_context=request_context) @@ -261,7 +267,11 @@ async def test_reflect_tool_loop_cached_ratio(self, memory_no_llm_verify, reques for r in allresp.items if allresp else []: print(f"\n[gemini-cache] reflect_tool_call status={r.status} error={r.error}") pytest.skip("reflect made no SUCCESSFUL reflect_tool_call iterations for this seed") - self._assert("reflect_tool_call", rows, min_calls=1, min_ratio=0.0) + # Floor guards the step-by-step cache: the old static system+tools prefix + # scored ~9% here; rolling the whole conversation forward lifts it well past + # that. Modest floor because the exact ratio depends on how many auto turns + # the seed drives (more auto turns → higher). + self._assert("reflect_tool_call", rows, min_calls=1, min_ratio=0.15) await mem.delete_bank(bank_id, request_context=request_context) diff --git a/hindsight-api-slim/tests/test_reflect_agent.py b/hindsight-api-slim/tests/test_reflect_agent.py index 63d49a2586..24263a30a4 100644 --- a/hindsight-api-slim/tests/test_reflect_agent.py +++ b/hindsight-api-slim/tests/test_reflect_agent.py @@ -1263,3 +1263,148 @@ async def test_real_stale_mental_model_forces_and_grounds_in_deeper_evidence(sel context="A stale mental model claimed the launch was still pending, but the freshly retrieved raw " "fact (deploy log A-1029) shows it shipped on Friday. The agent should correct the stale summary.", ) + + +class _StepCacheProvider: + """Fake provider that records the step-by-step incremental-cache protocol. + + Serves as both ``llm_config`` and its own ``_provider_impl``: the reflect loop + reaches the cache methods via ``llm_config._provider_impl`` and issues LLM + turns via ``llm_config.call_with_tools``. Every ``call_with_tools`` records the + ``cached_prefix`` / ``cached_prefix_message_count`` it was handed, so a test + can assert that each ``auto`` turn reuses exactly the previous turn's full + input and that the caches are torn down at the end. + """ + + def __init__(self, scripted: list[LLMToolCallResult]): + self._scripted = scripted + self._i = 0 + self._provider_impl = self + self.cache_counter = 0 + self.created: list[tuple[str, int]] = [] # (session_id, #messages covered) + self.deleted_sessions: list[str] = [] + self.calls: list[dict] = [] # per call_with_tools: tool_choice / cached_prefix / count / #messages + + # -- incremental cache capability -- + def supports_incremental_prompt_cache(self) -> bool: + return True + + async def create_incremental_cache(self, *, session_id, messages, tools=None): + self.cache_counter += 1 + self.created.append((session_id, len(messages))) + return f"cache-{self.cache_counter}" + + async def delete_cached_prefix(self, name): # pragma: no cover - not exercised here + pass + + async def delete_cache_session(self, session_id): + self.deleted_sessions.append(session_id) + + # -- llm surface -- + async def call_with_tools( + self, + *, + messages, + tools, + scope="tools", + tool_choice="auto", + cached_prefix=None, + cached_prefix_message_count=0, + **_, + ): + self.calls.append( + { + "tool_choice": tool_choice, + "cached_prefix": cached_prefix, + "cached_prefix_message_count": cached_prefix_message_count, + "n_messages": len(messages), + } + ) + res = self._scripted[self._i] + self._i += 1 + return res + + async def call(self, *args, **kwargs): # final-synthesis fallback (unused on the happy path) + return ("final", TokenUsage(input_tokens=1, output_tokens=1, total_tokens=2)) + + +class TestReflectIncrementalCache: + """The step-by-step Gemini context cache: each auto turn reuses the previous + turn's full input, and every per-reflect cache is deleted at the end.""" + + @pytest.mark.asyncio + async def test_each_auto_turn_reuses_previous_step_cache_and_cleans_up(self): + functions = { + "search_observations_fn": AsyncMock(return_value={"observations": [{"id": "obs-1"}]}), + "recall_fn": AsyncMock(return_value={"memories": [{"id": "mem-1", "content": "x"}]}), + "search_mental_models_fn": AsyncMock(return_value={"mental_models": []}), + "expand_fn": AsyncMock(return_value={"memories": []}), + } + + def _tc(cid, name): + # A real query arg so the stubbed tools return evidence (not an + # error), which lets the terminal ``done`` call be accepted. + return LLMToolCallResult( + tool_calls=[LLMToolCall(id=cid, name=name, arguments={"query": "q"})], finish_reason="tool_calls" + ) + + # Forced obs -> forced recall -> two auto recalls -> done. + provider = _StepCacheProvider( + scripted=[ + _tc("0", "search_observations"), + _tc("1", "recall"), + _tc("2", "recall"), + _tc("3", "recall"), + LLMToolCallResult( + tool_calls=[LLMToolCall(id="4", name="done", arguments={"answer": "A", "memory_ids": ["mem-1"]})], + finish_reason="tool_calls", + ), + ] + ) + + result = await run_reflect_agent( + llm_config=provider, + bank_id="cache-bank", + query="q", + bank_profile={"name": "T", "mission": "M"}, + has_mental_models=False, + include_observations=True, + include_recall=True, + budget="high", # keep the full forced path (no early release) so counts are deterministic + max_iterations=8, + **functions, + ) + assert result.text == "A" + + calls = provider.calls + # 2 forced turns (obs, recall) then 3 auto turns (recall, recall, done). + assert [c["tool_choice"] for c in calls[:2]] == [ + {"type": "function", "function": {"name": "search_observations"}}, + {"type": "function", "function": {"name": "recall"}}, + ] + assert all(c["tool_choice"] == "auto" for c in calls[2:]) + + # Forced turns never reference a cache (Gemini forbids cache + tool_config). + assert calls[0]["cached_prefix"] is None + assert calls[1]["cached_prefix"] is None + + # Every auto turn references a cache, and the count it was handed equals the + # PREVIOUS turn's full input length — i.e. it reuses the previous step entirely. + for i in range(2, len(calls)): + assert calls[i]["cached_prefix"] is not None, f"auto call {i} should use the cache" + assert calls[i]["cached_prefix_message_count"] == calls[i - 1]["n_messages"], ( + f"auto call {i} must cache exactly the previous step's input" + ) + + # Caches are created covering a strictly growing prefix, one per auto turn. + assert [n for _, n in provider.created] == [ + calls[1]["n_messages"], + calls[2]["n_messages"], + calls[3]["n_messages"], + ] + assert all(sid.startswith("reflect:") for sid, _ in provider.created) + + # Ephemeral: the session is torn down exactly once when the reflect ends. + assert len(provider.deleted_sessions) == 1 + assert provider.deleted_sessions[0].startswith("reflect:") + assert provider.deleted_sessions[0] == provider.created[0][0] diff --git a/hindsight-docs/docs/developer/configuration.md b/hindsight-docs/docs/developer/configuration.md index 52cb8e4bbc..f1cdfcfa2c 100644 --- a/hindsight-docs/docs/developer/configuration.md +++ b/hindsight-docs/docs/developer/configuration.md @@ -189,6 +189,7 @@ For non-English banks (especially CJK) and the language/extraction-language trad | `HINDSIGHT_API_LLM_STRICT_SCHEMA` | Grammar-enforce structured output via `json_schema` `strict: true` instead of the soft "schema-in-prompt + `json_object`" path. Use it with weaker self-hosted models that return prose preambles, markdown ` ```json ` fences, or invalid JSON — which otherwise fail to parse and wedge retain/consolidation. Applies to OpenAI-compatible backends (OpenAI, llama.cpp, vLLM) and LiteLLM; Gemini already enforces its native `response_schema` regardless, and providers without a strict mode ignore it. | `false` | | `HINDSIGHT_API_LLM_GEMINI_SAFETY_SETTINGS` | JSON-encoded list of `{category, threshold}` dicts for Gemini/VertexAI content safety filtering | `null` | | `HINDSIGHT_API_LLM_PROMPT_CACHE_ENABLED` | Reuse the fixed system prefix via the provider's explicit prompt cache, billed at the cached-input rate (Gemini/Vertex `CachedContent`). The cached prefix is shared across all banks and soft-fails to an uncached call. Set to `false` to disable. See [Models](./models#provider-capabilities). | `true` | +| `HINDSIGHT_API_REFLECT_PROMPT_CACHE_ENABLED` | For reflect specifically, roll a step-by-step context cache forward through the agent's tool loop so each turn reuses the whole prior conversation (system + tools + all prior tool results) at the cached-input rate instead of only the static prefix. Requires `HINDSIGHT_API_LLM_PROMPT_CACHE_ENABLED`. The per-reflect caches are ephemeral and deleted when the reflect ends. Set to `false` to run reflect uncached while leaving prompt caching on elsewhere. | `true` | **Provider Examples** diff --git a/skills/hindsight-docs/references/developer/configuration.md b/skills/hindsight-docs/references/developer/configuration.md index 42240315a0..23c46fd750 100644 --- a/skills/hindsight-docs/references/developer/configuration.md +++ b/skills/hindsight-docs/references/developer/configuration.md @@ -189,6 +189,7 @@ For non-English banks (especially CJK) and the language/extraction-language trad | `HINDSIGHT_API_LLM_STRICT_SCHEMA` | Grammar-enforce structured output via `json_schema` `strict: true` instead of the soft "schema-in-prompt + `json_object`" path. Use it with weaker self-hosted models that return prose preambles, markdown ` ```json ` fences, or invalid JSON — which otherwise fail to parse and wedge retain/consolidation. Applies to OpenAI-compatible backends (OpenAI, llama.cpp, vLLM) and LiteLLM; Gemini already enforces its native `response_schema` regardless, and providers without a strict mode ignore it. | `false` | | `HINDSIGHT_API_LLM_GEMINI_SAFETY_SETTINGS` | JSON-encoded list of `{category, threshold}` dicts for Gemini/VertexAI content safety filtering | `null` | | `HINDSIGHT_API_LLM_PROMPT_CACHE_ENABLED` | Reuse the fixed system prefix via the provider's explicit prompt cache, billed at the cached-input rate (Gemini/Vertex `CachedContent`). The cached prefix is shared across all banks and soft-fails to an uncached call. Set to `false` to disable. See [Models](./models#provider-capabilities). | `true` | +| `HINDSIGHT_API_REFLECT_PROMPT_CACHE_ENABLED` | For reflect specifically, roll a step-by-step context cache forward through the agent's tool loop so each turn reuses the whole prior conversation (system + tools + all prior tool results) at the cached-input rate instead of only the static prefix. Requires `HINDSIGHT_API_LLM_PROMPT_CACHE_ENABLED`. The per-reflect caches are ephemeral and deleted when the reflect ends. Set to `false` to run reflect uncached while leaving prompt caching on elsewhere. | `true` | **Provider Examples**