`ChatOpenAI`¶

参考文档

此页面包含 ChatOpenAI 的参考文档。有关使用 ChatOpenAI 的概念指南、教程和示例，请参阅文档。

langchain_openai.chat_models.ChatOpenAI ¶

基类：BaseChatOpenAI

OpenAI 聊天模型 API 的接口。

设置

安装 langchain-openai 并设置环境变量 OPENAI_API_KEY。

pip install -U langchain-openai

# or using uv
uv add langchain-openai

export OPENAI_API_KEY="your-api-key"

关键初始化参数 — 完成参数

参数	类型	描述
`model`	`str`	要使用的 OpenAI 模型名称。
`temperature`	`浮点数`	采样温度。
`max_tokens`	`整数 \| None`	要生成的最大 token 数。
`logprobs`	`布尔值 \| None`	是否返回 logprobs。
`stream_options`	`dict`	配置流式输出，例如是否在流式传输时返回 token 使用情况（`{"include_usage": True}`）。
`use_responses_api`	`布尔值 \| None`	是否使用 responses API。

请参阅下面支持的完整初始化参数及其描述列表。

关键初始化参数 — 客户端参数

参数	类型	描述
`timeout`	`浮点数 \| Tuple[浮点数, 浮点数] \| Any \| None`	请求超时。
`max_retries`	`整数 \| None`	最大重试次数。
`api_key`	`字符串 \| None`	OpenAI API 密钥。如果未传入，将从环境变量 `OPENAI_API_KEY` 中读取。
`base_url`	`字符串 \| None`	API 请求的基础 URL。仅在使用代理或服务模拟器时指定。
`organization`	`字符串 \| None`	OpenAI 组织 ID。如果未传入，将从环境变量 `OPENAI_ORG_ID` 中读取。

请参阅下面支持的完整初始化参数及其描述列表。

实例化

使用所需参数创建模型实例。例如：

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="...",
    temperature=0,
    max_tokens=None,
    timeout=None,
    max_retries=2,
    # api_key="...",
    # base_url="...",
    # organization="...",
    # other params...
)

请参阅下面所有可用参数。

保留参数

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

from langchain_openai import ChatOpenAI
import openai

ChatOpenAI(..., frequency_penalty=0.2).invoke(...)

# Results in underlying API call of:

openai.OpenAI(..).chat.completions.create(..., frequency_penalty=0.2)

# Which is also equivalent to:

ChatOpenAI(...).invoke(..., frequency_penalty=0.2)

调用

从模型生成响应

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

结果为 AIMessage 响应

AIMessage(
    content="J'adore la programmation.",
    response_metadata={
        "token_usage": {
            "completion_tokens": 5,
            "prompt_tokens": 31,
            "total_tokens": 36,
        },
        "model_name": "gpt-4o",
        "system_fingerprint": "fp_43dfabdef1",
        "finish_reason": "stop",
        "logprobs": None,
    },
    id="run-012cffe2-5d3d-424d-83b5-51c6d4a593d1-0",
    usage_metadata={"input_tokens": 31, "output_tokens": 5, "total_tokens": 36},
)

流

从模型流式传输响应

for chunk in model.stream(messages):
    print(chunk.text, end="")

结果为一系列带有部分内容的 AIMessageChunk 对象

AIMessageChunk(content="", id="run-9e1517e3-12bf-48f2-bb1b-2e824f7cd7b0")
AIMessageChunk(content="J", id="run-9e1517e3-12bf-48f2-bb1b-2e824f7cd7b0")
AIMessageChunk(content="'adore", id="run-9e1517e3-12bf-48f2-bb1b-2e824f7cd7b0")
AIMessageChunk(content=" la", id="run-9e1517e3-12bf-48f2-bb1b-2e824f7cd7b0")
AIMessageChunk(
    content=" programmation", id="run-9e1517e3-12bf-48f2-bb1b-2e824f7cd7b0"
)
AIMessageChunk(content=".", id="run-9e1517e3-12bf-48f2-bb1b-2e824f7cd7b0")
AIMessageChunk(
    content="",
    response_metadata={"finish_reason": "stop"},
    id="run-9e1517e3-12bf-48f2-bb1b-2e824f7cd7b0",
)

