langchain-milvus

langchain_milvus

BaseMilvusBuiltInFunction

基类: ABC

Milvus 内置函数的基类。

请参阅：https://milvus.org.cn/docs/manage-collections.md#Function

BM25BuiltInFunction

Bases: BaseMilvusBuiltInFunction

Milvus BM25 内置函数。

请参阅：https://milvus.org.cn/docs/full-text-search.md

方法	描述
__init__	参数

__init__

__init__(
    *,
    input_field_names: str = TEXT_FIELD,
    output_field_names: str = SPARSE_VECTOR_FIELD,
    analyzer_params: dict[Any, Any] | None = None,
    enable_match: bool = False,
    function_name: str | None = None,
)

参数	描述
input_field_names	输入字段的名称，默认为 'text'。 类型： str 默认值： TEXT_FIELD
output_field_names	输出字段的名称，默认为 'sparse'。 类型： str 默认值： SPARSE_VECTOR_FIELD
analyzer_params	分析器的参数。默认为 None。请参阅：https://milvus.org.cn/docs/analyzer-overview.md#Analyzer-Overview 类型： dict[Any, Any] | None 默认值： None
enable_match	是否启用匹配。 类型： bool 默认值： False
function_name	函数的名称。默认为 None，表示将生成一个随机名称。 类型: str | None 默认值: None

MilvusCollectionHybridSearchRetriever

基类： BaseRetriever

使用 Milvus Collection 基于多个字段检索文档的混合搜索检索器。

更多信息，请参阅：https://milvus.org.cn/docs/release_notes.md#Multi-Embedding---Hybrid-Search

方法	描述
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。
is_lc_serializable	这个类是否可序列化？
get_lc_namespace	获取 LangChain 对象的命名空间。
lc_id	为此类返回一个用于序列化目的的唯一标识符。
to_json	将 Runnable 序列化为 JSON。
to_json_not_implemented	序列化一个“未实现”的对象。
configurable_fields	在运行时配置特定的 Runnable 字段。
configurable_alternatives	为可在运行时设置的 Runnable 对象配置备选项。
__init__

name class-attribute instance-attribute

name: str | None = None

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

InputType property

InputType: type[Input]

输入类型。

此 Runnable 接受的输入类型，以类型注解形式指定。

引发	描述
TypeError	如果输入类型无法推断。

OutputType property

OutputType: type[Output]

输出类型。

此 Runnable 产生的输出类型，指定为类型注解。

引发	描述
TypeError	如果无法推断输出类型。

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 中的属性名称列表。

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

默认为空字典。

tags class-attribute instance-attribute

tags: list[str] | None = None

与检索器关联的可选标签列表。

这些标签将与对此检索器的每次调用相关联，并作为参数传递给在 callbacks 中定义的处理程序。

您可以使用这些标签来识别特定检索器实例及其用例。

metadata class-attribute instance-attribute

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

与检索器关联的可选元数据。

此元数据将与对此检索器的每次调用相关联，并作为参数传递给在 callbacks 中定义的处理程序。

您可以使用这些标签来识别特定检索器实例及其用例。

collection instance-attribute

collection: Collection

Milvus Collection 对象。

rerank instance-attribute

rerank: BaseRanker

Milvus ranker 对象。例如 WeightedRanker 或 RRFRanker。

anns_fields instance-attribute

anns_fields: list[str]

用于 ANNS 搜索的向量字段名称。

field_embeddings instance-attribute

field_embeddings: list[Embeddings | BaseSparseEmbedding]

每个向量字段的嵌入函数，可以是 Embeddings 或 BaseSparseEmbedding。

field_search_params class-attribute instance-attribute

field_search_params: list[dict] | None = None

每个向量字段的搜索参数。如果未指定，将使用默认的搜索参数。

field_limits class-attribute instance-attribute

field_limits: list[int] | None = None

限制每个 ANNS 字段的结果数量。如果未指定，将使用默认的 top_k。

field_exprs class-attribute instance-attribute

field_exprs: list[str | None] | None = None

用于过滤搜索结果的布尔表达式。

top_k class-attribute instance-attribute

top_k: int = 4

最终要检索的文档的 top-K 数量。

text_field class-attribute instance-attribute

text_field: str = 'text'

文本字段名称，将用作 Document 对象的 page_content。

output_fields class-attribute instance-attribute

output_fields: list[str] | None = _get_output_fields()

文档的最终输出字段。如果未指定，除向量字段外的所有字段都将用作输出字段，它们将成为 Document 对象的 metadata。

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: str, config: RunnableConfig | None = None, **kwargs: Any
) -> list[Document]

调用检索器以获取相关文档。

同步检索器调用的主入口点。

参数	描述
输入	查询字符串。 类型： str
配置	检索器的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给检索器的额外参数。 类型： Any 默认值： {}

返回	描述
list[Document]	相关文档的列表。

示例

retriever.invoke("query")

ainvoke async

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

异步调用检索器以获取相关文档。

异步检索器调用的主入口点。

参数	描述
输入	查询字符串。 类型： str
配置	检索器的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给检索器的额外参数。 类型： Any 默认值： {}

返回	描述
list[Document]	相关文档的列表。

示例

await retriever.ainvoke("query")

batch

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

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

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

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

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型： RunnableConfig | list[RunnableConfig] | None 默认值： None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

batch_as_completed

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

在输入列表上并行运行 invoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型： RunnableConfig | Sequence[RunnableConfig] | None 默认值： None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
tuple[int, Output | Exception]	由输入索引和 Runnable 输出组成的元组。

abatch async

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

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

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

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

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型： RunnableConfig | list[RunnableConfig] | None 默认值： None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

abatch_as_completed async

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

在输入列表上并行运行 ainvoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型： RunnableConfig | Sequence[RunnableConfig] | None 默认值： None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[tuple[int, Output | Exception]]	一个由输入索引和 Runnable 输出组成的元组。

stream

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

stream 的默认实现，它调用 invoke。

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

参数	描述
输入	Runnable 的输入。 类型： Input
配置	用于 Runnable 的配置。 类型： RunnableConfig | None 默认值： None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
输出	Runnable 的输出。

astream async

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

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

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
)

__init__

__init__(**kwargs: Any)

ZillizCloudPipelineRetriever

基类： BaseRetriever

Zilliz Cloud Pipeline 检索器。

参数	描述
pipeline_ids	一个包含 pipeline id 的字典。有效键值："ingestion", "search", "deletion"。
token	Zilliz Cloud 的令牌。默认为 ""。
cloud_region	Zilliz Cloud 集群的区域。默认为 'gcp-us-west1'。

方法	描述
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 对象配置备选项。
add_texts	添加文档以进行存储。
add_doc_url	从 url 添加文档。
delete	删除文档。仅 Zilliz Cloud 中的删除管道支持。

name class-attribute instance-attribute

name: str | None = None

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

InputType property

InputType: type[Input]

输入类型。

此 Runnable 接受的输入类型，以类型注解形式指定。

引发	描述
TypeError	如果输入类型无法推断。

OutputType property

OutputType: type[Output]

输出类型。

此 Runnable 产生的输出类型，指定为类型注解。

引发	描述
TypeError	如果无法推断输出类型。

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 中的属性名称列表。

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

默认为空字典。

tags class-attribute instance-attribute

tags: list[str] | None = None

与检索器关联的可选标签列表。

这些标签将与对此检索器的每次调用相关联，并作为参数传递给在 callbacks 中定义的处理程序。

您可以使用这些标签来识别特定检索器实例及其用例。

metadata class-attribute instance-attribute

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

与检索器关联的可选元数据。

此元数据将与对此检索器的每次调用相关联，并作为参数传递给在 callbacks 中定义的处理程序。

您可以使用这些标签来识别特定检索器实例及其用例。

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: str, config: RunnableConfig | None = None, **kwargs: Any
) -> list[Document]

调用检索器以获取相关文档。

同步检索器调用的主入口点。

参数	描述
输入	查询字符串。 类型： str
配置	检索器的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给检索器的额外参数。 类型： Any 默认值： {}

返回	描述
list[Document]	相关文档的列表。

示例

retriever.invoke("query")

ainvoke async

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

异步调用检索器以获取相关文档。

异步检索器调用的主入口点。

参数	描述
输入	查询字符串。 类型： str
配置	检索器的配置。 类型: RunnableConfig | None 默认值: None
**kwargs	要传递给检索器的额外参数。 类型： Any 默认值： {}

返回	描述
list[Document]	相关文档的列表。

示例

await retriever.ainvoke("query")

batch

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

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

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

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

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型： RunnableConfig | list[RunnableConfig] | None 默认值： None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

batch_as_completed

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

在输入列表上并行运行 invoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型： RunnableConfig | Sequence[RunnableConfig] | None 默认值： None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
tuple[int, Output | Exception]	由输入索引和 Runnable 输出组成的元组。

abatch async

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

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

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

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

参数	描述
inputs	Runnable 的输入列表。 类型: list[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型： RunnableConfig | list[RunnableConfig] | None 默认值： None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

返回	描述
list[Output]	来自 Runnable 的输出列表。

abatch_as_completed async

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

在输入列表上并行运行 ainvoke。

在结果完成时生成它们。

参数	描述
inputs	Runnable 的输入列表。 类型: Sequence[Input]
配置	调用 Runnable 时使用的配置。该配置支持标准键，如用于追踪目的的 'tags'、'metadata'，用于控制并行工作量的 'max_concurrency'，以及其他键。请参阅 RunnableConfig 以获取更多详细信息。 类型： RunnableConfig | Sequence[RunnableConfig] | None 默认值： None
返回异常	是否返回异常而不是引发它们。 类型： bool 默认值： False
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
AsyncIterator[tuple[int, Output | Exception]]	一个由输入索引和 Runnable 输出组成的元组。

stream

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

stream 的默认实现，它调用 invoke。

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

参数	描述
输入	Runnable 的输入。 类型： Input
配置	用于 Runnable 的配置。 类型： RunnableConfig | None 默认值： None
**kwargs	要传递给 Runnable 的其他关键字参数。 类型: Any | None 默认值: {}

YIELDS	描述
输出	Runnable 的输出。

astream async

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

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
)

