ChatVertexAI#

class langchain_google_vertexai.chat_models.ChatVertexAI[source]#

基础类：_VertexAICommon, BaseChatModel

Google Cloud Vertex AI 聊天模型集成。

Setup:
您必须安装 langchain-google-vertexai Python 包 .. code-block:: bash

pip install -U langchain-google-vertexai

And either:

为您的环境配置凭据（gcloud、工作负载身份等…）

将服务账户JSON文件的路径存储为GOOGLE_APPLICATION_CREDENTIALS环境变量

此代码库使用 google.auth 库，该库首先查找上述提到的应用程序凭据变量，然后查找系统级身份验证。

欲了解更多信息，请参阅： https://cloud.google.com/docs/authentication/application-default-credentials#GAC 和 https://googleapis.dev/python/google-auth/latest/reference/google.auth.html#module-google.auth。

Key init args — completion params:

model: str
使用的ChatVertexAI模型名称。例如“gemini-1.5-flash-001”、“gemini-1.5-pro-001”等。

temperature: Optional[float]
采样温度。

seed: Optional[int]
使用的采样整数。

max_tokens: Optional[int]
生成的最大令牌数。

stop: Optional[List[str]]
默认的停止序列。

safety_settings: Optional[Dict[vertexai.generative_models.HarmCategory, vertexai.generative_models.HarmBlockThreshold]]
用于所有生成的默认安全设置。

Key init args — client params:

max_retries: int
最大重试次数。

credentials: Optional[google.auth.credentials.Credentials]
在进行API调用时使用的默认自定义凭证。如果未提供，将从环境中获取凭证。

project: Optional[str]
在进行Vertex API调用时使用的默认GCP项目。

location: str = “us-central1”
进行API调用时使用的默认位置。

request_parallelism: int = 5
允许向VertexAI模型发出的请求的并行量。默认值为5。

base_url: Optional[str]
API请求的基础URL。

请参阅参数部分中支持的初始化参数及其描述的完整列表。
Instantiate:
from langchain_google_vertexai import ChatVertexAI

llm = ChatVertexAI(
    model="gemini-1.5-flash-001",
    temperature=0,
    max_tokens=None,
    max_retries=6,
    stop=None,
    # other params...
)
Invoke:
messages = [
    ("system", "You are a helpful translator. Translate the user sentence to French."),
    ("human", "I love programming."),
]
llm.invoke(messages)
AIMessage(content="J'adore programmer.

“, response_metadata={‘is_blocked’: False, ‘safety_ratings’: [{‘category’: ‘HARM_CATEGORY_HATE_SPEECH’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_DANGEROUS_CONTENT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_HARASSMENT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_SEXUALLY_EXPLICIT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}], ‘citation_metadata’: None, ‘usage_metadata’: {‘prompt_token_count’: 17, ‘candidates_token_count’: 7, ‘total_token_count’: 24}}, id=’run-925ce305-2268-44c4-875f-dde9128520ad-0’)

Stream:

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

AIMessageChunk(content='J', response_metadata={'is_blocked': False, 'safety_ratings': [], 'citation_metadata': None}, id='run-9df01d73-84d9-42db-9d6b-b1466a019e89')
AIMessageChunk(content="'adore programmer.

“, response_metadata={‘is_blocked’: False, ‘safety_ratings’: [{‘category’: ‘HARM_CATEGORY_HATE_SPEECH’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_DANGEROUS_CONTENT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_HARASSMENT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_SEXUALLY_EXPLICIT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}], ‘citation_metadata’: None}, id=’run-9df01d73-84d9-42db-9d6b-b1466a019e89’)

AIMessageChunk(content=’’, response_metadata={‘is_blocked’: False, ‘safety_ratings’: [], ‘citation_metadata’: None, ‘usage_metadata’: {‘prompt_token_count’: 17, ‘candidates_token_count’: 7, ‘total_token_count’: 24}}, id=’run-9df01d73-84d9-42db-9d6b-b1466a019e89’)

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

AIMessageChunk(content="J'adore programmer.

“, response_metadata={‘is_blocked’: False, ‘safety_ratings’: [{‘category’: ‘HARM_CATEGORY_HATE_SPEECH’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_DANGEROUS_CONTENT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_HARASSMENT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_SEXUALLY_EXPLICIT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}], ‘citation_metadata’: None, ‘usage_metadata’: {‘prompt_token_count’: 17, ‘candidates_token_count’: 7, ‘total_token_count’: 24}}, id=’run-b7f7492c-4cb5-42d0-8fc3-dce9b293b0fb’)

Async:

await llm.ainvoke(messages)

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

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

AIMessage(content="J'adore programmer.

“, response_metadata={‘is_blocked’: False, ‘safety_ratings’: [{‘category’: ‘HARM_CATEGORY_HATE_SPEECH’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_DANGEROUS_CONTENT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_HARASSMENT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}, {‘category’: ‘HARM_CATEGORY_SEXUALLY_EXPLICIT’, ‘probability_label’: ‘NEGLIGIBLE’, ‘probability_score’: 0.1, ‘blocked’: False, ‘severity’: ‘HARM_SEVERITY_NEGLIGIBLE’, ‘severity_score’: 0.1}], ‘citation_metadata’: None, ‘usage_metadata’: {‘prompt_token_count’: 17, ‘candidates_token_count’: 7, ‘total_token_count’: 24}}, id=’run-925ce305-2268-44c4-875f-dde9128520ad-0’)

Tool calling:

from pydantic import BaseModel, Field

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

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

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

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

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

[{'name': 'GetWeather',
  'args': {'location': 'Los Angeles, CA'},
  'id': '2a2401fa-40db-470d-83ce-4e52de910d9e'},
 {'name': 'GetWeather',
  'args': {'location': 'New York City, NY'},
  'id': '96761deb-ab7f-4ef9-b4b4-6d44562fc46e'},
 {'name': 'GetPopulation',
  'args': {'location': 'Los Angeles, CA'},
  'id': '9147d532-abee-43a2-adb5-12f164300484'},
 {'name': 'GetPopulation',
  'args': {'location': 'New York City, NY'},
  'id': 'c43374ea-bde5-49ca-8487-5b83ebeea1e6'}]

有关更多信息，请参见 ChatVertexAI.bind_tools() 方法。

Structured output:

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

structured_llm = llm.with_structured_output(Joke)
structured_llm.invoke("Tell me a joke about cats")

Joke(setup='What do you call a cat that loves to bowl?', punchline='An alley cat!', rating=None)

更多信息请参见 ChatVertexAI.with_structured_output()。

Image input:

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

'The weather in this image appears to be sunny and pleasant. The sky is a bright blue with scattered white clouds, suggesting a clear and mild day. The lush green grass indicates recent rainfall or sufficient moisture. The absence of strong shadows suggests that the sun is high in the sky, possibly late afternoon. Overall, the image conveys a sense of tranquility and warmth, characteristic of a beautiful summer day.

‘

您还可以指向GCS文件，这样更快/更高效，因为字节会来回传输。

llm.invoke(
    [
        HumanMessage(
            [
                "What's in the image?",
                {
                    "type": "media",
                    "file_uri": "gs://cloud-samples-data/generative-ai/image/scones.jpg",
                    "mime_type": "image/jpeg",
                },
            ]
        )
    ]
).content

'The image is of five blueberry scones arranged on a piece of baking paper.

以下是图片中的内容列表： * 五个蓝莓司康饼： 它们散落在烘焙纸上，撒上了糖粉。 * 两杯咖啡： 两个带有碟子的白色杯子。一个看起来是满的，另一个已经喝了一部分。 * 一碗蓝莓： 一个棕色的碗里装满了新鲜的蓝莓，放在司康饼旁边。 * 一个勺子： 一个银色的勺子，上面写着“Let’s Jam”，放在纸上。 * 粉色牡丹： 几朵粉色牡丹放在司康饼旁边，增添了一抹色彩。 * 烘焙纸： 司康饼、杯子、碗和勺子都摆在一张白色的烘焙纸上，纸上溅有紫色斑点。纸张皱褶，放在一个深色的表面上。

这张图片有一种乡村和美味的感觉，暗示了一个舒适愉快的早餐或早午餐场景。 ‘

Video input:

注意：目前仅支持gemini-...-vision模型。

llm = ChatVertexAI(model="gemini-1.0-pro-vision")

llm.invoke(
    [
        HumanMessage(
            [
                "What's in the video?",
                {
                    "type": "media",
                    "file_uri": "gs://cloud-samples-data/video/animals.mp4",
                    "mime_type": "video/mp4",
                },
            ]
        )
    ]
).content

'The video is about a new feature in Google Photos called "Zoomable Selfies". The feature allows users to take selfies with animals at the zoo. The video shows several examples of people taking selfies with animals, including a tiger, an elephant, and a sea otter. The video also shows how the feature works. Users simply need to open the Google Photos app and select the "Zoomable Selfies" option. Then, they need to choose an animal from the list of available animals. The app will then guide the user through the process of taking the selfie.'

Audio input:

from langchain_core.messages import HumanMessage

llm = ChatVertexAI(model="gemini-1.5-flash-001")

llm.invoke(
    [
        HumanMessage(
            [
                "What's this audio about?",
                {
                    "type": "media",
                    "file_uri": "gs://cloud-samples-data/generative-ai/audio/pixel.mp3",
                    "mime_type": "audio/mpeg",
                },
            ]
        )
    ]
).content

"This audio is an interview with two product managers from Google who work on Pixel feature drops. They discuss how feature drops are important for showcasing how Google devices are constantly improving and getting better. They also discuss some of the highlights of the January feature drop and the new features coming in the March drop for Pixel phones and Pixel watches. The interview concludes with discussion of how user feedback is extremely important to them in deciding which features to include in the feature drops. "

Token usage:

ai_msg = llm.invoke(messages)
ai_msg.usage_metadata

{'input_tokens': 17, 'output_tokens': 7, 'total_tokens': 24}

Logprobs:

llm = ChatVertexAI(model="gemini-1.5-flash-001", logprobs=True)
ai_msg = llm.invoke(messages)
ai_msg.response_metadata["logprobs_result"]