要收集完整的消息，您可以连接这些块

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"},
    id="run-bf917526-7f58-4683-84f7-36a6b671d140",
)

异步

invoke、stream 和 batch 的异步等效项也可用

# Invoke
await model.ainvoke(messages)

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

# Batch
await model.abatch([messages])

结果为 AIMessage 响应

AIMessage(
    content="J'adore la programmation.",
    response_metadata={
        "token_usage": {
            "completion_tokens": 5,
            "prompt_tokens": 31,
            "total_tokens": 36,
        },
        "model_name": "gpt-4o",
        "system_fingerprint": "fp_43dfabdef1",
        "finish_reason": "stop",
        "logprobs": None,
    },
    id="run-012cffe2-5d3d-424d-83b5-51c6d4a593d1-0",
    usage_metadata={
        "input_tokens": 31,
        "output_tokens": 5,
        "total_tokens": 36,
    },
)

对于批量调用，结果为 list[AIMessage]。

工具调用

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]
    # strict = True  # Enforce tool args schema is respected
)
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",
    },
]

并行工具调用

openai >= 1.32 支持 parallel_tool_calls 参数，默认为 True。可以将此参数设置为 False 以禁用并行工具调用。

ai_msg = model_with_tools.invoke(
    "What is the weather in LA and NY?", parallel_tool_calls=False
)
ai_msg.tool_calls

[
    {
        "name": "GetWeather",
        "args": {"location": "Los Angeles, CA"},
        "id": "call_4OoY0ZR99iEvC7fevsH8Uhtz",
    }
]

与其他运行时参数一样，parallel_tool_calls 可以使用 model.bind(parallel_tool_calls=False) 绑定到模型，或者在实例化时通过设置 model_kwargs 来绑定。

有关更多信息，请参阅 bind_tools。

内置（服务器端）工具

您可以访问 OpenAI Responses API 支持的内置工具。有关更多详细信息，请参阅 LangChain 文档。

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="...", output_version="responses/v1")

tool = {"type": "web_search"}
model_with_tools = model.bind_tools([tool])

response = model_with_tools.invoke("What was a positive news story from today?")
response.content

[
    {
        "type": "text",
        "text": "Today, a heartwarming story emerged from ...",
        "annotations": [
            {
                "end_index": 778,
                "start_index": 682,
                "title": "Title of story",
                "type": "url_citation",
                "url": "<url of story>",
            }
        ],
    }
]

于 0.3.9 版本中添加

在版本 0.3.26 中添加：更新了 AIMessage 格式

langchain-openai >= 0.3.26 允许用户在使用 Responses API 时选择加入更新的 AIMessage 格式。设置 ChatOpenAI(..., output_version="responses/v1") 会将推理摘要、内置工具调用和其他响应项的输出格式化到消息的 content 字段中，而不是 additional_kwargs。我们建议新应用程序使用此格式。

管理对话状态

OpenAI 的 Responses API 支持对话状态管理。传入先前消息的响应 ID 将继续对话线程。

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="...",
    use_responses_api=True,
    output_version="responses/v1",
)
response = model.invoke("Hi, I'm Bob.")
response.text

"Hi Bob! How can I assist you today?"

second_response = model.invoke(
    "What is my name?",
    previous_response_id=response.response_metadata["id"],
)
second_response.text

"Your name is Bob. How can I help you today, Bob?"

于 0.3.9 版本中添加

在 0.3.26 版本中添加

您还可以使用 use_previous_response_id 初始化 ChatOpenAI。然后，直到最新响应的输入消息将从请求负载中删除，并且 previous_response_id 将使用最新响应的 ID 进行设置。

model = ChatOpenAI(model="...", use_previous_response_id=True)

