`langchain-ollama`¶

注意

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

langchain_ollama ¶

这是 langchain_ollama 包。

提供与 Ollama 服务交互的基础设施。

注意

在 0.3.4 版本中新增： 所有模型上的 validate_model_on_init 参数。此参数允许您在初始化时验证模型是否存在于本地 Ollama 中。如果设置为 True，当模型在本地不存在时，它将引发错误。这对于确保模型在尝试使用前可用非常有用，尤其是在模型可能未预先下载的环境中。

ChatOllama ¶

基类：BaseChatModel

Ollama 聊天模型集成。

设置

安装 langchain-ollama 并从 ollama 下载您想要使用的任何模型。

ollama pull gpt-oss:20b
pip install -U langchain-ollama

关键初始化参数 — 完成参数：model: str 要使用的 Ollama 模型名称。reasoning: bool | None 控制支持的模型的推理/思考模式。

    - `True`: Enables reasoning mode. The model's reasoning process will be
        captured and returned separately in the `additional_kwargs` of the
        response message, under `reasoning_content`. The main response
        content will not include the reasoning tags.
    - `False`: Disables reasoning mode. The model will not perform any reasoning,
        and the response will not include any reasoning content.
    - `None` (Default): The model will use its default reasoning behavior. Note
        however, if the model's default behavior *is* to perform reasoning, think tags
        (`<think>` and `</think>`) will be present within the main response content
        unless you set `reasoning` to `True`.
temperature: float
    Sampling temperature. Ranges from `0.0` to `1.0`.
num_predict: int | None
    Max number of tokens to generate.

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

实例化

from langchain_ollama import ChatOllama

model = ChatOllama(
    model="gpt-oss:20b",
    validate_model_on_init=True,
    temperature=0.8,
    num_predict=256,
    # other params ...
)

调用

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