add_texts

add_texts(texts: list[str], metadata: dict[str, Any] | None = None) -> dict

添加文档以进行存储。仅 Zilliz Cloud 中的文本摄入管道支持。

参数	描述
texts	一个文本字符串列表。 类型: list[str]
metadata	一个键值对字典的元数据将被插入，作为摄入管道所需的保留字段。默认为 None。 类型: dict[str, Any] | None 默认值: None

add_doc_url

add_doc_url(doc_url: str, metadata: dict[str, Any] | None = None) -> dict

从 url 添加文档。仅 Zilliz Cloud 中的文档摄入管道支持。

参数	描述
doc_url	一个文档 url。 类型： str
metadata	一个键值对字典的元数据将被插入，作为摄入管道所需的保留字段。默认为 None。 类型: dict[str, Any] | None 默认值: None

delete

delete(key: str, value: Any) -> dict

删除文档。仅 Zilliz Cloud 中的删除管道支持。

参数	描述
key	用于运行删除管道的输入名称 类型： str
value	用于运行删除管道的输入值 类型： Any

Milvus

Bases: VectorStore

Milvus 向量存储集成。

设置

安装 `langchain_milvus` 包

pip install -qU  langchain_milvus

关键初始化参数 — 索引参数： collection_name: str Collection 的名称。 collection_description: str Collection 的描述。 embedding_function: Union[Embeddings, BaseSparseEmbedding] 要使用的嵌入函数。

关键初始化参数 — 客户端参数： connection_args: Optional[dict] 连接参数。

实例化

from langchain_milvus import Milvus
from langchain_openai import OpenAIEmbeddings

URI = "./milvus_example.db"

vector_store = Milvus(
    embedding_function=OpenAIEmbeddings(),
    connection_args={"uri": URI},
)

添加文档

from langchain_core.documents import Document

document_1 = Document(page_content="foo", metadata={"baz": "bar"})
document_2 = Document(page_content="thud", metadata={"baz": "baz"})
document_3 = Document(page_content="i will be deleted :(", metadata={"baz": "qux"})

documents = [document_1, document_2, document_3]
ids = ["1", "2", "3"]
vector_store.add_documents(documents=documents, ids=ids)

删除文档

vector_store.delete(ids=["3"])

搜索

results = vector_store.similarity_search(query="thud",k=1)
for doc in results:
    print(f"* {doc.page_content} [{doc.metadata}]")

* thud [{'baz': 'baz', 'pk': '2'}]

带分数的搜索

results = vector_store.similarity_search_with_score(query="qux",k=1)
for doc, score in results:
    print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")

* [SIM=0.335463] foo [{'baz': 'bar', 'pk': '1'}]

异步

# add documents
# await vector_store.aadd_documents(documents=documents, ids=ids)

# delete documents
# await vector_store.adelete(ids=["3"])

# search
# results = vector_store.asimilarity_search(query="thud",k=1)

# search with score
results = await vector_store.asimilarity_search_with_score(query="qux",k=1)
for doc,score in results:
    print(f"* [SIM={score:3f}] {doc.page_content} [{doc.metadata}]")

* [SIM=0.335463] foo [{'baz': 'bar', 'pk': '1'}]

用作检索器

retriever = vector_store.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 1, "fetch_k": 2, "lambda_mult": 0.5},
)
retriever.invoke("thud")

[Document(metadata={'baz': 'baz', 'pk': '2'}, page_content='thud')]

方法	描述
get_by_ids	根据 ID 获取文档。
aget_by_ids	通过 ID 异步获取文档。
search	使用指定的搜索类型返回与查询最相似的文档。
asearch	异步返回与查询最相似的文档，使用指定的搜索类型。
similarity_search_with_relevance_scores	返回文档和在 `[0, 1]` 范围内的相关性分数。
asimilarity_search_with_relevance_scores	异步返回文档和在 `[0, 1]` 范围内的相关性分数。
from_documents	返回从文档和嵌入初始化的 `VectorStore`。
afrom_documents	异步返回从文档和嵌入初始化的 `VectorStore`。
as_retriever	返回从此 `VectorStore` 初始化的 `VectorStoreRetriever`。
__init__	初始化 Milvus 向量存储。
add_texts	将文本数据插入 Milvus。
add_embeddings	将带有嵌入向量的文本数据插入 Milvus。
similarity_search	针对查询字符串执行相似性搜索。
similarity_search_by_vector	针对查询字符串执行相似性搜索。
similarity_search_with_score	对查询字符串执行搜索并返回带有分数的结果。
similarity_search_with_score_by_vector	对嵌入执行搜索并返回带有分数的结果。
max_marginal_relevance_search	执行搜索并返回由 MMR 重新排序的结果。
max_marginal_relevance_search_by_vector	执行搜索并返回由 MMR 重新排序的结果。
delete	通过向量 ID 或布尔表达式删除。
drop	通过删除（唯一的）collection 来删除索引中的所有内容。
from_texts	创建一个 Milvus collection，使用 HNSW 对其进行索引，并插入数据。
add_documents	通过嵌入运行更多文档并将其添加到向量存储中。
get_pks	使用表达式获取主键
upsert	更新/插入文档到向量存储。
search_by_metadata	根据元数据条件搜索 Milvus 向量存储。
aadd_texts	异步将文本数据插入 Milvus。
aadd_embeddings	异步将带有嵌入向量的文本数据插入 Milvus。
asimilarity_search	针对查询字符串执行异步相似性搜索。
asimilarity_search_by_vector	针对查询向量执行异步相似性搜索。
asimilarity_search_with_score	对查询字符串执行异步搜索并返回带有分数的结果。
asimilarity_search_with_score_by_vector	对嵌入执行异步搜索并返回带有分数的结果。
amax_marginal_relevance_search	执行异步搜索并返回由 MMR 重新排序的结果。
amax_marginal_relevance_search_by_vector	执行异步搜索并返回由 MMR 重新排序的结果。
adelete	异步通过向量 ID 或布尔表达式删除。
afrom_texts	创建一个 Milvus collection，使用 HNSW 对其进行索引，并插入数据
aadd_documents	通过嵌入运行更多文档并将其添加到向量存储中
aget_pks	异步使用表达式获取主键
aupsert	异步更新/插入文档到向量存储。
asearch_by_metadata	异步根据元数据条件搜索 Milvus 向量存储。

embeddings property

embeddings: EmbeddingType | list[EmbeddingType] | None

获取嵌入函数。

client property

client: MilvusClient

获取客户端。

aclient property

aclient: AsyncMilvusClient

获取异步客户端。

vector_fields property

vector_fields: list[str]

获取向量字段。

get_by_ids

get_by_ids(ids: Sequence[str]) -> list[Document]

根据 ID 获取文档。

返回的文档应将其 ID 字段设置为文档在向量存储中的 ID。

如果某些 ID 未找到或存在重复的 ID，返回的文档数量可能少于请求的数量。

用户不应假设返回文档的顺序与输入 ID 的顺序相匹配。相反，用户应依赖于返回文档的 ID 字段。

如果某些 ID 未找到文档，此方法不应**抛出异常**。

参数	描述
ids	要检索的 ID 列表。 类型： Sequence[str]

返回	描述
list[Document]	`Document` 对象列表。

aget_by_ids async

aget_by_ids(ids: Sequence[str]) -> list[Document]

通过 ID 异步获取文档。

返回的文档应将其 ID 字段设置为文档在向量存储中的 ID。

如果某些 ID 未找到或存在重复的 ID，返回的文档数量可能少于请求的数量。

用户不应假设返回文档的顺序与输入 ID 的顺序相匹配。相反，用户应依赖于返回文档的 ID 字段。

如果某些 ID 未找到文档，此方法不应**抛出异常**。

参数	描述
ids	要检索的 ID 列表。 类型： Sequence[str]

返回	描述
list[Document]	`Document` 对象列表。

search

search(query: str, search_type: str, **kwargs: Any) -> list[Document]

使用指定的搜索类型返回与查询最相似的文档。

参数	描述
query	输入文本。 类型： str
search_type	要执行的搜索类型。可以是 `'similarity'`、`'mmr'` 或 `'similarity_score_threshold'`。 类型： str
**kwargs	传递给搜索方法的参数。 类型： Any 默认值： {}

返回	描述
list[Document]	与查询最相似的 `Document` 对象列表。

引发	描述
ValueError	如果 `search_type` 不是 `'similarity'`、`'mmr'` 或 `'similarity_score_threshold'` 之一。

asearch async

asearch(query: str, search_type: str, **kwargs: Any) -> list[Document]

异步返回与查询最相似的文档，使用指定的搜索类型。

参数	描述
query	输入文本。 类型： str
search_type	要执行的搜索类型。可以是 `'similarity'`、`'mmr'` 或 `'similarity_score_threshold'`。 类型： str
**kwargs	传递给搜索方法的参数。 类型： Any 默认值： {}

返回	描述
list[Document]	与查询最相似的 `Document` 对象列表。

引发	描述
ValueError	如果 `search_type` 不是 `'similarity'`、`'mmr'` 或 `'similarity_score_threshold'` 之一。

similarity_search_with_relevance_scores

similarity_search_with_relevance_scores(
    query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]

返回文档和在 `[0, 1]` 范围内的相关性分数。

`0` 表示不相似，`1` 表示最相似。

参数	描述
query	输入文本。 类型： str
k	要返回的 `Document` 对象数量。 TYPE: int DEFAULT: 4
**kwargs	将传递给相似性搜索的kwargs。应包括`score_threshold`，一个可选的浮点值，介于`0`到`1`之间，用于筛选检索到的文档结果集。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	由 `(doc, similarity_score)` 组成的元组列表。