[
    {'token': 'J', 'logprob': -1.549651415189146e-06, 'top_logprobs': []},
    {'token': "'", 'logprob': -1.549651415189146e-06, 'top_logprobs': []},
    {'token': 'adore', 'logprob': 0.0, 'top_logprobs': []},
    {'token': ' programmer', 'logprob': -1.1922384146600962e-07, 'top_logprobs': []},
    {'token': '.', 'logprob': -4.827636439586058e-05, 'top_logprobs': []},
    {'token': ' ', 'logprob': -0.018011733889579773, 'top_logprobs': []},
    {'token': '

‘, ‘logprob’: -0.0008687592926435173, ‘top_logprobs’: []}

]

Response metadata

ai_msg = llm.invoke(messages)
ai_msg.response_metadata

{'is_blocked': False,
 'safety_ratings': [{'category': 'HARM_CATEGORY_HATE_SPEECH',
   'probability_label': 'NEGLIGIBLE',
   'probability_score': 0.1,
   'blocked': False,
   'severity': 'HARM_SEVERITY_NEGLIGIBLE',
   'severity_score': 0.1},
  {'category': 'HARM_CATEGORY_DANGEROUS_CONTENT',
   'probability_label': 'NEGLIGIBLE',
   'probability_score': 0.1,
   'blocked': False,
   'severity': 'HARM_SEVERITY_NEGLIGIBLE',
   'severity_score': 0.1},
  {'category': 'HARM_CATEGORY_HARASSMENT',
   'probability_label': 'NEGLIGIBLE',
   'probability_score': 0.1,
   'blocked': False,
   'severity': 'HARM_SEVERITY_NEGLIGIBLE',
   'severity_score': 0.1},
  {'category': 'HARM_CATEGORY_SEXUALLY_EXPLICIT',
   'probability_label': 'NEGLIGIBLE',
   'probability_score': 0.1,
   'blocked': False,
   'severity': 'HARM_SEVERITY_NEGLIGIBLE',
   'severity_score': 0.1}],
 'usage_metadata': {'prompt_token_count': 17,
  'candidates_token_count': 7,
  'total_token_count': 24}}

Safety settings

from langchain_google_vertexai import HarmBlockThreshold, HarmCategory

llm = ChatVertexAI(
    model="gemini-1.5-pro",
    safety_settings={
        HarmCategory.HARM_CATEGORY_HATE_SPEECH: HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
        HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
        HarmCategory.HARM_CATEGORY_HARASSMENT: HarmBlockThreshold.BLOCK_LOW_AND_ABOVE,
        HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: HarmBlockThreshold.BLOCK_ONLY_HIGH,
    },
)

llm.invoke(messages).response_metadata

{'is_blocked': False,
 'safety_ratings': [{'category': 'HARM_CATEGORY_HATE_SPEECH',
   'probability_label': 'NEGLIGIBLE',
   'probability_score': 0.1,
   'blocked': False,
   'severity': 'HARM_SEVERITY_NEGLIGIBLE',
   'severity_score': 0.1},
  {'category': 'HARM_CATEGORY_DANGEROUS_CONTENT',
   'probability_label': 'NEGLIGIBLE',
   'probability_score': 0.1,
   'blocked': False,
   'severity': 'HARM_SEVERITY_NEGLIGIBLE',
   'severity_score': 0.1},
  {'category': 'HARM_CATEGORY_HARASSMENT',
   'probability_label': 'NEGLIGIBLE',
   'probability_score': 0.1,
   'blocked': False,
   'severity': 'HARM_SEVERITY_NEGLIGIBLE',
   'severity_score': 0.1},
  {'category': 'HARM_CATEGORY_SEXUALLY_EXPLICIT',
   'probability_label': 'NEGLIGIBLE',
   'probability_score': 0.1,
   'blocked': False,
   'severity': 'HARM_SEVERITY_NEGLIGIBLE',
   'severity_score': 0.1}],
 'usage_metadata': {'prompt_token_count': 17,
  'candidates_token_count': 7,
  'total_token_count': 24}}

mypy 类型检查需要将 model_name 识别为有效参数。

注意

ChatVertexAI 实现了标准的 Runnable Interface。🏃

Runnable Interface 接口在可运行对象上提供了额外的方法，例如 with_types, with_retry, assign, bind, get_graph, 等等。

param additional_headers: Dict[str, str] | None = None#: 表示模型调用的附加头信息的键值字典

param api_endpoint: str | None = None (alias 'base_url')#: 所需的API端点，例如，us-central1-aiplatform.googleapis.com

param api_transport: str | None = None#: 所需的API传输方法，可以是‘grpc’或‘rest’。如果定义了vertexai.init中的默认参数，则使用该参数。

param cache: BaseCache | bool | None = None#

是否缓存响应。

如果为真，将使用全局缓存。
如果为false，将不使用缓存
如果为None，将使用全局缓存（如果已设置），否则不使用缓存。
如果是 BaseCache 的实例，将使用提供的缓存。

目前不支持对模型的流式方法进行缓存。

param cached_content: str | None = None#: 可选。在缓存模式下使用模型。仅在Gemini 1.5及更高版本中支持。必须是一个包含缓存名称的字符串（一串数字）

param callback_manager: BaseCallbackManager | None = None#: 自版本0.1.7起已弃用：请改用callbacks()。它将在pydantic==1.0中被移除。

