ChatGoogleGenerativeAI

Reference docs

This page contains reference documentation for ChatGoogleGenerativeAI. See the docs for conceptual guides, tutorials, and examples on using ChatGoogleGenerativeAI.

ChatGoogleGenerativeAI

Bases: _BaseGoogleGenerativeAI, BaseChatModel

Google GenAI chat model integration.

Setup

Vertex AI Platform Support

Added in langchain-google-genai 4.0.0.

ChatGoogleGenerativeAI now supports both the Gemini Developer API and Vertex AI Platform as backend options.

For Gemini Developer API (simplest):

1. Set the GOOGLE_API_KEY environment variable (recommended), or
2. Pass your API key using the api_key parameter

from langchain_google_genai import ChatGoogleGenerativeAI

model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview", api_key="...")

For Vertex AI Platform with API key:

export GEMINI_API_KEY='your-api-key'
export GOOGLE_GENAI_USE_VERTEXAI=true
export GOOGLE_CLOUD_PROJECT='your-project-id'

model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")
# Or explicitly:
model = ChatGoogleGenerativeAI(
    model="gemini-3-pro-preview",
    api_key="...",
    project="your-project-id",
    vertexai=True,
)

For Vertex AI with credentials:

model = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    project="your-project-id",
    # Uses Application Default Credentials (ADC)
)

Automatic backend detection (when vertexai=None / unspecified):

1. If GOOGLE_GENAI_USE_VERTEXAI env var is set, uses that value
2. If credentials parameter is provided, uses Vertex AI
3. If project parameter is provided, uses Vertex AI
4. Otherwise, uses Gemini Developer API

Environment variables

Variable	Purpose	Backend
GOOGLE_API_KEY	API key (primary)	Both (see GOOGLE_GENAI_USE_VERTEXAI)
GEMINI_API_KEY	API key (fallback)	Both (see GOOGLE_GENAI_USE_VERTEXAI)
GOOGLE_GENAI_USE_VERTEXAI	Force Vertex AI backend (true/false)	Vertex AI
GOOGLE_CLOUD_PROJECT	GCP project ID	Vertex AI
GOOGLE_CLOUD_LOCATION	GCP region (default: us-central1)	Vertex AI
HTTPS_PROXY	HTTP/HTTPS proxy URL	Both
SSL_CERT_FILE	Custom SSL certificate file	Both

GOOGLE_API_KEY is checked first for backwards compatibility. (GEMINI_API_KEY was introduced later to better reflect the API's branding.)

Proxy configuration

Set these before initializing:

export HTTPS_PROXY='http://username:password@proxy_uri:port'
export SSL_CERT_FILE='path/to/cert.pem'  # Optional: custom SSL certificate

For SOCKS5 proxies or advanced proxy configuration, use the client_args parameter:

model = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    client_args={"proxy": "socks5://user:pass@host:port"},
)

Instantiation

from langchain_google_genai import ChatGoogleGenerativeAI

model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")
model.invoke("Write me a ballad about LangChain")

Invoke

messages = [
    ("system", "Translate the user sentence to French."),
    ("human", "I love programming."),
]
model.invoke(messages)

AIMessage(
    content=[
        {
            "type": "text",
            "text": "**J'adore la programmation.**\n\nYou can also say:...",
            "extras": {"signature": "Eq0W..."},
        }
    ],
    additional_kwargs={},
    response_metadata={
        "prompt_feedback": {"block_reason": 0, "safety_ratings": []},
        "finish_reason": "STOP",
        "model_name": "gemini-3-pro-preview",
        "safety_ratings": [],
        "model_provider": "google_genai",
    },
    id="lc_run--63a04ced-6b63-4cf6-86a1-c32fa565938e-0",
    usage_metadata={
        "input_tokens": 12,
        "output_tokens": 826,
        "total_tokens": 838,
        "input_token_details": {"cache_read": 0},
        "output_token_details": {"reasoning": 777},
    },
)

content format

The shape of content may differ based on the model chosen. See the docs for more info.

Stream

from langchain_google_genai import ChatGoogleGenerativeAI

model = ChatGoogleGenerativeAI(model="gemini-2.5-flash")

for chunk in model.stream(messages):
    print(chunk)

AIMessageChunk(
    content="J",
    response_metadata={"finish_reason": "STOP", "safety_ratings": []},
    id="run-e905f4f4-58cb-4a10-a960-448a2bb649e3",
    usage_metadata={
        "input_tokens": 18,
        "output_tokens": 1,
        "total_tokens": 19,
    },
)
AIMessageChunk(
    content="'adore programmer. \\n",
    response_metadata={
        "finish_reason": "STOP",
        "safety_ratings": [
            {
                "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
                "probability": "NEGLIGIBLE",
                "blocked": False,
            },
            {
                "category": "HARM_CATEGORY_HATE_SPEECH",
                "probability": "NEGLIGIBLE",
                "blocked": False,
            },
            {
                "category": "HARM_CATEGORY_HARASSMENT",
                "probability": "NEGLIGIBLE",
                "blocked": False,
            },
            {
                "category": "HARM_CATEGORY_DANGEROUS_CONTENT",
                "probability": "NEGLIGIBLE",
                "blocked": False,
            },
        ],
    },
    id="run-e905f4f4-58cb-4a10-a960-448a2bb649e3",
    usage_metadata={
        "input_tokens": 18,
        "output_tokens": 5,
        "total_tokens": 23,
    },
)

To assemble a full AIMessage message from a stream of chunks:

stream = model.stream(messages)
full = next(stream)
for chunk in stream:
    full += chunk
full

AIMessageChunk(
    content="J'adore programmer. \\n",
    response_metadata={
        "finish_reason": "STOPSTOP",
        "safety_ratings": [
            {
                "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
                "probability": "NEGLIGIBLE",
                "blocked": False,
            },
            {
                "category": "HARM_CATEGORY_HATE_SPEECH",
                "probability": "NEGLIGIBLE",
                "blocked": False,
            },
            {
                "category": "HARM_CATEGORY_HARASSMENT",
                "probability": "NEGLIGIBLE",
                "blocked": False,
            },
            {
                "category": "HARM_CATEGORY_DANGEROUS_CONTENT",
                "probability": "NEGLIGIBLE",
                "blocked": False,
            },
        ],
    },
    id="run-3ce13a42-cd30-4ad7-a684-f1f0b37cdeec",
    usage_metadata={
        "input_tokens": 36,
        "output_tokens": 6,
        "total_tokens": 42,
    },
)

content format

The shape of content may differ based on the model chosen. See the docs for more info.

Async invocation

await model.ainvoke(messages)

# stream:
async for chunk in (await model.astream(messages))

# batch:
await model.abatch([messages])

Tool calling

See the docs for more info.

from pydantic import BaseModel, Field


class GetWeather(BaseModel):
    '''Get the current weather in a given location'''

    location: str = Field(
        ..., description="The city and state, e.g. San Francisco, CA"
    )


class GetPopulation(BaseModel):
    '''Get the current population in a given location'''

    location: str = Field(
        ..., description="The city and state, e.g. San Francisco, CA"
    )


llm_with_tools = llm.bind_tools([GetWeather, GetPopulation])
ai_msg = llm_with_tools.invoke(
    "Which city is hotter today and which is bigger: LA or NY?"
)
ai_msg.tool_calls

[
    {
        "name": "GetWeather",
        "args": {"location": "Los Angeles, CA"},
        "id": "c186c99f-f137-4d52-947f-9e3deabba6f6",
    },
    {
        "name": "GetWeather",
        "args": {"location": "New York City, NY"},
        "id": "cebd4a5d-e800-4fa5-babd-4aa286af4f31",
    },
    {
        "name": "GetPopulation",
        "args": {"location": "Los Angeles, CA"},
        "id": "4f92d897-f5e4-4d34-a3bc-93062c92591e",
    },
    {
        "name": "GetPopulation",
        "args": {"location": "New York City, NY"},
        "id": "634582de-5186-4e4b-968b-f192f0a93678",
    },
]

Structured output

See the docs for more info.

from typing import Optional

from pydantic import BaseModel, Field


class Joke(BaseModel):
    '''Joke to tell user.'''

    setup: str = Field(description="The setup of the joke")
    punchline: str = Field(description="The punchline to the joke")
    rating: Optional[int] = Field(
        description="How funny the joke is, from 1 to 10"
    )


# Default method uses json_schema for reliable structured output
structured_model = model.with_structured_output(Joke)
structured_model.invoke("Tell me a joke about cats")

# Alternative: use function_calling method (less reliable)
structured_model_fc = model.with_structured_output(
    Joke, method="function_calling"
)

Joke(
    setup="Why are cats so good at video games?",
    punchline="They have nine lives on the internet",
    rating=None,
)

Two methods are supported for structured output:

• method='json_schema' (default): Uses Gemini's native structured output API.

  The Google GenAI SDK automatically transforms schemas to ensure compatibility with Gemini. This includes:

  • Inlining $defs definitions (Union types work correctly)
  • Resolving $ref references for nested schemas
  • Property ordering preservation
  • Support for streaming partial JSON chunks

  Uses Gemini's response_json_schema API param. Refer to the Gemini API docs for more details. This method is recommended for better reliability as it constrains the model's generation process directly.

• method='function_calling': Uses tool calling to extract structured data. Less reliable than json_schema but compatible with all models.

Image input

See the docs for more info.

import base64
import httpx
from langchain.messages import HumanMessage

image_url = "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg"
image_data = base64.b64encode(httpx.get(image_url).content).decode("utf-8")
message = HumanMessage(
    content=[
        {"type": "text", "text": "describe the weather in this image"},
        {
            "type": "image_url",
            "image_url": {"url": f"data:image/jpeg;base64,{image_data}"},
        },
    ]
)
ai_msg = model.invoke([message])
ai_msg.content

The weather in this image appears to be sunny and pleasant. The sky is a bright
blue with scattered white clouds, suggesting fair weather. The lush green grass
and trees indicate a warm and possibly slightly breezy day. There are no...

PDF input

See the docs for more info.

import base64
from langchain.messages import HumanMessage

pdf_bytes = open("/path/to/your/test.pdf", "rb").read()
pdf_base64 = base64.b64encode(pdf_bytes).decode("utf-8")

message = HumanMessage(
    content=[
        {"type": "text", "text": "describe the document in a sentence"},
        {
            "type": "file",
            "source_type": "base64",
            "mime_type": "application/pdf",
            "data": pdf_base64,
        },
    ]
)
ai_msg = model.invoke([message])

Audio input

See the docs for more info.

import base64
from langchain.messages import HumanMessage