AIMessage(content='J'adore le programmation. (Note: "programming" can also refer to the act of writing code, so if you meant that, I could translate it as "J'adore programmer". But since you didn\'t specify, I assumed you were talking about the activity itself, which is what "le programmation" usually refers to.)', response_metadata={'model': 'llama3', 'created_at': '2024-07-04T03:37:50.182604Z', 'message': {'role': 'assistant', 'content': ''}, 'done_reason': 'stop', 'done': True, 'total_duration': 3576619666, 'load_duration': 788524916, 'prompt_eval_count': 32, 'prompt_eval_duration': 128125000, 'eval_count': 71, 'eval_duration': 2656556000}, id='run-ba48f958-6402-41a5-b461-5e250a4ebd36-0')

流

for chunk in model.stream("Return the words Hello World!"):
    print(chunk.text, end="")

content='Hello' id='run-327ff5ad-45c8-49fe-965c-0a93982e9be1'
content=' World' id='run-327ff5ad-45c8-49fe-965c-0a93982e9be1'
content='!' id='run-327ff5ad-45c8-49fe-965c-0a93982e9be1'
content='' response_metadata={'model': 'llama3', 'created_at': '2024-07-04T03:39:42.274449Z', 'message': {'role': 'assistant', 'content': ''}, 'done_reason': 'stop', 'done': True, 'total_duration': 411875125, 'load_duration': 1898166, 'prompt_eval_count': 14, 'prompt_eval_duration': 297320000, 'eval_count': 4, 'eval_duration': 111099000} id='run-327ff5ad-45c8-49fe-965c-0a93982e9be1'

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

AIMessageChunk(
    content='Je adore le programmation.(Note: "programmation" is the formal way to say "programming" in French, but informally, people might use the phrase "le développement logiciel" or simply "le code")',
    response_metadata={
        "model": "llama3",
        "created_at": "2024-07-04T03:38:54.933154Z",
        "message": {"role": "assistant", "content": ""},
        "done_reason": "stop",
        "done": True,
        "total_duration": 1977300042,
        "load_duration": 1345709,
        "prompt_eval_duration": 159343000,
        "eval_count": 47,
        "eval_duration": 1815123000,
    },
    id="run-3c81a3ed-3e79-4dd3-a796-04064d804890",
)

异步

await model.ainvoke("Hello how are you!")

AIMessage(
    content="Hi there! I'm just an AI, so I don't have feelings or emotions like humans do. But I'm functioning properly and ready to help with any questions or tasks you may have! How can I assist you today?",
    response_metadata={
        "model": "llama3",
        "created_at": "2024-07-04T03:52:08.165478Z",
        "message": {"role": "assistant", "content": ""},
        "done_reason": "stop",
        "done": True,
        "total_duration": 2138492875,
        "load_duration": 1364000,
        "prompt_eval_count": 10,
        "prompt_eval_duration": 297081000,
        "eval_count": 47,
        "eval_duration": 1838524000,
    },
    id="run-29c510ae-49a4-4cdd-8f23-b972bfab1c49-0",
)

async for chunk in model.astream("Say hello world!"):
    print(chunk.content)

HEL
LO
WORLD
!

messages = [("human", "Say hello world!"), ("human", "Say goodbye world!")]
await model.abatch(messages)

[
    AIMessage(
        content="HELLO, WORLD!",
        response_metadata={
            "model": "llama3",
            "created_at": "2024-07-04T03:55:07.315396Z",
            "message": {"role": "assistant", "content": ""},
            "done_reason": "stop",
            "done": True,
            "total_duration": 1696745458,
            "load_duration": 1505000,
            "prompt_eval_count": 8,
            "prompt_eval_duration": 111627000,
            "eval_count": 6,
            "eval_duration": 185181000,
        },
        id="run-da6c7562-e25a-4a44-987a-2c83cd8c2686-0",
    ),
    AIMessage(
        content="It's been a blast chatting with you! Say goodbye to the world for me, and don't forget to come back and visit us again soon!",
        response_metadata={
            "model": "llama3",
            "created_at": "2024-07-04T03:55:07.018076Z",
            "message": {"role": "assistant", "content": ""},
            "done_reason": "stop",
            "done": True,
            "total_duration": 1399391083,
            "load_duration": 1187417,
            "prompt_eval_count": 20,
            "prompt_eval_duration": 230349000,
            "eval_count": 31,
            "eval_duration": 1166047000,
        },
        id="run-96cad530-6f3e-4cf9-86b4-e0f8abba4cdb-0",
    ),
]

JSON 模式

json_model = ChatOllama(format="json")
json_model.invoke(
    "Return a query for the weather in a random location and time of day with two keys: location and time_of_day. "
    "Respond using JSON only."
).content

'{"location": "Pune, India", "time_of_day": "morning"}'

工具调用

from langchain_ollama import ChatOllama
from pydantic import BaseModel, Field


class Multiply(BaseModel):
    a: int = Field(..., description="First integer")
    b: int = Field(..., description="Second integer")


ans = await chat.invoke("What is 45*67")
ans.tool_calls

[
    {
        "name": "Multiply",
        "args": {"a": 45, "b": 67},
        "id": "420c3f3b-df10-4188-945f-eb3abdb40622",
        "type": "tool_call",
    }
]

思考/推理：您可以通过在构造函数或 invoke/stream 方法中将 reasoning 参数设置为 True 来为支持它的模型启用推理模式。这将使模型能够思考问题，并在响应消息的 additional_kwargs 中，在 reasoning_content 下单独返回推理过程。

If `reasoning` is set to `None`, the model will use its default reasoning
behavior, and any reasoning content will *not* be captured under the
`reasoning_content` key, but will be present within the main response content
as think tags (`<think>` and `</think>`).

!!! note
    This feature is only available for [models that support reasoning](https://ollama.ac.cn/search?c=thinking).

```python
from langchain_ollama import ChatOllama

model = ChatOllama(
    model="deepseek-r1:8b",
    validate_model_on_init=True,
    reasoning=True,
)

model.invoke("how many r in the word strawberry?")

# or, on an invocation basis:

model.invoke("how many r in the word strawberry?", reasoning=True)
# or model.stream("how many r in the word strawberry?", reasoning=True)

# If not provided, the invocation will default to the ChatOllama reasoning
# param provided (None by default).
```

```python
AIMessage(content='The word "strawberry" contains **three \'r\' letters**. Here\'s a breakdown for clarity:\n\n- The spelling of "strawberry" has two parts ... be 3.\n\nTo be thorough, let\'s confirm with an online source or common knowledge.\n\nI can recall that "strawberry" has: s-t-r-a-w-b-e-r-r-y — yes, three r\'s.\n\nPerhaps it\'s misspelled by some, but standard is correct.\n\nSo I think the response should be 3.\n'}, response_metadata={'model': 'deepseek-r1:8b', 'created_at': '2025-07-08T19:33:55.891269Z', 'done': True, 'done_reason': 'stop', 'total_duration': 98232561292, 'load_duration': 28036792, 'prompt_eval_count': 10, 'prompt_eval_duration': 40171834, 'eval_count': 3615, 'eval_duration': 98163832416, 'model_name': 'deepseek-r1:8b'}, id='run--18f8269f-6a35-4a7c-826d-b89d52c753b3-0', usage_metadata={'input_tokens': 10, 'output_tokens': 3615, 'total_tokens': 3625})

```

方法	描述
`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__`
`is_lc_serializable`	这个类是否可序列化？
`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 的字典。
`bind_tools`	将类工具（tool-like）对象绑定到此聊天模型。
`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 的可配置字段。

lc_secrets `属性` ¶

lc_secrets: dict[str, str]

构造函数参数名称到密钥 ID 的映射。

例如，{"openai_api_key": "OPENAI_API_KEY"}

lc_attributes `属性` ¶

lc_attributes: dict

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

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

默认为空字典。

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

支持的值

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

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

在版本 1.0 中添加

profile `属性` ¶

profile: ModelProfile

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

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

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

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

model `实例属性` ¶

model: str

要使用的模型名称。

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

reasoning: bool | str | None = None

控制支持的模型的推理/思考模式。

True：启用推理模式。模型的推理过程将被捕获并单独返回到响应消息的 additional_kwargs 中，位于 reasoning_content 下。主要响应内容将不包含推理标签。
False：禁用推理模式。模型将不执行任何推理，响应中也不会包含任何推理内容。
None（默认）：模型将使用其默认的推理行为。但请注意，如果模型的默认行为是执行推理，则思考标签（<think> 和 </think>）将出现在主要响应内容中，除非您将 reasoning 设置为 True。
str：例如 'low'、'medium'、'high'。以自定义强度级别启用推理。目前，这仅受 gpt-oss 支持。有关更多信息，请参阅 Ollama 文档。

validate_model_on_init `类属性` `实例属性` ¶

validate_model_on_init: bool = False

是否在初始化时验证模型是否存在于本地 Ollama 中。

在版本 0.3.4 中添加

mirostat `类属性` `实例属性` ¶

mirostat: int | None = None

启用 Mirostat 采样以控制困惑度。

（默认值：0，0 = 禁用，1 = Mirostat，2 = Mirostat 2.0）

mirostat_eta `类属性` `实例属性` ¶

mirostat_eta: float | None = None

影响算法对生成文本反馈的响应速度。

较低的学习率将导致较慢的调整，而较高的学习率将使算法更具响应性。

（默认值：0.1）

mirostat_tau `类属性` `实例属性` ¶

mirostat_tau: float | None = None

控制输出的连贯性和多样性之间的平衡。

较低的值将导致更集中和连贯的文本。

（默认值：5.0）

num_ctx `类属性` `实例属性` ¶

num_ctx: int | None = None

设置用于生成下一个词元（token）的上下文窗口大小。

（默认值：2048）

num_gpu `类属性` `实例属性` ¶

num_gpu: int | None = None

要使用的 GPU 数量。

在 macOS 上，默认为 1 以启用 Metal 支持，0 表示禁用。

num_thread `类属性` `实例属性` ¶

num_thread: int | None = None

设置计算期间要使用的线程数。

默认情况下，Ollama 会自动检测此值以获得最佳性能。建议将此值设置为您系统的物理 CPU 核心数（而不是逻辑核心数）。

num_predict `类属性` `实例属性` ¶

num_predict: int | None = None

生成文本时预测的最大词元（token）数。

（默认值：128，-1 = 无限生成，-2 = 填充上下文）

repeat_last_n `类属性` `实例属性` ¶

repeat_last_n: int | None = None

设置模型回顾以防止重复的范围。

（默认值：64，0 = 禁用，-1 = num_ctx）

repeat_penalty `类属性` `实例属性` ¶

repeat_penalty: float | None = None

设置对重复的惩罚强度。

较高的值（例如 1.5）将更强烈地惩罚重复，而较低的值（例如 0.9）将更宽松。（默认值：1.1）

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

temperature: float | None = None

模型的温度。

增加温度将使模型的回答更具创造性。

（默认值：0.8）

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

seed: int | None = None

设置用于生成的随机数种子。

将此设置为特定数字将使模型针对相同的提示生成相同的文本。

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

stop: list[str] | None = None

设置要使用的停止词元（stop token）。

tfs_z `类属性` `实例属性` ¶

tfs_z: float | None = None

尾部自由采样（Tail free sampling）。

用于减少输出中概率较低的词元（token）的影响。

较高的值（例如 2.0）将更多地减少影响，而 1.0 的值会禁用此设置。

（默认值：1）

top_k `类属性` `实例属性` ¶

top_k: int | None = None

减少生成无意义内容的概率。

较高的值（例如 100）将提供更多样化的答案，而较低的值（例如 10）将更为保守。

（默认值：40）

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

top_p: float | None = None

与 top-k 一起工作。

较高的值（例如 0.95）将导致更多样化的文本，而较低的值（例如 0.5）将生成更集中和保守的文本。

（默认值：0.9）

format `类属性` `实例属性` ¶

format: Literal['', 'json'] | JsonSchemaValue | None = None

指定输出格式（选项：'json'、JSON schema）。

keep_alive `类属性` `实例属性` ¶

keep_alive: int | str | None = None

模型将保持加载到内存中的时间。

base_url `类属性` `实例属性` ¶

base_url: str | None = None

模型托管的基础 URL。

如果没有，则默认为 Ollama 客户端的默认值。

支持 http://username:password@localhost:11434 格式的 userinfo 认证。如果您的 Ollama 服务器位于代理之后，这将非常有用。

警告

userinfo 不安全，只应用于本地测试或安全环境中。避免在生产环境或不安全的网络上使用它。

注意

如果使用 userinfo，请确保 Ollama 服务器已配置为接受并验证这些凭据。

注意

userinfo 标头会传递给同步和异步客户端。

client_kwargs `类属性` `实例属性` ¶

client_kwargs: dict | None = {}

传递给 httpx 客户端的额外关键字参数。在此处传递标头。

这些参数会传递给同步和异步客户端。

使用 sync_client_kwargs 和 async_client_kwargs 来为同步和异步客户端传递不同的参数。

async_client_kwargs `类属性` `实例属性` ¶

async_client_kwargs: dict | None = {}

在传递给 httpx 客户端之前，与 client_kwargs 合并的额外关键字参数。

这些是异步客户端独有的客户端参数；对于共享参数，请使用 client_kwargs。

有关参数的完整列表，请参阅 httpx 文档。

sync_client_kwargs `类属性` `实例属性` ¶

sync_client_kwargs: dict | None = {}

在传递给 httpx 客户端之前，与 client_kwargs 合并的额外关键字参数。

这些是同步客户端独有的客户端参数；对于共享参数，请使用 client_kwargs。

有关参数的完整列表，请参阅 httpx 文档。

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

is_lc_serializable `类方法` ¶

is_lc_serializable() -> bool

这个类是否可序列化？

根据设计，即使一个类继承自 Serializable，它默认也是不可序列化的。这是为了防止意外序列化不应被序列化的对象。

返回	描述
`bool`	类是否可序列化。默认为 `False`。

get_lc_namespace `类方法` ¶

get_lc_namespace() -> list[str]

获取 LangChain 对象的命名空间。

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

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

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]

返回文本中 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 `异步` ¶

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 | Literal["auto", "any"] | 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]`
`工具选择`	如果提供，模型将调用哪个工具。此参数当前被忽略，因为 Ollama 不支持。类型： `dict \| str \| Literal['auto', 'any'] \| bool \| None` 默认值： `None`
`kwargs`	任何其他参数都直接传递给 `self.bind(kwargs)`。类型： `Any` 默认值：** `{}`

with_structured_output ¶

with_structured_output(
    schema: dict | type,
    *,
    method: Literal["function_calling", "json_mode", "json_schema"] = "json_schema",
    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`。类型： `dict \| type`
