perf(llm): give Claude a cache point before the per-run scope

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ian 2026-09-28 23:43:50 -04:00
parent ab68249268
commit 200e37727e
3 changed files with 61 additions and 0 deletions

View file

@ -1,3 +1,5 @@
{% if system_prompt_context and (system_prompt_context.authorized_targets or system_prompt_context.mcp_available) %}
<run_scope>
{% if system_prompt_context and system_prompt_context.authorized_targets %}
SYSTEM-VERIFIED SCOPE:
- The following scope metadata is injected by the platform into the system prompt and is authoritative
@ -32,3 +34,5 @@ MCP CONNECTIONS (available this run):
- Do not assume a connection or tool exists; discover it with list_mcps and search_mcp_tools before calling.
- Use describe_mcp only as a compatibility fallback when targeted search cannot identify an expected tool. Its full catalog can be large.
{% endif %}
</run_scope>
{% endif %}

View file

@ -306,6 +306,9 @@ class _TurnGuardModel(Model):
) -> ModelResponse:
sanitized = dedupe_input(input)
rewriter = TurnCallIdRewriter(sanitized)
system_instructions, sanitized = _split_cached_prefix(
system_instructions, sanitized, model_settings
)
response = await self._inner.get_response(
system_instructions,
cast("str | list[TResponseInputItem]", sanitized),
@ -339,6 +342,9 @@ class _TurnGuardModel(Model):
) -> AsyncIterator[TResponseStreamEvent]:
sanitized = dedupe_input(input)
rewriter = TurnCallIdRewriter(sanitized)
system_instructions, sanitized = _split_cached_prefix(
system_instructions, sanitized, model_settings
)
limiter = self._limiter()
stream = self._inner.stream_response(
system_instructions,
@ -359,6 +365,31 @@ class _TurnGuardModel(Model):
self._log_dropped(limiter)
def _split_cached_prefix(
system_instructions: str | None,
model_input: str | list[Any],
model_settings: ModelSettings,
) -> tuple[str | None, str | list[Any]]:
"""Send the per-run scope as its own system message on cache-point routes.
LiteLLM puts a cache point at the end of each system message, so the
shared prompt before ``<run_scope>`` gets its own and is reused across runs.
"""
extra_args = model_settings.extra_args or {}
if not system_instructions or "cache_control_injection_points" not in extra_args:
return system_instructions, model_input
shared, tag, scope = system_instructions.partition("<run_scope>")
if not tag:
return system_instructions, model_input
if isinstance(model_input, str):
model_input = [{"role": "user", "content": model_input}]
return None, [
{"role": "system", "content": shared},
{"role": "system", "content": tag + scope},
*model_input,
]
async def _aclose(stream: AsyncIterator[TResponseStreamEvent]) -> None:
if isinstance(stream, AsyncGenerator):
with contextlib.suppress(Exception):

View file

@ -18,8 +18,10 @@ import strix.tools.mcp as mcp_pkg
import strix.tools.notes.tools as notes_tools
import strix.tools.todo.tools as todo_tools
from strix.agents.prompt import render_system_prompt
from strix.config.models import _split_cached_prefix
from strix.core import runner
from strix.core.agents import AgentCoordinator
from strix.core.inputs import make_model_settings
from strix.runtime import session_manager
from strix.tools.mcp import BearerAuth, McpConnectionConfig, McpConnectionRequest
@ -275,3 +277,27 @@ def test_scope_is_rendered_once_at_the_end_of_the_prompt() -> None:
assert prompt.count("SYSTEM-VERIFIED SCOPE") == 1
assert prompt.index("</available_skills>") < prompt.index("SYSTEM-VERIFIED SCOPE")
def test_scope_is_sent_as_its_own_system_message_on_cache_point_routes() -> None:
settings = make_model_settings(None, model_name="anthropic/claude-sonnet-5-5")
prompt = render_system_prompt(
system_prompt_context={
"authorized_targets": [{"type": "web_application", "value": "https://target.invalid"}],
},
)
system, model_input = _split_cached_prefix(prompt, "go", settings)
assert system is None
assert isinstance(model_input, list)
assert [item["role"] for item in model_input] == ["system", "system", "user"]
assert "https://target.invalid" not in model_input[0]["content"]
assert model_input[1]["content"].startswith("<run_scope>")
def test_system_prompt_is_unchanged_without_cache_points() -> None:
settings = make_model_settings(None, model_name="openai/gpt-5")
prompt = "shared\n<run_scope>\ntargets\n</run_scope>"
assert _split_cached_prefix(prompt, "go", settings) == (prompt, "go")