`langchain-groq`¶

注意

此软件包参考尚未完全迁移到 v1。

langchain_groq ¶

用于 LangChain 的 Groq 集成。

ChatGroq ¶

基类：BaseChatModel

Groq 聊天大型语言模型 API。

要使用该功能，您应该将环境变量 GROQ_API_KEY 设置为您的 API 密钥。

任何可以有效传递给 groq.create 调用的参数都可以传入，即使此类未明确保存这些参数。

设置

安装 langchain-groq 并设置环境变量 GROQ_API_KEY。

pip install -U langchain-groq
export GROQ_API_KEY="your-api-key"

关键初始化参数 — 完成参数： model：要使用的 Groq 模型名称，例如 llama-3.1-8b-instant。 temperature：采样温度。范围从 0.0 到 1.0。 max_tokens：生成的最大令牌数。 reasoning_format：推理输出的格式。如果未定义，Groq 将默认为 raw。

    - `'parsed'`: Separates reasoning into a dedicated field while keeping the
        response concise. Reasoning will be returned in the
        `additional_kwargs.reasoning_content` field of the response.
    - `'raw'`: Includes reasoning within think tags (e.g.
        `<think>{reasoning_content}</think>`).
    - `'hidden'`: Returns only the final answer content. Note: this only
        supresses reasoning content in the response; the model will still perform
        reasoning unless overridden in `reasoning_effort`.

    See the [Groq documentation](https://console.groq.com/docs/reasoning#reasoning)
    for more details and a list of supported models.
model_kwargs:
    Holds any model parameters valid for create call not
    explicitly specified.

关键初始化参数 — 客户端参数： timeout：请求超时时间。 max_retries：最大重试次数。 api_key：Groq API 密钥。如果未传入，将从环境变量 GROQ_API_KEY 中读取。 base_url：API 请求的基础 URL 路径，如果不使用代理或服务模拟器，则留空。 custom_get_token_ids：用于计算令牌的可选编码器。

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

实例化

from langchain_groq import ChatGroq

model = ChatGroq(
    model="llama-3.1-8b-instant",
    temperature=0.0,
    max_retries=2,
    # other params...
)

调用

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

AIMessage(content='The English sentence "I love programming" can
be translated to French as "J\'aime programmer". The word
"programming" is translated as "programmer" in French.',
response_metadata={'token_usage': {'completion_tokens': 38,
'prompt_tokens': 28, 'total_tokens': 66, 'completion_time':
0.057975474, 'prompt_time': 0.005366091, 'queue_time': None,
'total_time': 0.063341565}, 'model_name': 'llama-3.1-8b-instant',
'system_fingerprint': 'fp_c5f20b5bb1', 'finish_reason': 'stop',
'logprobs': None}, id='run-ecc71d70-e10c-4b69-8b8c-b8027d95d4b8-0')

流

# Streaming `text` for each content chunk received
for chunk in model.stream(messages):
    print(chunk.text, end="")

content='' id='run-4e9f926b-73f5-483b-8ef5-09533d925853'
content='The' id='run-4e9f926b-73f5-483b-8ef5-09533d925853'
content=' English' id='run-4e9f926b-73f5-483b-8ef5-09533d925853'
content=' sentence' id='run-4e9f926b-73f5-483b-8ef5-09533d925853'
...
content=' program' id='run-4e9f926b-73f5-483b-8ef5-09533d925853'
content='".' id='run-4e9f926b-73f5-483b-8ef5-09533d925853'
content='' response_metadata={'finish_reason': 'stop'}
id='run-4e9f926b-73f5-483b-8ef5-09533d925853

# Reconstructing a full response
stream = model.stream(messages)
full = next(stream)
for chunk in stream:
    full += chunk
full

AIMessageChunk(content='The English sentence "I love programming"
can be translated to French as "J\'aime programmer". Here\'s the
breakdown of the sentence: "J\'aime" is the French equivalent of "
I love", and "programmer" is the French infinitive for "to program".
So, the literal translation is "I love to program". However, in
English we often omit the "to" when talking about activities we
love, and the same applies to French. Therefore, "J\'aime
programmer" is the correct and natural way to express "I love
programming" in French.', response_metadata={'finish_reason':
'stop'}, id='run-a3c35ac4-0750-4d08-ac55-bfc63805de76')

异步

await model.ainvoke(messages)