推理输出

OpenAI 的 Responses API 支持推理模型，这些模型会公开内部推理过程的摘要。

from langchain_openai import ChatOpenAI

reasoning = {
    "effort": "medium",  # 'low', 'medium', or 'high'
    "summary": "auto",  # 'detailed', 'auto', or None
}

model = ChatOpenAI(
    model="...", reasoning=reasoning, output_version="responses/v1"
)
response = model.invoke("What is 3^3?")

# Response text
print(f"Output: {response.text}")

# Reasoning summaries
for block in response.content:
    if block["type"] == "reasoning":
        for summary in block["summary"]:
            print(summary["text"])

Output: 3³ = 27
Reasoning: The user wants to know...

在版本 0.3.26 中添加：更新了 AIMessage 格式

langchain-openai >= 0.3.26 允许用户在使用 Responses API 时选择加入更新的 AIMessage 格式。设置 ChatOpenAI(..., output_version="responses/v1") 会将推理摘要、内置工具调用和其他响应项的输出格式化到消息的 content 字段中，而不是 additional_kwargs。我们建议新应用程序使用此格式。

结构化输出

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

有关更多信息，请参阅 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.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 clear and pleasant. The sky is mostly blue with scattered, light clouds, suggesting a sunny day with minimal cloud cover. There is no indication of rain or strong winds, and the overall scene looks bright and calm. The lush green grass and clear visibility further indicate good weather conditions."

令牌使用情况

ai_msg = model.invoke(messages)
ai_msg.usage_metadata

```txt
{"input_tokens": 28, "output_tokens": 5, "total_tokens": 33}

流式传输时，设置 stream_usage 关键字参数

stream = model.stream(messages, stream_usage=True)
full = next(stream)
for chunk in stream:
    full += chunk
full.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"]

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

{
    "token_usage": {
        "completion_tokens": 5,
        "prompt_tokens": 28,
        "total_tokens": 33,
    },
    "model_name": "gpt-4o",
    "system_fingerprint": "fp_319be4768e",
    "finish_reason": "stop",
    "logprobs": None,
}

弹性处理

OpenAI 提供多种服务层级。“弹性 (flex)”层级为请求提供更便宜的价格，但代价是响应可能需要更长时间，并且资源可能不总是可用。这种方法最适合非关键任务，包括模型测试、数据增强或可以异步运行的作业。

要使用它，请使用 service_tier="flex" 初始化模型

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="...", service_tier="flex")

请注意，这是一个 beta 功能，仅适用于部分模型。有关更多详细信息，请参阅 OpenAI 弹性处理文档。

与 OpenAI 兼容的 API

ChatOpenAI 可以与兼容 OpenAI 的 API 一起使用，例如 LM Studio、vLLM、Ollama 等。

要使用这些提供商特有的自定义参数，请使用 extra_body 参数。

带 TTL（自动逐出）的 LM Studio 示例

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    base_url="https://:1234/v1",
    api_key="lm-studio",  # Can be any string
    model="mlx-community/QwQ-32B-4bit",
    temperature=0,
    extra_body={"ttl": 300},  # Auto-evict model after 5 minutes of inactivity
)

带自定义参数的 vLLM 示例

model = ChatOpenAI(
    base_url="https://:8000/v1",
    api_key="EMPTY",
    model="meta-llama/Llama-2-7b-chat-hf",
    extra_body={"use_beam_search": True, "best_of": 4},
)

model_kwargs 与 extra_body

针对不同类型的 API 参数使用正确的参数

使用 model_kwargs 用于：

未明确定义为类参数的标准 OpenAI API 参数
应平铺到顶级请求负载中的参数
示例：max_completion_tokens、stream_options、modalities、audio

# Standard OpenAI parameters
model = ChatOpenAI(
    model="...",
    model_kwargs={
        "stream_options": {"include_usage": True},
        "max_completion_tokens": 300,
        "modalities": ["text", "audio"],
        "audio": {"voice": "alloy", "format": "wav"},
    },
)

使用 extra_body 用于：

与 OpenAI 兼容的提供商（vLLM、LM Studio、OpenRouter 等）特有的自定义参数
需要在请求中嵌套在 extra_body 下的参数
任何非标准的 OpenAI API 参数

# Custom provider parameters
model = ChatOpenAI(
    base_url="https://:8000/v1",
    model="custom-model",
    extra_body={
        "use_beam_search": True,  # vLLM parameter
        "best_of": 4,  # vLLM parameter
        "ttl": 300,  # LM Studio parameter
    },
)

主要区别

model_kwargs：参数合并到顶级请求负载中
extra_body：参数在请求中嵌套在 extra_body 键下

警告

对于自定义参数，请始终使用 extra_body，而不是 model_kwargs。对非 OpenAI 参数使用 model_kwargs 会导致 API 错误。

提示缓存优化

对于具有重复提示的高容量应用程序，每次调用使用 prompt_cache_key 以提高缓存命中率并降低成本。

model = ChatOpenAI(model="...")

response = model.invoke(
    messages,
    prompt_cache_key="example-key-a",  # Routes to same machine for cache hits
)

customer_response = model.invoke(messages, prompt_cache_key="example-key-b")
support_response = model.invoke(messages, prompt_cache_key="example-key-c")

# Dynamic cache keys based on context
cache_key = f"example-key-{dynamic_suffix}"
response = model.invoke(messages, prompt_cache_key=cache_key)

缓存键有助于确保具有相同提示前缀的请求被路由到具有现有缓存的机器，从而为缓存的 token 提供成本降低和延迟改进。

方法	描述
`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`	向此 `Runnable` 的 `dict` 输出分配新字段。
`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-turbo` 和 `gpt-4` 的令牌数。
`generate`	将一系列提示传递给模型并返回模型生成的内容。
`agenerate`	异步地将一系列提示传递给模型并返回生成的内容。
`dict`	返回 LLM 的字典。
`bind_tools`	将类工具（tool-like）对象绑定到此聊天模型。
`build_extra`	根据传入的额外参数构建额外的关键字参数（kwargs）。
`validate_temperature`	验证不同模型的 temperature 参数。
`validate_environment`	验证 API 密钥和 Python 包是否存在于环境中。
`get_lc_namespace`	获取 LangChain 对象的命名空间。
`is_lc_serializable`	返回此模型是否可以被 LangChain 序列化。
`with_structured_output`	返回与给定模式匹配的格式化输出的模型包装器。

name `类属性` `实例属性` ¶

name: str | None = None

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

InputType `属性` ¶

InputType: TypeAlias

获取此 Runnable 的输入类型。

OutputType `属性` ¶

OutputType: Any

获取此 Runnable 的输出类型。

input_schema `属性` ¶

input_schema: type[BaseModel]

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

output_schema `属性` ¶

output_schema: type[BaseModel]

输出模式。

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

config_specs `属性` ¶

config_specs: list[ConfigurableFieldSpec]

列出此 Runnable 的可配置字段。

cache `类属性` `实例属性` ¶

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

是否缓存响应。

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

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

verbose `类属性` `实例属性` ¶

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

是否打印响应文本。

callbacks `类属性` `实例属性` ¶

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

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

tags `类属性` `实例属性` ¶

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

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

metadata `类属性` `实例属性` ¶

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

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

custom_get_token_ids `类属性` `实例属性` ¶

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

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

rate_limiter `类属性` `实例属性` ¶

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

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

disable_streaming `类属性` `实例属性` ¶

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

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

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

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

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

output_version `类属性` `实例属性` ¶

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 `属性` ¶

profile: ModelProfile

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

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

引发	描述
`ImportError`	如果未安装 `langchain-model-profiles`。

返回	描述
`ModelProfile`	一个 `ModelProfile` 对象，包含模型的性能分析信息。

model_name `类属性` `实例属性` ¶

model_name: str = Field(default='gpt-3.5-turbo', alias='model')

要使用的模型名称。

temperature `类属性` `实例属性` ¶

temperature: float | None = None

要使用的采样温度。

model_kwargs `类属性` `实例属性` ¶

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

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

openai_api_key `类属性` `实例属性` ¶

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

要使用的 API 密钥。

可以从 OPENAI_API_KEY 环境变量中推断，或指定为字符串，或返回字符串的同步或异步可调用对象。

使用环境变量指定

export OPENAI_API_KEY=...

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-5-nano")

使用字符串指定

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-5-nano", api_key="...")

使用同步可调用对象指定

from langchain_openai import ChatOpenAI

def get_api_key() -> str:
    # Custom logic to retrieve API key
    return "..."

model = ChatOpenAI(model="gpt-5-nano", api_key=get_api_key)

使用异步可调用对象指定

from langchain_openai import ChatOpenAI

async def get_api_key() -> str:
    # Custom async logic to retrieve API key
    return "..."

model = ChatOpenAI(model="gpt-5-nano", api_key=get_api_key)

openai_api_base `类属性` `实例属性` ¶

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

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

openai_organization `类属性` `实例属性` ¶

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

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

request_timeout `类属性` `实例属性` ¶

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

向 OpenAI 补全 API 发出请求的超时时间。可以是浮点数、`httpx.Timeout` 或 `None`。

stream_usage `类属性` `实例属性` ¶

stream_usage: bool | None = None

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

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

于 0.3.9 版本中添加

行为在 0.3.35 版本中发生变化

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

max_retries `类属性` `实例属性` ¶

max_retries: int | None = None

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

presence_penalty `类属性` `实例属性` ¶

presence_penalty: float | None = None

对重复的令牌进行惩罚。

frequency_penalty `类属性` `实例属性` ¶

frequency_penalty: float | None = None

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

seed `类属性` `实例属性` ¶

seed: int | None = None

生成的种子

logprobs `类属性` `实例属性` ¶

logprobs: bool | None = None

是否返回 logprobs。

top_logprobs `类属性` `实例属性` ¶

top_logprobs: int | None = None

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

logit_bias `类属性` `实例属性` ¶

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

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

streaming `类属性` `实例属性` ¶

streaming: bool = False

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

n `类属性` `实例属性` ¶

n: int | None = None

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

top_p `类属性` `实例属性` ¶

top_p: float | None = None

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

reasoning_effort `类属性` `实例属性` ¶

reasoning_effort: str | None = None

约束推理模型的推理工作量。用于 Chat Completions API。

仅限推理模型。

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

reasoning `类属性` `实例属性` ¶

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 `类属性` `实例属性` ¶

verbosity: str | None = None

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

当前支持的值为 `'low'`、`'medium'` 和 `'high'`。

在 0.3.28 版本中添加

tiktoken_model_name `类属性` `实例属性` ¶

tiktoken_model_name: str | None = None

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

http_client `类属性` `实例属性` ¶

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

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

http_async_client `类属性` `实例属性` ¶

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

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

stop `类属性` `实例属性` ¶

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

默认的停止序列。

extra_body `类属性` `实例属性` ¶

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 `类属性` `实例属性` ¶

include_response_headers: bool = False

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

disabled_params `类属性` `实例属性` ¶

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}。

如果一个参数被禁用，那么它将不会在任何方法中默认使用，例如在 with_structured_output 中。但这并不妨碍用户在调用期间直接传入该参数。

include `类属性` `实例属性` ¶

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 `类属性` `实例属性` ¶

service_tier: str | None = None

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

store `类属性` `实例属性` ¶

store: bool | None = None

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

对于 Responses API 默认为 `True`，对于 Chat Completions API 默认为 `False`。

在 0.3.24 版本中添加

truncation `类属性` `实例属性` ¶

truncation: str | None = None

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

在 0.3.24 版本中添加

use_previous_response_id `类属性` `实例属性` ¶

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 `类属性` `实例属性` ¶

use_responses_api: bool | None = None

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

如果未指定，将根据调用参数推断。

于 0.3.9 版本中添加

max_tokens `类属性` `实例属性` ¶

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

生成的最大令牌数。

lc_secrets `属性` ¶

lc_secrets: dict[str, str]

机密环境变量的映射。

lc_attributes `属性` ¶

lc_attributes: dict[str, Any]

获取 langchain 对象的属性。

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_fields 和 configurable_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_fields 和 configurable_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_fields 和 configurable_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 ¶

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

向此 Runnable 的 dict 输出分配新字段。

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 `异步` ¶

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 `异步` ¶

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 `异步` ¶

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 `异步` ¶

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 `异步` ¶

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]`	一个 `RunLogPatch` 或 `RunLog` 对象。

