TestMetadataMerging Class — langchain Architecture

Source Code

libs/langchain_v1/tests/unit_tests/agents/test_system_message.py lines 682–752

class TestMetadataMerging:
    """Test metadata merging behavior when updating system messages."""

    @pytest.mark.parametrize(
        ("metadata_type", "initial_metadata", "update_metadata", "expected_result"),
        [
            # additional_kwargs merging
            (
                "additional_kwargs",
                {"key1": "value1", "shared": "original"},
                {"key2": "value2", "shared": "updated"},
                {"key1": "value1", "key2": "value2", "shared": "updated"},
            ),
            # response_metadata merging
            (
                "response_metadata",
                {"model": "gpt-4", "region": "us-east"},
                {"tokens": 100, "region": "eu-west"},
                {"model": "gpt-4", "tokens": 100, "region": "eu-west"},
            ),
        ],
        ids=["additional_kwargs", "response_metadata"],
    )
    def test_metadata_merge_across_updates(
        self,
        metadata_type: str,
        initial_metadata: dict[str, Any],
        update_metadata: dict[str, Any],
        expected_result: dict[str, Any],
    ) -> None:
        """Test that metadata merges correctly when updating system message."""
        base_message = SystemMessage(
            content="Base",
            **{metadata_type: initial_metadata},
        )

        def update_middleware(
            request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
        ) -> ModelResponse:
            """Update system message, merging metadata."""
            current_metadata = getattr(request.system_message, metadata_type)
            new_metadata = {**current_metadata, **update_metadata}

            new_request = request.override(
                system_message=SystemMessage(content="Updated", **{metadata_type: new_metadata})
            )
            return handler(new_request)

        model = GenericFakeChatModel(messages=iter([AIMessage(content="response")]))
        request = ModelRequest(
            model=model,
            system_message=base_message,
            messages=[],
            tool_choice=None,
            tools=[],
            response_format=None,
            state=AgentState(messages=[]),
            runtime=Runtime(),
        )

        captured_request = None

        def mock_handler(req: ModelRequest) -> ModelResponse:
            nonlocal captured_request
            captured_request = req
            return ModelResponse(result=[AIMessage(content="response")])

        update_middleware(request, mock_handler)

        assert captured_request is not None
        assert getattr(captured_request.system_message, metadata_type) == expected_result