AIMessage(content='The English sentence "I love programming" can
be translated to French as "J\'aime programmer". The word
"programming" is translated as "programmer" in French. I hope
this helps! Let me know if you have any other questions.',
response_metadata={'token_usage': {'completion_tokens': 53,
'prompt_tokens': 28, 'total_tokens': 81, 'completion_time':
0.083623752, 'prompt_time': 0.007365126, 'queue_time': None,
'total_time': 0.090988878}, 'model_name': 'llama-3.1-8b-instant',
'system_fingerprint': 'fp_c5f20b5bb1', 'finish_reason': 'stop',
'logprobs': None}, id='run-897f3391-1bea-42e2-82e0-686e2367bcf8-0')

工具调用

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("What is the population of NY?")
ai_msg.tool_calls

[
    {
        "name": "GetPopulation",
        "args": {"location": "NY"},
        "id": "call_bb8d",
    }
]

更多信息请参见 ChatGroq.bind_tools() 方法。

结构化输出

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 don't cats play poker in the jungle?",
    punchline="Too many cheetahs!",
    rating=None,
)

更多信息请参见 ChatGroq.with_structured_output() 方法。

响应元数据

ai_msg = model.invoke(messages)
ai_msg.response_metadata

{
    "token_usage": {
        "completion_tokens": 70,
        "prompt_tokens": 28,
        "total_tokens": 98,
        "completion_time": 0.111956391,
        "prompt_time": 0.007518279,
        "queue_time": None,
        "total_time": 0.11947467,
    },
    "model_name": "llama-3.1-8b-instant",
    "system_fingerprint": "fp_c5f20b5bb1",
    "finish_reason": "stop",
    "logprobs": None,
}

方法	描述
`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__`
`get_lc_namespace`	获取 LangChain 对象的命名空间。
`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`	返回文本中 token 的有序 ID。
`get_num_tokens`	获取文本中存在的 token 数量。
`get_num_tokens_from_messages`	获取消息中的 token 数量。
`generate`	将一系列提示传递给模型并返回模型生成的内容。
`agenerate`	异步地将一系列提示传递给模型并返回生成的内容。
`dict`	返回 LLM 的字典。
`build_extra`	根据传入的额外参数构建额外的关键字参数（kwargs）。
`validate_environment`	验证 API 密钥和 Python 包是否存在于环境中。
`is_lc_serializable`	返回此模型是否可以被 LangChain 序列化。
`bind_tools`	将类工具（tool-like）对象绑定到此聊天模型。
`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 的可配置字段。

lc_attributes `property` ¶

lc_attributes: dict

应包含在序列化 kwargs 中的属性名称列表。

这些属性必须被构造函数接受。

默认为空字典。

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.content_blocks 将懒解析 content 的内容为标准格式。此标志可用于额外将标准格式存储在消息内容中，例如，用于序列化目的。

支持的值

'v0'：内容中特定于提供商的格式（可以使用 content_blocks 进行懒解析）
'v1'：内容中的标准化格式（与 content_blocks 一致）

合作伙伴包（例如 langchain-openai）也可以使用此字段以向后兼容的方式推出新的内容格式。

在版本 1.0 中添加

profile `property` ¶

profile: ModelProfile

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

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

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

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

model_name `class-attribute` `instance-attribute` ¶

model_name: str = Field(alias='model')

要使用的模型名称。

temperature `class-attribute` `instance-attribute` ¶

temperature: float = 0.7

要使用的采样温度。

stop `class-attribute` `instance-attribute` ¶

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

默认的停止序列。

reasoning_format `class-attribute` `instance-attribute` ¶

reasoning_format: Literal['parsed', 'raw', 'hidden'] | None = Field(default=None)

推理输出的格式。如果未定义，Groq 将默认为 raw。

'parsed'：将推理分离到一个专用字段，同时保持响应简洁。推理将返回在响应的 additional_kwargs.reasoning_content 字段中。
'raw'：在思考标签（例如 <think>{reasoning_content}</think>）内包含推理。
'hidden'：仅返回最终答案内容。注意：这只会在响应中抑制推理内容；除非在 reasoning_effort 中覆盖，否则模型仍将执行推理。

有关更多详细信息和支持的模型列表，请参阅 Groq 文档。

reasoning_effort `class-attribute` `instance-attribute` ¶

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

模型在推理上投入的努力程度。如果未定义，Groq 将默认启用推理。

有关更多详细信息以及支持设置推理努力的选项和模型列表，请参阅 Groq 文档。

model_kwargs `class-attribute` `instance-attribute` ¶

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

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

groq_api_key `class-attribute` `instance-attribute` ¶

groq_api_key: SecretStr | None = Field(
    alias="api_key", default_factory=secret_from_env("GROQ_API_KEY", default=None)
)

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

groq_api_base `class-attribute` `instance-attribute` ¶

groq_api_base: str | None = Field(
    alias="base_url", default_factory=from_env("GROQ_API_BASE", default=None)
)

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

request_timeout `class-attribute` `instance-attribute` ¶

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

向 Groq 完成 API 发出请求的超时时间。可以是浮点数、httpx.Timeout 或 None。

max_retries `class-attribute` `instance-attribute` ¶

max_retries: int = 2

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

streaming `class-attribute` `instance-attribute` ¶

streaming: bool = False

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

n `class-attribute` `instance-attribute` ¶

n: int = 1

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

max_tokens `class-attribute` `instance-attribute` ¶

max_tokens: int | None = None

生成的最大令牌数。

service_tier `class-attribute` `instance-attribute` ¶

service_tier: Literal['on_demand', 'flex', 'auto'] = Field(default='on_demand')

可选参数，您可以包含此参数以指定您希望用于请求的服务层级。

'on_demand'：默认值。
'flex'：当容量可用时进行按需处理，如果资源受限则快速超时。为不需要保证处理的工作负载提供了性能和可靠性之间的平衡。
'auto'：使用按需速率限制，如果超过这些限制，则回退到 'flex'

有关更多详细信息以及服务层级及其描述的列表，请参阅 Groq 文档。

http_client `class-attribute` `instance-attribute` ¶

http_client: Any | None = None

可选的 httpx.Client。

http_async_client `class-attribute` `instance-attribute` ¶

http_async_client: Any | None = None

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

lc_secrets `property` ¶

lc_secrets: dict[str, str]

机密环境变量的映射。

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

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 对象包含有关运行的信息，包括其 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

get_lc_namespace `classmethod` ¶

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

例如，如果类是 langchain.llms.openai.OpenAI，那么命名空间是 ["langchain", "llms", "openai"]

返回	描述
`list[str]`	命名空间。

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

当你想要

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

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

当你想要

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

参数	描述
`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]