astream_events `异步` ¶

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 `异步` ¶

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 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_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 对象包含有关运行的信息，包括其 id、type、input、output、error、start_time、end_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` 的参数。即：`initial`、`max`、`exp_base` 和 `jitter`（均为 `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 `类方法` ¶

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 的模型，此方法应利用批量调用。

当你想要

利用批量调用，
需要从模型获得比最高生成值更多的输出，
正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
`prompts`	`PromptValue` 对象列表。`PromptValue` 是一个可以转换为匹配任何语言模型格式的对象（对于纯文本生成模型是字符串，对于聊天模型是 `BaseMessage` 对象）。类型： `list[PromptValue]`
`停止`	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。类型: `list[str] \| None` 默认值: `None`
`回调`	要传递的 `Callbacks`。用于在生成过程中执行额外的功能，如日志记录或流式处理。类型： `Callbacks` 默认值： `None`
`**kwargs`	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。类型： `Any` 默认值： `{}`

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

agenerate_prompt `异步` ¶

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

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

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

当你想要

利用批量调用，
需要从模型获得比最高生成值更多的输出，
正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
`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-turbo` 和 `gpt-4` 的令牌数。

警告

如果您想计算图像令牌并且将图像指定为 base64 字符串，则必须安装 `pillow`；如果您将图像指定为 URL，则必须同时安装 `pillow` 和 `httpx`。如果未安装这些库，图像输入将在令牌计数中被忽略。