回调管理器以添加到运行跟踪中。

param callbacks: Callbacks = None#: 添加到运行跟踪的回调。

param client_cert_source: Callable[[], Tuple[bytes, bytes]] | None = None#: 一个回调函数，返回客户端证书字节和私钥字节

param convert_system_message_to_human: bool = False#: [已弃用] 由于新的Gemini模型支持设置系统消息，因此不建议将此参数设置为True。

param credentials: Any = None#: 默认使用的自定义凭据 (google.auth.credentials.Credentials)

param custom_get_token_ids: Callable[[str], list[int]] | None = None#: 用于计数标记的可选编码器。

param disable_streaming: bool | Literal['tool_calling'] = False#

是否禁用此模型的流式传输。

如果流式传输被绕过，那么 stream()/astream() 将依赖于 invoke()/ainvoke()。

如果为True，将始终绕过流式传输情况。
如果是“tool_calling”，只有在使用tools关键字参数调用模型时，才会绕过流式处理的情况。
如果为 False（默认值），将始终使用流式情况（如果可用）。

param examples: List[BaseMessage] | None = None#

param full_model_name: str | None = None#: 模型端点的全名。

param location: str = 'us-central1'#: 进行API调用时使用的默认位置。

param logprobs: bool | int = False#

是否将logprobs作为AIMessage.response_metadata的一部分返回。

如果为False，则不返回logprobs。如果为True，则返回最佳候选的logprobs。如果为整数，则返回前logprobs个候选的logprobs。

注意: 截至10.28.24，此功能仅支持gemini-1.5-flash模型。

param max_output_tokens: int | None = None (alias 'max_tokens')#: 令牌限制决定了从一个提示中输出的最大文本量。

param max_retries: int = 6#: 生成时的最大重试次数。

param metadata: dict[str, Any] | None = None#: 要添加到运行跟踪的元数据。

param model_name: str = 'chat-bison-default' (alias 'model')#: 底层模型名称。

param n: int = 1#: 为每个提示生成多少个完成项。

param project: str | None = None#: 在进行Vertex API调用时使用的默认GCP项目。

param rate_limiter: BaseRateLimiter | None = None#: 一个可选的速率限制器，用于限制请求的数量。

param request_parallelism: int = 5#: 允许向VertexAI模型发出的请求的并行量。

param response_mime_type: str | None = None#

Optional. Output response mimetype of the generated candidate text. Only

supported in Gemini 1.5 and later models. Supported mimetype:

“text/plain”: （默认）文本输出。
“application/json”: 候选者中的JSON响应。
“text/x.enum”: 纯文本中的枚举。

模型还需要被提示以输出适当的响应类型，否则行为是未定义的。这是一个预览功能。

param response_schema: Dict[str, Any] | None = None#: 可选。对输出强制执行一个模式。字典的格式应遵循Open API模式。

param safety_settings: 'SafetySettingsType' | None = None#

用于所有生成的默认安全设置。

例如：

从langchain_google_vertexai导入HarmBlockThreshold, HarmCategory

safety_settings = {
HarmCategory.HARM_CATEGORY_UNSPECIFIED: HarmBlockThreshold.BLOCK_NONE, 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,

}

param seed: int | None = None#: 用于生成的随机种子。

param stop: List[str] | None = None (alias 'stop_sequences')#: 生成时使用的可选停用词列表。

param streaming: bool = False#: 是否流式传输结果。

param tags: list[str] | None = None#: 要添加到运行跟踪的标签。

param temperature: float | None = None#: 采样温度，它控制着标记选择的随机程度。

param top_k: int | None = None#: 模型如何选择输出的标记，下一个标记是从

param top_p: float | None = None#: 从最可能到最不可能选择令牌，直到它们的总和

param tuned_model_name: str | None = None#: 调优模型的名称。如果传递了tuned_model_name，model_name将用于确定模型系列

param verbose: bool [Optional]#: 是否打印出响应文本。

__call__(messages: list[BaseMessage], stop: list[str] | None = None, callbacks: list[BaseCallbackHandler] | BaseCallbackManager | None = None, **kwargs: Any) → BaseMessage#

自版本0.1.7起已弃用：请改用invoke()。在langchain-core==1.0之前不会移除。

Parameters:

messages (列表[BaseMessage])
stop (列表[字符串] | 无)
callbacks (列表[BaseCallbackHandler] | BaseCallbackManager | 无)
kwargs (Any)

Return type:

BaseMessage

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

默认实现使用asyncio.gather并行运行ainvoke。

batch的默认实现对于IO绑定的runnables效果很好。

如果子类能够更高效地进行批处理，则应重写此方法；例如，如果底层的Runnable使用支持批处理模式的API。

Parameters:

inputs (list[Input]) – Runnable 的输入列表。
config (RunnableConfig | list[RunnableConfig] | None) – 调用Runnable时使用的配置。该配置支持标准键，如用于跟踪目的的‘tags’、‘metadata’，用于控制并行工作量的‘max_concurrency’，以及其他键。更多详情请参考RunnableConfig。默认为None。
return_exceptions (bool) – 是否返回异常而不是抛出它们。默认为 False。
kwargs (Any | None) – 传递给Runnable的额外关键字参数。

Returns:

Runnable 的输出列表。

Return type:

列表[输出]

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]]#

在输入列表上并行运行ainvoke，在它们完成时产生结果。

Parameters:

inputs (Sequence[Input]) – Runnable 的输入列表。
config (RunnableConfig | Sequence[RunnableConfig] | None) – 调用Runnable时使用的配置。该配置支持标准键，如用于跟踪目的的'tags'、'metadata'，用于控制并行工作量的'max_concurrency'，以及其他键。有关更多详细信息，请参阅RunnableConfig。默认为None。默认为None。
return_exceptions (bool) – 是否返回异常而不是抛出它们。默认为 False。
kwargs (Any | None) – 传递给Runnable的额外关键字参数。

Yields:

输入索引和Runnable输出的元组。

Return type:

AsyncIterator[元组[int, Output | 异常]]

async ainvoke(input: LanguageModelInput, config: RunnableConfig | None = None, *, stop: list[str] | None = None, **kwargs: Any) → BaseMessage#

ainvoke的默认实现，从线程调用invoke。

默认实现允许使用异步代码，即使Runnable没有实现本地的异步版本的invoke。

如果子类可以异步运行，则应重写此方法。

Parameters:

输入 (LanguageModelInput)
config (可选[RunnableConfig])
stop (可选[列表[字符串]])
kwargs (Any)

Return type:

BaseMessage

async astream(input: LanguageModelInput, config: RunnableConfig | None = None, *, stop: list[str] | None = None, **kwargs: Any) → AsyncIterator[BaseMessageChunk]#

astream的默认实现，调用ainvoke。如果子类支持流式输出，则应重写此方法。

Parameters:

input (LanguageModelInput) – Runnable 的输入。
config (可选[RunnableConfig]) – 用于Runnable的配置。默认为None。
kwargs (Any) – 传递给Runnable的额外关键字参数。
stop (可选[列表[字符串]])

Yields:

Runnable 的输出。

Return type:

异步迭代器[BaseMessageChunk]

async astream_events(input: Any, config: RunnableConfig | None = None, *, version: Literal['v1', '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[StandardStreamEvent | CustomStreamEvent]#

生成事件流。

用于创建一个迭代器，遍历提供实时信息的StreamEvents，包括来自中间结果的StreamEvents。

StreamEvent 是一个具有以下模式的字典：

event: str - 事件名称的格式为
格式: on_[runnable_type]_(start|stream|end).
name: str - 生成事件的 Runnable 的名称。
run_id: str - 与给定执行相关联的随机生成的ID
发出事件的Runnable。作为父Runnable执行的一部分被调用的子Runnable会被分配其自己唯一的ID。
parent_ids: List[str] - 生成事件的父可运行对象的ID。
根可运行对象将有一个空列表。父ID的顺序是从根到直接父对象。仅适用于API的v2版本。API的v1版本将返回一个空列表。
tags: Optional[List[str]] - 生成事件的Runnable的标签
事件。
metadata: Optional[Dict[str, Any]] - Runnable的元数据
生成事件的元数据。
data: Dict[str, Any]

下表展示了一些可能由不同链发出的事件。为了简洁起见，表中省略了元数据字段。链定义已包含在表后。

注意此参考表适用于V2版本的架构。

事件	名称	块	输入	输出
on_chat_model_start	[模型名称]		{“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	[模型名称]	‘Hello’
on_llm_end	[model name]		‘你好，人类！’
链上开始	格式化文档
on_chain_stream	format_docs	“你好世界！，再见世界！”
on_chain_end	format_docs		[Document(…)]	“你好世界！，再见世界！”
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, …])

除了标准事件外，用户还可以派发自定义事件（见下面的示例）。

自定义事件将仅在API的v2版本中显示！

自定义事件具有以下格式：

属性	类型	描述
name	str	用户定义的事件名称。
data	Any	与事件相关的数据。这可以是任何内容，但我们建议使其可JSON序列化。

以下是上述标准事件相关的声明：

format_docs:

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

format_docs = RunnableLambda(format_docs)

some_tool:

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

提示:

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": [],
    },
]

示例：分发自定义事件

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:

input (Any) – Runnable 的输入。
config (RunnableConfig | None) – 用于Runnable的配置。
version (Literal['v1', 'v2']) – 使用的模式版本，可以是 v2 或 v1。用户应使用 v2。 v1 是为了向后兼容，将在 0.4.0 版本中弃用。在 API 稳定之前不会分配默认值。自定义事件仅在 v2 中显示。
include_names (Sequence[str] | None) – 仅包含来自具有匹配名称的可运行对象的事件。
include_types (Sequence[str] | None) – 仅包含来自具有匹配类型的可运行对象的事件。
include_tags (Sequence[str] | None) – 仅包含具有匹配标签的可运行对象的事件。
exclude_names (Sequence[str] | None) – 排除具有匹配名称的可运行对象的事件。
exclude_types (Sequence[str] | None) – 排除具有匹配类型的可运行对象的事件。
exclude_tags (Sequence[str] | None) – 排除具有匹配标签的可运行对象的事件。
kwargs (Any) – 传递给 Runnable 的额外关键字参数。这些参数将传递给 astream_log，因为 astream_events 的实现是基于 astream_log 的。

