TestSystemMessageMiddlewareIntegration Class — langchain Architecture

Source Code

libs/langchain_v1/tests/unit_tests/agents/test_system_message.py lines 860–991

class TestSystemMessageMiddlewareIntegration:
    """Test integration of SystemMessage with middleware chain."""

    def test_multiple_middleware_can_modify_system_message(self) -> None:
        """Test that multiple middleware can modify system message in sequence."""

        def first_middleware(request: ModelRequest) -> ModelRequest:
            """First middleware adds base system message."""
            new_message = SystemMessage(
                content="You are an assistant.",
                additional_kwargs={"middleware_1": "applied"},
            )
            return request.override(system_message=new_message)

        def second_middleware(request: ModelRequest) -> ModelRequest:
            """Second middleware appends to system message."""
            assert request.system_message is not None
            current_content = request.system_message.text
            new_content = current_content + " Be helpful."

            merged_kwargs = {
                **request.system_message.additional_kwargs,
                "middleware_2": "applied",
            }

            new_message = SystemMessage(
                content=new_content,
                additional_kwargs=merged_kwargs,
            )
            return request.override(system_message=new_message)

        request = _make_request(system_message=None)

        # Apply middleware in sequence
        request = first_middleware(request)
        assert request.system_message is not None
        assert len(request.system_message.content_blocks) == 1
        assert request.system_message.content_blocks[0].get("text") == "You are an assistant."
        assert request.system_message.additional_kwargs["middleware_1"] == "applied"

        request = second_middleware(request)
        assert request.system_message is not None
        assert len(request.system_message.content_blocks) == 1
        assert (
            request.system_message.content_blocks[0].get("text")
            == "You are an assistant. Be helpful."
        )
        assert request.system_message.additional_kwargs["middleware_1"] == "applied"
        assert request.system_message.additional_kwargs["middleware_2"] == "applied"

    def test_middleware_preserves_system_message_metadata(self) -> None:
        """Test that metadata is preserved when middleware modifies system message."""
        base_message = SystemMessage(
            content="Base prompt",
            additional_kwargs={"key1": "value1", "key2": "value2"},
            response_metadata={"model": "gpt-4"},
        )

        def preserving_middleware(request: ModelRequest) -> ModelRequest:
            """Middleware that preserves existing metadata."""
            assert request.system_message is not None
            new_message = SystemMessage(
                content=request.system_message.text + " Extended.",
                additional_kwargs=request.system_message.additional_kwargs,
                response_metadata=request.system_message.response_metadata,
            )
            return request.override(system_message=new_message)

        request = _make_request(system_message=base_message)
        new_request = preserving_middleware(request)

        assert new_request.system_message is not None
        assert len(new_request.system_message.content_blocks) == 1
        assert new_request.system_message.content_blocks[0].get("text") == "Base prompt Extended."
        assert new_request.system_message.additional_kwargs == {
            "key1": "value1",
            "key2": "value2",
        }
        assert new_request.system_message.response_metadata == {"model": "gpt-4"}

    def test_backward_compatibility_with_string_system_prompt(self) -> None: