feat(langchain_v1): implement nicer devx for dynamic prompt (#33264)

Adding a `dynamic_prompt` decorator to support smoother devx for dynamic
system prompts

```py
from langchain.agents.middleware.types import dynamic_prompt, ModelRequest, AgentState
from langchain.agents.middleware_agent import create_agent
from langgraph.runtime import Runtime
from dataclasses import dataclass
from langchain_core.messages import HumanMessage


@DataClass
class Context:
    user_name: str


@dynamic_prompt
def my_prompt(request: ModelRequest, state: AgentState, runtime: Runtime[Context]) -> str:
    user_name = runtime.context.user_name
    return (
        f"You are a helpful assistant helping {user_name}. Please refer to the user as {user_name}."
    )


agent = create_agent(model="openai:gpt-4o", middleware=[my_prompt]).compile()

result = agent.invoke({"messages": [HumanMessage("Hello")]}, context=Context(user_name="Sydney"))
for msg in result["messages"]:
    msg.pretty_print()

"""
================================ Human Message =================================

Hello
================================== Ai Message ==================================

Hello Sydney! How can I assist you today?
"""

```

Author: sydney-runkle

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

@@ -10,6 +10,7 @@
     ModelRequest,
     after_model,
     before_model,
+    dynamic_prompt,
     hook_config,
     modify_model_request,
 )
@@ -25,6 +26,7 @@
     "SummarizationMiddleware",
     "after_model",
     "before_model",
+    "dynamic_prompt",
     "hook_config",
     "modify_model_request",
 ]

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

@@ -45,6 +45,7 @@
     "ModelRequest",
     "OmitFromSchema",
     "PublicAgentState",
+    "dynamic_prompt",
     "hook_config",
 ]
 