Yields:

一个异步的StreamEvents流。

Raises:

NotImplementedError – 如果版本不是v1或v2。

Return type:

AsyncIterator[StandardStreamEvent | CustomStreamEvent]

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

默认实现使用线程池执行器并行运行invoke。

batch的默认实现对于IO绑定的runnables效果很好。

如果子类能够更高效地进行批处理，则应重写此方法；例如，如果底层的Runnable使用支持批处理模式的API。

Parameters:

inputs (列表[Input])
config (RunnableConfig | list[RunnableConfig] | None)
return_exceptions (bool)
kwargs (任意 | 无)

Return type:

列表[输出]

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

在输入列表上并行运行invoke，在它们完成时产生结果。

Parameters:

inputs (Sequence[Input])
config (RunnableConfig | Sequence[RunnableConfig] | None)
return_exceptions (bool)
kwargs (任意 | 无)

Return type:

Iterator[元组[int, Output | 异常]]

bind(**kwargs: Any) → Runnable[Input, Output]#

将参数绑定到Runnable，返回一个新的Runnable。

当链中的Runnable需要一个不在前一个Runnable输出中或用户输入中的参数时，这很有用。

Parameters:: kwargs (Any) – 绑定到Runnable的参数。
Returns:: 一个新的Runnable，参数已绑定。
Return type:: Runnable[Input, Output]

示例：

from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import StrOutputParser

llm = ChatOllama(model='llama2')

# 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'

将类似工具的对象绑定到此聊天模型。

假设模型与Vertex工具调用API兼容。

Parameters:

tools (Sequence[Tool | Tool | _ToolDictLike | BaseTool | Type[BaseModel] | FunctionDescription | Callable | FunctionDeclaration | Dict[str, Any]]) – 绑定到此聊天模型的工具定义列表。可以是pydantic模型、可调用对象或BaseTool。Pydantic 模型、可调用对象和BaseTools将自动转换为它们的模式字典表示。
**kwargs (Any) – 传递给Runnable构造函数的任何额外参数。
tool_config (_ToolConfigDict | None)
tool_choice (dict | List[str] | str | Literal['auto', 'none', 'any'] | ~typing.Literal[True] | bool | None)
**kwargs

Return type:

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

配置可以在运行时设置的Runnables的替代方案。

Parameters:

which (ConfigurableField) – 将用于选择替代项的ConfigurableField实例。
default_key (str) – 如果没有选择其他选项，则使用的默认键。默认为“default”。
prefix_keys (bool) – 是否在键前加上 ConfigurableField 的 id。默认为 False。
**kwargs (Runnable[Input, Output] | Callable[[], Runnable[Input, Output]]) – 一个字典，键为Runnable实例或返回Runnable实例的可调用对象。

Returns:

一个新的Runnable，配置了替代方案。

Return type:

RunnableSerializable

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

model = ChatAnthropic(
    model_name="claude-3-sonnet-20240229"
).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
)

configurable_fields(**kwargs: ConfigurableField | ConfigurableFieldSingleOption | ConfigurableFieldMultiOption) → RunnableSerializable#

在运行时配置特定的Runnable字段。

Parameters:: **kwargs (ConfigurableField | ConfigurableFieldSingleOption | ConfigurableFieldMultiOption) – 一个包含ConfigurableField实例的字典，用于配置。
Returns:: 一个新的Runnable，其字段已配置。
Return type:: RunnableSerializable

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
)

get_num_tokens(text: str) → int[source]#

获取文本中存在的标记数量。

Parameters:: 文本 (str)
Return type:: 整数

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

获取消息中的令牌数量。

用于检查输入是否适合模型的上下文窗口。

注意: get_num_tokens_from_messages 的基本实现忽略了工具模式。

Parameters:

messages (list[BaseMessage]) – 要标记化的消息输入。
工具 (序列 | 无) – 如果提供，则为字典、BaseModel、函数或BaseTools的序列，将被转换为工具模式。

Returns:

消息中令牌数量的总和。

Return type:

整数

get_token_ids(text: str) → list[int]#

返回文本中标记的有序ID。

Parameters:

文本 (字符串) – 要分词的字符串输入。

Returns:

与文本中的标记对应的ID列表，按它们在文本中出现的顺序排列: 在文本中。

Return type:

列表[int]

invoke(input: LanguageModelInput, config: RunnableConfig | None = None, *, stop: list[str] | None = None, **kwargs: Any) → BaseMessage#

将单个输入转换为输出。重写以实现。

Parameters:

input (LanguageModelInput) – Runnable 的输入。
config (可选[RunnableConfig]) – 调用Runnable时使用的配置。该配置支持标准键，如用于跟踪目的的‘tags’、‘metadata’，用于控制并行工作量的‘max_concurrency’，以及其他键。更多详情请参考RunnableConfig。
stop (可选[列表[字符串]])
kwargs (Any)

Returns:

Runnable 的输出。

Return type:

BaseMessage

stream(input: LanguageModelInput, config: RunnableConfig | None = None, *, stop: list[str] | None = None, **kwargs: Any) → Iterator[BaseMessageChunk]#

