test_agent_hooks_run_once_with_multiple_model_calls() — langchain Function Reference

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_framework.py lines 958–1041

    def test_agent_hooks_run_once_with_multiple_model_calls(self) -> None:
        """Test that before_agent and after_agent run only once per thread.

        This test verifies that agent-level hooks (before_agent, after_agent) execute
        exactly once per agent invocation, regardless of how many tool calling loops occur.
        This is different from model-level hooks (before_model, after_model) which run
        on every model invocation within the tool calling loop.
        """
        execution_log = []

        @tool
        def sample_tool_agent(query: str) -> str:
            """A sample tool for testing."""
            return f"Result for: {query}"

        @before_agent
        def log_before_agent(*_args: Any, **_kwargs: Any) -> None:
            execution_log.append("before_agent")

        @before_model
        def log_before_model(*_args: Any, **_kwargs: Any) -> None:
            execution_log.append("before_model")

        @after_agent
        def log_after_agent(*_args: Any, **_kwargs: Any) -> None:
            execution_log.append("after_agent")

        @after_model
        def log_after_model(*_args: Any, **_kwargs: Any) -> None:
            execution_log.append("after_model")

        # Model will call a tool twice, then respond with final answer
        # This creates 3 model invocations total, but agent hooks should still run once
        model = FakeToolCallingModel(
            tool_calls=[
                [{"name": "sample_tool_agent", "args": {"query": "first"}, "id": "1"}],
                [{"name": "sample_tool_agent", "args": {"query": "second"}, "id": "2"}],
                [],  # Third call returns no tool calls (final answer)
            ]
        )

        agent = create_agent(
            model=model,
            tools=[sample_tool_agent],
            middleware=[log_before_agent, log_before_model, log_after_model, log_after_agent],
        )

        agent.invoke(
            {"messages": [HumanMessage("Test")]}, config={"configurable": {"thread_id": "abc"}}
        )

        assert execution_log == [
            "before_agent",
            "before_model",
            "after_model",
            "before_model",
            "after_model",
            "before_model",
            "after_model",
            "after_agent",
        ]

        agent.invoke(
            {"messages": [HumanMessage("Test")]}, config={"configurable": {"thread_id": "abc"}}
        )

        assert execution_log == [
            "before_agent",
            "before_model",
            "after_model",
            "before_model",
            "after_model",
            "before_model",
            "after_model",
            "after_agent",
            "before_agent",
            "before_model",
            "after_model",
            "before_model",
            "after_model",
            "before_model",