`method`	用于引导模型生成的方法，可选值之一： `'json_schema'`：使用 Ollama 的结构化输出 API `'function_calling'`：使用 Ollama 的工具调用 API `'json_mode'`：指定 `format='json'`。请注意，如果使用 JSON 模式，则必须在模型调用中包含将输出格式化为所需模式的指令。类型： `Literal['function_calling', 'json_mode', 'json_schema']` 默认值： `'json_schema'`
`包含原始数据`	如果为 `False`，则仅返回解析后的结构化输出。如果在模型输出解析期间发生错误，它将被引发。如果为 `True`，则将返回原始模型响应（一个 `BaseMessage`）和解析后的模型响应。如果在输出解析期间发生错误，它将被捕获并一并返回。最终输出总是一个带有键 `'raw'`、`'parsed'` 和 `'parsing_error'` 的 `dict`。类型： `bool` 默认值： `False`
`kwargs`	不支持额外的关键字参数。类型： `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

行为在 0.2.2 版本中发生变化

通过 format 参数增加了对结构化输出 API 的支持。

行为在 0.3.0 版本中发生变化

将默认的 method 更新为 'json_schema'。

示例：schema=Pydantic 类，method='json_schema'，include_raw=False

from typing import Optional

from langchain_ollama import ChatOllama
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 = ChatOllama(model="llama3.1", 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='json_schema'，include_raw=True

from langchain_ollama import ChatOllama
from pydantic import BaseModel


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

    answer: str
    justification: str


model = ChatOllama(model="llama3.1", 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=Pydantic 类，method='function_calling'，include_raw=False

from typing import Optional

from langchain_ollama import ChatOllama
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 = ChatOllama(model="llama3.1", 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=TypedDict 类，method='function_calling'，include_raw=False

from typing_extensions import Annotated, TypedDict

from langchain_ollama import ChatOllama


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 = ChatOllama(model="llama3.1", 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_ollama import ChatOllama

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 = ChatOllama(model="llama3.1", 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_ollama import ChatOllama
from pydantic import BaseModel


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


model = ChatOllama(model="llama3.1", 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
# }

OllamaEmbeddings ¶

基类：BaseModel, Embeddings

Ollama 嵌入模型集成。

设置本地 Ollama 实例

安装 Ollama 包并设置本地 Ollama 实例。

您需要选择一个模型来提供服务。

您可以通过模型库查看可用模型列表。

要从 Ollama 模型库获取模型，请使用 ollama pull <name-of-model>。

例如，要拉取 llama3 模型

ollama pull llama3

这将下载模型的默认标记版本。通常，默认指向最新、最小参数大小的模型。

在 Mac 上，模型将下载到 ~/.ollama/models
在 Linux（或 WSL）上，模型将存储在 /usr/share/ollama/.ollama/models

您可以指定感兴趣的模型的具体版本，例如 ollama pull vicuna:13b-v1.5-16k-q4_0。

查看已拉取的模型

ollama list

开始提供服务

ollama serve

查看 Ollama 文档以获取更多命令。

ollama help

安装 langchain-ollama 集成包

pip install -U langchain_ollama

关键初始化参数 — 完成参数：model: str 要使用的 Ollama 模型名称。base_url: str | None 模型托管的基础 URL。

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

实例化

from langchain_ollama import OllamaEmbeddings

embed = OllamaEmbeddings(model="llama3")

嵌入单个文本

input_text = "The meaning of life is 42"
vector = embed.embed_query(input_text)
print(vector[:3])

[-0.024603435769677162, -0.007543657906353474, 0.0039630369283258915]

嵌入多个文本

input_texts = ["Document 1...", "Document 2..."]
vectors = embed.embed_documents(input_texts)
print(len(vectors))
# The first 3 coordinates for the first vector
print(vectors[0][:3])

2
[-0.024603435769677162, -0.007543657906353474, 0.0039630369283258915]

异步

vector = await embed.aembed_query(input_text)
print(vector[:3])

# multiple:
# await embed.aembed_documents(input_texts)

[-0.009100092574954033, 0.005071679595857859, -0.0029193938244134188]

方法	描述
`embed_documents`	嵌入搜索文档。
`embed_query`	嵌入查询文本。
`aembed_documents`	嵌入搜索文档。
`aembed_query`	嵌入查询文本。

model `实例属性` ¶

model: str

要使用的模型名称。

validate_model_on_init `类属性` `实例属性` ¶

validate_model_on_init: bool = False

是否在初始化时验证模型是否存在于本地 ollama 中。

在版本 0.3.4 中添加

base_url `类属性` `实例属性` ¶

base_url: str | None = None

模型托管的基础 URL。

如果没有，则默认为 Ollama 客户端的默认值。

支持 http://username:password@localhost:11434 格式的 userinfo 认证。如果您的 Ollama 服务器位于代理之后，这将非常有用。

警告

userinfo 不安全，只应用于本地测试或安全环境中。避免在生产环境或不安全的网络上使用它。

注意

如果使用 userinfo，请确保 Ollama 服务器已配置为接受并验证这些凭据。

注意

userinfo 标头会传递给同步和异步客户端。

client_kwargs `class-attribute` `instance-attribute` ¶

client_kwargs: dict | None = {}

传递给 httpx 客户端的额外关键字参数。在此处传递标头。

这些参数会传递给同步和异步客户端。

使用 sync_client_kwargs 和 async_client_kwargs 来为同步和异步客户端传递不同的参数。

async_client_kwargs `class-attribute` `instance-attribute` ¶

async_client_kwargs: dict | None = {}

在传递给 httpx 客户端之前，与 client_kwargs 合并的额外关键字参数。

这些是异步客户端独有的客户端参数；对于共享参数，请使用 client_kwargs。

有关参数的完整列表，请参阅 httpx 文档。

sync_client_kwargs `class-attribute` `instance-attribute` ¶

sync_client_kwargs: dict | None = {}

在传递给 httpx 客户端之前，与 client_kwargs 合并的额外关键字参数。

这些是同步客户端独有的客户端参数；对于共享参数，请使用 client_kwargs。

有关参数的完整列表，请参阅 httpx 文档。

mirostat `class-attribute` `instance-attribute` ¶

mirostat: int | None = None

启用 Mirostat 采样以控制困惑度。（默认值：0，0 = 禁用，1 = Mirostat，2 = Mirostat 2.0）

mirostat_eta `class-attribute` `instance-attribute` ¶

mirostat_eta: float | None = None

影响算法对生成文本反馈的响应速度。较低的学习率将导致较慢的调整，而较高的学习率将使算法更具响应性。（默认值：0.1）

mirostat_tau `class-attribute` `instance-attribute` ¶

mirostat_tau: float | None = None

控制输出的连贯性和多样性之间的平衡。较低的值将产生更集中和连贯的文本。（默认值：5.0）

num_ctx `class-attribute` `instance-attribute` ¶

num_ctx: int | None = None

设置用于生成下一个词元（token）的上下文窗口大小。（默认值：2048）

num_gpu `class-attribute` `instance-attribute` ¶

num_gpu: int | None = None

要使用的 GPU 数量。在 macOS 上，默认为 1 以启用 Metal 支持，0 为禁用。

keep_alive `class-attribute` `instance-attribute` ¶

keep_alive: int | None = None

控制模型在请求后保持加载到内存中的时间（默认值：5m）

num_thread `class-attribute` `instance-attribute` ¶

num_thread: int | None = None

设置计算期间要使用的线程数。默认情况下，Ollama 会自动检测以获得最佳性能。建议将此值设置为系统拥有的物理 CPU 核心数（而不是逻辑核心数）。

repeat_last_n `class-attribute` `instance-attribute` ¶

repeat_last_n: int | None = None

设置模型为防止重复而回顾的距离。（默认值：64，0 = 禁用，-1 = num_ctx）

repeat_penalty `class-attribute` `instance-attribute` ¶

repeat_penalty: float | None = None

设置对重复的惩罚强度。较高的值（例如 1.5）将更强烈地惩罚重复，而较低的值（例如 0.9）将更宽松。（默认值：1.1）

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

temperature: float | None = None

模型的温度。增加温度会使模型的回答更具创造性。（默认值：0.8）

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

stop: list[str] | None = None

设置要使用的停止词元（stop token）。

tfs_z `class-attribute` `instance-attribute` ¶

tfs_z: float | None = None

尾部自由采样（Tail free sampling）用于减少输出中可能性较低的词元的影响。较高的值（例如 2.0）会更多地减少影响，而值为 1.0 则禁用此设置。（默认值：1）

top_k `class-attribute` `instance-attribute` ¶

top_k: int | None = None

减少生成无意义内容的概率。较高的值（例如 100）会给出更多样化的答案，而较低的值（例如 10）会更保守。（默认值：40）

top_p `class-attribute` `instance-attribute` ¶

top_p: float | None = None

与 top-k 一起工作。较高的值（例如 0.95）将导致更多样化的文本，而较低的值（例如 0.5）将生成更集中和保守的文本。（默认值：0.9）

embed_documents ¶

embed_documents(texts: list[str]) -> list[list[float]]

嵌入搜索文档。

embed_query ¶

embed_query(text: str) -> list[float]

嵌入查询文本。

aembed_documents `async` ¶

aembed_documents(texts: list[str]) -> list[list[float]]

嵌入搜索文档。

aembed_query `async` ¶

aembed_query(text: str) -> list[float]

嵌入查询文本。

OllamaLLM ¶

基类：BaseLLM

Ollama 大型语言模型。

设置

安装 langchain-ollama 并在本地安装/运行 Ollama 服务器

pip install -U langchain-ollama
# Visit https://ollama.ac.cn/download to download and install Ollama
# (Linux users): start the server with `ollama serve`

下载一个模型以供使用

ollama pull llama3.1

关键初始化参数 — 客户端参数： base_url: Ollama 服务器托管的基础 URL。 keep_alive: 模型在内存中保持加载的时间。 format: 指定输出的格式。

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

实例化

from langchain_ollama import OllamaLLM

model = OllamaLLM(
    model="llama3.1",
    temperature=0.7,
    num_predict=256,
    # base_url="https://:11434",
    # other params...
)

调用

input_text = "The meaning of life is "
response = model.invoke(input_text)
print(response)

"a philosophical question that has been contemplated by humans for
centuries..."

流

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

a philosophical question that has been contemplated by humans for
centuries...

异步

response = await model.ainvoke(input_text)

# stream:
# async for chunk in model.astream(input_text):
#     print(chunk, end="")

方法	描述
`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__`
`is_lc_serializable`	这个类是否可序列化？
`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`	异步地将一系列提示传递并返回模型生成的内容。
`with_structured_output`	此类未实现。
`get_token_ids`	返回文本中 token 的有序 ID。
`get_num_tokens`	获取文本中存在的 token 数量。
`get_num_tokens_from_messages`	获取消息中的 token 数量。
`generate`	向模型传递一系列提示并返回生成结果。
`agenerate`	异步地将一系列提示传递给模型并返回生成的内容。
`__str__`	返回对象的字符串表示形式以供打印。
`dict`	返回 LLM 的字典。
`save`	保存 LLM。

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

name: str | None = None

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

InputType `property` ¶

InputType: TypeAlias

获取此 Runnable 的输入类型。

OutputType `property` ¶

OutputType: type[str]

获取此 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_secrets `property` ¶

lc_secrets: dict[str, str]

构造函数参数名称到密钥 ID 的映射。

例如，{"openai_api_key": "OPENAI_API_KEY"}

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 的可选编码器。

model `instance-attribute` ¶

model: str

要使用的模型名称。

reasoning `class-attribute` `instance-attribute` ¶

reasoning: bool | None = None

控制支持的模型的推理/思考模式。

True：启用推理模式。模型的推理过程将被捕获并单独返回到响应消息的 additional_kwargs 中，位于 reasoning_content 下。主要响应内容将不包含推理标签。
False：禁用推理模式。模型将不执行任何推理，响应中也不会包含任何推理内容。
None（默认值）：模型将使用其默认的推理行为。如果模型执行推理，<think> 和 </think> 标签将直接出现在主响应内容中。

validate_model_on_init `class-attribute` `instance-attribute` ¶

validate_model_on_init: bool = False

是否在初始化时验证模型是否存在于本地 ollama 中。

在版本 0.3.4 中添加

mirostat `class-attribute` `instance-attribute` ¶

mirostat: int | None = None

启用 Mirostat 采样以控制困惑度。（默认值：0，0 = 禁用，1 = Mirostat，2 = Mirostat 2.0）

mirostat_eta `class-attribute` `instance-attribute` ¶

mirostat_eta: float | None = None

影响算法对生成文本反馈的响应速度。较低的学习率将导致较慢的调整，而较高的学习率将使算法更具响应性。（默认值：0.1）

mirostat_tau `class-attribute` `instance-attribute` ¶

mirostat_tau: float | None = None

控制输出的连贯性和多样性之间的平衡。较低的值将产生更集中和连贯的文本。（默认值：5.0）

num_ctx `class-attribute` `instance-attribute` ¶

num_ctx: int | None = None

设置用于生成下一个词元（token）的上下文窗口大小。（默认值：2048）

num_gpu `class-attribute` `instance-attribute` ¶

num_gpu: int | None = None

要使用的 GPU 数量。在 macOS 上，默认为 1 以启用 Metal 支持，0 为禁用。

num_thread `class-attribute` `instance-attribute` ¶

num_thread: int | None = None

设置计算期间要使用的线程数。默认情况下，Ollama 会自动检测以获得最佳性能。建议将此值设置为系统拥有的物理 CPU 核心数（而不是逻辑核心数）。

num_predict `class-attribute` `instance-attribute` ¶

num_predict: int | None = None

生成文本时预测的最大词元数。（默认值：128，-1 = 无限生成，-2 = 填充上下文）

repeat_last_n `class-attribute` `instance-attribute` ¶

repeat_last_n: int | None = None

设置模型为防止重复而回顾的距离。（默认值：64，0 = 禁用，-1 = num_ctx）

repeat_penalty `class-attribute` `instance-attribute` ¶

repeat_penalty: float | None = None

设置对重复的惩罚强度。较高的值（例如 1.5）将更强烈地惩罚重复，而较低的值（例如 0.9）将更宽松。（默认值：1.1）

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

temperature: float | None = None

模型的温度。增加温度会使模型的回答更具创造性。（默认值：0.8）

seed `class-attribute` `instance-attribute` ¶

seed: int | None = None

设置用于生成的随机数种子。将其设置为特定数字将使模型为相同的提示生成相同的文本。

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

stop: list[str] | None = None

设置要使用的停止词元（stop token）。

tfs_z `class-attribute` `instance-attribute` ¶

tfs_z: float | None = None

尾部自由采样（Tail free sampling）用于减少输出中可能性较低的词元的影响。较高的值（例如 2.0）会更多地减少影响，而值为 1.0 则禁用此设置。（默认值：1）

top_k `class-attribute` `instance-attribute` ¶

top_k: int | None = None

减少生成无意义内容的概率。较高的值（例如 100）会给出更多样化的答案，而较低的值（例如 10）会更保守。（默认值：40）

top_p `class-attribute` `instance-attribute` ¶

top_p: float | None = None

与 top-k 一起工作。较高的值（例如 0.95）将导致更多样化的文本，而较低的值（例如 0.5）将生成更集中和保守的文本。（默认值：0.9）

format `class-attribute` `instance-attribute` ¶

format: Literal['', 'json'] = ''

指定输出的格式（选项：'json'）

keep_alive `class-attribute` `instance-attribute` ¶

keep_alive: int | str | None = None

模型将保持加载到内存中的时间。

base_url `class-attribute` `instance-attribute` ¶

base_url: str | None = None

模型托管的基础 URL。

如果没有，则默认为 Ollama 客户端的默认值。

支持 http://username:password@localhost:11434 格式的 userinfo 认证。如果您的 Ollama 服务器位于代理之后，这将非常有用。

警告

userinfo 不安全，只应用于本地测试或安全环境中。避免在生产环境或不安全的网络上使用它。

注意

如果使用 userinfo，请确保 Ollama 服务器已配置为接受并验证这些凭据。

注意

userinfo 标头会传递给同步和异步客户端。

client_kwargs `class-attribute` `instance-attribute` ¶

client_kwargs: dict | None = {}

传递给 httpx 客户端的额外关键字参数。在此处传递标头。

这些参数会传递给同步和异步客户端。

使用 sync_client_kwargs 和 async_client_kwargs 来为同步和异步客户端传递不同的参数。

async_client_kwargs `class-attribute` `instance-attribute` ¶

async_client_kwargs: dict | None = {}

在传递给 httpx 客户端之前，与 client_kwargs 合并的额外关键字参数。

这些是异步客户端独有的客户端参数；对于共享参数，请使用 client_kwargs。

有关参数的完整列表，请参阅 httpx 文档。

sync_client_kwargs `class-attribute` `instance-attribute` ¶

sync_client_kwargs: dict | None = {}

在传递给 httpx 客户端之前，与 client_kwargs 合并的额外关键字参数。

这些是同步客户端独有的客户端参数；对于共享参数，请使用 client_kwargs。

有关参数的完整列表，请参阅 httpx 文档。

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

将单个输入转换为输出。

参数	描述
`输入`	`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,
) -> str

