跳转到内容

AzureChatOpenAI

参考文档

本页包含 AzureChatOpenAI参考文档。关于使用 AzureChatOpenAI 的概念指南、教程和示例,请参阅文档

langchain_openai.chat_models.AzureChatOpenAI

Bases: BaseChatOpenAI

Azure OpenAI 聊天模型集成。

设置

前往 Azure OpenAI 快速入门指南,创建你的 Azure OpenAI 部署。

然后安装 langchain-openai 并设置环境变量 AZURE_OPENAI_API_KEYAZURE_OPENAI_ENDPOINT

pip install -U langchain-openai

export AZURE_OPENAI_API_KEY="your-api-key"
export AZURE_OPENAI_ENDPOINT="https://your-endpoint.openai.azure.com/"

关键初始化参数 — 完成参数:azure_deployment:要使用的 Azure OpenAI 部署名称。temperature:采样温度。max_tokens:要生成的最大令牌数。logprobs:是否返回 logprobs。

关键初始化参数 — 客户端参数:api_version:要使用的 Azure OpenAI REST API 版本(与底层模型的版本不同)。查看更多关于不同版本的信息。timeout:请求超时时间。max_retries:最大重试次数。organization:OpenAI 组织 ID。如果未传入,将从环境变量 OPENAI_ORG_ID 中读取。model:底层 OpenAI 模型的名称。用于跟踪和令牌计数。不影响完成结果。例如 'gpt-4''gpt-35-turbo' 等。model_version:底层 OpenAI 模型的版本。用于跟踪和令牌计数。不影响完成结果。例如 '0125''0125-preview' 等。

有关支持的初始化参数及其描述的完整列表,请参见参数部分。

实例化
from langchain_openai import AzureChatOpenAI

model = AzureChatOpenAI(
    azure_deployment="your-deployment",
    api_version="2024-05-01-preview",
    temperature=0,
    max_tokens=None,
    timeout=None,
    max_retries=2,
    # organization="...",
    # model="gpt-35-turbo",
    # model_version="0125",
    # other params...
)

注意

任何未明确支持的参数将在每次调用模型时直接传递给 openai.AzureOpenAI.chat.completions.create(...) API。

例如

from langchain_openai import AzureChatOpenAI
import openai

AzureChatOpenAI(..., logprobs=True).invoke(...)

# results in underlying API call of:

openai.AzureOpenAI(..).chat.completions.create(..., logprobs=True)

# which is also equivalent to:

AzureChatOpenAI(...).invoke(..., logprobs=True)
调用
messages = [
    (
        "system",
        "You are a helpful translator. Translate the user sentence to French.",
    ),
    ("human", "I love programming."),
]
model.invoke(messages)
AIMessage(
    content="J'adore programmer.",
    usage_metadata={
        "input_tokens": 28,
        "output_tokens": 6,
        "total_tokens": 34,
    },
    response_metadata={
        "token_usage": {
            "completion_tokens": 6,
            "prompt_tokens": 28,
            "total_tokens": 34,
        },
        "model_name": "gpt-4",
        "system_fingerprint": "fp_7ec89fabc6",
        "prompt_filter_results": [
            {
                "prompt_index": 0,
                "content_filter_results": {
                    "hate": {"filtered": False, "severity": "safe"},
                    "self_harm": {"filtered": False, "severity": "safe"},
                    "sexual": {"filtered": False, "severity": "safe"},
                    "violence": {"filtered": False, "severity": "safe"},
                },
            }
        ],
        "finish_reason": "stop",
        "logprobs": None,
        "content_filter_results": {
            "hate": {"filtered": False, "severity": "safe"},
            "self_harm": {"filtered": False, "severity": "safe"},
            "sexual": {"filtered": False, "severity": "safe"},
            "violence": {"filtered": False, "severity": "safe"},
        },
    },
    id="run-6d7a5282-0de0-4f27-9cc0-82a9db9a3ce9-0",
)
for chunk in model.stream(messages):
    print(chunk.text, end="")
AIMessageChunk(content="", id="run-a6f294d3-0700-4f6a-abc2-c6ef1178c37f")
AIMessageChunk(content="J", id="run-a6f294d3-0700-4f6a-abc2-c6ef1178c37f")
AIMessageChunk(content="'", id="run-a6f294d3-0700-4f6a-abc2-c6ef1178c37f")
AIMessageChunk(content="ad", id="run-a6f294d3-0700-4f6a-abc2-c6ef1178c37f")
AIMessageChunk(content="ore", id="run-a6f294d3-0700-4f6a-abc2-c6ef1178c37f")
AIMessageChunk(content=" la", id="run-a6f294d3-0700-4f6a-abc2-c6ef1178c37f")
AIMessageChunk(
    content=" programm", id="run-a6f294d3-0700-4f6a-abc2-c6ef1178c37f"
)
AIMessageChunk(content="ation", id="run-a6f294d3-0700-4f6a-abc2-c6ef1178c37f")
AIMessageChunk(content=".", id="run-a6f294d3-0700-4f6a-abc2-c6ef1178c37f")
AIMessageChunk(
    content="",
    response_metadata={
        "finish_reason": "stop",
        "model_name": "gpt-4",
        "system_fingerprint": "fp_811936bd4f",
    },
    id="run-a6f294d3-0700-4f6a-abc2-c6ef1178c37f",
)
stream = model.stream(messages)
full = next(stream)
for chunk in stream:
    full += chunk
full
AIMessageChunk(
    content="J'adore la programmation.",
    response_metadata={
        "finish_reason": "stop",
        "model_name": "gpt-4",
        "system_fingerprint": "fp_811936bd4f",
    },
    id="run-ba60e41c-9258-44b8-8f3a-2f10599643b3",
)
异步
await model.ainvoke(messages)

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

# batch:
# await model.abatch([messages])
工具调用
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"
    )


model_with_tools = model.bind_tools([GetWeather, GetPopulation])
ai_msg = model_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": "call_6XswGD5Pqk8Tt5atYr7tfenU",
    },
    {
        "name": "GetWeather",
        "args": {"location": "New York, NY"},
        "id": "call_ZVL15vA8Y7kXqOy3dtmQgeCi",
    },
    {
        "name": "GetPopulation",
        "args": {"location": "Los Angeles, CA"},
        "id": "call_49CFW8zqC9W7mh7hbMLSIrXw",
    },
    {
        "name": "GetPopulation",
        "args": {"location": "New York, NY"},
        "id": "call_6ghfKxV264jEfe1mRIkS3PE7",
    },
]
结构化输出
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: int | None = Field(
        description="How funny the joke is, from 1 to 10"
    )


structured_model = model.with_structured_output(Joke)
structured_model.invoke("Tell me a joke about cats")
Joke(
    setup="Why was the cat sitting on the computer?",
    punchline="To keep an eye on the mouse!",
    rating=None,
)

更多信息请参见 AzureChatOpenAI.with_structured_output()

JSON 模式
json_model = model.bind(response_format={"type": "json_object"})
ai_msg = json_model.invoke(
    "Return a JSON object with key 'random_ints' and a value of 10 random ints in [0-99]"
)
ai_msg.content
'\\n{\\n  "random_ints": [23, 87, 45, 12, 78, 34, 56, 90, 11, 67]\\n}'
图像输入
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 = model.invoke([message])
ai_msg.content
"The weather in the image appears to be quite pleasant. The sky is mostly clear"
令牌使用情况
ai_msg = model.invoke(messages)
ai_msg.usage_metadata
{"input_tokens": 28, "output_tokens": 5, "total_tokens": 33}

Logprobs

logprobs_model = model.bind(logprobs=True)
ai_msg = logprobs_model.invoke(messages)
ai_msg.response_metadata["logprobs"]

```python
{
    "content": [
        {
            "token": "J",
            "bytes": [74],
            "logprob": -4.9617593e-06,
            "top_logprobs": [],
        },
        {
            "token": "'adore",
            "bytes": [39, 97, 100, 111, 114, 101],
            "logprob": -0.25202933,
            "top_logprobs": [],
        },
        {
            "token": " la",
            "bytes": [32, 108, 97],
            "logprob": -0.20141791,
            "top_logprobs": [],
        },
        {
            "token": " programmation",
            "bytes": [
                32,
                112,
                114,
                111,
                103,
                114,
                97,
                109,
                109,
                97,
                116,
                105,
                111,
                110,
            ],
            "logprob": -1.9361265e-07,
            "top_logprobs": [],
        },
        {
            "token": ".",
            "bytes": [46],
            "logprob": -1.2233183e-05,
            "top_logprobs": [],
        },
    ]
}
```

