feat(langchain): dynamic system prompt middleware (#33006)

# Changes

## Adds support for `DynamicSystemPromptMiddleware`

```py
from langchain.agents.middleware import DynamicSystemPromptMiddleware
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

class Context(TypedDict):
    user_name: str

def system_prompt(state: AgentState, runtime: Runtime[Context]) -> str:
    user_name = runtime.context.get("user_name", "n/a")
    return f"You are a helpful assistant. Always address the user by their name: {user_name}"

middleware = DynamicSystemPromptMiddleware(system_prompt)
```

## Adds support for `runtime` in middleware hooks

```py
class AgentMiddleware(Generic[StateT, ContextT]):
    def modify_model_request(
        self,
        request: ModelRequest,
        state: StateT,
        runtime: Runtime[ContextT],  # Optional runtime parameter
    ) -> ModelRequest:
        # upgrade model if runtime.context.subscription is `top-tier` or whatever
```

## Adds support for omitting state attributes from input / output
schemas

```py
from typing import Annotated, NotRequired
from langchain.agents.middleware.types import PrivateStateAttr, OmitFromInput, OmitFromOutput

class CustomState(AgentState):
    # Private field - not in input or output schemas
    internal_counter: NotRequired[Annotated[int, PrivateStateAttr]]
    
    # Input-only field - not in output schema
    user_input: NotRequired[Annotated[str, OmitFromOutput]]
    
    # Output-only field - not in input schema  
    computed_result: NotRequired[Annotated[str, OmitFromInput]]
```

## Additionally
* Removes filtering of state before passing into middleware hooks

Typing is not foolproof here, still need to figure out some of the
generics stuff w/ state and context schema extensions for middleware.

TODO:
* More docs for middleware, should hold off on this until other prios
like MCP and deepagents are met

---------

Co-authored-by: Eugene Yurtsev <[email protected]>

Author: sydney-runkle

libs/langchain_v1/langchain/agents/middleware/__init__.py

@@ -1,5 +1,6 @@
 """Middleware plugins for agents."""
 
+from .dynamic_system_prompt import DynamicSystemPromptMiddleware
 from .human_in_the_loop import HumanInTheLoopMiddleware
 from .prompt_caching import AnthropicPromptCachingMiddleware
 from .summarization import SummarizationMiddleware
@@ -8,7 +9,9 @@
 __all__ = [
     "AgentMiddleware",
     "AgentState",
+    # should move to langchain-anthropic if we decide to keep it
     "AnthropicPromptCachingMiddleware",
+    "DynamicSystemPromptMiddleware",
     "HumanInTheLoopMiddleware",
     "ModelRequest",
     "SummarizationMiddleware",

libs/langchain_v1/langchain/agents/middleware/dynamic_system_prompt.py

@@ -0,0 +1,105 @@
+"""Dynamic System Prompt Middleware.
+
+Allows setting the system prompt dynamically right before each model invocation.
+Useful when the prompt depends on the current agent state or per-invocation context.
+"""
+
+from __future__ import annotations
+
+from inspect import signature
+from typing import TYPE_CHECKING, Protocol, TypeAlias, cast
+
+from langgraph.typing import ContextT
+
+from langchain.agents.middleware.types import (
+    AgentMiddleware,
+    AgentState,
+    ModelRequest,
+)
+
+if TYPE_CHECKING:
+    from langgraph.runtime import Runtime
+
+
+class DynamicSystemPromptWithoutRuntime(Protocol):
+    """Dynamic system prompt without runtime in call signature."""
+
+    def __call__(self, state: AgentState) -> str:
+        """Return the system prompt for the next model call."""
+        ...
+
+
+class DynamicSystemPromptWithRuntime(Protocol[ContextT]):
+    """Dynamic system prompt with runtime in call signature."""
+
+    def __call__(self, state: AgentState, runtime: Runtime[ContextT]) -> str:
+        """Return the system prompt for the next model call."""
+        ...
+
+
+DynamicSystemPrompt: TypeAlias = (
+    DynamicSystemPromptWithoutRuntime | DynamicSystemPromptWithRuntime[ContextT]
+)
+
+
+class DynamicSystemPromptMiddleware(AgentMiddleware):
+    """Dynamic System Prompt Middleware.
+
+    Allows setting the system prompt dynamically right before each model invocation.
+    Useful when the prompt depends on the current agent state or per-invocation context.
+
+    Example:
+        ```python
+        from langchain.agents.middleware import DynamicSystemPromptMiddleware
+
+
+        class Context(TypedDict):
+            user_name: str
+
+
+        def system_prompt(state: AgentState, runtime: Runtime[Context]) -> str:
+            user_name = runtime.context.get("user_name", "n/a")
+            return (
+                f"You are a helpful assistant. Always address the user by their name: {user_name}"
+            )
+
+
+        middleware = DynamicSystemPromptMiddleware(system_prompt)
+        ```
+    """
+
+    _accepts_runtime: bool
+
+    def __init__(
+        self,
+        dynamic_system_prompt: DynamicSystemPrompt[ContextT],
+    ) -> None:
+        """Initialize the dynamic system prompt middleware.
+
+        Args:
+            dynamic_system_prompt: Function that receives the current agent state
+                and optionally runtime with context, and returns the system prompt for
+                the next model call. Returns a string.
+        """
+        super().__init__()
+        self.dynamic_system_prompt = dynamic_system_prompt
+        self._accepts_runtime = "runtime" in signature(dynamic_system_prompt).parameters
+
+    def modify_model_request(
+        self,
+        request: ModelRequest,
+        state: AgentState,
+        runtime: Runtime[ContextT],
+    ) -> ModelRequest:
+        """Modify the model request to include the dynamic system prompt."""
+        if self._accepts_runtime:
+            system_prompt = cast(
+                "DynamicSystemPromptWithRuntime[ContextT]", self.dynamic_system_prompt
+            )(state, runtime)
+        else:
+            system_prompt = cast("DynamicSystemPromptWithoutRuntime", self.dynamic_system_prompt)(
+                state
+            )
+
+        request.system_prompt = system_prompt
+        return request

libs/langchain_v1/langchain/agents/middleware/human_in_the_loop.py

@@ -143,7 +143,7 @@ def __init__(
         self.tool_configs = resolved_tool_configs
         self.description_prefix = description_prefix
 
-    def after_model(self, state: AgentState) -> dict[str, Any] | None:
+    def after_model(self, state: AgentState) -> dict[str, Any] | None:  # type: ignore[override]
         """Trigger HITL flows for relevant tool calls after an AIMessage."""
         messages = state["messages"]
         if not messages:

libs/langchain_v1/langchain/agents/middleware/prompt_caching.py

@@ -2,7 +2,7 @@
 
 from typing import Literal
 
-from langchain.agents.middleware.types import AgentMiddleware, AgentState, ModelRequest
+from langchain.agents.middleware.types import AgentMiddleware, ModelRequest
 
 
 class AnthropicPromptCachingMiddleware(AgentMiddleware):
@@ -32,7 +32,10 @@ def __init__(
         self.ttl = ttl
         self.min_messages_to_cache = min_messages_to_cache
 
-    def modify_model_request(self, request: ModelRequest, state: AgentState) -> ModelRequest:  # noqa: ARG002
+    def modify_model_request(  # type: ignore[override]
+        self,
+        request: ModelRequest,
+    ) -> ModelRequest:
         """Modify the model request to add cache control blocks."""
         try:
             from langchain_anthropic import ChatAnthropic

libs/langchain_v1/langchain/agents/middleware/summarization.py

@@ -98,7 +98,7 @@ def __init__(
         self.summary_prompt = summary_prompt
         self.summary_prefix = summary_prefix
 
-    def before_model(self, state: AgentState) -> dict[str, Any] | None:
+    def before_model(self, state: AgentState) -> dict[str, Any] | None:  # type: ignore[override]
         """Process messages before model invocation, potentially triggering summarization."""
         messages = state["messages"]
         self._ensure_message_ids(messages)

libs/langchain_v1/langchain/agents/middleware/types.py

@@ -8,15 +8,27 @@
 # needed as top level import for pydantic schema generation on AgentState
 from langchain_core.messages import AnyMessage  # noqa: TC002
 from langgraph.channels.ephemeral_value import EphemeralValue
-from langgraph.graph.message import Messages, add_messages
+from langgraph.graph.message import add_messages
+from langgraph.runtime import Runtime
+from langgraph.typing import ContextT
 from typing_extensions import NotRequired, Required, TypedDict, TypeVar
 
 if TYPE_CHECKING:
     from langchain_core.language_models.chat_models import BaseChatModel
     from langchain_core.tools import BaseTool
+    from langgraph.runtime import Runtime
 
     from langchain.agents.structured_output import ResponseFormat
 
+__all__ = [
+    "AgentMiddleware",
+    "AgentState",
+    "ContextT",
+    "ModelRequest",
+    "OmitFromSchema",
+    "PublicAgentState",
+]
+
 JumpTo = Literal["tools", "model", "__end__"]
 """Destination to jump to when a middleware node returns."""
 
@@ -36,26 +48,49 @@ class ModelRequest:
     model_settings: dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass
+class OmitFromSchema:
+    """Annotation used to mark state attributes as omitted from input or output schemas."""
+
+    input: bool = True
+    """Whether to omit the attribute from the input schema."""
+
+    output: bool = True
+    """Whether to omit the attribute from the output schema."""
+
+
+OmitFromInput = OmitFromSchema(input=True, output=False)
+"""Annotation used to mark state attributes as omitted from input schema."""
+
+OmitFromOutput = OmitFromSchema(input=False, output=True)
+"""Annotation used to mark state attributes as omitted from output schema."""
+
+PrivateStateAttr = OmitFromSchema(input=True, output=True)
+"""Annotation used to mark state attributes as purely internal for a given middleware."""
+
+
 class AgentState(TypedDict, Generic[ResponseT]):
     """State schema for the agent."""
 
     messages: Required[Annotated[list[AnyMessage], add_messages]]
-    model_request: NotRequired[Annotated[ModelRequest | None, EphemeralValue]]
-    jump_to: NotRequired[Annotated[JumpTo | None, EphemeralValue]]
+    jump_to: NotRequired[Annotated[JumpTo | None, EphemeralValue, PrivateStateAttr]]
     response: NotRequired[ResponseT]
 
 
 class PublicAgentState(TypedDict, Generic[ResponseT]):
-    """Input / output schema for the agent."""
+    """Public state schema for the agent.
 
-    messages: Required[Messages]
+    Just used for typing purposes.
+    """
+
+    messages: Required[Annotated[list[AnyMessage], add_messages]]
     response: NotRequired[ResponseT]
 
 
-StateT = TypeVar("StateT", bound=AgentState)
+StateT = TypeVar("StateT", bound=AgentState, default=AgentState)
 
 
-class AgentMiddleware(Generic[StateT]):
+class AgentMiddleware(Generic[StateT, ContextT]):
     """Base middleware class for an agent.
 
     Subclass this and implement any of the defined methods to customize agent behavior
@@ -68,12 +103,17 @@ class AgentMiddleware(Generic[StateT]):
     tools: list[BaseTool]
     """Additional tools registered by the middleware."""
 
-    def before_model(self, state: StateT) -> dict[str, Any] | None:
+    def before_model(self, state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None:
         """Logic to run before the model is called."""
 
-    def modify_model_request(self, request: ModelRequest, state: StateT) -> ModelRequest:  # noqa: ARG002
+    def modify_model_request(
+        self,
+        request: ModelRequest,
+        state: StateT,  # noqa: ARG002
+        runtime: Runtime[ContextT],  # noqa: ARG002
+    ) -> ModelRequest:
         """Logic to modify request kwargs before the model is called."""
         return request
 
-    def after_model(self, state: StateT) -> dict[str, Any] | None:
+    def after_model(self, state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None:
         """Logic to run after the model is called."""