langchain Architecture Docs
Home / Class/ TestAsyncComposition Class — langchain Architecture

TestAsyncComposition Class — langchain Architecture

Architecture documentation for the TestAsyncComposition class in test_wrap_model_call_state_update.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  0f39b70e_2926_951e_853d_3344387750a6["TestAsyncComposition"]
  949c7cf4_56fe_f3b4_cd89_9631a7e9cb1e["AgentMiddleware"]
  0f39b70e_2926_951e_853d_3344387750a6 -->|extends| 949c7cf4_56fe_f3b4_cd89_9631a7e9cb1e
  707a7c14_db9d_706f_20af_75c8ede73799["ModelResponse"]
  0f39b70e_2926_951e_853d_3344387750a6 -->|extends| 707a7c14_db9d_706f_20af_75c8ede73799
  e4070722_d3c6_f73d_d727_5779ab576a2a["AgentState"]
  0f39b70e_2926_951e_853d_3344387750a6 -->|extends| e4070722_d3c6_f73d_d727_5779ab576a2a
  08361e76_a995_1e57_f200_e292474e2f00["test_wrap_model_call_state_update.py"]
  0f39b70e_2926_951e_853d_3344387750a6 -->|defined in| 08361e76_a995_1e57_f200_e292474e2f00
  e4a9b953_7f2b_c6fa_b196_05f6e3c3dd78["test_async_inner_command_propagated()"]
  0f39b70e_2926_951e_853d_3344387750a6 -->|method| e4a9b953_7f2b_c6fa_b196_05f6e3c3dd78
  27773a64_cc86_16f8_6afe_6a894b80f58b["test_async_both_commands_additive_messages()"]
  0f39b70e_2926_951e_853d_3344387750a6 -->|method| 27773a64_cc86_16f8_6afe_6a894b80f58b
  0d52416f_581d_2540_a7c2_aaffcd76d1b8["test_async_inner_command_retry_safe()"]
  0f39b70e_2926_951e_853d_3344387750a6 -->|method| 0d52416f_581d_2540_a7c2_aaffcd76d1b8

Relationship Graph

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_wrap_model_call_state_update.py lines 639–779

class TestAsyncComposition:
    """Test async ExtendedModelResponse propagation through composed middleware."""

    async def test_async_inner_command_propagated(self) -> None:
        """Async: inner middleware's ExtendedModelResponse command is propagated."""

        class OuterMiddleware(AgentMiddleware):
            async def awrap_model_call(
                self,
                request: ModelRequest,
                handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
            ) -> ModelResponse:
                response = await handler(request)
                assert isinstance(response, ModelResponse)
                return response

        class InnerMiddleware(AgentMiddleware):
            async def awrap_model_call(
                self,
                request: ModelRequest,
                handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
            ) -> ExtendedModelResponse:
                response = await handler(request)
                return ExtendedModelResponse(
                    model_response=response,
                    command=Command(
                        update={
                            "messages": [
                                HumanMessage(content="Inner msg", id="inner"),
                            ]
                        }
                    ),
                )

        model = GenericFakeChatModel(messages=iter([AIMessage(content="Hello")]))
        agent = create_agent(
            model=model,
            middleware=[OuterMiddleware(), InnerMiddleware()],
        )

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

        # Model messages + inner command messages (additive)
        messages = result["messages"]
        assert len(messages) == 3
        assert messages[0].content == "Hi"
        assert messages[1].content == "Hello"
        assert messages[2].content == "Inner msg"

    async def test_async_both_commands_additive_messages(self) -> None:
        """Async: both inner and outer command messages are added alongside model."""

        class OuterMiddleware(AgentMiddleware):
            async def awrap_model_call(
                self,
                request: ModelRequest,
                handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
            ) -> ExtendedModelResponse:
                response = await handler(request)
                return ExtendedModelResponse(
                    model_response=response,
                    command=Command(
                        update={"messages": [HumanMessage(content="Outer msg", id="outer")]}
                    ),
                )

        class InnerMiddleware(AgentMiddleware):
            async def awrap_model_call(
                self,
                request: ModelRequest,
                handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
            ) -> ExtendedModelResponse:
                response = await handler(request)
                return ExtendedModelResponse(
                    model_response=response,
                    command=Command(
                        update={"messages": [HumanMessage(content="Inner msg", id="inner")]}
                    ),
                )

        model = GenericFakeChatModel(messages=iter([AIMessage(content="Hello")]))

Defined In

Extends

Source

Frequently Asked Questions

What is the TestAsyncComposition class?
TestAsyncComposition is a class in the langchain codebase, defined in libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_wrap_model_call_state_update.py.
Where is TestAsyncComposition defined?
TestAsyncComposition is defined in libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_wrap_model_call_state_update.py at line 639.
What does TestAsyncComposition extend?
TestAsyncComposition extends AgentMiddleware, ModelResponse, AgentState.

Analyze Your Own Codebase

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

Try Supermodel Free