langchain-google-genai¶
LangChain Google Generative AI Integration.
This module integrates Google's Generative AI models, specifically the Gemini series, with the LangChain framework. It provides classes for interacting with chat models and generating embeddings, leveraging Google's advanced AI capabilities.
Chat Models
The ChatGoogleGenerativeAI class is the primary interface for interacting with Google's Gemini chat models. It allows users to send and receive messages using a specified Gemini model, suitable for various conversational AI applications.
LLMs
The GoogleGenerativeAI class is the primary interface for interacting with Google's Gemini LLMs. It allows users to generate text using a specified Gemini model.
Embeddings
The GoogleGenerativeAIEmbeddings class provides functionalities to generate embeddings using Google's models.
These embeddings can be used for a range of NLP tasks, including semantic analysis, similarity comparisons, and more.
Installation
To install the package, use pip:
.. code-block:: python pip install -U langchain-google-genai
Using Chat Models
After setting up your environment with the required API key, you can interact with the Google Gemini models.
.. code-block:: python
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(model="gemini-2.5-pro")
llm.invoke("Sing a ballad of LangChain.")
Using LLMs
The package also supports generating text with Google's models.
.. code-block:: python
from langchain_google_genai import GoogleGenerativeAI
llm = GoogleGenerativeAI(model="gemini-2.5-pro")
llm.invoke("Once upon a time, a library called LangChain")
Embedding Generation
The package also supports creating embeddings with Google's models, useful for textual similarity and other NLP applications.
.. code-block:: python
from langchain_google_genai import GoogleGenerativeAIEmbeddings
embeddings = GoogleGenerativeAIEmbeddings(model="models/gemini-embedding-001")
embeddings.embed_query("hello, world!")
Modules:
| Name | Description |
|---|---|
chat_models |
|
embeddings |
|
genai_aqa |
Google GenerativeAI Attributed Question and Answering (AQA) service. |
google_vector_store |
Google Generative AI Vector Store. |
llms |
|
Classes:
| Name | Description |
|---|---|
ChatGoogleGenerativeAI |
|
GoogleGenerativeAIEmbeddings |
|
AqaInput |
Input to |
AqaOutput |
Output from |
GenAIAqa |
Google's Attributed Question and Answering service. |
GoogleVectorStore |
Google GenerativeAI Vector Store. |
GoogleGenerativeAI |
Google GenerativeAI models. |
Bases: _BaseGoogleGenerativeAI, BaseChatModel
Google AI chat models integration.
Instantiation
To use, you must have either:
1. The ``GOOGLE_API_KEY`` environment variable set with your API key, or
2. Pass your API key using the ``google_api_key`` kwarg to the
ChatGoogleGenerativeAI constructor.
.. code-block:: python
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash")
llm.invoke("Write me a ballad about LangChain")
Invoke
.. code-block:: python
messages = [
("system", "Translate the user sentence to French."),
("human", "I love programming."),
]
llm.invoke(messages)
.. code-block:: python
AIMessage(
content="J'adore programmer. \\n",
response_metadata={
"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,
},
],
},
id="run-56cecc34-2e54-4b52-a974-337e47008ad2-0",
usage_metadata={
"input_tokens": 18,
"output_tokens": 5,
"total_tokens": 23,
},
)
Stream
.. code-block:: python
for chunk in llm.stream(messages):
print(chunk)
.. code-block:: python
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,
},
)
.. code-block:: python
stream = llm.stream(messages)
full = next(stream)
for chunk in stream:
full += chunk
full
.. code-block:: python
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,
},
)
Async
.. code-block:: python
await llm.ainvoke(messages)
# stream:
# async for chunk in (await llm.astream(messages))
# batch:
# await llm.abatch([messages])
Context Caching
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. 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.
.. code-block:: python
from google import genai
from google.genai import types
import time
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage
client = genai.Client()
# Upload file
file = client.files.upload(file="./example_file")
while file.state.name == "PROCESSING":
time.sleep(2)
file = client.files.get(name=file.name)
# Create cache
model = "models/gemini-2.5-flash"
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.
.. code-block:: python
from google import genai
from google.genai.types import CreateCachedContentConfig, Content, Part
import time
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.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-2.5-flash"
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])
Tool calling
.. code-block:: python
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
.. code-block:: python
[
{
"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",
},
]
Use Search with Gemini 2
.. code-block:: python
from google.ai.generativelanguage_v1beta.types import Tool as GenAITool
llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash")
resp = llm.invoke(
"When is the next total solar eclipse in US?",
tools=[GenAITool(google_search={})],
)
Structured output
.. code-block:: python
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 function calling
structured_llm = llm.with_structured_output(Joke)
# For more reliable output, use json_schema with native responseSchema
structured_llm_json = llm.with_structured_output(Joke, method="json_schema")
structured_llm_json.invoke("Tell me a joke about cats")
.. code-block:: python
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="function_calling"(default): Uses tool calling to extract structured data. Compatible with all models.method="json_schema": Uses Gemini's native structured output with responseSchema. More reliable but requires Gemini 1.5+ models.method="json_mode"also works for backwards compatibility but is a misnomer.
The json_schema method is recommended for better reliability as it
constrains the model's generation process directly rather than relying on
post-processing tool calls.
Image input
.. code-block:: python
import base64
import httpx
from langchain_core.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 = llm.invoke([message])
ai_msg.content
.. code-block:: python
"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 signs of rain or storms."
PDF input
.. code-block:: python
import base64
from langchain_core.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 = llm.invoke([message])
ai_msg.content
.. code-block:: python
"This research paper describes a system developed for SemEval-2025 Task 9,
which aims to automate the detection of food hazards from recall reports,
addressing the class imbalance problem by leveraging LLM-based data
augmentation techniques and transformer-based models to improve
performance."
Video input
.. code-block:: python
import base64
from langchain_core.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 = llm.invoke([message])
ai_msg.content
.. code-block:: python
"Tom and Jerry, along with a turkey, engage in a chaotic Thanksgiving-themed
adventure involving a corn-on-the-cob chase, maze antics, and a disastrous
attempt to prepare a turkey dinner."
You can also pass YouTube URLs directly:
.. code-block:: python
from langchain_core.messages import HumanMessage
message = HumanMessage(
content=[
{"type": "text", "text": "summarize the video in 3 sentences."},
{
"type": "media",
"file_uri": "https://www.youtube.com/watch?v=9hE5-98ZeCg",
"mime_type": "video/mp4",
},
]
)
ai_msg = llm.invoke([message])
ai_msg.content
.. code-block:: python
"The video is a demo of multimodal live streaming in Gemini 2.0. The
narrator is sharing his screen in AI Studio and asks if the AI can see it.
The AI then reads text that is highlighted on the screen, defines the word
“multimodal,” and summarizes everything that was seen and heard."
Audio input
.. code-block:: python
import base64
from langchain_core.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 = llm.invoke([message])
ai_msg.content
.. code-block:: python
"In this episode of the Made by Google podcast, Stephen Johnson and Simon
Tokumine discuss NotebookLM, a tool designed to help users understand
complex material in various modalities, with a focus on its unexpected uses,
the development of audio overviews, and the implementation of new features
like mind maps and source discovery."
File upload (URI-based): You can also upload files to Google's servers and reference them by URI. This works for PDFs, images, videos, and audio files.
.. code-block:: python
import time
from google import genai
from langchain_core.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 = llm.invoke([message])
ai_msg.content
.. code-block:: python
"This research paper assesses and mitigates multi-turn jailbreak
vulnerabilities in large language models using the Crescendo attack study,
evaluating attack success rates and mitigation strategies like prompt
hardening and LLM-as-guardrail."
Token usage
.. code-block:: python
ai_msg = llm.invoke(messages)
ai_msg.usage_metadata
.. code-block:: python
{"input_tokens": 18, "output_tokens": 5, "total_tokens": 23}
Response metadata .. code-block:: python
ai_msg = llm.invoke(messages)
ai_msg.response_metadata
.. code-block:: python
{
"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,
},
],
}
Methods:
| Name | Description |
|---|---|
get_name |
Get the name of the |
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 |
get_output_schema |
Get a pydantic model that can be used to validate output to the |
get_output_jsonschema |
Get a JSON schema that represents the output of the |
config_schema |
The type of config this |
get_config_jsonschema |
Get a JSON schema that represents the config of the |
get_graph |
Return a graph representation of this |
get_prompts |
Return a list of prompts used by this |
__or__ |
Runnable "or" operator. |
__ror__ |
Runnable "reverse-or" operator. |
pipe |
Pipe runnables. |
pick |
Pick keys from the output dict of this |
assign |
Assigns new fields to the dict output of this |
batch |
Default implementation runs invoke in parallel using a thread pool executor. |
batch_as_completed |
Run |
abatch |
Default implementation runs |
abatch_as_completed |
Run |
astream_log |
Stream all output from a |
astream_events |
Generate a stream of events. |
transform |
Transform inputs to outputs. |
atransform |
Transform inputs to outputs. |
bind |
Bind arguments to a |
with_config |
Bind config to a |
with_listeners |
Bind lifecycle listeners to a |
with_alisteners |
Bind async lifecycle listeners to a |
with_types |
Bind input and output types to a |
with_retry |
Create a new Runnable that retries the original Runnable on exceptions. |
map |
Return a new |
with_fallbacks |
Add fallbacks to a |
as_tool |
Create a |
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 |
to_json_not_implemented |
Serialize a "not implemented" object. |
configurable_fields |
Configure particular |
configurable_alternatives |
Configure alternatives for |
set_verbose |
If verbose is None, set it. |
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. |
build_extra |
Build extra kwargs from additional params that were passed in. |
validate_environment |
Validates params and passes them to google-generativeai package. |
invoke |
Enable code execution. Supported on: gemini-1.5-pro, gemini-1.5-flash, |
get_num_tokens |
Get the number of tokens present in the text. |
bind_tools |
Bind tool-like objects to this chat model. |
Attributes:
| Name | Type | Description |
|---|---|---|
InputType |
TypeAlias
|
Get the input type for this runnable. |
OutputType |
Any
|
Get the output type for this runnable. |
input_schema |
type[BaseModel]
|
The type of input this |
output_schema |
type[BaseModel]
|
Output schema. |
config_specs |
list[ConfigurableFieldSpec]
|
List configurable fields for this |
lc_attributes |
dict
|
List of attribute names that should be included in the serialized kwargs. |
cache |
BaseCache | bool | None
|
Whether to cache the response. |
verbose |
bool
|
Whether to print out response text. |
callbacks |
Callbacks
|
Callbacks to add to the run trace. |
tags |
list[str] | None
|
Tags to add to the run trace. |
metadata |
dict[str, Any] | None
|
Metadata to add to the run trace. |
custom_get_token_ids |
Callable[[str], list[int]] | None
|
Optional encoder to use for counting tokens. |
rate_limiter |
BaseRateLimiter | None
|
An optional rate limiter to use for limiting the number of requests. |
disable_streaming |
bool | Literal['tool_calling']
|
Whether to disable streaming for this model. |
output_version |
str | None
|
Version of |
model |
str
|
Model name to use. |
google_api_key |
Optional[SecretStr]
|
Google AI API key. |
credentials |
Any
|
The default custom credentials (google.auth.credentials.Credentials) to use |
temperature |
float
|
Run inference with this temperature. Must be within |
top_p |
Optional[float]
|
Decode using nucleus sampling: consider the smallest set of tokens whose |
top_k |
Optional[int]
|
Decode using top-k sampling: consider the set of |
max_output_tokens |
Optional[int]
|
Maximum number of tokens to include in a candidate. Must be greater than zero. |
n |
int
|
Number of chat completions to generate for each prompt. Note that the API may |
max_retries |
int
|
The maximum number of retries to make when generating. If unset, will default |
timeout |
Optional[float]
|
The maximum number of seconds to wait for a response. |
safety_settings |
Optional[Dict[HarmCategory, HarmBlockThreshold]]
|
The default safety settings to use for all generations. |
convert_system_message_to_human |
bool
|
Whether to merge any leading SystemMessage into the following HumanMessage. |
response_mime_type |
Optional[str]
|
Optional. Output response mimetype of the generated candidate text. Only |
response_schema |
Optional[Dict[str, Any]]
|
Optional. Enforce an schema to the output. The format of the dictionary should |
cached_content |
Optional[str]
|
The name of the cached content used as context to serve the prediction. |
stop |
Optional[List[str]]
|
Stop sequences for the model. |
streaming |
Optional[bool]
|
Whether to stream responses from the model. |
model_kwargs |
dict[str, Any]
|
Holds any unexpected initialization parameters. |
property
¶
input_schema: type[BaseModel]
The type of input this Runnable accepts specified as a pydantic model.
property
¶
output_schema: type[BaseModel]
Output schema.
The type of output this Runnable produces specified as a pydantic model.
property
¶
config_specs: list[ConfigurableFieldSpec]
List configurable fields for this Runnable.
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.
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.
class-attribute
instance-attribute
¶
verbose: bool = Field(
default_factory=_get_verbosity, exclude=True, repr=False
)
Whether to print out response text.
class-attribute
instance-attribute
¶
Callbacks to add to the run trace.
class-attribute
instance-attribute
¶
Tags to add to the run trace.
class-attribute
instance-attribute
¶
Metadata to add to the run trace.
class-attribute
instance-attribute
¶
Optional encoder to use for counting tokens.
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.
class-attribute
instance-attribute
¶
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 atoolskeyword 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.
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 version 1.0
class-attribute
instance-attribute
¶
model: str = Field(
...,
description="The name of the model to use.\nExamples:\n - gemini-2.5-flash\n - models/text-bison-001",
)
Model name to use.
class-attribute
instance-attribute
¶
google_api_key: Optional[SecretStr] = Field(
alias="api_key",
default_factory=secret_from_env(
"GOOGLE_API_KEY", default=None
),
)
Google AI API key.
If not specified will be read from env var GOOGLE_API_KEY.
class-attribute
instance-attribute
¶
credentials: Any = None
The default custom credentials (google.auth.credentials.Credentials) to use
class-attribute
instance-attribute
¶
temperature: float = 0.7
Run inference with this temperature. Must be within [0.0, 2.0]. If unset,
will default to 0.7.
class-attribute
instance-attribute
¶
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].
class-attribute
instance-attribute
¶
Decode using top-k sampling: consider the set of top_k most probable tokens.
Must be positive.
class-attribute
instance-attribute
¶
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 https://ai.google.dev/gemini-api/docs/models for model-specific limits.
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.
class-attribute
instance-attribute
¶
max_retries: int = Field(default=6, alias='retries')
The maximum number of retries to make when generating. If unset, will default
to 6.
class-attribute
instance-attribute
¶
The maximum number of seconds to wait for a response.
class-attribute
instance-attribute
¶
The default safety settings to use for all generations.
For example:
.. code-block:: python from google.generativeai.types.safety_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,
}
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.
class-attribute
instance-attribute
¶
Optional. Output response mimetype of the generated candidate text. Only supported in Gemini 1.5 and later models.
Supported mimetype
'text/plain': (default) Text output.'application/json': JSON response in the candidates.'text/x.enum': Enum in plain text.
The model also needs to be prompted to output the appropriate response type, otherwise the behavior is undefined. This is a preview feature.
class-attribute
instance-attribute
¶
Optional. Enforce an schema to the output. The format of the dictionary should follow Open API schema.
class-attribute
instance-attribute
¶
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}.
class-attribute
instance-attribute
¶
Stop sequences for the model.
class-attribute
instance-attribute
¶
Whether to stream responses from the model.
class-attribute
instance-attribute
¶
Holds any unexpected initialization parameters.
get_input_schema(
config: RunnableConfig | None = None,
) -> type[BaseModel]
Get a pydantic model that can be used to validate input to the Runnable.
Runnables 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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
type[BaseModel]
|
A pydantic model that can be used to validate input. |
Get a JSON schema that represents the input to the Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A JSON schema that represents the input to the |
Example
Added in version 0.3.0
get_output_schema(
config: RunnableConfig | None = None,
) -> type[BaseModel]
Get a pydantic model that can be used to validate output to the Runnable.
Runnables 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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
type[BaseModel]
|
A pydantic model that can be used to validate output. |
Get a JSON schema that represents the output of the Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A JSON schema that represents the output of the |
Example
Added in version 0.3.0
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include
|
Sequence[str] | None
|
A list of fields to include in the config schema. |
None
|
Returns:
| Type | Description |
|---|---|
type[BaseModel]
|
A pydantic model that can be used to validate config. |
Get a JSON schema that represents the config of the Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include
|
Sequence[str] | None
|
A list of fields to include in the config schema. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A JSON schema that represents the config of the |
Added in version 0.3.0
Return a graph representation of this Runnable.
get_prompts(
config: RunnableConfig | None = None,
) -> list[BasePromptTemplate]
Return a list of prompts used by this Runnable.
__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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
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]
|
Another |
required |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
__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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
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]
|
Another |
required |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Other, Output]
|
A new |
pipe(
*others: Runnable[Any, Other] | Callable[[Any], Other],
name: str | None = None
) -> RunnableSerializable[Input, Other]
Pipe runnables.
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]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*others
|
Runnable[Any, Other] | Callable[[Any], Other]
|
Other |
()
|
name
|
str | None
|
An optional name for the resulting |
None
|
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
Pick keys from the output dict of this Runnable.
Pick single key:
```python
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 list of keys:
```python
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]"}
```
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
str | list[str]
|
A key or list of keys to pick from the output dict. |
required |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Any, Any]
|
a new |
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_community.llms.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}"
)
llm = FakeStreamingListLLM(responses=["foo-lish"])
chain: Runnable = prompt | llm | {"str": StrOutputParser()}
chain_with_assign = chain.assign(hello=itemgetter("str") | llm)
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'}}}
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any] | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]]
|
A mapping of keys to |
{}
|
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Any, Any]
|
A new |
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 should override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
list[Input]
|
A list of inputs to the |
required |
config
|
RunnableConfig | list[RunnableConfig] | None
|
A config to use when invoking the |
None
|
return_exceptions
|
bool
|
Whether to return exceptions instead of raising them. Defaults to False. |
False
|
**kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Returns:
| Type | Description |
|---|---|
list[Output]
|
A list of outputs from the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
Sequence[Input]
|
A list of inputs to the |
required |
config
|
RunnableConfig | Sequence[RunnableConfig] | None
|
A config to use when invoking the |
None
|
return_exceptions
|
bool
|
Whether to return exceptions instead of raising them. Defaults to False. |
False
|
**kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
tuple[int, Output | Exception]
|
Tuples of the index of the input and the output from the |
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 should override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
list[Input]
|
A list of inputs to the |
required |
config
|
RunnableConfig | list[RunnableConfig] | None
|
A config to use when invoking the |
None
|
return_exceptions
|
bool
|
Whether to return exceptions instead of raising them. Defaults to False. |
False
|
**kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Returns:
| Type | Description |
|---|---|
list[Output]
|
A list of outputs from the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
Sequence[Input]
|
A list of inputs to the |
required |
config
|
RunnableConfig | Sequence[RunnableConfig] | None
|
A config to use when invoking the |
None
|
return_exceptions
|
bool
|
Whether to return exceptions instead of raising them. Defaults to False. |
False
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[tuple[int, Output | Exception]]
|
A tuple of the index of the input and the output from the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Any
|
The input to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
diff
|
bool
|
Whether to yield diffs between each step or the current state. |
True
|
with_streamed_output_list
|
bool
|
Whether to yield the |
True
|
include_names
|
Sequence[str] | None
|
Only include logs with these names. |
None
|
include_types
|
Sequence[str] | None
|
Only include logs with these types. |
None
|
include_tags
|
Sequence[str] | None
|
Only include logs with these tags. |
None
|
exclude_names
|
Sequence[str] | None
|
Exclude logs with these names. |
None
|
exclude_types
|
Sequence[str] | None
|
Exclude logs with these types. |
None
|
exclude_tags
|
Sequence[str] | None
|
Exclude logs with these tags. |
None
|
kwargs
|
Any
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
|
A |
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 StreamEvents that provide real-time information
about the progress of the Runnable, including StreamEvents from intermediate
results.
A StreamEvent is a dictionary with the following schema:
event: str - Event names are of the format:on_[runnable_type]_(start|stream|end).name: str - The name of theRunnablethat generated the event.run_id: str - randomly generated ID associated with the given execution of theRunnablethat emitted the event. A childRunnablethat gets invoked as part of the execution of a parentRunnableis assigned its own unique ID.parent_ids: list[str] - The IDs of the parent runnables that generated the event. The rootRunnablewill 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: Optional[list[str]] - The tags of theRunnablethat generated the event.metadata: Optional[dict[str, Any]] - The metadata of theRunnablethat generated the event.data: dict[str, Any]
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:
prompt:
template = ChatPromptTemplate.from_messages(
[
("system", "You are Cat Agent 007"),
("human", "{question}"),
]
).with_config({"run_name": "my_template", "tags": ["my_template"]})
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": [],
},
]
Example: 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)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Any
|
The input to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
version
|
Literal['v1', 'v2']
|
The version of the schema to use either |
'v2'
|
include_names
|
Sequence[str] | None
|
Only include events from |
None
|
include_types
|
Sequence[str] | None
|
Only include events from |
None
|
include_tags
|
Sequence[str] | None
|
Only include events from |
None
|
exclude_names
|
Sequence[str] | None
|
Exclude events from |
None
|
exclude_types
|
Sequence[str] | None
|
Exclude events from |
None
|
exclude_tags
|
Sequence[str] | None
|
Exclude events from |
None
|
kwargs
|
Any
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[StreamEvent]
|
An async stream of |
Raises:
| Type | Description |
|---|---|
NotImplementedError
|
If the version is not |
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 should override this method if they can start producing output while input is still being generated.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Iterator[Input]
|
An iterator of inputs to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
Output
|
The output of the |
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 should override this method if they can start producing output while input is still being generated.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
AsyncIterator[Input]
|
An async iterator of inputs to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[Output]
|
The output of the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
kwargs
|
Any
|
The arguments to bind to the |
{}
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
llm = ChatOllama(model="llama3.1")
# Without bind.
chain = llm | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'
# With bind.
chain = llm.bind(stop=["three"]) | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'
with_config(
config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]
Bind config to a Runnable, returning a new Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
The config to bind to the |
None
|
kwargs
|
Any
|
Additional keyword arguments to pass to the |
{}
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
on_start
|
Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None
|
Called before the |
None
|
on_end
|
Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None
|
Called after the |
None
|
on_error
|
Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None
|
Called if the |
None
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
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(
*,
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
on_start
|
AsyncListener | None
|
Called asynchronously before the |
None
|
on_end
|
AsyncListener | None
|
Called asynchronously after the |
None
|
on_error
|
AsyncListener | None
|
Called asynchronously if the |
None
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
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(
*,
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input_type
|
type[Input] | None
|
The input type to bind to the |
None
|
output_type
|
type[Output] | None
|
The output type to bind to the |
None
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new Runnable with the types bound. |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
retry_if_exception_type
|
tuple[type[BaseException], ...]
|
A tuple of exception types to retry on. Defaults to (Exception,). |
(Exception,)
|
wait_exponential_jitter
|
bool
|
Whether to add jitter to the wait time between retries. Defaults to True. |
True
|
stop_after_attempt
|
int
|
The maximum number of attempts to make before giving up. Defaults to 3. |
3
|
exponential_jitter_params
|
ExponentialJitterParams | None
|
Parameters for
|
None
|
Returns:
| Type | 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
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fallbacks
|
Sequence[Runnable[Input, Output]]
|
A sequence of runnables to try if the original |
required |
exceptions_to_handle
|
tuple[type[BaseException], ...]
|
A tuple of exception types to handle.
Defaults to |
(Exception,)
|
exception_key
|
str | None
|
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 |
None
|
Returns:
| Type | Description |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
RunnableWithFallbacks[Input, Output]
|
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
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fallbacks
|
Sequence[Runnable[Input, Output]]
|
A sequence of runnables to try if the original |
required |
exceptions_to_handle
|
tuple[type[BaseException], ...]
|
A tuple of exception types to handle. |
(Exception,)
|
exception_key
|
str | None
|
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 |
None
|
Returns:
| Type | Description |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
RunnableWithFallbacks[Input, Output]
|
fallback in order, upon failures. |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
args_schema
|
type[BaseModel] | None
|
The schema for the tool. Defaults to None. |
None
|
name
|
str | None
|
The name of the tool. Defaults to None. |
None
|
description
|
str | None
|
The description of the tool. Defaults to None. |
None
|
arg_types
|
dict[str, type] | None
|
A dictionary of argument names to types. Defaults to None. |
None
|
Returns:
| Type | Description |
|---|---|
BaseTool
|
A |
Typed dict 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]})
String 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")
Added in version 0.2.14
classmethod
¶
classmethod
¶
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"].
Serialize the Runnable to JSON.
Returns:
| Type | Description |
|---|---|
SerializedConstructor | SerializedNotImplemented
|
A JSON-serializable representation of the |
Serialize a "not implemented" object.
Returns:
| Type | Description |
|---|---|
SerializedNotImplemented
|
SerializedNotImplemented. |
Configure particular Runnable fields at runtime.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
AnyConfigurableField
|
A dictionary of |
{}
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If a configuration key is not found in the |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
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(
which: ConfigurableField,
*,
default_key: str = "default",
prefix_keys: bool = False,
**kwargs: (
Runnable[Input, Output]
| Callable[[], Runnable[Input, Output]]
)
) -> RunnableSerializable[Input, Output]
Configure alternatives for Runnables that can be set at runtime.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
which
|
ConfigurableField
|
The |
required |
default_key
|
str
|
The default key to use if no alternative is selected.
Defaults to |
'default'
|
prefix_keys
|
bool
|
Whether to prefix the keys with the |
False
|
**kwargs
|
Runnable[Input, Output] | Callable[[], Runnable[Input, Output]]
|
A dictionary of keys to |
{}
|
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatAnthropic(
model_name="claude-3-7-sonnet-20250219"
).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
)
Get the number of tokens in the messages.
Useful for checking if an input fits in a model's context window.
Note
The base implementation of get_num_tokens_from_messages ignores tool
schemas.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
messages
|
list[BaseMessage]
|
The message inputs to tokenize. |
required |
tools
|
Sequence | None
|
If provided, sequence of dict, |
None
|
Returns:
| Type | Description |
|---|---|
int
|
The sum of the number of tokens across the messages. |
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:
- Take advantage of batched calls,
- Need more output from the model than just the top generated value,
- Are building chains that are agnostic to the underlying language model type (e.g., pure text completion models vs chat models).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
messages
|
list[list[BaseMessage]]
|
List of list of messages. |
required |
stop
|
list[str] | None
|
Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. |
None
|
callbacks
|
Callbacks
|
Callbacks to pass through. Used for executing additional functionality, such as logging or streaming, throughout generation. |
None
|
tags
|
list[str] | None
|
The tags to apply. |
None
|
metadata
|
dict[str, Any] | None
|
The metadata to apply. |
None
|
run_name
|
str | None
|
The name of the run. |
None
|
run_id
|
UUID | None
|
The ID of the run. |
None
|
**kwargs
|
Any
|
Arbitrary additional keyword arguments. These are usually passed to the model provider API call. |
{}
|
Returns:
| Type | Description |
|---|---|
LLMResult
|
An LLMResult, which contains a list of candidate Generations for each input |
LLMResult
|
prompt and additional model provider-specific output. |
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:
- Take advantage of batched calls,
- Need more output from the model than just the top generated value,
- Are building chains that are agnostic to the underlying language model type (e.g., pure text completion models vs chat models).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
messages
|
list[list[BaseMessage]]
|
List of list of messages. |
required |
stop
|
list[str] | None
|
Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. |
None
|
callbacks
|
Callbacks
|
Callbacks to pass through. Used for executing additional functionality, such as logging or streaming, throughout generation. |
None
|
tags
|
list[str] | None
|
The tags to apply. |
None
|
metadata
|
dict[str, Any] | None
|
The metadata to apply. |
None
|
run_name
|
str | None
|
The name of the run. |
None
|
run_id
|
UUID | None
|
The ID of the run. |
None
|
**kwargs
|
Any
|
Arbitrary additional keyword arguments. These are usually passed to the model provider API call. |
{}
|
Returns:
| Type | Description |
|---|---|
LLMResult
|
An LLMResult, which contains a list of candidate Generations for each input |
LLMResult
|
prompt and additional model provider-specific output. |
classmethod
¶
Build extra kwargs from additional params that were passed in.
Validates params and passes them to google-generativeai package.
invoke(
input: LanguageModelInput,
config: Optional[RunnableConfig] = None,
*,
code_execution: Optional[bool] = None,
stop: Optional[list[str]] = None,
**kwargs: Any
) -> BaseMessage
Enable code execution. Supported on: gemini-1.5-pro, gemini-1.5-flash, gemini-2.0-flash, and gemini-2.0-pro. When enabled, the model can execute code to solve problems.
bind_tools(
tools: Sequence[
dict[str, Any]
| type
| Callable[..., Any]
| BaseTool
| Tool
],
tool_config: Optional[
Union[Dict, _ToolConfigDict]
] = None,
*,
tool_choice: Optional[
Union[_ToolChoiceType, bool]
] = None,
**kwargs: Any
) -> Runnable[LanguageModelInput, BaseMessage]
Bind tool-like objects to this chat model.
Assumes model is compatible with google-generativeAI tool-calling API.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tools
|
Sequence[dict[str, Any] | type | Callable[..., Any] | BaseTool | Tool]
|
A list of tool definitions to bind to this chat model.
Can be a pydantic model, callable, or BaseTool. Pydantic models,
callables, and BaseTools will be automatically converted to their schema
dictionary representation. Tools with Union types in their arguments are
now supported and converted to |
required |
**kwargs
|
Any
|
Any additional parameters to pass to the
:class: |
{}
|
Bases: BaseModel, Embeddings
Google Generative AI Embeddings.
To use, you must have either:
1. The ``GOOGLE_API_KEY`` environment variable set with your API key, or
2. Pass your API key using the google_api_key kwarg to the
GoogleGenerativeAIEmbeddings constructor.
Example
.. code-block:: python
from langchain_google_genai import GoogleGenerativeAIEmbeddings
embeddings = GoogleGenerativeAIEmbeddings(model="gemini-embedding-001")
embeddings.embed_query("What's our Q1 revenue?")
Methods:
| Name | Description |
|---|---|
validate_environment |
Validates params and passes them to google-generativeai package. |
embed_documents |
Embed a list of strings using the |
embed_query |
Embed a text, using the |
aembed_documents |
Embed a list of strings using the |
aembed_query |
Embed a text, using the |
Validates params and passes them to google-generativeai package.
embed_documents(
texts: List[str],
*,
batch_size: int = _DEFAULT_BATCH_SIZE,
task_type: Optional[str] = None,
titles: Optional[List[str]] = None,
output_dimensionality: Optional[int] = None
) -> List[List[float]]
Embed a list of strings using the batch endpoint <https://ai.google.dev/api/embeddings#method:-models.batchembedcontents>__.
Google Generative AI currently sets a max batch size of 100 strings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
texts
|
List[str]
|
List[str] The list of strings to embed. |
required |
batch_size
|
int
|
[int] The batch size of embeddings to send to the model |
_DEFAULT_BATCH_SIZE
|
task_type
|
Optional[str]
|
|
None
|
titles
|
Optional[List[str]]
|
An optional list of titles for texts provided.
Only applicable when TaskType is |
None
|
output_dimensionality
|
Optional[int]
|
Optional |
None
|
Returns:
| Type | Description |
|---|---|
List[List[float]]
|
List of embeddings, one for each text. |
embed_query(
text: str,
*,
task_type: Optional[str] = None,
title: Optional[str] = None,
output_dimensionality: Optional[int] = None
) -> List[float]
Embed a text, using the non-batch endpoint <https://ai.google.dev/api/embeddings#method:-models.embedcontent>__.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
The text to embed. |
required |
task_type
|
Optional[str]
|
|
None
|
title
|
Optional[str]
|
An optional title for the text.
Only applicable when TaskType is |
None
|
output_dimensionality
|
Optional[int]
|
Optional |
None
|
Returns:
| Type | Description |
|---|---|
List[float]
|
Embedding for the text. |
async
¶
aembed_documents(
texts: List[str],
*,
batch_size: int = _DEFAULT_BATCH_SIZE,
task_type: Optional[str] = None,
titles: Optional[List[str]] = None,
output_dimensionality: Optional[int] = None
) -> List[List[float]]
Embed a list of strings using the batch endpoint <https://ai.google.dev/api/embeddings#method:-models.batchembedcontents>__.
Google Generative AI currently sets a max batch size of 100 strings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
texts
|
List[str]
|
List[str] The list of strings to embed. |
required |
batch_size
|
int
|
[int] The batch size of embeddings to send to the model |
_DEFAULT_BATCH_SIZE
|
task_type
|
Optional[str]
|
|
None
|
titles
|
Optional[List[str]]
|
An optional list of titles for texts provided.
Only applicable when TaskType is |
None
|
output_dimensionality
|
Optional[int]
|
Optional |
None
|
Returns:
| Type | Description |
|---|---|
List[List[float]]
|
List of embeddings, one for each text. |
async
¶
aembed_query(
text: str,
*,
task_type: Optional[str] = None,
title: Optional[str] = None,
output_dimensionality: Optional[int] = None
) -> List[float]
Embed a text, using the non-batch endpoint <https://ai.google.dev/api/embeddings#method:-models.embedcontent>__.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
The text to embed. |
required |
task_type
|
Optional[str]
|
|
None
|
title
|
Optional[str]
|
An optional title for the text.
Only applicable when TaskType is |
None
|
output_dimensionality
|
Optional[int]
|
Optional |
None
|
Returns:
| Type | Description |
|---|---|
List[float]
|
Embedding for the text. |
Bases: BaseModel
Output from GenAIAqa.invoke.
Attributes:
| Name | Type | Description |
|---|---|---|
answer |
str
|
The answer to the user's inquiry. |
attributed_passages |
List[str]
|
A list of passages that the LLM used to construct the answer. |
answerable_probability |
float
|
The probability of the question being answered from the provided passages. |
Bases: RunnableSerializable[AqaInput, AqaOutput]
Google's Attributed Question and Answering service.
Given a user's query and a list of passages, Google's server will return a response that is grounded to the provided list of passages. It will not base the response on parametric memory.
Attributes:
| Name | Type | Description |
|---|---|---|
answer_style |
int
|
keyword-only argument. See
|
Methods:
| Name | Description |
|---|---|
get_name |
Get the name of the |
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 |
get_output_schema |
Get a pydantic model that can be used to validate output to the |
get_output_jsonschema |
Get a JSON schema that represents the output of the |
config_schema |
The type of config this |
get_config_jsonschema |
Get a JSON schema that represents the config of the |
get_graph |
Return a graph representation of this |
get_prompts |
Return a list of prompts used by this |
__or__ |
Runnable "or" operator. |
__ror__ |
Runnable "reverse-or" operator. |
pipe |
Pipe runnables. |
pick |
Pick keys from the output dict of this |
assign |
Assigns new fields to the dict output of this |
ainvoke |
Transform a single input into an output. |
batch |
Default implementation runs invoke in parallel using a thread pool executor. |
batch_as_completed |
Run |
abatch |
Default implementation runs |
abatch_as_completed |
Run |
stream |
Default implementation of |
astream |
Default implementation of |
astream_log |
Stream all output from a |
astream_events |
Generate a stream of events. |
transform |
Transform inputs to outputs. |
atransform |
Transform inputs to outputs. |
bind |
Bind arguments to a |
with_config |
Bind config to a |
with_listeners |
Bind lifecycle listeners to a |
with_alisteners |
Bind async lifecycle listeners to a |
with_types |
Bind input and output types to a |
with_retry |
Create a new Runnable that retries the original Runnable on exceptions. |
map |
Return a new |
with_fallbacks |
Add fallbacks to a |
as_tool |
Create a |
is_lc_serializable |
Is this class serializable? |
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 |
to_json_not_implemented |
Serialize a "not implemented" object. |
configurable_fields |
Configure particular |
configurable_alternatives |
Configure alternatives for |
__init__ |
Construct a Google Generative AI AQA model. |
invoke |
Generates a grounded response using the provided passages. |
property
¶
InputType: type[Input]
Input type.
The type of input this Runnable accepts specified as a type annotation.
Raises:
| Type | Description |
|---|---|
TypeError
|
If the input type cannot be inferred. |
property
¶
OutputType: type[Output]
Output Type.
The type of output this Runnable produces specified as a type annotation.
Raises:
| Type | Description |
|---|---|
TypeError
|
If the output type cannot be inferred. |
property
¶
input_schema: type[BaseModel]
The type of input this Runnable accepts specified as a pydantic model.
property
¶
output_schema: type[BaseModel]
Output schema.
The type of output this Runnable produces specified as a pydantic model.
property
¶
config_specs: list[ConfigurableFieldSpec]
List configurable fields for this Runnable.
property
¶
A map of constructor argument names to secret ids.
For example,
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.
get_input_schema(
config: RunnableConfig | None = None,
) -> type[BaseModel]
Get a pydantic model that can be used to validate input to the Runnable.
Runnables 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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
type[BaseModel]
|
A pydantic model that can be used to validate input. |
Get a JSON schema that represents the input to the Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A JSON schema that represents the input to the |
Example
Added in version 0.3.0
get_output_schema(
config: RunnableConfig | None = None,
) -> type[BaseModel]
Get a pydantic model that can be used to validate output to the Runnable.
Runnables 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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
type[BaseModel]
|
A pydantic model that can be used to validate output. |
Get a JSON schema that represents the output of the Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A JSON schema that represents the output of the |
Example
Added in version 0.3.0
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include
|
Sequence[str] | None
|
A list of fields to include in the config schema. |
None
|
Returns:
| Type | Description |
|---|---|
type[BaseModel]
|
A pydantic model that can be used to validate config. |
Get a JSON schema that represents the config of the Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include
|
Sequence[str] | None
|
A list of fields to include in the config schema. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A JSON schema that represents the config of the |
Added in version 0.3.0
Return a graph representation of this Runnable.
get_prompts(
config: RunnableConfig | None = None,
) -> list[BasePromptTemplate]
Return a list of prompts used by this Runnable.
__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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
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]
|
Another |
required |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
__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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
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]
|
Another |
required |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Other, Output]
|
A new |
pipe(
*others: Runnable[Any, Other] | Callable[[Any], Other],
name: str | None = None
) -> RunnableSerializable[Input, Other]
Pipe runnables.
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]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*others
|
Runnable[Any, Other] | Callable[[Any], Other]
|
Other |
()
|
name
|
str | None
|
An optional name for the resulting |
None
|
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
Pick keys from the output dict of this Runnable.
Pick single key:
```python
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 list of keys:
```python
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]"}
```
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
str | list[str]
|
A key or list of keys to pick from the output dict. |
required |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Any, Any]
|
a new |
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_community.llms.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}"
)
llm = FakeStreamingListLLM(responses=["foo-lish"])
chain: Runnable = prompt | llm | {"str": StrOutputParser()}
chain_with_assign = chain.assign(hello=itemgetter("str") | llm)
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'}}}
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any] | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]]
|
A mapping of keys to |
{}
|
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Any, Any]
|
A new |
async
¶
ainvoke(
input: Input,
config: RunnableConfig | None = None,
**kwargs: Any
) -> Output
Transform a single input into an output.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Input
|
The input to the |
required |
config
|
RunnableConfig | None
|
A config to use when invoking the |
None
|
Returns:
| Type | Description |
|---|---|
Output
|
The output of the |
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 should override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
list[Input]
|
A list of inputs to the |
required |
config
|
RunnableConfig | list[RunnableConfig] | None
|
A config to use when invoking the |
None
|
return_exceptions
|
bool
|
Whether to return exceptions instead of raising them. Defaults to False. |
False
|
**kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Returns:
| Type | Description |
|---|---|
list[Output]
|
A list of outputs from the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
Sequence[Input]
|
A list of inputs to the |
required |
config
|
RunnableConfig | Sequence[RunnableConfig] | None
|
A config to use when invoking the |
None
|
return_exceptions
|
bool
|
Whether to return exceptions instead of raising them. Defaults to False. |
False
|
**kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
tuple[int, Output | Exception]
|
Tuples of the index of the input and the output from the |
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 should override this method if they can batch more efficiently;
e.g., if the underlying Runnable uses an API which supports a batch mode.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
list[Input]
|
A list of inputs to the |
required |
config
|
RunnableConfig | list[RunnableConfig] | None
|
A config to use when invoking the |
None
|
return_exceptions
|
bool
|
Whether to return exceptions instead of raising them. Defaults to False. |
False
|
**kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Returns:
| Type | Description |
|---|---|
list[Output]
|
A list of outputs from the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
Sequence[Input]
|
A list of inputs to the |
required |
config
|
RunnableConfig | Sequence[RunnableConfig] | None
|
A config to use when invoking the |
None
|
return_exceptions
|
bool
|
Whether to return exceptions instead of raising them. Defaults to False. |
False
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[tuple[int, Output | Exception]]
|
A tuple of the index of the input and the output from the |
stream(
input: Input,
config: RunnableConfig | None = None,
**kwargs: Any | None
) -> Iterator[Output]
Default implementation of stream, which calls invoke.
Subclasses should override this method if they support streaming output.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Input
|
The input to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
Output
|
The output of the |
async
¶
astream(
input: Input,
config: RunnableConfig | None = None,
**kwargs: Any | None
) -> AsyncIterator[Output]
Default implementation of astream, which calls ainvoke.
Subclasses should override this method if they support streaming output.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Input
|
The input to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[Output]
|
The output of the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Any
|
The input to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
diff
|
bool
|
Whether to yield diffs between each step or the current state. |
True
|
with_streamed_output_list
|
bool
|
Whether to yield the |
True
|
include_names
|
Sequence[str] | None
|
Only include logs with these names. |
None
|
include_types
|
Sequence[str] | None
|
Only include logs with these types. |
None
|
include_tags
|
Sequence[str] | None
|
Only include logs with these tags. |
None
|
exclude_names
|
Sequence[str] | None
|
Exclude logs with these names. |
None
|
exclude_types
|
Sequence[str] | None
|
Exclude logs with these types. |
None
|
exclude_tags
|
Sequence[str] | None
|
Exclude logs with these tags. |
None
|
kwargs
|
Any
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
|
A |
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 StreamEvents that provide real-time information
about the progress of the Runnable, including StreamEvents from intermediate
results.
A StreamEvent is a dictionary with the following schema:
event: str - Event names are of the format:on_[runnable_type]_(start|stream|end).name: str - The name of theRunnablethat generated the event.run_id: str - randomly generated ID associated with the given execution of theRunnablethat emitted the event. A childRunnablethat gets invoked as part of the execution of a parentRunnableis assigned its own unique ID.parent_ids: list[str] - The IDs of the parent runnables that generated the event. The rootRunnablewill 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: Optional[list[str]] - The tags of theRunnablethat generated the event.metadata: Optional[dict[str, Any]] - The metadata of theRunnablethat generated the event.data: dict[str, Any]
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:
prompt:
template = ChatPromptTemplate.from_messages(
[
("system", "You are Cat Agent 007"),
("human", "{question}"),
]
).with_config({"run_name": "my_template", "tags": ["my_template"]})
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": [],
},
]
Example: 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)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Any
|
The input to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
version
|
Literal['v1', 'v2']
|
The version of the schema to use either |
'v2'
|
include_names
|
Sequence[str] | None
|
Only include events from |
None
|
include_types
|
Sequence[str] | None
|
Only include events from |
None
|
include_tags
|
Sequence[str] | None
|
Only include events from |
None
|
exclude_names
|
Sequence[str] | None
|
Exclude events from |
None
|
exclude_types
|
Sequence[str] | None
|
Exclude events from |
None
|
exclude_tags
|
Sequence[str] | None
|
Exclude events from |
None
|
kwargs
|
Any
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[StreamEvent]
|
An async stream of |
Raises:
| Type | Description |
|---|---|
NotImplementedError
|
If the version is not |
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 should override this method if they can start producing output while input is still being generated.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Iterator[Input]
|
An iterator of inputs to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
Output
|
The output of the |
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 should override this method if they can start producing output while input is still being generated.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
AsyncIterator[Input]
|
An async iterator of inputs to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[Output]
|
The output of the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
kwargs
|
Any
|
The arguments to bind to the |
{}
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
llm = ChatOllama(model="llama3.1")
# Without bind.
chain = llm | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'
# With bind.
chain = llm.bind(stop=["three"]) | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'
with_config(
config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]
Bind config to a Runnable, returning a new Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
The config to bind to the |
None
|
kwargs
|
Any
|
Additional keyword arguments to pass to the |
{}
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
on_start
|
Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None
|
Called before the |
None
|
on_end
|
Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None
|
Called after the |
None
|
on_error
|
Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None
|
Called if the |
None
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
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(
*,
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
on_start
|
AsyncListener | None
|
Called asynchronously before the |
None
|
on_end
|
AsyncListener | None
|
Called asynchronously after the |
None
|
on_error
|
AsyncListener | None
|
Called asynchronously if the |
None
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
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(
*,
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input_type
|
type[Input] | None
|
The input type to bind to the |
None
|
output_type
|
type[Output] | None
|
The output type to bind to the |
None
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new Runnable with the types bound. |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
retry_if_exception_type
|
tuple[type[BaseException], ...]
|
A tuple of exception types to retry on. Defaults to (Exception,). |
(Exception,)
|
wait_exponential_jitter
|
bool
|
Whether to add jitter to the wait time between retries. Defaults to True. |
True
|
stop_after_attempt
|
int
|
The maximum number of attempts to make before giving up. Defaults to 3. |
3
|
exponential_jitter_params
|
ExponentialJitterParams | None
|
Parameters for
|
None
|
Returns:
| Type | 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
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fallbacks
|
Sequence[Runnable[Input, Output]]
|
A sequence of runnables to try if the original |
required |
exceptions_to_handle
|
tuple[type[BaseException], ...]
|
A tuple of exception types to handle.
Defaults to |
(Exception,)
|
exception_key
|
str | None
|
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 |
None
|
Returns:
| Type | Description |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
RunnableWithFallbacks[Input, Output]
|
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
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fallbacks
|
Sequence[Runnable[Input, Output]]
|
A sequence of runnables to try if the original |
required |
exceptions_to_handle
|
tuple[type[BaseException], ...]
|
A tuple of exception types to handle. |
(Exception,)
|
exception_key
|
str | None
|
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 |
None
|
Returns:
| Type | Description |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
RunnableWithFallbacks[Input, Output]
|
fallback in order, upon failures. |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
args_schema
|
type[BaseModel] | None
|
The schema for the tool. Defaults to None. |
None
|
name
|
str | None
|
The name of the tool. Defaults to None. |
None
|
description
|
str | None
|
The description of the tool. Defaults to None. |
None
|
arg_types
|
dict[str, type] | None
|
A dictionary of argument names to types. Defaults to None. |
None
|
Returns:
| Type | Description |
|---|---|
BaseTool
|
A |
Typed dict 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]})
String 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")
Added in version 0.2.14
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:
| Type | Description |
|---|---|
bool
|
Whether the class is serializable. Default is False. |
classmethod
¶
classmethod
¶
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"].
Serialize the Runnable to JSON.
Returns:
| Type | Description |
|---|---|
SerializedConstructor | SerializedNotImplemented
|
A JSON-serializable representation of the |
Serialize a "not implemented" object.
Returns:
| Type | Description |
|---|---|
SerializedNotImplemented
|
SerializedNotImplemented. |
Configure particular Runnable fields at runtime.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
AnyConfigurableField
|
A dictionary of |
{}
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If a configuration key is not found in the |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
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(
which: ConfigurableField,
*,
default_key: str = "default",
prefix_keys: bool = False,
**kwargs: (
Runnable[Input, Output]
| Callable[[], Runnable[Input, Output]]
)
) -> RunnableSerializable[Input, Output]
Configure alternatives for Runnables that can be set at runtime.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
which
|
ConfigurableField
|
The |
required |
default_key
|
str
|
The default key to use if no alternative is selected.
Defaults to |
'default'
|
prefix_keys
|
bool
|
Whether to prefix the keys with the |
False
|
**kwargs
|
Runnable[Input, Output] | Callable[[], Runnable[Input, Output]]
|
A dictionary of keys to |
{}
|
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatAnthropic(
model_name="claude-3-7-sonnet-20250219"
).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
)
__init__(**kwargs: Any) -> None
Construct a Google Generative AI AQA model.
All arguments are optional.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
answer_style
|
See
|
required | |
safety_settings
|
See |
required | |
temperature
|
0.0 to 1.0. |
required |
Bases: VectorStore
Google GenerativeAI Vector Store.
Currently, it computes the embedding vectors on the server side.
Example: Add texts to an existing corpus.
store = GoogleVectorStore(corpus_id="123")
store.add_documents(documents, document_id="456")
Example: Create a new corpus.
store = GoogleVectorStore.create_corpus(
corpus_id="123", display_name="My Google corpus")
Example: Query the corpus for relevant passages.
store.as_retriever() .get_relevant_documents("Who caught the gingerbread man?")
Example: Ask the corpus for grounded responses!
aqa = store.as_aqa()
response = aqa.invoke("Who caught the gingerbread man?")
print(response.answer)
print(response.attributed_passages)
print(response.answerability_probability)
You can also operate at Google's Document level.
Example: Add texts to an existing Google Vector Store Document.
doc_store = GoogleVectorStore(corpus_id="123", document_id="456")
doc_store.add_documents(documents)
Example: Create a new Google Vector Store Document.
doc_store = GoogleVectorStore.create_document(
corpus_id="123", document_id="456", display_name="My Google document")
Example: Query the Google document.
doc_store.as_retriever() .get_relevant_documents("Who caught the gingerbread man?")
For more details, see the class's methods.
Methods:
| Name | Description |
|---|---|
get_by_ids |
Get documents by their IDs. |
aget_by_ids |
Async get documents by their IDs. |
aadd_texts |
Async run more texts through the embeddings and add to the vectorstore. |
add_documents |
Add or update documents in the vectorstore. |
aadd_documents |
Async run more documents through the embeddings and add to the vectorstore. |
search |
Return docs most similar to query using a specified search type. |
asearch |
Async return docs most similar to query using a specified search type. |
asimilarity_search_with_score |
Async run similarity search with distance. |
similarity_search_with_relevance_scores |
Return docs and relevance scores in the range [0, 1]. |
asimilarity_search_with_relevance_scores |
Async return docs and relevance scores in the range [0, 1]. |
asimilarity_search |
Async return docs most similar to query. |
similarity_search_by_vector |
Return docs most similar to embedding vector. |
asimilarity_search_by_vector |
Async return docs most similar to embedding vector. |
max_marginal_relevance_search |
Return docs selected using the maximal marginal relevance. |
amax_marginal_relevance_search |
Async return docs selected using the maximal marginal relevance. |
max_marginal_relevance_search_by_vector |
Return docs selected using the maximal marginal relevance. |
amax_marginal_relevance_search_by_vector |
Async return docs selected using the maximal marginal relevance. |
from_documents |
Return VectorStore initialized from documents and embeddings. |
afrom_documents |
Async return VectorStore initialized from documents and embeddings. |
afrom_texts |
Async return VectorStore initialized from texts and embeddings. |
as_retriever |
Return VectorStoreRetriever initialized from this VectorStore. |
__init__ |
Returns an existing Google Semantic Retriever corpus or document. |
create_corpus |
Create a Google Semantic Retriever corpus. |
create_document |
Create a Google Semantic Retriever document. |
from_texts |
Returns a vector store of an existing document with the specified text. |
add_texts |
Add texts to the vector store. |
similarity_search |
Search the vector store for relevant texts. |
similarity_search_with_score |
Run similarity search with distance. |
delete |
Delete chunks. |
adelete |
Delete chunks asynchronously. |
as_aqa |
Construct a Google Generative AI AQA engine. |
Attributes:
| Name | Type | Description |
|---|---|---|
embeddings |
Embeddings | None
|
Access the query embedding object if available. |
name |
str
|
Returns the name of the Google entity. |
corpus_id |
str
|
Returns the corpus ID managed by this vector store. |
document_id |
Optional[str]
|
Returns the document ID managed by this vector store. |
property
¶
name: str
Returns the name of the Google entity.
You shouldn't need to care about this unless you want to access your corpus or document via Google Generative AI API.
property
¶
Returns the document ID managed by this vector store.
Get documents by their IDs.
The returned documents are expected to have the ID field set to the ID of the document in the vector store.
Fewer documents may be returned than requested if some IDs are not found or if there are duplicated IDs.
Users should not assume that the order of the returned documents matches the order of the input IDs. Instead, users should rely on the ID field of the returned documents.
This method should NOT raise exceptions if no documents are found for some IDs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ids
|
Sequence[str]
|
List of ids to retrieve. |
required |
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents. |
Added in version 0.2.11
async
¶
Async get documents by their IDs.
The returned documents are expected to have the ID field set to the ID of the document in the vector store.
Fewer documents may be returned than requested if some IDs are not found or if there are duplicated IDs.
Users should not assume that the order of the returned documents matches the order of the input IDs. Instead, users should rely on the ID field of the returned documents.
This method should NOT raise exceptions if no documents are found for some IDs.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ids
|
Sequence[str]
|
List of ids to retrieve. |
required |
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents. |
Added in version 0.2.11
async
¶
aadd_texts(
texts: Iterable[str],
metadatas: list[dict] | None = None,
*,
ids: list[str] | None = None,
**kwargs: Any
) -> list[str]
Async run more texts through the embeddings and add to the vectorstore.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
texts
|
Iterable[str]
|
Iterable of strings to add to the vectorstore. |
required |
metadatas
|
list[dict] | None
|
Optional list of metadatas associated with the texts. Default is None. |
None
|
ids
|
list[str] | None
|
Optional list |
None
|
**kwargs
|
Any
|
vectorstore specific parameters. |
{}
|
Returns:
| Type | Description |
|---|---|
list[str]
|
List of ids from adding the texts into the vectorstore. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the number of metadatas does not match the number of texts. |
ValueError
|
If the number of ids does not match the number of texts. |
Add or update documents in the vectorstore.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
documents
|
list[Document]
|
Documents to add to the vectorstore. |
required |
kwargs
|
Any
|
Additional keyword arguments. if kwargs contains ids and documents contain ids, the ids in the kwargs will receive precedence. |
{}
|
Returns:
| Type | Description |
|---|---|
list[str]
|
List of IDs of the added texts. |
async
¶
Async run more documents through the embeddings and add to the vectorstore.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
documents
|
list[Document]
|
Documents to add to the vectorstore. |
required |
kwargs
|
Any
|
Additional keyword arguments. |
{}
|
Returns:
| Type | Description |
|---|---|
list[str]
|
List of IDs of the added texts. |
Return docs most similar to query using a specified search type.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
Input text |
required |
search_type
|
str
|
Type of search to perform. Can be "similarity", "mmr", or "similarity_score_threshold". |
required |
**kwargs
|
Any
|
Arguments to pass to the search method. |
{}
|
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents most similar to the query. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If search_type is not one of "similarity", "mmr", or "similarity_score_threshold". |
async
¶
Async return docs most similar to query using a specified search type.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
Input text. |
required |
search_type
|
str
|
Type of search to perform. Can be "similarity", "mmr", or "similarity_score_threshold". |
required |
**kwargs
|
Any
|
Arguments to pass to the search method. |
{}
|
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents most similar to the query. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If search_type is not one of "similarity", "mmr", or "similarity_score_threshold". |
async
¶
similarity_search_with_relevance_scores(
query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]
Return docs and relevance scores in the range [0, 1].
0 is dissimilar, 1 is most similar.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
Input text. |
required |
k
|
int
|
Number of Documents to return. Defaults to 4. |
4
|
**kwargs
|
Any
|
kwargs to be passed to similarity search. Should include: score_threshold: Optional, a floating point value between 0 to 1 to filter the resulting set of retrieved docs. |
{}
|
Returns:
| Type | Description |
|---|---|
list[tuple[Document, float]]
|
List of Tuples of (doc, similarity_score). |
async
¶
asimilarity_search_with_relevance_scores(
query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]
Async return docs and relevance scores in the range [0, 1].
0 is dissimilar, 1 is most similar.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
Input text. |
required |
k
|
int
|
Number of Documents to return. Defaults to 4. |
4
|
**kwargs
|
Any
|
kwargs to be passed to similarity search. Should include: score_threshold: Optional, a floating point value between 0 to 1 to filter the resulting set of retrieved docs |
{}
|
Returns:
| Type | Description |
|---|---|
list[tuple[Document, float]]
|
List of Tuples of (doc, similarity_score) |
async
¶
Async return docs most similar to query.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
Input text. |
required |
k
|
int
|
Number of Documents to return. Defaults to 4. |
4
|
**kwargs
|
Any
|
Arguments to pass to the search method. |
{}
|
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents most similar to the query. |
Return docs most similar to embedding vector.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
embedding
|
list[float]
|
Embedding to look up documents similar to. |
required |
k
|
int
|
Number of Documents to return. Defaults to 4. |
4
|
**kwargs
|
Any
|
Arguments to pass to the search method. |
{}
|
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents most similar to the query vector. |
async
¶
Async return docs most similar to embedding vector.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
embedding
|
list[float]
|
Embedding to look up documents similar to. |
required |
k
|
int
|
Number of Documents to return. Defaults to 4. |
4
|
**kwargs
|
Any
|
Arguments to pass to the search method. |
{}
|
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents most similar to the query vector. |
max_marginal_relevance_search(
query: str,
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
**kwargs: Any
) -> list[Document]
Return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
Text to look up documents similar to. |
required |
k
|
int
|
Number of Documents to return. Defaults to 4. |
4
|
fetch_k
|
int
|
Number of Documents to fetch to pass to MMR algorithm. Default is 20. |
20
|
lambda_mult
|
float
|
Number between 0 and 1 that determines the degree of diversity among the results with 0 corresponding to maximum diversity and 1 to minimum diversity. Defaults to 0.5. |
0.5
|
**kwargs
|
Any
|
Arguments to pass to the search method. |
{}
|
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents selected by maximal marginal relevance. |
async
¶
amax_marginal_relevance_search(
query: str,
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
**kwargs: Any
) -> list[Document]
Async return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
query
|
str
|
Text to look up documents similar to. |
required |
k
|
int
|
Number of Documents to return. Defaults to 4. |
4
|
fetch_k
|
int
|
Number of Documents to fetch to pass to MMR algorithm. Default is 20. |
20
|
lambda_mult
|
float
|
Number between 0 and 1 that determines the degree of diversity among the results with 0 corresponding to maximum diversity and 1 to minimum diversity. Defaults to 0.5. |
0.5
|
**kwargs
|
Any
|
Arguments to pass to the search method. |
{}
|
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents selected by maximal marginal relevance. |
max_marginal_relevance_search_by_vector(
embedding: list[float],
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
**kwargs: Any
) -> list[Document]
Return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
embedding
|
list[float]
|
Embedding to look up documents similar to. |
required |
k
|
int
|
Number of Documents to return. Defaults to 4. |
4
|
fetch_k
|
int
|
Number of Documents to fetch to pass to MMR algorithm. Default is 20. |
20
|
lambda_mult
|
float
|
Number between 0 and 1 that determines the degree of diversity among the results with 0 corresponding to maximum diversity and 1 to minimum diversity. Defaults to 0.5. |
0.5
|
**kwargs
|
Any
|
Arguments to pass to the search method. |
{}
|
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents selected by maximal marginal relevance. |
async
¶
amax_marginal_relevance_search_by_vector(
embedding: list[float],
k: int = 4,
fetch_k: int = 20,
lambda_mult: float = 0.5,
**kwargs: Any
) -> list[Document]
Async return docs selected using the maximal marginal relevance.
Maximal marginal relevance optimizes for similarity to query AND diversity among selected documents.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
embedding
|
list[float]
|
Embedding to look up documents similar to. |
required |
k
|
int
|
Number of Documents to return. Defaults to 4. |
4
|
fetch_k
|
int
|
Number of Documents to fetch to pass to MMR algorithm. Default is 20. |
20
|
lambda_mult
|
float
|
Number between 0 and 1 that determines the degree of diversity among the results with 0 corresponding to maximum diversity and 1 to minimum diversity. Defaults to 0.5. |
0.5
|
**kwargs
|
Any
|
Arguments to pass to the search method. |
{}
|
Returns:
| Type | Description |
|---|---|
list[Document]
|
List of Documents selected by maximal marginal relevance. |
classmethod
¶
from_documents(
documents: list[Document],
embedding: Embeddings,
**kwargs: Any
) -> Self
Return VectorStore initialized from documents and embeddings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
documents
|
list[Document]
|
List of Documents to add to the vectorstore. |
required |
embedding
|
Embeddings
|
Embedding function to use. |
required |
kwargs
|
Any
|
Additional keyword arguments. |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
VectorStore |
Self
|
VectorStore initialized from documents and embeddings. |
async
classmethod
¶
afrom_documents(
documents: list[Document],
embedding: Embeddings,
**kwargs: Any
) -> Self
Async return VectorStore initialized from documents and embeddings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
documents
|
list[Document]
|
List of Documents to add to the vectorstore. |
required |
embedding
|
Embeddings
|
Embedding function to use. |
required |
kwargs
|
Any
|
Additional keyword arguments. |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
VectorStore |
Self
|
VectorStore initialized from documents and embeddings. |
async
classmethod
¶
afrom_texts(
texts: list[str],
embedding: Embeddings,
metadatas: list[dict] | None = None,
*,
ids: list[str] | None = None,
**kwargs: Any
) -> Self
Async return VectorStore initialized from texts and embeddings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
texts
|
list[str]
|
Texts to add to the vectorstore. |
required |
embedding
|
Embeddings
|
Embedding function to use. |
required |
metadatas
|
list[dict] | None
|
Optional list of metadatas associated with the texts. Default is None. |
None
|
ids
|
list[str] | None
|
Optional list of IDs associated with the texts. |
None
|
kwargs
|
Any
|
Additional keyword arguments. |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
VectorStore |
Self
|
VectorStore initialized from texts and embeddings. |
as_retriever(**kwargs: Any) -> VectorStoreRetriever
Return VectorStoreRetriever initialized from this VectorStore.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
Any
|
Keyword arguments to pass to the search function. Can include: search_type (Optional[str]): Defines the type of search that the Retriever should perform. Can be "similarity" (default), "mmr", or "similarity_score_threshold". search_kwargs (Optional[Dict]): Keyword arguments to pass to the search function. Can include things like: k: Amount of documents to return (Default: 4) score_threshold: Minimum relevance threshold for similarity_score_threshold fetch_k: Amount of documents to pass to MMR algorithm (Default: 20) lambda_mult: Diversity of results returned by MMR; 1 for minimum diversity and 0 for maximum. (Default: 0.5) filter: Filter by document metadata |
{}
|
Returns:
| Name | Type | Description |
|---|---|---|
VectorStoreRetriever |
VectorStoreRetriever
|
Retriever class for VectorStore. |
Examples:
.. code-block:: python
# Retrieve more documents with higher diversity
# Useful if your dataset has many similar documents
docsearch.as_retriever(
search_type="mmr", search_kwargs={"k": 6, "lambda_mult": 0.25}
)
# Fetch more documents for the MMR algorithm to consider
# But only return the top 5
docsearch.as_retriever(
search_type="mmr", search_kwargs={"k": 5, "fetch_k": 50}
)
# Only retrieve documents that have a relevance score
# Above a certain threshold
docsearch.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={"score_threshold": 0.8},
)
# Only get the single most similar document from the dataset
docsearch.as_retriever(search_kwargs={"k": 1})
# Use a filter to only retrieve documents from a specific paper
docsearch.as_retriever(
search_kwargs={"filter": {"paper_title": "GPT-4 Technical Report"}}
)
Returns an existing Google Semantic Retriever corpus or document.
If just the corpus ID is provided, the vector store operates over all documents within that corpus.
If the document ID is provided, the vector store operates over just that document.
classmethod
¶
create_corpus(
corpus_id: Optional[str] = None,
display_name: Optional[str] = None,
) -> GoogleVectorStore
Create a Google Semantic Retriever corpus.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
corpus_id
|
Optional[str]
|
The ID to use to create the new corpus. If not provided, Google server will provide one. |
None
|
display_name
|
Optional[str]
|
The title of the new corpus. If not provided, Google server will provide one. |
None
|
Returns:
| Type | Description |
|---|---|
GoogleVectorStore
|
An instance of vector store that points to the newly created corpus. |
classmethod
¶
create_document(
corpus_id: str,
document_id: Optional[str] = None,
display_name: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> GoogleVectorStore
Create a Google Semantic Retriever document.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
corpus_id
|
str
|
ID of an existing corpus. |
required |
document_id
|
Optional[str]
|
The ID to use to create the new Google Semantic Retriever document. If not provided, Google server will provide one. |
None
|
display_name
|
Optional[str]
|
The title of the new document. If not provided, Google server will provide one. |
None
|
Returns:
| Type | Description |
|---|---|
GoogleVectorStore
|
An instance of vector store that points to the newly created |
GoogleVectorStore
|
document. |
classmethod
¶
from_texts(
texts: List[str],
embedding: Optional[Embeddings] = None,
metadatas: Optional[List[dict[str, Any]]] = None,
*,
corpus_id: Optional[str] = None,
document_id: Optional[str] = None,
**kwargs: Any
) -> GoogleVectorStore
Returns a vector store of an existing document with the specified text.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
corpus_id
|
Optional[str]
|
REQUIRED. Must be an existing corpus. |
None
|
document_id
|
Optional[str]
|
REQUIRED. Must be an existing document. |
None
|
texts
|
List[str]
|
Texts to be loaded into the vector store. |
required |
Returns:
| Type | Description |
|---|---|
GoogleVectorStore
|
A vector store pointing to the specified Google Semantic Retriever |
GoogleVectorStore
|
Document. |
similarity_search(
query: str,
k: int = 4,
filter: Optional[Dict[str, Any]] = None,
**kwargs: Any
) -> List[Document]
Search the vector store for relevant texts.
similarity_search_with_score(
query: str,
k: int = 4,
filter: Optional[Dict[str, Any]] = None,
**kwargs: Any
) -> List[Tuple[Document, float]]
Run similarity search with distance.
async
¶
Construct a Google Generative AI AQA engine.
All arguments are optional.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
answer_style
|
See
|
required | |
safety_settings
|
See |
required | |
temperature
|
0.0 to 1.0. |
required |
Bases: _BaseGoogleGenerativeAI, BaseLLM
Google GenerativeAI models.
Example
.. code-block:: python
from langchain_google_genai import GoogleGenerativeAI
llm = GoogleGenerativeAI(model="gemini-2.5-pro")
Methods:
| Name | Description |
|---|---|
get_name |
Get the name of the |
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 |
get_output_schema |
Get a pydantic model that can be used to validate output to the |
get_output_jsonschema |
Get a JSON schema that represents the output of the |
config_schema |
The type of config this |
get_config_jsonschema |
Get a JSON schema that represents the config of the |
get_graph |
Return a graph representation of this |
get_prompts |
Return a list of prompts used by this |
__or__ |
Runnable "or" operator. |
__ror__ |
Runnable "reverse-or" operator. |
pipe |
Pipe runnables. |
pick |
Pick keys from the output dict of this |
assign |
Assigns new fields to the dict output of this |
batch_as_completed |
Run |
abatch_as_completed |
Run |
astream_log |
Stream all output from a |
astream_events |
Generate a stream of events. |
transform |
Transform inputs to outputs. |
atransform |
Transform inputs to outputs. |
bind |
Bind arguments to a |
with_config |
Bind config to a |
with_listeners |
Bind lifecycle listeners to a |
with_alisteners |
Bind async lifecycle listeners to a |
with_types |
Bind input and output types to a |
with_retry |
Create a new Runnable that retries the original Runnable on exceptions. |
map |
Return a new |
with_fallbacks |
Add fallbacks to a |
as_tool |
Create a |
is_lc_serializable |
Is this class serializable? |
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 |
to_json_not_implemented |
Serialize a "not implemented" object. |
configurable_fields |
Configure particular |
configurable_alternatives |
Configure alternatives for |
set_verbose |
If verbose is None, set it. |
with_structured_output |
Not implemented on this class. |
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 a model and return generations. |
agenerate |
Asynchronously pass a sequence of prompts to a model and return generations. |
__str__ |
Return a string representation of the object for printing. |
dict |
Return a dictionary of the LLM. |
save |
Save the LLM. |
__init__ |
Needed for arg validation. |
validate_environment |
Validates params and passes them to google-generativeai package. |
get_num_tokens |
Get the number of tokens present in the text. |
Attributes:
| Name | Type | Description |
|---|---|---|
InputType |
TypeAlias
|
Get the input type for this runnable. |
OutputType |
type[str]
|
Get the input type for this runnable. |
input_schema |
type[BaseModel]
|
The type of input this |
output_schema |
type[BaseModel]
|
Output schema. |
config_specs |
list[ConfigurableFieldSpec]
|
List configurable fields for this |
lc_attributes |
dict
|
List of attribute names that should be included in the serialized kwargs. |
cache |
BaseCache | bool | None
|
Whether to cache the response. |
verbose |
bool
|
Whether to print out response text. |
callbacks |
Callbacks
|
Callbacks to add to the run trace. |
tags |
list[str] | None
|
Tags to add to the run trace. |
metadata |
dict[str, Any] | None
|
Metadata to add to the run trace. |
custom_get_token_ids |
Callable[[str], list[int]] | None
|
Optional encoder to use for counting tokens. |
model |
str
|
Model name to use. |
google_api_key |
Optional[SecretStr]
|
Google AI API key. |
credentials |
Any
|
The default custom credentials (google.auth.credentials.Credentials) to use |
temperature |
float
|
Run inference with this temperature. Must be within |
top_p |
Optional[float]
|
Decode using nucleus sampling: consider the smallest set of tokens whose |
top_k |
Optional[int]
|
Decode using top-k sampling: consider the set of |
max_output_tokens |
Optional[int]
|
Maximum number of tokens to include in a candidate. Must be greater than zero. |
n |
int
|
Number of chat completions to generate for each prompt. Note that the API may |
max_retries |
int
|
The maximum number of retries to make when generating. If unset, will default |
timeout |
Optional[float]
|
The maximum number of seconds to wait for a response. |
safety_settings |
Optional[Dict[HarmCategory, HarmBlockThreshold]]
|
The default safety settings to use for all generations. |
property
¶
input_schema: type[BaseModel]
The type of input this Runnable accepts specified as a pydantic model.
property
¶
output_schema: type[BaseModel]
Output schema.
The type of output this Runnable produces specified as a pydantic model.
property
¶
config_specs: list[ConfigurableFieldSpec]
List configurable fields for this Runnable.
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.
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.
class-attribute
instance-attribute
¶
verbose: bool = Field(
default_factory=_get_verbosity, exclude=True, repr=False
)
Whether to print out response text.
class-attribute
instance-attribute
¶
Callbacks to add to the run trace.
class-attribute
instance-attribute
¶
Tags to add to the run trace.
class-attribute
instance-attribute
¶
Metadata to add to the run trace.
class-attribute
instance-attribute
¶
Optional encoder to use for counting tokens.
class-attribute
instance-attribute
¶
model: str = Field(
...,
description="The name of the model to use.\nExamples:\n - gemini-2.5-flash\n - models/text-bison-001",
)
Model name to use.
class-attribute
instance-attribute
¶
google_api_key: Optional[SecretStr] = Field(
alias="api_key",
default_factory=secret_from_env(
"GOOGLE_API_KEY", default=None
),
)
Google AI API key.
If not specified will be read from env var GOOGLE_API_KEY.
class-attribute
instance-attribute
¶
credentials: Any = None
The default custom credentials (google.auth.credentials.Credentials) to use
class-attribute
instance-attribute
¶
temperature: float = 0.7
Run inference with this temperature. Must be within [0.0, 2.0]. If unset,
will default to 0.7.
class-attribute
instance-attribute
¶
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].
class-attribute
instance-attribute
¶
Decode using top-k sampling: consider the set of top_k most probable tokens.
Must be positive.
class-attribute
instance-attribute
¶
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 https://ai.google.dev/gemini-api/docs/models for model-specific limits.
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.
class-attribute
instance-attribute
¶
max_retries: int = Field(default=6, alias='retries')
The maximum number of retries to make when generating. If unset, will default
to 6.
class-attribute
instance-attribute
¶
The maximum number of seconds to wait for a response.
class-attribute
instance-attribute
¶
The default safety settings to use for all generations.
For example:
.. code-block:: python from google.generativeai.types.safety_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,
}
get_input_schema(
config: RunnableConfig | None = None,
) -> type[BaseModel]
Get a pydantic model that can be used to validate input to the Runnable.
Runnables 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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
type[BaseModel]
|
A pydantic model that can be used to validate input. |
Get a JSON schema that represents the input to the Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A JSON schema that represents the input to the |
Example
Added in version 0.3.0
get_output_schema(
config: RunnableConfig | None = None,
) -> type[BaseModel]
Get a pydantic model that can be used to validate output to the Runnable.
Runnables 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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
type[BaseModel]
|
A pydantic model that can be used to validate output. |
Get a JSON schema that represents the output of the Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
A config to use when generating the schema. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A JSON schema that represents the output of the |
Example
Added in version 0.3.0
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include
|
Sequence[str] | None
|
A list of fields to include in the config schema. |
None
|
Returns:
| Type | Description |
|---|---|
type[BaseModel]
|
A pydantic model that can be used to validate config. |
Get a JSON schema that represents the config of the Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include
|
Sequence[str] | None
|
A list of fields to include in the config schema. |
None
|
Returns:
| Type | Description |
|---|---|
dict[str, Any]
|
A JSON schema that represents the config of the |
Added in version 0.3.0
Return a graph representation of this Runnable.
get_prompts(
config: RunnableConfig | None = None,
) -> list[BasePromptTemplate]
Return a list of prompts used by this Runnable.
__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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
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]
|
Another |
required |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
__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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
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]
|
Another |
required |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Other, Output]
|
A new |
pipe(
*others: Runnable[Any, Other] | Callable[[Any], Other],
name: str | None = None
) -> RunnableSerializable[Input, Other]
Pipe runnables.
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]
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
*others
|
Runnable[Any, Other] | Callable[[Any], Other]
|
Other |
()
|
name
|
str | None
|
An optional name for the resulting |
None
|
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Other]
|
A new |
Pick keys from the output dict of this Runnable.
Pick single key:
```python
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 list of keys:
```python
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]"}
```
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
keys
|
str | list[str]
|
A key or list of keys to pick from the output dict. |
required |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Any, Any]
|
a new |
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_community.llms.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}"
)
llm = FakeStreamingListLLM(responses=["foo-lish"])
chain: Runnable = prompt | llm | {"str": StrOutputParser()}
chain_with_assign = chain.assign(hello=itemgetter("str") | llm)
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'}}}
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any] | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]]
|
A mapping of keys to |
{}
|
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Any, Any]
|
A new |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
Sequence[Input]
|
A list of inputs to the |
required |
config
|
RunnableConfig | Sequence[RunnableConfig] | None
|
A config to use when invoking the |
None
|
return_exceptions
|
bool
|
Whether to return exceptions instead of raising them. Defaults to False. |
False
|
**kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
tuple[int, Output | Exception]
|
Tuples of the index of the input and the output from the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
inputs
|
Sequence[Input]
|
A list of inputs to the |
required |
config
|
RunnableConfig | Sequence[RunnableConfig] | None
|
A config to use when invoking the |
None
|
return_exceptions
|
bool
|
Whether to return exceptions instead of raising them. Defaults to False. |
False
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[tuple[int, Output | Exception]]
|
A tuple of the index of the input and the output from the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Any
|
The input to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
diff
|
bool
|
Whether to yield diffs between each step or the current state. |
True
|
with_streamed_output_list
|
bool
|
Whether to yield the |
True
|
include_names
|
Sequence[str] | None
|
Only include logs with these names. |
None
|
include_types
|
Sequence[str] | None
|
Only include logs with these types. |
None
|
include_tags
|
Sequence[str] | None
|
Only include logs with these tags. |
None
|
exclude_names
|
Sequence[str] | None
|
Exclude logs with these names. |
None
|
exclude_types
|
Sequence[str] | None
|
Exclude logs with these types. |
None
|
exclude_tags
|
Sequence[str] | None
|
Exclude logs with these tags. |
None
|
kwargs
|
Any
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]
|
A |
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 StreamEvents that provide real-time information
about the progress of the Runnable, including StreamEvents from intermediate
results.
A StreamEvent is a dictionary with the following schema:
event: str - Event names are of the format:on_[runnable_type]_(start|stream|end).name: str - The name of theRunnablethat generated the event.run_id: str - randomly generated ID associated with the given execution of theRunnablethat emitted the event. A childRunnablethat gets invoked as part of the execution of a parentRunnableis assigned its own unique ID.parent_ids: list[str] - The IDs of the parent runnables that generated the event. The rootRunnablewill 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: Optional[list[str]] - The tags of theRunnablethat generated the event.metadata: Optional[dict[str, Any]] - The metadata of theRunnablethat generated the event.data: dict[str, Any]
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:
prompt:
template = ChatPromptTemplate.from_messages(
[
("system", "You are Cat Agent 007"),
("human", "{question}"),
]
).with_config({"run_name": "my_template", "tags": ["my_template"]})
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": [],
},
]
Example: 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)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Any
|
The input to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
version
|
Literal['v1', 'v2']
|
The version of the schema to use either |
'v2'
|
include_names
|
Sequence[str] | None
|
Only include events from |
None
|
include_types
|
Sequence[str] | None
|
Only include events from |
None
|
include_tags
|
Sequence[str] | None
|
Only include events from |
None
|
exclude_names
|
Sequence[str] | None
|
Exclude events from |
None
|
exclude_types
|
Sequence[str] | None
|
Exclude events from |
None
|
exclude_tags
|
Sequence[str] | None
|
Exclude events from |
None
|
kwargs
|
Any
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[StreamEvent]
|
An async stream of |
Raises:
| Type | Description |
|---|---|
NotImplementedError
|
If the version is not |
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 should override this method if they can start producing output while input is still being generated.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
Iterator[Input]
|
An iterator of inputs to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
Output
|
The output of the |
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 should override this method if they can start producing output while input is still being generated.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input
|
AsyncIterator[Input]
|
An async iterator of inputs to the |
required |
config
|
RunnableConfig | None
|
The config to use for the |
None
|
kwargs
|
Any | None
|
Additional keyword arguments to pass to the |
{}
|
Yields:
| Type | Description |
|---|---|
AsyncIterator[Output]
|
The output of the |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
kwargs
|
Any
|
The arguments to bind to the |
{}
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
Example
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
llm = ChatOllama(model="llama3.1")
# Without bind.
chain = llm | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two three four five.'
# With bind.
chain = llm.bind(stop=["three"]) | StrOutputParser()
chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# Output is 'One two'
with_config(
config: RunnableConfig | None = None, **kwargs: Any
) -> Runnable[Input, Output]
Bind config to a Runnable, returning a new Runnable.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config
|
RunnableConfig | None
|
The config to bind to the |
None
|
kwargs
|
Any
|
Additional keyword arguments to pass to the |
{}
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
on_start
|
Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None
|
Called before the |
None
|
on_end
|
Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None
|
Called after the |
None
|
on_error
|
Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None
|
Called if the |
None
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
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(
*,
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
on_start
|
AsyncListener | None
|
Called asynchronously before the |
None
|
on_end
|
AsyncListener | None
|
Called asynchronously after the |
None
|
on_error
|
AsyncListener | None
|
Called asynchronously if the |
None
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new |
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(
*,
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
input_type
|
type[Input] | None
|
The input type to bind to the |
None
|
output_type
|
type[Output] | None
|
The output type to bind to the |
None
|
Returns:
| Type | Description |
|---|---|
Runnable[Input, Output]
|
A new Runnable with the types bound. |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
retry_if_exception_type
|
tuple[type[BaseException], ...]
|
A tuple of exception types to retry on. Defaults to (Exception,). |
(Exception,)
|
wait_exponential_jitter
|
bool
|
Whether to add jitter to the wait time between retries. Defaults to True. |
True
|
stop_after_attempt
|
int
|
The maximum number of attempts to make before giving up. Defaults to 3. |
3
|
exponential_jitter_params
|
ExponentialJitterParams | None
|
Parameters for
|
None
|
Returns:
| Type | 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
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fallbacks
|
Sequence[Runnable[Input, Output]]
|
A sequence of runnables to try if the original |
required |
exceptions_to_handle
|
tuple[type[BaseException], ...]
|
A tuple of exception types to handle.
Defaults to |
(Exception,)
|
exception_key
|
str | None
|
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 |
None
|
Returns:
| Type | Description |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
RunnableWithFallbacks[Input, Output]
|
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
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fallbacks
|
Sequence[Runnable[Input, Output]]
|
A sequence of runnables to try if the original |
required |
exceptions_to_handle
|
tuple[type[BaseException], ...]
|
A tuple of exception types to handle. |
(Exception,)
|
exception_key
|
str | None
|
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 |
None
|
Returns:
| Type | Description |
|---|---|
RunnableWithFallbacks[Input, Output]
|
A new |
RunnableWithFallbacks[Input, Output]
|
fallback in order, upon failures. |
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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
args_schema
|
type[BaseModel] | None
|
The schema for the tool. Defaults to None. |
None
|
name
|
str | None
|
The name of the tool. Defaults to None. |
None
|
description
|
str | None
|
The description of the tool. Defaults to None. |
None
|
arg_types
|
dict[str, type] | None
|
A dictionary of argument names to types. Defaults to None. |
None
|
Returns:
| Type | Description |
|---|---|
BaseTool
|
A |
Typed dict 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]})
String 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")
Added in version 0.2.14
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:
| Type | Description |
|---|---|
bool
|
Whether the class is serializable. Default is False. |
classmethod
¶
classmethod
¶
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"].
Serialize the Runnable to JSON.
Returns:
| Type | Description |
|---|---|
SerializedConstructor | SerializedNotImplemented
|
A JSON-serializable representation of the |
Serialize a "not implemented" object.
Returns:
| Type | Description |
|---|---|
SerializedNotImplemented
|
SerializedNotImplemented. |
Configure particular Runnable fields at runtime.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
**kwargs
|
AnyConfigurableField
|
A dictionary of |
{}
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If a configuration key is not found in the |
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
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(
which: ConfigurableField,
*,
default_key: str = "default",
prefix_keys: bool = False,
**kwargs: (
Runnable[Input, Output]
| Callable[[], Runnable[Input, Output]]
)
) -> RunnableSerializable[Input, Output]
Configure alternatives for Runnables that can be set at runtime.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
which
|
ConfigurableField
|
The |
required |
default_key
|
str
|
The default key to use if no alternative is selected.
Defaults to |
'default'
|
prefix_keys
|
bool
|
Whether to prefix the keys with the |
False
|
**kwargs
|
Runnable[Input, Output] | Callable[[], Runnable[Input, Output]]
|
A dictionary of keys to |
{}
|
Returns:
| Type | Description |
|---|---|
RunnableSerializable[Input, Output]
|
A new |
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI
model = ChatAnthropic(
model_name="claude-3-7-sonnet-20250219"
).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
)
with_structured_output(
schema: dict | type, **kwargs: Any
) -> Runnable[LanguageModelInput, dict | BaseModel]
Not implemented on this class.
Get the number of tokens in the messages.
Useful for checking if an input fits in a model's context window.
Note
The base implementation of get_num_tokens_from_messages ignores tool
schemas.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
messages
|
list[BaseMessage]
|
The message inputs to tokenize. |
required |
tools
|
Sequence | None
|
If provided, sequence of dict, |
None
|
Returns:
| Type | Description |
|---|---|
int
|
The sum of the number of tokens across the messages. |
generate(
prompts: list[str],
stop: list[str] | None = None,
callbacks: Callbacks | list[Callbacks] | None = None,
*,
tags: list[str] | list[list[str]] | None = None,
metadata: (
dict[str, Any] | list[dict[str, Any]] | None
) = None,
run_name: str | list[str] | None = None,
run_id: UUID | list[UUID | None] | None = None,
**kwargs: Any
) -> LLMResult
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:
- Take advantage of batched calls,
- Need more output from the model than just the top generated value,
- Are building chains that are agnostic to the underlying language model type (e.g., pure text completion models vs chat models).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompts
|
list[str]
|
List of string prompts. |
required |
stop
|
list[str] | None
|
Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. |
None
|
callbacks
|
Callbacks | list[Callbacks] | None
|
Callbacks to pass through. Used for executing additional functionality, such as logging or streaming, throughout generation. |
None
|
tags
|
list[str] | list[list[str]] | None
|
List of tags to associate with each prompt. If provided, the length of the list must match the length of the prompts list. |
None
|
metadata
|
dict[str, Any] | list[dict[str, Any]] | None
|
List of metadata dictionaries to associate with each prompt. If provided, the length of the list must match the length of the prompts list. |
None
|
run_name
|
str | list[str] | None
|
List of run names to associate with each prompt. If provided, the length of the list must match the length of the prompts list. |
None
|
run_id
|
UUID | list[UUID | None] | None
|
List of run IDs to associate with each prompt. If provided, the length of the list must match the length of the prompts list. |
None
|
**kwargs
|
Any
|
Arbitrary additional keyword arguments. These are usually passed to the model provider API call. |
{}
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If prompts is not a list. |
ValueError
|
If the length of |
Returns:
| Type | Description |
|---|---|
LLMResult
|
An LLMResult, which contains a list of candidate Generations for each input prompt and additional model provider-specific output. |
async
¶
agenerate(
prompts: list[str],
stop: list[str] | None = None,
callbacks: Callbacks | list[Callbacks] | None = None,
*,
tags: list[str] | list[list[str]] | None = None,
metadata: (
dict[str, Any] | list[dict[str, Any]] | None
) = None,
run_name: str | list[str] | None = None,
run_id: UUID | list[UUID | None] | 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:
- Take advantage of batched calls,
- Need more output from the model than just the top generated value,
- Are building chains that are agnostic to the underlying language model type (e.g., pure text completion models vs chat models).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prompts
|
list[str]
|
List of string prompts. |
required |
stop
|
list[str] | None
|
Stop words to use when generating. Model output is cut off at the first occurrence of any of these substrings. |
None
|
callbacks
|
Callbacks | list[Callbacks] | None
|
Callbacks to pass through. Used for executing additional functionality, such as logging or streaming, throughout generation. |
None
|
tags
|
list[str] | list[list[str]] | None
|
List of tags to associate with each prompt. If provided, the length of the list must match the length of the prompts list. |
None
|
metadata
|
dict[str, Any] | list[dict[str, Any]] | None
|
List of metadata dictionaries to associate with each prompt. If provided, the length of the list must match the length of the prompts list. |
None
|
run_name
|
str | list[str] | None
|
List of run names to associate with each prompt. If provided, the length of the list must match the length of the prompts list. |
None
|
run_id
|
UUID | list[UUID | None] | None
|
List of run IDs to associate with each prompt. If provided, the length of the list must match the length of the prompts list. |
None
|
**kwargs
|
Any
|
Arbitrary additional keyword arguments. These are usually passed to the model provider API call. |
{}
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If the length of |
Returns:
| Type | Description |
|---|---|
LLMResult
|
An LLMResult, which contains a list of candidate Generations for each input prompt and additional model provider-specific output. |
Save the LLM.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
file_path
|
Path | str
|
Path to file to save the LLM to. |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the file path is not a string or Path object. |
Example:
.. code-block:: python
llm.save(file_path="path/llm.yaml")
Validates params and passes them to google-generativeai package.