TestLLMToolEmulatorAsync Class — langchain Architecture

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_tool_emulator.py lines 421–623

class TestLLMToolEmulatorAsync:
    """Test async tool emulator functionality."""

    async def test_async_emulates_specified_tool_by_name(self) -> None:
        """Test that tools specified by name are emulated in async mode."""
        agent_model = FakeModel(
            messages=cycle(
                [
                    AIMessage(
                        content="",
                        tool_calls=[
                            {"name": "get_weather", "id": "1", "args": {"location": "Paris"}}
                        ],
                    ),
                    AIMessage(content="The weather has been retrieved."),
                ]
            )
        )

        emulator_model = FakeEmulatorModel(responses=["Emulated: 72°F, sunny in Paris"])

        emulator = LLMToolEmulator(tools=["get_weather"], model=emulator_model)

        agent = create_agent(
            model=agent_model,
            tools=[get_weather, calculator],
            middleware=[emulator],
        )

        result = await agent.ainvoke({"messages": [HumanMessage("What's the weather in Paris?")]})

        # Should complete without raising NotImplementedError
        assert isinstance(result["messages"][-1], AIMessage)

    async def test_async_emulates_specified_tool_by_instance(self) -> None:
        """Test that tools specified by BaseTool instance are emulated in async mode."""
        agent_model = FakeModel(
            messages=cycle(
                [
                    AIMessage(
                        content="",
                        tool_calls=[{"name": "search_web", "id": "1", "args": {"query": "Python"}}],
                    ),
                    AIMessage(content="Search results retrieved."),
                ]
            )
        )

        emulator_model = FakeEmulatorModel(responses=["Emulated: Python is a programming language"])

        emulator = LLMToolEmulator(tools=[search_web], model=emulator_model)

        agent = create_agent(
            model=agent_model,
            tools=[search_web, calculator],
            middleware=[emulator],
        )

        result = await agent.ainvoke({"messages": [HumanMessage("Search for Python")]})

        assert isinstance(result["messages"][-1], AIMessage)

    async def test_async_non_emulated_tools_execute_normally(self) -> None:
        """Test that tools not in tools_to_emulate execute normally in async mode."""
        agent_model = FakeModel(
            messages=cycle(
                [
                    AIMessage(
                        content="",
                        tool_calls=[
                            {"name": "calculator", "id": "1", "args": {"expression": "2+2"}}
                        ],
                    ),
                    AIMessage(content="The calculation is complete."),
                ]
            )
        )

        emulator_model = FakeEmulatorModel(responses=["Should not be used"])

        # Only emulate get_weather, not calculator