audio_bytes = open("/path/to/your/audio.mp3", "rb").read()
audio_base64 = base64.b64encode(audio_bytes).decode("utf-8")

message = HumanMessage(
    content=[
        {"type": "text", "text": "summarize this audio in a sentence"},
        {
            "type": "file",
            "source_type": "base64",
            "mime_type": "audio/mp3",
            "data": audio_base64,
        },
    ]
)
ai_msg = model.invoke([message])

Video input

See the docs for more info.

import base64
from langchain.messages import HumanMessage

video_bytes = open("/path/to/your/video.mp4", "rb").read()
video_base64 = base64.b64encode(video_bytes).decode("utf-8")

message = HumanMessage(
    content=[
        {
            "type": "text",
            "text": "describe what's in this video in a sentence",
        },
        {
            "type": "file",
            "source_type": "base64",
            "mime_type": "video/mp4",
            "data": video_base64,
        },
    ]
)
ai_msg = model.invoke([message])

You can also pass YouTube URLs directly:

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage

model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")

message = HumanMessage(
    content=[
        {"type": "text", "text": "Summarize the video in 3 sentences."},
        {
            "type": "media",
            "file_uri": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
            "mime_type": "video/mp4",
        },
    ]
)
response = model.invoke([message])
print(response.text)

Image generation

See the docs for more info.

Audio generation

See the docs for more info.

Vertex compatibility

Audio generation models (TTS) are currently in preview on Vertex AI and may require allowlist access. If you receive an INVALID_ARGUMENT error when using TTS models with vertexai=True, your project may need to be allowlisted.

See this post on the Google AI forum for more details.

File upload

You can also upload files to Google's servers and reference them by URI.

This works for PDFs, images, videos, and audio files.

import time
from google import genai
from langchain.messages import HumanMessage

client = genai.Client()

myfile = client.files.upload(file="/path/to/your/sample.pdf")
while myfile.state.name == "PROCESSING":
    time.sleep(2)
    myfile = client.files.get(name=myfile.name)

message = HumanMessage(
    content=[
        {"type": "text", "text": "What is in the document?"},
        {
            "type": "media",
            "file_uri": myfile.uri,
            "mime_type": "application/pdf",
        },
    ]
)
ai_msg = model.invoke([message])

Thinking

See the docs for more info.

Gemini 3+ models use thinking_level ('low', 'medium', or 'high') to control reasoning depth. If not specified, defaults to 'high'.

model = ChatGoogleGenerativeAI(
    model="gemini-3-pro-preview",
    thinking_level="low",  # For faster, lower-latency responses
)

Gemini 2.5 models use thinking_budget (an integer token count) to control reasoning. Set to 0 to disable thinking (where supported), or -1 for dynamic thinking.

See the Gemini API docs for more details on thinking models.

To see a thinking model's thoughts, set include_thoughts=True to have the model's reasoning summaries included in the response.

model = ChatGoogleGenerativeAI(
    model="gemini-3-pro-preview",
    include_thoughts=True,
)
ai_msg = model.invoke("How many 'r's are in the word 'strawberry'?")

Thought signatures

Gemini 3+ models return thought signatures—encrypted representations of the model's internal reasoning.

For multi-turn conversations involving tool calls, you must pass the full AIMessage back to the model so that these signatures are preserved. This happens automatically when you append the AIMessage to your message list.

See the LangChain docs for more info as well as a code example.

See the Gemini API docs for more details on thought signatures.

Google search

See the docs for more info.

model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")
response = model.invoke(
    "When is the next total solar eclipse in US?",
    tools=[{"google_search": {}}],
)
response.content_blocks

Alternatively, you can bind the tool to the model for easier reuse across calls:

model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")

model_with_search = model.bind_tools([{"google_search": {}}])
response = model_with_search.invoke(
    "When is the next total solar eclipse in US?"
)

response.content_blocks

Google Maps

See the docs for more info.

Code execution

See the docs for more info.

from langchain_google_genai import ChatGoogleGenerativeAI

model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")

model_with_code_interpreter = model.bind_tools([{"code_execution": {}}])
response = model_with_code_interpreter.invoke("Use Python to calculate 3^3.")

response.content_blocks

[{'type': 'server_tool_call',
  'name': 'code_interpreter',
  'args': {'code': 'print(3**3)', 'language': <Language.PYTHON: 1>},
  'id': '...'},
 {'type': 'server_tool_result',
  'tool_call_id': '',
  'status': 'success',
  'output': '27\n',
  'extras': {'block_type': 'code_execution_result',
   'outcome': 1}},
 {'type': 'text', 'text': 'The calculation of 3 to the power of 3 is 27.'}]

Computer use

See the docs for more info.

Preview model limitations

The Computer Use model is in preview and may produce unexpected behavior.

Always supervise automated tasks and avoid use with sensitive data or critical operations. See the Gemini API docs for safety best practices.

Token usage

See the docs for more info.

ai_msg = model.invoke(messages)
ai_msg.usage_metadata

{"input_tokens": 18, "output_tokens": 5, "total_tokens": 23}

Safety settings

Gemini models have default safety settings that can be overridden. If you are receiving lots of "Safety Warnings" from your models, you can try tweaking the safety_settings attribute of the model. For example, to turn off safety blocking for dangerous content, you can construct your LLM as follows:

from langchain_google_genai import (
    ChatGoogleGenerativeAI,
    HarmBlockThreshold,
    HarmCategory,
)

llm = ChatGoogleGenerativeAI(
    model="gemini-3-pro-preview",
    safety_settings={
        HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_NONE,
    },
)

For an enumeration of the categories and thresholds available, see Google's safety setting types.

Context caching

See the docs for more info.

Context caching allows you to store and reuse content (e.g., PDFs, images) for faster processing. The cached_content parameter accepts a cache name created via the Google Generative AI API.

See the Gemini docs for more details on cached content.

Below are two examples: caching a single file directly and caching multiple files using Part.

Single file example

This caches a single file and queries it.

from google import genai
from google.genai import types
import time
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.messages import HumanMessage

client = genai.Client()

# Upload file
file = client.files.upload(file="path/to/your/file")
while file.state.name == "PROCESSING":
    time.sleep(2)
    file = client.files.get(name=file.name)

# Create cache
model = "gemini-3-pro-preview"
cache = client.caches.create(
    model=model,
    config=types.CreateCachedContentConfig(
        display_name="Cached Content",
        system_instruction=(
            "You are an expert content analyzer, and your job is to answer "
            "the user's query based on the file you have access to."
        ),
        contents=[file],
        ttl="300s",
    ),
)

# Query with LangChain
llm = ChatGoogleGenerativeAI(
    model=model,
    cached_content=cache.name,
)
message = HumanMessage(content="Summarize the main points of the content.")
llm.invoke([message])

Multiple files example

This caches two files using Part and queries them together.

from google import genai
from google.genai.types import CreateCachedContentConfig, Content, Part
import time
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.messages import HumanMessage

client = genai.Client()

# Upload files
file_1 = client.files.upload(file="./file1")
while file_1.state.name == "PROCESSING":
    time.sleep(2)
    file_1 = client.files.get(name=file_1.name)

file_2 = client.files.upload(file="./file2")
while file_2.state.name == "PROCESSING":
    time.sleep(2)
    file_2 = client.files.get(name=file_2.name)

# Create cache with multiple files
contents = [
    Content(
        role="user",
        parts=[
            Part.from_uri(file_uri=file_1.uri, mime_type=file_1.mime_type),
            Part.from_uri(file_uri=file_2.uri, mime_type=file_2.mime_type),
        ],
    )
]
model = "gemini-3-pro-preview"
cache = client.caches.create(
    model=model,
    config=CreateCachedContentConfig(
        display_name="Cached Contents",
        system_instruction=(
            "You are an expert content analyzer, and your job is to answer "
            "the user's query based on the files you have access to."
        ),
        contents=contents,
        ttl="300s",
    ),
)

# Query with LangChain
llm = ChatGoogleGenerativeAI(
    model=model,
    cached_content=cache.name,
)
message = HumanMessage(
    content="Provide a summary of the key information across both files."
)
llm.invoke([message])

Response metadata

ai_msg = model.invoke(messages)
ai_msg.response_metadata

{
    "model_name": "gemini-3-pro-preview",
    "model_provider": "google_genai",
    "prompt_feedback": {"block_reason": 0, "safety_ratings": []},
    "finish_reason": "STOP",
    "safety_ratings": [
        {
            "category": "HARM_CATEGORY_SEXUALLY_EXPLICIT",
            "probability": "NEGLIGIBLE",
            "blocked": False,
        },
        {
            "category": "HARM_CATEGORY_HATE_SPEECH",
            "probability": "NEGLIGIBLE",
            "blocked": False,
        },
        {
            "category": "HARM_CATEGORY_HARASSMENT",
            "probability": "NEGLIGIBLE",
            "blocked": False,
        },
        {
            "category": "HARM_CATEGORY_DANGEROUS_CONTENT",
            "probability": "NEGLIGIBLE",
            "blocked": False,
        },
    ],
}