响应元数据

ai_msg = model.invoke(messages)
ai_msg.response_metadata

```python
{
    "token_usage": {
        "completion_tokens": 6,
        "prompt_tokens": 28,
        "total_tokens": 34,
    },
    "model_name": "gpt-35-turbo",
    "system_fingerprint": None,
    "prompt_filter_results": [
        {
            "prompt_index": 0,
            "content_filter_results": {
                "hate": {"filtered": False, "severity": "safe"},
                "self_harm": {"filtered": False, "severity": "safe"},
                "sexual": {"filtered": False, "severity": "safe"},
                "violence": {"filtered": False, "severity": "safe"},
            },
        }
    ],
    "finish_reason": "stop",
    "logprobs": None,
    "content_filter_results": {
        "hate": {"filtered": False, "severity": "safe"},
        "self_harm": {"filtered": False, "severity": "safe"},
        "sexual": {"filtered": False, "severity": "safe"},
        "violence": {"filtered": False, "severity": "safe"},
    },
}
```
方法 描述
get_name

获取 Runnable 的名称。

get_input_schema

获取可用于验证 Runnable 输入的 Pydantic 模型。

get_input_jsonschema

获取表示 Runnable 输入的 JSON 模式。

get_output_schema

获取可用于验证 Runnable 输出的 Pydantic 模型。

get_output_jsonschema

获取表示 Runnable 输出的 JSON 模式。

config_schema

Runnable 接受的配置类型,指定为 Pydantic 模型。

get_config_jsonschema

获取表示 Runnable 配置的 JSON 模式。

get_graph

返回此 Runnable 的图形表示。

get_prompts

返回此 Runnable 使用的提示列表。

__or__

Runnable "or" 运算符。

__ror__

Runnable "reverse-or" 运算符。

pipe

管道连接 Runnable 对象。

pick

从此 Runnable 的输出 dict 中选择键。

assign

向此 Runnabledict 输出分配新字段。

invoke

将单个输入转换为输出。

ainvoke

将单个输入转换为输出。

batch

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

batch_as_completed

在输入列表上并行运行 invoke

abatch

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

abatch_as_completed

在输入列表上并行运行 ainvoke

stream

stream 的默认实现,它调用 invoke

astream

astream 的默认实现,它调用 ainvoke

astream_log

流式传输 Runnable 的所有输出,如回调系统所报告。

astream_events

生成事件流。

transform

将输入转换为输出。

atransform

将输入转换为输出。

bind

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

with_config

将配置绑定到 Runnable,返回一个新的 Runnable

with_listeners

将生命周期侦听器绑定到 Runnable,返回一个新的 Runnable

with_alisteners

将异步生命周期侦听器绑定到 Runnable

with_types

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

with_retry

创建一个新的 Runnable,它在发生异常时重试原始的 Runnable

map

返回一个新的 Runnable,它将输入列表映射到输出列表。

with_fallbacks

Runnable 添加回退机制,返回一个新的 Runnable

as_tool

Runnable 创建一个 BaseTool

__init__
lc_id

为此类返回一个用于序列化目的的唯一标识符。

to_json

Runnable 序列化为 JSON。

to_json_not_implemented

序列化一个“未实现”的对象。

configurable_fields

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

configurable_alternatives

为可在运行时设置的 Runnable 对象配置备选项。

set_verbose

如果 verbose 是 None,则设置它。

generate_prompt

将一系列提示传递给模型并返回模型生成的内容。

agenerate_prompt

异步地将一系列提示传递并返回模型生成的内容。

get_token_ids

使用 tiktoken 包获取文本中存在的令牌。

get_num_tokens

获取文本中存在的 token 数量。

get_num_tokens_from_messages

使用 tiktoken 包计算 gpt-3.5-turbogpt-4 的令牌数。

generate

将一系列提示传递给模型并返回模型生成的内容。

agenerate

异步地将一系列提示传递给模型并返回生成的内容。

dict

返回 LLM 的字典。

bind_tools

将类工具(tool-like)对象绑定到此聊天模型。

build_extra

根据传入的额外参数构建额外的关键字参数(kwargs)。

validate_temperature

验证不同模型的温度参数。

get_lc_namespace

获取 LangChain 对象的命名空间。

is_lc_serializable

检查该类在 langchain 中是否可序列化。

validate_environment

验证 API 密钥和 Python 包是否存在于环境中。

with_structured_output

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

name class-attribute instance-attribute

name: str | None = None

Runnable 的名称。用于调试和追踪。

InputType property

InputType: TypeAlias

获取此 Runnable 的输入类型。

OutputType property

OutputType: Any

获取此 Runnable 的输出类型。

input_schema property

input_schema: type[BaseModel]

Runnable 接受的输入类型,指定为 Pydantic 模型。

output_schema property

output_schema: type[BaseModel]

输出模式。

Runnable 产生的输出类型,指定为 Pydantic 模型。

config_specs property

config_specs: list[ConfigurableFieldSpec]

列出此 Runnable 的可配置字段。

cache class-attribute instance-attribute

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

是否缓存响应。

  • 如果为 True,将使用全局缓存。
  • 如果为 False,将不使用缓存
  • 如果为 None,如果设置了全局缓存,则使用全局缓存,否则不使用缓存。
  • 如果是 BaseCache 的实例,将使用提供的缓存。

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

verbose class-attribute instance-attribute

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

是否打印响应文本。

callbacks class-attribute instance-attribute

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

添加到运行跟踪中的回调。

tags class-attribute instance-attribute

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

添加到运行跟踪中的标签。

metadata class-attribute instance-attribute

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

添加到运行跟踪中的元数据。

custom_get_token_ids class-attribute instance-attribute

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

用于计算 token 的可选编码器。

rate_limiter class-attribute instance-attribute

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

一个可选的速率限制器,用于限制请求数量。

disable_streaming class-attribute instance-attribute

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

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

如果绕过流式传输,则 stream/astream/astream_events 将转而调用 invoke/ainvoke

  • 如果为 True,将始终绕过流式传输情况。
  • 如果为 'tool_calling',仅当模型被调用时带有 tools 关键字参数时,才会绕过流式传输情况。换句话说,仅当提供 tools 参数时,LangChain 才会自动切换到非流式行为 (invoke)。这提供了两全其美的方案。
  • 如果为 False(默认值),则在可用时始终使用流式传输情况。

此标志的主要原因是,代码可能是使用 stream 编写的,而用户可能希望将给定模型换成另一个实现不完全支持流式传输的模型。

output_version class-attribute instance-attribute

output_version: str | None = Field(
    default_factory=from_env("LC_OUTPUT_VERSION", default=None)
)

要使用的 AIMessage 输出格式的版本。

该字段用于以向后兼容的方式推出聊天模型 AIMessage 响应的新输出格式。

支持的值

  • 'v0'langchain-openai 0.3.x 版本的 AIMessage 格式。
  • 'responses/v1':将 Responses API 输出项格式化为 AIMessage 内容块(仅限 Responses API)。
  • 'v1':LangChain 跨提供商标​​准的 v1 版本。

行为在 1.0.0 中已更改

默认更新为 "responses/v1"

profile property

profile: ModelProfile

返回模型的性能分析信息。

此属性将依赖于 langchain-model-profiles 包来检索聊天模型的能力,例如上下文窗口大小和支持的功能。

引发 描述
ImportError

如果未安装 langchain-model-profiles

返回 描述
ModelProfile

一个 ModelProfile 对象,包含模型的性能分析信息。

temperature class-attribute instance-attribute

temperature: float | None = None

要使用的采样温度。

model_kwargs class-attribute instance-attribute

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

保存任何未明确指定的、对 create 调用有效的模型参数。

openai_api_base class-attribute instance-attribute

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

API 请求的基础 URL 路径,如果不使用代理或服务模拟器,请留空。

openai_organization class-attribute instance-attribute

openai_organization: str | None = Field(default=None, alias='organization')

如果未提供,则从环境变量 `OPENAI_ORG_ID` 自动推断。

request_timeout class-attribute instance-attribute

request_timeout: float | tuple[float, float] | Any | None = Field(
    default=None, alias="timeout"
)

请求 OpenAI 完成 API 的超时时间。可以是浮点数、httpx.TimeoutNone

stream_usage class-attribute instance-attribute

stream_usage: bool | None = None

是否在流式输出中包含使用情况元数据。如果启用,将在流式传输期间生成一个额外的消息块,其中包含使用情况元数据。

除非设置了 openai_api_base 或模型使用自定义客户端初始化,否则此参数是启用的,因为许多聊天完成 API 不支持流式传输令牌使用情况。

于 0.3.9 版本中添加

行为在 0.3.35 中已更改

对默认基础 URL 和客户端启用。

max_retries class-attribute instance-attribute

max_retries: int | None = None

生成时允许的最大重试次数。

presence_penalty class-attribute instance-attribute

presence_penalty: float | None = None

对重复的令牌进行惩罚。

frequency_penalty class-attribute instance-attribute

frequency_penalty: float | None = None

根据频率对重复的令牌进行惩罚。

seed class-attribute instance-attribute

seed: int | None = None

生成的种子

logprobs class-attribute instance-attribute

logprobs: bool | None = None

是否返回 logprobs。

top_logprobs class-attribute instance-attribute

top_logprobs: int | None = None

在每个令牌位置返回的最可能令牌的数量,每个令牌都有一个相关的对数概率。如果使用此参数,`logprobs` 必须设置为 true。

logit_bias class-attribute instance-attribute

logit_bias: dict[int, int] | None = None

修改指定令牌出现在补全中的可能性。

streaming class-attribute instance-attribute

streaming: bool = False

是否以流式方式返回结果。

n class-attribute instance-attribute

n: int | None = None

为每个提示生成的聊天补全数量。

top_p class-attribute instance-attribute

top_p: float | None = None

每一步要考虑的令牌的总概率质量。

reasoning_effort class-attribute instance-attribute

reasoning_effort: str | None = None

限制推理模型的推理工作量。用于 Chat Completions API。

仅限推理模型。

当前支持的值为 'minimal''low''medium''high'。减少推理工作量可以加快响应速度,并减少响应中用于推理的令牌数量。

reasoning class-attribute instance-attribute

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

推理模型的推理参数。用于 Responses API。

reasoning={
    "effort": "medium",  # Can be "low", "medium", or "high"
    "summary": "auto",  # Can be "auto", "concise", or "detailed"
}

在 0.3.24 版本中添加

verbosity class-attribute instance-attribute

verbosity: str | None = None

控制推理模型响应的详细程度。用于 Responses API。

当前支持的值为 'low''medium''high'

在 0.3.28 版本中添加

tiktoken_model_name class-attribute instance-attribute

tiktoken_model_name: str | None = None

使用此类时传递给 tiktoken 的模型名称。Tiktoken 用于计算文档中的令牌数量,以将其限制在特定限制以下。默认情况下,当设置为 None 时,这将与嵌入模型名称相同。但是,在某些情况下,您可能希望将此 Embedding 类与 tiktoken 不支持的模型名称一起使用。这可能包括使用 Azure 嵌入时,或使用众多提供类似 OpenAI API 但具有不同模型的模型提供商之一时。在这些情况下,为了避免在调用 tiktoken 时出错,您可以在此处指定要使用的模型名称。

http_client class-attribute instance-attribute

http_client: Any | None = Field(default=None, exclude=True)

可选的 `httpx.Client`。仅用于同步调用。如果您希望为异步调用使用自定义客户端,则还必须指定 `http_async_client`。

http_async_client class-attribute instance-attribute

http_async_client: Any | None = Field(default=None, exclude=True)

可选的 `httpx.AsyncClient`。仅用于异步调用。如果您希望为同步调用使用自定义客户端,则还必须指定 `http_client`。

stop class-attribute instance-attribute

stop: list[str] | str | None = Field(default=None, alias='stop_sequences')

默认的停止序列。

extra_body class-attribute instance-attribute

extra_body: Mapping[str, Any] | None = None

在向与 OpenAI 兼容的 API(如 vLLM、LM Studio 或其他提供商)发出请求时,要在请求参数中包含的可选附加 JSON 属性。

这是传递特定于您的 OpenAI 兼容 API 提供商但不属于标准 OpenAI API 的自定义参数的推荐方法。

示例

  • LM Studio TTL 参数:extra_body={"ttl": 300}
  • vLLM 自定义参数:extra_body={"use_beam_search": True}
  • 任何其他特定于提供商的参数

警告

不要使用 model_kwargs 来传递不属于标准 OpenAI API 的自定义参数,因为这会在进行 API 调用时导致错误。请改用 extra_body

include_response_headers class-attribute instance-attribute

include_response_headers: bool = False

是否在输出消息的 response_metadata 中包含响应头。

include class-attribute instance-attribute

include: list[str] | None = None

从 Responses API 的生成结果中要包含的额外字段。

支持的值

  • 'file_search_call.results'
  • 'message.input_image.image_url'
  • 'computer_call_output.output.image_url'
  • 'reasoning.encrypted_content'
  • 'code_interpreter_call.outputs'

在 0.3.24 版本中添加

service_tier class-attribute instance-attribute

service_tier: str | None = None

请求的延迟层级。选项有 'auto''default''flex'。与 OpenAI 的 scale tier 服务用户相关。

store class-attribute instance-attribute

store: bool | None = None

如果为 True,OpenAI 可能会存储响应数据以备将来使用。

对于 Responses API,默认为 True;对于 Chat Completions API,默认为 False

在 0.3.24 版本中添加

truncation class-attribute instance-attribute

truncation: str | None = None

截断策略 (Responses API)。可以是 'auto''disabled' (默认)。如果为 'auto',模型可能会从消息序列的中间丢弃输入项以适应上下文窗口。

在 0.3.24 版本中添加

use_previous_response_id class-attribute instance-attribute

use_previous_response_id: bool = False

如果为 True,则始终使用最近一次响应的 ID 传递 previous_response_id。仅限 Responses API。

直到最近一次响应的输入消息将从请求负载中被丢弃。

例如,以下两种情况是等效的

model = ChatOpenAI(
    model="...",
    use_previous_response_id=True,
)
model.invoke(
    [
        HumanMessage("Hello"),
        AIMessage("Hi there!", response_metadata={"id": "resp_123"}),
        HumanMessage("How are you?"),
    ]
)
model = ChatOpenAI(model="...", use_responses_api=True)
model.invoke([HumanMessage("How are you?")], previous_response_id="resp_123")

在 0.3.26 版本中添加

use_responses_api class-attribute instance-attribute

use_responses_api: bool | None = None

是否使用 Responses API 而不是 Chat API。

如果未指定,则将根据调用参数进行推断。

于 0.3.9 版本中添加

azure_endpoint class-attribute instance-attribute

azure_endpoint: str | None = Field(
    default_factory=from_env("AZURE_OPENAI_ENDPOINT", default=None)
)

您的 Azure 端点,包括资源。

如果未提供,则自动从环境变量 AZURE_OPENAI_ENDPOINT 推断。

示例:https://example-resource.azure.openai.com/

deployment_name class-attribute instance-attribute

deployment_name: str | None = Field(default=None, alias='azure_deployment')

一个模型部署。

如果给定,则将基础客户端 URL 设置为包含 /deployments/{azure_deployment}

注意

这意味着您将无法使用非部署端点。

openai_api_version class-attribute instance-attribute

openai_api_version: str | None = Field(
    alias="api_version", default_factory=from_env("OPENAI_API_VERSION", default=None)
)

如果未提供,则从环境变量 `OPENAI_API_VERSION` 自动推断。

openai_api_key class-attribute instance-attribute

openai_api_key: SecretStr | None = Field(
    alias="api_key",
    default_factory=secret_from_env(
        ["AZURE_OPENAI_API_KEY", "OPENAI_API_KEY"], default=None
    ),
)

如果未提供,则自动从环境变量 AZURE_OPENAI_API_KEY 推断。

azure_ad_token class-attribute instance-attribute

azure_ad_token: SecretStr | None = Field(
    default_factory=secret_from_env("AZURE_OPENAI_AD_TOKEN", default=None)
)

您的 Azure Active Directory 令牌。

如果未提供,则自动从环境变量 AZURE_OPENAI_AD_TOKEN 推断。

更多信息,请参阅此页面

azure_ad_token_provider class-attribute instance-attribute

azure_ad_token_provider: Callable[[], str] | None = None

一个返回 Azure Active Directory 令牌的函数。

将在每个同步请求中调用。对于异步请求,如果未提供 `azure_ad_async_token_provider`,则会调用此函数。

azure_ad_async_token_provider class-attribute instance-attribute

azure_ad_async_token_provider: Callable[[], Awaitable[str]] | None = None

一个返回 Azure Active Directory 令牌的函数。

将在每个异步请求中调用。

model_version class-attribute instance-attribute

model_version: str = ''

模型版本(例如 '0125' 代表 'gpt-3.5-0125')。

Azure OpenAI 默认不返回模型版本,因此如果您想在下游使用此信息(例如计算成本时),必须手动指定。

当您指定版本时,它将被附加到响应中的模型名称后。设置正确的版本将帮助您正确计算成本。模型版本不会被验证,因此请确保您正确设置以获得正确的成本。

openai_api_type class-attribute instance-attribute

openai_api_type: str | None = Field(
    default_factory=from_env("OPENAI_API_TYPE", default="azure")
)

旧版,用于支持 openai<1.0.0

validate_base_url class-attribute instance-attribute

validate_base_url: bool = True

如果传入了旧版参数 openai_api_base,则尝试推断它是 base_url 还是 azure_endpoint,并相应地更新客户端参数。

model_name class-attribute instance-attribute

model_name: str | None = Field(default=None, alias='model')

已部署的 OpenAI 模型的名称,例如 'gpt-4o''gpt-35-turbo' 等。

与由 Azure 用户设置的 Azure 部署名称不同。用于跟踪和令牌计数。

警告

不影响完成结果。

disabled_params class-attribute instance-attribute

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

对于给定模型应禁用的 OpenAI 客户端或 chat.completions 端点的参数。

应指定为 {"param": None | ['val1', 'val2']},其中键是参数,值可以是 None(表示该参数永远不应使用),也可以是该参数的禁用值列表。

例如,旧版模型可能根本不支持 'parallel_tool_calls' 参数,在这种情况下可以传入 disabled_params={"parallel_tool_calls: None}

如果一个参数被禁用,那么它在任何方法中都不会被默认使用,例如在 langchain_openai.chat_models.azure.AzureChatOpenAI.with_structured_output 中。但这并不妨碍用户在调用时直接传入该参数。

默认情况下,除非指定了 model_name="gpt-4o",否则 'parallel_tools_calls' 将被禁用。

max_tokens class-attribute instance-attribute

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

生成的最大令牌数。

lc_secrets property

lc_secrets: dict[str, str]

获取机密环境变量的映射。

lc_attributes property

lc_attributes: dict[str, Any]

获取与追踪相关的属性。

get_name

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

获取 Runnable 的名称。

参数 描述
后缀

一个可选的后缀,附加到名称上。

类型: str | None 默认值: None

name

一个可选的名称,用于代替 Runnable 的名称。

类型: str | None 默认值: None

返回 描述
str

Runnable 的名称。

get_input_schema

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

获取可用于验证 Runnable 输入的 Pydantic 模型。

利用 configurable_fieldsconfigurable_alternatives 方法的 Runnable 对象将具有一个动态输入模式,该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输入模式。

参数 描述
配置

生成模式时使用的配置。

类型: RunnableConfig | None 默认值: None

返回 描述
type[BaseModel]

一个可用于验证输入的 Pydantic 模型。

get_input_jsonschema

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

获取表示 Runnable 输入的 JSON 模式。

参数 描述
配置

生成模式时使用的配置。

类型: RunnableConfig | None 默认值: None

返回 描述
dict[str, Any]

表示 Runnable 输入的 JSON 模式。

示例
from langchain_core.runnables import RunnableLambda


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


runnable = RunnableLambda(add_one)

print(runnable.get_input_jsonschema())

在 0.3.0 版本中新增。

get_output_schema

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

获取可用于验证 Runnable 输出的 Pydantic 模型。

利用 configurable_fieldsconfigurable_alternatives 方法的 Runnable 对象将具有一个动态输出模式,该模式取决于调用 Runnable 时使用的配置。

此方法允许获取特定配置的输出模式。

参数 描述
配置

生成模式时使用的配置。

类型: RunnableConfig | None 默认值: None

返回 描述
type[BaseModel]

一个可用于验证输出的 Pydantic 模型。

get_output_jsonschema

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

获取表示 Runnable 输出的 JSON 模式。

参数 描述
配置

生成模式时使用的配置。

类型: RunnableConfig | None 默认值: None

返回 描述
dict[str, Any]

表示 Runnable 输出的 JSON 模式。

示例
from langchain_core.runnables import RunnableLambda


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


runnable = RunnableLambda(add_one)

print(runnable.get_output_jsonschema())

在 0.3.0 版本中新增。

config_schema

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

Runnable 接受的配置类型,指定为 Pydantic 模型。

要将字段标记为可配置,请参阅 configurable_fieldsconfigurable_alternatives 方法。

参数 描述
包含

要包含在配置模式中的字段列表。

类型: Sequence[str] | None 默认值: None

返回 描述
type[BaseModel]

一个可用于验证配置的 Pydantic 模型。

get_config_jsonschema

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

获取表示 Runnable 配置的 JSON 模式。

参数 描述
包含

要包含在配置模式中的字段列表。

类型: Sequence[str] | None 默认值: None

返回 描述
dict[str, Any]

表示 Runnable 配置的 JSON 模式。

在 0.3.0 版本中新增。

get_graph

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

返回此 Runnable 的图形表示。

get_prompts

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

返回此 Runnable 使用的提示列表。

__or__

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

Runnable "or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence

参数 描述
other

另一个 Runnable 或类 Runnable 对象。

类型: 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

__ror__

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

Runnable "reverse-or" 运算符。

将此 Runnable 与另一个对象组合以创建 RunnableSequence

参数 描述
other

另一个 Runnable 或类 Runnable 对象。

类型: 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

pipe

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

管道连接 Runnable 对象。

将此 Runnable 与类 Runnable 对象组合以构成一个 RunnableSequence

等同于 RunnableSequence(self, *others)self | others[0] | ...

示例
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]
参数 描述
*其他

其他要组合的 Runnable 或类 Runnable 对象

类型: Runnable[Any, Other] | Callable[[Any], Other] 默认值: ()

name

生成的 RunnableSequence 的一个可选名称。

类型: str | None 默认值: None

返回 描述
RunnableSerializable[Input, Other]

一个新的 Runnable

pick

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

从此 Runnable 的输出 dict 中选择键。

选择单个键

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]

选择键列表

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]"}
参数 描述
keys

从输出字典中选择的一个键或键列表。

类型: str | list[str]

返回 描述
RunnableSerializable[Any, Any]

一个新的 Runnable

assign

向此 Runnabledict 输出分配新字段。

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}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])

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

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

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

一个键到 Runnable 或类 Runnable 对象的映射,这些对象将使用此 Runnable 的整个输出字典来调用。

类型: 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]

一个新的 Runnable

invoke

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

将单个输入转换为输出。

参数 描述
输入

Runnable 的输入。

类型: Input

配置

调用 Runnable 时使用的配置。该配置支持标准键,如用于追踪目的的 'tags''metadata',用于控制并行工作量的 'max_concurrency',以及其他键。请参阅 RunnableConfig 以获取更多详细信息。

类型: RunnableConfig | None 默认值: None

返回 描述
输出

Runnable 的输出。

ainvoke async

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

将单个输入转换为输出。

参数 描述
输入

Runnable 的输入。

类型: Input

配置

调用 Runnable 时使用的配置。该配置支持标准键,如用于追踪目的的 'tags''metadata',用于控制并行工作量的 'max_concurrency',以及其他键。请参阅 RunnableConfig 以获取更多详细信息。

类型: RunnableConfig | None 默认值: None

返回 描述
输出

Runnable 的输出。

batch

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

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

批处理的默认实现对于 IO 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理,则必须重写此方法;例如,如果底层的 Runnable 使用支持批处理模式的 API。

参数 描述
inputs

Runnable 的输入列表。

类型: list[Input]

配置

调用 Runnable 时使用的配置。该配置支持标准键,如用于追踪目的的 'tags''metadata',用于控制并行工作量的 'max_concurrency',以及其他键。请参阅 RunnableConfig 以获取更多详细信息。

类型: RunnableConfig | list[RunnableConfig] | None 默认值: None

返回异常

是否返回异常而不是引发它们。

类型: bool 默认值: False

**kwargs

要传递给 Runnable 的其他关键字参数。

类型: Any | None 默认值: {}

返回 描述
list[Output]

来自 Runnable 的输出列表。

batch_as_completed

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

在输入列表上并行运行 invoke

在结果完成时生成它们。

参数 描述
inputs

Runnable 的输入列表。

类型: Sequence[Input]

配置

调用 Runnable 时使用的配置。该配置支持标准键,如用于追踪目的的 'tags''metadata',用于控制并行工作量的 'max_concurrency',以及其他键。请参阅 RunnableConfig 以获取更多详细信息。

类型: RunnableConfig | Sequence[RunnableConfig] | None 默认值: None

返回异常

是否返回异常而不是引发它们。

类型: bool 默认值: False

**kwargs

要传递给 Runnable 的其他关键字参数。

类型: Any | None 默认值: {}

YIELDS 描述
tuple[int, Output | Exception]

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

abatch 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 密集型的 runnable 效果很好。

如果子类能够更有效地进行批处理,则必须重写此方法;例如,如果底层的 Runnable 使用支持批处理模式的 API。

参数 描述
inputs

Runnable 的输入列表。

类型: list[Input]

配置

调用 Runnable 时使用的配置。该配置支持标准键,如用于追踪目的的 'tags''metadata',用于控制并行工作量的 'max_concurrency',以及其他键。请参阅 RunnableConfig 以获取更多详细信息。

类型: RunnableConfig | list[RunnableConfig] | None 默认值: None

返回异常

是否返回异常而不是引发它们。

类型: bool 默认值: False

**kwargs

要传递给 Runnable 的其他关键字参数。

类型: Any | None 默认值: {}

返回 描述
list[Output]

来自 Runnable 的输出列表。

abatch_as_completed async

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

在输入列表上并行运行 ainvoke

在结果完成时生成它们。

参数 描述
inputs

Runnable 的输入列表。

类型: Sequence[Input]

配置

调用 Runnable 时使用的配置。该配置支持标准键,如用于追踪目的的 'tags''metadata',用于控制并行工作量的 'max_concurrency',以及其他键。请参阅 RunnableConfig 以获取更多详细信息。

类型: RunnableConfig | Sequence[RunnableConfig] | None 默认值: None

返回异常

是否返回异常而不是引发它们。

类型: bool 默认值: False

**kwargs

要传递给 Runnable 的其他关键字参数。

类型: Any | None 默认值: {}

YIELDS 描述
AsyncIterator[tuple[int, Output | Exception]]

一个由输入索引和 Runnable 输出组成的元组。

stream

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

stream 的默认实现,它调用 invoke

如果子类支持流式输出,则必须重写此方法。

参数 描述
输入

Runnable 的输入。

类型: Input

配置

用于 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

**kwargs

要传递给 Runnable 的其他关键字参数。

类型: Any | None 默认值: {}

YIELDS 描述
输出

Runnable 的输出。

astream async

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

astream 的默认实现,它调用 ainvoke

如果子类支持流式输出,则必须重写此方法。

参数 描述
输入

Runnable 的输入。

类型: Input

配置

用于 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

**kwargs

要传递给 Runnable 的其他关键字参数。

类型: Any | None 默认值: {}

YIELDS 描述
AsyncIterator[Output]

Runnable 的输出。

astream_log async

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

流式传输 Runnable 的所有输出,如回调系统所报告。

这包括 LLM、检索器、工具等的所有内部运行。

输出以 Log 对象的形式流式传输,其中包括一个 Jsonpatch 操作列表,描述了运行状态在每一步中如何变化,以及运行的最终状态。

可以按顺序应用 Jsonpatch 操作来构造状态。

参数 描述
输入

Runnable 的输入。

类型: Any

配置

用于 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

差异

是生成每一步之间的差异还是当前状态。

类型: bool 默认值: True

带流式输出列表

是否生成 streamed_output 列表。

类型: bool 默认值: True

包含名称

仅包含具有这些名称的日志。

类型: Sequence[str] | None 默认值: None

包含类型

仅包含具有这些类型的日志。

类型: Sequence[str] | None 默认值: None

包含标签

仅包含具有这些标签的日志。

类型: Sequence[str] | None 默认值: None

排除名称

排除具有这些名称的日志。

类型: Sequence[str] | None 默认值: None

排除类型

排除具有这些类型的日志。

类型: Sequence[str] | None 默认值: None

排除标签

排除具有这些标签的日志。

类型: Sequence[str] | None 默认值: None

**kwargs

要传递给 Runnable 的其他关键字参数。

类型: Any 默认值: {}

YIELDS 描述
AsyncIterator[RunLogPatch] | AsyncIterator[RunLog]

一个 RunLogPatchRunLog 对象。

astream_events async

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

生成事件流。

用于创建一个 StreamEvent 的迭代器,提供有关 Runnable 进度的实时信息,包括来自中间结果的 StreamEvent

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

  • event:事件名称的格式为:on_[runnable_type]_(start|stream|end)
  • name:生成事件的 Runnable 的名称。
  • run_id:与发出事件的 Runnable 的给定执行相关联的随机生成的 ID。作为父 Runnable 执行的一部分被调用的子 Runnable 被分配其自己的唯一 ID。
  • parent_ids:生成事件的父可运行对象的 ID。根 Runnable 将有一个空列表。父 ID 的顺序是从根到直接父级。仅适用于 API 的 v2 版本。API 的 v1 版本将返回一个空列表。
  • tags:生成事件的 Runnable 的标签。
  • metadata:生成事件的 Runnable 的元数据。
  • data:与事件关联的数据。此字段的内容取决于事件的类型。有关更多详细信息,请参见下表。

下表说明了各种链可能发出的某些事件。为简洁起见,已从表中省略了元数据字段。链定义已包含在表之后。

注意

此参考表适用于模式的 v2 版本。

事件 name chunk 输入 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]' '你好'
on_llm_end '[model name]' '你好,人类!'
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, ...])

除了标准事件外,用户还可以分派自定义事件(见下例)。

自定义事件将仅在 API 的 v2 版本中出现!

自定义事件具有以下格式

属性 类型 描述
name str 用户为事件定义的名称。
data 任意 与事件关联的数据。这可以是任何东西,但我们建议使其可 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}

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": [],
    },
]
示例:分派自定义事件
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)
参数 描述
输入

Runnable 的输入。

类型: Any

配置

用于 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

版本

要使用的模式版本,可以是 'v2''v1'。用户应使用 'v2''v1' 是为了向后兼容,并将在 0.4.0 中弃用。在 API 稳定之前不会分配默认值。自定义事件将仅在 'v2' 中出现。

类型: Literal['v1', 'v2'] 默认值: 'v2'

包含名称

仅包括来自具有匹配名称的 Runnable 对象的事件。

类型: Sequence[str] | None 默认值: None

包含类型

仅包括来自具有匹配类型的 Runnable 对象的事件。

类型: Sequence[str] | None 默认值: None

包含标签

仅包括来自具有匹配标签的 Runnable 对象的事件。

类型: Sequence[str] | None 默认值: None

排除名称

排除来自具有匹配名称的 Runnable 对象的事件。

类型: Sequence[str] | None 默认值: None

排除类型

排除来自具有匹配类型的 Runnable 对象的事件。

类型: Sequence[str] | None 默认值: None

排除标签

排除来自具有匹配标签的 Runnable 对象的事件。

类型: Sequence[str] | None 默认值: None

**kwargs

要传递给 Runnable 的其他关键字参数。这些将传递给 astream_log,因为 astream_events 的此实现是建立在 astream_log 之上的。

类型: Any 默认值: {}

YIELDS 描述
AsyncIterator[StreamEvent]

StreamEvent 的异步流。

引发 描述
NotImplementedError

如果版本不是 'v1''v2'

transform

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

将输入转换为输出。

transform 的默认实现,它缓冲输入并调用 astream

如果子类可以在输入仍在生成时开始产生输出,则必须重写此方法。

参数 描述
输入

Runnable 输入的迭代器。

类型: Iterator[Input]

配置

用于 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

**kwargs

要传递给 Runnable 的其他关键字参数。

类型: Any | None 默认值: {}

YIELDS 描述
输出

Runnable 的输出。

atransform async

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

将输入转换为输出。

atransform 的默认实现,它缓冲输入并调用 astream

如果子类可以在输入仍在生成时开始产生输出,则必须重写此方法。

参数 描述
输入

Runnable 输入的异步迭代器。

类型: AsyncIterator[Input]

配置

用于 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

**kwargs

要传递给 Runnable 的其他关键字参数。

类型: Any | None 默认值: {}

YIELDS 描述
AsyncIterator[Output]

Runnable 的输出。

bind

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

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

当链中的 Runnable 需要一个不在前一个 Runnable 的输出中或未包含在用户输入中的参数时非常有用。

参数 描述
**kwargs

要绑定到 Runnable 的参数。

类型: Any 默认值: {}

返回 描述
Runnable[Input, Output]

一个绑定了参数的新 Runnable

示例
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser

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

# Without bind
chain = model | StrOutputParser()

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

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

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

with_config

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

将配置绑定到 Runnable,返回一个新的 Runnable

参数 描述
配置

要绑定到 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

**kwargs

要传递给 Runnable 的其他关键字参数。

类型: Any 默认值: {}

返回 描述
Runnable[Input, Output]

一个绑定了配置的新 Runnable

with_listeners

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

将生命周期侦听器绑定到 Runnable,返回一个新的 Runnable

Run 对象包含有关运行的信息,包括其 idtypeinputoutputerrorstart_timeend_time 以及添加到运行中的任何标签或元数据。

参数 描述
开始时

Runnable 开始运行之前调用,并传入 Run 对象。

类型: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值: None

结束时

Runnable 完成运行后调用,并传入 Run 对象。

类型: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值: None

出错时

如果 Runnable 抛出错误,则调用此函数,并传入 Run 对象。

类型: Callable[[Run], None] | Callable[[Run, RunnableConfig], None] | None 默认值: None

返回 描述
Runnable[Input, Output]

一个绑定了监听器的新 Runnable

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

import time


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


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


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


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

with_alisteners

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

将异步生命周期侦听器绑定到 Runnable

返回一个新的 Runnable

Run 对象包含有关运行的信息,包括其 idtypeinputoutputerrorstart_timeend_time 以及添加到运行中的任何标签或元数据。

参数 描述
开始时

Runnable 开始运行之前异步调用,并传入 Run 对象。

类型: AsyncListener | None 默认值: None

结束时

Runnable 完成运行后异步调用,并传入 Run 对象。

类型: AsyncListener | None 默认值: None

出错时

如果 Runnable 抛出错误,则异步调用此函数,并传入 Run 对象。

类型: AsyncListener | None 默认值: None

返回 描述
Runnable[Input, Output]

一个绑定了监听器的新 Runnable

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

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

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

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

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

runnable = RunnableLambda(test_runnable).with_alisteners(
    on_start=fn_start,
    on_end=fn_end
)
async def concurrent_runs():
    await asyncio.gather(runnable.ainvoke(2), runnable.ainvoke(3))

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

with_types

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

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

参数 描述
输入类型

要绑定到 Runnable 的输入类型。

类型: type[Input] | None 默认值: None

输出类型

要绑定到 Runnable 的输出类型。

类型: type[Output] | None 默认值: None

返回 描述
Runnable[Input, Output]

一个绑定了类型的新 Runnable

with_retry

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

创建一个新的 Runnable,它在发生异常时重试原始的 Runnable

参数 描述
如果异常类型则重试

一个用于重试的异常类型元组。

类型: tuple[type[BaseException], ...] 默认值: (Exception,)

指数等待抖动

是否在两次重试之间的等待时间中添加抖动。

类型: bool 默认值: True

尝试后停止

放弃前尝试的最大次数。

类型: int 默认值: 3

指数抖动参数

tenacity.wait_exponential_jitter 的参数。即:initialmaxexp_basejitter(均为 float 值)。

类型: ExponentialJitterParams | None 默认值: None

返回 描述
Runnable[Input, Output]

一个新的 Runnable,它会在发生异常时重试原始的 Runnable。

示例
from langchain_core.runnables import RunnableLambda

count = 0


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


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

assert count == 2

map

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

返回一个新的 Runnable,它将输入列表映射到输出列表。

使用每个输入调用 invoke

返回 描述
Runnable[list[Input], list[Output]]

一个新的 Runnable,它将输入列表映射到输出列表。

示例
from langchain_core.runnables import RunnableLambda


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


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

with_fallbacks

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

Runnable 添加回退机制,返回一个新的 Runnable

新的 Runnable 将在失败时先尝试原始的 Runnable,然后按顺序尝试每个备选方案。

参数 描述
备用方案

如果原始 Runnable 失败,将尝试的一系列 runnable。

类型: Sequence[Runnable[Input, Output]]

要处理的异常

一个要处理的异常类型元组。

类型: tuple[type[BaseException], ...] 默认值: (Exception,)

异常键

如果指定了 string,则已处理的异常将作为输入的一部分,在指定的键下传递给备选方案。如果为 None,异常将不会传递给备选方案。如果使用,基础 Runnable 及其备选方案必须接受一个字典作为输入。

类型: str | None 默认值: None

返回 描述
RunnableWithFallbacks[Input, Output]

一个新的 Runnable,它会在失败时先尝试原始的 Runnable,然后按顺序尝试每个备选方案。

示例
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
参数 描述
备用方案

如果原始 Runnable 失败,将尝试的一系列 runnable。

类型: Sequence[Runnable[Input, Output]]

要处理的异常

一个要处理的异常类型元组。

类型: tuple[type[BaseException], ...] 默认值: (Exception,)

异常键

如果指定了 string,则已处理的异常将作为输入的一部分,在指定的键下传递给备选方案。如果为 None,异常将不会传递给备选方案。如果使用,基础 Runnable 及其备选方案必须接受一个字典作为输入。

类型: str | None 默认值: None

返回 描述
RunnableWithFallbacks[Input, Output]

一个新的 Runnable,它会在失败时先尝试原始的 Runnable,然后按顺序尝试每个备选方案。

as_tool

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

Runnable 创建一个 BaseTool

as_tool 将从一个 Runnable 实例化一个 BaseTool,该工具具有名称、描述和 args_schema。在可能的情况下,模式会从 runnable.get_input_schema 中推断。或者(例如,如果 Runnable 接受一个字典作为输入,并且特定的字典键没有类型),模式可以通过 args_schema 直接指定。你也可以传递 arg_types 来仅指定必需的参数及其类型。

参数 描述
参数模式

工具的模式。

类型: type[BaseModel] | None 默认值: None

name

工具的名称。

类型: str | None 默认值: None

描述

工具的描述。

类型: str | None 默认值: None

参数类型

一个从参数名称到类型的字典。

类型: dict[str, type] | None 默认值: None

返回 描述
BaseTool

一个 BaseTool 实例。

类型化字典输入

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 输入,通过 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 输入,通过 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]})

字符串输入

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

__init__

__init__(*args: Any, **kwargs: Any) -> None

lc_id classmethod

lc_id() -> list[str]

为此类返回一个用于序列化目的的唯一标识符。

唯一标识符是一个描述对象路径的字符串列表。

例如,对于类 langchain.llms.openai.OpenAI,id 是 ["langchain", "llms", "openai", "OpenAI"]

to_json

to_json() -> SerializedConstructor | SerializedNotImplemented

Runnable 序列化为 JSON。

返回 描述
SerializedConstructor | SerializedNotImplemented

一个 Runnable 的 JSON 可序列化表示。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

序列化一个“未实现”的对象。

返回 描述
SerializedNotImplemented

SerializedNotImplemented.

configurable_fields

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

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

参数 描述
**kwargs

一个要配置的 ConfigurableField 实例的字典。

类型: AnyConfigurableField 默认值: {}

引发 描述
ValueError

如果在 Runnable 中找不到配置键。

返回 描述
RunnableSerializable[Input, Output]

一个配置了字段的新 Runnable

from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

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

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

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

configurable_alternatives

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

为可在运行时设置的 Runnable 对象配置备选项。

参数 描述
哪个

将用于选择备选项的 ConfigurableField 实例。

类型: ConfigurableField

默认键

如果未选择备选项,则使用的默认键。

类型: str 默认值: 'default'

前缀键

是否用 ConfigurableField id 作为键的前缀。

类型: bool 默认值: False

**kwargs

一个从键到 Runnable 实例或返回 Runnable 实例的可调用对象的字典。

类型: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]] 默认值: {}

返回 描述
RunnableSerializable[Input, Output]

一个配置了备选项的新 Runnable

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

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

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

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

set_verbose

set_verbose(verbose: bool | None) -> bool

如果 verbose 是 None,则设置它。

这允许用户传入 None 作为 verbose 来访问全局设置。

参数 描述
verbose

要使用的详细程度设置。

类型: bool | None

返回 描述
bool

要使用的详细程度设置。

generate_prompt

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

将一系列提示传递给模型并返回模型生成的内容。

对于提供批量 API 的模型,此方法应利用批量调用。

当你想要

  1. 利用批量调用,
  2. 需要从模型获得比最高生成值更多的输出,
  3. 正在构建与底层语言模型类型无关的链(例如,纯文本补全模型与聊天模型)。
参数 描述
prompts

PromptValue 对象列表。PromptValue 是一个可以转换为匹配任何语言模型格式的对象(对于纯文本生成模型是字符串,对于聊天模型是 BaseMessage 对象)。