@@ -180,6 +181,16 @@ def __call__(
         ...
 
 
+class _CallableReturningPromptString(Protocol[StateT_contra, ContextT]):
+    """Callable that returns a prompt string given ModelRequest, AgentState, and Runtime."""
+
+    def __call__(
+        self, request: ModelRequest, state: StateT_contra, runtime: Runtime[ContextT]
+    ) -> str | Awaitable[str]:
+        """Generate a system prompt string based on the request, state, and runtime."""
+        ...
+
+
 CallableT = TypeVar("CallableT", bound=Callable[..., Any])
 
 
@@ -639,3 +650,126 @@ def wrapped(
     if func is not None:
         return decorator(func)
     return decorator
+
+
+@overload
+def dynamic_prompt(
+    func: _CallableReturningPromptString[StateT, ContextT],
+) -> AgentMiddleware[StateT, ContextT]: ...
+
+
+@overload
+def dynamic_prompt(
+    func: None = None,
+) -> Callable[
+    [_CallableReturningPromptString[StateT, ContextT]],
+    AgentMiddleware[StateT, ContextT],
+]: ...
+
+
+def dynamic_prompt(
+    func: _CallableReturningPromptString[StateT, ContextT] | None = None,
+) -> (
+    Callable[
+        [_CallableReturningPromptString[StateT, ContextT]],
+        AgentMiddleware[StateT, ContextT],
+    ]
+    | AgentMiddleware[StateT, ContextT]
+):
+    """Decorator used to dynamically generate system prompts for the model.
+
+    This is a convenience decorator that creates middleware using `modify_model_request`
+    specifically for dynamic prompt generation. The decorated function should return
+    a string that will be set as the system prompt for the model request.
+
+    Args:
+        func: The function to be decorated. Must accept:
+            `request: ModelRequest, state: StateT, runtime: Runtime[ContextT]` -
+            Model request, state, and runtime context
+
+    Returns:
+        Either an AgentMiddleware instance (if func is provided) or a decorator function
+        that can be applied to a function.
+
+    The decorated function should return:
+        - `str` - The system prompt to use for the model request
+
+    Examples:
+        Basic usage with dynamic content:
+        ```python
+        @dynamic_prompt
+        def my_prompt(request: ModelRequest, state: AgentState, runtime: Runtime) -> str:
+            user_name = runtime.context.get("user_name", "User")
+            return f"You are a helpful assistant helping {user_name}."
+        ```
+
+        Using state to customize the prompt:
+        ```python
+        @dynamic_prompt
+        def context_aware_prompt(request: ModelRequest, state: AgentState, runtime: Runtime) -> str:
+            msg_count = len(state["messages"])
+            if msg_count > 10:
+                return "You are in a long conversation. Be concise."
+            return "You are a helpful assistant."
+        ```
+
+        Using with agent:
+        ```python
+        agent = create_agent(model, middleware=[my_prompt])
+        ```
+    """
+
+    def decorator(
+        func: _CallableReturningPromptString[StateT, ContextT],
+    ) -> AgentMiddleware[StateT, ContextT]:
+        is_async = iscoroutinefunction(func)
+
+        if is_async:
+
+            async def async_wrapped(
+                self: AgentMiddleware[StateT, ContextT],  # noqa: ARG001
+                request: ModelRequest,
+                state: StateT,
+                runtime: Runtime[ContextT],
+            ) -> ModelRequest:
+                prompt = await func(request, state, runtime)  # type: ignore[misc]
+                request.system_prompt = prompt
+                return request
+
+            middleware_name = cast("str", getattr(func, "__name__", "DynamicPromptMiddleware"))
+
+            return type(
+                middleware_name,
+                (AgentMiddleware,),
+                {
+                    "state_schema": AgentState,
+                    "tools": [],
+                    "amodify_model_request": async_wrapped,
+                },
+            )()
+
+        def wrapped(
+            self: AgentMiddleware[StateT, ContextT],  # noqa: ARG001
+            request: ModelRequest,
+            state: StateT,
+            runtime: Runtime[ContextT],
+        ) -> ModelRequest:
+            prompt = cast("str", func(request, state, runtime))
+            request.system_prompt = prompt
+            return request
+
+        middleware_name = cast("str", getattr(func, "__name__", "DynamicPromptMiddleware"))
+
+        return type(
+            middleware_name,
+            (AgentMiddleware,),
+            {
+                "state_schema": AgentState,
+                "tools": [],
+                "modify_model_request": wrapped,
+            },
+        )()
+
+    if func is not None:
+        return decorator(func)
+    return decorator

libs/langchain_v1/tests/unit_tests/agents/test_middleware_decorators.py

@@ -16,6 +16,7 @@
     ModelRequest,
     before_model,
     after_model,
+    dynamic_prompt,
     modify_model_request,
     hook_config,
 )
@@ -572,3 +573,145 @@ async def async_after_with_jumps(state: AgentState, runtime: Runtime) -> dict[st
     )
 
     assert agent_mixed.compile().get_graph().draw_mermaid() == snapshot
+
+
+def test_dynamic_prompt_decorator() -> None:
+    """Test dynamic_prompt decorator with basic usage."""
+
+    @dynamic_prompt
+    def my_prompt(request: ModelRequest, state: AgentState, runtime: Runtime) -> str:
+        return "Dynamic test prompt"
+
+    assert isinstance(my_prompt, AgentMiddleware)
+    assert my_prompt.state_schema == AgentState
+    assert my_prompt.tools == []
+    assert my_prompt.__class__.__name__ == "my_prompt"
+
+    # Verify it modifies the request correctly
+    original_request = ModelRequest(
+        model="test-model",
+        system_prompt="Original",
+        messages=[HumanMessage("Hello")],
+        tool_choice=None,
+        tools=[],
+        response_format=None,
+    )
+    result = my_prompt.modify_model_request(
+        original_request, {"messages": [HumanMessage("Hello")]}, None
+    )
+    assert result.system_prompt == "Dynamic test prompt"
+
+
+def test_dynamic_prompt_uses_state() -> None:
+    """Test that dynamic_prompt can use state information."""
+
+    @dynamic_prompt
+    def custom_prompt(request: ModelRequest, state: AgentState, runtime: Runtime) -> str:
+        msg_count = len(state["messages"])
+        return f"Prompt with {msg_count} messages"
+
+    # Verify it uses state correctly
+    original_request = ModelRequest(
+        model="test-model",
+        system_prompt="Original",
+        messages=[HumanMessage("Hello")],
+        tool_choice=None,
+        tools=[],
+        response_format=None,
+    )
+    result = custom_prompt.modify_model_request(
+        original_request, {"messages": [HumanMessage("Hello"), HumanMessage("World")]}, None
+    )
+    assert result.system_prompt == "Prompt with 2 messages"
+
+
+def test_dynamic_prompt_integration() -> None:
+    """Test dynamic_prompt decorator in a full agent."""
+
+    prompt_calls = 0
+
+    @dynamic_prompt
+    def context_aware_prompt(request: ModelRequest, state: AgentState, runtime: Runtime) -> str:
+        nonlocal prompt_calls
+        prompt_calls += 1
+        return f"you are a helpful assistant."
+
+    agent = create_agent(model=FakeToolCallingModel(), middleware=[context_aware_prompt])
+    agent = agent.compile()
+
+    result = agent.invoke({"messages": [HumanMessage("Hello")]})
+
+    assert prompt_calls == 1
+    assert result["messages"][-1].content == "you are a helpful assistant.-Hello"
+
+
+async def test_async_dynamic_prompt_decorator() -> None:
+    """Test dynamic_prompt decorator with async function."""
+
+    @dynamic_prompt
+    async def async_prompt(request: ModelRequest, state: AgentState, runtime: Runtime) -> str:
+        return "Async dynamic prompt"
+
+    assert isinstance(async_prompt, AgentMiddleware)
+    assert async_prompt.state_schema == AgentState
+    assert async_prompt.tools == []
+    assert async_prompt.__class__.__name__ == "async_prompt"
+
+
+async def test_async_dynamic_prompt_integration() -> None:
+    """Test async dynamic_prompt decorator in a full agent."""
+
+    prompt_calls = 0
+
+    @dynamic_prompt
+    async def async_context_prompt(
+        request: ModelRequest, state: AgentState, runtime: Runtime
+    ) -> str:
+        nonlocal prompt_calls
+        prompt_calls += 1
+        return f"Async assistant."
+
+    agent = create_agent(model=FakeToolCallingModel(), middleware=[async_context_prompt])
+    agent = agent.compile()
+
+    result = await agent.ainvoke({"messages": [HumanMessage("Hello")]})
+    assert prompt_calls == 1
+    assert result["messages"][-1].content == "Async assistant.-Hello"
+
+
+def test_dynamic_prompt_overwrites_system_prompt() -> None:
+    """Test that dynamic_prompt overwrites the original system_prompt."""
+
+    @dynamic_prompt
+    def override_prompt(request: ModelRequest, state: AgentState, runtime: Runtime) -> str:
+        return "Overridden prompt."
+
+    agent = create_agent(
+        model=FakeToolCallingModel(),
+        system_prompt="Original static prompt",
+        middleware=[override_prompt],
+    )
+    agent = agent.compile()
+
+    result = agent.invoke({"messages": [HumanMessage("Hello")]})
+    assert result["messages"][-1].content == "Overridden prompt.-Hello"
+
+
+def test_dynamic_prompt_multiple_in_sequence() -> None:
+    """Test multiple dynamic_prompt decorators in sequence (last wins)."""
+
+    @dynamic_prompt
+    def first_prompt(request: ModelRequest, state: AgentState, runtime: Runtime) -> str:
+        return "First prompt."
+
+    @dynamic_prompt
+    def second_prompt(request: ModelRequest, state: AgentState, runtime: Runtime) -> str:
+        return "Second prompt."
+
+    # When used together, the last middleware in the list should win
+    # since they're both modify_model_request hooks executed in sequence
+    agent = create_agent(model=FakeToolCallingModel(), middleware=[first_prompt, second_prompt])
+    agent = agent.compile()
+
+    result = agent.invoke({"messages": [HumanMessage("Hello")]})
+    assert result["messages"][-1].content == "Second prompt.-Hello"