METHOD	DESCRIPTION
get_name	Get the name of the Runnable.
get_input_schema	Get a Pydantic model that can be used to validate input to the Runnable.
get_input_jsonschema	Get a JSON schema that represents the input to the Runnable.
get_output_schema	Get a Pydantic model that can be used to validate output to the Runnable.
get_output_jsonschema	Get a JSON schema that represents the output of the Runnable.
config_schema	The type of config this Runnable accepts specified as a Pydantic model.
get_config_jsonschema	Get a JSON schema that represents the config of the Runnable.
get_graph	Return a graph representation of this Runnable.
get_prompts	Return a list of prompts used by this Runnable.
__or__	Runnable "or" operator.
__ror__	Runnable "reverse-or" operator.
pipe	Pipe Runnable objects.
pick	Pick keys from the output dict of this Runnable.
assign	Assigns new fields to the dict output of this Runnable.
ainvoke	Transform a single input into an output.
batch	Default implementation runs invoke in parallel using a thread pool executor.
batch_as_completed	Run invoke in parallel on a list of inputs.
abatch	Default implementation runs ainvoke in parallel using asyncio.gather.
abatch_as_completed	Run ainvoke in parallel on a list of inputs.
stream	Default implementation of stream, which calls invoke.
astream	Default implementation of astream, which calls ainvoke.
astream_log	Stream all output from a Runnable, as reported to the callback system.
astream_events	Generate a stream of events.
transform	Transform inputs to outputs.
atransform	Transform inputs to outputs.
bind	Bind arguments to a Runnable, returning a new Runnable.
with_config	Bind config to a Runnable, returning a new Runnable.
with_listeners	Bind lifecycle listeners to a Runnable, returning a new Runnable.
with_alisteners	Bind async lifecycle listeners to a Runnable.
with_types	Bind input and output types to a Runnable, returning a new Runnable.
with_retry	Create a new Runnable that retries the original Runnable on exceptions.
map	Return a new Runnable that maps a list of inputs to a list of outputs.
with_fallbacks	Add fallbacks to a Runnable, returning a new Runnable.
as_tool	Create a BaseTool from a Runnable.
get_lc_namespace	Get the namespace of the LangChain object.
lc_id	Return a unique identifier for this class for serialization purposes.
to_json	Serialize the Runnable to JSON.
to_json_not_implemented	Serialize a "not implemented" object.
configurable_fields	Configure particular Runnable fields at runtime.
configurable_alternatives	Configure alternatives for Runnable objects that can be set at runtime.
set_verbose	If verbose is None, set it.
generate_prompt	Pass a sequence of prompts to the model and return model generations.
agenerate_prompt	Asynchronously pass a sequence of prompts and return model generations.
get_token_ids	Return the ordered IDs of the tokens in a text.
get_num_tokens_from_messages	Get the number of tokens in the messages.
generate	Pass a sequence of prompts to the model and return model generations.
agenerate	Asynchronously pass a sequence of prompts to a model and return generations.
dict	Return a dictionary of the LLM.
__init__	Needed for arg validation.
is_lc_serializable	Is this class serializable?
build_extra	Build extra kwargs from additional params that were passed in.
validate_environment	Validates params and builds client.
__del__	Clean up the client on deletion.
invoke	Override invoke on ChatGoogleGenerativeAI to add code_execution.
get_num_tokens	Get the number of tokens present in the text. Uses the model's tokenizer.
with_structured_output	Return a Runnable that constrains model output to a given schema.
bind_tools	Bind tool-like objects to this chat model.

name class-attribute instance-attribute

name: str | None = None

The name of the Runnable. Used for debugging and tracing.

InputType property

InputType: TypeAlias

Get the input type for this Runnable.

OutputType property

OutputType: Any

Get the output type for this Runnable.

input_schema property

input_schema: type[BaseModel]

The type of input this Runnable accepts specified as a Pydantic model.

output_schema property

output_schema: type[BaseModel]

Output schema.

The type of output this Runnable produces specified as a Pydantic model.

config_specs property

config_specs: list[ConfigurableFieldSpec]

List configurable fields for this Runnable.

lc_attributes property

lc_attributes: dict

List of attribute names that should be included in the serialized kwargs.

These attributes must be accepted by the constructor.

Default is an empty dictionary.

cache class-attribute instance-attribute

cache: BaseCache | bool | None = Field(default=None, exclude=True)

Whether to cache the response.

• If True, will use the global cache.
• If False, will not use a cache
• If None, will use the global cache if it's set, otherwise no cache.
• If instance of BaseCache, will use the provided cache.

Caching is not currently supported for streaming methods of models.

verbose class-attribute instance-attribute

verbose: bool = Field(default_factory=_get_verbosity, exclude=True, repr=False)

Whether to print out response text.

callbacks class-attribute instance-attribute

callbacks: Callbacks = Field(default=None, exclude=True)

Callbacks to add to the run trace.

tags class-attribute instance-attribute

tags: list[str] | None = Field(default=None, exclude=True)

Tags to add to the run trace.

metadata class-attribute instance-attribute

metadata: dict[str, Any] | None = Field(default=None, exclude=True)

Metadata to add to the run trace.

custom_get_token_ids class-attribute instance-attribute

custom_get_token_ids: Callable[[str], list[int]] | None = Field(
    default=None, exclude=True
)

Optional encoder to use for counting tokens.

rate_limiter class-attribute instance-attribute

rate_limiter: BaseRateLimiter | None = Field(default=None, exclude=True)

An optional rate limiter to use for limiting the number of requests.

disable_streaming class-attribute instance-attribute

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 the implementation does not properly support streaming.

output_version class-attribute instance-attribute

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) can also use this field to roll out new content formats in a backward-compatible way.

Added in langchain-core 1.0.0

profile class-attribute instance-attribute

profile: ModelProfile | None = Field(default=None, exclude=True)

Profile detailing model capabilities.

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.

Added in langchain-core 1.1.0

google_api_key class-attribute instance-attribute

google_api_key: SecretStr | None = Field(
    alias="api_key",
    default_factory=secret_from_env(["GOOGLE_API_KEY", "GEMINI_API_KEY"], default=None),
)

API key for authentication.

If not specified, will check the env vars GOOGLE_API_KEY and GEMINI_API_KEY with precedence given to GOOGLE_API_KEY.

• Gemini Developer API: API key is required (default when no project is set)
• Vertex AI: API key is optional (set vertexai=True or provide project)
  • If provided, uses API key for authentication
  • If not provided, uses Application Default Credentials (ADC) or credentials parameter

Vertex AI with API key

You can now use Vertex AI with API key authentication instead of service account credentials. Set GOOGLE_GENAI_USE_VERTEXAI=true or vertexai=True along with your API key and project.

credentials class-attribute instance-attribute

credentials: Any = None

Custom credentials for Vertex AI authentication.

When provided, forces Vertex AI backend (regardless of API key presence in google_api_key/api_key).

Accepts a google.auth.credentials.Credentials object.

If omitted and no API key is found, the SDK uses Application Default Credentials (ADC).

Service account credentials

from google.oauth2 import service_account

credentials = service_account.Credentials.from_service_account_file(
    "path/to/service-account.json",
    scopes=["https://www.googleapis.com/auth/cloud-platform"],
)

llm = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    credentials=credentials,
    project="my-project-id",
)

vertexai class-attribute instance-attribute

vertexai: bool | None = Field(default=None)

Whether to use Vertex AI backend.

If None (default), backend is automatically determined as follows:

1. If the GOOGLE_GENAI_USE_VERTEXAI env var is set, uses Vertex AI
2. If the credentials parameter is provided, uses Vertex AI
3. If the project parameter is provided, uses Vertex AI
4. Otherwise, uses Gemini Developer API

Set explicitly to True or False to override auto-detection.

Vertex AI with API key

You can use Vertex AI with API key authentication by setting:

export GEMINI_API_KEY='your-api-key'
export GOOGLE_GENAI_USE_VERTEXAI=true
export GOOGLE_CLOUD_PROJECT='your-project-id'

Or programmatically:

llm = ChatGoogleGenerativeAI(
    model="gemini-3-pro-preview",
    api_key="your-api-key",
    project="your-project-id",
    vertexai=True,
)

This allows for simpler authentication compared to service account JSON files.

project class-attribute instance-attribute

project: str | None = Field(default=None)

Google Cloud project ID (Vertex AI only).

Required when using Vertex AI.

Falls back to GOOGLE_CLOUD_PROJECT env var if not provided.

location class-attribute instance-attribute

location: str | None = Field(
    default_factory=from_env("GOOGLE_CLOUD_LOCATION", default=None)
)

Google Cloud region (Vertex AI only).

If not provided, falls back to the GOOGLE_CLOUD_LOCATION env var, then 'us-central1'.

base_url class-attribute instance-attribute

base_url: str | dict | None = Field(default=None, alias='client_options')

Custom base URL for the API client.

If not provided, defaults depend on the API being used:

• Gemini Developer API ( api_key/ google_api_key ): https://generativelanguage.googleapis.com/
• Vertex AI ( credentials): https://{location}-aiplatform.googleapis.com/

Backwards compatibility

Typed to accept dict to support backwards compatibility for the (now removed) client_options param.

If a dict is passed in, it will only extract the 'api_endpoint' key.

additional_headers class-attribute instance-attribute

additional_headers: dict[str, str] | None = Field(default=None)

Additional HTTP headers to include in API requests.

Passed as headers to HttpOptions when creating the client.

Example

llm = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    additional_headers={
        "X-Custom-Header": "value",
    },
)

client_args class-attribute instance-attribute

client_args: dict[str, Any] | None = Field(default=None)

Additional arguments to pass to the underlying HTTP client.

Applied to both sync and async clients.

SOCKS5 proxy

llm = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    client_args={"proxy": "socks5://user:pass@host:port"},
)

model class-attribute instance-attribute

model: str = Field(...)

Model name to use.

temperature class-attribute instance-attribute

temperature: float = 0.7

Run inference with this temperature.

Must be within [0.0, 2.0].

Automatic override for Gemini 3.0+ models

If temperature is not explicitly set and the model is Gemini 3.0 or later, it will be automatically set to 1.0 instead of the default 0.7 per the Google GenAI API best practices, as it can cause infinite loops, degraded reasoning performance, and failure on complex tasks.

top_p class-attribute instance-attribute

top_p: float | None = None

Decode using nucleus sampling.

Consider the smallest set of tokens whose probability sum is at least top_p.

Must be within [0.0, 1.0].

top_k class-attribute instance-attribute

top_k: int | None = None

Decode using top-k sampling: consider the set of top_k most probable tokens.

Must be positive.

max_output_tokens class-attribute instance-attribute

max_output_tokens: int | None = Field(default=None, alias='max_tokens')

Maximum number of tokens to include in a candidate.

Must be greater than zero.

If unset, will use the model's default value, which varies by model.

See docs for model-specific limits.

To constrain the number of thinking tokens to use when generating a response, see the thinking_budget parameter.

n class-attribute instance-attribute

n: int = 1

Number of chat completions to generate for each prompt.

Note that the API may not return the full n completions if duplicates are generated.

max_retries class-attribute instance-attribute

max_retries: int = Field(default=6, alias='retries')

The maximum number of retries to make when generating.

Disabling retries

To disable retries, set max_retries=1 (not 0) due to a quirk in the underlying Google SDK. max_retries=0 is interpreted as "use the (Google) default" (5 retries).

Setting max_retries=1 means only the initial request is made with no retries.

Handling rate limits (429 errors)

When you exceed quota limits, the API returns a 429 error with a suggested retry_delay. The SDK's built-in retry logic ignores this value and uses fixed exponential backoff instead. This is a known issue in Google's SDK and an issue has been raised upstream. We plan to implement proper handling once it's supported.

If you need to respect the server's suggested retry delay, disable SDK retries with max_retries=1 and implement custom retry logic:

import re
import time

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_google_genai.chat_models import ChatGoogleGenerativeAIError

llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", max_retries=1)

try:
    response = llm.invoke("Hello")
except ChatGoogleGenerativeAIError as e:
    if "429" in str(e):
        # Parse retry_delay from error: "[retry_delay { seconds: N }]"
        match = re.search(r"retry_delay\s*\{\s*seconds:\s*(\d+)", str(e))
        delay = int(match.group(1)) if match else 60
        time.sleep(delay)
        # Retry...

timeout class-attribute instance-attribute

timeout: float | None = Field(default=None, alias='request_timeout')

The maximum number of seconds to wait for a response.

response_modalities class-attribute instance-attribute

response_modalities: list[Modality] | None = Field(default=None)

A list of modalities of the response

media_resolution class-attribute instance-attribute

media_resolution: MediaResolution | None = Field(default=None)

Media resolution for the input media.

May be defined at the individual part level, allowing for mixed-resolution requests (e.g., images and videos of different resolutions in the same request).

May be 'low', 'medium', or 'high'.

Can be set either per-part or globally for all media inputs in the request. To set globally, set in the generation_config.

Model compatibility

Setting per-part media resolution requests to Gemini 2.5 models is not supported.

image_config class-attribute instance-attribute

image_config: dict[str, Any] | None = Field(default=None)

Configuration for image generation.

Provides control over generated image dimensions and quality for image generation models.

See genai.types.ImageConfig for a list of supported fields and their values.

Model compatibility

This parameter only applies to image generation models. Supported parameters vary by model and backend (Gemini Developer API and Vertex AI each support different subsets of parameters and models).

See the docs for more details and examples.

thinking_budget class-attribute instance-attribute

thinking_budget: int | None = Field(default=None)

Indicates the thinking budget in tokens.

Used to disable thinking for supported models (when set to 0) or to constrain the number of tokens used for thinking.

Dynamic thinking (allowing the model to decide how many tokens to use) is enabled when set to -1.

More information, including per-model limits, can be found in the Gemini API docs.

include_thoughts class-attribute instance-attribute

include_thoughts: bool | None = Field(default=None)

Indicates whether to include thoughts in the response.

Note

This parameter is only applicable for models that support thinking.

This does not disable thinking; to disable thinking, set thinking_budget to 0. for supported models. See the thinking_budget parameter for more details.

safety_settings class-attribute instance-attribute

safety_settings: SafetySettingDict | None = None

Default safety settings to use for all generations.

Example

from google.genai.types import HarmBlockThreshold, HarmCategory

safety_settings = {
    HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_ONLY_HIGH,
    HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
    HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_NONE,
}

seed class-attribute instance-attribute

seed: int | None = Field(default=None)

Seed used in decoding for reproducible generations.

By default, a random number is used.

Note

Using the same seed does not guarantee identical outputs, but makes them more deterministic. Reproducibility is "best effort" based on the model and infrastructure.

labels class-attribute instance-attribute

labels: dict[str, str] | None = Field(default=None)

User-defined key-value metadata for organizing and filtering billing reports.

Attach labels to categorize API usage by team, environment, or feature.

Can be overridden per-request via invoke kwargs.

See: https://cloud.google.com/vertex-ai/generative-ai/docs/multimodal/add-labels-to-api-calls

model_kwargs class-attribute instance-attribute

model_kwargs: dict[str, Any] = Field(default_factory=dict)

Holds any unexpected initialization parameters.

streaming class-attribute instance-attribute

streaming: bool | None = None

Whether to stream responses from the model.

convert_system_message_to_human class-attribute instance-attribute

convert_system_message_to_human: bool = False

Whether to merge any leading SystemMessage into the following HumanMessage.

Gemini does not support system messages; any unsupported messages will raise an error.

stop class-attribute instance-attribute

stop: list[str] | None = None

Stop sequences for the model.

response_mime_type class-attribute instance-attribute

response_mime_type: str | None = None

Output response MIME type of the generated candidate text.

Supported MIME types

• 'text/plain': (default) Text output.
• 'application/json': JSON response in the candidates.
• 'text/x.enum': Enum in plain text. (legacy; use JSON schema output instead)

Note

The model also needs to be prompted to output the appropriate response type, otherwise the behavior is undefined.

(In other words, simply setting this param doesn't force the model to comply; it only tells the model the kind of output expected. You still need to prompt it correctly.)

response_schema class-attribute instance-attribute

response_schema: dict[str, Any] | None = None

Enforce a schema to the output.

The format of the dictionary should follow JSON Schema specification.

Schema Transformation

The Google GenAI SDK automatically transforms schemas for Gemini compatibility:

• Inlines $defs definitions (enables Union types with anyOf)
• Resolves $ref pointers for nested/recursive schemas
• Preserves property ordering
• Supports constraints like minimum/maximum, minItems/maxItems

Using Union Types

Union types in Pydantic models (e.g., field: Union[TypeA, TypeB]) are automatically converted to anyOf schemas and work correctly with the json_schema method.

Refer to the Gemini API docs for more details on supported JSON Schema features.

thinking_level class-attribute instance-attribute

thinking_level: Literal["minimal", "low", "medium", "high"] | None = Field(default=None)

Indicates the thinking level.

Supported values

• 'low': Minimizes latency and cost.
• 'medium': Balances latency/cost with reasoning depth.
• 'high': Maximizes reasoning depth.

Replaces thinking_budget

thinking_budget is deprecated for Gemini 3+ models. If both parameters are provided, thinking_level takes precedence.

If left unspecified, the model's default thinking level is used. For Gemini 3+, this defaults to 'high'.

cached_content class-attribute instance-attribute

cached_content: str | None = None

The name of the cached content used as context to serve the prediction.

Note

Only used in explicit caching, where users can have control over caching (e.g. what content to cache) and enjoy guaranteed cost savings. Format: cachedContents/{cachedContent}.

async_client property

async_client: Any

Async client for Google GenAI operations..

RETURNS	DESCRIPTION
Any	The async client interface that exposes async versions of all client methods.

RAISES	DESCRIPTION
ValueError	If the client has not been initialized.

get_name

get_name(suffix: str | None = None, *, name: str | None = None) -> str

Get the name of the Runnable.

PARAMETER	DESCRIPTION
suffix	An optional suffix to append to the name. TYPE: str | None DEFAULT: None
name	An optional name to use instead of the Runnable's name. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
str	The name of the Runnable.

get_input_schema

get_input_schema(config: RunnableConfig | None = None) -> type[BaseModel]

Get a Pydantic model that can be used to validate input to the Runnable.

Runnable objects that leverage the configurable_fields and configurable_alternatives methods will have a dynamic input schema that depends on which configuration the Runnable is invoked with.

This method allows to get an input schema for a specific configuration.

PARAMETER	DESCRIPTION
config	A config to use when generating the schema. TYPE: RunnableConfig | None DEFAULT: None

RETURNS	DESCRIPTION
type[BaseModel]	A Pydantic model that can be used to validate input.

get_input_jsonschema

get_input_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

Get a JSON schema that represents the input to the Runnable.

PARAMETER	DESCRIPTION
config	A config to use when generating the schema. TYPE: RunnableConfig | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, Any]	A JSON schema that represents the input to the Runnable.

Example

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_input_jsonschema())

Added in langchain-core 0.3.0

get_output_schema

get_output_schema(config: RunnableConfig | None = None) -> type[BaseModel]

Get a Pydantic model that can be used to validate output to the Runnable.

Runnable objects that leverage the configurable_fields and configurable_alternatives methods will have a dynamic output schema that depends on which configuration the Runnable is invoked with.

This method allows to get an output schema for a specific configuration.

PARAMETER	DESCRIPTION
config	A config to use when generating the schema. TYPE: RunnableConfig | None DEFAULT: None

RETURNS	DESCRIPTION
type[BaseModel]	A Pydantic model that can be used to validate output.

get_output_jsonschema

get_output_jsonschema(config: RunnableConfig | None = None) -> dict[str, Any]

Get a JSON schema that represents the output of the Runnable.

PARAMETER	DESCRIPTION
config	A config to use when generating the schema. TYPE: RunnableConfig | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, Any]	A JSON schema that represents the output of the Runnable.

Example

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


runnable = RunnableLambda(add_one)

print(runnable.get_output_jsonschema())

Added in langchain-core 0.3.0

config_schema

config_schema(*, include: Sequence[str] | None = None) -> type[BaseModel]

The type of config this Runnable accepts specified as a Pydantic model.

To mark a field as configurable, see the configurable_fields and configurable_alternatives methods.

PARAMETER	DESCRIPTION
include	A list of fields to include in the config schema. TYPE: Sequence[str] | None DEFAULT: None

RETURNS	DESCRIPTION
type[BaseModel]	A Pydantic model that can be used to validate config.

get_config_jsonschema

get_config_jsonschema(*, include: Sequence[str] | None = None) -> dict[str, Any]

Get a JSON schema that represents the config of the Runnable.

PARAMETER	DESCRIPTION
include	A list of fields to include in the config schema. TYPE: Sequence[str] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, Any]	A JSON schema that represents the config of the Runnable.

Added in langchain-core 0.3.0

get_graph

get_graph(config: RunnableConfig | None = None) -> Graph

Return a graph representation of this Runnable.

get_prompts

get_prompts(config: RunnableConfig | None = None) -> list[BasePromptTemplate]

Return a list of prompts used by this Runnable.

__or__

__or__(
    other: Runnable[Any, Other]
    | Callable[[Iterator[Any]], Iterator[Other]]
    | Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
    | Callable[[Any], Other]
    | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]

Runnable "or" operator.

Compose this Runnable with another object to create a RunnableSequence.

PARAMETER	DESCRIPTION
other	Another Runnable or a Runnable-like object. TYPE: Runnable[Any, Other] | Callable[[Iterator[Any]], Iterator[Other]] | Callable[[AsyncIterator[Any]], AsyncIterator[Other]] | Callable[[Any], Other] | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any]

RETURNS	DESCRIPTION
RunnableSerializable[Input, Other]	A new Runnable.

__ror__

__ror__(
    other: Runnable[Other, Any]
    | Callable[[Iterator[Other]], Iterator[Any]]
    | Callable[[AsyncIterator[Other]], AsyncIterator[Any]]
    | Callable[[Other], Any]
    | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any],
) -> RunnableSerializable[Other, Output]

Runnable "reverse-or" operator.

Compose this Runnable with another object to create a RunnableSequence.

PARAMETER	DESCRIPTION
other	Another Runnable or a Runnable-like object. TYPE: Runnable[Other, Any] | Callable[[Iterator[Other]], Iterator[Any]] | Callable[[AsyncIterator[Other]], AsyncIterator[Any]] | Callable[[Other], Any] | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any]

RETURNS	DESCRIPTION
RunnableSerializable[Other, Output]	A new Runnable.

pipe

pipe(
    *others: Runnable[Any, Other] | Callable[[Any], Other], name: str | None = None
) -> RunnableSerializable[Input, Other]

Pipe Runnable objects.

Compose this Runnable with Runnable-like objects to make a RunnableSequence.

Equivalent to RunnableSequence(self, *others) or self | others[0] | ...

Example

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


def mul_two(x: int) -> int:
    return x * 2


runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4

sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]

PARAMETER	DESCRIPTION
*others	Other Runnable or Runnable-like objects to compose TYPE: Runnable[Any, Other] | Callable[[Any], Other] DEFAULT: ()
name	An optional name for the resulting RunnableSequence. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
RunnableSerializable[Input, Other]	A new Runnable.

pick

pick(keys: str | list[str]) -> RunnableSerializable[Any, Any]

Pick keys from the output dict of this Runnable.

Pick a single key

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}

json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]

Pick a list of keys

from typing import Any

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)


def as_bytes(x: Any) -> bytes:
    return bytes(x, "utf-8")


chain = RunnableMap(
    str=as_str, json=as_json, bytes=RunnableLambda(as_bytes)
)

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

PARAMETER	DESCRIPTION
keys	A key or list of keys to pick from the output dict. TYPE: str | list[str]

RETURNS	DESCRIPTION
RunnableSerializable[Any, Any]	a new Runnable.

assign

assign(
    **kwargs: Runnable[dict[str, Any], Any]
    | Callable[[dict[str, Any]], Any]
    | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]],
) -> RunnableSerializable[Any, Any]

Assigns new fields to the dict output of this Runnable.

from langchain_core.language_models.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter

prompt = (
    SystemMessagePromptTemplate.from_template("You are a nice assistant.")
    + "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])

chain: Runnable = prompt | model | {"str": StrOutputParser()}

chain_with_assign = chain.assign(hello=itemgetter("str") | model)

print(chain_with_assign.input_schema.model_json_schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.model_json_schema())
# {'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}

PARAMETER	DESCRIPTION
**kwargs	A mapping of keys to Runnable or Runnable-like objects that will be invoked with the entire output dict of this Runnable. TYPE: Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any] | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]] DEFAULT: {}

RETURNS	DESCRIPTION
RunnableSerializable[Any, Any]	A new Runnable.

ainvoke async

ainvoke(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AIMessage

Transform a single input into an output.

PARAMETER	DESCRIPTION
input	The input to the Runnable. TYPE: Input
config	A config to use when invoking the Runnable. The config supports standard keys like 'tags', 'metadata' for tracing purposes, 'max_concurrency' for controlling how much work to do in parallel, and other keys. Please refer to RunnableConfig for more details. TYPE: RunnableConfig | None DEFAULT: None

RETURNS	DESCRIPTION
Output	The output of the Runnable.

batch

batch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

Default implementation runs invoke in parallel using a thread pool executor.

The default implementation of batch works well for IO bound runnables.

Subclasses must override this method if they can batch more efficiently; e.g., if the underlying Runnable uses an API which supports a batch mode.

PARAMETER	DESCRIPTION
inputs	A list of inputs to the Runnable. TYPE: list[Input]
config	A config to use when invoking the Runnable. The config supports standard keys like 'tags', 'metadata' for tracing purposes, 'max_concurrency' for controlling how much work to do in parallel, and other keys. Please refer to RunnableConfig for more details. TYPE: RunnableConfig | list[RunnableConfig] | None DEFAULT: None
return_exceptions	Whether to return exceptions instead of raising them. TYPE: bool DEFAULT: False
**kwargs	Additional keyword arguments to pass to the Runnable. TYPE: Any | None DEFAULT: {}

RETURNS	DESCRIPTION
list[Output]	A list of outputs from the Runnable.

batch_as_completed

batch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> Iterator[tuple[int, Output | Exception]]

Run invoke in parallel on a list of inputs.

Yields results as they complete.

PARAMETER	DESCRIPTION
inputs	A list of inputs to the Runnable. TYPE: Sequence[Input]
config	A config to use when invoking the Runnable. The config supports standard keys like 'tags', 'metadata' for tracing purposes, 'max_concurrency' for controlling how much work to do in parallel, and other keys. Please refer to RunnableConfig for more details. TYPE: RunnableConfig | Sequence[RunnableConfig] | None DEFAULT: None
return_exceptions	Whether to return exceptions instead of raising them. TYPE: bool DEFAULT: False
**kwargs	Additional keyword arguments to pass to the Runnable. TYPE: Any | None DEFAULT: {}

YIELDS	DESCRIPTION
tuple[int, Output | Exception]	Tuples of the index of the input and the output from the Runnable.

abatch async

abatch(
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> list[Output]

Default implementation runs ainvoke in parallel using asyncio.gather.

The default implementation of batch works well for IO bound runnables.

Subclasses must override this method if they can batch more efficiently; e.g., if the underlying Runnable uses an API which supports a batch mode.

PARAMETER	DESCRIPTION
inputs	A list of inputs to the Runnable. TYPE: list[Input]
config	A config to use when invoking the Runnable. The config supports standard keys like 'tags', 'metadata' for tracing purposes, 'max_concurrency' for controlling how much work to do in parallel, and other keys. Please refer to RunnableConfig for more details. TYPE: RunnableConfig | list[RunnableConfig] | None DEFAULT: None
return_exceptions	Whether to return exceptions instead of raising them. TYPE: bool DEFAULT: False
**kwargs	Additional keyword arguments to pass to the Runnable. TYPE: Any | None DEFAULT: {}

RETURNS	DESCRIPTION
list[Output]	A list of outputs from the Runnable.

abatch_as_completed async

abatch_as_completed(
    inputs: Sequence[Input],
    config: RunnableConfig | Sequence[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any | None,
) -> AsyncIterator[tuple[int, Output | Exception]]

Run ainvoke in parallel on a list of inputs.

Yields results as they complete.

PARAMETER	DESCRIPTION
inputs	A list of inputs to the Runnable. TYPE: Sequence[Input]
config	A config to use when invoking the Runnable. The config supports standard keys like 'tags', 'metadata' for tracing purposes, 'max_concurrency' for controlling how much work to do in parallel, and other keys. Please refer to RunnableConfig for more details. TYPE: RunnableConfig | Sequence[RunnableConfig] | None DEFAULT: None
return_exceptions	Whether to return exceptions instead of raising them. TYPE: bool DEFAULT: False
**kwargs	Additional keyword arguments to pass to the Runnable. TYPE: Any | None DEFAULT: {}

YIELDS	DESCRIPTION
AsyncIterator[tuple[int, Output | Exception]]	A tuple of the index of the input and the output from the Runnable.

stream

stream(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> Iterator[AIMessageChunk]

Default implementation of stream, which calls invoke.

Subclasses must override this method if they support streaming output.

PARAMETER	DESCRIPTION
input	The input to the Runnable. TYPE: Input
config	The config to use for the Runnable. TYPE: RunnableConfig | None DEFAULT: None
**kwargs	Additional keyword arguments to pass to the Runnable. TYPE: Any | None DEFAULT: {}

YIELDS	DESCRIPTION
Output	The output of the Runnable.

astream async

astream(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[AIMessageChunk]

Default implementation of astream, which calls ainvoke.

Subclasses must override this method if they support streaming output.

PARAMETER	DESCRIPTION
input	The input to the Runnable. TYPE: Input
config	The config to use for the Runnable. TYPE: RunnableConfig | None DEFAULT: None
**kwargs	Additional keyword arguments to pass to the Runnable. TYPE: Any | None DEFAULT: {}

YIELDS	DESCRIPTION
AsyncIterator[Output]	The output of the Runnable.

astream_log async

astream_log(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    diff: bool = True,
    with_streamed_output_list: bool = True,
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]

Stream all output from a Runnable, as reported to the callback system.

This includes all inner runs of LLMs, Retrievers, Tools, etc.

Output is streamed as Log objects, which include a list of Jsonpatch ops that describe how the state of the run has changed in each step, and the final state of the run.

The Jsonpatch ops can be applied in order to construct state.

PARAMETER	DESCRIPTION
input	The input to the Runnable. TYPE: Any
config	The config to use for the Runnable. TYPE: RunnableConfig | None DEFAULT: None
diff	Whether to yield diffs between each step or the current state. TYPE: bool DEFAULT: True
with_streamed_output_list	Whether to yield the streamed_output list. TYPE: bool DEFAULT: True
include_names	Only include logs with these names. TYPE: Sequence[str] | None DEFAULT: None
include_types	Only include logs with these types. TYPE: Sequence[str] | None DEFAULT: None
include_tags	Only include logs with these tags. TYPE: Sequence[str] | None DEFAULT: None
exclude_names	Exclude logs with these names. TYPE: Sequence[str] | None DEFAULT: None
exclude_types	Exclude logs with these types. TYPE: Sequence[str] | None DEFAULT: None
exclude_tags	Exclude logs with these tags. TYPE: Sequence[str] | None DEFAULT: None
**kwargs	Additional keyword arguments to pass to the Runnable. TYPE: Any DEFAULT: {}

YIELDS	DESCRIPTION
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]	A RunLogPatch or RunLog object.

astream_events async

astream_events(
    input: Any,
    config: RunnableConfig | None = None,
    *,
    version: Literal["v1", "v2"] = "v2",
    include_names: Sequence[str] | None = None,
    include_types: Sequence[str] | None = None,
    include_tags: Sequence[str] | None = None,
    exclude_names: Sequence[str] | None = None,
    exclude_types: Sequence[str] | None = None,
    exclude_tags: Sequence[str] | None = None,
    **kwargs: Any,
) -> AsyncIterator[StreamEvent]

Generate a stream of events.

Use to create an iterator over StreamEvent that provide real-time information about the progress of the Runnable, including StreamEvent from intermediate results.

A StreamEvent is a dictionary with the following schema:

• event: Event names are of the format: on_[runnable_type]_(start|stream|end).
• name: The name of the Runnable that generated the event.
• run_id: Randomly generated ID associated with the given execution of the Runnable that emitted the event. A child Runnable that gets invoked as part of the execution of a parent Runnable is assigned its own unique ID.
• parent_ids: The IDs of the parent runnables that generated the event. The root Runnable will have an empty list. The order of the parent IDs is from the root to the immediate parent. Only available for v2 version of the API. The v1 version of the API will return an empty list.
• tags: The tags of the Runnable that generated the event.
• metadata: The metadata of the Runnable that generated the event.
• data: The data associated with the event. The contents of this field depend on the type of event. See the table below for more details.

Below is a table that illustrates some events that might be emitted by various chains. Metadata fields have been omitted from the table for brevity. Chain definitions have been included after the table.

Note

This reference table is for the v2 version of the schema.

event	name	chunk	input	output
on_chat_model_start	'[model name]'		{"messages": [[SystemMessage, HumanMessage]]}
on_chat_model_stream	'[model name]'	AIMessageChunk(content="hello")
on_chat_model_end	'[model name]'		{"messages": [[SystemMessage, HumanMessage]]}	AIMessageChunk(content="hello world")
on_llm_start	'[model name]'		{'input': 'hello'}
on_llm_stream	'[model name]'	'Hello'
on_llm_end	'[model name]'		'Hello human!'
on_chain_start	'format_docs'
on_chain_stream	'format_docs'	'hello world!, goodbye world!'
on_chain_end	'format_docs'		[Document(...)]	'hello world!, goodbye world!'
on_tool_start	'some_tool'		{"x": 1, "y": "2"}
on_tool_end	'some_tool'			{"x": 1, "y": "2"}
on_retriever_start	'[retriever name]'		{"query": "hello"}
on_retriever_end	'[retriever name]'		{"query": "hello"}	[Document(...), ..]
on_prompt_start	'[template_name]'		{"question": "hello"}
on_prompt_end	'[template_name]'		{"question": "hello"}	ChatPromptValue(messages: [SystemMessage, ...])

In addition to the standard events, users can also dispatch custom events (see example below).

Custom events will be only be surfaced with in the v2 version of the API!

A custom event has following format:

Attribute	Type	Description
name	str	A user defined name for the event.
data	Any	The data associated with the event. This can be anything, though we suggest making it JSON serializable.

Here are declarations associated with the standard events shown above:

format_docs:

def format_docs(docs: list[Document]) -> str:
    '''Format the docs.'''
    return ", ".join([doc.page_content for doc in docs])


format_docs = RunnableLambda(format_docs)

some_tool:

@tool
def some_tool(x: int, y: str) -> dict:
    '''Some_tool.'''
    return {"x": x, "y": y}

prompt:

template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are Cat Agent 007"),
        ("human", "{question}"),
    ]
).with_config({"run_name": "my_template", "tags": ["my_template"]})

Example

from langchain_core.runnables import RunnableLambda


async def reverse(s: str) -> str:
    return s[::-1]


chain = RunnableLambda(func=reverse)

events = [
    event async for event in chain.astream_events("hello", version="v2")
]

# Will produce the following events
# (run_id, and parent_ids has been omitted for brevity):
[
    {
        "data": {"input": "hello"},
        "event": "on_chain_start",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"chunk": "olleh"},
        "event": "on_chain_stream",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"output": "olleh"},
        "event": "on_chain_end",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
]

Dispatch custom event

from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio


async def slow_thing(some_input: str, config: RunnableConfig) -> str:
    """Do something that takes a long time."""
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 1 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 2 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    return "Done"

slow_thing = RunnableLambda(slow_thing)

async for event in slow_thing.astream_events("some_input", version="v2"):
    print(event)

PARAMETER	DESCRIPTION
input	The input to the Runnable. TYPE: Any
config	The config to use for the Runnable. TYPE: RunnableConfig | None DEFAULT: None
version	The version of the schema to use, either 'v2' or 'v1'. Users should use 'v2'. 'v1' is for backwards compatibility and will be deprecated in 0.4.0. No default will be assigned until the API is stabilized. custom events will only be surfaced in 'v2'. TYPE: Literal['v1', 'v2'] DEFAULT: 'v2'
include_names	Only include events from Runnable objects with matching names. TYPE: Sequence[str] | None DEFAULT: None
include_types	Only include events from Runnable objects with matching types. TYPE: Sequence[str] | None DEFAULT: None
include_tags	Only include events from Runnable objects with matching tags. TYPE: Sequence[str] | None DEFAULT: None
exclude_names	Exclude events from Runnable objects with matching names. TYPE: Sequence[str] | None DEFAULT: None
exclude_types	Exclude events from Runnable objects with matching types. TYPE: Sequence[str] | None DEFAULT: None
exclude_tags	Exclude events from Runnable objects with matching tags. TYPE: Sequence[str] | None DEFAULT: None
**kwargs	Additional keyword arguments to pass to the Runnable. These will be passed to astream_log as this implementation of astream_events is built on top of astream_log. TYPE: Any DEFAULT: {}

YIELDS	DESCRIPTION
AsyncIterator[StreamEvent]	An async stream of StreamEvent.

RAISES	DESCRIPTION
NotImplementedError	If the version is not 'v1' or 'v2'.

transform

transform(
    input: Iterator[Input], config: RunnableConfig | None = None, **kwargs: Any | None
) -> Iterator[Output]

Transform inputs to outputs.

Default implementation of transform, which buffers input and calls astream.

Subclasses must override this method if they can start producing output while input is still being generated.

PARAMETER	DESCRIPTION
input	An iterator of inputs to the Runnable. TYPE: Iterator[Input]
config	The config to use for the Runnable. TYPE: RunnableConfig | None DEFAULT: None
**kwargs	Additional keyword arguments to pass to the Runnable. TYPE: Any | None DEFAULT: {}

YIELDS	DESCRIPTION
Output	The output of the Runnable.

atransform async

atransform(
    input: AsyncIterator[Input],
    config: RunnableConfig | None = None,
    **kwargs: Any | None,
) -> AsyncIterator[Output]

Transform inputs to outputs.

Default implementation of atransform, which buffers input and calls astream.

Subclasses must override this method if they can start producing output while input is still being generated.

PARAMETER	DESCRIPTION
input	An async iterator of inputs to the Runnable. TYPE: AsyncIterator[Input]
config	The config to use for the Runnable. TYPE: RunnableConfig | None DEFAULT: None
**kwargs	Additional keyword arguments to pass to the Runnable. TYPE: Any | None DEFAULT: {}

YIELDS	DESCRIPTION
AsyncIterator[Output]	The output of the Runnable.

bind

bind(**kwargs: Any) -> Runnable[Input, Output]

Bind arguments to a Runnable, returning a new Runnable.

Useful when a Runnable in a chain requires an argument that is not in the output of the previous Runnable or included in the user input.

PARAMETER	DESCRIPTION
**kwargs	The arguments to bind to the Runnable. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Runnable[Input, Output]	A new Runnable with the arguments bound.

Example

from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser

model = ChatOllama(model="llama3.1")

# Without bind
chain = model | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'

# With bind
chain = model.bind(stop=["three"]) | StrOutputParser()

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'

with_config

with_config(
    config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]

Bind config to a Runnable, returning a new Runnable.

PARAMETER	DESCRIPTION
config	The config to bind to the Runnable. TYPE: RunnableConfig | None DEFAULT: None
**kwargs	Additional keyword arguments to pass to the Runnable. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Runnable[Input, Output]	A new Runnable with the config bound.

with_listeners

with_listeners(
    *,
    on_start: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
    on_end: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None = None,
    on_error: Callable[[Run], None]
    | Callable[[Run, RunnableConfig], None]
    | None = None,
) -> Runnable[Input, Output]

Bind lifecycle listeners to a Runnable, returning a new Runnable.

The Run object contains information about the run, including its id, type, input, output, error, start_time, end_time, and any tags or metadata added to the run.

PARAMETER	DESCRIPTION
on_start	Called before the Runnable starts running, with the Run object. TYPE: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None DEFAULT: None
on_end	Called after the Runnable finishes running, with the Run object. TYPE: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None DEFAULT: None
on_error	Called if the Runnable throws an error, with the Run object. TYPE: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None DEFAULT: None

RETURNS	DESCRIPTION
Runnable[Input, Output]	A new Runnable with the listeners bound.

Example

from langchain_core.runnables import RunnableLambda
from langchain_core.tracers.schemas import Run

import time


def test_runnable(time_to_sleep: int):
    time.sleep(time_to_sleep)


def fn_start(run_obj: Run):
    print("start_time:", run_obj.start_time)


def fn_end(run_obj: Run):
    print("end_time:", run_obj.end_time)


chain = RunnableLambda(test_runnable).with_listeners(
    on_start=fn_start, on_end=fn_end
)
chain.invoke(2)

with_alisteners

with_alisteners(
    *,
    on_start: AsyncListener | None = None,
    on_end: AsyncListener | None = None,
    on_error: AsyncListener | None = None,
) -> Runnable[Input, Output]

Bind async lifecycle listeners to a Runnable.

Returns a new Runnable.

The Run object contains information about the run, including its id, type, input, output, error, start_time, end_time, and any tags or metadata added to the run.

PARAMETER	DESCRIPTION
on_start	Called asynchronously before the Runnable starts running, with the Run object. TYPE: AsyncListener | None DEFAULT: None
on_end	Called asynchronously after the Runnable finishes running, with the Run object. TYPE: AsyncListener | None DEFAULT: None
on_error	Called asynchronously if the Runnable throws an error, with the Run object. TYPE: AsyncListener | None DEFAULT: None

RETURNS	DESCRIPTION
Runnable[Input, Output]	A new Runnable with the listeners bound.

Example

from langchain_core.runnables import RunnableLambda, Runnable
from datetime import datetime, timezone
import time
import asyncio


def format_t(timestamp: float) -> str:
    return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()


async def test_runnable(time_to_sleep: int):
    print(f"Runnable[{time_to_sleep}s]: starts at {format_t(time.time())}")
    await asyncio.sleep(time_to_sleep)
    print(f"Runnable[{time_to_sleep}s]: ends at {format_t(time.time())}")


async def fn_start(run_obj: Runnable):
    print(f"on start callback starts at {format_t(time.time())}")
    await asyncio.sleep(3)
    print(f"on start callback ends at {format_t(time.time())}")


async def fn_end(run_obj: Runnable):
    print(f"on end callback starts at {format_t(time.time())}")
    await asyncio.sleep(2)
    print(f"on end callback ends at {format_t(time.time())}")


runnable = RunnableLambda(test_runnable).with_alisteners(
    on_start=fn_start, on_end=fn_end
)


async def concurrent_runs():
    await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))


