TestAgentHooksCombined Class — langchain Architecture

Source Code

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

class TestAgentHooksCombined:
    """Test before_agent and after_agent hooks working together."""

    @pytest.mark.parametrize("is_async", [False, True])
    async def test_execution_order(self, *, is_async: bool) -> None:
        """Test that before_agent executes before after_agent in both sync and async modes."""
        execution_log: list[str] = []

        if is_async:

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

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

        else:

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

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

        model = GenericFakeChatModel(messages=iter([AIMessage(content="Response")]))
        agent = create_agent(model=model, tools=[], middleware=[log_before, log_after])

        if is_async:
            await agent.ainvoke({"messages": [HumanMessage("Test")]})
        else:
            agent.invoke({"messages": [HumanMessage("Test")]})

        assert execution_log == ["before", "after"]

    def test_state_passthrough(self) -> None:
        """Test that state modifications in before_agent are visible to after_agent."""

        @before_agent
        def modify_in_before(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
            return {"messages": [HumanMessage("Added by before_agent")]}

        model = GenericFakeChatModel(messages=iter([AIMessage(content="Response")]))
        agent = create_agent(model=model, tools=[], middleware=[modify_in_before])
        result = agent.invoke({"messages": [HumanMessage("Original")]})

        message_contents = [msg.content for msg in result["messages"]]
        assert message_contents[1] == "Added by before_agent"

    def test_multiple_middleware_instances(self) -> None:
        """Test multiple before_agent and after_agent middleware instances."""
        execution_log = []

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

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

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

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

        model = GenericFakeChatModel(messages=iter([AIMessage(content="Response")]))
        agent = create_agent(
            model=model, tools=[], middleware=[before_one, before_two, after_one, after_two]
        )
        agent.invoke({"messages": [HumanMessage("Test")]})

        assert execution_log == ["before_1", "before_2", "after_2", "after_1"]

    def test_agent_hooks_run_once_with_multiple_model_calls(self) -> None: