test_inner_handler_retry() — langchain Function Reference

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_composition.py lines 173–210

    def test_inner_handler_retry(self) -> None:
        """Test inner handler retrying before outer sees response."""
        inner_attempts = []

        def outer_passthrough(
            request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
        ) -> ModelResponse:
            return handler(request)

        def inner_with_retry(
            request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
        ) -> ModelResponse | AIMessage:
            for attempt in range(3):
                inner_attempts.append(attempt)
                try:
                    return handler(request)
                except ValueError:
                    if attempt == 2:
                        raise
            return AIMessage(content="should not reach")

        composed = _chain_model_call_handlers([outer_passthrough, inner_with_retry])
        assert composed is not None

        call_count = {"value": 0}

        def mock_base_handler(req: ModelRequest) -> ModelResponse:
            call_count["value"] += 1
            if call_count["value"] < 3:
                msg = "fail"
                raise ValueError(msg)
            return ModelResponse(result=[AIMessage(content="success")], structured_response=None)

        result = composed(create_test_request(), mock_base_handler)

        assert inner_attempts == [0, 1, 2]
        assert isinstance(result, _ComposedExtendedModelResponse)
        assert result.model_response.result[0].content == "success"