_chain_async_model_call_handlers() — langchain Function Reference

Architecture documentation for the _chain_async_model_call_handlers() function in factory.py from the langchain codebase.

Function python LangChainCore LanguageModelBase calls 1 called by 1

Entity Profile

LangChainCore→ LanguageModelBase→ _chain_async_model_call_handlers() — langchain Function Reference

Dependency Diagram

graph TD
  611bf38d_1c0a_8d53_2a61_0d70eb6cac51["_chain_async_model_call_handlers()"]
  fd7a28b1_3772_169b_6524_1342f35143b1["factory.py"]
  611bf38d_1c0a_8d53_2a61_0d70eb6cac51 -->|defined in| fd7a28b1_3772_169b_6524_1342f35143b1
  f4b66c38_651c_807a_caca_41f73fbbe516["create_agent()"]
  f4b66c38_651c_807a_caca_41f73fbbe516 -->|calls| 611bf38d_1c0a_8d53_2a61_0d70eb6cac51
  d2970fae_d131_e7b8_2f45_5fcefc4ea166["_normalize_to_model_response()"]
  611bf38d_1c0a_8d53_2a61_0d70eb6cac51 -->|calls| d2970fae_d131_e7b8_2f45_5fcefc4ea166
  style 611bf38d_1c0a_8d53_2a61_0d70eb6cac51 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/factory.py lines 296–384

def _chain_async_model_call_handlers(
    handlers: Sequence[_AsyncModelCallHandler[ContextT]],
) -> _ComposedAsyncModelCallHandler[ContextT] | None:
    """Compose multiple async ``wrap_model_call`` handlers into single middleware stack.

    Commands from each layer are accumulated into a list (inner-first, then outer)
    without merging.

    Args:
        handlers: List of async handlers.

            First handler wraps all others.

    Returns:
        Composed async handler returning ``_ComposedExtendedModelResponse``,
        or ``None`` if handlers empty.
    """
    if not handlers:
        return None

    def _to_composed_result(
        result: ModelResponse | AIMessage | ExtendedModelResponse | _ComposedExtendedModelResponse,
        extra_commands: list[Command[Any]] | None = None,
    ) -> _ComposedExtendedModelResponse:
        """Normalize any handler result to _ComposedExtendedModelResponse."""
        commands: list[Command[Any]] = list(extra_commands or [])
        if isinstance(result, _ComposedExtendedModelResponse):
            commands.extend(result.commands)
            model_response = result.model_response
        elif isinstance(result, ExtendedModelResponse):
            model_response = result.model_response
            if result.command is not None:
                commands.append(result.command)
        else:
            model_response = _normalize_to_model_response(result)

        return _ComposedExtendedModelResponse(model_response=model_response, commands=commands)

    if len(handlers) == 1:
        single_handler = handlers[0]

        async def normalized_single(
            request: ModelRequest[ContextT],
            handler: Callable[[ModelRequest[ContextT]], Awaitable[ModelResponse]],
        ) -> _ComposedExtendedModelResponse:
            return _to_composed_result(await single_handler(request, handler))

        return normalized_single

    def compose_two(
        outer: _AsyncModelCallHandler[ContextT] | _ComposedAsyncModelCallHandler[ContextT],
        inner: _AsyncModelCallHandler[ContextT] | _ComposedAsyncModelCallHandler[ContextT],
    ) -> _ComposedAsyncModelCallHandler[ContextT]:
        """Compose two async handlers where outer wraps inner."""

        async def composed(
            request: ModelRequest[ContextT],
            handler: Callable[[ModelRequest[ContextT]], Awaitable[ModelResponse]],
        ) -> _ComposedExtendedModelResponse:
            # Closure variable to capture inner's commands before normalizing
            accumulated_commands: list[Command[Any]] = []

            async def inner_handler(req: ModelRequest[ContextT]) -> ModelResponse:
                # Clear on each call for retry safety
                accumulated_commands.clear()
                inner_result = await inner(req, handler)
                if isinstance(inner_result, _ComposedExtendedModelResponse):
                    accumulated_commands.extend(inner_result.commands)
                    return inner_result.model_response
                if isinstance(inner_result, ExtendedModelResponse):
                    if inner_result.command is not None:
                        accumulated_commands.append(inner_result.command)
                    return inner_result.model_response
                return _normalize_to_model_response(inner_result)

            outer_result = await outer(request, inner_handler)
            return _to_composed_result(
                outer_result,
                extra_commands=accumulated_commands or None,
            )