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

test_async_retry() — langchain Function Reference

Architecture documentation for the test_async_retry() function in test_wrap_model_call.py from the langchain codebase.

Function python LangChainCore LanguageModelBase calls 2

Entity Profile

Dependency Diagram

graph TD
  a22d9207_3949_0a51_b352_3861c56117b7["test_async_retry()"]
  75e76fa8_8e23_8ada_b149_8fa3fd341b70["TestAsyncWrapModelCall"]
  a22d9207_3949_0a51_b352_3861c56117b7 -->|defined in| 75e76fa8_8e23_8ada_b149_8fa3fd341b70
  76f9c2cf_408e_f96a_c3ed_7090a3a6c165["_agenerate()"]
  a22d9207_3949_0a51_b352_3861c56117b7 -->|calls| 76f9c2cf_408e_f96a_c3ed_7090a3a6c165
  8b14d6df_29e1_d984_6a2a_6e2bb0e2e4b2["awrap_model_call()"]
  a22d9207_3949_0a51_b352_3861c56117b7 -->|calls| 8b14d6df_29e1_d984_6a2a_6e2bb0e2e4b2
  style a22d9207_3949_0a51_b352_3861c56117b7 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_wrap_model_call.py lines 1405–1441

    async def test_async_retry(self) -> None:
        """Test retry logic with async execution."""
        call_count = {"value": 0}

        class AsyncFailOnceThenSucceed(GenericFakeChatModel):
            @override
            async def _agenerate(
                self,
                messages: list[BaseMessage],
                stop: list[str] | None = None,
                run_manager: AsyncCallbackManagerForLLMRun | None = None,
                **kwargs: Any,
            ) -> ChatResult:
                call_count["value"] += 1
                if call_count["value"] == 1:
                    msg = "First async call fails"
                    raise ValueError(msg)
                return await super()._agenerate(messages, **kwargs)

        class RetryMiddleware(AgentMiddleware):
            async def awrap_model_call(
                self,
                request: ModelRequest,
                handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
            ) -> ModelCallResult:
                try:
                    return await handler(request)
                except Exception:
                    return await handler(request)

        model = AsyncFailOnceThenSucceed(messages=iter([AIMessage(content="Async success")]))
        agent = create_agent(model=model, middleware=[RetryMiddleware()])

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

        assert call_count["value"] == 2
        assert result["messages"][1].content == "Async success"

Domain

Subdomains

Defined In

Calls

Source

Frequently Asked Questions

What does test_async_retry() do?
test_async_retry() is a function in the langchain codebase, defined in libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_wrap_model_call.py.
Where is test_async_retry() defined?
test_async_retry() is defined in libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_wrap_model_call.py at line 1405.
What does test_async_retry() call?
test_async_retry() calls 2 function(s): _agenerate, awrap_model_call.

Analyze Your Own Codebase

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

Try Supermodel Free