test_sync_invoke_with_mixed_middleware() — langchain Function Reference

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_wrap_model_call.py lines 1490–1532

    def test_sync_invoke_with_mixed_middleware(self) -> None:
        """Test that sync invoke works with mixed sync/async middleware when sync versions exist."""
        calls = []

        class MixedMiddleware(AgentMiddleware):
            @override
            def before_model(self, state: AgentState[Any], runtime: Runtime[Any]) -> None:
                calls.append("MixedMiddleware.before_model")

            @override
            async def abefore_model(self, state: AgentState[Any], runtime: Runtime[Any]) -> None:
                calls.append("MixedMiddleware.abefore_model")

            def wrap_model_call(
                self,
                request: ModelRequest,
                handler: Callable[[ModelRequest], ModelResponse],
            ) -> ModelCallResult:
                calls.append("MixedMiddleware.wrap_model_call")
                return handler(request)

            async def awrap_model_call(
                self,
                request: ModelRequest,
                handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
            ) -> ModelCallResult:
                calls.append("MixedMiddleware.awrap_model_call")
                return await handler(request)

        agent = create_agent(
            model=FakeToolCallingModel(),
            tools=[],
            system_prompt="You are a helpful assistant.",
            middleware=[MixedMiddleware()],
        )

        agent.invoke({"messages": [HumanMessage("hello")]})

        # In sync mode, only sync methods should be called
        assert calls == [
            "MixedMiddleware.before_model",
            "MixedMiddleware.wrap_model_call",
        ]