将单个输入转换为输出。

参数	描述
`输入`	`Runnable` 的输入。类型： `Input`
`配置`	调用 `Runnable` 时使用的配置。该配置支持标准键，如用于追踪目的的 `'tags'`、`'metadata'`，用于控制并行工作量的 `'max_concurrency'`，以及其他键。请参阅 `RunnableConfig` 以获取更多详细信息。类型： `RunnableConfig \| None` 默认值： `None`

返回	描述
`输出`	`Runnable` 的输出。

batch ¶

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

默认实现使用线程池执行器并行运行 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[LanguageModelInput],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    *,
    return_exceptions: bool = False,
    **kwargs: Any,
) -> list[str]

默认实现使用 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[str]

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[str]

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

is_lc_serializable `classmethod` ¶

is_lc_serializable() -> bool

这个类是否可序列化？

根据设计，即使一个类继承自 Serializable，它默认也是不可序列化的。这是为了防止意外序列化不应被序列化的对象。

返回	描述
`bool`	类是否可序列化。默认为 `False`。

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 | list[Callbacks] | None = 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 | list[Callbacks] | None = None,
    **kwargs: Any,
) -> LLMResult

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

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

当你想要

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

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

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

with_structured_output ¶

with_structured_output(
    schema: dict | type, **kwargs: Any
) -> Runnable[LanguageModelInput, dict | BaseModel]

