"""Chat models for conversational AI."""

from __future__ import annotations

import asyncio
import builtins  # noqa: TC003  # runtime-evaluated; subclass `dict()` shadows the builtin
import contextlib
import inspect
import json
from abc import ABC, abstractmethod
from collections.abc import AsyncIterator, Callable, Iterator, Sequence
from functools import cached_property
from operator import itemgetter
from typing import TYPE_CHECKING, Any, Literal, cast, overload

from langchain_protocol.protocol import MessageFinishData
from pydantic import BaseModel, ConfigDict, Field, model_validator
from typing_extensions import Self, override

from langchain_core._api import beta, deprecated, suppress_langchain_deprecation_warning
from langchain_core.caches import BaseCache
from langchain_core.callbacks import (
    AsyncCallbackManager,
    AsyncCallbackManagerForLLMRun,
    CallbackManager,
    CallbackManagerForLLMRun,
    Callbacks,
)
from langchain_core.globals import get_llm_cache
from langchain_core.language_models._compat_bridge import (
    achunks_to_events,
    amessage_to_events,
    chunks_to_events,
    message_to_events,
)
from langchain_core.language_models._utils import (
    _filter_invocation_params_for_tracing,
    _normalize_messages,
    _update_message_content_to_blocks,
)
from langchain_core.language_models.base import (
    BaseLanguageModel,
    LangSmithParams,
    LanguageModelInput,
)
from langchain_core.language_models.chat_model_stream import (
    AsyncChatModelStream,
    ChatModelStream,
)
from langchain_core.language_models.model_profile import (
    ModelProfile,
    _warn_unknown_profile_keys,
)
from langchain_core.load import dumpd, dumps
from langchain_core.messages import (
    AIMessage,
    AIMessageChunk,
    AnyMessage,
    BaseMessage,
    convert_to_messages,
    is_data_content_block,
    message_chunk_to_message,
)
from langchain_core.messages import content as types
from langchain_core.messages.block_translators.openai import (
    convert_to_openai_image_block,
)
from langchain_core.output_parsers.openai_tools import (
    JsonOutputKeyToolsParser,
    JsonOutputToolsParser,
    PydanticToolsParser,
)
from langchain_core.outputs import (
    ChatGeneration,
    ChatGenerationChunk,
    ChatResult,
    Generation,
    LLMResult,
    RunInfo,
)
from langchain_core.outputs.chat_generation import merge_chat_generation_chunks
from langchain_core.prompt_values import ChatPromptValue, PromptValue, StringPromptValue
from langchain_core.rate_limiters import BaseRateLimiter
from langchain_core.runnables import RunnableBinding, RunnableMap, RunnablePassthrough
from langchain_core.runnables.config import ensure_config, run_in_executor
from langchain_core.tracers._streaming import (
    _StreamingCallbackHandler,
    _V2StreamingCallbackHandler,
)
from langchain_core.utils.function_calling import (
    convert_to_json_schema,
    convert_to_openai_tool,
)
from langchain_core.utils.pydantic import is_basemodel_subclass
from langchain_core.utils.utils import LC_ID_PREFIX, from_env

if TYPE_CHECKING:
    import uuid
    from collections.abc import Awaitable

    from langchain_protocol.protocol import MessagesData

    from langchain_core.runnables import Runnable, RunnableConfig
    from langchain_core.runnables.schema import StreamEvent
    from langchain_core.tools import BaseTool


def _generate_response_from_error(error: BaseException) -> list[ChatGeneration]:
    if hasattr(error, "response"):
        response = error.response
        metadata: dict[str, Any] = {}
        if hasattr(response, "json"):
            try:
                metadata["body"] = response.json()
            except Exception:
                try:
                    metadata["body"] = getattr(response, "text", None)
                except Exception:
                    metadata["body"] = None
        if hasattr(response, "headers"):
            try:
                metadata["headers"] = dict(response.headers)
            except Exception:
                metadata["headers"] = None
        if hasattr(response, "status_code"):
            metadata["status_code"] = response.status_code
        if hasattr(error, "request_id"):
            metadata["request_id"] = error.request_id
        generations = [
            ChatGeneration(message=AIMessage(content="", response_metadata=metadata))
        ]
    else:
        generations = []

    return generations


def _format_for_tracing(messages: list[BaseMessage]) -> list[BaseMessage]:
    """Format messages for tracing in `on_chat_model_start`.

    - Update image content blocks to OpenAI Chat Completions format (backward
    compatibility).
    - Add `type` key to content blocks that have a single key.

    Args:
        messages: List of messages to format.

    Returns:
        List of messages formatted for tracing.

    """
    messages_to_trace = []
    for message in messages:
        message_to_trace = message
        if isinstance(message.content, list):
            for idx, block in enumerate(message.content):
                if isinstance(block, dict):
                    # Update image content blocks to OpenAI # Chat Completions format.
                    if (
                        block.get("type") == "image"
                        and is_data_content_block(block)
                        and not ("file_id" in block or block.get("source_type") == "id")
                    ):
                        if message_to_trace is message:
                            # Shallow copy
                            message_to_trace = message.model_copy()
                            message_to_trace.content = list(message_to_trace.content)

                        message_to_trace.content[idx] = (  # type: ignore[index]  # mypy confused by .model_copy
                            convert_to_openai_image_block(block)
                        )
                    elif (
                        block.get("type") == "file"
                        and is_data_content_block(block)  # v0 (image/audio/file) or v1
                        and "base64" in block
                        # Backward compat: convert v1 base64 blocks to v0
                    ):
                        if message_to_trace is message:
                            # Shallow copy
                            message_to_trace = message.model_copy()
                            message_to_trace.content = list(message_to_trace.content)

                        message_to_trace.content[idx] = {  # type: ignore[index]
                            **{k: v for k, v in block.items() if k != "base64"},
                            "data": block["base64"],
                            "source_type": "base64",
                        }
                    elif len(block) == 1 and "type" not in block:
                        # Tracing assumes all content blocks have a "type" key. Here
                        # we add this key if it is missing, and there's an obvious
                        # choice for the type (e.g., a single key in the block).
                        if message_to_trace is message:
                            # Shallow copy
                            message_to_trace = message.model_copy()
                            message_to_trace.content = list(message_to_trace.content)
                        key = next(iter(block))
                        message_to_trace.content[idx] = {  # type: ignore[index]
                            "type": key,
                            key: block[key],
                        }
        messages_to_trace.append(message_to_trace)

    return messages_to_trace


def generate_from_stream(stream: Iterator[ChatGenerationChunk]) -> ChatResult:
    """Generate from a stream.

    Args:
        stream: Iterator of `ChatGenerationChunk`.

    Raises:
        ValueError: If no generations are found in the stream.

    Returns:
        Chat result.

    """
    generation = next(stream, None)
    if generation:
        generation += list(stream)
    if generation is None:
        msg = "No generations found in stream."
        raise ValueError(msg)
    return ChatResult(
        generations=[
            ChatGeneration(
                message=message_chunk_to_message(generation.message),
                generation_info=generation.generation_info,
            )
        ]
    )


async def agenerate_from_stream(
    stream: AsyncIterator[ChatGenerationChunk],
) -> ChatResult:
    """Async generate from a stream.

    Args:
        stream: AsyncIterator of `ChatGenerationChunk`.

    Returns:
        Chat result.

    """
    chunks = [chunk async for chunk in stream]
    return await run_in_executor(None, generate_from_stream, iter(chunks))


def _format_ls_structured_output(
    ls_structured_output_format: dict[str, Any] | None,
) -> dict[str, Any]:
    if ls_structured_output_format:
        try:
            ls_structured_output_format_dict = {
                "ls_structured_output_format": {
                    "kwargs": ls_structured_output_format.get("kwargs", {}),
                    "schema": convert_to_json_schema(
                        ls_structured_output_format["schema"]
                    ),
                }
            }
        except ValueError:
            ls_structured_output_format_dict = {}
    else:
        ls_structured_output_format_dict = {}

    return ls_structured_output_format_dict