类型: list[PromptValue]

停止

生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。

类型: list[str] | None 默认值: None

回调

要传递的 Callbacks。用于在生成过程中执行额外的功能,如日志记录或流式处理。

类型: Callbacks 默认值: None

**kwargs

任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。

类型: Any 默认值: {}

返回 描述
LLMResult

一个 LLMResult,其中包含每个输入提示的候选 Generation 对象列表以及额外的模型提供商特定的输出。

agenerate_prompt async

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

异步地将一系列提示传递并返回模型生成的内容。

对于提供批量 API 的模型,此方法应利用批量调用。

当你想要

  1. 利用批量调用,
  2. 需要从模型获得比最高生成值更多的输出,
  3. 正在构建与底层语言模型类型无关的链(例如,纯文本补全模型与聊天模型)。
参数 描述
prompts

PromptValue 对象列表。PromptValue 是一个可以转换为匹配任何语言模型格式的对象(对于纯文本生成模型是字符串,对于聊天模型是 BaseMessage 对象)。

类型: list[PromptValue]

停止

生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。

类型: list[str] | None 默认值: None

回调

要传递的 Callbacks。用于在生成过程中执行额外的功能,如日志记录或流式处理。

类型: Callbacks 默认值: None

**kwargs

任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。

类型: Any 默认值: {}

返回 描述
LLMResult

一个 LLMResult,其中包含每个输入提示的候选 Generation 对象列表以及额外的模型提供商特定的输出。

get_token_ids

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

使用 tiktoken 包获取文本中存在的令牌。

get_num_tokens

get_num_tokens(text: str) -> int

获取文本中存在的 token 数量。

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

参数 描述
text

要进行分词的字符串输入。

类型: str

返回 描述
int

文本中的词元整数数量。

get_num_tokens_from_messages

get_num_tokens_from_messages(
    messages: Sequence[BaseMessage],
    tools: Sequence[dict[str, Any] | type | Callable | BaseTool] | None = None,
) -> int

使用 tiktoken 包计算 gpt-3.5-turbogpt-4 的令牌数。

警告

如果您以 base64 字符串形式指定图像并希望计算图像令牌,则必须安装 pillow;如果您以 URL 形式指定图像,则必须同时安装 pillowhttpx。如果未安装这些库,图像输入将在令牌计数中被忽略。

OpenAI 参考.

参数 描述
messages

要进行分词的消息输入。

类型: Sequence[BaseMessage]

工具

如果提供,则为 dictBaseModel、函数或 BaseTool 的序列,将转换为工具模式。

类型: Sequence[dict[str, Any] | type | Callable | BaseTool] | None 默认值: None

generate

generate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: UUID | None = None,
    **kwargs: Any,
) -> LLMResult

将一系列提示传递给模型并返回模型生成的内容。

对于提供批量 API 的模型,此方法应利用批量调用。

当你想要

  1. 利用批量调用,
  2. 需要从模型获得比最高生成值更多的输出,
  3. 正在构建与底层语言模型类型无关的链(例如,纯文本补全模型与聊天模型)。
参数 描述
messages

消息列表的列表。

类型: list[list[BaseMessage]]

停止

生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。

类型: list[str] | None 默认值: None

回调

要传递的 Callbacks。用于在生成过程中执行额外的功能,如日志记录或流式处理。

类型: Callbacks 默认值: None

tags

要应用的标签。

类型: list[str] | None 默认值: None

metadata

要应用的元数据。

类型: dict[str, Any] | None 默认值: None

运行名称

运行的名称。

类型: str | None 默认值: None

run_id

运行的 ID。

类型: UUID | None 默认值: None

**kwargs

任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。

类型: Any 默认值: {}

返回 描述
LLMResult

一个 LLMResult,其中包含每个输入提示的候选 Generations 列表以及额外的模型提供商特定的输出。

agenerate async

agenerate(
    messages: list[list[BaseMessage]],
    stop: list[str] | None = None,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    run_id: UUID | None = None,
    **kwargs: Any,
) -> LLMResult

异步地将一系列提示传递给模型并返回生成的内容。

对于提供批量 API 的模型,此方法应利用批量调用。

当你想要

  1. 利用批量调用,
  2. 需要从模型获得比最高生成值更多的输出,
  3. 正在构建与底层语言模型类型无关的链(例如,纯文本补全模型与聊天模型)。
参数 描述
messages

消息列表的列表。

类型: list[list[BaseMessage]]

停止

生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。

类型: list[str] | None 默认值: None

回调

要传递的 Callbacks。用于在生成过程中执行额外的功能,如日志记录或流式处理。

类型: Callbacks 默认值: None

tags

要应用的标签。

类型: list[str] | None 默认值: None

metadata

要应用的元数据。

类型: dict[str, Any] | None 默认值: None

运行名称

运行的名称。

类型: str | None 默认值: None

run_id

运行的 ID。

类型: UUID | None 默认值: None

**kwargs

任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。

类型: Any 默认值: {}

返回 描述
LLMResult

一个 LLMResult,其中包含每个输入提示的候选 Generations 列表以及额外的模型提供商特定的输出。

dict

dict(**kwargs: Any) -> dict

返回 LLM 的字典。

bind_tools

bind_tools(
    tools: Sequence[dict[str, Any] | type | Callable | BaseTool],
    *,
    tool_choice: dict | str | bool | None = None,
    strict: bool | None = None,
    parallel_tool_calls: bool | None = None,
    **kwargs: Any,
) -> Runnable[LanguageModelInput, AIMessage]

将类工具(tool-like)对象绑定到此聊天模型。

假定模型与 OpenAI 工具调用 API 兼容。

参数 描述
工具

要绑定到此聊天模型的工具定义列表。支持 `langchain_core.utils.function_calling.convert_to_openai_tool` 处理的任何工具定义。

类型: Sequence[dict[str, Any] | type | Callable | BaseTool]

工具选择

要求模型调用的工具。选项包括

  • 格式为 '<<tool_name>>'str:调用 <<tool_name>> 工具。
  • 'auto':自动选择一个工具(包括不选择工具)。
  • 'none':不调用工具。
  • 'any''required'True:强制至少调用一个工具。
  • 格式为 {"type": "function", "function": {"name": <<tool_name>>}}dict:调用 <<tool_name>> 工具。
  • FalseNone:无效果,默认 OpenAI 行为。

类型: dict | str | bool | None 默认值: None

strict

如果为 True,则保证模型输出与工具定义中提供的 JSON Schema 完全匹配。输入模式也将根据支持的模式进行验证。如果为 False,则不验证输入模式,也不验证模型输出。如果为 None,则不会将 strict 参数传递给模型。

类型: bool | None 默认值: None

parallel_tool_calls

设置为 False 以禁用并行工具使用。默认为 None(无指定,允许并行工具使用)。

类型: bool | None 默认值: None

kwargs

任何其他参数都直接传递给 bind

类型: Any 默认值: {}

build_extra classmethod

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

根据传入的额外参数构建额外的关键字参数(kwargs)。

validate_temperature classmethod

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

验证不同模型的温度参数。

  • gpt-5 模型(不包括 gpt-5-chat)只允许 `temperature=1` 或不设置(默认为 1)

get_lc_namespace classmethod

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

返回 描述
list[str]

["langchain", "chat_models", "azure_openai"]

is_lc_serializable classmethod

is_lc_serializable() -> bool

检查该类在 langchain 中是否可序列化。

validate_environment

validate_environment() -> Self

验证 API 密钥和 Python 包是否存在于环境中。

with_structured_output

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

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

参数 描述
模式

输出模式。可以作为以下形式传入:

  • 一个 JSON Schema,
  • 一个 TypedDict 类,
  • 一个 Pydantic 类,
  • 或一个 OpenAI 函数/工具模式。

如果 schema 是一个 Pydantic 类,那么模型输出将是该类的 Pydantic 实例,并且模型生成的字段将由 Pydantic 类进行验证。否则,模型输出将是一个字典,并且不会被验证。

有关在指定 Pydantic 或 TypedDict 类时如何正确指定模式字段的类型和描述的更多信息,请参阅 langchain_core.utils.function_calling.convert_to_openai_tool

类型: _DictOrPydanticClass | None 默认值: None

method

用于引导模型生成的方法,可选值之一:

  • 'json_schema':使用 OpenAI 的结构化输出 API。支持 'gpt-4o-mini''gpt-4o-2024-08-06''o1' 及更高版本的模型。
  • 'function_calling':使用 OpenAI 的工具调用(以前称为函数调用)API
  • 'json_mode':使用 OpenAI 的JSON 模式。请注意,如果使用 JSON 模式,则必须在模型调用中包含将输出格式化为所需模式的指令