asyncio.run(concurrent_runs())
# Result:
# on start callback starts at 2025-03-01T07:05:22.875378+00:00
# on start callback starts at 2025-03-01T07:05:22.875495+00:00
# on start callback ends at 2025-03-01T07:05:25.878862+00:00
# on start callback ends at 2025-03-01T07:05:25.878947+00:00
# Runnable[2s]: starts at 2025-03-01T07:05:25.879392+00:00
# Runnable[3s]: starts at 2025-03-01T07:05:25.879804+00:00
# Runnable[2s]: ends at 2025-03-01T07:05:27.881998+00:00
# on end callback starts at 2025-03-01T07:05:27.882360+00:00
# Runnable[3s]: ends at 2025-03-01T07:05:28.881737+00:00
# on end callback starts at 2025-03-01T07:05:28.882428+00:00
# on end callback ends at 2025-03-01T07:05:29.883893+00:00
# on end callback ends at 2025-03-01T07:05:30.884831+00:00

with_types

with_types(
    *, input_type: type[Input] | None = None, output_type: type[Output] | None = None
) -> Runnable[Input, Output]

Bind input and output types to a Runnable, returning a new Runnable.

PARAMETER	DESCRIPTION
input_type	The input type to bind to the Runnable. TYPE: type[Input] | None DEFAULT: None
output_type	The output type to bind to the Runnable. TYPE: type[Output] | None DEFAULT: None

RETURNS	DESCRIPTION
Runnable[Input, Output]	A new Runnable with the types bound.

with_retry

with_retry(
    *,
    retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
    wait_exponential_jitter: bool = True,
    exponential_jitter_params: ExponentialJitterParams | None = None,
    stop_after_attempt: int = 3,
) -> Runnable[Input, Output]

Create a new Runnable that retries the original Runnable on exceptions.

PARAMETER	DESCRIPTION
retry_if_exception_type	A tuple of exception types to retry on. TYPE: tuple[type[BaseException], ...] DEFAULT: (Exception,)
wait_exponential_jitter	Whether to add jitter to the wait time between retries. TYPE: bool DEFAULT: True
stop_after_attempt	The maximum number of attempts to make before giving up. TYPE: int DEFAULT: 3
exponential_jitter_params	Parameters for tenacity.wait_exponential_jitter. Namely: initial, max, exp_base, and jitter (all float values). TYPE: ExponentialJitterParams | None DEFAULT: None

RETURNS	DESCRIPTION
Runnable[Input, Output]	A new Runnable that retries the original Runnable on exceptions.

Example

from langchain_core.runnables import RunnableLambda

count = 0


def _lambda(x: int) -> None:
    global count
    count = count + 1
    if x == 1:
        raise ValueError("x is 1")
    else:
        pass


runnable = RunnableLambda(_lambda)
try:
    runnable.with_retry(
        stop_after_attempt=2,
        retry_if_exception_type=(ValueError,),
    ).invoke(1)
except ValueError:
    pass

assert count == 2

map

map() -> Runnable[list[Input], list[Output]]

Return a new Runnable that maps a list of inputs to a list of outputs.

Calls invoke with each input.

RETURNS	DESCRIPTION
Runnable[list[Input], list[Output]]	A new Runnable that maps a list of inputs to a list of outputs.

Example

from langchain_core.runnables import RunnableLambda


def _lambda(x: int) -> int:
    return x + 1


runnable = RunnableLambda(_lambda)
print(runnable.map().invoke([1, 2, 3]))  # [2, 3, 4]

with_fallbacks

with_fallbacks(
    fallbacks: Sequence[Runnable[Input, Output]],
    *,
    exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
    exception_key: str | None = None,
) -> RunnableWithFallbacks[Input, Output]

Add fallbacks to a Runnable, returning a new Runnable.

The new Runnable will try the original Runnable, and then each fallback in order, upon failures.

PARAMETER	DESCRIPTION
fallbacks	A sequence of runnables to try if the original Runnable fails. TYPE: Sequence[Runnable[Input, Output]]
exceptions_to_handle	A tuple of exception types to handle. TYPE: tuple[type[BaseException], ...] DEFAULT: (Exception,)
exception_key	If string is specified then handled exceptions will be passed to fallbacks as part of the input under the specified key. If None, exceptions will not be passed to fallbacks. If used, the base Runnable and its fallbacks must accept a dictionary as input. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
RunnableWithFallbacks[Input, Output]	A new Runnable that will try the original Runnable, and then each Fallback in order, upon failures.

Example

from typing import Iterator

from langchain_core.runnables import RunnableGenerator


def _generate_immediate_error(input: Iterator) -> Iterator[str]:
    raise ValueError()
    yield ""


def _generate(input: Iterator) -> Iterator[str]:
    yield from "foo bar"


runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
    [RunnableGenerator(_generate)]
)
print("".join(runnable.stream({})))  # foo bar

PARAMETER	DESCRIPTION
fallbacks	A sequence of runnables to try if the original Runnable fails. TYPE: Sequence[Runnable[Input, Output]]
exceptions_to_handle	A tuple of exception types to handle. TYPE: tuple[type[BaseException], ...] DEFAULT: (Exception,)
exception_key	If string is specified then handled exceptions will be passed to fallbacks as part of the input under the specified key. If None, exceptions will not be passed to fallbacks. If used, the base Runnable and its fallbacks must accept a dictionary as input. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
RunnableWithFallbacks[Input, Output]	A new Runnable that will try the original Runnable, and then each Fallback in order, upon failures.

as_tool

as_tool(
    args_schema: type[BaseModel] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    arg_types: dict[str, type] | None = None,
) -> BaseTool

Create a BaseTool from a Runnable.

as_tool will instantiate a BaseTool with a name, description, and args_schema from a Runnable. Where possible, schemas are inferred from runnable.get_input_schema.

Alternatively (e.g., if the Runnable takes a dict as input and the specific dict keys are not typed), the schema can be specified directly with args_schema.

You can also pass arg_types to just specify the required arguments and their types.

PARAMETER	DESCRIPTION
args_schema	The schema for the tool. TYPE: type[BaseModel] | None DEFAULT: None
name	The name of the tool. TYPE: str | None DEFAULT: None
description	The description of the tool. TYPE: str | None DEFAULT: None
arg_types	A dictionary of argument names to types. TYPE: dict[str, type] | None DEFAULT: None

RETURNS	DESCRIPTION
BaseTool	A BaseTool instance.

TypedDict input

from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda


class Args(TypedDict):
    a: int
    b: list[int]


def f(x: Args) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})

dict input, specifying schema via args_schema

from typing import Any
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda

def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))

class FSchema(BaseModel):
    """Apply a function to an integer and list of integers."""

    a: int = Field(..., description="Integer")
    b: list[int] = Field(..., description="List of ints")

runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})

dict input, specifying schema via arg_types

from typing import Any
from langchain_core.runnables import RunnableLambda


def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})

str input

from langchain_core.runnables import RunnableLambda


def f(x: str) -> str:
    return x + "a"


def g(x: str) -> str:
    return x + "z"


runnable = RunnableLambda(f) | g
as_tool = runnable.as_tool()
as_tool.invoke("b")

get_lc_namespace classmethod

get_lc_namespace() -> list[str]

Get the namespace of the LangChain object.

For example, if the class is langchain.llms.openai.OpenAI, then the namespace is ["langchain", "llms", "openai"]

RETURNS	DESCRIPTION
list[str]	The namespace.

lc_id classmethod

lc_id() -> list[str]

Return a unique identifier for this class for serialization purposes.

The unique identifier is a list of strings that describes the path to the object.

For example, for the class langchain.llms.openai.OpenAI, the id is ["langchain", "llms", "openai", "OpenAI"].

to_json

to_json() -> SerializedConstructor | SerializedNotImplemented

Serialize the Runnable to JSON.

RETURNS	DESCRIPTION
SerializedConstructor | SerializedNotImplemented	A JSON-serializable representation of the Runnable.

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

Serialize a "not implemented" object.

RETURNS	DESCRIPTION
SerializedNotImplemented	SerializedNotImplemented.

configurable_fields

configurable_fields(
    **kwargs: AnyConfigurableField,
) -> RunnableSerializable[Input, Output]

Configure particular Runnable fields at runtime.

PARAMETER	DESCRIPTION
**kwargs	A dictionary of ConfigurableField instances to configure. TYPE: AnyConfigurableField DEFAULT: {}

RAISES	DESCRIPTION
ValueError	If a configuration key is not found in the Runnable.

RETURNS	DESCRIPTION
RunnableSerializable[Input, Output]	A new Runnable with the fields configured.

Example

from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(max_tokens=20).configurable_fields(
    max_tokens=ConfigurableField(
        id="output_token_number",
        name="Max tokens in the output",
        description="The maximum number of tokens in the output",
    )
)

# max_tokens = 20
print(
    "max_tokens_20: ", model.invoke("tell me something about chess").content
)

# max_tokens = 200
print(
    "max_tokens_200: ",
    model.with_config(configurable={"output_token_number": 200})
    .invoke("tell me something about chess")
    .content,
)

configurable_alternatives

configurable_alternatives(
    which: ConfigurableField,
    *,
    default_key: str = "default",
    prefix_keys: bool = False,
    **kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]],
) -> RunnableSerializable[Input, Output]

Configure alternatives for Runnable objects that can be set at runtime.

PARAMETER	DESCRIPTION
which	The ConfigurableField instance that will be used to select the alternative. TYPE: ConfigurableField
default_key	The default key to use if no alternative is selected. TYPE: str DEFAULT: 'default'
prefix_keys	Whether to prefix the keys with the ConfigurableField id. TYPE: bool DEFAULT: False
**kwargs	A dictionary of keys to Runnable instances or callables that return Runnable instances. TYPE: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]] DEFAULT: {}

RETURNS	DESCRIPTION
RunnableSerializable[Input, Output]	A new Runnable with the alternatives configured.

Example

from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatAnthropic(
    model_name="claude-sonnet-4-5-20250929"
).configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="anthropic",
    openai=ChatOpenAI(),
)

# uses the default model ChatAnthropic
print(model.invoke("which organization created you?").content)

# uses ChatOpenAI
print(
    model.with_config(configurable={"llm": "openai"})
    .invoke("which organization created you?")
    .content
)

set_verbose

set_verbose(verbose: bool | None) -> bool

If verbose is None, set it.

This allows users to pass in None as verbose to access the global setting.

PARAMETER	DESCRIPTION
verbose	The verbosity setting to use. TYPE: bool | None

RETURNS	DESCRIPTION
bool	The verbosity setting to use.

generate_prompt

generate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = 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).

PARAMETER	DESCRIPTION
prompts	List of PromptValue objects. A PromptValue is an object that can be converted to match the format of any language model (string for pure text generation models and BaseMessage objects for chat models). TYPE: list[PromptValue]
stop	Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. TYPE: list[str] | None DEFAULT: None
callbacks	Callbacks to pass through. Used for executing additional functionality, such as logging or streaming, throughout generation. TYPE: Callbacks DEFAULT: None
**kwargs	Arbitrary additional keyword arguments. These are usually passed to the model provider API call. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
LLMResult	An LLMResult, which contains a list of candidate Generation objects for each input prompt and additional model provider-specific output.