class BaseChatModel(BaseLanguageModel[AIMessage], ABC):
    r"""Base class for chat models.

    Key imperative methods:
        Methods that actually call the underlying model.

        This table provides a brief overview of the main imperative methods. Please see the base `Runnable` reference for full documentation.

        | Method                 | Input                                                        | Output                                                     | Description                                                                      |
        | ---------------------- | ------------------------------------------------------------ | ---------------------------------------------------------- | -------------------------------------------------------------------------------- |
        | `invoke`               | `str` \| `list[dict | tuple | BaseMessage]` \| `PromptValue` | `BaseMessage`                                              | A single chat model call.                                                        |
        | `ainvoke`              | `'''`                                                        | `BaseMessage`                                              | Defaults to running `invoke` in an async executor.                               |
        | `stream`               | `'''`                                                        | `Iterator[BaseMessageChunk]`                               | Defaults to yielding output of `invoke`.                                         |
        | `astream`              | `'''`                                                        | `AsyncIterator[BaseMessageChunk]`                          | Defaults to yielding output of `ainvoke`.                                        |
        | `astream_events`       | `'''`                                                        | `AsyncIterator[StreamEvent]`                               | Event types: `on_chat_model_start`, `on_chat_model_stream`, `on_chat_model_end`. |
        | `batch`                | `list[''']`                                                  | `list[BaseMessage]`                                        | Defaults to running `invoke` in concurrent threads.                              |
        | `abatch`               | `list[''']`                                                  | `list[BaseMessage]`                                        | Defaults to running `ainvoke` in concurrent threads.                             |
        | `batch_as_completed`   | `list[''']`                                                  | `Iterator[tuple[int, Union[BaseMessage, Exception]]]`      | Defaults to running `invoke` in concurrent threads.                              |
        | `abatch_as_completed`  | `list[''']`                                                  | `AsyncIterator[tuple[int, Union[BaseMessage, Exception]]]` | Defaults to running `ainvoke` in concurrent threads.                             |

    Key declarative methods:
        Methods for creating another `Runnable` using the chat model.

        This table provides a brief overview of the main declarative methods. Please see the reference for each method for full documentation.

        | Method                       | Description                                                                                |
        | ---------------------------- | ------------------------------------------------------------------------------------------ |
        | `bind_tools`                 | Create chat model that can call tools.                                                     |
        | `with_structured_output`     | Create wrapper that structures model output using schema.                                  |
        | `with_retry`                 | Create wrapper that retries model calls on failure.                                        |
        | `with_fallbacks`             | Create wrapper that falls back to other models on failure.                                 |
        | `configurable_fields`        | Specify init args of the model that can be configured at runtime via the `RunnableConfig`. |
        | `configurable_alternatives`  | Specify alternative models which can be swapped in at runtime via the `RunnableConfig`.    |

    Creating custom chat model:
        Custom chat model implementations should inherit from this class.
        Please reference the table below for information about which
        methods and properties are required or optional for implementations.

        | Method/Property                  | Description                                                        | Required          |
        | -------------------------------- | ------------------------------------------------------------------ | ----------------- |
        | `_generate`                      | Use to generate a chat result from a prompt                        | Required          |
        | `_llm_type` (property)           | Used to uniquely identify the type of the model. Used for logging. | Required          |
        | `_identifying_params` (property) | Represent model parameterization for tracing purposes.             | Optional          |
        | `_stream`                        | Use to implement streaming                                         | Optional          |
        | `_agenerate`                     | Use to implement a native async method                             | Optional          |
        | `_astream`                       | Use to implement async version of `_stream`                        | Optional          |

    """  # noqa: E501

    rate_limiter: BaseRateLimiter | None = Field(default=None, exclude=True)
    "An optional rate limiter to use for limiting the number of requests."

    disable_streaming: bool | Literal["tool_calling"] = False
    """Whether to disable streaming for this model.

    If streaming is bypassed, then `stream`/`astream`/`astream_events` will
    defer to `invoke`/`ainvoke`.

    - If `True`, will always bypass streaming case.
    - If `'tool_calling'`, will bypass streaming case only when the model is called
        with a `tools` keyword argument. In other words, LangChain will automatically
        switch to non-streaming behavior (`invoke`) only when the tools argument is
        provided. This offers the best of both worlds.
    - If `False` (Default), will always use streaming case if available.

    The main reason for this flag is that code might be written using `stream` and
    a user may want to swap out a given model for another model whose implementation
    does not properly support streaming.
    """

    output_version: str | None = Field(
        default_factory=from_env("LC_OUTPUT_VERSION", default=None)
    )
    """Version of `AIMessage` output format to store in message content.

    `AIMessage.content_blocks` will lazily parse the contents of `content` into a
    standard format. This flag can be used to additionally store the standard format
    in message content, e.g., for serialization purposes.

    Supported values:

    - `'v0'`: provider-specific format in content (can lazily-parse with
        `content_blocks`)
    - `'v1'`: standardized format in content (consistent with `content_blocks`)

    Partner packages (e.g.,
    [`langchain-openai`](https://pypi.org/project/langchain-openai)) can also use this
    field to roll out new content formats in a backward-compatible way.

    !!! version-added "Added in `langchain-core` 1.0.0"

    """

    profile: ModelProfile | None = Field(default=None, exclude=True)
    """Profile detailing model capabilities.

    !!! warning "Beta feature"

        This is a beta feature. The format of model profiles is subject to change.

    If not specified, automatically loaded from the provider package on initialization
    if data is available.

    Example profile data includes context window sizes, supported modalities, or support
    for tool calling, structured output, and other features.

    !!! version-added "Added in `langchain-core` 1.1.0"
    """

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )

    def _resolve_model_profile(self) -> ModelProfile | None:
        """Return the default model profile, or `None` if unavailable.

        Override this in subclasses instead of `_set_model_profile`. The base
        validator calls it automatically and handles assignment. This avoids
        coupling partner code to Pydantic validator mechanics.

        Each partner needs its own override because things can vary per-partner,
        such as the attribute that identifies the model (e.g., `model`,
        `model_name`, `model_id`, `deployment_name`) and the partner-local
        `_get_default_model_profile` function that reads from each partner's own
        profile data.
        """
        # TODO: consider adding a `_model_identifier` property on BaseChatModel
        # to standardize how partners identify their model, which could allow a
        # default implementation here that calls a shared
        # profile-loading mechanism.
        return None

    @model_validator(mode="after")
    def _set_model_profile(self) -> Self:
        """Populate `profile` from `_resolve_model_profile` if not provided.

        Partners should override `_resolve_model_profile` rather than this
        validator. Overriding this with a new `@model_validator` replaces the
        base validator (Pydantic v2 behavior), bypassing the standard resolution
        path. A plain method override does not prevent the base validator from
        running.
        """
        if self.profile is None:
            # Suppress errors from partner overrides (e.g., missing profile
            # files, broken imports) so model construction never fails over an
            # optional field.
            with contextlib.suppress(Exception):
                self.profile = self._resolve_model_profile()
        return self

    # NOTE: _check_profile_keys must be defined AFTER _set_model_profile.
    # Pydantic v2 runs mode="after" validators in definition order.
    @model_validator(mode="after")
    def _check_profile_keys(self) -> Self:
        """Warn on unrecognized profile keys."""
        # isinstance guard: ModelProfile is a TypedDict (always a dict), but
        # protects against unexpected types from partner overrides.
        if self.profile and isinstance(self.profile, dict):
            _warn_unknown_profile_keys(self.profile)
        return self

    @cached_property
    def _serialized(self) -> builtins.dict[str, Any]:
        # self is always a Serializable object in this case, thus the result is
        # guaranteed to be a dict since dumpd uses the default callback, which uses
        # obj.to_json which always returns TypedDict subclasses
        return cast("builtins.dict[str, Any]", dumpd(self))

    # --- Runnable methods ---

    @property
    @override
    def OutputType(self) -> Any:
        """Get the output type for this `Runnable`."""
        return AnyMessage

    def _convert_input(self, model_input: LanguageModelInput) -> PromptValue:
        if isinstance(model_input, PromptValue):
            return model_input
        if isinstance(model_input, str):
            return StringPromptValue(text=model_input)
        if isinstance(model_input, Sequence):
            return ChatPromptValue(messages=convert_to_messages(model_input))
        msg = (  # type: ignore[unreachable]
            f"Invalid input type {type(model_input)}. "
            "Must be a PromptValue, str, or list of BaseMessages."
        )
        raise ValueError(msg)

    @override
    def invoke(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> AIMessage:
        config = ensure_config(config)
        return cast(
            "AIMessage",
            cast(
                "ChatGeneration",
                self.generate_prompt(
                    [self._convert_input(input)],
                    stop=stop,
                    callbacks=config.get("callbacks"),
                    tags=config.get("tags"),
                    metadata=config.get("metadata"),
                    run_name=config.get("run_name"),
                    run_id=config.pop("run_id", None),
                    **kwargs,
                ).generations[0][0],
            ).message,
        )

    @override
    async def ainvoke(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> AIMessage:
        config = ensure_config(config)
        llm_result = await self.agenerate_prompt(
            [self._convert_input(input)],
            stop=stop,
            callbacks=config.get("callbacks"),
            tags=config.get("tags"),
            metadata=config.get("metadata"),
            run_name=config.get("run_name"),
            run_id=config.pop("run_id", None),
            **kwargs,
        )
        return cast(
            "AIMessage", cast("ChatGeneration", llm_result.generations[0][0]).message
        )

    def _streaming_disabled(self, **kwargs: Any) -> bool:
        """Return whether streaming is hard-disabled for this call.

        Shared opt-outs honored by both `_should_stream` and
        `_should_use_protocol_streaming` — these override any affirmative trigger
        (attached handler, `stream=True`, etc.):

        - `self.disable_streaming is True`
        - `self.disable_streaming == "tool_calling"` with `tools` passed
        - `stream=<falsy>` in call kwargs
        - `self.streaming is False` on the instance
        """
        if self.disable_streaming is True:
            return True
        # We assume tools are passed in via "tools" kwarg in all models.
        if self.disable_streaming == "tool_calling" and kwargs.get("tools"):
            return True
        if "stream" in kwargs and not kwargs["stream"]:
            return True
        return (
            "streaming" in self.model_fields_set
            and getattr(self, "streaming", None) is False
        )

    def _should_stream(
        self,
        *,
        async_api: bool,
        run_manager: CallbackManagerForLLMRun
        | AsyncCallbackManagerForLLMRun
        | None = None,
        **kwargs: Any,
    ) -> bool:
        """Determine if a given model call should hit the streaming API."""
        sync_not_implemented = type(self)._stream == BaseChatModel._stream  # noqa: SLF001
        async_not_implemented = type(self)._astream == BaseChatModel._astream  # noqa: SLF001

        # Check if streaming is implemented.
        if (not async_api) and sync_not_implemented:
            return False
        # Note, since async falls back to sync we check both here.
        if async_api and async_not_implemented and sync_not_implemented:
            return False

        if self._streaming_disabled(**kwargs):
            return False

        # Affirmative: explicit `stream=<truthy>` kwarg.
        if kwargs.get("stream"):
            return True

        # Affirmative: instance-level `streaming=True` attribute.
        if (
            "streaming" in self.model_fields_set
            and getattr(self, "streaming", None) is True
        ):
            return True

        # Affirmative: a v1 streaming callback handler is attached.
        handlers = run_manager.handlers if run_manager else []
        return any(isinstance(h, _StreamingCallbackHandler) for h in handlers)

    def _should_use_protocol_streaming(
        self,
        *,
        async_api: bool,
        run_manager: CallbackManagerForLLMRun
        | AsyncCallbackManagerForLLMRun
        | None = None,
        **kwargs: Any,
    ) -> bool:
        """Determine whether an invoke should route through the v2 event path.

        Runs alongside `_should_stream` inside `_generate_with_cache` /
        `_agenerate_with_cache` — after the run manager is open — and
        wins over the v1 streaming branch when a handler has declared
        itself a `_V2StreamingCallbackHandler`. Parallel to
        `_should_stream` rather than a delegation — v1 and v2 have
        disjoint affirmative triggers.

        Args:
            async_api: Whether the caller is on the async path.
            run_manager: The active LLM run manager.
            **kwargs: Call kwargs; inspected for `disable_streaming`
                semantics and an explicit `stream=False` override.

        Returns:
            `True` if any attached handler inherits
            `_V2StreamingCallbackHandler` and the model can drive the v2
            event generator (natively or via the `_stream` compat
            bridge).
        """
        # Opt-in: only route through v2 when a v2 handler is attached.
        handlers = run_manager.handlers if run_manager else []
        if not any(isinstance(h, _V2StreamingCallbackHandler) for h in handlers):
            return False

        # Need a source of v2 events on the requested flavor. A native
        # `_(a)stream_chat_model_events` hook bypasses the bridge;
        # otherwise the bridge wraps `_stream` / `_astream`. Async can
        # fall back to sync.
        #
        # `cls._stream is not BaseChatModel._stream` is an identity
        # check for "subclass overrode `_stream`" — same pattern as
        # `_should_stream`.
        cls = type(self)
        has_native_sync = getattr(cls, "_stream_chat_model_events", None) is not None
        has_native_async = getattr(cls, "_astream_chat_model_events", None) is not None
        overrides_sync = cls._stream is not BaseChatModel._stream
        overrides_async = cls._astream is not BaseChatModel._astream
        has_sync_source = has_native_sync or overrides_sync
        has_async_source = has_native_async or overrides_async
        has_source = (
            (has_sync_source or has_async_source) if async_api else has_sync_source
        )
        if not has_source:
            return False

        return not self._streaming_disabled(**kwargs)

    def _iter_v2_events(
        self,
        messages: list[BaseMessage],
        *,
        run_manager: CallbackManagerForLLMRun,
        stream: ChatModelStream,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> Iterator[MessagesData]:
        """Drive the v2 event generator with per-event dispatch.

        Shared between the `stream_events(version="v3")` pump and the
        invoke-time v2 branch in `_generate_with_cache`. Picks the native
        `_stream_chat_model_events` hook when the subclass provides one,
        else bridges `_stream` chunks via `chunks_to_events`. Each event
        is dispatched into `stream` and fired as `on_stream_event` on
        the run manager. Run-lifecycle callbacks
        (`on_chat_model_start` / `on_llm_end` / `on_llm_error`) and
        rate-limiter acquisition are the caller's responsibility.

        Args:
            messages: Normalized input messages.
            run_manager: Active LLM run manager; receives
                `on_stream_event` per event.
            stream: Accumulator owned by the caller; receives each
                event via `stream.dispatch`.
            stop: Optional stop sequences.
            **kwargs: Forwarded to the event producer.

        Yields:
            Each protocol event produced by the model.
        """
        native = cast(
            "Callable[..., Iterator[MessagesData]] | None",
            getattr(self, "_stream_chat_model_events", None),
        )
        if native is not None:
            event_iter: Iterator[MessagesData] = native(
                messages, stop=stop, run_manager=run_manager, **kwargs
            )
        else:
            event_iter = chunks_to_events(
                self._stream(messages, stop=stop, run_manager=run_manager, **kwargs),
                message_id=stream.message_id,
            )
        for event in event_iter:
            stream.dispatch(event)
            run_manager.on_stream_event(event)
            yield event

    async def _aiter_v2_events(
        self,
        messages: list[BaseMessage],
        *,
        run_manager: AsyncCallbackManagerForLLMRun,
        stream: AsyncChatModelStream,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[MessagesData]:
        """Async counterpart to `_iter_v2_events`.

        See `_iter_v2_events` for the shared contract.
        """
        native = cast(
            "Callable[..., AsyncIterator[MessagesData]] | None",
            getattr(self, "_astream_chat_model_events", None),
        )
        if native is not None:
            event_iter: AsyncIterator[MessagesData] = native(
                messages, stop=stop, run_manager=run_manager, **kwargs
            )
        else:
            event_iter = achunks_to_events(
                self._astream(messages, stop=stop, run_manager=run_manager, **kwargs),
                message_id=stream.message_id,
            )
        async for event in event_iter:
            stream.dispatch(event)
            await run_manager.on_stream_event(event)
            yield event

    @override
    def stream(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> Iterator[AIMessageChunk]:
        if not self._should_stream(async_api=False, **{**kwargs, "stream": True}):
            # Model doesn't implement streaming, so use default implementation
            yield cast(
                "AIMessageChunk",
                self.invoke(input, config=config, stop=stop, **kwargs),
            )
        else:
            config = ensure_config(config)
            messages = self._convert_input(input).to_messages()
            ls_structured_output_format = kwargs.pop(
                "ls_structured_output_format", None
            ) or kwargs.pop("structured_output_format", None)
            ls_structured_output_format_dict = _format_ls_structured_output(
                ls_structured_output_format
            )

            params = self._get_invocation_params(stop=stop, **kwargs)
            options = {"stop": stop, **kwargs, **ls_structured_output_format_dict}
            inheritable_metadata = {
                **(config.get("metadata") or {}),
                **self._get_ls_params_with_defaults(stop=stop, **kwargs),
            }
            callback_manager = CallbackManager.configure(
                config.get("callbacks"),
                self.callbacks,
                self.verbose,
                config.get("tags"),
                self.tags,
                inheritable_metadata,
                self.metadata,
                langsmith_inheritable_metadata=_filter_invocation_params_for_tracing(
                    params
                ),
            )
            (run_manager,) = callback_manager.on_chat_model_start(
                self._serialized,
                [_format_for_tracing(messages)],
                invocation_params=params,
                options=options,
                name=config.get("run_name"),
                run_id=config.pop("run_id", None),
                batch_size=1,
            )

            chunks: list[ChatGenerationChunk] = []

            if self.rate_limiter:
                self.rate_limiter.acquire(blocking=True)

            try:
                input_messages = _normalize_messages(messages)
                run_id = "-".join((LC_ID_PREFIX, str(run_manager.run_id)))
                yielded = False
                index = -1
                index_type = ""
                for chunk in self._stream(input_messages, stop=stop, **kwargs):
                    if chunk.message.id is None:
                        chunk.message.id = run_id
                    chunk.message.response_metadata = _gen_info_and_msg_metadata(chunk)
                    if self.output_version == "v1":
                        # Overwrite .content with .content_blocks
                        chunk.message = _update_message_content_to_blocks(
                            chunk.message, "v1"
                        )
                        for block in cast(
                            "list[types.ContentBlock]", chunk.message.content
                        ):
                            if block["type"] != index_type:
                                index_type = block["type"]
                                index += 1
                            if "index" not in block:
                                block["index"] = index
                    run_manager.on_llm_new_token(chunk.message.content, chunk=chunk)
                    chunks.append(chunk)
                    yield cast("AIMessageChunk", chunk.message)
                    yielded = True

                # Yield a final empty chunk with chunk_position="last" if not yet
                # yielded
                if (
                    yielded
                    and isinstance(chunk.message, AIMessageChunk)
                    and not chunk.message.chunk_position
                ):
                    empty_content: str | list[str | dict[str, Any]] = (
                        "" if isinstance(chunk.message.content, str) else []
                    )
                    msg_chunk = AIMessageChunk(
                        content=empty_content, chunk_position="last", id=run_id
                    )
                    run_manager.on_llm_new_token(
                        "", chunk=ChatGenerationChunk(message=msg_chunk)
                    )
                    yield msg_chunk
            except BaseException as e:
                generations_with_error_metadata = _generate_response_from_error(e)
                chat_generation_chunk = merge_chat_generation_chunks(chunks)
                if chat_generation_chunk:
                    generations = [
                        [chat_generation_chunk],
                        generations_with_error_metadata,
                    ]
                else:
                    generations = [generations_with_error_metadata]
                run_manager.on_llm_error(
                    e,
                    response=LLMResult(generations=generations),
                )
                raise

            generation = merge_chat_generation_chunks(chunks)
            if generation is None:
                err = ValueError("No generation chunks were returned")
                run_manager.on_llm_error(err, response=LLMResult(generations=[]))
                raise err

            run_manager.on_llm_end(LLMResult(generations=[[generation]]))

    @override
    async def astream(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[AIMessageChunk]:
        if not self._should_stream(async_api=True, **{**kwargs, "stream": True}):
            # No async or sync stream is implemented, so fall back to ainvoke
            yield cast(
                "AIMessageChunk",
                await self.ainvoke(input, config=config, stop=stop, **kwargs),
            )
            return

        config = ensure_config(config)
        messages = self._convert_input(input).to_messages()

        ls_structured_output_format = kwargs.pop(
            "ls_structured_output_format", None
        ) or kwargs.pop("structured_output_format", None)
        ls_structured_output_format_dict = _format_ls_structured_output(
            ls_structured_output_format
        )

        params = self._get_invocation_params(stop=stop, **kwargs)
        options = {"stop": stop, **kwargs, **ls_structured_output_format_dict}
        inheritable_metadata = {
            **(config.get("metadata") or {}),
            **self._get_ls_params_with_defaults(stop=stop, **kwargs),
        }
        callback_manager = AsyncCallbackManager.configure(
            config.get("callbacks"),
            self.callbacks,
            self.verbose,
            config.get("tags"),
            self.tags,
            inheritable_metadata,
            self.metadata,
            langsmith_inheritable_metadata=_filter_invocation_params_for_tracing(
                params
            ),
        )
        (run_manager,) = await callback_manager.on_chat_model_start(
            self._serialized,
            [_format_for_tracing(messages)],
            invocation_params=params,
            options=options,
            name=config.get("run_name"),
            run_id=config.pop("run_id", None),
            batch_size=1,
        )

        if self.rate_limiter:
            await self.rate_limiter.aacquire(blocking=True)

        chunks: list[ChatGenerationChunk] = []

        try:
            input_messages = _normalize_messages(messages)
            run_id = "-".join((LC_ID_PREFIX, str(run_manager.run_id)))
            yielded = False
            index = -1
            index_type = ""
            async for chunk in self._astream(
                input_messages,
                stop=stop,
                **kwargs,
            ):
                if chunk.message.id is None:
                    chunk.message.id = run_id
                chunk.message.response_metadata = _gen_info_and_msg_metadata(chunk)
                if self.output_version == "v1":
                    # Overwrite .content with .content_blocks
                    chunk.message = _update_message_content_to_blocks(
                        chunk.message, "v1"
                    )
                    for block in cast(
                        "list[types.ContentBlock]", chunk.message.content
                    ):
                        if block["type"] != index_type:
                            index_type = block["type"]
                            index += 1
                        if "index" not in block:
                            block["index"] = index
                await run_manager.on_llm_new_token(chunk.message.content, chunk=chunk)
                chunks.append(chunk)
                yield cast("AIMessageChunk", chunk.message)
                yielded = True

            # Yield a final empty chunk with chunk_position="last" if not yet yielded
            if (
                yielded
                and isinstance(chunk.message, AIMessageChunk)
                and not chunk.message.chunk_position
            ):
                empty_content: str | list[str | dict[str, Any]] = (
                    "" if isinstance(chunk.message.content, str) else []
                )
                msg_chunk = AIMessageChunk(
                    content=empty_content, chunk_position="last", id=run_id
                )
                await run_manager.on_llm_new_token(
                    "", chunk=ChatGenerationChunk(message=msg_chunk)
                )
                yield msg_chunk
        except BaseException as e:
            generations_with_error_metadata = _generate_response_from_error(e)
            chat_generation_chunk = merge_chat_generation_chunks(chunks)
            if chat_generation_chunk:
                generations = [[chat_generation_chunk], generations_with_error_metadata]
            else:
                generations = [generations_with_error_metadata]
            await run_manager.on_llm_error(
                e,
                response=LLMResult(generations=generations),
            )
            raise

        generation = merge_chat_generation_chunks(chunks)
        if not generation:
            err = ValueError("No generation chunks were returned")
            await run_manager.on_llm_error(err, response=LLMResult(generations=[]))
            raise err

        await run_manager.on_llm_end(
            LLMResult(generations=[[generation]]),
        )

    # --- stream_events v3 ---

    @beta()
    def _chat_model_stream_v3(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> ChatModelStream:
        """Internal v3 sync streaming implementation.

        Public entry point: `stream_events(version='v3')`.
        """
        config = ensure_config(config)
        messages = self._convert_input(input).to_messages()
        input_messages = _normalize_messages(messages)

        # Strip tracing-only kwargs before forwarding to `_stream` — matches
        # `stream()` / `astream()`. Provider clients reject unknown kwargs,
        # so `.with_structured_output().stream_events(version="v3", ...)`
        # and any other binding that carries `ls_structured_output_format`
        # / `structured_output_format` would raise without this pop.
        ls_structured_output_format = kwargs.pop(
            "ls_structured_output_format", None
        ) or kwargs.pop("structured_output_format", None)
        ls_structured_output_format_dict = _format_ls_structured_output(
            ls_structured_output_format
        )

        params = self._get_invocation_params(stop=stop, **kwargs)
        options = {"stop": stop, **kwargs, **ls_structured_output_format_dict}
        inheritable_metadata = {
            **(config.get("metadata") or {}),
            **self._get_ls_params_with_defaults(stop=stop, **kwargs),
        }
        callback_manager = CallbackManager.configure(
            config.get("callbacks"),
            self.callbacks,
            self.verbose,
            config.get("tags"),
            self.tags,
            inheritable_metadata,
            self.metadata,
            langsmith_inheritable_metadata=_filter_invocation_params_for_tracing(
                params
            ),
        )
        stream = ChatModelStream()
        run_manager: CallbackManagerForLLMRun | None = None
        event_iter_ref: Iterator[MessagesData] | None = None
        rate_limiter_acquired = self.rate_limiter is None
        run_name = config.get("run_name")
        run_id = config.pop("run_id", None)

        def ensure_started() -> None:
            nonlocal event_iter_ref, run_manager
            if event_iter_ref is not None:
                return

            (run_manager,) = callback_manager.on_chat_model_start(
                self._serialized,
                [_format_for_tracing(messages)],
                invocation_params=params,
                options=options,
                name=run_name,
                run_id=run_id,
                batch_size=1,
            )
            stream.set_message_id("-".join((LC_ID_PREFIX, str(run_manager.run_id))))
            event_iter_ref = iter(
                self._iter_v2_events(
                    input_messages,
                    run_manager=run_manager,
                    stream=stream,
                    stop=stop,
                    **kwargs,
                )
            )

        def pump_one() -> bool:
            nonlocal rate_limiter_acquired
            ensure_started()
            if not rate_limiter_acquired:
                assert self.rate_limiter is not None  # noqa: S101
                self.rate_limiter.acquire(blocking=True)
                rate_limiter_acquired = True
            assert event_iter_ref is not None  # noqa: S101
            assert run_manager is not None  # noqa: S101
            try:
                next(event_iter_ref)
            except StopIteration:
                if not stream.done:
                    if stream.has_events:
                        # Native event producers may omit the terminal
                        # `message-finish`. Close the lifecycle here so
                        # `on_llm_end` still observes the assembled
                        # message. A truly empty stream remains an error
                        # for parity with `stream()`.
                        stream.dispatch(MessageFinishData(event="message-finish"))
                    else:
                        err = ValueError("No generation chunks were returned")
                        stream.fail(err)
                        run_manager.on_llm_error(
                            err,
                            response=LLMResult(generations=[]),
                        )
                        return False
                if stream.done and stream.output_message is not None:
                    run_manager.on_llm_end(
                        LLMResult(
                            generations=[
                                [ChatGeneration(message=stream.output_message)],
                            ],
                        ),
                    )
                return False
            except BaseException as exc:
                stream.fail(exc)
                run_manager.on_llm_error(
                    exc,
                    response=LLMResult(generations=[]),
                )
                return False
            if stream.done and stream.output_message is not None:
                run_manager.on_llm_end(
                    LLMResult(
                        generations=[
                            [ChatGeneration(message=stream.output_message)],
                        ],
                    ),
                )
            return True

        stream.set_start(ensure_started)
        stream.bind_pump(pump_one)
        return stream

    @beta()
    async def _achat_model_stream_v3(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> AsyncChatModelStream:
        """Internal v3 async streaming implementation.

        Public entry point: `astream_events(version='v3')`.
        """
        config = ensure_config(config)
        messages = self._convert_input(input).to_messages()
        input_messages = _normalize_messages(messages)

        # Strip tracing-only kwargs before forwarding — see the sync v3
        # implementation for the full rationale.
        ls_structured_output_format = kwargs.pop(
            "ls_structured_output_format", None
        ) or kwargs.pop("structured_output_format", None)
        ls_structured_output_format_dict = _format_ls_structured_output(
            ls_structured_output_format
        )

        params = self._get_invocation_params(stop=stop, **kwargs)
        options = {"stop": stop, **kwargs, **ls_structured_output_format_dict}
        inheritable_metadata = {
            **(config.get("metadata") or {}),
            **self._get_ls_params_with_defaults(stop=stop, **kwargs),
        }
        callback_manager = AsyncCallbackManager.configure(
            config.get("callbacks"),
            self.callbacks,
            self.verbose,
            config.get("tags"),
            self.tags,
            inheritable_metadata,
            self.metadata,
            langsmith_inheritable_metadata=_filter_invocation_params_for_tracing(
                params
            ),
        )
        stream = AsyncChatModelStream()
        run_manager: AsyncCallbackManagerForLLMRun | None = None
        run_name = config.get("run_name")
        run_id = config.pop("run_id", None)
        start_lock = asyncio.Lock()

        async def _produce() -> None:
            assert run_manager is not None  # noqa: S101
            try:
                if self.rate_limiter:
                    await self.rate_limiter.aacquire(blocking=True)

                async for _event in self._aiter_v2_events(
                    input_messages,
                    run_manager=run_manager,
                    stream=stream,
                    stop=stop,
                    **kwargs,
                ):
                    pass
                if not stream.done:
                    if stream.has_events:
                        # Native event producers may omit the terminal
                        # `message-finish`. Close the lifecycle here so
                        # `on_llm_end` sees the finalized message. A
                        # truly empty stream remains an error for parity
                        # with `astream()`.
                        stream.dispatch(MessageFinishData(event="message-finish"))
                    else:
                        err = ValueError("No generation chunks were returned")
                        stream.fail(err)
                        await run_manager.on_llm_error(
                            err,
                            response=LLMResult(generations=[]),
                        )
                        return
                if stream.done and stream.output_message is not None:
                    await run_manager.on_llm_end(
                        LLMResult(
                            generations=[
                                [ChatGeneration(message=stream.output_message)],
                            ],
                        ),
                    )
            except asyncio.CancelledError as exc:
                stream.fail(exc)
                # Close the callback lifecycle so tracing observes a
                # matching end event for the earlier `on_chat_model_start`.
                # `on_llm_error` is `@shielded`, so the callback runs to
                # completion in the background even though the `await`
                # here re-raises our cancellation.
                with contextlib.suppress(Exception):
                    await run_manager.on_llm_error(
                        exc,
                        response=LLMResult(generations=[]),
                    )
                raise
            except BaseException as exc:
                stream.fail(exc)
                await run_manager.on_llm_error(
                    exc,
                    response=LLMResult(generations=[]),
                )

        async def ensure_started() -> None:
            nonlocal run_manager
            if stream._producer_task is not None:  # noqa: SLF001
                return

            async with start_lock:
                if stream._producer_task is not None:  # noqa: SLF001
                    return  # type: ignore[unreachable]

                (run_manager,) = await callback_manager.on_chat_model_start(
                    self._serialized,
                    [_format_for_tracing(messages)],
                    invocation_params=params,
                    options=options,
                    name=run_name,
                    run_id=run_id,
                    batch_size=1,
                )
                stream.set_message_id("-".join((LC_ID_PREFIX, str(run_manager.run_id))))
                stream._producer_task = asyncio.get_running_loop().create_task(  # noqa: SLF001
                    _produce()
                )

        async def _on_aclose_fail(exc: BaseException) -> None:
            assert run_manager is not None  # noqa: S101
            # Invoked by `stream.aclose()` only when the producer was
            # cancelled before `_produce` ran — so `on_llm_error` from
            # the CancelledError handler never fired. Shielded by the
            # callback manager; runs to completion even if our caller
            # is being cancelled.
            await run_manager.on_llm_error(
                exc,
                response=LLMResult(generations=[]),
            )

        stream.set_start(ensure_started)
        stream._on_aclose_fail = _on_aclose_fail  # noqa: SLF001
        return stream

    @overload  # type: ignore[override]
    def stream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v1", "v2"] = "v2",
        **kwargs: Any,
    ) -> Iterator[StreamEvent]: ...

    @overload
    def stream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v3"],
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> ChatModelStream: ...

    def stream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v1", "v2", "v3"] = "v2",
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> Iterator[StreamEvent] | ChatModelStream:
        """Stream events from this chat model.

        For `version="v1"` / `"v2"`, yields `StreamEvent` dicts (see
        `Runnable.stream_events`). For `version="v3"`, returns a
        `ChatModelStream` exposing typed projections (`.text`,
        `.reasoning`, `.tool_calls`, `.output`).

        !!! warning "Beta"

            `version="v3"` is in beta. The protocol shape, return type,
            and surface area may change in future releases. Calling it
            emits a `LangChainBetaWarning` at runtime.

        !!! note "v3 always produces v1-shaped content"

            `ChatModelStream.output.content` is always a list of v1
            content blocks (text / reasoning / tool_call / image / …),
            regardless of the model's `output_version` attribute. The
            setting only affects the legacy `stream()` / `astream()` /
            `invoke()` paths. If you're mixing
            `stream_events(version="v3")` with those paths in the same
            pipeline and need a consistent output shape across them,
            set `output_version="v1"` on the model.

        Args:
            input: The model input.
            config: Optional runnable config.
            version: Streaming-event schema version. `"v3"` selects the
                content-block-centric streaming protocol.
            stop: Optional stop sequences. Only used for `version="v3"`;
                ignored otherwise.
            **kwargs: Additional keyword arguments. For `version="v3"`,
                forwarded to the model.

        Returns:
            For `version="v3"`, a `ChatModelStream` with typed
            projections. Otherwise an `Iterator[StreamEvent]`.
        """
        if version == "v3":
            return self._chat_model_stream_v3(input, config, stop=stop, **kwargs)
        return super().stream_events(
            input, config, version=version, stop=stop, **kwargs
        )

    @overload
    def astream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v1", "v2"] = "v2",
        **kwargs: Any,
    ) -> AsyncIterator[StreamEvent]: ...

    @overload
    def astream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v3"],
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> Awaitable[AsyncChatModelStream]: ...

    def astream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v1", "v2", "v3"] = "v2",
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[StreamEvent] | Awaitable[AsyncChatModelStream]:
        """Async variant of `stream_events`. See `stream_events` for full docs."""
        if version == "v3":
            return self._achat_model_stream_v3(input, config, stop=stop, **kwargs)
        # v1/v2: forward to Runnable.astream_events (async generator).
        return super().astream_events(
            input, config, version=version, stop=stop, **kwargs
        )

    # --- Custom methods ---

    def _combine_llm_outputs(
        self, _llm_outputs: list[builtins.dict[str, Any] | None], /
    ) -> builtins.dict[str, Any]:
        return {}

    def _convert_cached_generations(
        self, cache_val: list[Generation]
    ) -> list[ChatGeneration]:
        """Convert cached Generation objects to ChatGeneration objects.

        Handle case where cache contains Generation objects instead of
        ChatGeneration objects. This can happen due to serialization/deserialization
        issues or legacy cache data (see #22389).

        Args:
            cache_val: List of cached generation objects.

        Returns:
            List of ChatGeneration objects.

        """
        converted_generations = []
        for gen in cache_val:
            if isinstance(gen, Generation) and not isinstance(gen, ChatGeneration):
                # Convert Generation to ChatGeneration by creating AIMessage
                # from the text content
                chat_gen = ChatGeneration(
                    message=AIMessage(content=gen.text),
                    generation_info=gen.generation_info,
                )
                converted_generations.append(chat_gen)
            else:
                # Already a ChatGeneration or other expected type
                if hasattr(gen, "message") and isinstance(gen.message, AIMessage):
                    # We zero out cost on cache hits
                    gen.message = gen.message.model_copy(
                        update={
                            "usage_metadata": {
                                **(gen.message.usage_metadata or {}),
                                "total_cost": 0,
                            }
                        }
                    )
                converted_generations.append(gen)
        return converted_generations

    def _replay_v2_events_for_cache_hit(
        self,
        generations: list[ChatGeneration],
        *,
        run_manager: CallbackManagerForLLMRun | None,
        **kwargs: Any,
    ) -> None:
        """Replay cached messages as v2 events when a v2 handler is attached.

        A warm cache must produce the same `on_stream_event` stream as a
        cold call so LangGraph-style consumers do not observe behavior
        that depends on cache state. Gated by
        `_should_use_protocol_streaming` so a `disable_streaming` config
        that suppresses v2 on cold calls also suppresses it here.
        """
        if run_manager is None or not self._should_use_protocol_streaming(
            async_api=False, run_manager=run_manager, **kwargs
        ):
            return
        message_id = f"{LC_ID_PREFIX}-{run_manager.run_id}"
        for gen in generations:
            msg = getattr(gen, "message", None)
            if not isinstance(msg, AIMessage):
                continue
            for event in message_to_events(msg, message_id=message_id):
                run_manager.on_stream_event(event)

    async def _areplay_v2_events_for_cache_hit(
        self,
        generations: list[ChatGeneration],
        *,
        run_manager: AsyncCallbackManagerForLLMRun | None,
        **kwargs: Any,
    ) -> None:
        """Async counterpart to `_replay_v2_events_for_cache_hit`."""
        if run_manager is None or not self._should_use_protocol_streaming(
            async_api=True, run_manager=run_manager, **kwargs
        ):
            return
        message_id = f"{LC_ID_PREFIX}-{run_manager.run_id}"
        for gen in generations:
            msg = getattr(gen, "message", None)
            if not isinstance(msg, AIMessage):
                continue
            async for event in amessage_to_events(msg, message_id=message_id):
                await run_manager.on_stream_event(event)

    def _get_invocation_params(
        self,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> builtins.dict[str, Any]:
        params = self._dict_for_compat()
        params["stop"] = stop
        return {**params, **kwargs}

    def _get_ls_params(
        self,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> LangSmithParams:
        """Get standard params for LangSmith tracing.

        Subclasses **should override** this method to populate `ls_provider`
        and `ls_model_name` from provider-specific attributes (e.g. `self.model`,
        `self.model_name`, `self.model_id`) and to honor per-call overrides
        passed via `kwargs["model"]` so that runtime `bind`/`invoke` model
        changes are reflected in traces.

        The implementation here is a best-effort fallback for subclasses that
        do not override it. It is not part of a stable contract and the
        derivation rules may change:

        - `ls_provider` is derived from the class name by stripping a leading
            or trailing `"Chat"` and lowercasing the remainder. This produces
            ugly values for multi-word providers (e.g. `ChatGoogleGenerativeAI`
            would become `"googlegenerativeai"`).

            Override to set a stable, conventional value
            such as `"google_genai"`.
        - `ls_model_name` is resolved from `kwargs["model"]`, then
            `self.model`, then `self.model_name`.

            Subclasses whose model attribute has a different name
            (`model_id`, `deployment_name`, ...) must override.
        """
        # get default provider from class name
        default_provider = self.__class__.__name__
        if default_provider.startswith("Chat"):
            default_provider = default_provider[4:].lower()
        elif default_provider.endswith("Chat"):
            default_provider = default_provider[:-4]
        default_provider = default_provider.lower()

        ls_params = LangSmithParams(ls_provider=default_provider, ls_model_type="chat")
        if stop:
            ls_params["ls_stop"] = stop

        # model
        if "model" in kwargs and isinstance(kwargs["model"], str):
            ls_params["ls_model_name"] = kwargs["model"]
        elif hasattr(self, "model") and isinstance(self.model, str):
            ls_params["ls_model_name"] = self.model
        elif hasattr(self, "model_name") and isinstance(self.model_name, str):
            ls_params["ls_model_name"] = self.model_name

        # temperature
        if "temperature" in kwargs and isinstance(kwargs["temperature"], (int, float)):
            ls_params["ls_temperature"] = kwargs["temperature"]
        elif hasattr(self, "temperature") and isinstance(
            self.temperature, (int, float)
        ):
            ls_params["ls_temperature"] = self.temperature

        # max_tokens
        if "max_tokens" in kwargs and isinstance(kwargs["max_tokens"], int):
            ls_params["ls_max_tokens"] = kwargs["max_tokens"]
        elif hasattr(self, "max_tokens") and isinstance(self.max_tokens, int):
            ls_params["ls_max_tokens"] = self.max_tokens

        return ls_params

    def _get_ls_params_with_defaults(
        self,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> LangSmithParams:
        """Wrap _get_ls_params to always include ls_integration."""
        ls_params = self._get_ls_params(stop=stop, **kwargs)
        ls_params["ls_integration"] = "langchain_chat_model"
        return ls_params

    def _get_llm_string(self, stop: list[str] | None = None, **kwargs: Any) -> str:
        if self.is_lc_serializable():
            params = {**kwargs, "stop": stop}
            param_string = str(sorted(params.items()))
            # This code is not super efficient as it goes back and forth between
            # json and dict.
            serialized_repr = self._serialized
            _cleanup_llm_representation(serialized_repr, 1)
            llm_string = json.dumps(serialized_repr, sort_keys=True)
            return llm_string + "---" + param_string
        params = self._get_invocation_params(stop=stop, **kwargs)
        params = {**params, **kwargs}
        return str(sorted(params.items()))

    def generate(
        self,
        messages: list[list[BaseMessage]],
        stop: list[str] | None = None,
        callbacks: Callbacks = None,
        *,
        tags: list[str] | None = None,
        metadata: builtins.dict[str, Any] | None = None,
        run_name: str | None = None,
        run_id: uuid.UUID | None = None,
        **kwargs: Any,
    ) -> LLMResult:
        """Pass a sequence of prompts to the model and return model generations.

        This method should make use of batched calls for models that expose a batched
        API.

        Use this method when you want to:

        1. Take advantage of batched calls,
        2. Need more output from the model than just the top generated value,
        3. Are building chains that are agnostic to the underlying language model
            type (e.g., pure text completion models vs chat models).

        Args:
            messages: List of list of messages.
            stop: Stop words to use when generating.

                Model output is cut off at the first occurrence of any of these
                substrings.
            callbacks: `Callbacks` to pass through.

                Used for executing additional functionality, such as logging or
                streaming, throughout generation.
            tags: The tags to apply.
            metadata: The metadata to apply.
            run_name: The name of the run.
            run_id: The ID of the run.
            **kwargs: Arbitrary additional keyword arguments.

                These are usually passed to the model provider API call.

        Returns:
            An `LLMResult`, which contains a list of candidate `Generations` for each
                input prompt and additional model provider-specific output.

        """
        ls_structured_output_format = kwargs.pop(
            "ls_structured_output_format", None
        ) or kwargs.pop("structured_output_format", None)
        ls_structured_output_format_dict = _format_ls_structured_output(
            ls_structured_output_format
        )

        params = self._get_invocation_params(stop=stop, **kwargs)
        options = {"stop": stop, **ls_structured_output_format_dict}
        inheritable_metadata = {
            **(metadata or {}),
            **self._get_ls_params_with_defaults(stop=stop, **kwargs),
        }

        callback_manager = CallbackManager.configure(
            callbacks,
            self.callbacks,
            self.verbose,
            tags,
            self.tags,
            inheritable_metadata,
            self.metadata,
            langsmith_inheritable_metadata=_filter_invocation_params_for_tracing(
                params
            ),
        )
        messages_to_trace = [
            _format_for_tracing(message_list) for message_list in messages
        ]
        run_managers = callback_manager.on_chat_model_start(
            self._serialized,
            messages_to_trace,
            invocation_params=params,
            options=options,
            name=run_name,
            run_id=run_id,
            batch_size=len(messages),
        )
        results = []
        input_messages = [
            _normalize_messages(message_list) for message_list in messages
        ]
        for i, m in enumerate(input_messages):
            try:
                results.append(
                    self._generate_with_cache(
                        m,
                        stop=stop,
                        run_manager=run_managers[i] if run_managers else None,
                        **kwargs,
                    )
                )
            except BaseException as e:
                if run_managers:
                    generations_with_error_metadata = _generate_response_from_error(e)
                    run_managers[i].on_llm_error(
                        e,
                        response=LLMResult(
                            generations=[generations_with_error_metadata]
                        ),
                    )
                raise
        flattened_outputs = [
            LLMResult(generations=[res.generations], llm_output=res.llm_output)
            for res in results
        ]
        llm_output = self._combine_llm_outputs([res.llm_output for res in results])
        generations = [res.generations for res in results]
        output = LLMResult(generations=generations, llm_output=llm_output)
        if run_managers:
            run_infos = []
            for manager, flattened_output in zip(
                run_managers, flattened_outputs, strict=False
            ):
                manager.on_llm_end(flattened_output)
                run_infos.append(RunInfo(run_id=manager.run_id))
            output.run = run_infos
        return output

    async def agenerate(
        self,
        messages: list[list[BaseMessage]],
        stop: list[str] | None = None,
        callbacks: Callbacks = None,
        *,
        tags: list[str] | None = None,
        metadata: builtins.dict[str, Any] | None = None,
        run_name: str | None = None,
        run_id: uuid.UUID | None = None,
        **kwargs: Any,
    ) -> LLMResult:
        """Asynchronously pass a sequence of prompts to a model and return generations.

        This method should make use of batched calls for models that expose a batched
        API.

        Use this method when you want to:

        1. Take advantage of batched calls,
        2. Need more output from the model than just the top generated value,
        3. Are building chains that are agnostic to the underlying language model
            type (e.g., pure text completion models vs chat models).

        Args:
            messages: List of list of messages.
            stop: Stop words to use when generating.

                Model output is cut off at the first occurrence of any of these
                substrings.
            callbacks: `Callbacks` to pass through.

                Used for executing additional functionality, such as logging or
                streaming, throughout generation.
            tags: The tags to apply.
            metadata: The metadata to apply.
            run_name: The name of the run.
            run_id: The ID of the run.
            **kwargs: Arbitrary additional keyword arguments.

                These are usually passed to the model provider API call.

        Returns:
            An `LLMResult`, which contains a list of candidate `Generations` for each
                input prompt and additional model provider-specific output.

        """
        ls_structured_output_format = kwargs.pop(
            "ls_structured_output_format", None
        ) or kwargs.pop("structured_output_format", None)
        ls_structured_output_format_dict = _format_ls_structured_output(
            ls_structured_output_format
        )

        params = self._get_invocation_params(stop=stop, **kwargs)
        options = {"stop": stop, **ls_structured_output_format_dict}
        inheritable_metadata = {
            **(metadata or {}),
            **self._get_ls_params_with_defaults(stop=stop, **kwargs),
        }

        callback_manager = AsyncCallbackManager.configure(
            callbacks,
            self.callbacks,
            self.verbose,
            tags,
            self.tags,
            inheritable_metadata,
            self.metadata,
            langsmith_inheritable_metadata=_filter_invocation_params_for_tracing(
                params
            ),
        )

        messages_to_trace = [
            _format_for_tracing(message_list) for message_list in messages
        ]
        run_managers = await callback_manager.on_chat_model_start(
            self._serialized,
            messages_to_trace,
            invocation_params=params,
            options=options,
            name=run_name,
            batch_size=len(messages),
            run_id=run_id,
        )

        input_messages = [
            _normalize_messages(message_list) for message_list in messages
        ]
        results = await asyncio.gather(
            *[
                self._agenerate_with_cache(
                    m,
                    stop=stop,
                    run_manager=run_managers[i] if run_managers else None,
                    **kwargs,
                )
                for i, m in enumerate(input_messages)
            ],
            return_exceptions=True,
        )
        exceptions = []
        for i, res in enumerate(results):
            if isinstance(res, BaseException):
                if run_managers:
                    generations_with_error_metadata = _generate_response_from_error(res)
                    await run_managers[i].on_llm_error(
                        res,
                        response=LLMResult(
                            generations=[generations_with_error_metadata]
                        ),
                    )
                exceptions.append(res)
        if exceptions:
            if run_managers:
                await asyncio.gather(
                    *[
                        run_manager.on_llm_end(
                            LLMResult(
                                generations=[res.generations],  # type: ignore[union-attr]
                                llm_output=res.llm_output,  # type: ignore[union-attr]
                            )
                        )
                        for run_manager, res in zip(run_managers, results, strict=False)
                        if not isinstance(res, Exception)
                    ]
                )
            raise exceptions[0]
        flattened_outputs = [
            LLMResult(generations=[res.generations], llm_output=res.llm_output)  # type: ignore[union-attr]
            for res in results
        ]
        llm_output = self._combine_llm_outputs([res.llm_output for res in results])  # type: ignore[union-attr]
        generations = [res.generations for res in results]  # type: ignore[union-attr]
        output = LLMResult(generations=generations, llm_output=llm_output)
        await asyncio.gather(
            *[
                run_manager.on_llm_end(flattened_output)
                for run_manager, flattened_output in zip(
                    run_managers, flattened_outputs, strict=False
                )
            ]
        )
        if run_managers:
            output.run = [
                RunInfo(run_id=run_manager.run_id) for run_manager in run_managers
            ]
        return output

    @override
    def generate_prompt(
        self,
        prompts: list[PromptValue],
        stop: list[str] | None = None,
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> LLMResult:
        prompt_messages = [p.to_messages() for p in prompts]
        return self.generate(prompt_messages, stop=stop, callbacks=callbacks, **kwargs)

    @override
    async def agenerate_prompt(
        self,
        prompts: list[PromptValue],
        stop: list[str] | None = None,
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> LLMResult:
        prompt_messages = [p.to_messages() for p in prompts]
        return await self.agenerate(
            prompt_messages, stop=stop, callbacks=callbacks, **kwargs
        )

    def _generate_with_cache(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: CallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> ChatResult:
        llm_cache = self.cache if isinstance(self.cache, BaseCache) else get_llm_cache()
        # We should check the cache unless it's explicitly set to False
        # A None cache means we should use the default global cache
        # if it's configured.
        check_cache = self.cache or self.cache is None
        if check_cache:
            if llm_cache:
                llm_string = self._get_llm_string(stop=stop, **kwargs)
                normalized_messages = [
                    (
                        msg.model_copy(update={"id": None})
                        if getattr(msg, "id", None) is not None
                        else msg
                    )
                    for msg in messages
                ]
                prompt = dumps(normalized_messages)
                cache_val = llm_cache.lookup(prompt, llm_string)
                if isinstance(cache_val, list):
                    converted_generations = self._convert_cached_generations(cache_val)
                    self._replay_v2_events_for_cache_hit(
                        converted_generations,
                        run_manager=run_manager,
                        **kwargs,
                    )
                    return ChatResult(generations=converted_generations)
            elif self.cache is None:
                pass
            else:
                msg = "Asked to cache, but no cache found at `langchain.cache`."
                raise ValueError(msg)

        # Apply the rate limiter after checking the cache, since
        # we usually don't want to rate limit cache lookups, but
        # we do want to rate limit API requests.
        if self.rate_limiter:
            self.rate_limiter.acquire(blocking=True)

        # v2 streaming: preferred over v1 when any attached handler opts in via
        # `_V2StreamingCallbackHandler`. Drives the protocol event generator
        # (native or `_stream` compat bridge) through the shared helper so
        # `on_stream_event` fires per event, then returns a normal `ChatResult`
        # so caching / `on_llm_end` stay on the existing generate path.
        if self._should_use_protocol_streaming(
            async_api=False,
            run_manager=run_manager,
            **kwargs,
        ):
            stream_accum = ChatModelStream(
                message_id=(
                    f"{LC_ID_PREFIX}-{run_manager.run_id}" if run_manager else None
                )
            )
            assert run_manager is not None  # noqa: S101
            for _event in self._iter_v2_events(
                messages,
                run_manager=run_manager,
                stream=stream_accum,
                stop=stop,
                **kwargs,
            ):
                pass
            if stream_accum.output_message is None:
                msg = "v2 stream finished without producing a message"
                raise RuntimeError(msg)
            result = ChatResult(
                generations=[ChatGeneration(message=stream_accum.output_message)]
            )
        # If stream is not explicitly set, check if implicitly requested by
        # astream_events() or astream_log(). Bail out if _stream not implemented
        elif self._should_stream(
            async_api=False,
            run_manager=run_manager,
            **kwargs,
        ):
            chunks: list[ChatGenerationChunk] = []
            run_id: str | None = (
                f"{LC_ID_PREFIX}-{run_manager.run_id}" if run_manager else None
            )
            yielded = False
            index = -1
            index_type = ""
            for chunk in self._stream(messages, stop=stop, **kwargs):
                chunk.message.response_metadata = _gen_info_and_msg_metadata(chunk)
                if self.output_version == "v1":
                    # Overwrite .content with .content_blocks
                    chunk.message = _update_message_content_to_blocks(
                        chunk.message, "v1"
                    )
                    for block in cast(
                        "list[types.ContentBlock]", chunk.message.content
                    ):
                        if block["type"] != index_type:
                            index_type = block["type"]
                            index += 1
                        if "index" not in block:
                            block["index"] = index
                if run_manager:
                    if chunk.message.id is None:
                        chunk.message.id = run_id
                    run_manager.on_llm_new_token(chunk.message.content, chunk=chunk)
                chunks.append(chunk)
                yielded = True

            # Yield a final empty chunk with chunk_position="last" if not yet yielded
            if (
                yielded
                and isinstance(chunk.message, AIMessageChunk)
                and not chunk.message.chunk_position
            ):
                empty_content: str | list[str | dict[str, Any]] = (
                    "" if isinstance(chunk.message.content, str) else []
                )
                chunk = ChatGenerationChunk(
                    message=AIMessageChunk(
                        content=empty_content, chunk_position="last", id=run_id
                    )
                )
                if run_manager:
                    run_manager.on_llm_new_token("", chunk=chunk)
                chunks.append(chunk)
            result = generate_from_stream(iter(chunks))
        elif inspect.signature(self._generate).parameters.get("run_manager"):
            result = self._generate(
                messages, stop=stop, run_manager=run_manager, **kwargs
            )
        else:
            result = self._generate(messages, stop=stop, **kwargs)

        if self.output_version == "v1":
            # Overwrite .content with .content_blocks
            for generation in result.generations:
                generation.message = _update_message_content_to_blocks(
                    generation.message, "v1"
                )

        # Add response metadata to each generation
        for idx, generation in enumerate(result.generations):
            if run_manager and generation.message.id is None:
                generation.message.id = f"{LC_ID_PREFIX}-{run_manager.run_id}-{idx}"
            generation.message.response_metadata = _gen_info_and_msg_metadata(
                generation
            )
        if len(result.generations) == 1 and result.llm_output is not None:
            result.generations[0].message.response_metadata = {
                **result.llm_output,
                **result.generations[0].message.response_metadata,
            }
        if check_cache and llm_cache:
            llm_cache.update(prompt, llm_string, result.generations)
        return result

    async def _agenerate_with_cache(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: AsyncCallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> ChatResult:
        llm_cache = self.cache if isinstance(self.cache, BaseCache) else get_llm_cache()
        # We should check the cache unless it's explicitly set to False
        # A None cache means we should use the default global cache
        # if it's configured.
        check_cache = self.cache or self.cache is None
        if check_cache:
            if llm_cache:
                llm_string = self._get_llm_string(stop=stop, **kwargs)
                normalized_messages = [
                    (
                        msg.model_copy(update={"id": None})
                        if getattr(msg, "id", None) is not None
                        else msg
                    )
                    for msg in messages
                ]
                prompt = dumps(normalized_messages)
                cache_val = await llm_cache.alookup(prompt, llm_string)
                if isinstance(cache_val, list):
                    converted_generations = self._convert_cached_generations(cache_val)
                    await self._areplay_v2_events_for_cache_hit(
                        converted_generations,
                        run_manager=run_manager,
                        **kwargs,
                    )
                    return ChatResult(generations=converted_generations)
            elif self.cache is None:
                pass
            else:
                msg = "Asked to cache, but no cache found at `langchain.cache`."
                raise ValueError(msg)

        # Apply the rate limiter after checking the cache, since
        # we usually don't want to rate limit cache lookups, but
        # we do want to rate limit API requests.
        if self.rate_limiter:
            await self.rate_limiter.aacquire(blocking=True)

        # v2 streaming: see sync counterpart in `_generate_with_cache`.
        if self._should_use_protocol_streaming(
            async_api=True,
            run_manager=run_manager,
            **kwargs,
        ):
            stream_accum = AsyncChatModelStream(
                message_id=(
                    f"{LC_ID_PREFIX}-{run_manager.run_id}" if run_manager else None
                )
            )
            assert run_manager is not None  # noqa: S101
            async for _event in self._aiter_v2_events(
                messages,
                run_manager=run_manager,
                stream=stream_accum,
                stop=stop,
                **kwargs,
            ):
                pass
            if stream_accum.output_message is None:
                msg = "v2 stream finished without producing a message"
                raise RuntimeError(msg)
            result = ChatResult(
                generations=[ChatGeneration(message=stream_accum.output_message)]
            )
        # If stream is not explicitly set, check if implicitly requested by
        # astream_events() or astream_log(). Bail out if _astream not implemented
        elif self._should_stream(
            async_api=True,
            run_manager=run_manager,
            **kwargs,
        ):
            chunks: list[ChatGenerationChunk] = []
            run_id: str | None = (
                f"{LC_ID_PREFIX}-{run_manager.run_id}" if run_manager else None
            )
            yielded = False
            index = -1
            index_type = ""
            async for chunk in self._astream(messages, stop=stop, **kwargs):
                chunk.message.response_metadata = _gen_info_and_msg_metadata(chunk)
                if self.output_version == "v1":
                    # Overwrite .content with .content_blocks
                    chunk.message = _update_message_content_to_blocks(
                        chunk.message, "v1"
                    )
                    for block in cast(
                        "list[types.ContentBlock]", chunk.message.content
                    ):
                        if block["type"] != index_type:
                            index_type = block["type"]
                            index += 1
                        if "index" not in block:
                            block["index"] = index
                if run_manager:
                    if chunk.message.id is None:
                        chunk.message.id = run_id
                    await run_manager.on_llm_new_token(
                        chunk.message.content, chunk=chunk
                    )
                chunks.append(chunk)
                yielded = True

            # Yield a final empty chunk with chunk_position="last" if not yet yielded
            if (
                yielded
                and isinstance(chunk.message, AIMessageChunk)
                and not chunk.message.chunk_position
            ):
                empty_content: str | list[str | dict[str, Any]] = (
                    "" if isinstance(chunk.message.content, str) else []
                )
                chunk = ChatGenerationChunk(
                    message=AIMessageChunk(
                        content=empty_content, chunk_position="last", id=run_id
                    )
                )
                if run_manager:
                    await run_manager.on_llm_new_token("", chunk=chunk)
                chunks.append(chunk)
            result = generate_from_stream(iter(chunks))
        elif inspect.signature(self._agenerate).parameters.get("run_manager"):
            result = await self._agenerate(
                messages, stop=stop, run_manager=run_manager, **kwargs
            )
        else:
            result = await self._agenerate(messages, stop=stop, **kwargs)

        if self.output_version == "v1":
            # Overwrite .content with .content_blocks
            for generation in result.generations:
                generation.message = _update_message_content_to_blocks(
                    generation.message, "v1"
                )

        # Add response metadata to each generation
        for idx, generation in enumerate(result.generations):
            if run_manager and generation.message.id is None:
                generation.message.id = f"{LC_ID_PREFIX}-{run_manager.run_id}-{idx}"
            generation.message.response_metadata = _gen_info_and_msg_metadata(
                generation
            )
        if len(result.generations) == 1 and result.llm_output is not None:
            result.generations[0].message.response_metadata = {
                **result.llm_output,
                **result.generations[0].message.response_metadata,
            }
        if check_cache and llm_cache:
            await llm_cache.aupdate(prompt, llm_string, result.generations)
        return result

    @abstractmethod
    def _generate(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: CallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Generate the result.

        Args:
            messages: The messages to generate from.
            stop: Optional list of stop words to use when generating.
            run_manager: Optional callback manager to use for this call.
            **kwargs: Additional keyword arguments to pass to the model.

        Returns:
            The chat result.
        """

    async def _agenerate(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: AsyncCallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> ChatResult:
        """Generate the result.

        Args:
            messages: The messages to generate from.
            stop: Optional list of stop words to use when generating.
            run_manager: Optional callback manager to use for this call.
            **kwargs: Additional keyword arguments to pass to the model.

        Returns:
            The chat result.
        """
        return await run_in_executor(
            None,
            self._generate,
            messages,
            stop,
            run_manager.get_sync() if run_manager else None,
            **kwargs,
        )

    def _stream(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: CallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> Iterator[ChatGenerationChunk]:
        """Stream the output of the model.

        Args:
            messages: The messages to generate from.
            stop: Optional list of stop words to use when generating.
            run_manager: Optional callback manager to use for this call.
            **kwargs: Additional keyword arguments to pass to the model.

        Yields:
            The chat generation chunks.
        """
        raise NotImplementedError

    async def _astream(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: AsyncCallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[ChatGenerationChunk]:
        """Stream the output of the model.

        Args:
            messages: The messages to generate from.
            stop: Optional list of stop words to use when generating.
            run_manager: Optional callback manager to use for this call.
            **kwargs: Additional keyword arguments to pass to the model.

        Yields:
            The chat generation chunks.
        """
        iterator = await run_in_executor(
            None,
            self._stream,
            messages,
            stop,
            run_manager.get_sync() if run_manager else None,
            **kwargs,
        )
        done = object()
        while True:
            item = await run_in_executor(
                None,
                next,
                iterator,
                done,
            )
            if item is done:
                break
            yield item  # type: ignore[misc]

    async def _call_async(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> BaseMessage:
        result = await self.agenerate(
            [messages], stop=stop, callbacks=callbacks, **kwargs
        )
        generation = result.generations[0][0]
        if isinstance(generation, ChatGeneration):
            return generation.message
        msg = "Unexpected generation type"
        raise ValueError(msg)

    @property
    @abstractmethod
    def _llm_type(self) -> str:
        """Return type of chat model."""

    @deprecated("1.4.2", alternative="asdict", removal="2.0.0")
    @override
    def dict(self, **_kwargs: Any) -> builtins.dict[str, Any]:
        """DEPRECATED - use `asdict()` instead.

        Return a dictionary representation of the chat model.
        """
        return self.asdict()

    def asdict(self) -> builtins.dict[str, Any]:
        """Return a dictionary representation of the chat model."""
        starter_dict = dict(self._identifying_params)
        starter_dict["_type"] = self._llm_type
        return starter_dict

    def _dict_for_compat(self) -> builtins.dict[str, Any]:
        """Return the chat model dictionary while preserving deprecated overrides."""
        with suppress_langchain_deprecation_warning():
            return self.dict()

    @override
    def bind(self, **kwargs: Any) -> _ChatModelBinding:
        """Bind kwargs to this chat model, returning a typed `_ChatModelBinding`.

        Overrides `Runnable.bind` so the result preserves chat-model-specific
        `stream_events` / `astream_events` overloads. Without this override,
        `model.bind(...).stream_events(version="v3")` would type as
        `Iterator[Any]` and `await model.bind(...).astream_events(version="v3")`
        as `Any`, forcing callers to `cast`.
        """
        return _ChatModelBinding(bound=self, kwargs=kwargs, config={})

    def bind_tools(
        self,
        tools: Sequence[builtins.dict[str, Any] | type | Callable[..., Any] | BaseTool],
        *,
        tool_choice: str | None = None,
        **kwargs: Any,
    ) -> Runnable[LanguageModelInput, AIMessage]:
        """Bind tools to the model.

        Args:
            tools: Sequence of tools to bind to the model.
            tool_choice: The tool to use. If "any" then any tool can be used.

        Returns:
            A Runnable that returns a message.

        """
        raise NotImplementedError

    def with_structured_output(
        self,
        schema: builtins.dict[str, Any] | type,
        *,
        include_raw: bool = False,
        **kwargs: Any,
    ) -> Runnable[LanguageModelInput, builtins.dict[str, Any] | BaseModel]:
        """Model wrapper that returns outputs formatted to match the given schema.

        Args:
            schema: The output schema. Can be passed in as:

                - An OpenAI function/tool schema,
                - A JSON Schema,
                - A `TypedDict` class,
                - Or a Pydantic class.

                If `schema` is a Pydantic class then the model output will be a
                Pydantic instance of that class, and the model-generated fields will be
                validated by the Pydantic class. Otherwise the model output will be a
                dict and will not be validated.

                See `langchain_core.utils.function_calling.convert_to_openai_tool` for
                more on how to properly specify types and descriptions of schema fields
                when specifying a Pydantic or `TypedDict` class.

            include_raw:
                If `False` then only the parsed structured output is returned.

                If an error occurs during model output parsing it will be raised.

                If `True` then both the raw model response (a `BaseMessage`) and the
                parsed model response will be returned.

                If an error occurs during output parsing it will be caught and returned
                as well.

                The final output is always a `dict` with keys `'raw'`, `'parsed'`, and
                `'parsing_error'`.

        Raises:
            ValueError: If there are any unsupported `kwargs`.
            NotImplementedError: If the model does not implement
                `with_structured_output()`.

        Returns:
            A `Runnable` that takes same inputs as a
                `langchain_core.language_models.chat.BaseChatModel`. If `include_raw` is
                `False` and `schema` is a Pydantic class, `Runnable` outputs an instance
                of `schema` (i.e., a Pydantic object). Otherwise, if `include_raw` is
                `False` then `Runnable` outputs a `dict`.

                If `include_raw` is `True`, then `Runnable` outputs a `dict` with keys:

                - `'raw'`: `BaseMessage`
                - `'parsed'`: `None` if there was a parsing error, otherwise the type
                    depends on the `schema` as described above.
                - `'parsing_error'`: `BaseException | None`

        ???+ example "Pydantic schema (`include_raw=False`)"

            ```python
            from pydantic import BaseModel


            class AnswerWithJustification(BaseModel):
                '''An answer to the user question along with justification for the answer.'''

                answer: str
                justification: str


            model = ChatModel(model="model-name", temperature=0)
            structured_model = model.with_structured_output(AnswerWithJustification)

            structured_model.invoke(
                "What weighs more a pound of bricks or a pound of feathers"
            )

            # -> AnswerWithJustification(
            #     answer='They weigh the same',
            #     justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'
            # )
            ```

        ??? example "Pydantic schema (`include_raw=True`)"

            ```python
            from pydantic import BaseModel


            class AnswerWithJustification(BaseModel):
                '''An answer to the user question along with justification for the answer.'''

                answer: str
                justification: str


            model = ChatModel(model="model-name", temperature=0)
            structured_model = model.with_structured_output(
                AnswerWithJustification, include_raw=True
            )

            structured_model.invoke(
                "What weighs more a pound of bricks or a pound of feathers"
            )
            # -> {
            #     'raw': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_Ao02pnFYXD6GN1yzc0uXPsvF', 'function': {'arguments': '{"answer":"They weigh the same.","justification":"Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ."}', 'name': 'AnswerWithJustification'}, 'type': 'function'}]}),
            #     'parsed': AnswerWithJustification(answer='They weigh the same.', justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'),
            #     'parsing_error': None
            # }
            ```

        ??? example "Dictionary schema (`include_raw=False`)"

            ```python
            from pydantic import BaseModel
            from langchain_core.utils.function_calling import convert_to_openai_tool


            class AnswerWithJustification(BaseModel):
                '''An answer to the user question along with justification for the answer.'''

                answer: str
                justification: str


            dict_schema = convert_to_openai_tool(AnswerWithJustification)
            model = ChatModel(model="model-name", temperature=0)
            structured_model = model.with_structured_output(dict_schema)

            structured_model.invoke(
                "What weighs more a pound of bricks or a pound of feathers"
            )
            # -> {
            #     'answer': 'They weigh the same',
            #     'justification': 'Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume and density of the two substances differ.'
            # }
            ```

        !!! warning "Behavior changed in `langchain-core` 0.2.26"

            Added support for `TypedDict` class.

        """  # noqa: E501
        _ = kwargs.pop("method", None)
        _ = kwargs.pop("strict", None)
        if kwargs:
            msg = f"Received unsupported arguments {kwargs}"
            raise ValueError(msg)

        if type(self).bind_tools is BaseChatModel.bind_tools:
            msg = "with_structured_output is not implemented for this model."
            raise NotImplementedError(msg)

        llm = self.bind_tools(
            [schema],
            tool_choice="any",
            ls_structured_output_format={
                "kwargs": {"method": "function_calling"},
                "schema": schema,
            },
        )
        output_parser: JsonOutputToolsParser
        if isinstance(schema, type) and is_basemodel_subclass(schema):
            output_parser = PydanticToolsParser(tools=[schema], first_tool_only=True)
        else:
            key_name = convert_to_openai_tool(schema)["function"]["name"]
            output_parser = JsonOutputKeyToolsParser(
                key_name=key_name, first_tool_only=True
            )
        if include_raw:
            parser_assign = RunnablePassthrough.assign(
                parsed=itemgetter("raw") | output_parser, parsing_error=lambda _: None
            )
            parser_none = RunnablePassthrough.assign(parsed=lambda _: None)
            parser_with_fallback = parser_assign.with_fallbacks(
                [parser_none], exception_key="parsing_error"
            )
            return RunnableMap(raw=llm) | parser_with_fallback
        return llm | output_parser


class _ChatModelBinding(RunnableBinding[LanguageModelInput, AIMessage]):  # type: ignore[no-redef]
    """`RunnableBinding` that preserves chat-model-typed v3 overloads.

    Returned by `BaseChatModel.bind` so that callers of the bound runnable's
    `stream_events(version="v3")` / `astream_events(version="v3")` get the
    typed `ChatModelStream` / `AsyncChatModelStream` back without needing
    `cast`. At runtime this is a plain `RunnableBinding`; the subclass
    exists purely to give the type checker a more specific surface.

    The chat-model narrowing is preserved across further `bind` /
    `with_config` calls because `RunnableBinding.bind` constructs its
    result via `self.__class__(...)`.
    """

    @classmethod
    @override
    def lc_id(cls) -> list[str]:
        """Serialize as `RunnableBinding`.

        At runtime this class is behaviorally identical to `RunnableBinding`;
        keeping the serialized id stable means existing snapshots and the
        load mapping continue to work without registering a new entry.
        """
        return [*cls.get_lc_namespace(), "RunnableBinding"]

    @overload  # type: ignore[override]
    def stream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v1", "v2"] = "v2",
        **kwargs: Any,
    ) -> Iterator[StreamEvent]: ...

    @overload
    def stream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v3"],
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> ChatModelStream: ...

    def stream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v1", "v2", "v3"] = "v2",
        **kwargs: Any,
    ) -> Iterator[StreamEvent] | ChatModelStream:
        return super().stream_events(input, config, version=version, **kwargs)

    @overload
    def astream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v1", "v2"] = "v2",
        **kwargs: Any,
    ) -> AsyncIterator[StreamEvent]: ...

    @overload
    def astream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        *,
        version: Literal["v3"],
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> Awaitable[AsyncChatModelStream]: ...

    def astream_events(
        self,
        input: LanguageModelInput,
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> AsyncIterator[StreamEvent] | Awaitable[AsyncChatModelStream]:
        return cast(
            "AsyncIterator[StreamEvent] | Awaitable[AsyncChatModelStream]",
            super().astream_events(input, config, **kwargs),
        )


class SimpleChatModel(BaseChatModel):
    """Simplified implementation for a chat model to inherit from.

    !!! note
        This implementation is primarily here for backwards compatibility. For new
        implementations, please use `BaseChatModel` directly.

    """

    def _generate(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: CallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> ChatResult:
        output_str = self._call(messages, stop=stop, run_manager=run_manager, **kwargs)
        message = AIMessage(content=output_str)
        generation = ChatGeneration(message=message)
        return ChatResult(generations=[generation])

    @abstractmethod
    def _call(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: CallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> str:
        """Simpler interface."""

    async def _agenerate(
        self,
        messages: list[BaseMessage],
        stop: list[str] | None = None,
        run_manager: AsyncCallbackManagerForLLMRun | None = None,
        **kwargs: Any,
    ) -> ChatResult:
        return await run_in_executor(
            None,
            self._generate,
            messages,
            stop=stop,
            run_manager=run_manager.get_sync() if run_manager else None,
            **kwargs,
        )


def _gen_info_and_msg_metadata(
    generation: ChatGeneration | ChatGenerationChunk,
) -> dict[str, Any]:
    return {
        **(generation.generation_info or {}),
        **generation.message.response_metadata,
    }


_MAX_CLEANUP_DEPTH = 100


def _cleanup_llm_representation(serialized: Any, depth: int) -> None:
    """Remove non-serializable objects from a serialized object."""
    if depth > _MAX_CLEANUP_DEPTH:  # Don't cooperate for pathological cases
        return

    if not isinstance(serialized, dict):
        return

    if (
        "type" in serialized
        and serialized["type"] == "not_implemented"
        and "repr" in serialized
    ):
        del serialized["repr"]

    if "graph" in serialized:
        del serialized["graph"]

    if "kwargs" in serialized:
        kwargs = serialized["kwargs"]

        for value in kwargs.values():
            _cleanup_llm_representation(value, depth + 1)