asimilarity_search_with_relevance_scores async

asimilarity_search_with_relevance_scores(
    query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]

异步返回文档和在 `[0, 1]` 范围内的相关性分数。

`0` 表示不相似，`1` 表示最相似。

参数	描述
query	输入文本。 类型： str
k	要返回的 `Document` 对象数量。 TYPE: int DEFAULT: 4
**kwargs	将传递给相似性搜索的kwargs。应包括`score_threshold`，一个可选的浮点值，介于`0`到`1`之间，用于筛选检索到的文档结果集。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	元组列表 `(doc, similarity_score)`

from_documents classmethod

from_documents(documents: list[Document], embedding: Embeddings, **kwargs: Any) -> Self

返回从文档和嵌入初始化的 `VectorStore`。

参数	描述
documents	要添加到 `VectorStore` 的 `Document` 对象列表。 TYPE: list[Document]
embedding	要使用的嵌入函数。 TYPE: Embeddings
**kwargs	附加的关键字参数。 类型： Any 默认值： {}

返回	描述
Self	从文档和嵌入初始化的 `VectorStore`。

afrom_documents async classmethod

afrom_documents(
    documents: list[Document], embedding: Embeddings, **kwargs: Any
) -> Self

异步返回从文档和嵌入初始化的 `VectorStore`。

参数	描述
documents	要添加到 `VectorStore` 的 `Document` 对象列表。 TYPE: list[Document]
embedding	要使用的嵌入函数。 TYPE: Embeddings
**kwargs	附加的关键字参数。 类型： Any 默认值： {}

返回	描述
Self	从文档和嵌入初始化的 `VectorStore`。

as_retriever

as_retriever(**kwargs: Any) -> VectorStoreRetriever

返回从此 `VectorStore` 初始化的 `VectorStoreRetriever`。

参数	描述
**kwargs	传递给搜索函数的关键字参数。可以包括 `search_type`：定义检索器应执行的搜索类型。可以是 `'similarity'` (默认)、`'mmr'` 或 `'similarity_score_threshold'`。 `search_kwargs`：传递给搜索函数的关键字参数。可以包括诸如 `k`：要返回的文档数量（默认：`4`） `score_threshold`：`similarity_score_threshold` 的最小相关性阈值 `fetch_k`：传递给 MMR 算法的文档数量（默认值：`20`） `lambda_mult`：MMR 返回结果的多样性；`1` 表示最小多样性，0 表示最大多样性。（默认值：`0.5`） `filter`：按文档元数据过滤 类型： Any 默认值： {}

返回	描述
VectorStoreRetriever	`VectorStore` 的检索器类。

示例

# Retrieve more documents with higher diversity
# Useful if your dataset has many similar documents
docsearch.as_retriever(
    search_type="mmr", search_kwargs={"k": 6, "lambda_mult": 0.25}
)

# Fetch more documents for the MMR algorithm to consider
# But only return the top 5
docsearch.as_retriever(search_type="mmr", search_kwargs={"k": 5, "fetch_k": 50})

# Only retrieve documents that have a relevance score
# Above a certain threshold
docsearch.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.8},
)

# Only get the single most similar document from the dataset
docsearch.as_retriever(search_kwargs={"k": 1})

# Use a filter to only retrieve documents from a specific paper
docsearch.as_retriever(
    search_kwargs={"filter": {"paper_title": "GPT-4 Technical Report"}}
)

__init__

__init__(
    embedding_function: EmbeddingType | list[EmbeddingType] | None,
    collection_name: str = "LangChainCollection",
    collection_description: str = "",
    collection_properties: dict[str, Any] | None = None,
    connection_args: dict[str, Any] | None = None,
    consistency_level: str = "Session",
    index_params: dict | list[dict] | None = None,
    search_params: dict | list[dict] | None = None,
    drop_old: bool | None = False,
    auto_id: bool = False,
    *,
    primary_field: str = PRIMARY_FIELD,
    text_field: str = TEXT_FIELD,
    vector_field: str | list[str] = VECTOR_FIELD,
    enable_dynamic_field: bool = False,
    metadata_field: str | None = None,
    partition_key_field: str | None = None,
    num_partitions: int | None = None,
    partition_names: list | None = None,
    replica_number: int = 1,
    timeout: float | None = None,
    num_shards: int | None = None,
    vector_schema: dict[str, Any] | list[dict[str, Any]] | None = None,
    metadata_schema: dict[str, Any] | None = None,
    builtin_function: BaseMilvusBuiltInFunction
    | list[BaseMilvusBuiltInFunction]
    | None = None,
)

初始化 Milvus 向量存储。

add_texts

add_texts(
    texts: Iterable[str],
    metadatas: list[dict] | None = None,
    timeout: float | None = None,
    batch_size: int = 1000,
    *,
    ids: list[str] | None = None,
    **kwargs: Any,
) -> list[str]

将文本数据插入 Milvus。

在尚未创建集合（collection）时插入数据，将导致创建一个新的集合。第一个实体的数据决定了新集合的模式（schema），维度（dim）从第一个嵌入（embedding）中提取，列（columns）由第一个元数据字典决定。元数据键（Metadata keys）需要存在于所有插入的值中。目前 Milvus 中没有与 None 等效的值。

参数	描述
texts	要嵌入的文本，假定它们全部都能装入内存。 类型： Iterable[str]
metadatas	附加到每个文本的元数据字典。默认为 None。 类型： list[dict] | None 默认值： None
timeout	每次批量插入的超时时间。默认为 None。 类型： float | None 默认值： None
batch_size	用于插入的批处理大小。默认为 1000。 类型： int 默认值： 1000
ids	文本 ID 列表。每个项目的长度 类型: list[str] | None 默认值: None

引发	描述
MilvusException	添加文本失败

返回	描述
list[str]	List[str]：每个插入元素的最终键列表。

add_embeddings

add_embeddings(
    texts: list[str],
    embeddings: List[List[float]] | List[List[List[float]]],
    metadatas: list[dict] | None = None,
    timeout: float | None = None,
    batch_size: int = 1000,
    *,
    ids: list[str] | None = None,
    **kwargs: Any,
) -> list[str]

将带有嵌入向量的文本数据插入 Milvus。

此方法将一批文本嵌入插入到 Milvus 集合中。如果集合尚未初始化，它将根据嵌入、元数据和其他参数自动初始化集合。嵌入应使用兼容的嵌入函数预先生成，与每个文本关联的元数据是可选的，但数量必须与文本数量匹配。

参数	描述
texts	要插入的文本 类型: list[str]
embeddings	每个文本的向量嵌入（如果是单个向量）或每个文本的向量列表（如果是多向量）。 类型： List[List[float]] | List[List[List[float]]]
metadatas	附加到每个文本的元数据字典。默认为 None。 类型： list[dict] | None 默认值： None
timeout	每次批量插入的超时时间。默认为 None。 类型： float | None 默认值： None
batch_size	用于插入的批处理大小。默认为 1000。 类型： int 默认值： 1000
ids	文本 ID 列表。每个项目的长度 类型: list[str] | None 默认值: None

引发	描述
MilvusException	添加文本和嵌入失败

返回	描述
list[str]	List[str]：每个插入元素的最终键列表。

similarity_search