点击这里了解更多关于这些方法的区别以及哪些模型支持哪些方法。

类型: Literal['function_calling', 'json_mode', 'json_schema'] 默认值: 'json_schema'

包含原始数据

如果为 False,则仅返回解析后的结构化输出。如果在模型输出解析期间发生错误,它将被引发。如果为 True,则将返回原始模型响应(一个 BaseMessage)和解析后的模型响应。如果在输出解析期间发生错误,它将被捕获并一并返回。

最终输出总是一个带有键 'raw''parsed''parsing_error'dict

类型: bool 默认值: False

strict
  • True:保证模型输出与模式完全匹配。输入模式也将根据支持的模式进行验证。
  • False:不验证输入模式,也不验证模型输出。
  • None:不会将 `strict` 参数传递给模型。

如果通过 TypedDict 或 JSON 模式指定模式,则默认不启用 `strict`。传递 `strict=True` 来启用它。

注意

strict 只有在 method'json_schema''function_calling' 时才能非空。

类型: bool | None 默认值: None

工具

要绑定到聊天模型的类工具对象的列表。要求

  • method'json_schema'(默认)。
  • strict=True
  • include_raw=True

如果模型选择调用工具,则 'raw' 中的结果 AIMessage 将包含工具调用。

示例
from langchain.chat_models import init_chat_model
from pydantic import BaseModel


class ResponseSchema(BaseModel):
    response: str


def get_weather(location: str) -> str:
    \"\"\"Get weather at a location.\"\"\"
    pass

model = init_chat_model("openai:gpt-4o-mini")

structured_model = model.with_structured_output(
    ResponseSchema,
    tools=[get_weather],
    strict=True,
    include_raw=True,
)

structured_model.invoke("What's the weather in Boston?")
{
    "raw": AIMessage(content="", tool_calls=[...], ...),
    "parsing_error": None,
    "parsed": None,
}

kwargs

额外的关键字参数会传递给模型。

类型: Any 默认值: {}

返回 描述
Runnable[LanguageModelInput, _DictOrPydantic]

一个接受与 langchain_core.language_models.chat.BaseChatModel 相同输入的 Runnable。如果 include_rawFalseschema 是一个 Pydantic 类,则 Runnable 输出一个 schema 的实例(即一个 Pydantic 对象)。否则,如果 include_rawFalse,则 Runnable 输出一个 dict

如果 include_rawTrue,则 Runnable 输出一个带有以下键的 dict

  • 'raw': BaseMessage
  • 'parsed': 如果出现解析错误则为 None,否则类型取决于上面描述的 schema
  • 'parsing_error': BaseException | None

行为在 0.3.0 版本中发生变化

method 的默认值从 "function_calling" 更改为 "json_schema"。

行为在 0.3.12 中已更改

增加了对 tools 的支持。

行为在 0.3.21 中已更改

kwargs 传递给模型。

示例:schema=Pydantic 类,method='json_schema'include_raw=Falsestrict=True

注意,如果 strict = True,OpenAI 对可提供的模式类型有一些限制。使用 Pydantic 时,我们的模型不能指定任何字段元数据(如最小/最大约束),并且字段不能有默认值。

点击这里查看所有约束。

from typing import Optional

from langchain_openai import AzureChatOpenAI
from pydantic import BaseModel, Field


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

    answer: str
    justification: str | None = Field(
        default=..., description="A justification for the answer."
    )


model = AzureChatOpenAI(
    azure_deployment="...", model="gpt-4o", temperature=0
)
structured_model = model.with_structured_output(AnswerWithJustification)

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

# -> AnswerWithJustification(
#     answer='They weigh the same',
#     justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'
# )
示例:schema=Pydantic 类,method='function_calling'include_raw=Falsestrict=False
from typing import Optional

from langchain_openai import AzureChatOpenAI
from pydantic import BaseModel, Field


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

    answer: str
    justification: str | None = Field(
        default=..., description="A justification for the answer."
    )


model = AzureChatOpenAI(
    azure_deployment="...", model="gpt-4o", temperature=0
)
structured_model = model.with_structured_output(
    AnswerWithJustification, method="function_calling"
)

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

# -> AnswerWithJustification(
#     answer='They weigh the same',
#     justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'
# )
示例:schema=Pydantic 类,method='json_schema'include_raw=True
from langchain_openai import AzureChatOpenAI
from pydantic import BaseModel


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

    answer: str
    justification: str


model = AzureChatOpenAI(
    azure_deployment="...", model="gpt-4o", temperature=0
)
structured_model = model.with_structured_output(
    AnswerWithJustification, include_raw=True
)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)
# -> {
#     'raw': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_Ao02pnFYXD6GN1yzc0uXPsvF', 'function': {'arguments': '{"answer":"They weigh the same.","justification":"Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ."}', 'name': 'AnswerWithJustification'}, 'type': 'function'}]}),
#     'parsed': AnswerWithJustification(answer='They weigh the same.', justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'),
#     'parsing_error': None
# }
示例:schema=TypedDict 类,method='json_schema'include_raw=Falsestrict=False
from typing_extensions import Annotated, TypedDict

from langchain_openai import AzureChatOpenAI


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

    answer: str
    justification: Annotated[
        str | None, None, "A justification for the answer."
    ]


model = AzureChatOpenAI(
    azure_deployment="...", model="gpt-4o", temperature=0
)
structured_model = model.with_structured_output(AnswerWithJustification)

structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)
# -> {
#     'answer': 'They weigh the same',
#     'justification': 'Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume and density of the two substances differ.'
# }
示例:schema=OpenAI 函数模式,method='json_schema'include_raw=False
from langchain_openai import AzureChatOpenAI

oai_schema = {
    'name': 'AnswerWithJustification',
    'description': 'An answer to the user question along with justification for the answer.',
    'parameters': {
        'type': 'object',
        'properties': {
            'answer': {'type': 'string'},
            'justification': {'description': 'A justification for the answer.', 'type': 'string'}
        },
        'required': ['answer']
    }

    model = AzureChatOpenAI(
        azure_deployment="...",
        model="gpt-4o",
        temperature=0,
    )
    structured_model = model.with_structured_output(oai_schema)

    structured_model.invoke(
        "What weighs more a pound of bricks or a pound of feathers"
    )
    # -> {
    #     'answer': 'They weigh the same',
    #     'justification': 'Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume and density of the two substances differ.'
    # }
示例:schema=Pydantic 类,method='json_mode'include_raw=True
from langchain_openai import AzureChatOpenAI
from pydantic import BaseModel


class AnswerWithJustification(BaseModel):
    answer: str
    justification: str


model = AzureChatOpenAI(
    azure_deployment="...",
    model="gpt-4o",
    temperature=0,
)
structured_model = model.with_structured_output(
    AnswerWithJustification, method="json_mode", include_raw=True
)

structured_model.invoke(
    "Answer the following question. "
    "Make sure to return a JSON blob with keys 'answer' and 'justification'.\\n\\n"
    "What's heavier a pound of bricks or a pound of feathers?"
)
# -> {
#     'raw': AIMessage(content='{\\n    "answer": "They are both the same weight.",\\n    "justification": "Both a pound of bricks and a pound of feathers weigh one pound. The difference lies in the volume and density of the materials, not the weight." \\n}'),
#     'parsed': AnswerWithJustification(answer='They are both the same weight.', justification='Both a pound of bricks and a pound of feathers weigh one pound. The difference lies in the volume and density of the materials, not the weight.'),
#     'parsing_error': None
# }
示例:schema=Nonemethod='json_mode'include_raw=True
structured_model = model.with_structured_output(
    method="json_mode", include_raw=True
)

structured_model.invoke(
    "Answer the following question. "
    "Make sure to return a JSON blob with keys 'answer' and 'justification'.\\n\\n"
    "What's heavier a pound of bricks or a pound of feathers?"
)
# -> {
#     'raw': AIMessage(content='{\\n    "answer": "They are both the same weight.",\\n    "justification": "Both a pound of bricks and a pound of feathers weigh one pound. The difference lies in the volume and density of the materials, not the weight." \\n}'),
#     'parsed': {
#         'answer': 'They are both the same weight.',
#         'justification': 'Both a pound of bricks and a pound of feathers weigh one pound. The difference lies in the volume and density of the materials, not the weight.'
#     },
#     'parsing_error': None
# }
© . This site is unofficial and not affiliated with LangChain, Inc.