此类未实现。

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(
    prompts: list[str],
    stop: list[str] | None = None,
    callbacks: Callbacks | list[Callbacks] | None = None,
    *,
    tags: list[str] | list[list[str]] | None = None,
    metadata: dict[str, Any] | list[dict[str, Any]] | None = None,
    run_name: str | list[str] | None = None,
    run_id: UUID | list[UUID | None] | None = None,
    **kwargs: Any,
) -> LLMResult

向模型传递一系列提示并返回生成结果。

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

当你想要

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

参数	描述
`prompts`	字符串提示列表。类型: `list[str]`
`停止`	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。类型: `list[str] \| None` 默认值: `None`
`回调`	要传递的 `Callbacks`。用于在生成过程中执行额外的功能，如日志记录或流式处理。类型： `Callbacks \| list[Callbacks] \| None` 默认值： `None`
`tags`	要与每个提示关联的标签列表。如果提供，列表的长度必须与提示列表的长度匹配。类型： `list[str] \| list[list[str]] \| None` 默认值： `None`
`metadata`	要与每个提示关联的元数据字典列表。如果提供，列表的长度必须与提示列表的长度匹配。类型： `dict[str, Any] \| list[dict[str, Any]] \| None` 默认值： `None`
`运行名称`	要与每个提示关联的运行名称列表。如果提供，列表的长度必须与提示列表的长度匹配。类型： `str \| list[str] \| None` 默认值： `None`
`run_id`	要与每个提示关联的运行 ID 列表。如果提供，列表的长度必须与提示列表的长度匹配。类型： `UUID \| list[UUID \| None] \| None` 默认值： `None`
`**kwargs`	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。类型： `Any` 默认值： `{}`