OpenAI 参考.

参数	描述
`messages`	要进行分词的消息输入。类型： `Sequence[BaseMessage]`
`工具`	如果提供，将要转换为工具模式的 `dict`、`BaseModel`、函数或 `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 的模型，此方法应利用批量调用。

当你想要

利用批量调用，
需要从模型获得比最高生成值更多的输出，
正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
`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 `异步` ¶

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 的模型，此方法应利用批量调用。

当你想要

利用批量调用，
需要从模型获得比最高生成值更多的输出，
正在构建与底层语言模型类型无关的链（例如，纯文本补全模型与聊天模型）。

参数	描述
`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>>` 工具。 `False` 或 `None`：无效果，默认 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 `类方法` ¶

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

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

validate_temperature `类方法` ¶

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

验证不同模型的 temperature 参数。

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

validate_environment ¶

validate_environment() -> Self

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

get_lc_namespace `类方法` ¶

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

返回	描述
`list[str]`	`["langchain", "chat_models", "openai"]`

is_lc_serializable `类方法` ¶

is_lc_serializable() -> bool

返回此模型是否可以被 LangChain 序列化。

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]

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

参数	描述
`模式`	输出模式。可以作为以下形式传入：一个 OpenAI 函数/工具模式，一个 JSON Schema，一个 `TypedDict` 类，或一个 Pydantic 类。如果 `schema` 是一个 Pydantic 类，那么模型输出将是该类的 Pydantic 实例，并且模型生成的字段将由 Pydantic 类进行验证。否则，模型输出将是一个字典，并且不会被验证。有关在指定 Pydantic 或 `TypedDict` 类时如何正确指定模式字段的类型和描述的更多信息，请参阅 `langchain_core.utils.function_calling.convert_to_openai_tool`。类型： `_DictOrPydanticClass \| None` 默认值： `None`
`method`	用于引导模型生成的方法，可选值之一： `'json_schema'`：使用 OpenAI 的结构化输出 API。有关支持的模型列表，请参阅文档。 `'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 schema 指定了 schema，`strict` 模式默认不开启。传入 `strict=True` 来启用它。注意仅当 `method` 为 `'json_schema'` 或 `'function_calling'` 时，`strict` 才能为非空值。类型： `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_raw 为 False 且 schema 是一个 Pydantic 类，则 Runnable 输出一个 schema 的实例（即一个 Pydantic 对象）。否则，如果 include_raw 为 False，则 Runnable 输出一个 dict。

