wrap_model_call() — langchain Function Reference

Architecture documentation for the wrap_model_call() function in model_retry.py from the langchain codebase.

Function python LangChainCore LanguageModelBase calls 1

Entity Profile

LangChainCore→ LanguageModelBase→ wrap_model_call() — langchain Function Reference

Dependency Diagram

graph TD
  05371c56_6c46_808b_c839_e45e7d2e464e["wrap_model_call()"]
  5e783bd0_a142_0fd6_3238_fc2d9985440d["ModelRetryMiddleware"]
  05371c56_6c46_808b_c839_e45e7d2e464e -->|defined in| 5e783bd0_a142_0fd6_3238_fc2d9985440d
  5f0811d1_ab8b_7642_81d3_ed55469793d6["_handle_failure()"]
  05371c56_6c46_808b_c839_e45e7d2e464e -->|calls| 5f0811d1_ab8b_7642_81d3_ed55469793d6
  style 05371c56_6c46_808b_c839_e45e7d2e464e fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/middleware/model_retry.py lines 214–262

    def wrap_model_call(
        self,
        request: ModelRequest[ContextT],
        handler: Callable[[ModelRequest[ContextT]], ModelResponse[ResponseT]],
    ) -> ModelResponse[ResponseT] | AIMessage:
        """Intercept model execution and retry on failure.

        Args:
            request: Model request with model, messages, state, and runtime.
            handler: Callable to execute the model (can be called multiple times).

        Returns:
            `ModelResponse` or `AIMessage` (the final result).

        Raises:
            RuntimeError: If the retry loop completes without returning. (This should not happen.)
        """
        # Initial attempt + retries
        for attempt in range(self.max_retries + 1):
            try:
                return handler(request)
            except Exception as exc:
                attempts_made = attempt + 1  # attempt is 0-indexed

                # Check if we should retry this exception
                if not should_retry_exception(exc, self.retry_on):
                    # Exception is not retryable, handle failure immediately
                    return self._handle_failure(exc, attempts_made)

                # Check if we have more retries left
                if attempt < self.max_retries:
                    # Calculate and apply backoff delay
                    delay = calculate_delay(
                        attempt,
                        backoff_factor=self.backoff_factor,
                        initial_delay=self.initial_delay,
                        max_delay=self.max_delay,
                        jitter=self.jitter,
                    )
                    if delay > 0:
                        time.sleep(delay)
                    # Continue to next retry
                else:
                    # No more retries, handle failure
                    return self._handle_failure(exc, attempts_made)

        # Unreachable: loop always returns via handler success or _handle_failure
        msg = "Unexpected: retry loop completed without returning"
        raise RuntimeError(msg)