"""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=` 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=` 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)