流的默认实现，调用invoke。如果子类支持流输出，则应重写此方法。

Parameters:

input (LanguageModelInput) – Runnable 的输入。
config (可选[RunnableConfig]) – 用于Runnable的配置。默认为None。
kwargs (Any) – 传递给Runnable的额外关键字参数。
stop (可选[列表[字符串]])

Yields:

Runnable 的输出。

Return type:

迭代器[BaseMessageChunk]

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

将异步生命周期监听器绑定到一个Runnable，返回一个新的Runnable。

on_start: 在Runnable开始运行之前异步调用。 on_end: 在Runnable完成运行之后异步调用。 on_error: 如果Runnable抛出错误，则异步调用。

Run对象包含有关运行的信息，包括其id、类型、输入、输出、错误、开始时间、结束时间以及添加到运行中的任何标签或元数据。

Parameters:

on_start (Optional[AsyncListener]) – 在Runnable开始运行之前异步调用。默认为None。
on_end (Optional[AsyncListener]) – 在Runnable运行结束后异步调用。默认为None。
on_error (可选[AsyncListener]) – 如果Runnable抛出错误，则异步调用。默认为None。

Returns:

一个新的Runnable，绑定了监听器。

Return type:

Runnable[Input, Output]

示例：

from langchain_core.runnables import RunnableLambda
import time

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 2024-05-16T14:20:29.637053+00:00
on start callback starts at 2024-05-16T14:20:29.637150+00:00
on start callback ends at 2024-05-16T14:20:32.638305+00:00
on start callback ends at 2024-05-16T14:20:32.638383+00:00
Runnable[3s]: starts at 2024-05-16T14:20:32.638849+00:00
Runnable[5s]: starts at 2024-05-16T14:20:32.638999+00:00
Runnable[3s]: ends at 2024-05-16T14:20:35.640016+00:00
on end callback starts at 2024-05-16T14:20:35.640534+00:00
Runnable[5s]: ends at 2024-05-16T14:20:37.640169+00:00
on end callback starts at 2024-05-16T14:20:37.640574+00:00
on end callback ends at 2024-05-16T14:20:37.640654+00:00
on end callback ends at 2024-05-16T14:20:39.641751+00:00

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

将配置绑定到一个可运行对象，返回一个新的可运行对象。

Parameters:

config (RunnableConfig | None) – 绑定到Runnable的配置。
kwargs (Any) – 传递给Runnable的额外关键字参数。

Returns:

一个新的Runnable，带有绑定的配置。

Return type:

Runnable[Input, Output]

with_fallbacks(fallbacks: Sequence[Runnable[Input, Output]], *, exceptions_to_handle: tuple[type[BaseException], ...] = (<class 'Exception'>,), exception_key: Optional[str] = None) → RunnableWithFallbacksT[Input, Output]#

为Runnable添加回退，返回一个新的Runnable。

新的Runnable将尝试原始的Runnable，然后在失败时依次尝试每个回退。

Parameters:

fallbacks (Sequence[Runnable[Input, Output]]) – 如果原始 Runnable 失败，将尝试的一系列 runnables。
exceptions_to_handle (tuple[type[BaseException], ...]) – 要处理的异常类型的元组。默认为 (Exception,)。
exception_key (Optional[str]) – 如果指定了字符串，则处理的异常将作为输入的一部分传递给后备函数，使用指定的键。如果为 None，异常将不会传递给后备函数。如果使用此参数，基础 Runnable 及其后备函数必须接受字典作为输入。默认为 None。

Returns:

一个新的Runnable，它将在失败时尝试原始的Runnable，然后依次尝试每个回退。

Return type:

RunnableWithFallbacksT[Input, Output]

示例

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:

fallbacks (Sequence[Runnable[Input, Output]]) – 如果原始 Runnable 失败，将尝试的一系列 runnables。
exceptions_to_handle (tuple[type[BaseException], ...]) – 要处理的异常类型的元组。
exception_key (Optional[str]) – 如果指定了字符串，则处理的异常将作为输入的一部分传递给后备函数，使用指定的键。如果为 None，异常将不会传递给后备函数。如果使用此参数，基础 Runnable 及其后备函数必须接受字典作为输入。

Returns:

一个新的Runnable，它将在失败时尝试原始的Runnable，然后依次尝试每个回退。

Return type:

RunnableWithFallbacksT[Input, Output]

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]#

将生命周期监听器绑定到一个Runnable，返回一个新的Runnable。

on_start: 在Runnable开始运行之前调用，带有Run对象。 on_end: 在Runnable完成运行之后调用，带有Run对象。 on_error: 如果Runnable抛出错误时调用，带有Run对象。

Run对象包含有关运行的信息，包括其id、类型、输入、输出、错误、开始时间、结束时间以及添加到运行中的任何标签或元数据。

Parameters:

on_start (可选[联合[可调用[[运行], 无], 可调用[[运行, RunnableConfig], 无]]]) – 在Runnable开始运行之前调用。默认为无。
on_end (可选[联合[可调用[[运行], 无], 可调用[[运行, RunnableConfig], 无]]]) – 在Runnable完成运行后调用。默认为无。
on_error (可选[联合[可调用[[运行], 无], 可调用[[运行, RunnableConfig], 无]]]) – 如果Runnable抛出错误时调用。默认为无。

