test_wrap_tool_call_caching_pattern() — langchain Function Reference

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_wrap_tool_call.py lines 760–815

def test_wrap_tool_call_caching_pattern() -> None:
    """Test caching pattern with wrap_tool_call decorator."""
    cache: dict[tuple[str, str], Any] = {}
    handler_calls = []

    @wrap_tool_call
    def with_cache(
        request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]]
    ) -> ToolMessage | Command[Any]:
        assert isinstance(request.tool, BaseTool)
        # Create cache key from tool name and args
        cache_key = (request.tool.name, str(request.tool_call["args"]))

        # Check cache
        if cache_key in cache:
            return ToolMessage(
                content=cache[cache_key],
                tool_call_id=request.tool_call["id"],
                name=request.tool_call["name"],
            )

        # Execute tool and cache result
        handler_calls.append("executed")
        response = handler(request)

        if isinstance(response, ToolMessage):
            cache[cache_key] = response.content

        return response

    model = FakeToolCallingModel(
        tool_calls=[
            [ToolCall(name="search", args={"query": "test"}, id="1")],
            [ToolCall(name="search", args={"query": "test"}, id="2")],  # Same query
            [],
        ]
    )

    agent = create_agent(
        model=model,
        tools=[search],
        middleware=[with_cache],
        checkpointer=InMemorySaver(),
    )

    result = agent.invoke(
        {"messages": [HumanMessage("Search twice")]},
        {"configurable": {"thread_id": "test"}},
    )

    # Handler should only be called once (second call uses cache)
    assert len(handler_calls) == 1

    tool_messages = [m for m in result["messages"] if isinstance(m, ToolMessage)]
    # Both tool calls should have messages
    assert len(tool_messages) >= 1