chore(langchain): Support for SystemMessage in create_agent (#33640)

- **Description:** Updated Function Signature of `create_agent`, the
system prompt can be both a list and string. I see no harm in doing
this, since SystemMessage accepts both.
- **Issue:** #33630

---------

Co-authored-by: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com>

Author: keenborder786

libs/langchain_v1/langchain/agents/factory.py

@@ -516,7 +516,7 @@ def create_agent(  # noqa: PLR0915
     model: str | BaseChatModel,
     tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None,
     *,
-    system_prompt: str | None = None,
+    system_prompt: str | SystemMessage | None = None,
     middleware: Sequence[AgentMiddleware[StateT_co, ContextT]] = (),
     response_format: ResponseFormat[ResponseT] | type[ResponseT] | None = None,
     state_schema: type[AgentState[ResponseT]] | None = None,
@@ -548,11 +548,9 @@ def create_agent(  # noqa: PLR0915
 
             If `None` or an empty list, the agent will consist of a model node without a
             tool calling loop.
-        system_prompt: An optional system prompt for the LLM.
+        system_prompt: An optional system prompt for the LLM or
+        can already be a [`SystemMessage`][langchain.messages.SystemMessage] object.
 
-            Prompts are converted to a
-            [`SystemMessage`][langchain.messages.SystemMessage] and added to the
-            beginning of the message list.
         middleware: A sequence of middleware instances to apply to the agent.
 
             Middleware can intercept and modify agent behavior at various stages. See
@@ -1040,8 +1038,10 @@ def _execute_model_sync(request: ModelRequest) -> ModelResponse:
         # Get the bound model (with auto-detection if needed)
         model_, effective_response_format = _get_bound_model(request)
         messages = request.messages
-        if request.system_prompt:
-            messages = [SystemMessage(request.system_prompt), *messages]
+        if request.system_prompt and not isinstance(request.system_prompt, SystemMessage):
+            messages = [SystemMessage(content=request.system_prompt), *messages]
+        elif request.system_prompt and isinstance(request.system_prompt, SystemMessage):
+            messages = [request.system_prompt, *messages]
 
         output = model_.invoke(messages)
 
@@ -1093,8 +1093,10 @@ async def _execute_model_async(request: ModelRequest) -> ModelResponse:
         # Get the bound model (with auto-detection if needed)
         model_, effective_response_format = _get_bound_model(request)
         messages = request.messages
-        if request.system_prompt:
-            messages = [SystemMessage(request.system_prompt), *messages]
+        if request.system_prompt and not isinstance(request.system_prompt, SystemMessage):
+            messages = [SystemMessage(content=request.system_prompt), *messages]
+        elif request.system_prompt and isinstance(request.system_prompt, SystemMessage):
+            messages = [request.system_prompt, *messages]
 
         output = await model_.ainvoke(messages)
 

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

@@ -225,9 +225,11 @@ def wrap_model_call(
             def count_tokens(messages: Sequence[BaseMessage]) -> int:
                 return count_tokens_approximately(messages)
         else:
-            system_msg = (
-                [SystemMessage(content=request.system_prompt)] if request.system_prompt else []
-            )
+            system_msg = []
+            if request.system_prompt and not isinstance(request.system_prompt, SystemMessage):
+                system_msg = [SystemMessage(content=request.system_prompt)]
+            elif request.system_prompt and isinstance(request.system_prompt, SystemMessage):
+                system_msg = [request.system_prompt]
 
             def count_tokens(messages: Sequence[BaseMessage]) -> int:
                 return request.model.get_num_tokens_from_messages(
@@ -253,9 +255,12 @@ async def awrap_model_call(
             def count_tokens(messages: Sequence[BaseMessage]) -> int:
                 return count_tokens_approximately(messages)
         else:
-            system_msg = (
-                [SystemMessage(content=request.system_prompt)] if request.system_prompt else []
-            )
+            system_msg = []
+
+            if request.system_prompt and not isinstance(request.system_prompt, SystemMessage):
+                system_msg = [SystemMessage(content=request.system_prompt)]
+            elif request.system_prompt and isinstance(request.system_prompt, SystemMessage):
+                system_msg = [request.system_prompt]
 
             def count_tokens(messages: Sequence[BaseMessage]) -> int:
                 return request.model.get_num_tokens_from_messages(

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

@@ -8,7 +8,7 @@
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
 
-from langchain_core.messages import ToolMessage
+from langchain_core.messages import SystemMessage, ToolMessage
 from langchain_core.tools import tool
 from langgraph.types import Command
 from typing_extensions import NotRequired, TypedDict
@@ -199,11 +199,22 @@ def wrap_model_call(
         handler: Callable[[ModelRequest], ModelResponse],
     ) -> ModelCallResult:
         """Update the system prompt to include the todo system prompt."""
-        request.system_prompt = (
-            request.system_prompt + "\n\n" + self.system_prompt
-            if request.system_prompt
-            else self.system_prompt
-        )
+        if request.system_prompt is None:
+            request.system_prompt = self.system_prompt
+        elif isinstance(request.system_prompt, str):
+            request.system_prompt = request.system_prompt + "\n\n" + self.system_prompt
+        elif isinstance(request.system_prompt, SystemMessage) and isinstance(
+            request.system_prompt.content, str
+        ):
+            request.system_prompt = SystemMessage(
+                content=request.system_prompt.content + self.system_prompt
+            )
+        elif isinstance(request.system_prompt, SystemMessage) and isinstance(
+            request.system_prompt.content, list
+        ):
+            request.system_prompt = SystemMessage(
+                content=[*request.system_prompt.content, self.system_prompt]
+            )
         return handler(request)
 
     async def awrap_model_call(
@@ -212,9 +223,20 @@ async def awrap_model_call(
         handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
     ) -> ModelCallResult:
         """Update the system prompt to include the todo system prompt (async version)."""
-        request.system_prompt = (
-            request.system_prompt + "\n\n" + self.system_prompt
-            if request.system_prompt
-            else self.system_prompt
-        )
+        if request.system_prompt is None:
+            request.system_prompt = self.system_prompt
+        elif isinstance(request.system_prompt, str):
+            request.system_prompt = request.system_prompt + "\n\n" + self.system_prompt
+        elif isinstance(request.system_prompt, SystemMessage) and isinstance(
+            request.system_prompt.content, str
+        ):
+            request.system_prompt = SystemMessage(
+                content=request.system_prompt.content + self.system_prompt
+            )
+        elif isinstance(request.system_prompt, SystemMessage) and isinstance(
+            request.system_prompt.content, list
+        ):
+            request.system_prompt = SystemMessage(
+                content=[*request.system_prompt.content, self.system_prompt]
+            )
         return await handler(request)

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

@@ -26,6 +26,7 @@
     AIMessage,
     AnyMessage,
     BaseMessage,
+    SystemMessage,
     ToolMessage,
 )
 from langgraph.channels.ephemeral_value import EphemeralValue
@@ -85,7 +86,7 @@ class ModelRequest:
     """Model request information for the agent."""
 
     model: BaseChatModel
-    system_prompt: str | None
+    system_prompt: str | SystemMessage | None
     messages: list[AnyMessage]  # excluding system prompt
     tool_choice: Any | None
     tools: list[BaseTool | dict]
@@ -103,7 +104,7 @@ def override(self, **overrides: Unpack[_ModelRequestOverrides]) -> ModelRequest:
         Args:
             **overrides: Keyword arguments for attributes to override. Supported keys:
                 - model: BaseChatModel instance
-                - system_prompt: Optional system prompt string
+                - system_prompt: Optional system prompt string or SystemMessage object
                 - messages: List of messages
                 - tool_choice: Tool choice configuration
                 - tools: List of available tools
@@ -1256,7 +1257,7 @@ def wrapped(
             request: ModelRequest,
             handler: Callable[[ModelRequest], ModelResponse],
         ) -> ModelCallResult:
-            prompt = cast("str", func(request))
+            prompt = cast("str | SystemMessage", func(request))
             request.system_prompt = prompt
             return handler(request)
 
@@ -1266,7 +1267,7 @@ async def async_wrapped_from_sync(
             handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
         ) -> ModelCallResult:
             # Delegate to sync function
-            prompt = cast("str", func(request))
+            prompt = cast("str | SystemMessage", func(request))
             request.system_prompt = prompt
             return await handler(request)
 

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

@@ -13,6 +13,7 @@
 from langchain_core.messages import (
     AIMessage,
     MessageLikeRepresentation,
+    SystemMessage,
     ToolMessage,
 )
 from langgraph.runtime import Runtime
@@ -399,3 +400,126 @@ async def mock_handler(req: ModelRequest) -> AIMessage:
 
     assert isinstance(calc_tool, ToolMessage)
     assert calc_tool.content == "[cleared]"
+
+
+# ==============================================================================
+# SystemMessage Tests
+# ==============================================================================
+
+
+def test_handles_system_message_prompt() -> None:
+    """Test that middleware handles SystemMessage as system_prompt correctly."""
+    tool_call_id = "call-1"
+    ai_message = AIMessage(
+        content="",
+        tool_calls=[{"id": tool_call_id, "name": "search", "args": {}}],
+    )
+    tool_message = ToolMessage(content="12345", tool_call_id=tool_call_id)
+
+    system_prompt = SystemMessage(content="You are a helpful assistant.")
+    state, request = _make_state_and_request([ai_message, tool_message], system_prompt=None)
+    # Manually set SystemMessage as system_prompt
+    request.system_prompt = system_prompt
+
+    middleware = ContextEditingMiddleware(
+        edits=[ClearToolUsesEdit(trigger=50)],
+        token_count_method="model",
+    )
+
+    def mock_handler(req: ModelRequest) -> AIMessage:
+        return AIMessage(content="mock response")
+
+    # Call wrap_model_call - should not fail with SystemMessage
+    middleware.wrap_model_call(request, mock_handler)
+
+    # Request should have processed without errors
+    assert request.system_prompt == system_prompt
+    assert isinstance(request.system_prompt, SystemMessage)
+
+
+def test_does_not_double_wrap_system_message() -> None:
+    """Test that middleware doesn't wrap SystemMessage in another SystemMessage."""
+    tool_call_id = "call-1"
+    ai_message = AIMessage(
+        content="",
+        tool_calls=[{"id": tool_call_id, "name": "search", "args": {}}],
+    )
+    tool_message = ToolMessage(content="x" * 100, tool_call_id=tool_call_id)
+
+    system_prompt = SystemMessage(content="Original system prompt")
+    state, request = _make_state_and_request([ai_message, tool_message], system_prompt=None)
+    request.system_prompt = system_prompt
+
+    middleware = ContextEditingMiddleware(
+        edits=[ClearToolUsesEdit(trigger=50)],
+        token_count_method="model",
+    )
+
+    def mock_handler(req: ModelRequest) -> AIMessage:
+        return AIMessage(content="mock response")
+
+    middleware.wrap_model_call(request, mock_handler)
+
+    # System prompt should still be the same SystemMessage, not wrapped
+    assert request.system_prompt == system_prompt
+    assert isinstance(request.system_prompt, SystemMessage)
+    assert request.system_prompt.content == "Original system prompt"
+
+
+async def test_handles_system_message_prompt_async() -> None:
+    """Test async version - middleware handles SystemMessage as system_prompt correctly."""
+    tool_call_id = "call-1"
+    ai_message = AIMessage(
+        content="",
+        tool_calls=[{"id": tool_call_id, "name": "search", "args": {}}],
+    )
+    tool_message = ToolMessage(content="12345", tool_call_id=tool_call_id)
+
+    system_prompt = SystemMessage(content="You are a helpful assistant.")
+    state, request = _make_state_and_request([ai_message, tool_message], system_prompt=None)
+    # Manually set SystemMessage as system_prompt
+    request.system_prompt = system_prompt
+
+    middleware = ContextEditingMiddleware(
+        edits=[ClearToolUsesEdit(trigger=50)],
+        token_count_method="model",
+    )
+
+    async def mock_handler(req: ModelRequest) -> AIMessage:
+        return AIMessage(content="mock response")
+
+    # Call awrap_model_call - should not fail with SystemMessage
+    await middleware.awrap_model_call(request, mock_handler)
+
+    # Request should have processed without errors
+    assert request.system_prompt == system_prompt
+    assert isinstance(request.system_prompt, SystemMessage)
+
+
+async def test_does_not_double_wrap_system_message_async() -> None:
+    """Test async version - middleware doesn't wrap SystemMessage in another SystemMessage."""
+    tool_call_id = "call-1"
+    ai_message = AIMessage(
+        content="",
+        tool_calls=[{"id": tool_call_id, "name": "search", "args": {}}],
+    )
+    tool_message = ToolMessage(content="x" * 100, tool_call_id=tool_call_id)
+
+    system_prompt = SystemMessage(content="Original system prompt")
+    state, request = _make_state_and_request([ai_message, tool_message], system_prompt=None)
+    request.system_prompt = system_prompt
+
+    middleware = ContextEditingMiddleware(
+        edits=[ClearToolUsesEdit(trigger=50)],
+        token_count_method="model",
+    )
+
+    async def mock_handler(req: ModelRequest) -> AIMessage:
+        return AIMessage(content="mock response")
+
+    await middleware.awrap_model_call(request, mock_handler)
+
+    # System prompt should still be the same SystemMessage, not wrapped
+    assert request.system_prompt == system_prompt
+    assert isinstance(request.system_prompt, SystemMessage)
+    assert request.system_prompt.content == "Original system prompt"