引发	描述
`ValueError`	如果提示不是列表。
`ValueError`	如果 `callbacks`、`tags`、`metadata` 或 `run_name`（如果提供）的长度与提示的长度不匹配。

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

agenerate `async` ¶

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

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

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

当你想要

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

参数	描述
`prompts`	字符串提示列表。类型: `list[str]`
`停止`	生成时使用的停止词。模型输出在首次出现这些子字符串中的任何一个时被截断。类型: `list[str] \| None` 默认值: `None`
`回调`	要传递的 `Callbacks`。用于在生成过程中执行额外的功能，如日志记录或流式处理。类型： `Callbacks \| list[Callbacks] \| None` 默认值： `None`
`tags`	要与每个提示关联的标签列表。如果提供，列表的长度必须与提示列表的长度匹配。类型： `list[str] \| list[list[str]] \| None` 默认值： `None`
`metadata`	要与每个提示关联的元数据字典列表。如果提供，列表的长度必须与提示列表的长度匹配。类型： `dict[str, Any] \| list[dict[str, Any]] \| None` 默认值： `None`
`运行名称`	要与每个提示关联的运行名称列表。如果提供，列表的长度必须与提示列表的长度匹配。类型： `str \| list[str] \| None` 默认值： `None`
`run_id`	要与每个提示关联的运行 ID 列表。如果提供，列表的长度必须与提示列表的长度匹配。类型： `UUID \| list[UUID \| None] \| None` 默认值： `None`
`**kwargs`	任意附加的关键字参数。这些通常会传递给模型提供商的 API 调用。类型： `Any` 默认值： `{}`