如果 include_raw 为 True，则 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=False, strict=True

请注意，如果 strict = True，OpenAI 对可提供的 schema 类型有许多限制。使用 Pydantic 时，我们的模型不能指定任何字段元数据（如最小/最大值约束），并且字段不能有默认值。

查看所有约束。

from langchain_openai import ChatOpenAI
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 = ChatOpenAI(model="...", 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=False, strict=False

from langchain_openai import ChatOpenAI
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 = ChatOpenAI(model="...", 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 ChatOpenAI
from pydantic import BaseModel


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

    answer: str
    justification: str


model = ChatOpenAI(model="...", 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=False, strict=False

from typing_extensions import Annotated, TypedDict

from langchain_openai import ChatOpenAI


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 = ChatOpenAI(model="...", 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 函数 schema, method='json_schema', include_raw=False

from langchain_openai import ChatOpenAI

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 = ChatOpenAI(model="...", 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 ChatOpenAI
from pydantic import BaseModel


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


model = ChatOpenAI(model="...", 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=None，method='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,
}

ChatOpenAI¶

langchain_openai.chat_models.ChatOpenAI ¶

name 类属性 实例属性 ¶

InputType 属性 ¶

OutputType 属性 ¶

input_schema 属性 ¶

output_schema 属性 ¶

config_specs 属性 ¶

cache 类属性 实例属性 ¶

verbose 类属性 实例属性 ¶

callbacks 类属性 实例属性 ¶

tags 类属性 实例属性 ¶

metadata 类属性 实例属性 ¶

custom_get_token_ids 类属性 实例属性 ¶

rate_limiter 类属性 实例属性 ¶

disable_streaming 类属性 实例属性 ¶

output_version 类属性 实例属性 ¶

profile 属性 ¶

model_name 类属性 实例属性 ¶

temperature 类属性 实例属性 ¶

model_kwargs 类属性 实例属性 ¶

openai_api_key 类属性 实例属性 ¶

openai_api_base 类属性 实例属性 ¶

openai_organization 类属性 实例属性 ¶

request_timeout 类属性 实例属性 ¶

stream_usage 类属性 实例属性 ¶

max_retries 类属性 实例属性 ¶

presence_penalty 类属性 实例属性 ¶

frequency_penalty 类属性 实例属性 ¶

seed 类属性 实例属性 ¶

logprobs 类属性 实例属性 ¶

top_logprobs 类属性 实例属性 ¶

logit_bias 类属性 实例属性 ¶

streaming 类属性 实例属性 ¶

n 类属性 实例属性 ¶

top_p 类属性 实例属性 ¶

reasoning_effort 类属性 实例属性 ¶

reasoning 类属性 实例属性 ¶

verbosity 类属性 实例属性 ¶

tiktoken_model_name 类属性 实例属性 ¶

http_client 类属性 实例属性 ¶

http_async_client 类属性 实例属性 ¶

stop 类属性 实例属性 ¶

extra_body 类属性 实例属性 ¶

include_response_headers 类属性 实例属性 ¶

disabled_params 类属性 实例属性 ¶

include 类属性 实例属性 ¶

service_tier 类属性 实例属性 ¶

store 类属性 实例属性 ¶

truncation 类属性 实例属性 ¶

use_previous_response_id 类属性 实例属性 ¶

use_responses_api 类属性 实例属性 ¶

max_tokens 类属性 实例属性 ¶

lc_secrets 属性 ¶

lc_attributes 属性 ¶

get_name ¶

get_input_schema ¶

get_input_jsonschema ¶

get_output_schema ¶

get_output_jsonschema ¶

config_schema ¶

get_config_jsonschema ¶

get_graph ¶

get_prompts ¶

__or__ ¶

__ror__ ¶

pipe ¶

pick ¶

assign ¶

invoke ¶

ainvoke 异步 ¶

batch ¶

batch_as_completed ¶

abatch 异步 ¶

abatch_as_completed 异步 ¶

stream ¶

astream 异步 ¶

astream_log 异步 ¶

astream_events 异步 ¶

transform ¶

`ChatOpenAI`¶

name `类属性` `实例属性` ¶

InputType `属性` ¶

OutputType `属性` ¶

input_schema `属性` ¶

output_schema `属性` ¶

config_specs `属性` ¶

cache `类属性` `实例属性` ¶

verbose `类属性` `实例属性` ¶

callbacks `类属性` `实例属性` ¶

tags `类属性` `实例属性` ¶

metadata `类属性` `实例属性` ¶

custom_get_token_ids `类属性` `实例属性` ¶

rate_limiter `类属性` `实例属性` ¶

disable_streaming `类属性` `实例属性` ¶

output_version `类属性` `实例属性` ¶

profile `属性` ¶

model_name `类属性` `实例属性` ¶

temperature `类属性` `实例属性` ¶

model_kwargs `类属性` `实例属性` ¶

openai_api_key `类属性` `实例属性` ¶

openai_api_base `类属性` `实例属性` ¶

openai_organization `类属性` `实例属性` ¶

request_timeout `类属性` `实例属性` ¶

stream_usage `类属性` `实例属性` ¶

max_retries `类属性` `实例属性` ¶

presence_penalty `类属性` `实例属性` ¶

frequency_penalty `类属性` `实例属性` ¶

seed `类属性` `实例属性` ¶

logprobs `类属性` `实例属性` ¶

top_logprobs `类属性` `实例属性` ¶

logit_bias `类属性` `实例属性` ¶

streaming `类属性` `实例属性` ¶

n `类属性` `实例属性` ¶

top_p `类属性` `实例属性` ¶

reasoning_effort `类属性` `实例属性` ¶

reasoning `类属性` `实例属性` ¶

verbosity `类属性` `实例属性` ¶

tiktoken_model_name `类属性` `实例属性` ¶

http_client `类属性` `实例属性` ¶

http_async_client `类属性` `实例属性` ¶

stop `类属性` `实例属性` ¶

extra_body `类属性` `实例属性` ¶

include_response_headers `类属性` `实例属性` ¶

disabled_params `类属性` `实例属性` ¶

include `类属性` `实例属性` ¶

service_tier `类属性` `实例属性` ¶

store `类属性` `实例属性` ¶

truncation `类属性` `实例属性` ¶

use_previous_response_id `类属性` `实例属性` ¶

use_responses_api `类属性` `实例属性` ¶

max_tokens `类属性` `实例属性` ¶

lc_secrets `属性` ¶

lc_attributes `属性` ¶

or ¶

ror ¶

ainvoke `异步` ¶

abatch `异步` ¶

abatch_as_completed `异步` ¶

astream `异步` ¶

astream_log `异步` ¶

astream_events `异步` ¶

atransform `异步` ¶

init ¶

lc_id `类方法` ¶

agenerate_prompt `异步` ¶

agenerate `异步` ¶

build_extra `类方法` ¶

validate_temperature `类方法` ¶

get_lc_namespace `类方法` ¶

is_lc_serializable `类方法` ¶