agenerate_prompt async

agenerate_prompt(
    prompts: list[PromptValue],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> LLMResult

Asynchronously pass a sequence of prompts 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).

PARAMETER	DESCRIPTION
prompts	List of PromptValue objects. A PromptValue is an object that can be converted to match the format of any language model (string for pure text generation models and BaseMessage objects for chat models). TYPE: list[PromptValue]
stop	Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. TYPE: list[str] | None DEFAULT: None
callbacks	Callbacks to pass through. Used for executing additional functionality, such as logging or streaming, throughout generation. TYPE: Callbacks DEFAULT: None
**kwargs	Arbitrary additional keyword arguments. These are usually passed to the model provider API call. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
LLMResult	An LLMResult, which contains a list of candidate Generation objects for each input prompt and additional model provider-specific output.

get_token_ids

get_token_ids(text: str) -> list[int]

Return the ordered IDs of the tokens in a text.

PARAMETER	DESCRIPTION
text	The string input to tokenize. TYPE: str

RETURNS	DESCRIPTION
list[int]	A list of IDs corresponding to the tokens in the text, in order they occur in the text.

get_num_tokens_from_messages

get_num_tokens_from_messages(
    messages: list[BaseMessage], tools: Sequence | None = None
) -> int

Get the number of tokens in the messages.

Useful for checking if an input fits in a model's context window.

This should be overridden by model-specific implementations to provide accurate token counts via model-specific tokenizers.

Note

• The base implementation of get_num_tokens_from_messages ignores tool schemas.
• The base implementation of get_num_tokens_from_messages adds additional prefixes to messages in represent user roles, which will add to the overall token count. Model-specific implementations may choose to handle this differently.

PARAMETER	DESCRIPTION
messages	The message inputs to tokenize. TYPE: list[BaseMessage]
tools	If provided, sequence of dict, BaseModel, function, or BaseTool objects to be converted to tool schemas. TYPE: Sequence | None DEFAULT: None

RETURNS	DESCRIPTION
int	The sum of the number of tokens across the messages.

generate

generate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: 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).

PARAMETER	DESCRIPTION
messages	List of list of messages. TYPE: list[list[BaseMessage]]
stop	Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. TYPE: list[str] | None DEFAULT: None
callbacks	Callbacks to pass through. Used for executing additional functionality, such as logging or streaming, throughout generation. TYPE: Callbacks DEFAULT: None
tags	The tags to apply. TYPE: list[str] | None DEFAULT: None
metadata	The metadata to apply. TYPE: dict[str, Any] | None DEFAULT: None
run_name	The name of the run. TYPE: str | None DEFAULT: None
run_id	The ID of the run. TYPE: UUID | None DEFAULT: None
**kwargs	Arbitrary additional keyword arguments. These are usually passed to the model provider API call. TYPE: Any DEFAULT: {}

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

agenerate async

agenerate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: 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).

PARAMETER	DESCRIPTION
messages	List of list of messages. TYPE: list[list[BaseMessage]]
stop	Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. TYPE: list[str] | None DEFAULT: None
callbacks	Callbacks to pass through. Used for executing additional functionality, such as logging or streaming, throughout generation. TYPE: Callbacks DEFAULT: None
tags	The tags to apply. TYPE: list[str] | None DEFAULT: None
metadata	The metadata to apply. TYPE: dict[str, Any] | None DEFAULT: None
run_name	The name of the run. TYPE: str | None DEFAULT: None
run_id	The ID of the run. TYPE: UUID | None DEFAULT: None
**kwargs	Arbitrary additional keyword arguments. These are usually passed to the model provider API call. TYPE: Any DEFAULT: {}

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

dict

dict(**kwargs: Any) -> dict

Return a dictionary of the LLM.

__init__

__init__(**kwargs: Any) -> None

Needed for arg validation.

is_lc_serializable classmethod

is_lc_serializable() -> bool

Is this class serializable?

By design, even if a class inherits from Serializable, it is not serializable by default. This is to prevent accidental serialization of objects that should not be serialized.

RETURNS	DESCRIPTION
bool	Whether the class is serializable. Default is False.

build_extra classmethod

build_extra(values: dict[str, Any]) -> Any

Build extra kwargs from additional params that were passed in.

(In other words, handle additional params that aren't explicitly defined as model fields. Used to pass extra config to underlying APIs without defining them all here.)

validate_environment

validate_environment() -> Self

Validates params and builds client.

We override temperature to 1.0 for Gemini 3+ models if not explicitly set. This is to prevent infinite loops and degraded performance that can occur with temperature < 1.0 on these models.

__del__

__del__() -> None

Clean up the client on deletion.

invoke

invoke(
    input: LanguageModelInput,
    config: RunnableConfig | None = None,
    *,
    code_execution: bool | None = None,
    stop: list[str] | None = None,
    **kwargs: Any,
) -> AIMessage

Override invoke on ChatGoogleGenerativeAI to add code_execution.

get_num_tokens

get_num_tokens(text: str) -> int

Get the number of tokens present in the text. Uses the model's tokenizer.

Useful for checking if an input will fit in a model's context window.

PARAMETER	DESCRIPTION
text	The string input to tokenize. TYPE: str

RETURNS	DESCRIPTION
int	The integer number of tokens in the text.

Example

llm = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")
num_tokens = llm.get_num_tokens("Hello, world!")
print(num_tokens)
# -> 4

with_structured_output

with_structured_output(
    schema: dict | type[BaseModel],
    method: Literal["function_calling", "json_mode", "json_schema"]
    | None = "json_schema",
    *,
    include_raw: bool = False,
    **kwargs: Any,
) -> Runnable[LanguageModelInput, dict | BaseModel]

Return a Runnable that constrains model output to a given schema.

Constrains the model to return output conforming to the provided schema.

Supports Pydantic models, TypedDict, and JSON schema dictionaries.

PARAMETER	DESCRIPTION
schema	The output schema as a Pydantic BaseModel class, a TypedDict class, or a JSON schema dictionary. TYPE: dict | type[BaseModel]
method	The method to use for structured output. Options: 'json_schema' (recommended): Uses native JSON schema support for reliable structured output. Supports streaming with fully-parsed Pydantic objects. 'json_mode': Deprecated alias for 'json_schema'. 'function_calling': Uses tool/function calling. Less reliable than 'json_schema' and not recommended for new code. TYPE: Literal['function_calling', 'json_mode', 'json_schema'] | None DEFAULT: 'json_schema'
include_raw	If True, returns a dict with both the raw model output and the parsed structured output. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
Runnable[LanguageModelInput, dict | BaseModel]	A Runnable that takes the same input as the chat model but returns the structured output. When streaming, emits fully-parsed objects of the specified schema type (not incremental JSON strings).

Example

Basic usage with Pydantic model

from pydantic import BaseModel
from langchain_google_genai import ChatGoogleGenerativeAI


class Person(BaseModel):
    name: str
    age: int


model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")
structured_model = model.with_structured_output(
    Person,
    method="json_schema",
)

result = structured_model.invoke(
    "Tell me about a person named Alice, age 30"
)
print(result)  # Person(name="Alice", age=30)

Streaming structured output

from pydantic import BaseModel
from langchain_google_genai import ChatGoogleGenerativeAI


class Recipe(BaseModel):
    name: str
    ingredients: list[str]
    steps: list[str]


model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")
structured_model = model.with_structured_output(
    Recipe, method="json_schema"
)

# Emits fully-parsed Recipe objects, not incremental JSON strings
for chunk in structured_model.stream(
    "Give me a recipe for chocolate chip cookies"
):
    print(chunk)  # Recipe(name=..., ingredients=[...], steps=[...])

Using with dict schema

model = ChatGoogleGenerativeAI(model="gemini-3-pro-preview")

schema = {
    "type": "object",
    "properties": {
        "title": {"type": "string"},
        "priority": {"type": "integer"},
    },
    "required": ["title", "priority"],
}

structured_model = model.with_structured_output(
    schema, method="json_schema"
)
result = structured_model.invoke("Create a task: finish report, priority 1")
print(result)  # {"title": "finish report", "priority": 1}

Including raw output

structured_model = model.with_structured_output(
    Person, method="json_schema", include_raw=True
)

result = structured_model.invoke("Tell me about Bob, age 25")
print(result["parsed"])  # Person(name="Bob", age=25)
print(result["raw"])  # AIMessage with full model response

bind_tools

bind_tools(
    tools: Sequence[dict[str, Any] | type | Callable[..., Any] | BaseTool | Tool],
    tool_config: dict | ToolConfig | None = None,
    *,
    tool_choice: _ToolChoiceType | bool | None = None,
    **kwargs: Any,
) -> Runnable[LanguageModelInput, AIMessage]

Bind tool-like objects to this chat model.

PARAMETER	DESCRIPTION
tools	A list of tool definitions to bind to this chat model. Can be a pydantic model, Callable, or BaseTool. Pydantic models, Callable, and BaseTool objects will be automatically converted to their schema dictionary representation. Tools with Union types in their arguments are now supported and converted to anyOf schemas. TYPE: Sequence[dict[str, Any] | type | Callable[..., Any] | BaseTool | Tool]
tool_config	Optional tool configuration for additional settings like retrieval_config (for Google Maps/Google Search grounding). Can be used together with tool_choice, but cannot specify function_calling_config in tool_config if tool_choice is also provided (they would conflict). Example with Google Maps grounding from langchain_google_genai import ChatGoogleGenerativeAI model = ChatGoogleGenerativeAI(model="gemini-2.5-pro") response = model.invoke( "What Italian restaurants are near here?", tools=[{"google_maps": {}}], tool_choice="required", tool_config={ "retrieval_config": { "lat_lng": { "latitude": 48.858844, "longitude": 2.294351, } } }, ) TYPE: dict | ToolConfig | None DEFAULT: None
tool_choice	Control how the model uses tools. Options: 'auto' (default): Model decides whether to call functions 'any' or 'required': Model must call a function (both are equivalent) 'none': Model cannot call functions 'function_name': Model must call the specified function ['fn1', 'fn2']: Model must call one of the specified functions True: Same as 'any' Can be used together with tool_config to control function calling while also providing additional configuration like retrieval_config. TYPE: _ToolChoiceType | bool | None DEFAULT: None
**kwargs	Any additional parameters to pass to the Runnable constructor. TYPE: Any DEFAULT: {}