similarity_search(
    query: str,
    k: int = 4,
    param: dict | list[dict] | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

针对查询字符串执行相似性搜索。

参数	描述
query	要搜索的文本。 类型： str
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
param	索引类型的搜索参数。默认为 None。 类型： dict | list[dict] 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： int 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

similarity_search_by_vector

similarity_search_by_vector(
    embedding: list[float],
    k: int = 4,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

针对查询字符串执行相似性搜索。

参数	描述
embedding	要搜索的嵌入向量。 类型： list[float]
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
param	索引类型的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： int 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

similarity_search_with_score

similarity_search_with_score(
    query: str,
    k: int = 4,
    param: dict | list[dict] | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[tuple[Document, float]]

对查询字符串执行搜索并返回带有分数的结果。

有关搜索参数的更多信息，请参阅此处的 pymilvus 文档：https://milvus.org.cn/api-reference/pymilvus/v2.5.x/ORM/Collection/search.md

参数	描述
query	正在搜索的文本。 类型： str
k	要返回的结果数量。默认为 4。 TYPE: int DEFAULT: 4
param	指定索引的搜索参数。 类型： dict | list[dict] 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 或 hybrid_search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	List[Tuple[Document, float]]：结果文档和分数的列表。

similarity_search_with_score_by_vector

similarity_search_with_score_by_vector(
    embedding: List[float] | Dict[int, float],
    k: int = 4,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[tuple[Document, float]]

对嵌入执行搜索并返回带有分数的结果。

有关搜索参数的更多信息，请参阅此处的 pymilvus 文档：https://milvus.org.cn/api-reference/pymilvus/v2.5.x/ORM/Collection/search.md

参数	描述
embedding	正在搜索的嵌入向量。 类型： List[float] | Dict[int, float]
k	要返回的结果数量。默认为 4。 TYPE: int DEFAULT: 4
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	List[Tuple[Document, float]]：结果文档和分数。

max_marginal_relevance_search

max_marginal_relevance_search(
    query: str,
    k: int = 4,
    fetch_k: int = 20,
    lambda_mult: float = 0.5,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

执行搜索并返回由 MMR 重新排序的结果。

参数	描述
query	正在搜索的文本。 类型： str
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
fetch_k	从中选择 k 个结果的总结果数。默认为 20。 类型： int 默认值： 20
lambda_mult	一个介于 0 和 1 之间的数字，用于确定结果之间的多样性程度，其中 0 对应最大多样性，1 对应最小多样性。默认为 0.5 类型： float 默认值： 0.5
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

max_marginal_relevance_search_by_vector

max_marginal_relevance_search_by_vector(
    embedding: list[float] | dict[int, float],
    k: int = 4,
    fetch_k: int = 20,
    lambda_mult: float = 0.5,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

执行搜索并返回由 MMR 重新排序的结果。

参数	描述
embedding	正在搜索的嵌入向量。 类型： list[float] | dict[int, float]
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
fetch_k	从中选择 k 个结果的总结果数。默认为 20。 类型： int 默认值： 20
lambda_mult	一个介于 0 和 1 之间的数字，用于确定结果之间的多样性程度，其中 0 对应最大多样性，1 对应最小多样性。默认为 0.5 类型： float 默认值： 0.5
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

delete

delete(
    ids: list[str] | None = None, expr: str | None = None, **kwargs: str
) -> bool | None

通过向量 ID 或布尔表达式删除。有关表达式的说明和示例，请参阅 Milvus 文档。

参数	描述
ids	要删除的 ID 列表。 类型: list[str] | None 默认值: None
expr	指定要删除的实体的布尔表达式。 类型: str | None 默认值: None
kwargs	Milvus 删除 API 中的其他参数。 类型： str 默认值： {}

返回	描述
bool | None	Optional[bool]：如果删除成功，则为 True，
bool | None	否则为 False。

drop

drop() -> None

通过删除（唯一的）collection 来删除索引中的所有内容。

from_texts classmethod

from_texts(
    texts: list[str],
    embedding: EmbeddingType | list[EmbeddingType] | None,
    metadatas: list[dict] | None = None,
    collection_name: str = "LangChainCollection",
    connection_args: dict[str, Any] | None = None,
    consistency_level: str = "Session",
    index_params: dict | list[dict] | None = None,
    search_params: dict | list[dict] | None = None,
    drop_old: bool = False,
    *,
    ids: list[str] | None = None,
    auto_id: bool = False,
    builtin_function: BaseMilvusBuiltInFunction
    | list[BaseMilvusBuiltInFunction]
    | None = None,
    **kwargs: Any,
) -> Milvus

创建一个 Milvus collection，使用 HNSW 对其进行索引，并插入数据。

参数	描述
texts	文本数据。 类型: list[str]
embedding	嵌入函数。 类型： Embeddings | BaseSparseEmbedding | None
metadatas	每个文本的元数据（如果存在）。默认为 None。 类型： list[dict] | None 默认值： None
collection_name	要使用的集合名称。默认为 "LangChainCollection"。 类型： str 默认值： 'LangChainCollection'
connection_args	要使用的连接参数。默认为 DEFAULT_MILVUS_CONNECTION。 类型： dict[str, Any] 默认值： None
consistency_level	使用哪种一致性级别。默认为 "Session"。 类型： str 默认值： 'Session'
index_params	使用哪个 index_params。默认为 None。 类型： dict | None 默认值： None
search_params	使用哪个 search_params。默认为 None。 类型： dict | None 默认值： None
drop_old	如果存在同名集合，是否删除它。默认为 False。 类型： bool | None 默认值： False
ids	文本 ID 列表。默认为 None。 类型: list[str] | None 默认值: None
auto_id	是否为主键启用自动 ID。默认为 False。如果为 False，您需要提供文本 ID（小于 65535 字节的字符串）。如果为 True，Milvus 将生成唯一的整数作为主键。 类型： bool 默认值： False
**kwargs	Milvus 集合中的其他参数。 类型： Any 默认值： {}

返回： Milvus：Milvus 向量存储

add_documents

add_documents(documents: list[Document], **kwargs: Any) -> list[str]

通过嵌入运行更多文档并将其添加到向量存储中。

参数	描述
documents	要添加到向量存储的文档。 TYPE: list[Document]

返回	描述
list[str]	已添加文本的 ID 列表。

get_pks

get_pks(expr: str, **kwargs: Any) -> List[int] | None

使用表达式获取主键

参数	描述
expr	表达式 - 例如："id in [1, 2]" 或 "title LIKE 'Abc%'" 类型： str

返回	描述
List[int] | None	List[int]：ID 列表（主键）

upsert

upsert(
    ids: list[str] | None = None,
    documents: List[Document] | None = None,
    batch_size: int = 1000,
    timeout: float | None = None,
    **kwargs: Any,
) -> None

更新/插入文档到向量存储。

参数	描述
ids	要更新的 ID - 调用 get_pks 以通过表达式获取 ID 类型: list[str] | None 默认值: None
documents	要添加到向量存储的文档。 类型： list[Document] 默认值： None

search_by_metadata

search_by_metadata(
    expr: str, fields: list[str] | None = None, limit: int = 10
) -> list[Document]

根据元数据条件搜索 Milvus 向量存储。

此函数使用表达式执行基于元数据的查询，该表达式过滤存储的文档，而无需进行向量相似度比较。

参数	描述
expr	一个过滤表达式（例如，`"city == 'Seoul'"`）。 类型： str
fields	要检索的字段列表。如果为 None，则检索所有可用字段。 类型: list[str] | None 默认值: None
limit	要返回的最大结果数。 类型： int 默认值： 10

返回	描述
list[Document]	List[Document]：匹配元数据过滤器的文档列表。

aadd_texts async

aadd_texts(
    texts: Iterable[str],
    metadatas: list[dict] | None = None,
    timeout: float | None = None,
    batch_size: int = 1000,
    *,
    ids: list[str] | None = None,
    **kwargs: Any,
) -> list[str]

异步将文本数据插入 Milvus。

在尚未创建集合（collection）时插入数据，将导致创建一个新的集合。第一个实体的数据决定了新集合的模式（schema），维度（dim）从第一个嵌入（embedding）中提取，列（columns）由第一个元数据字典决定。元数据键（Metadata keys）需要存在于所有插入的值中。目前 Milvus 中没有与 None 等效的值。

参数	描述
texts	要嵌入的文本，假定它们全部都能装入内存。 类型： Iterable[str]
metadatas	附加到每个文本的元数据字典。默认为 None。 类型： list[dict] | None 默认值： None
timeout	每次批量插入的超时时间。默认为 None。 类型： float | None 默认值： None
batch_size	用于插入的批处理大小。默认为 1000。 类型： int 默认值： 1000
ids	文本 ID 列表。每个项目的长度 类型: list[str] | None 默认值: None

引发	描述
MilvusException	添加文本失败

返回	描述
list[str]	List[str]：每个插入元素的最终键列表。

aadd_embeddings async

aadd_embeddings(
    texts: list[str],
    embeddings: List[List[float]] | List[List[List[float]]],
    metadatas: list[dict] | None = None,
    timeout: float | None = None,
    batch_size: int = 1000,
    *,
    ids: list[str] | None = None,
    **kwargs: Any,
) -> list[str]

异步将带有嵌入向量的文本数据插入 Milvus。

此方法将一批文本嵌入插入到 Milvus 集合中。如果集合尚未初始化，它将根据嵌入、元数据和其他参数自动初始化集合。嵌入应使用兼容的嵌入函数预先生成，与每个文本关联的元数据是可选的，但数量必须与文本数量匹配。

参数	描述
texts	要插入的文本 类型: list[str]
embeddings	每个文本的向量嵌入（如果是单个向量）或每个文本的向量列表（如果是多向量）。 类型： List[List[float]] | List[List[List[float]]]
metadatas	附加到每个文本的元数据字典。默认为 None。 类型： list[dict] | None 默认值： None
timeout	每次批量插入的超时时间。默认为 None。 类型： float | None 默认值： None
batch_size	用于插入的批处理大小。默认为 1000。 类型： int 默认值： 1000
ids	文本 ID 列表。每个项目的长度 类型: list[str] | None 默认值: None

引发	描述
MilvusException	添加文本和嵌入失败

返回	描述
list[str]	List[str]：每个插入元素的最终键列表。

asimilarity_search async

asimilarity_search(
    query: str,
    k: int = 4,
    param: dict | list[dict] | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

针对查询字符串执行异步相似性搜索。

参数	描述
query	要搜索的文本。 类型： str
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
param	索引类型的搜索参数。默认为 None。 类型： dict | list[dict] 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： int 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

asimilarity_search_by_vector async

asimilarity_search_by_vector(
    embedding: list[float],
    k: int = 4,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

针对查询向量执行异步相似性搜索。

参数	描述
embedding	要搜索的嵌入向量。 类型： list[float]
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
param	索引类型的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： int 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

asimilarity_search_with_score async

asimilarity_search_with_score(
    query: str,
    k: int = 4,
    param: dict | list[dict] | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[tuple[Document, float]]

对查询字符串执行异步搜索并返回带有分数的结果。

有关搜索参数的更多信息，请参阅此处的 pymilvus 文档：https://milvus.org.cn/api-reference/pymilvus/v2.5.x/ORM/Collection/search.md

参数	描述
query	正在搜索的文本。 类型： str
k	要返回的结果数量。默认为 4。 TYPE: int DEFAULT: 4
param	指定索引的搜索参数。 类型： dict | list[dict] 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 或 hybrid_search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	List[Tuple[Document, float]]：结果文档和分数的列表。

asimilarity_search_with_score_by_vector async

asimilarity_search_with_score_by_vector(
    embedding: List[float] | Dict[int, float],
    k: int = 4,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[tuple[Document, float]]

对嵌入执行异步搜索并返回带有分数的结果。

有关搜索参数的更多信息，请参阅此处的 pymilvus 文档：https://milvus.org.cn/api-reference/pymilvus/v2.5.x/ORM/Collection/search.md

参数	描述
embedding	正在搜索的嵌入向量。 类型： List[float] | Dict[int, float]
k	要返回的结果数量。默认为 4。 TYPE: int DEFAULT: 4
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	List[Tuple[Document, float]]：结果文档和分数。

amax_marginal_relevance_search async

amax_marginal_relevance_search(
    query: str,
    k: int = 4,
    fetch_k: int = 20,
    lambda_mult: float = 0.5,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

执行异步搜索并返回由 MMR 重新排序的结果。

参数	描述
query	正在搜索的文本。 类型： str
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
fetch_k	从中选择 k 个结果的总结果数。默认为 20。 类型： int 默认值： 20
lambda_mult	一个介于 0 和 1 之间的数字，用于确定结果之间的多样性程度，其中 0 对应最大多样性，1 对应最小多样性。默认为 0.5 类型： float 默认值： 0.5
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

amax_marginal_relevance_search_by_vector async

amax_marginal_relevance_search_by_vector(
    embedding: list[float] | dict[int, float],
    k: int = 4,
    fetch_k: int = 20,
    lambda_mult: float = 0.5,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

执行异步搜索并返回由 MMR 重新排序的结果。

参数	描述
embedding	正在搜索的嵌入向量。 类型： list[float] | dict[int, float]
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
fetch_k	从中选择 k 个结果的总结果数。默认为 20。 类型： int 默认值： 20
lambda_mult	一个介于 0 和 1 之间的数字，用于确定结果之间的多样性程度，其中 0 对应最大多样性，1 对应最小多样性。默认为 0.5 类型： float 默认值： 0.5
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

adelete async

adelete(
    ids: list[str] | None = None, expr: str | None = None, **kwargs: Any
) -> bool | None

通过向量 ID 或布尔表达式异步删除。有关表达式的说明和示例，请参阅 Milvus 文档。

参数	描述
ids	要删除的 ID 列表。 类型: list[str] | None 默认值: None
expr	指定要删除的实体的布尔表达式。 类型: str | None 默认值: None
kwargs	Milvus 删除 API 中的其他参数。 类型： Any 默认值： {}

返回	描述
bool | None	Optional[bool]：如果删除成功，则为 True，
bool | None	否则为 False。

afrom_texts async classmethod

afrom_texts(
    texts: list[str],
    embedding: EmbeddingType | list[EmbeddingType] | None,
    metadatas: list[dict] | None = None,
    collection_name: str = "LangChainCollection",
    connection_args: dict[str, Any] | None = None,
    consistency_level: str = "Session",
    index_params: dict | list[dict] | None = None,
    search_params: dict | list[dict] | None = None,
    drop_old: bool = False,
    *,
    ids: list[str] | None = None,
    auto_id: bool = False,
    builtin_function: BaseMilvusBuiltInFunction
    | list[BaseMilvusBuiltInFunction]
    | None = None,
    **kwargs: Any,
) -> Milvus

创建一个 Milvus 集合，使用 HNSW 对其进行索引，并异步插入数据。

参数	描述
texts	文本数据。 类型: list[str]
embedding	嵌入函数。 类型： Embeddings | BaseSparseEmbedding | None
metadatas	每个文本的元数据（如果存在）。默认为 None。 类型： list[dict] | None 默认值： None
collection_name	要使用的集合名称。默认为 "LangChainCollection"。 类型： str 默认值： 'LangChainCollection'
connection_args	要使用的连接参数。默认为 DEFAULT_MILVUS_CONNECTION。 类型： dict[str, Any] 默认值： None
consistency_level	使用哪种一致性级别。默认为 "Session"。 类型： str 默认值： 'Session'
index_params	使用哪个 index_params。默认为 None。 类型： dict | None 默认值： None
search_params	使用哪个 search_params。默认为 None。 类型： dict | None 默认值： None
drop_old	如果存在同名集合，是否删除它。默认为 False。 类型： bool | None 默认值： False
ids	文本 ID 列表。默认为 None。 类型: list[str] | None 默认值: None
auto_id	是否为主键启用自动 ID。默认为 False。如果为 False，您需要提供文本 ID（小于 65535 字节的字符串）。如果为 True，Milvus 将生成唯一的整数作为主键。 类型： bool 默认值： False
**kwargs	Milvus 集合中的其他参数。 类型： Any 默认值： {}

返回： Milvus：Milvus 向量存储

aadd_documents async

aadd_documents(documents: list[Document], **kwargs: Any) -> list[str]

通过嵌入处理更多文档并异步添加到向量存储。

参数	描述
documents	要添加到向量存储的文档。 TYPE: list[Document]

返回	描述
list[str]	已添加文本的 ID 列表。

aget_pks async

aget_pks(expr: str, **kwargs: Any) -> List[int] | None

异步使用表达式获取主键

参数	描述
expr	表达式 - 例如："id in [1, 2]" 或 "title LIKE 'Abc%'" 类型： str

返回	描述
List[int] | None	List[int]：ID 列表（主键）

aupsert async

aupsert(
    ids: list[str] | None = None,
    documents: List[Document] | None = None,
    batch_size: int = 1000,
    timeout: float | None = None,
    **kwargs: Any,
) -> None

异步更新/插入文档到向量存储。

参数	描述
ids	要更新的 ID - 调用 aget_pks 以通过表达式获取 ID 类型: list[str] | None 默认值: None
documents	要添加到向量存储的文档。 类型： list[Document] 默认值： None
batch_size	用于 upsert 的批处理大小。默认为 1000。 类型： int 默认值： 1000
timeout	每次批量 upsert 的超时时间。默认为 None。 类型： float | None 默认值： None
**kwargs	Milvus upsert API 中的其他参数。 类型： Any 默认值： {}

asearch_by_metadata async

asearch_by_metadata(
    expr: str, fields: list[str] | None = None, limit: int = 10
) -> list[Document]

异步根据元数据条件搜索 Milvus 向量存储。

此函数使用表达式执行基于元数据的查询，该表达式过滤存储的文档，而无需进行向量相似度比较。

参数	描述
expr	一个过滤表达式（例如，`"city == 'Seoul'"`）。 类型： str
fields	要检索的字段列表。如果为 None，则检索所有可用字段。 类型: list[str] | None 默认值: None
limit	要返回的最大结果数。 类型： int 默认值： 10

返回	描述
list[Document]	List[Document]：匹配元数据过滤器的文档列表。

Zilliz

基类： Milvus

Zilliz 向量存储。

您需要安装 pymilvus 并有一个正在运行的 Zilliz 数据库。

有关如何运行 Zilliz 实例，请参阅以下文档：https://docs.zilliz.com/docs/create-cluster

如果使用 L2/IP 度量标准，强烈建议对数据进行归一化。

参数	描述
embedding_function	用于嵌入文本的函数。 类型： Embeddings
collection_name	使用哪个 Zilliz 集合。默认为 "LangChainCollection"。 类型： str
connection_args	此类使用的连接参数以字典形式提供。 类型： Optional[dict[str, any]]
consistency_level	集合使用的一致性级别。默认为 "Session"。 类型： str
index_params	使用哪个索引参数。根据服务默认为 HNSW/AUTOINDEX。 类型： Optional[dict]
search_params	使用哪个搜索参数。默认为索引的默认值。 类型： Optional[dict]
drop_old	是否删除当前集合。默认为 False。 类型： Optional[bool]
auto_id	是否为主键启用自动 ID。默认为 False。如果为 False，您需要提供文本 ID（小于 65535 字节的字符串）。如果为 True，Milvus 将生成唯一的整数作为主键。 类型： bool

此类使用的连接参数以字典形式提供，两个主要参数是：uri (str)：Zilliz 实例的公共端点。示例 uri："https://in03-ba4234asae.api.gcp-us-west1.zillizcloud.com"，token (str)：API 密钥，用于 serverless 集群，可替代用户名和密码。更多信息，请参阅：https://docs.zilliz.com/docs/on-zilliz-cloud-console#cluster-details 和 https://docs.zilliz.com/reference/python/python/Connections-connect

示例

from langchain_milvus import Zilliz
from langchain_openai import OpenAIEmbeddings

embedding = OpenAIEmbeddings()
# Connect to a Zilliz instance
milvus_store = Zilliz(
    embedding_function = embedding,
    collection_name = "LangChainCollection",
    connection_args = {
        "uri": "https://in03-ba4234asae.api.gcp-us-west1.zillizcloud.com",
        "token": "temp", # API key
    }
    drop_old: True,
)

引发	描述
ValueError	如果未安装 pymilvus python 包。

方法	描述
add_texts	将文本数据插入 Milvus。
delete	通过向量 ID 或布尔表达式删除。
get_by_ids	根据 ID 获取文档。
aget_by_ids	通过 ID 异步获取文档。
adelete	异步通过向量 ID 或布尔表达式删除。
aadd_texts	异步将文本数据插入 Milvus。
add_documents	通过嵌入运行更多文档并将其添加到向量存储中。
aadd_documents	通过嵌入运行更多文档并将其添加到向量存储中
search	使用指定的搜索类型返回与查询最相似的文档。
asearch	异步返回与查询最相似的文档，使用指定的搜索类型。
similarity_search	针对查询字符串执行相似性搜索。
similarity_search_with_score	对查询字符串执行搜索并返回带有分数的结果。
asimilarity_search_with_score	对查询字符串执行异步搜索并返回带有分数的结果。
similarity_search_with_relevance_scores	返回文档和在 `[0, 1]` 范围内的相关性分数。
asimilarity_search_with_relevance_scores	异步返回文档和在 `[0, 1]` 范围内的相关性分数。
asimilarity_search	针对查询字符串执行异步相似性搜索。
similarity_search_by_vector	针对查询字符串执行相似性搜索。
asimilarity_search_by_vector	针对查询向量执行异步相似性搜索。
max_marginal_relevance_search	执行搜索并返回由 MMR 重新排序的结果。
amax_marginal_relevance_search	执行异步搜索并返回由 MMR 重新排序的结果。
max_marginal_relevance_search_by_vector	执行搜索并返回由 MMR 重新排序的结果。
amax_marginal_relevance_search_by_vector	执行异步搜索并返回由 MMR 重新排序的结果。
from_documents	返回从文档和嵌入初始化的 `VectorStore`。
afrom_documents	异步返回从文档和嵌入初始化的 `VectorStore`。
from_texts	创建一个 Milvus collection，使用 HNSW 对其进行索引，并插入数据。
afrom_texts	创建一个 Milvus collection，使用 HNSW 对其进行索引，并插入数据
as_retriever	返回从此 `VectorStore` 初始化的 `VectorStoreRetriever`。
add_embeddings	将带有嵌入向量的文本数据插入 Milvus。
similarity_search_with_score_by_vector	对嵌入执行搜索并返回带有分数的结果。
drop	通过删除（唯一的）collection 来删除索引中的所有内容。
get_pks	使用表达式获取主键
upsert	更新/插入文档到向量存储。
search_by_metadata	根据元数据条件搜索 Milvus 向量存储。
aadd_embeddings	异步将带有嵌入向量的文本数据插入 Milvus。
asimilarity_search_with_score_by_vector	对嵌入执行异步搜索并返回带有分数的结果。
aget_pks	异步使用表达式获取主键
aupsert	异步更新/插入文档到向量存储。
asearch_by_metadata	异步根据元数据条件搜索 Milvus 向量存储。
__init__	初始化 Milvus 向量存储。

embeddings property

embeddings: EmbeddingType | list[EmbeddingType] | None

获取嵌入函数。

client property

client: MilvusClient

获取客户端。

aclient property

aclient: AsyncMilvusClient

获取异步客户端。

vector_fields property

vector_fields: list[str]

获取向量字段。

add_texts

add_texts(
    texts: Iterable[str],
    metadatas: list[dict] | None = None,
    timeout: float | None = None,
    batch_size: int = 1000,
    *,
    ids: list[str] | None = None,
    **kwargs: Any,
) -> list[str]

将文本数据插入 Milvus。

在尚未创建集合（collection）时插入数据，将导致创建一个新的集合。第一个实体的数据决定了新集合的模式（schema），维度（dim）从第一个嵌入（embedding）中提取，列（columns）由第一个元数据字典决定。元数据键（Metadata keys）需要存在于所有插入的值中。目前 Milvus 中没有与 None 等效的值。

参数	描述
texts	要嵌入的文本，假定它们全部都能装入内存。 类型： Iterable[str]
metadatas	附加到每个文本的元数据字典。默认为 None。 类型： list[dict] | None 默认值： None
timeout	每次批量插入的超时时间。默认为 None。 类型： float | None 默认值： None
batch_size	用于插入的批处理大小。默认为 1000。 类型： int 默认值： 1000
ids	文本 ID 列表。每个项目的长度 类型: list[str] | None 默认值: None

引发	描述
MilvusException	添加文本失败

返回	描述
list[str]	List[str]：每个插入元素的最终键列表。

delete

delete(
    ids: list[str] | None = None, expr: str | None = None, **kwargs: str
) -> bool | None

通过向量 ID 或布尔表达式删除。有关表达式的说明和示例，请参阅 Milvus 文档。

参数	描述
ids	要删除的 ID 列表。 类型: list[str] | None 默认值: None
expr	指定要删除的实体的布尔表达式。 类型: str | None 默认值: None
kwargs	Milvus 删除 API 中的其他参数。 类型： str 默认值： {}

返回	描述
bool | None	Optional[bool]：如果删除成功，则为 True，
bool | None	否则为 False。

get_by_ids

get_by_ids(ids: Sequence[str]) -> list[Document]

根据 ID 获取文档。

返回的文档应将其 ID 字段设置为文档在向量存储中的 ID。

如果某些 ID 未找到或存在重复的 ID，返回的文档数量可能少于请求的数量。

用户不应假设返回文档的顺序与输入 ID 的顺序相匹配。相反，用户应依赖于返回文档的 ID 字段。

如果某些 ID 未找到文档，此方法不应**抛出异常**。

参数	描述
ids	要检索的 ID 列表。 类型： Sequence[str]

返回	描述
list[Document]	`Document` 对象列表。

aget_by_ids async

aget_by_ids(ids: Sequence[str]) -> list[Document]

通过 ID 异步获取文档。

返回的文档应将其 ID 字段设置为文档在向量存储中的 ID。

如果某些 ID 未找到或存在重复的 ID，返回的文档数量可能少于请求的数量。

用户不应假设返回文档的顺序与输入 ID 的顺序相匹配。相反，用户应依赖于返回文档的 ID 字段。

如果某些 ID 未找到文档，此方法不应**抛出异常**。

参数	描述
ids	要检索的 ID 列表。 类型： Sequence[str]

返回	描述
list[Document]	`Document` 对象列表。

adelete async

adelete(
    ids: list[str] | None = None, expr: str | None = None, **kwargs: Any
) -> bool | None

通过向量 ID 或布尔表达式异步删除。有关表达式的说明和示例，请参阅 Milvus 文档。

参数	描述
ids	要删除的 ID 列表。 类型: list[str] | None 默认值: None
expr	指定要删除的实体的布尔表达式。 类型: str | None 默认值: None
kwargs	Milvus 删除 API 中的其他参数。 类型： Any 默认值： {}

返回	描述
bool | None	Optional[bool]：如果删除成功，则为 True，
bool | None	否则为 False。

aadd_texts async

aadd_texts(
    texts: Iterable[str],
    metadatas: list[dict] | None = None,
    timeout: float | None = None,
    batch_size: int = 1000,
    *,
    ids: list[str] | None = None,
    **kwargs: Any,
) -> list[str]

异步将文本数据插入 Milvus。

在尚未创建集合（collection）时插入数据，将导致创建一个新的集合。第一个实体的数据决定了新集合的模式（schema），维度（dim）从第一个嵌入（embedding）中提取，列（columns）由第一个元数据字典决定。元数据键（Metadata keys）需要存在于所有插入的值中。目前 Milvus 中没有与 None 等效的值。

参数	描述
texts	要嵌入的文本，假定它们全部都能装入内存。 类型： Iterable[str]
metadatas	附加到每个文本的元数据字典。默认为 None。 类型： list[dict] | None 默认值： None
timeout	每次批量插入的超时时间。默认为 None。 类型： float | None 默认值： None
batch_size	用于插入的批处理大小。默认为 1000。 类型： int 默认值： 1000
ids	文本 ID 列表。每个项目的长度 类型: list[str] | None 默认值: None

引发	描述
MilvusException	添加文本失败

返回	描述
list[str]	List[str]：每个插入元素的最终键列表。

add_documents

add_documents(documents: list[Document], **kwargs: Any) -> list[str]

通过嵌入运行更多文档并将其添加到向量存储中。

参数	描述
documents	要添加到向量存储的文档。 TYPE: list[Document]

返回	描述
list[str]	已添加文本的 ID 列表。

aadd_documents async

aadd_documents(documents: list[Document], **kwargs: Any) -> list[str]

通过嵌入处理更多文档并异步添加到向量存储。

参数	描述
documents	要添加到向量存储的文档。 TYPE: list[Document]

返回	描述
list[str]	已添加文本的 ID 列表。

search

search(query: str, search_type: str, **kwargs: Any) -> list[Document]

使用指定的搜索类型返回与查询最相似的文档。

参数	描述
query	输入文本。 类型： str
search_type	要执行的搜索类型。可以是 `'similarity'`、`'mmr'` 或 `'similarity_score_threshold'`。 类型： str
**kwargs	传递给搜索方法的参数。 类型： Any 默认值： {}

返回	描述
list[Document]	与查询最相似的 `Document` 对象列表。

引发	描述
ValueError	如果 `search_type` 不是 `'similarity'`、`'mmr'` 或 `'similarity_score_threshold'` 之一。

asearch async

asearch(query: str, search_type: str, **kwargs: Any) -> list[Document]

异步返回与查询最相似的文档，使用指定的搜索类型。

参数	描述
query	输入文本。 类型： str
search_type	要执行的搜索类型。可以是 `'similarity'`、`'mmr'` 或 `'similarity_score_threshold'`。 类型： str
**kwargs	传递给搜索方法的参数。 类型： Any 默认值： {}

返回	描述
list[Document]	与查询最相似的 `Document` 对象列表。

引发	描述
ValueError	如果 `search_type` 不是 `'similarity'`、`'mmr'` 或 `'similarity_score_threshold'` 之一。

similarity_search

similarity_search(
    query: str,
    k: int = 4,
    param: dict | list[dict] | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

针对查询字符串执行相似性搜索。

参数	描述
query	要搜索的文本。 类型： str
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
param	索引类型的搜索参数。默认为 None。 类型： dict | list[dict] 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： int 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

similarity_search_with_score

similarity_search_with_score(
    query: str,
    k: int = 4,
    param: dict | list[dict] | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[tuple[Document, float]]

对查询字符串执行搜索并返回带有分数的结果。

有关搜索参数的更多信息，请参阅此处的 pymilvus 文档：https://milvus.org.cn/api-reference/pymilvus/v2.5.x/ORM/Collection/search.md

参数	描述
query	正在搜索的文本。 类型： str
k	要返回的结果数量。默认为 4。 TYPE: int DEFAULT: 4
param	指定索引的搜索参数。 类型： dict | list[dict] 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 或 hybrid_search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	List[Tuple[Document, float]]：结果文档和分数的列表。

asimilarity_search_with_score async

asimilarity_search_with_score(
    query: str,
    k: int = 4,
    param: dict | list[dict] | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[tuple[Document, float]]

对查询字符串执行异步搜索并返回带有分数的结果。

有关搜索参数的更多信息，请参阅此处的 pymilvus 文档：https://milvus.org.cn/api-reference/pymilvus/v2.5.x/ORM/Collection/search.md

参数	描述
query	正在搜索的文本。 类型： str
k	要返回的结果数量。默认为 4。 TYPE: int DEFAULT: 4
param	指定索引的搜索参数。 类型： dict | list[dict] 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 或 hybrid_search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	List[Tuple[Document, float]]：结果文档和分数的列表。

similarity_search_with_relevance_scores

similarity_search_with_relevance_scores(
    query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]

返回文档和在 `[0, 1]` 范围内的相关性分数。

`0` 表示不相似，`1` 表示最相似。

参数	描述
query	输入文本。 类型： str
k	要返回的 `Document` 对象数量。 TYPE: int DEFAULT: 4
**kwargs	将传递给相似性搜索的kwargs。应包括`score_threshold`，一个可选的浮点值，介于`0`到`1`之间，用于筛选检索到的文档结果集。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	由 `(doc, similarity_score)` 组成的元组列表。

asimilarity_search_with_relevance_scores async

asimilarity_search_with_relevance_scores(
    query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]

异步返回文档和在 `[0, 1]` 范围内的相关性分数。

`0` 表示不相似，`1` 表示最相似。

参数	描述
query	输入文本。 类型： str
k	要返回的 `Document` 对象数量。 TYPE: int DEFAULT: 4
**kwargs	将传递给相似性搜索的kwargs。应包括`score_threshold`，一个可选的浮点值，介于`0`到`1`之间，用于筛选检索到的文档结果集。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	元组列表 `(doc, similarity_score)`

asimilarity_search async

asimilarity_search(
    query: str,
    k: int = 4,
    param: dict | list[dict] | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

针对查询字符串执行异步相似性搜索。

参数	描述
query	要搜索的文本。 类型： str
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
param	索引类型的搜索参数。默认为 None。 类型： dict | list[dict] 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： int 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

similarity_search_by_vector

similarity_search_by_vector(
    embedding: list[float],
    k: int = 4,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

针对查询字符串执行相似性搜索。

参数	描述
embedding	要搜索的嵌入向量。 类型： list[float]
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
param	索引类型的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： int 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

asimilarity_search_by_vector async

asimilarity_search_by_vector(
    embedding: list[float],
    k: int = 4,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

针对查询向量执行异步相似性搜索。

参数	描述
embedding	要搜索的嵌入向量。 类型： list[float]
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
param	索引类型的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： int 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

max_marginal_relevance_search

max_marginal_relevance_search(
    query: str,
    k: int = 4,
    fetch_k: int = 20,
    lambda_mult: float = 0.5,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

执行搜索并返回由 MMR 重新排序的结果。

参数	描述
query	正在搜索的文本。 类型： str
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
fetch_k	从中选择 k 个结果的总结果数。默认为 20。 类型： int 默认值： 20
lambda_mult	一个介于 0 和 1 之间的数字，用于确定结果之间的多样性程度，其中 0 对应最大多样性，1 对应最小多样性。默认为 0.5 类型： float 默认值： 0.5
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

amax_marginal_relevance_search async

amax_marginal_relevance_search(
    query: str,
    k: int = 4,
    fetch_k: int = 20,
    lambda_mult: float = 0.5,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

执行异步搜索并返回由 MMR 重新排序的结果。

参数	描述
query	正在搜索的文本。 类型： str
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
fetch_k	从中选择 k 个结果的总结果数。默认为 20。 类型： int 默认值： 20
lambda_mult	一个介于 0 和 1 之间的数字，用于确定结果之间的多样性程度，其中 0 对应最大多样性，1 对应最小多样性。默认为 0.5 类型： float 默认值： 0.5
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

max_marginal_relevance_search_by_vector

max_marginal_relevance_search_by_vector(
    embedding: list[float] | dict[int, float],
    k: int = 4,
    fetch_k: int = 20,
    lambda_mult: float = 0.5,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

执行搜索并返回由 MMR 重新排序的结果。

参数	描述
embedding	正在搜索的嵌入向量。 类型： list[float] | dict[int, float]
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
fetch_k	从中选择 k 个结果的总结果数。默认为 20。 类型： int 默认值： 20
lambda_mult	一个介于 0 和 1 之间的数字，用于确定结果之间的多样性程度，其中 0 对应最大多样性，1 对应最小多样性。默认为 0.5 类型： float 默认值： 0.5
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

amax_marginal_relevance_search_by_vector async

amax_marginal_relevance_search_by_vector(
    embedding: list[float] | dict[int, float],
    k: int = 4,
    fetch_k: int = 20,
    lambda_mult: float = 0.5,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[Document]

执行异步搜索并返回由 MMR 重新排序的结果。

参数	描述
embedding	正在搜索的嵌入向量。 类型： list[float] | dict[int, float]
k	返回多少个结果。默认为 4。 TYPE: int DEFAULT: 4
fetch_k	从中选择 k 个结果的总结果数。默认为 20。 类型： int 默认值： 20
lambda_mult	一个介于 0 和 1 之间的数字，用于确定结果之间的多样性程度，其中 0 对应最大多样性，1 对应最小多样性。默认为 0.5 类型： float 默认值： 0.5
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[Document]	List[Document]：搜索的文档结果。

from_documents classmethod

from_documents(documents: list[Document], embedding: Embeddings, **kwargs: Any) -> Self

返回从文档和嵌入初始化的 `VectorStore`。

参数	描述
documents	要添加到 `VectorStore` 的 `Document` 对象列表。 TYPE: list[Document]
embedding	要使用的嵌入函数。 TYPE: Embeddings
**kwargs	附加的关键字参数。 类型： Any 默认值： {}

返回	描述
Self	从文档和嵌入初始化的 `VectorStore`。

afrom_documents async classmethod

afrom_documents(
    documents: list[Document], embedding: Embeddings, **kwargs: Any
) -> Self

异步返回从文档和嵌入初始化的 `VectorStore`。

参数	描述
documents	要添加到 `VectorStore` 的 `Document` 对象列表。 TYPE: list[Document]
embedding	要使用的嵌入函数。 TYPE: Embeddings
**kwargs	附加的关键字参数。 类型： Any 默认值： {}

返回	描述
Self	从文档和嵌入初始化的 `VectorStore`。

from_texts classmethod

from_texts(
    texts: list[str],
    embedding: EmbeddingType | list[EmbeddingType] | None,
    metadatas: list[dict] | None = None,
    collection_name: str = "LangChainCollection",
    connection_args: dict[str, Any] | None = None,
    consistency_level: str = "Session",
    index_params: dict | list[dict] | None = None,
    search_params: dict | list[dict] | None = None,
    drop_old: bool = False,
    *,
    ids: list[str] | None = None,
    auto_id: bool = False,
    builtin_function: BaseMilvusBuiltInFunction
    | list[BaseMilvusBuiltInFunction]
    | None = None,
    **kwargs: Any,
) -> Milvus

创建一个 Milvus collection，使用 HNSW 对其进行索引，并插入数据。

参数	描述
texts	文本数据。 类型: list[str]
embedding	嵌入函数。 类型： Embeddings | BaseSparseEmbedding | None
metadatas	每个文本的元数据（如果存在）。默认为 None。 类型： list[dict] | None 默认值： None
collection_name	要使用的集合名称。默认为 "LangChainCollection"。 类型： str 默认值： 'LangChainCollection'
connection_args	要使用的连接参数。默认为 DEFAULT_MILVUS_CONNECTION。 类型： dict[str, Any] 默认值： None
consistency_level	使用哪种一致性级别。默认为 "Session"。 类型： str 默认值： 'Session'
index_params	使用哪个 index_params。默认为 None。 类型： dict | None 默认值： None
search_params	使用哪个 search_params。默认为 None。 类型： dict | None 默认值： None
drop_old	如果存在同名集合，是否删除它。默认为 False。 类型： bool | None 默认值： False
ids	文本 ID 列表。默认为 None。 类型: list[str] | None 默认值: None
auto_id	是否为主键启用自动 ID。默认为 False。如果为 False，您需要提供文本 ID（小于 65535 字节的字符串）。如果为 True，Milvus 将生成唯一的整数作为主键。 类型： bool 默认值： False
**kwargs	Milvus 集合中的其他参数。 类型： Any 默认值： {}

返回： Milvus：Milvus 向量存储

afrom_texts async classmethod

afrom_texts(
    texts: list[str],
    embedding: EmbeddingType | list[EmbeddingType] | None,
    metadatas: list[dict] | None = None,
    collection_name: str = "LangChainCollection",
    connection_args: dict[str, Any] | None = None,
    consistency_level: str = "Session",
    index_params: dict | list[dict] | None = None,
    search_params: dict | list[dict] | None = None,
    drop_old: bool = False,
    *,
    ids: list[str] | None = None,
    auto_id: bool = False,
    builtin_function: BaseMilvusBuiltInFunction
    | list[BaseMilvusBuiltInFunction]
    | None = None,
    **kwargs: Any,
) -> Milvus

创建一个 Milvus 集合，使用 HNSW 对其进行索引，并异步插入数据。

参数	描述
texts	文本数据。 类型: list[str]
embedding	嵌入函数。 类型： Embeddings | BaseSparseEmbedding | None
metadatas	每个文本的元数据（如果存在）。默认为 None。 类型： list[dict] | None 默认值： None
collection_name	要使用的集合名称。默认为 "LangChainCollection"。 类型： str 默认值： 'LangChainCollection'
connection_args	要使用的连接参数。默认为 DEFAULT_MILVUS_CONNECTION。 类型： dict[str, Any] 默认值： None
consistency_level	使用哪种一致性级别。默认为 "Session"。 类型： str 默认值： 'Session'
index_params	使用哪个 index_params。默认为 None。 类型： dict | None 默认值： None
search_params	使用哪个 search_params。默认为 None。 类型： dict | None 默认值： None
drop_old	如果存在同名集合，是否删除它。默认为 False。 类型： bool | None 默认值： False
ids	文本 ID 列表。默认为 None。 类型: list[str] | None 默认值: None
auto_id	是否为主键启用自动 ID。默认为 False。如果为 False，您需要提供文本 ID（小于 65535 字节的字符串）。如果为 True，Milvus 将生成唯一的整数作为主键。 类型： bool 默认值： False
**kwargs	Milvus 集合中的其他参数。 类型： Any 默认值： {}

返回： Milvus：Milvus 向量存储

as_retriever

as_retriever(**kwargs: Any) -> VectorStoreRetriever

返回从此 `VectorStore` 初始化的 `VectorStoreRetriever`。

参数	描述
**kwargs	传递给搜索函数的关键字参数。可以包括 `search_type`：定义检索器应执行的搜索类型。可以是 `'similarity'` (默认)、`'mmr'` 或 `'similarity_score_threshold'`。 `search_kwargs`：传递给搜索函数的关键字参数。可以包括诸如 `k`：要返回的文档数量（默认：`4`） `score_threshold`：`similarity_score_threshold` 的最小相关性阈值 `fetch_k`：传递给 MMR 算法的文档数量（默认值：`20`） `lambda_mult`：MMR 返回结果的多样性；`1` 表示最小多样性，0 表示最大多样性。（默认值：`0.5`） `filter`：按文档元数据过滤 类型： Any 默认值： {}

返回	描述
VectorStoreRetriever	`VectorStore` 的检索器类。

示例

# Retrieve more documents with higher diversity
# Useful if your dataset has many similar documents
docsearch.as_retriever(
    search_type="mmr", search_kwargs={"k": 6, "lambda_mult": 0.25}
)

# Fetch more documents for the MMR algorithm to consider
# But only return the top 5
docsearch.as_retriever(search_type="mmr", search_kwargs={"k": 5, "fetch_k": 50})

# Only retrieve documents that have a relevance score
# Above a certain threshold
docsearch.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.8},
)

# Only get the single most similar document from the dataset
docsearch.as_retriever(search_kwargs={"k": 1})

# Use a filter to only retrieve documents from a specific paper
docsearch.as_retriever(
    search_kwargs={"filter": {"paper_title": "GPT-4 Technical Report"}}
)

add_embeddings

add_embeddings(
    texts: list[str],
    embeddings: List[List[float]] | List[List[List[float]]],
    metadatas: list[dict] | None = None,
    timeout: float | None = None,
    batch_size: int = 1000,
    *,
    ids: list[str] | None = None,
    **kwargs: Any,
) -> list[str]

将带有嵌入向量的文本数据插入 Milvus。

此方法将一批文本嵌入插入到 Milvus 集合中。如果集合尚未初始化，它将根据嵌入、元数据和其他参数自动初始化集合。嵌入应使用兼容的嵌入函数预先生成，与每个文本关联的元数据是可选的，但数量必须与文本数量匹配。

参数	描述
texts	要插入的文本 类型: list[str]
embeddings	每个文本的向量嵌入（如果是单个向量）或每个文本的向量列表（如果是多向量）。 类型： List[List[float]] | List[List[List[float]]]
metadatas	附加到每个文本的元数据字典。默认为 None。 类型： list[dict] | None 默认值： None
timeout	每次批量插入的超时时间。默认为 None。 类型： float | None 默认值： None
batch_size	用于插入的批处理大小。默认为 1000。 类型： int 默认值： 1000
ids	文本 ID 列表。每个项目的长度 类型: list[str] | None 默认值: None

引发	描述
MilvusException	添加文本和嵌入失败

返回	描述
list[str]	List[str]：每个插入元素的最终键列表。

similarity_search_with_score_by_vector

similarity_search_with_score_by_vector(
    embedding: List[float] | Dict[int, float],
    k: int = 4,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[tuple[Document, float]]

对嵌入执行搜索并返回带有分数的结果。

有关搜索参数的更多信息，请参阅此处的 pymilvus 文档：https://milvus.org.cn/api-reference/pymilvus/v2.5.x/ORM/Collection/search.md

参数	描述
embedding	正在搜索的嵌入向量。 类型： List[float] | Dict[int, float]
k	要返回的结果数量。默认为 4。 TYPE: int DEFAULT: 4
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	List[Tuple[Document, float]]：结果文档和分数。

drop

drop() -> None

通过删除（唯一的）collection 来删除索引中的所有内容。

get_pks

get_pks(expr: str, **kwargs: Any) -> List[int] | None

使用表达式获取主键

参数	描述
expr	表达式 - 例如："id in [1, 2]" 或 "title LIKE 'Abc%'" 类型： str

返回	描述
List[int] | None	List[int]：ID 列表（主键）

upsert

upsert(
    ids: list[str] | None = None,
    documents: List[Document] | None = None,
    batch_size: int = 1000,
    timeout: float | None = None,
    **kwargs: Any,
) -> None

更新/插入文档到向量存储。

参数	描述
ids	要更新的 ID - 调用 get_pks 以通过表达式获取 ID 类型: list[str] | None 默认值: None
documents	要添加到向量存储的文档。 类型： list[Document] 默认值： None

search_by_metadata

search_by_metadata(
    expr: str, fields: list[str] | None = None, limit: int = 10
) -> list[Document]

根据元数据条件搜索 Milvus 向量存储。

此函数使用表达式执行基于元数据的查询，该表达式过滤存储的文档，而无需进行向量相似度比较。

参数	描述
expr	一个过滤表达式（例如，`"city == 'Seoul'"`）。 类型： str
fields	要检索的字段列表。如果为 None，则检索所有可用字段。 类型: list[str] | None 默认值: None
limit	要返回的最大结果数。 类型： int 默认值： 10

返回	描述
list[Document]	List[Document]：匹配元数据过滤器的文档列表。

aadd_embeddings async

aadd_embeddings(
    texts: list[str],
    embeddings: List[List[float]] | List[List[List[float]]],
    metadatas: list[dict] | None = None,
    timeout: float | None = None,
    batch_size: int = 1000,
    *,
    ids: list[str] | None = None,
    **kwargs: Any,
) -> list[str]

异步将带有嵌入向量的文本数据插入 Milvus。

此方法将一批文本嵌入插入到 Milvus 集合中。如果集合尚未初始化，它将根据嵌入、元数据和其他参数自动初始化集合。嵌入应使用兼容的嵌入函数预先生成，与每个文本关联的元数据是可选的，但数量必须与文本数量匹配。

参数	描述
texts	要插入的文本 类型: list[str]
embeddings	每个文本的向量嵌入（如果是单个向量）或每个文本的向量列表（如果是多向量）。 类型： List[List[float]] | List[List[List[float]]]
metadatas	附加到每个文本的元数据字典。默认为 None。 类型： list[dict] | None 默认值： None
timeout	每次批量插入的超时时间。默认为 None。 类型： float | None 默认值： None
batch_size	用于插入的批处理大小。默认为 1000。 类型： int 默认值： 1000
ids	文本 ID 列表。每个项目的长度 类型: list[str] | None 默认值: None

引发	描述
MilvusException	添加文本和嵌入失败

返回	描述
list[str]	List[str]：每个插入元素的最终键列表。

asimilarity_search_with_score_by_vector async

asimilarity_search_with_score_by_vector(
    embedding: List[float] | Dict[int, float],
    k: int = 4,
    param: dict | None = None,
    expr: str | None = None,
    timeout: float | None = None,
    **kwargs: Any,
) -> list[tuple[Document, float]]

对嵌入执行异步搜索并返回带有分数的结果。

有关搜索参数的更多信息，请参阅此处的 pymilvus 文档：https://milvus.org.cn/api-reference/pymilvus/v2.5.x/ORM/Collection/search.md

参数	描述
embedding	正在搜索的嵌入向量。 类型： List[float] | Dict[int, float]
k	要返回的结果数量。默认为 4。 TYPE: int DEFAULT: 4
param	指定索引的搜索参数。默认为 None。 类型： dict 默认值： None
expr	过滤表达式。默认为 None。 类型： str 默认值： None
timeout	在超时错误前等待多长时间。默认为 None。 类型： float 默认值： None
kwargs	Collection.search() 的关键字参数。 类型： Any 默认值： {}

返回	描述
list[tuple[Document, float]]	List[Tuple[Document, float]]：结果文档和分数。

aget_pks async

aget_pks(expr: str, **kwargs: Any) -> List[int] | None

异步使用表达式获取主键

参数	描述
expr	表达式 - 例如："id in [1, 2]" 或 "title LIKE 'Abc%'" 类型： str

返回	描述
List[int] | None	List[int]：ID 列表（主键）

aupsert async

aupsert(
    ids: list[str] | None = None,
    documents: List[Document] | None = None,
    batch_size: int = 1000,
    timeout: float | None = None,
    **kwargs: Any,
) -> None

异步更新/插入文档到向量存储。

参数	描述
ids	要更新的 ID - 调用 aget_pks 以通过表达式获取 ID 类型: list[str] | None 默认值: None
documents	要添加到向量存储的文档。 类型： list[Document] 默认值： None
batch_size	用于 upsert 的批处理大小。默认为 1000。 类型： int 默认值： 1000
timeout	每次批量 upsert 的超时时间。默认为 None。 类型： float | None 默认值： None
**kwargs	Milvus upsert API 中的其他参数。 类型： Any 默认值： {}

asearch_by_metadata async

asearch_by_metadata(
    expr: str, fields: list[str] | None = None, limit: int = 10
) -> list[Document]

异步根据元数据条件搜索 Milvus 向量存储。

此函数使用表达式执行基于元数据的查询，该表达式过滤存储的文档，而无需进行向量相似度比较。

参数	描述
expr	一个过滤表达式（例如，`"city == 'Seoul'"`）。 类型： str
fields	要检索的字段列表。如果为 None，则检索所有可用字段。 类型: list[str] | None 默认值: None
limit	要返回的最大结果数。 类型： int 默认值： 10

返回	描述
list[Document]	List[Document]：匹配元数据过滤器的文档列表。

__init__

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

初始化 Milvus 向量存储。