引发	描述
`ValueError`	如果 `callbacks`、`tags`、`metadata` 或 `run_name`（如果提供）的长度与提示的长度不匹配。

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

str ¶

__str__() -> str

返回对象的字符串表示形式以供打印。

dict ¶

dict(**kwargs: Any) -> dict

返回 LLM 的字典。

save ¶

save(file_path: Path | str) -> None

保存 LLM。

参数	描述
`file_path`	用于保存 LLM 的文件路径。类型： `Path \| str`

引发	描述
`ValueError`	如果文件路径不是字符串或 Path 对象。

示例

llm.save(file_path="path/llm.yaml")

langchain-ollama¶

langchain_ollama ¶

ChatOllama ¶

name 类属性 实例属性 ¶

InputType 属性 ¶

OutputType 属性 ¶

input_schema 属性 ¶

output_schema 属性 ¶

config_specs 属性 ¶

lc_secrets 属性 ¶

lc_attributes 属性 ¶

cache 类属性 实例属性 ¶

verbose 类属性 实例属性 ¶

callbacks 类属性 实例属性 ¶

tags 类属性 实例属性 ¶

metadata 类属性 实例属性 ¶

custom_get_token_ids 类属性 实例属性 ¶

rate_limiter 类属性 实例属性 ¶

disable_streaming 类属性 实例属性 ¶

