_chain_model_call_handlers() — langchain Function Reference

Architecture documentation for the _chain_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_model_call_handlers() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/factory.py lines 204–293

def _chain_model_call_handlers(
    handlers: Sequence[_ModelCallHandler[ContextT]],
) -> _ComposedModelCallHandler[ContextT] | None:
    """Compose multiple ``wrap_model_call`` handlers into single middleware stack.

    Composes handlers so first in list becomes outermost layer. Each handler receives a
    handler callback to execute inner layers. Commands from each layer are accumulated
    into a list (inner-first, then outer) without merging.

    Args:
        handlers: List of handlers.

            First handler wraps all others.

    Returns:
        Composed 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]

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

        return normalized_single

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

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

            def inner_handler(req: ModelRequest[ContextT]) -> ModelResponse:
                # Clear on each call for retry safety
                accumulated_commands.clear()
                inner_result = 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 = outer(request, inner_handler)
            return _to_composed_result(
                outer_result,
                extra_commands=accumulated_commands or None,
            )