perf(sdk): cache filesystem system prompts (#3889)

Adds per-instance caching for the dynamic filesystem system prompt keyed
by whether execution instructions are included.

The prompt only depends on immutable middleware configuration and the
execute-tool support flag, so this avoids rebuilding identical prompt
text before every model call.
This commit is contained in:
Mason Daugherty
2026-06-11 19:06:16 -04:00
committed by GitHub
parent 85fd7c283d
commit 2f432ba636
2 changed files with 80 additions and 24 deletions
@@ -774,6 +774,11 @@ class FilesystemMiddleware(AgentMiddleware[FilesystemState, ContextT, ResponseT]
self._large_tool_results_prefix = f"{_root}/large_tool_results"
self._conversation_history_prefix = f"{_root}/conversation_history"
# Cache for dynamic system prompts keyed on the `include_execution`
# flag. The text depends only on that flag and immutable config, so it
# is computed at most twice per instance.
self._dynamic_system_prompt_cache: dict[bool, str] = {}
# Store configuration (private - internal implementation details)
self._custom_system_prompt = system_prompt
self._custom_tool_descriptions = custom_tool_descriptions or {}
@@ -801,6 +806,25 @@ class FilesystemMiddleware(AgentMiddleware[FilesystemState, ContextT, ResponseT]
self._create_execute_tool(),
]
def _build_dynamic_system_prompt(self, *, include_execution: bool) -> str:
"""Build (and memoize) the dynamic system prompt.
The result depends only on `include_execution` and immutable config,
so it is cached per instance to avoid rebuilding on every model call.
The cache is intentionally lock-free even though sync and async model
calls share it: writes are idempotent (a given flag always yields the
same string), so a race at worst recomputes and re-stores that value.
"""
cached = self._dynamic_system_prompt_cache.get(include_execution)
if cached is not None:
return cached
prompt_parts = [_FILESYSTEM_SYSTEM_PROMPT_TEMPLATE.format(large_tool_results_prefix=self._large_tool_results_prefix)]
if include_execution:
prompt_parts.append(EXECUTION_SYSTEM_PROMPT)
system_prompt = "\n\n".join(prompt_parts).strip()
self._dynamic_system_prompt_cache[include_execution] = system_prompt
return system_prompt
def _get_backend(self, runtime: ToolRuntime[Any, Any]) -> BackendProtocol:
"""Get the resolved backend instance from backend or factory.
@@ -1776,18 +1800,9 @@ class FilesystemMiddleware(AgentMiddleware[FilesystemState, ContextT, ResponseT]
if self._custom_system_prompt is not None:
system_prompt = self._custom_system_prompt
else:
# Build dynamic system prompt based on available tools
prompt_parts = [
_FILESYSTEM_SYSTEM_PROMPT_TEMPLATE.format(
large_tool_results_prefix=self._large_tool_results_prefix,
)
]
# Add execution instructions if execute tool is available
if has_execute_tool and backend_supports_execution:
prompt_parts.append(EXECUTION_SYSTEM_PROMPT)
system_prompt = "\n\n".join(prompt_parts).strip()
system_prompt = self._build_dynamic_system_prompt(
include_execution=has_execute_tool and backend_supports_execution,
)
if system_prompt:
new_system_message = append_to_system_message(request.system_message, system_prompt)
@@ -1841,18 +1856,9 @@ class FilesystemMiddleware(AgentMiddleware[FilesystemState, ContextT, ResponseT]
if self._custom_system_prompt is not None:
system_prompt = self._custom_system_prompt
else:
# Build dynamic system prompt based on available tools
prompt_parts = [
_FILESYSTEM_SYSTEM_PROMPT_TEMPLATE.format(
large_tool_results_prefix=self._large_tool_results_prefix,
)
]
# Add execution instructions if execute tool is available
if has_execute_tool and backend_supports_execution:
prompt_parts.append(EXECUTION_SYSTEM_PROMPT)
system_prompt = "\n\n".join(prompt_parts).strip()
system_prompt = self._build_dynamic_system_prompt(
include_execution=has_execute_tool and backend_supports_execution,
)
if system_prompt:
new_system_message = append_to_system_message(request.system_message, system_prompt)
@@ -3,20 +3,70 @@
from typing import Any
from langchain.agents import create_agent
from langchain.agents.middleware.types import ModelRequest, ModelResponse
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.store.memory import InMemoryStore
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
from deepagents.middleware.filesystem import (
EXECUTION_SYSTEM_PROMPT,
WRITE_FILE_TOOL_DESCRIPTION,
FilesystemMiddleware,
)
from tests.unit_tests.chat_model import GenericFakeChatModel
def build_composite_state_backend(*, routes: dict[str, Any]) -> CompositeBackend:
return CompositeBackend(default=StateBackend(), routes=routes)
class TestDynamicSystemPromptCache:
"""`_build_dynamic_system_prompt` caches per `include_execution` flag."""
def test_returns_identical_cached_object(self) -> None:
mw = FilesystemMiddleware(backend=StateBackend())
first = mw._build_dynamic_system_prompt(include_execution=False)
second = mw._build_dynamic_system_prompt(include_execution=False)
assert first is second
def test_execution_flag_changes_output(self) -> None:
mw = FilesystemMiddleware(backend=StateBackend())
without = mw._build_dynamic_system_prompt(include_execution=False)
with_exec = mw._build_dynamic_system_prompt(include_execution=True)
assert without != with_exec
assert EXECUTION_SYSTEM_PROMPT not in without
assert EXECUTION_SYSTEM_PROMPT in with_exec
async def test_awrap_model_call_emits_dynamic_prompt(self) -> None:
"""`awrap_model_call` appends the same memoized prompt as the sync path.
The cache call site is duplicated across `wrap_model_call` and
`awrap_model_call`; this guards the async path against drift.
"""
mw = FilesystemMiddleware(backend=StateBackend())
# StateBackend has no execution support, so the execute tool (if any)
# is filtered out and `include_execution` resolves to False.
expected = mw._build_dynamic_system_prompt(include_execution=False)
captured: list[ModelRequest] = []
async def handler(request: ModelRequest) -> ModelResponse:
captured.append(request)
return ModelResponse(result=[AIMessage(content="ok")])
request = ModelRequest(
model=GenericFakeChatModel(messages=iter([AIMessage(content="ok")])),
messages=[HumanMessage(content="hi")],
tools=[],
)
await mw.awrap_model_call(request, handler)
assert len(captured) == 1
assert captured[0].system_prompt == expected
class TestFilesystemMiddlewareInit:
"""Tests for FilesystemMiddleware initialization that don't require LLM invocation."""