TestDynamicModelWithResponseFormat Class — langchain Architecture

Source Code

libs/langchain_v1/tests/unit_tests/agents/test_response_format.py lines 789–871

class TestDynamicModelWithResponseFormat:
    """Test response_format with middleware that modifies the model."""

    def test_middleware_model_swap_provider_to_tool_strategy(self) -> None:
        """Test that strategy resolution is deferred until after middleware modifies the model.

        Verifies that when a raw schema is provided, `_supports_provider_strategy` is called
        on the middleware-modified model (not the original), ensuring the correct strategy is
        selected based on the final model's capabilities.
        """

        # Custom model that we'll use to test whether the tool strategy is applied
        # correctly at runtime.
        class CustomModel(GenericFakeChatModel):
            tool_bindings: list[Any] = Field(default_factory=list)

            def bind_tools(
                self,
                tools: Sequence[dict[str, Any] | type[BaseModel] | Callable[..., Any] | BaseTool],
                **kwargs: Any,
            ) -> Runnable[LanguageModelInput, AIMessage]:
                # Record every tool binding event.
                self.tool_bindings.append(tools)
                return self

        model = CustomModel(
            messages=iter(
                [
                    # Simulate model returning structured output directly
                    # (this is what provider strategy would do)
                    json.dumps(WEATHER_DATA),
                ]
            )
        )

        # Create middleware that swaps the model in the request
        class ModelSwappingMiddleware(AgentMiddleware):
            def wrap_model_call(
                self,
                request: ModelRequest,
                handler: Callable[[ModelRequest], ModelResponse],
            ) -> ModelCallResult:
                # Replace the model with our custom test model
                return handler(request.override(model=model))

        # Track which model is checked for provider strategy support
        calls = []

        def mock_supports_provider_strategy(
            model: str | BaseChatModel, tools: list[Any] | None = None
        ) -> bool:
            """Track which model is checked and return True for ProviderStrategy."""
            calls.append(model)
            return True

        # Use raw Pydantic model (not wrapped in ToolStrategy or ProviderStrategy)
        # This should auto-detect strategy based on model capabilities
        agent = create_agent(
            model=model,
            tools=[],
            # Raw schema - should auto-detect strategy
            response_format=WeatherBaseModel,
            middleware=[ModelSwappingMiddleware()],
        )

        with patch(
            "langchain.agents.factory._supports_provider_strategy",
            side_effect=mock_supports_provider_strategy,
        ):
            response = agent.invoke({"messages": [HumanMessage("What's the weather?")]})

        # Verify strategy resolution was deferred: check was called once during _get_bound_model
        assert len(calls) == 1

        # Verify successful parsing of JSON as structured output via ProviderStrategy
        assert response["structured_response"] == EXPECTED_WEATHER_PYDANTIC
        # Two messages: Human input message and AI response with JSON content
        assert len(response["messages"]) == 2
        ai_message = response["messages"][1]
        assert isinstance(ai_message, AIMessage)
        # ProviderStrategy doesn't use tool calls - it parses content directly