langchain Architecture Docs
Home / Function/ test_hook_execution() — langchain Function Reference

test_hook_execution() — langchain Function Reference

Architecture documentation for the test_hook_execution() function in test_framework.py from the langchain codebase.

Function python LangChainCore Runnables

Entity Profile

Dependency Diagram

graph TD
  e16a31b9_f39c_c0b9_3ca8_fc42ed96cc32["test_hook_execution()"]
  9e2d20bd_cff7_3577_a41c_be4a0e929bf8["TestAgentMiddlewareHooks"]
  e16a31b9_f39c_c0b9_3ca8_fc42ed96cc32 -->|defined in| 9e2d20bd_cff7_3577_a41c_be4a0e929bf8
  style e16a31b9_f39c_c0b9_3ca8_fc42ed96cc32 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_framework.py lines 774–828

    async def test_hook_execution(self, *, is_async: bool, hook_type: str) -> None:
        """Test that agent hooks are called in both sync and async modes."""
        execution_log: list[str] = []

        if is_async:
            if hook_type == "before":

                @before_agent
                async def log_hook(
                    state: AgentState[Any], *_args: Any, **_kwargs: Any
                ) -> dict[str, Any] | None:
                    execution_log.append(f"{hook_type}_agent_called")
                    execution_log.append(f"message_count: {len(state['messages'])}")
                    return None

            else:

                @after_agent
                async def log_hook(
                    state: AgentState[Any], *_args: Any, **_kwargs: Any
                ) -> dict[str, Any] | None:
                    execution_log.append(f"{hook_type}_agent_called")
                    execution_log.append(f"message_count: {len(state['messages'])}")
                    return None

        elif hook_type == "before":

            @before_agent
            def log_hook(
                state: AgentState[Any], *_args: Any, **_kwargs: Any
            ) -> dict[str, Any] | None:
                execution_log.append(f"{hook_type}_agent_called")
                execution_log.append(f"message_count: {len(state['messages'])}")
                return None

        else:

            @after_agent
            def log_hook(
                state: AgentState[Any], *_args: Any, **_kwargs: Any
            ) -> dict[str, Any] | None:
                execution_log.append(f"{hook_type}_agent_called")
                execution_log.append(f"message_count: {len(state['messages'])}")
                return None

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

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

        assert f"{hook_type}_agent_called" in execution_log
        assert any("message_count:" in log for log in execution_log)

Domain

Subdomains

Defined In

Source

Frequently Asked Questions

What does test_hook_execution() do?
test_hook_execution() is a function in the langchain codebase, defined in libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_framework.py.
Where is test_hook_execution() defined?
test_hook_execution() is defined in libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_framework.py at line 774.

Analyze Your Own Codebase

Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.

Try Supermodel Free