Returns:

一个新的Runnable，绑定了监听器。

Return type:

Runnable[Input, Output]

示例：

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_retry(*, retry_if_exception_type: tuple[type[BaseException], ...] = (<class 'Exception'>,), wait_exponential_jitter: bool = True, stop_after_attempt: int = 3) → Runnable[Input, Output]#

创建一个新的Runnable，在异常时重试原始的Runnable。

Parameters:

retry_if_exception_type (tuple[type[BaseException], ...]) – 一个异常类型的元组，用于重试。默认值为 (Exception,)。
wait_exponential_jitter (bool) – 是否在重试之间的等待时间中添加抖动。默认为 True。
stop_after_attempt (int) – 在放弃之前尝试的最大次数。默认为3。

Returns:

一个新的Runnable，在异常时重试原始的Runnable。

Return type:

Runnable[Input, Output]

示例：

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)

Parameters:

retry_if_exception_type (tuple[type[BaseException], ...]) – 一个异常类型的元组，用于在发生这些异常时重试
wait_exponential_jitter (bool) – 是否在重试之间为等待时间添加抖动
stop_after_attempt (int) – 在放弃之前尝试的最大次数

Returns:

一个新的Runnable，在异常时重试原始的Runnable。

Return type:

Runnable[Input, Output]

模型包装器，返回格式化为匹配给定模式的输出。

在版本1.1.0中更改：在版本1.1.0中修正了返回类型。以前如果提供了字典模式，则输出形式为 [{"args": {}, "name": "schema_name"}]，其中输出是一个包含单个字典的列表，并且该字典的“args”对应于模式。从1.1.0开始，这个问题已经修复，现在直接返回模式（对应于旧的“args”键的值）。

Parameters:

schema (Dict | Type[BaseModel]) – 输出模式可以是一个字典或一个Pydantic类。如果是一个Pydantic类，那么模型输出将是该类的对象。如果是一个字典，那么模型输出将是一个字典。使用Pydantic类时，返回的属性将被验证，而使用字典时则不会。如果method是“function_calling”并且schema是一个字典，那么该字典必须符合OpenAI函数调用规范。
include_raw (bool) – 如果为False，则仅返回解析后的结构化输出。如果在模型输出解析过程中发生错误，将会抛出。如果为True，则返回原始模型响应（一个BaseMessage）和解析后的模型响应。如果在输出解析过程中发生错误，它将被捕获并返回。最终输出始终是一个包含“raw”、“parsed”和“parsing_error”键的字典。
method (Literal['json_mode'] | None) – 如果设置为‘json_schema’，它将使用受控生成来生成响应，而不是函数调用。不适用于具有引用或自引用Pydantic模型的模式。
kwargs (Any)

Returns:

一个接受任何ChatModel输入的Runnable。如果include_raw为True，则返回一个包含键的字典——raw: BaseMessage, parsed: Optional[_DictOrPydantic], parsing_error: Optional[BaseException]。如果include_raw为False，则只返回_DictOrPydantic，其中_DictOrPydantic取决于模式。如果模式是一个Pydantic类，则_DictOrPydantic是该Pydantic类。如果模式是一个字典，则_DictOrPydantic是一个字典。

Return type:

Example: Pydantic schema, exclude raw:

from pydantic import BaseModel
from langchain_google_vertexai import ChatVertexAI

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

llm = ChatVertexAI(model_name="gemini-pro", temperature=0)
structured_llm = llm.with_structured_output(AnswerWithJustification)

structured_llm.invoke("What weighs more a pound of bricks or a pound of feathers")
# -> AnswerWithJustification(
#     answer='They weigh the same.', justification='A pound is a pound.'
# )

Example: Pydantic schema, include raw:

from pydantic import BaseModel
from langchain_google_vertexai import ChatVertexAI

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

llm = ChatVertexAI(model_name="gemini-pro", temperature=0)
structured_llm = llm.with_structured_output(AnswerWithJustification, include_raw=True)

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

Example: Dict schema, exclude raw:

from pydantic import BaseModel
from langchain_core.utils.function_calling import convert_to_openai_function
from langchain_google_vertexai import ChatVertexAI

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

dict_schema = convert_to_openai_function(AnswerWithJustification)
llm = ChatVertexAI(model_name="gemini-pro", temperature=0)
structured_llm = llm.with_structured_output(dict_schema)

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

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

将输入和输出类型绑定到一个Runnable，返回一个新的Runnable。

Parameters:

input_type (type[Input] | None) – 要绑定到Runnable的输入类型。默认为None。
output_type (type[Output] | None) – 要绑定到Runnable的输出类型。默认为None。

Returns:

一个带有类型绑定的新Runnable。

Return type:

Runnable[Input, Output]

property async_prediction_client: PredictionServiceAsyncClient#: 返回 PredictionServiceClient。

property prediction_client: PredictionServiceClient#: 返回 PredictionServiceClient。

task_executor: ClassVar[Executor | None] = FieldInfo(annotation=NoneType, required=False, default=None, exclude=True)#