返回文本中 token 的有序 ID。

参数	描述
`text`	要进行分词的字符串输入。类型： `str`

返回	描述
`list[int]`	与文本中的词元相对应的 ID 列表，按它们在文本中出现的顺序列出。

get_num_tokens ¶

get_num_tokens(text: str) -> int

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

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

参数	描述
`text`	要进行分词的字符串输入。类型： `str`

返回	描述
`int`	文本中的词元整数数量。

get_num_tokens_from_messages ¶

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

获取消息中的 token 数量。

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

注意

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

参数	描述
`messages`	要进行分词的消息输入。类型： `list[BaseMessage]`
`工具`	如果提供，则为要转换为工具模式的 dict、`BaseModel`、函数或 `BaseTool` 对象的序列。类型： `Sequence \| None` 默认值： `None`

返回	描述
`int`	所有消息中词元数量的总和。

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

当你想要

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

参数	描述
`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 的字典。

build_extra `classmethod` ¶

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

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

validate_environment ¶

validate_environment() -> Self

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

is_lc_serializable `classmethod` ¶

is_lc_serializable() -> bool

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

bind_tools ¶

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

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

参数	描述
`工具`	要绑定到此聊天模型的工具定义列表。支持 `langchain_core.utils.function_calling.convert_to_openai_tool` 处理的任何工具定义。类型： `Sequence[dict[str, Any] \| type[BaseModel] \| Callable \| BaseTool]`
`工具选择`	指定要求模型调用的工具。必须是提供的单个函数的名称，`'auto'` 表示自动确定调用哪个函数（可以选择不调用任何函数），`'any'` 表示强制调用某个函数，或是一个形如 `{"type": "function", "function": {"name": <<tool_name>>}}` 的字典。类型： `dict \| str \| bool \| None` 默认值： `None`
`**kwargs`	任何要传递给 `langchain.runnable.Runnable` 构造函数的额外参数。类型： `Any` 默认值： `{}`

with_structured_output ¶

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

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

参数	描述
`模式`	输出模式。可以作为以下形式传入：一个 OpenAI 函数/工具模式，一个 JSON Schema，一个 `TypedDict` 类，或一个 Pydantic 类。如果 `schema` 是一个 Pydantic 类，那么模型输出将是该类的 Pydantic 实例，并且模型生成的字段将由 Pydantic 类进行验证。否则，模型输出将是一个字典，并且不会被验证。有关在指定 Pydantic 或 `TypedDict` 类时如何正确指定模式字段的类型和描述的更多信息，请参阅 `langchain_core.utils.function_calling.convert_to_openai_tool`。行为在 0.3.8 版本中发生变化通过 `method="json_schema"` 添加了对 Groq 专用结构化输出功能的支持。类型： `dict \| type[BaseModel] \| None` 默认值： `None`
`method`	用于引导模型生成的方法，可选值之一： `'function_calling'`：使用 Groq 的工具调用 API `'json_schema'`：使用 Groq 的结构化输出 API。支持部分模型，包括 `openai/gpt-oss`、`moonshotai/kimi-k2-instruct-0905` 和一些 `meta-llama/llama-4` 模型。详情请参阅文档。 `'json_mode'`：使用 Groq 的JSON 模式。请注意，如果使用 JSON 模式，您必须在模型调用中包含将输出格式化为所需模式的指令。在此处了解更多关于这些方法之间的差异以及哪些模型支持哪些方法的信息。类型： `Literal['function_calling', 'json_mode', 'json_schema']` 默认值： `'function_calling'`
`method`	引导模型生成的方法，可以是 `'function_calling'` 或 `'json_mode'`。如果为 `'function_calling'`，则模式将被转换为 OpenAI 函数，并且返回的模型将使用函数调用 API。如果为 `'json_mode'`，则将使用 JSON 模式。注意如果使用 `'json_mode'`，您必须在模型调用中包含将输出格式化为所需模式的指令（可以通过提示本身或在系统消息/提示/指令中）。警告 `'json_mode'` 不支持流式响应的停止序列。类型： `Literal['function_calling', 'json_mode', 'json_schema']` 默认值： `'function_calling'`
`包含原始数据`	如果为 `False`，则仅返回解析后的结构化输出。如果在模型输出解析期间发生错误，它将被引发。如果为 `True`，则将返回原始模型响应（一个 `BaseMessage`）和解析后的模型响应。如果在输出解析期间发生错误，它将被捕获并一并返回。最终输出总是一个带有键 `'raw'`、`'parsed'` 和 `'parsing_error'` 的 `dict`。类型： `bool` 默认值： `False`
`kwargs`	任何要传递给 `langchain.runnable.Runnable` 构造函数的额外参数。类型： `Any` 默认值： `{}`

返回描述

Runnable[LanguageModelInput, dict | BaseModel]

一个接受与 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

示例：schema=Pydantic 类, method="function_calling", include_raw=False

from typing import Optional

from langchain_groq import ChatGroq
from pydantic import BaseModel, Field


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

    answer: str
    # If we provide default values and/or descriptions for fields, these will be passed
    # to the model. This is an important part of improving a model's ability to
    # correctly return structured outputs.
    justification: str | None = Field(default=None, description="A justification for the answer.")


model = ChatGroq(model="openai/gpt-oss-120b", 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=True

from langchain_groq import ChatGroq
from pydantic import BaseModel


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

    answer: str
    justification: str


model = ChatGroq(model="openai/gpt-oss-120b", 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="function_calling", include_raw=False

from typing_extensions import Annotated, TypedDict

from langchain_groq import ChatGroq


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 = ChatGroq(model="openai/gpt-oss-120b", 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="function_calling", include_raw=False

from langchain_groq import ChatGroq

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 = ChatGroq(model="openai/gpt-oss-120b", 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_schema"，include_raw=False

from typing import Optional

from langchain_groq import ChatGroq
from pydantic import BaseModel, Field


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

    answer: str
    # If we provide default values and/or descriptions for fields, these will be passed
    # to the model. This is an important part of improving a model's ability to
    # correctly return structured outputs.
    justification: str | None = Field(default=None, description="A justification for the answer.")


model = ChatGroq(model="openai/gpt-oss-120b", temperature=0)
structured_model = model.with_structured_output(
    AnswerWithJustification,
    method="json_schema",
)

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_mode", include_raw=True

from langchain_groq import ChatGroq
from pydantic import BaseModel


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


model = ChatGroq(model="openai/gpt-oss-120b", 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
# }

langchain-groq¶

langchain_groq ¶

ChatGroq ¶

name class-attribute instance-attribute ¶

InputType property ¶

OutputType property ¶

input_schema property ¶

output_schema property ¶

config_specs property ¶

lc_attributes property ¶

cache class-attribute instance-attribute ¶

verbose class-attribute instance-attribute ¶

callbacks class-attribute instance-attribute ¶

tags class-attribute instance-attribute ¶

metadata class-attribute instance-attribute ¶

custom_get_token_ids class-attribute instance-attribute ¶

rate_limiter class-attribute instance-attribute ¶

disable_streaming class-attribute instance-attribute ¶

output_version class-attribute instance-attribute ¶

profile property ¶

model_name class-attribute instance-attribute ¶

temperature class-attribute instance-attribute ¶

stop class-attribute instance-attribute ¶

reasoning_format class-attribute instance-attribute ¶

reasoning_effort class-attribute instance-attribute ¶

model_kwargs class-attribute instance-attribute ¶

groq_api_key class-attribute instance-attribute ¶

groq_api_base class-attribute instance-attribute ¶

request_timeout class-attribute instance-attribute ¶

max_retries class-attribute instance-attribute ¶

streaming class-attribute instance-attribute ¶

n class-attribute instance-attribute ¶

max_tokens class-attribute instance-attribute ¶

service_tier class-attribute instance-attribute ¶

http_client class-attribute instance-attribute ¶

http_async_client class-attribute instance-attribute ¶

lc_secrets property ¶

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 async ¶

batch ¶

batch_as_completed ¶

abatch async ¶

abatch_as_completed async ¶

stream ¶

astream async ¶

astream_log async ¶

astream_events async ¶

transform ¶

atransform async ¶

bind ¶

with_config ¶

with_listeners ¶

with_alisteners ¶

with_types ¶

with_retry ¶

map ¶

with_fallbacks ¶

as_tool ¶

__init__ ¶

get_lc_namespace classmethod ¶

lc_id classmethod ¶

to_json ¶

to_json_not_implemented ¶

configurable_fields ¶

configurable_alternatives ¶

set_verbose ¶

`langchain-groq`¶

name `class-attribute` `instance-attribute` ¶

InputType `property` ¶

OutputType `property` ¶

input_schema `property` ¶

output_schema `property` ¶

config_specs `property` ¶

lc_attributes `property` ¶

cache `class-attribute` `instance-attribute` ¶

verbose `class-attribute` `instance-attribute` ¶

callbacks `class-attribute` `instance-attribute` ¶

tags `class-attribute` `instance-attribute` ¶

metadata `class-attribute` `instance-attribute` ¶

custom_get_token_ids `class-attribute` `instance-attribute` ¶

rate_limiter `class-attribute` `instance-attribute` ¶

disable_streaming `class-attribute` `instance-attribute` ¶

output_version `class-attribute` `instance-attribute` ¶

profile `property` ¶

model_name `class-attribute` `instance-attribute` ¶

temperature `class-attribute` `instance-attribute` ¶

stop `class-attribute` `instance-attribute` ¶

reasoning_format `class-attribute` `instance-attribute` ¶

reasoning_effort `class-attribute` `instance-attribute` ¶

model_kwargs `class-attribute` `instance-attribute` ¶

groq_api_key `class-attribute` `instance-attribute` ¶

groq_api_base `class-attribute` `instance-attribute` ¶

request_timeout `class-attribute` `instance-attribute` ¶

max_retries `class-attribute` `instance-attribute` ¶

streaming `class-attribute` `instance-attribute` ¶

n `class-attribute` `instance-attribute` ¶

max_tokens `class-attribute` `instance-attribute` ¶

service_tier `class-attribute` `instance-attribute` ¶

http_client `class-attribute` `instance-attribute` ¶

http_async_client `class-attribute` `instance-attribute` ¶

lc_secrets `property` ¶

or ¶

ror ¶

ainvoke `async` ¶

abatch `async` ¶

abatch_as_completed `async` ¶

astream `async` ¶

astream_log `async` ¶

astream_events `async` ¶

atransform `async` ¶

init ¶

get_lc_namespace `classmethod` ¶

lc_id `classmethod` ¶

agenerate_prompt `async` ¶

agenerate `async` ¶

build_extra `classmethod` ¶

is_lc_serializable `classmethod` ¶