output_version 类属性 实例属性 ¶

profile 属性 ¶

model 实例属性 ¶

reasoning 类属性 实例属性 ¶

validate_model_on_init 类属性 实例属性 ¶

mirostat 类属性 实例属性 ¶

mirostat_eta 类属性 实例属性 ¶

mirostat_tau 类属性 实例属性 ¶

num_ctx 类属性 实例属性 ¶

num_gpu 类属性 实例属性 ¶

num_thread 类属性 实例属性 ¶

num_predict 类属性 实例属性 ¶

repeat_last_n 类属性 实例属性 ¶

repeat_penalty 类属性 实例属性 ¶

temperature 类属性 实例属性 ¶

seed 类属性 实例属性 ¶

stop 类属性 实例属性 ¶

tfs_z 类属性 实例属性 ¶

top_k 类属性 实例属性 ¶

top_p 类属性 实例属性 ¶

format 类属性 实例属性 ¶

keep_alive 类属性 实例属性 ¶

base_url 类属性 实例属性 ¶

client_kwargs 类属性 实例属性 ¶

async_client_kwargs 类属性 实例属性 ¶

sync_client_kwargs 类属性 实例属性 ¶

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 ¶

atransform 异步 ¶

bind ¶

with_config ¶

with_listeners ¶

with_alisteners ¶

with_types ¶

with_retry ¶

map ¶

with_fallbacks ¶

as_tool ¶

`langchain-ollama`¶

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

InputType `属性` ¶

OutputType `属性` ¶

input_schema `属性` ¶

output_schema `属性` ¶

config_specs `属性` ¶

lc_secrets `属性` ¶

lc_attributes `属性` ¶

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

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

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

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

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

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

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

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

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

profile `属性` ¶

model `实例属性` ¶

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

validate_model_on_init `类属性` `实例属性` ¶

mirostat `类属性` `实例属性` ¶

mirostat_eta `类属性` `实例属性` ¶

mirostat_tau `类属性` `实例属性` ¶

num_ctx `类属性` `实例属性` ¶

num_gpu `类属性` `实例属性` ¶

num_thread `类属性` `实例属性` ¶

num_predict `类属性` `实例属性` ¶

repeat_last_n `类属性` `实例属性` ¶

repeat_penalty `类属性` `实例属性` ¶

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

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

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

tfs_z `类属性` `实例属性` ¶

top_k `类属性` `实例属性` ¶

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

format `类属性` `实例属性` ¶

keep_alive `类属性` `实例属性` ¶

base_url `类属性` `实例属性` ¶

client_kwargs `类属性` `实例属性` ¶

async_client_kwargs `类属性` `实例属性` ¶

sync_client_kwargs `类属性` `实例属性` ¶

or ¶

ror ¶

ainvoke `异步` ¶

abatch `异步` ¶

abatch_as_completed `异步` ¶

astream `异步` ¶

astream_log `异步` ¶

astream_events `异步` ¶

atransform `异步` ¶

init ¶

is_lc_serializable `类方法` ¶

get_lc_namespace `类方法` ¶

lc_id `类方法` ¶

agenerate_prompt `异步` ¶

agenerate `异步` ¶

model `实例属性` ¶

validate_model_on_init `类属性` `实例属性` ¶

base_url `类属性` `实例属性` ¶

client_kwargs `class-attribute` `instance-attribute` ¶

async_client_kwargs `class-attribute` `instance-attribute` ¶

sync_client_kwargs `class-attribute` `instance-attribute` ¶

mirostat `class-attribute` `instance-attribute` ¶

mirostat_eta `class-attribute` `instance-attribute` ¶

mirostat_tau `class-attribute` `instance-attribute` ¶

num_ctx `class-attribute` `instance-attribute` ¶

num_gpu `class-attribute` `instance-attribute` ¶

keep_alive `class-attribute` `instance-attribute` ¶

num_thread `class-attribute` `instance-attribute` ¶

repeat_last_n `class-attribute` `instance-attribute` ¶

repeat_penalty `class-attribute` `instance-attribute` ¶

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

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

tfs_z `class-attribute` `instance-attribute` ¶

top_k `class-attribute` `instance-attribute` ¶

top_p `class-attribute` `instance-attribute` ¶

aembed_documents `async` ¶

aembed_query `async` ¶