跳转到内容

LangSmith 实用工具

langchain-classic 文档

本文档涵盖 langchain-classic 包。该包将持续维护安全漏洞至 2026 年 12 月。我们鼓励用户迁移到 langchain 包以获取最新的功能和改进。查看 langchain 的文档

langchain_classic.smith

LangSmith 实用工具。

该模块提供用于连接 LangSmith 的实用工具。

评估

LangSmith 帮助您使用多种 LangChain 评估器来评估链(Chains)和其他语言模型应用组件。下面展示了一个示例,假设您已创建了一个名为 <my_dataset_name> 的 LangSmith 数据集。

from langsmith import Client
from langchain_openai import ChatOpenAI
from langchain_classic.chains import LLMChain
from langchain_classic.smith import RunEvalConfig, run_on_dataset


# Chains may have memory. Passing in a constructor function lets the
# evaluation framework avoid cross-contamination between runs.
def construct_chain():
    model = ChatOpenAI(temperature=0)
    chain = LLMChain.from_string(model, "What's the answer to {your_input_key}")
    return chain


# Load off-the-shelf evaluators via config or the EvaluatorType (string or enum)
evaluation_config = RunEvalConfig(
    evaluators=[
        "qa",  # "Correctness" against a reference answer
        "embedding_distance",
        RunEvalConfig.Criteria("helpfulness"),
        RunEvalConfig.Criteria(
            {
                "fifth-grader-score": "Do you have to be smarter than a fifth "
                "grader to answer this question?"
            }
        ),
    ]
)

client = Client()
run_on_dataset(
    client,
    "<my_dataset_name>",
    construct_chain,
    evaluation=evaluation_config,
)

您还可以通过继承 StringEvaluator <langchain.evaluation.schema.StringEvaluator> 或 LangSmith 的 RunEvaluator 类来创建自定义评估器。

from typing import Optional
from langchain_classic.evaluation import StringEvaluator


class MyStringEvaluator(StringEvaluator):
    @property
    def requires_input(self) -> bool:
        return False

    @property
    def requires_reference(self) -> bool:
        return True

    @property
    def evaluation_name(self) -> str:
        return "exact_match"

    def _evaluate_strings(
        self, prediction, reference=None, input=None, **kwargs
    ) -> dict:
        return {"score": prediction == reference}


evaluation_config = RunEvalConfig(
    custom_evaluators=[MyStringEvaluator()],
)

run_on_dataset(
    client,
    "<my_dataset_name>",
    construct_chain,
    evaluation=evaluation_config,
)

主要函数

  • arun_on_dataset <langchain.smith.evaluation.runner_utils.arun_on_dataset>:一个异步函数,用于在数据集上评估链、代理或其他 LangChain 组件。
  • run_on_dataset <langchain.smith.evaluation.runner_utils.run_on_dataset>:一个函数,用于在数据集上评估链、代理或其他 LangChain 组件。
  • RunEvalConfig <langchain.smith.evaluation.config.RunEvalConfig>:代表运行评估配置的类。您可以通过 EvaluatorType <langchain.evaluation.schema.EvaluatorType> 或配置来选择评估器,也可以传入 custom_evaluators
函数 描述
arun_on_dataset

在数据集上运行。

run_on_dataset

在数据集上运行。

RunEvalConfig

基类: BaseModel

运行评估的配置。

evaluators class-attribute instance-attribute

evaluators: list[SINGLE_EVAL_CONFIG_TYPE | CUSTOM_EVALUATOR_TYPE] = Field(
    default_factory=list
)

配置哪些评估器应用于数据集运行。每个评估器可以是 EvaluatorType <langchain.evaluation.schema.EvaluatorType> 的字符串,例如 EvaluatorType.QA、评估器类型字符串("qa"),或给定评估器的配置(例如 RunEvalConfig.QA <langchain.smith.evaluation.config.RunEvalConfig.QA>)。

custom_evaluators class-attribute instance-attribute

custom_evaluators: list[CUSTOM_EVALUATOR_TYPE] | None = None

应用于数据集运行的自定义评估器。

batch_evaluators class-attribute instance-attribute

batch_evaluators: list[BATCH_EVALUATOR_LIKE] | None = None

在聚合/批处理级别运行的评估器。

这些评估器会生成一个或多个指标,这些指标被分配给整个测试运行。因此,它们不与单个跟踪(trace)相关联。

reference_key class-attribute instance-attribute

reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute

prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute

input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

eval_llm class-attribute instance-attribute

eval_llm: BaseLanguageModel | None = None

传递给任何需要语言模型的评估器的语言模型。

Criteria

基类:SingleKeyEvalConfig

无参考标准评估器的配置。

属性 描述
criteria

要评估的标准。

类型: CRITERIA_TYPE | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

LabeledCriteria

基类:SingleKeyEvalConfig

带标签(有参考)的标准评估器的配置。

属性 描述
criteria

要评估的标准。

类型: CRITERIA_TYPE | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

EmbeddingDistance

基类:SingleKeyEvalConfig

嵌入距离评估器的配置。

属性 描述
embeddings

用于计算距离的嵌入模型。

类型: Embeddings | None

distance_metric

用于计算距离的度量标准。

类型: EmbeddingDistance | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

StringDistance

基类:SingleKeyEvalConfig

字符串距离评估器的配置。

属性 描述
distance

要使用的字符串距离度量(damerau_levenshteinlevenshteinjarojaro_winkler)。

类型: StringDistance | None

normalize_score

是否将距离归一化到 0 和 1 之间。仅适用于 Levenshtein 和 Damerau-Levenshtein 距离。

类型: bool

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

QA

基类:SingleKeyEvalConfig

问答(QA)评估器的配置。

属性 描述
prompt

用于生成问题的提示模板。

类型: BasePromptTemplate | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

ContextQA

基类:SingleKeyEvalConfig

基于上下文的问答(QA)评估器的配置。

属性 描述
prompt

用于生成问题的提示模板。

类型: BasePromptTemplate | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

CoTQA

基类:SingleKeyEvalConfig

基于上下文的问答(QA)评估器的配置。

属性 描述
prompt

用于生成问题的提示模板。

类型: BasePromptTemplate | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

JsonValidity

基类:SingleKeyEvalConfig

JSON 有效性评估器的配置。

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

JsonEqualityEvaluator

基类:EvalConfig

JSON 相等性评估器的配置。

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

ExactMatch

基类:SingleKeyEvalConfig

精确匹配字符串评估器的配置。

属性 描述
ignore_case

比较字符串时是否忽略大小写。

类型: bool

ignore_punctuation

比较字符串时是否忽略标点符号。

类型: bool

ignore_numbers

比较字符串时是否忽略数字。

类型: bool

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

RegexMatch

基类:SingleKeyEvalConfig

正则表达式匹配字符串评估器的配置。

属性 描述
flags

传递给正则表达式的标志。示例:re.IGNORECASE

类型: int

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

ScoreString

基类:SingleKeyEvalConfig

字符串评分评估器的配置。

这类似于标准评估器,但默认配置为返回 1-10 范围内的分数。

建议通过将 normalize_by 设置为 10 来对这些分数进行归一化。

属性 描述
criteria

要评估的标准。

类型: CRITERIA_TYPE | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

normalize_by

如果要对分数进行归一化,则使用此分母。如果未提供,分数将在 1 到 10 之间。

类型: float | None

prompt

用于评估的提示模板。

类型: BasePromptTemplate | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

LabeledScoreString

基类:ScoreString

带标签的字符串评分评估器的配置。

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

arun_on_dataset async

arun_on_dataset(
    client: Client | None,
    dataset_name: str,
    llm_or_chain_factory: MODEL_OR_CHAIN_FACTORY,
    *,
    evaluation: RunEvalConfig | None = None,
    dataset_version: datetime | str | None = None,
    concurrency_level: int = 5,
    project_name: str | None = None,
    project_metadata: dict[str, Any] | None = None,
    verbose: bool = False,
    revision_id: str | None = None,
    **kwargs: Any,
) -> dict[str, Any]

在数据集上运行。

在数据集上运行链(Chain)或语言模型,并将跟踪信息存储到指定的项目名称中。

有关此函数(通常更快)的异步版本,请参阅 arun_on_dataset

参数 描述
dataset_name

要在其上运行链的数据集的名称。

类型: str

llm_or_chain_factory

用于在数据集上运行的语言模型或链构造函数。链构造函数用于允许对每个示例进行独立调用,而不会继承状态。

类型: MODEL_OR_CHAIN_FACTORY

evaluation

用于对链的结果运行评估器的配置。

类型: RunEvalConfig | None 默认值: None

dataset_version

数据集的可选版本。

类型: datetime | str | None 默认值: None

concurrency_level

并发运行的异步任务数。

类型: int 默认值: 5

project_name

用于存储跟踪信息的项目名称。默认为 {dataset_name}-{chain class name}-{datetime}

类型: str | None 默认值: None

project_metadata

要添加到项目的可选元数据。可用于存储测试变体的信息。(例如提示版本、模型版本等)

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

client

用于访问数据集以及记录反馈和运行跟踪的 LangSmith 客户端。

类型: Client | None

verbose

是否打印进度。

类型: bool 默认值: False

revision_id

可选的修订标识符,用于将此测试运行分配给跟踪系统不同版本性能的标识符。

类型: str | None 默认值: None

**kwargs

不应使用,但为向后兼容而提供。

类型: Any 默认值: {}

返回 描述
dict[str, Any]

包含运行的项目名称和生成的模型输出的 dict

示例

from langsmith import Client
from langchain_openai import ChatOpenAI
from langchain_classic.chains import LLMChain
from langchain_classic.smith import smith_eval.RunEvalConfig, run_on_dataset

# Chains may have memory. Passing in a constructor function lets the
# evaluation framework avoid cross-contamination between runs.
def construct_chain():
    model = ChatOpenAI(temperature=0)
    chain = LLMChain.from_string(
        model,
        "What's the answer to {your_input_key}"
    )
    return chain

# Load off-the-shelf evaluators via config or the EvaluatorType (string or enum)
evaluation_config = smith_eval.RunEvalConfig(
    evaluators=[
        "qa",  # "Correctness" against a reference answer
        "embedding_distance",
        smith_eval.RunEvalConfig.Criteria("helpfulness"),
        smith_eval.RunEvalConfig.Criteria({
            "fifth-grader-score": "Do you have to be smarter than a fifth "
            "grader to answer this question?"
        }),
    ]
)

client = Client()
await arun_on_dataset(
    client,
    dataset_name="<my_dataset_name>",
    llm_or_chain_factory=construct_chain,
    evaluation=evaluation_config,
)
您还可以通过继承 StringEvaluator 或 LangSmith 的RunEvaluator` 类来创建自定义评估器。

from typing import Optional
from langchain_classic.evaluation import StringEvaluator


class MyStringEvaluator(StringEvaluator):
    @property
    def requires_input(self) -> bool:
        return False

    @property
    def requires_reference(self) -> bool:
        return True

    @property
    def evaluation_name(self) -> str:
        return "exact_match"

    def _evaluate_strings(
        self, prediction, reference=None, input=None, **kwargs
    ) -> dict:
        return {"score": prediction == reference}


evaluation_config = smith_eval.RunEvalConfig(
    custom_evaluators=[MyStringEvaluator()],
)

await arun_on_dataset(
    client,
    dataset_name="<my_dataset_name>",
    llm_or_chain_factory=construct_chain,
    evaluation=evaluation_config,
)

run_on_dataset

run_on_dataset(
    client: Client | None,
    dataset_name: str,
    llm_or_chain_factory: MODEL_OR_CHAIN_FACTORY,
    *,
    evaluation: RunEvalConfig | None = None,
    dataset_version: datetime | str | None = None,
    concurrency_level: int = 5,
    project_name: str | None = None,
    project_metadata: dict[str, Any] | None = None,
    verbose: bool = False,
    revision_id: str | None = None,
    **kwargs: Any,
) -> dict[str, Any]

在数据集上运行。

在数据集上运行链(Chain)或语言模型,并将跟踪信息存储到指定的项目名称中。

有关此函数(通常更快)的异步版本,请参阅 arun_on_dataset

参数 描述
dataset_name

要在其上运行链的数据集的名称。

类型: str

llm_or_chain_factory

用于在数据集上运行的语言模型或链构造函数。链构造函数用于允许对每个示例进行独立调用,而不会继承状态。

类型: MODEL_OR_CHAIN_FACTORY

evaluation

用于对链的结果运行评估器的配置。

类型: RunEvalConfig | None 默认值: None

dataset_version

数据集的可选版本。

类型: datetime | str | None 默认值: None

concurrency_level

并发运行的异步任务数。

类型: int 默认值: 5

project_name

用于存储跟踪信息的项目名称。默认为 {dataset_name}-{chain class name}-{datetime}

类型: str | None 默认值: None

project_metadata

要添加到项目的可选元数据。可用于存储测试变体的信息。(例如提示版本、模型版本等)

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

client

用于访问数据集以及记录反馈和运行跟踪的 LangSmith 客户端。

类型: Client | None

verbose

是否打印进度。

类型: bool 默认值: False

revision_id

可选的修订标识符,用于将此测试运行分配给跟踪系统不同版本性能的标识符。

类型: str | None 默认值: None

**kwargs

不应使用,但为向后兼容而提供。

类型: Any 默认值: {}

返回 描述
dict[str, Any]

包含运行的项目名称和生成的模型输出的 dict

示例

from langsmith import Client
from langchain_openai import ChatOpenAI
from langchain_classic.chains import LLMChain
from langchain_classic.smith import smith_eval.RunEvalConfig, run_on_dataset

# Chains may have memory. Passing in a constructor function lets the
# evaluation framework avoid cross-contamination between runs.
def construct_chain():
    model = ChatOpenAI(temperature=0)
    chain = LLMChain.from_string(
        model,
        "What's the answer to {your_input_key}"
    )
    return chain

# Load off-the-shelf evaluators via config or the EvaluatorType (string or enum)
evaluation_config = smith_eval.RunEvalConfig(
    evaluators=[
        "qa",  # "Correctness" against a reference answer
        "embedding_distance",
        smith_eval.RunEvalConfig.Criteria("helpfulness"),
        smith_eval.RunEvalConfig.Criteria({
            "fifth-grader-score": "Do you have to be smarter than a fifth "
            "grader to answer this question?"
        }),
    ]
)

client = Client()
run_on_dataset(
    client,
    dataset_name="<my_dataset_name>",
    llm_or_chain_factory=construct_chain,
    evaluation=evaluation_config,
)

您还可以通过继承 StringEvaluator 或 LangSmith 的 RunEvaluator 类来创建自定义评估器。

from typing import Optional
from langchain_classic.evaluation import StringEvaluator


class MyStringEvaluator(StringEvaluator):
    @property
    def requires_input(self) -> bool:
        return False

    @property
    def requires_reference(self) -> bool:
        return True

    @property
    def evaluation_name(self) -> str:
        return "exact_match"

    def _evaluate_strings(
        self, prediction, reference=None, input=None, **kwargs
    ) -> dict:
        return {"score": prediction == reference}


evaluation_config = smith_eval.RunEvalConfig(
    custom_evaluators=[MyStringEvaluator()],
)

run_on_dataset(
    client,
    dataset_name="<my_dataset_name>",
    llm_or_chain_factory=construct_chain,
    evaluation=evaluation_config,
)

langchain_classic.smith.evaluation.config

运行评估器的配置。

EvalConfig

基类: BaseModel

给定运行评估器的配置。

属性 描述
evaluator_type

要使用的评估器类型。

类型: EvaluatorType

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

get_kwargs

get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

RunEvalConfig

基类: BaseModel

运行评估的配置。

evaluators class-attribute instance-attribute

evaluators: list[SINGLE_EVAL_CONFIG_TYPE | CUSTOM_EVALUATOR_TYPE] = Field(
    default_factory=list
)

配置哪些评估器应用于数据集运行。每个评估器可以是 EvaluatorType <langchain.evaluation.schema.EvaluatorType> 的字符串,例如 EvaluatorType.QA、评估器类型字符串("qa"),或给定评估器的配置(例如 RunEvalConfig.QA <langchain.smith.evaluation.config.RunEvalConfig.QA>)。

custom_evaluators class-attribute instance-attribute

custom_evaluators: list[CUSTOM_EVALUATOR_TYPE] | None = None

应用于数据集运行的自定义评估器。

batch_evaluators class-attribute instance-attribute

batch_evaluators: list[BATCH_EVALUATOR_LIKE] | None = None

在聚合/批处理级别运行的评估器。

这些评估器会生成一个或多个指标,这些指标被分配给整个测试运行。因此,它们不与单个跟踪(trace)相关联。

reference_key class-attribute instance-attribute

reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute

prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute

input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

eval_llm class-attribute instance-attribute

eval_llm: BaseLanguageModel | None = None

传递给任何需要语言模型的评估器的语言模型。

Criteria

基类:SingleKeyEvalConfig

无参考标准评估器的配置。

属性 描述
criteria

要评估的标准。

类型: CRITERIA_TYPE | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

LabeledCriteria

基类:SingleKeyEvalConfig

带标签(有参考)的标准评估器的配置。

属性 描述
criteria

要评估的标准。

类型: CRITERIA_TYPE | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

EmbeddingDistance

基类:SingleKeyEvalConfig

嵌入距离评估器的配置。

属性 描述
embeddings

用于计算距离的嵌入模型。

类型: Embeddings | None

distance_metric

用于计算距离的度量标准。

类型: EmbeddingDistance | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

StringDistance

基类:SingleKeyEvalConfig

字符串距离评估器的配置。

属性 描述
distance

要使用的字符串距离度量(damerau_levenshteinlevenshteinjarojaro_winkler)。

类型: StringDistance | None

normalize_score

是否将距离归一化到 0 和 1 之间。仅适用于 Levenshtein 和 Damerau-Levenshtein 距离。

类型: bool

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

QA

基类:SingleKeyEvalConfig

问答(QA)评估器的配置。

属性 描述
prompt

用于生成问题的提示模板。

类型: BasePromptTemplate | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

ContextQA

基类:SingleKeyEvalConfig

基于上下文的问答(QA)评估器的配置。

属性 描述
prompt

用于生成问题的提示模板。

类型: BasePromptTemplate | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

CoTQA

基类:SingleKeyEvalConfig

基于上下文的问答(QA)评估器的配置。

属性 描述
prompt

用于生成问题的提示模板。

类型: BasePromptTemplate | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

JsonValidity

基类:SingleKeyEvalConfig

JSON 有效性评估器的配置。

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

JsonEqualityEvaluator

基类:EvalConfig

JSON 相等性评估器的配置。

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

ExactMatch

基类:SingleKeyEvalConfig

精确匹配字符串评估器的配置。

属性 描述
ignore_case

比较字符串时是否忽略大小写。

类型: bool

ignore_punctuation

比较字符串时是否忽略标点符号。

类型: bool

ignore_numbers

比较字符串时是否忽略数字。

类型: bool

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

RegexMatch

基类:SingleKeyEvalConfig

正则表达式匹配字符串评估器的配置。

属性 描述
flags

传递给正则表达式的标志。示例:re.IGNORECASE

类型: int

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

ScoreString

基类:SingleKeyEvalConfig

字符串评分评估器的配置。

这类似于标准评估器,但默认配置为返回 1-10 范围内的分数。

建议通过将 normalize_by 设置为 10 来对这些分数进行归一化。

属性 描述
criteria

要评估的标准。

类型: CRITERIA_TYPE | None

llm

用于评估链的语言模型。

类型: BaseLanguageModel | None

normalize_by

如果要对分数进行归一化,则使用此分母。如果未提供,分数将在 1 到 10 之间。

类型: float | None

prompt

用于评估的提示模板。

类型: BasePromptTemplate | None

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

LabeledScoreString

基类:ScoreString

带标签的字符串评分评估器的配置。

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute
reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute
prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute
input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs
get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

SingleKeyEvalConfig

基类:EvalConfig

仅需单个键的运行评估器的配置。

方法 描述
get_kwargs

获取 load_evaluator 调用的关键字参数。

reference_key class-attribute instance-attribute

reference_key: str | None = None

在数据集运行中用作参考字符串的键。如果未提供,将尝试自动推断。

prediction_key class-attribute instance-attribute

prediction_key: str | None = None

从跟踪的运行(traced run)的输出字典中,用于表示预测的键。如果未提供,将自动推断。

input_key class-attribute instance-attribute

input_key: str | None = None

从跟踪的运行(traced run)的输入字典中,用于表示输入的键。如果未提供,将自动推断。

get_kwargs

get_kwargs() -> dict[str, Any]

获取 load_evaluator 调用的关键字参数。

返回 描述
dict[str, Any]

load_evaluator 调用的关键字参数。

langchain_classic.smith.evaluation.progress

一个用于控制台的简单进度条。

ProgressBarCallback

基类:BaseCallbackHandler

一个用于控制台的简单进度条。

方法 描述
__init__

初始化进度条。

increment

增加计数器并更新进度条。

on_chain_error

当链出错时运行。

on_chain_end

当链结束运行时运行。

on_retriever_error

当检索器(Retriever)出错时运行。

on_retriever_end

当检索器(Retriever)结束运行时运行。

on_llm_error

当语言模型(LLM)出错时运行。

on_llm_end

当语言模型(LLM)结束运行时运行。

on_tool_error

当工具出错时运行。

on_tool_end

当工具结束运行时运行。

on_text

在任意文本上运行。

on_retry

在重试事件上运行。

on_custom_event

重写以定义自定义事件的处理程序。

on_llm_start

当语言模型(LLM)开始运行时运行。

on_chat_model_start

当聊天模型开始运行时运行。

on_retriever_start

当检索器(Retriever)开始运行时运行。

on_chain_start

当链开始运行时运行。

on_tool_start

当工具开始运行时运行。

on_agent_action

在代理(agent)执行动作时运行。

on_agent_finish

在代理(agent)结束时运行。

on_llm_new_token

在新输出的 token 上运行。仅在启用流式传输时可用。

raise_error class-attribute instance-attribute

raise_error: bool = False

是否在发生异常时抛出错误。

run_inline class-attribute instance-attribute

run_inline: bool = False

是否内联运行回调。

ignore_llm property

ignore_llm: bool

是否忽略语言模型(LLM)回调。

ignore_retry property

ignore_retry: bool

是否忽略重试回调。

ignore_chain property

ignore_chain: bool

是否忽略链回调。

ignore_agent property

ignore_agent: bool

是否忽略代理回调。

ignore_retriever property

ignore_retriever: bool

是否忽略检索器回调。

ignore_chat_model property

ignore_chat_model: bool

是否忽略聊天模型回调。

ignore_custom_event property

ignore_custom_event: bool

忽略自定义事件。

__init__

__init__(total: int, ncols: int = 50, end_with: str = '\n')

初始化进度条。

参数 描述
total

待处理项目的总数。

类型: int

ncols

进度条的字符宽度。

类型: int 默认值: 50

end_with

进度条到达末尾后打印的最后一个字符串。

类型: str 默认值: '\n'

**kwargs

附加的关键字参数。

increment

increment() -> None

增加计数器并更新进度条。

on_chain_error

on_chain_error(
    error: BaseException,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

当链出错时运行。

参数 描述
error

发生的错误。

类型: BaseException

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_chain_end

on_chain_end(
    outputs: dict[str, Any],
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

当链结束运行时运行。

参数 描述
outputs

链的输出。

类型: dict[str, Any]

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_retriever_error

on_retriever_error(
    error: BaseException,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

当检索器(Retriever)出错时运行。

参数 描述
error

发生的错误。

类型: BaseException

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_retriever_end

on_retriever_end(
    documents: Sequence[Document],
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

当检索器(Retriever)结束运行时运行。

参数 描述
documents

检索到的文档。

类型: Sequence[Document]

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_llm_error

on_llm_error(
    error: BaseException,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

当语言模型(LLM)出错时运行。

参数 描述
error

发生的错误。

类型: BaseException

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_llm_end

on_llm_end(
    response: LLMResult,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

当语言模型(LLM)结束运行时运行。

参数 描述
response

生成的响应。

类型: LLMResult

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_tool_error

on_tool_error(
    error: BaseException,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

当工具出错时运行。

参数 描述
error

发生的错误。

类型: BaseException

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_tool_end

on_tool_end(
    output: str, *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs: Any
) -> Any

当工具结束运行时运行。

参数 描述
output

工具的输出。

类型: Any

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_text

on_text(
    text: str, *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs: Any
) -> Any

在任意文本上运行。

参数 描述
text

文本。

类型: str

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_retry

on_retry(
    retry_state: RetryCallState,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

在重试事件上运行。

参数 描述
retry_state

重试状态。

类型: RetryCallState

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_custom_event

on_custom_event(
    name: str,
    data: Any,
    *,
    run_id: UUID,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> Any

重写以定义自定义事件的处理程序。

参数 描述
name

自定义事件的名称。

类型: str

data

自定义事件的数据。格式将与用户指定的格式匹配。

类型: Any

run_id

运行的 ID。

类型: UUID

tags

与自定义事件关联的标签(包括继承的标签)。

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

metadata

与自定义事件关联的元数据(包括继承的元数据)。

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

on_llm_start

on_llm_start(
    serialized: dict[str, Any],
    prompts: list[str],
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> Any

当语言模型(LLM)开始运行时运行。

警告

此方法用于非聊天模型(常规LLM)。如果您正在为聊天模型实现处理程序,则应改用 on_chat_model_start

参数 描述
serialized

序列化的LLM。

类型: dict[str, Any]

prompts

提示。

类型: list[str]

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

tags

标签。

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

metadata

元数据。

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

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_chat_model_start

on_chat_model_start(
    serialized: dict[str, Any],
    messages: list[list[BaseMessage]],
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> Any

当聊天模型开始运行时运行。

警告

此方法用于聊天模型。如果您正在为非聊天模型实现处理程序,则应改用 on_llm_start

参数 描述
serialized

序列化的聊天模型。

类型: dict[str, Any]

messages

消息。

类型: list[list[BaseMessage]]

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

tags

标签。

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

metadata

元数据。

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

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_retriever_start

on_retriever_start(
    serialized: dict[str, Any],
    query: str,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> Any

当检索器(Retriever)开始运行时运行。

参数 描述
serialized

序列化的检索器。

类型: dict[str, Any]

query

查询。

类型: str

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

tags

标签。

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

metadata

元数据。

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

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_chain_start

on_chain_start(
    serialized: dict[str, Any],
    inputs: dict[str, Any],
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> Any

当链开始运行时运行。

参数 描述
serialized

序列化的链。

类型: dict[str, Any]

inputs

输入。

类型: dict[str, Any]

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

tags

标签。

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

metadata

元数据。

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

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_tool_start

on_tool_start(
    serialized: dict[str, Any],
    input_str: str,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    inputs: dict[str, Any] | None = None,
    **kwargs: Any,
) -> Any

当工具开始运行时运行。

参数 描述
serialized

序列化的链。

类型: dict[str, Any]

input_str

输入字符串。

类型: str

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

tags

标签。

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

metadata

元数据。

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

inputs

输入。

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

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_agent_action

on_agent_action(
    action: AgentAction,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

在代理(agent)执行动作时运行。

参数 描述
action

代理操作。

类型: AgentAction

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_agent_finish

on_agent_finish(
    finish: AgentFinish,
    *,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

在代理(agent)结束时运行。

参数 描述
finish

代理完成。

类型: AgentFinish

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

on_llm_new_token

on_llm_new_token(
    token: str,
    *,
    chunk: GenerationChunk | ChatGenerationChunk | None = None,
    run_id: UUID,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> Any

在新输出的 token 上运行。仅在启用流式传输时可用。

适用于聊天模型和非聊天模型(传统LLM)。

参数 描述
token

新令牌。

类型: str

chunk

新生成的块,包含内容和其他信息。

类型: GenerationChunk | ChatGenerationChunk | None 默认值: None

run_id

运行ID。这是当前运行的ID。

类型: UUID

parent_run_id

父运行ID。这是父运行的ID。

类型: UUID | None 默认值: None

**kwargs

附加的关键字参数。

类型: Any 默认值: {}

langchain_classic.smith.evaluation.runner_utils

用于在数据集上运行语言模型或链的实用工具。

函数 描述
arun_on_dataset

在数据集上运行。

run_on_dataset

在数据集上运行。

ChatModelInput

基类:TypedDict

聊天模型的输入。

参数 描述
messages

聊天消息列表。

EvalError

基类: dict

您的架构引发了一个错误。

方法 描述
__init__

使用错误和其他属性初始化 EvalError

__getattr__

EvalError 中获取一个属性。

__init__

__init__(Error: BaseException, **kwargs: Any) -> None

使用错误和其他属性初始化 EvalError

参数 描述
错误

发生的错误。

类型: BaseException

**kwargs

要包含在错误中的其他属性。

类型: Any 默认值: {}

__getattr__

__getattr__(name: str) -> Any

EvalError 中获取一个属性。

参数 描述
name

要获取的属性的名称。

类型: str

返回 描述
任意

属性的值。

引发 描述
AttributeError

如果属性不存在。

InputFormatError

基类: Exception

当输入格式无效时引发。

TestResult

基类: dict

单次测试运行结果的字典。

方法 描述
get_aggregate_feedback

返回反馈分数的分位数。

to_dataframe

将结果转换为数据帧。

get_aggregate_feedback

get_aggregate_feedback() -> DataFrame

返回反馈分数的分位数。

此方法计算并打印所有反馈键的反馈分数的分位数。

返回 描述
DataFrame

一个包含每个反馈键的分位数的 DataFrame。

to_dataframe

to_dataframe() -> DataFrame

将结果转换为数据帧。

arun_on_dataset async

arun_on_dataset(
    client: Client | None,
    dataset_name: str,
    llm_or_chain_factory: MODEL_OR_CHAIN_FACTORY,
    *,
    evaluation: RunEvalConfig | None = None,
    dataset_version: datetime | str | None = None,
    concurrency_level: int = 5,
    project_name: str | None = None,
    project_metadata: dict[str, Any] | None = None,
    verbose: bool = False,
    revision_id: str | None = None,
    **kwargs: Any,
) -> dict[str, Any]

在数据集上运行。

在数据集上运行链(Chain)或语言模型,并将跟踪信息存储到指定的项目名称中。

有关此函数(通常更快)的异步版本,请参阅 arun_on_dataset

参数 描述
dataset_name

要在其上运行链的数据集的名称。

类型: str

llm_or_chain_factory

用于在数据集上运行的语言模型或链构造函数。链构造函数用于允许对每个示例进行独立调用,而不会继承状态。

类型: MODEL_OR_CHAIN_FACTORY

evaluation

用于对链的结果运行评估器的配置。

类型: RunEvalConfig | None 默认值: None

dataset_version

数据集的可选版本。

类型: datetime | str | None 默认值: None

concurrency_level

并发运行的异步任务数。

类型: int 默认值: 5

project_name

用于存储跟踪信息的项目名称。默认为 {dataset_name}-{chain class name}-{datetime}

类型: str | None 默认值: None

project_metadata

要添加到项目的可选元数据。可用于存储测试变体的信息。(例如提示版本、模型版本等)

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

client

用于访问数据集以及记录反馈和运行跟踪的 LangSmith 客户端。

类型: Client | None

verbose

是否打印进度。

类型: bool 默认值: False

revision_id

可选的修订标识符,用于将此测试运行分配给跟踪系统不同版本性能的标识符。

类型: str | None 默认值: None

**kwargs

不应使用,但为向后兼容而提供。

类型: Any 默认值: {}

返回 描述
dict[str, Any]

包含运行的项目名称和生成的模型输出的 dict

示例

from langsmith import Client
from langchain_openai import ChatOpenAI
from langchain_classic.chains import LLMChain
from langchain_classic.smith import smith_eval.RunEvalConfig, run_on_dataset

# Chains may have memory. Passing in a constructor function lets the
# evaluation framework avoid cross-contamination between runs.
def construct_chain():
    model = ChatOpenAI(temperature=0)
    chain = LLMChain.from_string(
        model,
        "What's the answer to {your_input_key}"
    )
    return chain

# Load off-the-shelf evaluators via config or the EvaluatorType (string or enum)
evaluation_config = smith_eval.RunEvalConfig(
    evaluators=[
        "qa",  # "Correctness" against a reference answer
        "embedding_distance",
        smith_eval.RunEvalConfig.Criteria("helpfulness"),
        smith_eval.RunEvalConfig.Criteria({
            "fifth-grader-score": "Do you have to be smarter than a fifth "
            "grader to answer this question?"
        }),
    ]
)

client = Client()
await arun_on_dataset(
    client,
    dataset_name="<my_dataset_name>",
    llm_or_chain_factory=construct_chain,
    evaluation=evaluation_config,
)
您还可以通过继承 StringEvaluator 或 LangSmith 的RunEvaluator` 类来创建自定义评估器。

from typing import Optional
from langchain_classic.evaluation import StringEvaluator


class MyStringEvaluator(StringEvaluator):
    @property
    def requires_input(self) -> bool:
        return False

    @property
    def requires_reference(self) -> bool:
        return True

    @property
    def evaluation_name(self) -> str:
        return "exact_match"

    def _evaluate_strings(
        self, prediction, reference=None, input=None, **kwargs
    ) -> dict:
        return {"score": prediction == reference}


evaluation_config = smith_eval.RunEvalConfig(
    custom_evaluators=[MyStringEvaluator()],
)

await arun_on_dataset(
    client,
    dataset_name="<my_dataset_name>",
    llm_or_chain_factory=construct_chain,
    evaluation=evaluation_config,
)

run_on_dataset

run_on_dataset(
    client: Client | None,
    dataset_name: str,
    llm_or_chain_factory: MODEL_OR_CHAIN_FACTORY,
    *,
    evaluation: RunEvalConfig | None = None,
    dataset_version: datetime | str | None = None,
    concurrency_level: int = 5,
    project_name: str | None = None,
    project_metadata: dict[str, Any] | None = None,
    verbose: bool = False,
    revision_id: str | None = None,
    **kwargs: Any,
) -> dict[str, Any]

在数据集上运行。

在数据集上运行链(Chain)或语言模型,并将跟踪信息存储到指定的项目名称中。

有关此函数(通常更快)的异步版本,请参阅 arun_on_dataset

参数 描述
dataset_name

要在其上运行链的数据集的名称。

类型: str

llm_or_chain_factory

用于在数据集上运行的语言模型或链构造函数。链构造函数用于允许对每个示例进行独立调用,而不会继承状态。

类型: MODEL_OR_CHAIN_FACTORY

evaluation

用于对链的结果运行评估器的配置。

类型: RunEvalConfig | None 默认值: None

dataset_version

数据集的可选版本。

类型: datetime | str | None 默认值: None

concurrency_level

并发运行的异步任务数。

类型: int 默认值: 5

project_name

用于存储跟踪信息的项目名称。默认为 {dataset_name}-{chain class name}-{datetime}

类型: str | None 默认值: None

project_metadata

要添加到项目的可选元数据。可用于存储测试变体的信息。(例如提示版本、模型版本等)

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

client

用于访问数据集以及记录反馈和运行跟踪的 LangSmith 客户端。

类型: Client | None

verbose

是否打印进度。

类型: bool 默认值: False

revision_id

可选的修订标识符,用于将此测试运行分配给跟踪系统不同版本性能的标识符。

类型: str | None 默认值: None

**kwargs

不应使用,但为向后兼容而提供。

类型: Any 默认值: {}

返回 描述
dict[str, Any]

包含运行的项目名称和生成的模型输出的 dict

示例

from langsmith import Client
from langchain_openai import ChatOpenAI
from langchain_classic.chains import LLMChain
from langchain_classic.smith import smith_eval.RunEvalConfig, run_on_dataset

# Chains may have memory. Passing in a constructor function lets the
# evaluation framework avoid cross-contamination between runs.
def construct_chain():
    model = ChatOpenAI(temperature=0)
    chain = LLMChain.from_string(
        model,
        "What's the answer to {your_input_key}"
    )
    return chain

# Load off-the-shelf evaluators via config or the EvaluatorType (string or enum)
evaluation_config = smith_eval.RunEvalConfig(
    evaluators=[
        "qa",  # "Correctness" against a reference answer
        "embedding_distance",
        smith_eval.RunEvalConfig.Criteria("helpfulness"),
        smith_eval.RunEvalConfig.Criteria({
            "fifth-grader-score": "Do you have to be smarter than a fifth "
            "grader to answer this question?"
        }),
    ]
)

client = Client()
run_on_dataset(
    client,
    dataset_name="<my_dataset_name>",
    llm_or_chain_factory=construct_chain,
    evaluation=evaluation_config,
)

您还可以通过继承 StringEvaluator 或 LangSmith 的 RunEvaluator 类来创建自定义评估器。

from typing import Optional
from langchain_classic.evaluation import StringEvaluator


class MyStringEvaluator(StringEvaluator):
    @property
    def requires_input(self) -> bool:
        return False

    @property
    def requires_reference(self) -> bool:
        return True

    @property
    def evaluation_name(self) -> str:
        return "exact_match"

    def _evaluate_strings(
        self, prediction, reference=None, input=None, **kwargs
    ) -> dict:
        return {"score": prediction == reference}


evaluation_config = smith_eval.RunEvalConfig(
    custom_evaluators=[MyStringEvaluator()],
)

run_on_dataset(
    client,
    dataset_name="<my_dataset_name>",
    llm_or_chain_factory=construct_chain,
    evaluation=evaluation_config,
)

langchain_classic.smith.evaluation.string_run_evaluator

字符串评估器的运行评估器包装器。

ChainStringRunMapper

基类: StringRunMapper

从链的运行对象中提取要评估的项目。

方法 描述
map

将运行映射到一个字典。

__init__
is_lc_serializable

这个类是否可序列化?

get_lc_namespace

获取 LangChain 对象的命名空间。

lc_id

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

to_json

将对象序列化为 JSON。

to_json_not_implemented

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

__call__

将运行映射到一个字典。

input_key class-attribute instance-attribute

input_key: str | None = None

模型运行输入中用作评估输入的键。如果未提供,将使用唯一的输入键,或者在有多个输入键时引发错误。

prediction_key class-attribute instance-attribute

prediction_key: str | None = None

模型运行输出中用作评估预测的键。如果未提供,将使用唯一的输出键,或者在有多个输出键时引发错误。

lc_secrets property

lc_secrets: dict[str, str]

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

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

lc_attributes property

lc_attributes: dict

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

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

默认为空字典。

output_keys property

output_keys: list[str]

要从运行中提取的键。

map

map(run: Run) -> dict[str, str]

将运行映射到一个字典。

__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

将对象序列化为 JSON。

引发 描述
ValueError

如果类有已弃用的属性。

返回 描述
SerializedConstructor | SerializedNotImplemented

一个可 JSON 序列化的对象或一个 SerializedNotImplemented 对象。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

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

返回 描述
SerializedNotImplemented

SerializedNotImplemented.

__call__

__call__(run: Run) -> dict[str, str]

将运行映射到一个字典。

LLMStringRunMapper

基类: StringRunMapper

从运行对象中提取要评估的项目。

方法 描述
serialize_chat_messages

从运行中提取输入消息。

serialize_inputs

序列化输入。

serialize_outputs

序列化输出。

map

将运行映射到一个字典。

__init__
is_lc_serializable

这个类是否可序列化?

get_lc_namespace

获取 LangChain 对象的命名空间。

lc_id

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

to_json

将对象序列化为 JSON。

to_json_not_implemented

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

__call__

将运行映射到一个字典。

lc_secrets property

lc_secrets: dict[str, str]

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

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

lc_attributes property

lc_attributes: dict

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

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

默认为空字典。

output_keys property

output_keys: list[str]

要从运行中提取的键。

serialize_chat_messages

serialize_chat_messages(messages: list[dict] | list[list[dict]]) -> str

从运行中提取输入消息。

serialize_inputs

serialize_inputs(inputs: dict) -> str

序列化输入。

参数 描述
inputs

来自运行的输入,应包含提示或消息。

类型: dict

返回 描述
str

来自提示或消息的序列化输入文本。

引发 描述
ValueError

如果在输入中既没有找到提示也没有找到消息。

serialize_outputs

serialize_outputs(outputs: dict) -> str

序列化输出。

参数 描述
outputs

来自运行的输出,应包含生成内容。

类型: dict

返回 描述
str

来自第一次生成的序列化输出文本。

引发 描述
ValueError

如果在输出中没有找到生成内容,

map

map(run: Run) -> dict[str, str]

将运行映射到一个字典。

__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

将对象序列化为 JSON。

引发 描述
ValueError

如果类有已弃用的属性。

返回 描述
SerializedConstructor | SerializedNotImplemented

一个可 JSON 序列化的对象或一个 SerializedNotImplemented 对象。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

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

返回 描述
SerializedNotImplemented

SerializedNotImplemented.

__call__

__call__(run: Run) -> dict[str, str]

将运行映射到一个字典。

StringExampleMapper

基类: Serializable

将一个示例或数据集中的一行映射到评估的输入。

方法 描述
serialize_chat_messages

从运行中提取输入消息。

map

将示例或数据集行映射到字典。

__call__

将运行和示例映射到字典。

__init__
is_lc_serializable

这个类是否可序列化?

get_lc_namespace

获取 LangChain 对象的命名空间。

lc_id

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

to_json

将对象序列化为 JSON。

to_json_not_implemented

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

output_keys property

output_keys: list[str]

要从运行中提取的键。

lc_secrets property

lc_secrets: dict[str, str]

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

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

lc_attributes property

lc_attributes: dict

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

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

默认为空字典。

serialize_chat_messages

serialize_chat_messages(messages: list[dict]) -> str

从运行中提取输入消息。

map

map(example: Example) -> dict[str, str]

将示例或数据集行映射到字典。

__call__

__call__(example: Example) -> dict[str, str]

将运行和示例映射到字典。

__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

将对象序列化为 JSON。

引发 描述
ValueError

如果类有已弃用的属性。

返回 描述
SerializedConstructor | SerializedNotImplemented

一个可 JSON 序列化的对象或一个 SerializedNotImplemented 对象。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

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

返回 描述
SerializedNotImplemented

SerializedNotImplemented.

StringRunEvaluatorChain

基类: Chain, RunEvaluator

评估运行和可选示例。

方法 描述
evaluate_run

评估一个示例。

aevaluate_run

评估一个示例。

from_run_and_data_type

创建一个 StringRunEvaluatorChain。

get_name

获取 Runnable 的名称。

get_input_schema

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

get_input_jsonschema

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

get_output_schema

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

get_output_jsonschema

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

config_schema

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

get_config_jsonschema

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

get_graph

返回此 Runnable 的图形表示。

get_prompts

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

__or__

Runnable "or" 运算符。

__ror__

Runnable "reverse-or" 运算符。

pipe

管道连接 Runnable 对象。

pick

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

assign

向此 Runnabledict 输出分配新字段。

invoke

将单个输入转换为输出。

ainvoke

将单个输入转换为输出。

batch

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

batch_as_completed

在输入列表上并行运行 invoke

abatch

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

abatch_as_completed

在输入列表上并行运行 ainvoke

stream

stream 的默认实现,它调用 invoke

astream

astream 的默认实现,它调用 ainvoke

astream_log

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

astream_events

生成事件流。

transform

将输入转换为输出。

atransform

将输入转换为输出。

bind

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

with_config

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

with_listeners

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

with_alisteners

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

with_types

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

with_retry

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

map

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

with_fallbacks

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

as_tool

Runnable 创建一个 BaseTool

__init__
is_lc_serializable

这个类是否可序列化?

get_lc_namespace

获取 LangChain 对象的命名空间。

lc_id

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

to_json

Runnable 序列化为 JSON。

to_json_not_implemented

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

configurable_fields

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

configurable_alternatives

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

raise_callback_manager_deprecation

如果使用 callback_manager 则发出弃用警告。

set_verbose

设置链的详细程度。

__call__

执行链。

acall

异步执行链。

prep_outputs

验证并准备链的输出,并将此次运行的信息保存到内存中。

aprep_outputs

验证并准备链的输出,并将此次运行的信息保存到内存中。

prep_inputs

准备链的输入,包括从内存中添加输入。

aprep_inputs

准备链的输入,包括从内存中添加输入。

run

执行链的便捷方法。

arun

执行链的便捷方法。

dict

链的字典表示。

save

保存链。

apply

在列表中的所有输入上调用链。

run_mapper instance-attribute

run_mapper: StringRunMapper

将运行映射到一个包含'input'和'prediction'字符串的字典。

example_mapper class-attribute instance-attribute

example_mapper: StringExampleMapper | None = None

将示例(数据集行)映射到一个包含'reference'字符串的字典。

name instance-attribute

name: str

评估指标的名称。

string_evaluator instance-attribute

string_evaluator: StringEvaluator

评估链。

input_keys property

input_keys: list[str]

预期在链输入中的键。

output_keys property

output_keys: list[str]

预期在链输出中的键。

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

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

默认为空字典。

memory class-attribute instance-attribute

memory: BaseMemory | None = None

可选的内存对象。内存是一个类,在每个链的开始和结束时被调用。在开始时,内存加载变量并在链中传递它们。在结束时,它保存任何返回的变量。有许多不同类型的内存 - 请参阅内存文档以获取完整目录。

callbacks class-attribute instance-attribute

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

可选的回调处理程序列表(或回调管理器)。回调处理程序在链的调用生命周期中被调用,从 on_chain_start 开始,到 on_chain_end 或 on_chain_error 结束。每个自定义链可以选择性地调用额外的回调方法,详情请参见回调文档。

verbose class-attribute instance-attribute

verbose: bool = Field(default_factory=_get_verbosity)

是否以详细模式运行。在详细模式下,一些中间日志将被打印到控制台。默认为全局 verbose 值,可通过 langchain.globals.get_verbose() 访问。

tags class-attribute instance-attribute

tags: list[str] | None = None

与链关联的可选标签列表。这些标签将与对此链的每次调用相关联,并作为参数传递给在 callbacks 中定义的处理程序。您可以使用这些标签来识别特定链实例及其用例。

metadata class-attribute instance-attribute

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

与链关联的可选元数据。此元数据将与对此链的每次调用相关联,并作为参数传递给在 callbacks 中定义的处理程序。您可以使用这些元数据来识别特定链实例及其用例。

callback_manager class-attribute instance-attribute

callback_manager: BaseCallbackManager | None = Field(default=None, exclude=True)

[已弃用] 请改用 callbacks

evaluate_run

evaluate_run(
    run: Run, example: Example | None = None, evaluator_run_id: UUID | None = None
) -> EvaluationResult

评估一个示例。

aevaluate_run async

aevaluate_run(
    run: Run, example: Example | None = None, evaluator_run_id: UUID | None = None
) -> EvaluationResult

评估一个示例。

from_run_and_data_type classmethod

from_run_and_data_type(
    evaluator: StringEvaluator,
    run_type: str,
    data_type: DataType,
    input_key: str | None = None,
    prediction_key: str | None = None,
    reference_key: str | None = None,
    tags: list[str] | None = None,
) -> StringRunEvaluatorChain

创建一个 StringRunEvaluatorChain。

根据评估器以及运行和数据类型创建一个 StringRunEvaluatorChain。

此方法提供了一种简单的方式来实例化 StringRunEvaluatorChain,通过接受一个评估器以及有关运行类型和数据的信息。该方法支持 LLM 和链运行。

参数 描述
evaluator

要使用的字符串评估器。

类型: StringEvaluator

run_type

正在评估的运行类型。支持的类型有 LLM 和 Chain。

类型: str

data_type

运行中使用的数据集类型。

类型: DataType

input_key

用于从运行中映射输入的键。

类型: str | None 默认值: None

prediction_key

用于从运行中映射预测的键。

类型: str | None 默认值: None

reference_key

用于从数据集中映射引用的键。

类型: str | None 默认值: None

tags

附加到评估链的标签列表。

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

返回 描述
StringRunEvaluatorChain

实例化的评估链。

get_name

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

获取 Runnable 的名称。

参数 描述
后缀

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

类型: str | None 默认值: None

name

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

类型: str | None 默认值: None

返回 描述
str

Runnable 的名称。

get_input_schema

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

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

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

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

参数 描述
配置

生成模式时使用的配置。

类型: RunnableConfig | None 默认值: None

返回 描述
type[BaseModel]

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

get_input_jsonschema

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

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

参数 描述
配置

生成模式时使用的配置。

类型: RunnableConfig | None 默认值: None

返回 描述
dict[str, Any]

表示 Runnable 输入的 JSON 模式。

示例
from langchain_core.runnables import RunnableLambda


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


runnable = RunnableLambda(add_one)

print(runnable.get_input_jsonschema())

在 0.3.0 版本中新增。

get_output_schema

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

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

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

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

参数 描述
配置

生成模式时使用的配置。

类型: RunnableConfig | None 默认值: None

返回 描述
type[BaseModel]

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

get_output_jsonschema

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

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

参数 描述
配置

生成模式时使用的配置。

类型: RunnableConfig | None 默认值: None

返回 描述
dict[str, Any]

表示 Runnable 输出的 JSON 模式。

示例
from langchain_core.runnables import RunnableLambda


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


runnable = RunnableLambda(add_one)

print(runnable.get_output_jsonschema())

在 0.3.0 版本中新增。

config_schema

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

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

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

参数 描述
包含

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

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

返回 描述
type[BaseModel]

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

get_config_jsonschema

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

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

参数 描述
包含

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

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

返回 描述
dict[str, Any]

表示 Runnable 配置的 JSON 模式。

在 0.3.0 版本中新增。

get_graph

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

返回此 Runnable 的图形表示。

get_prompts

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

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

__or__

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

Runnable "or" 运算符。

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

参数 描述
other

另一个 Runnable 或类 Runnable 对象。

类型: Runnable[Any, Other] | Callable[[Iterator[Any]], Iterator[Other]] | Callable[[AsyncIterator[Any]], AsyncIterator[Other]] | Callable[[Any], Other] | Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any]

返回 描述
RunnableSerializable[Input, Other]

一个新的 Runnable

__ror__

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

Runnable "reverse-or" 运算符。

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

参数 描述
other

另一个 Runnable 或类 Runnable 对象。

类型: Runnable[Other, Any] | Callable[[Iterator[Other]], Iterator[Any]] | Callable[[AsyncIterator[Other]], AsyncIterator[Any]] | Callable[[Other], Any] | Mapping[str, Runnable[Other, Any] | Callable[[Other], Any] | Any]

返回 描述
RunnableSerializable[Other, Output]

一个新的 Runnable

pipe

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

管道连接 Runnable 对象。

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

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

示例
from langchain_core.runnables import RunnableLambda


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


def mul_two(x: int) -> int:
    return x * 2


runnable_1 = RunnableLambda(add_one)
runnable_2 = RunnableLambda(mul_two)
sequence = runnable_1.pipe(runnable_2)
# Or equivalently:
# sequence = runnable_1 | runnable_2
# sequence = RunnableSequence(first=runnable_1, last=runnable_2)
sequence.invoke(1)
await sequence.ainvoke(1)
# -> 4

sequence.batch([1, 2, 3])
await sequence.abatch([1, 2, 3])
# -> [4, 6, 8]
参数 描述
*其他

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

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

name

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

类型: str | None 默认值: None

返回 描述
RunnableSerializable[Input, Other]

一个新的 Runnable

pick

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

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

选择单个键

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)
chain = RunnableMap(str=as_str, json=as_json)

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3]}

json_only_chain = chain.pick("json")
json_only_chain.invoke("[1, 2, 3]")
# -> [1, 2, 3]

选择键列表

from typing import Any

import json

from langchain_core.runnables import RunnableLambda, RunnableMap

as_str = RunnableLambda(str)
as_json = RunnableLambda(json.loads)


def as_bytes(x: Any) -> bytes:
    return bytes(x, "utf-8")


chain = RunnableMap(str=as_str, json=as_json, bytes=RunnableLambda(as_bytes))

chain.invoke("[1, 2, 3]")
# -> {"str": "[1, 2, 3]", "json": [1, 2, 3], "bytes": b"[1, 2, 3]"}

json_and_bytes_chain = chain.pick(["json", "bytes"])
json_and_bytes_chain.invoke("[1, 2, 3]")
# -> {"json": [1, 2, 3], "bytes": b"[1, 2, 3]"}
参数 描述
keys

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

类型: str | list[str]

返回 描述
RunnableSerializable[Any, Any]

一个新的 Runnable

assign

向此 Runnabledict 输出分配新字段。

from langchain_community.llms.fake import FakeStreamingListLLM
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import SystemMessagePromptTemplate
from langchain_core.runnables import Runnable
from operator import itemgetter

prompt = (
    SystemMessagePromptTemplate.from_template("You are a nice assistant.")
    + "{question}"
)
model = FakeStreamingListLLM(responses=["foo-lish"])

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

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

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

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

类型: Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any] | Mapping[str, Runnable[dict[str, Any], Any] | Callable[[dict[str, Any]], Any]] 默认值: {}

返回 描述
RunnableSerializable[Any, Any]

一个新的 Runnable

invoke

invoke(
    input: dict[str, Any], config: RunnableConfig | None = None, **kwargs: Any
) -> dict[str, Any]

将单个输入转换为输出。

参数 描述
输入

Runnable 的输入。

类型: Input

配置

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

类型: RunnableConfig | None 默认值: None

返回 描述
输出

Runnable 的输出。

ainvoke async

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

将单个输入转换为输出。

参数 描述
输入

Runnable 的输入。

类型: Input

配置

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

类型: RunnableConfig | None 默认值: None

返回 描述
输出

Runnable 的输出。

batch

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

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

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

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

参数 描述
inputs

Runnable 的输入列表。

类型: list[Input]

配置

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

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

返回异常

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

类型: bool 默认值: False

**kwargs

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

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

返回 描述
list[Output]

来自 Runnable 的输出列表。

batch_as_completed

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

在输入列表上并行运行 invoke

在结果完成时生成它们。

参数 描述
inputs

Runnable 的输入列表。

类型: Sequence[Input]

配置

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

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

返回异常

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

类型: bool 默认值: False

**kwargs

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

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

YIELDS 描述
tuple[int, Output | Exception]

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

abatch async

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

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

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

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

参数 描述
inputs

Runnable 的输入列表。

类型: list[Input]

配置

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

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

返回异常

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

类型: bool 默认值: False

**kwargs

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

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

返回 描述
list[Output]

来自 Runnable 的输出列表。

abatch_as_completed async

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

在输入列表上并行运行 ainvoke

在结果完成时生成它们。

参数 描述
inputs

Runnable 的输入列表。

类型: Sequence[Input]

配置

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

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

返回异常

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

类型: bool 默认值: False

**kwargs

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

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

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

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

stream

stream(
    input: 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]

一个 RunLogPatchRunLog 对象。

astream_events async

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

生成事件流。

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

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

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

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

注意

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

事件 name chunk 输入 output
on_chat_model_start '[model name]' {"messages": [[SystemMessage, HumanMessage]]}
on_chat_model_stream '[model name]' AIMessageChunk(content="hello")
on_chat_model_end '[model name]' {"messages": [[SystemMessage, HumanMessage]]} AIMessageChunk(content="hello world")
on_llm_start '[model name]' {'input': 'hello'}
on_llm_stream '[model name]' '你好'
on_llm_end '[model name]' '你好,人类!'
on_chain_start 'format_docs'
on_chain_stream 'format_docs' 'hello world!, goodbye world!'
on_chain_end 'format_docs' [Document(...)] 'hello world!, goodbye world!'
on_tool_start 'some_tool' {"x": 1, "y": "2"}
on_tool_end 'some_tool' {"x": 1, "y": "2"}
on_retriever_start '[retriever name]' {"query": "hello"}
on_retriever_end '[retriever name]' {"query": "hello"} [Document(...), ..]
on_prompt_start '[template_name]' {"question": "hello"}
on_prompt_end '[template_name]' {"question": "hello"} ChatPromptValue(messages: [SystemMessage, ...])

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

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

自定义事件具有以下格式

属性 类型 描述
name str 用户为事件定义的名称。
data 任意 与事件关联的数据。这可以是任何东西,但我们建议使其可 JSON 序列化。

以下是与上面显示的标准事件相关的声明

format_docs:

def format_docs(docs: list[Document]) -> str:
    '''Format the docs.'''
    return ", ".join([doc.page_content for doc in docs])


format_docs = RunnableLambda(format_docs)

some_tool:

@tool
def some_tool(x: int, y: str) -> dict:
    '''Some_tool.'''
    return {"x": x, "y": y}

prompt:

template = ChatPromptTemplate.from_messages(
    [
        ("system", "You are Cat Agent 007"),
        ("human", "{question}"),
    ]
).with_config({"run_name": "my_template", "tags": ["my_template"]})

例如

from langchain_core.runnables import RunnableLambda


async def reverse(s: str) -> str:
    return s[::-1]


chain = RunnableLambda(func=reverse)

events = [event async for event in chain.astream_events("hello", version="v2")]

# Will produce the following events
# (run_id, and parent_ids has been omitted for brevity):
[
    {
        "data": {"input": "hello"},
        "event": "on_chain_start",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"chunk": "olleh"},
        "event": "on_chain_stream",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
    {
        "data": {"output": "olleh"},
        "event": "on_chain_end",
        "metadata": {},
        "name": "reverse",
        "tags": [],
    },
]
示例:分派自定义事件
from langchain_core.callbacks.manager import (
    adispatch_custom_event,
)
from langchain_core.runnables import RunnableLambda, RunnableConfig
import asyncio


async def slow_thing(some_input: str, config: RunnableConfig) -> str:
    """Do something that takes a long time."""
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 1 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    await adispatch_custom_event(
        "progress_event",
        {"message": "Finished step 2 of 3"},
        config=config # Must be included for python < 3.10
    )
    await asyncio.sleep(1) # Placeholder for some slow operation
    return "Done"

slow_thing = RunnableLambda(slow_thing)

async for event in slow_thing.astream_events("some_input", version="v2"):
    print(event)
参数 描述
输入

Runnable 的输入。

类型: Any

配置

用于 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

版本

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

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

包含名称

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

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

包含类型

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

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

包含标签

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

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

排除名称

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

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

排除类型

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

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

排除标签

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

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

**kwargs

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

类型: Any 默认值: {}

YIELDS 描述
AsyncIterator[StreamEvent]

StreamEvent 的异步流。

引发 描述
NotImplementedError

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

transform

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

将输入转换为输出。

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

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

参数 描述
输入

Runnable 输入的迭代器。

类型: Iterator[Input]

配置

用于 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

**kwargs

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

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

YIELDS 描述
输出

Runnable 的输出。

atransform async

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

将输入转换为输出。

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

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

参数 描述
输入

Runnable 输入的异步迭代器。

类型: AsyncIterator[Input]

配置

用于 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

**kwargs

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

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

YIELDS 描述
AsyncIterator[Output]

Runnable 的输出。

bind

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

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

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

参数 描述
**kwargs

要绑定到 Runnable 的参数。

类型: Any 默认值: {}

返回 描述
Runnable[Input, Output]

一个绑定了参数的新 Runnable

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

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

# Without bind
chain = model | StrOutputParser()

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

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

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

with_config

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

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

参数 描述
配置

要绑定到 Runnable 的配置。

类型: RunnableConfig | None 默认值: None

**kwargs

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

类型: Any 默认值: {}

返回 描述
Runnable[Input, Output]

一个绑定了配置的新 Runnable

with_listeners

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

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

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

参数 描述
开始时

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

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

结束时

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

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

出错时

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

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

返回 描述
Runnable[Input, Output]

一个绑定了监听器的新 Runnable

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

import time


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


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


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


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

with_alisteners

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

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

返回一个新的 Runnable

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

参数 描述
开始时

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

类型: AsyncListener | None 默认值: None

结束时

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

类型: AsyncListener | None 默认值: None

出错时

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

类型: AsyncListener | None 默认值: None

返回 描述
Runnable[Input, Output]

一个绑定了监听器的新 Runnable

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

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

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

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

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

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

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

with_types

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

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

参数 描述
输入类型

要绑定到 Runnable 的输入类型。

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

输出类型

要绑定到 Runnable 的输出类型。

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

返回 描述
Runnable[Input, Output]

一个绑定了类型的新 Runnable

with_retry

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

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

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

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

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

指数等待抖动

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

类型: bool 默认值: True

尝试后停止

放弃前尝试的最大次数。

类型: int 默认值: 3

指数抖动参数

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

类型: ExponentialJitterParams | None 默认值: None

返回 描述
Runnable[Input, Output]

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

示例
from langchain_core.runnables import RunnableLambda

count = 0


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


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

assert count == 2

map

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

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

使用每个输入调用 invoke

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

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

示例
from langchain_core.runnables import RunnableLambda


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


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

with_fallbacks

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

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

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

参数 描述
备用方案

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

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

要处理的异常

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

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

异常键

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

类型: str | None 默认值: None

返回 描述
RunnableWithFallbacks[Input, Output]

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

示例
from typing import Iterator

from langchain_core.runnables import RunnableGenerator


def _generate_immediate_error(input: Iterator) -> Iterator[str]:
    raise ValueError()
    yield ""


def _generate(input: Iterator) -> Iterator[str]:
    yield from "foo bar"


runnable = RunnableGenerator(_generate_immediate_error).with_fallbacks(
    [RunnableGenerator(_generate)]
)
print("".join(runnable.stream({})))  # foo bar
参数 描述
备用方案

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

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

要处理的异常

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

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

异常键

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

类型: str | None 默认值: None

返回 描述
RunnableWithFallbacks[Input, Output]

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

as_tool

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

Runnable 创建一个 BaseTool

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

参数 描述
参数模式

工具的模式。

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

name

工具的名称。

类型: str | None 默认值: None

描述

工具的描述。

类型: str | None 默认值: None

参数类型

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

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

返回 描述
BaseTool

一个 BaseTool 实例。

类型化字典输入

from typing_extensions import TypedDict
from langchain_core.runnables import RunnableLambda


class Args(TypedDict):
    a: int
    b: list[int]


def f(x: Args) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool()
as_tool.invoke({"a": 3, "b": [1, 2]})

dict 输入,通过 args_schema 指定模式

from typing import Any
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda

def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))

class FSchema(BaseModel):
    """Apply a function to an integer and list of integers."""

    a: int = Field(..., description="Integer")
    b: list[int] = Field(..., description="List of ints")

runnable = RunnableLambda(f)
as_tool = runnable.as_tool(FSchema)
as_tool.invoke({"a": 3, "b": [1, 2]})

dict 输入,通过 arg_types 指定模式

from typing import Any
from langchain_core.runnables import RunnableLambda


def f(x: dict[str, Any]) -> str:
    return str(x["a"] * max(x["b"]))


runnable = RunnableLambda(f)
as_tool = runnable.as_tool(arg_types={"a": int, "b": list[int]})
as_tool.invoke({"a": 3, "b": [1, 2]})

字符串输入

from langchain_core.runnables import RunnableLambda


def f(x: str) -> str:
    return x + "a"


def g(x: str) -> str:
    return x + "z"


runnable = RunnableLambda(f) | g
as_tool = runnable.as_tool()
as_tool.invoke("b")

__init__

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

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
)

raise_callback_manager_deprecation classmethod

raise_callback_manager_deprecation(values: dict) -> Any

如果使用 callback_manager 则发出弃用警告。

set_verbose classmethod

set_verbose(verbose: bool | None) -> bool

设置链的详细程度。

如果用户未指定,则默认为全局设置。

__call__

__call__(
    inputs: dict[str, Any] | Any,
    return_only_outputs: bool = False,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    include_run_info: bool = False,
) -> dict[str, Any]

执行链。

参数 描述
inputs

输入的字典,如果链只接受一个参数,则为单个输入。应包含 `Chain.input_keys` 中指定的所有输入,除非这些输入将由链的内存设置。

类型: dict[str, Any] | Any

return_only_outputs

响应中是否只返回输出。如果为 True,则只返回此链生成的新键。如果为 False,则返回输入键和此链生成的新键。

类型: bool 默认值: False

回调

用于此链运行的回调。这些回调将在链构建时传入的回调之外被调用,但只有这些运行时回调会传播到对其他对象的调用。

类型: Callbacks 默认值: None

tags

要传递给所有回调的字符串标签列表。这些标签将在构建链时传递的标签之外额外传递,但只有这些运行时标签会传播到对其他对象的调用中。

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

metadata

与链关联的可选元数据。

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

运行名称

本次链运行的可选名称。

类型: str | None 默认值: None

include_run_info

是否在响应中包含运行信息。默认为 False。

类型: bool 默认值: False

返回 描述
dict[str, Any]

一个包含命名输出的字典。应包含在 `Chain.output_keys` 中指定的所有输出。

acall async

acall(
    inputs: dict[str, Any] | Any,
    return_only_outputs: bool = False,
    callbacks: Callbacks = None,
    *,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    run_name: str | None = None,
    include_run_info: bool = False,
) -> dict[str, Any]

异步执行链。

参数 描述
inputs

输入的字典,如果链只接受一个参数,则为单个输入。应包含 `Chain.input_keys` 中指定的所有输入,除非这些输入将由链的内存设置。

类型: dict[str, Any] | Any

return_only_outputs

响应中是否只返回输出。如果为 True,则只返回此链生成的新键。如果为 False,则返回输入键和此链生成的新键。

类型: bool 默认值: False

回调

用于此链运行的回调。这些回调将在链构建时传入的回调之外被调用,但只有这些运行时回调会传播到对其他对象的调用。

类型: Callbacks 默认值: None

tags

要传递给所有回调的字符串标签列表。这些标签将在构建链时传递的标签之外额外传递,但只有这些运行时标签会传播到对其他对象的调用中。

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

metadata

与链关联的可选元数据。

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

运行名称

本次链运行的可选名称。

类型: str | None 默认值: None

include_run_info

是否在响应中包含运行信息。默认为 False。

类型: bool 默认值: False

返回 描述
dict[str, Any]

一个包含命名输出的字典。应包含在 `Chain.output_keys` 中指定的所有输出。

prep_outputs

prep_outputs(
    inputs: dict[str, str], outputs: dict[str, str], return_only_outputs: bool = False
) -> dict[str, str]

验证并准备链的输出,并将此次运行的信息保存到内存中。

参数 描述
inputs

链输入的字典,包括由链内存添加的任何输入。

类型: dict[str, str]

outputs

初始链输出的字典。

类型: dict[str, str]

return_only_outputs

是否只返回链输出。如果为 `False`,输入也会被添加到最终输出中。

类型: bool 默认值: False

返回 描述
dict[str, str]

一个包含最终链输出的字典。

aprep_outputs async

aprep_outputs(
    inputs: dict[str, str], outputs: dict[str, str], return_only_outputs: bool = False
) -> dict[str, str]

验证并准备链的输出,并将此次运行的信息保存到内存中。

参数 描述
inputs

链输入的字典,包括由链内存添加的任何输入。

类型: dict[str, str]

outputs

初始链输出的字典。

类型: dict[str, str]

return_only_outputs

是否只返回链输出。如果为 `False`,输入也会被添加到最终输出中。

类型: bool 默认值: False

返回 描述
dict[str, str]

一个包含最终链输出的字典。

prep_inputs

prep_inputs(inputs: dict[str, Any] | Any) -> dict[str, str]

准备链的输入,包括从内存中添加输入。

参数 描述
inputs

原始输入的字典,如果链只需要一个参数,则为单个输入。应包含 `Chain.input_keys` 中指定的所有输入,但不包括将由链的内存设置的输入。

类型: dict[str, Any] | Any

返回 描述
dict[str, str]

一个包含所有输入的字典,包括由链的内存添加的输入。

aprep_inputs async

aprep_inputs(inputs: dict[str, Any] | Any) -> dict[str, str]

准备链的输入,包括从内存中添加输入。

参数 描述
inputs

原始输入的字典,如果链只需要一个参数,则为单个输入。应包含 `Chain.input_keys` 中指定的所有输入,但不包括将由链的内存设置的输入。

类型: dict[str, Any] | Any

返回 描述
dict[str, str]

一个包含所有输入的字典,包括由链的内存添加的输入。

run

run(
    *args: Any,
    callbacks: Callbacks = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> Any

执行链的便捷方法。

此方法与 `Chain.__call__` 的主要区别在于,此方法期望输入直接作为位置参数或关键字参数传递,而 `Chain.__call__` 期望一个包含所有输入的单一输入字典。

参数 描述
*args

如果链期望单个输入,可以将其作为唯一的位置参数传入。

类型: Any 默认值: ()

回调

用于此链运行的回调。这些回调将在链构建时传入的回调之外被调用,但只有这些运行时回调会传播到对其他对象的调用。

类型: Callbacks 默认值: None

tags

要传递给所有回调的字符串标签列表。这些标签将在构建链时传递的标签之外额外传递,但只有这些运行时标签会传播到对其他对象的调用中。

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

metadata

与链关联的可选元数据。

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

**kwargs

如果链期望多个输入,可以直接作为关键字参数传入。

类型: Any 默认值: {}

返回 描述
任意

链输出。

示例
# Suppose we have a single-input chain that takes a 'question' string:
chain.run("What's the temperature in Boise, Idaho?")
# -> "The temperature in Boise is..."

# Suppose we have a multi-input chain that takes a 'question' string
# and 'context' string:
question = "What's the temperature in Boise, Idaho?"
context = "Weather report for Boise, Idaho on 07/03/23..."
chain.run(question=question, context=context)
# -> "The temperature in Boise is..."

arun async

arun(
    *args: Any,
    callbacks: Callbacks = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> Any

执行链的便捷方法。

此方法与 `Chain.__call__` 的主要区别在于,此方法期望输入直接作为位置参数或关键字参数传递,而 `Chain.__call__` 期望一个包含所有输入的单一输入字典。

参数 描述
*args

如果链期望单个输入,可以将其作为唯一的位置参数传入。

类型: Any 默认值: ()

回调

用于此链运行的回调。这些回调将在链构建时传入的回调之外被调用,但只有这些运行时回调会传播到对其他对象的调用。

类型: Callbacks 默认值: None

tags

要传递给所有回调的字符串标签列表。这些标签将在构建链时传递的标签之外额外传递,但只有这些运行时标签会传播到对其他对象的调用中。

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

metadata

与链关联的可选元数据。

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

**kwargs

如果链期望多个输入,可以直接作为关键字参数传入。

类型: Any 默认值: {}

返回 描述
任意

链输出。

示例
# Suppose we have a single-input chain that takes a 'question' string:
await chain.arun("What's the temperature in Boise, Idaho?")
# -> "The temperature in Boise is..."

# Suppose we have a multi-input chain that takes a 'question' string
# and 'context' string:
question = "What's the temperature in Boise, Idaho?"
context = "Weather report for Boise, Idaho on 07/03/23..."
await chain.arun(question=question, context=context)
# -> "The temperature in Boise is..."

dict

dict(**kwargs: Any) -> dict

链的字典表示。

期望 `Chain._chain_type` 属性已实现,且内存为空。

参数 描述
**kwargs

传递给默认 `pydantic.BaseModel.dict` 方法的关键字参数。

类型: Any 默认值: {}

返回 描述
dict

链的字典表示。

示例
chain.model_dump(exclude_unset=True)
# -> {"_type": "foo", "verbose": False, ...}

save

save(file_path: Path | str) -> None

保存链。

期望 `Chain._chain_type` 属性已实现,且内存为空。

参数 描述
file_path

保存链的文件路径。

类型: Path | str

示例
chain.save(file_path="path/chain.yaml")

apply

apply(
    input_list: list[dict[str, Any]], callbacks: Callbacks = None
) -> list[dict[str, str]]

在列表中的所有输入上调用链。

StringRunMapper

基类: Serializable

从运行对象中提取要评估的项目。

方法 描述
map

将运行映射到一个字典。

__call__

将运行映射到一个字典。

__init__
is_lc_serializable

这个类是否可序列化?

get_lc_namespace

获取 LangChain 对象的命名空间。

lc_id

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

to_json

将对象序列化为 JSON。

to_json_not_implemented

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

output_keys property

output_keys: list[str]

要从运行中提取的键。

lc_secrets property

lc_secrets: dict[str, str]

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

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

lc_attributes property

lc_attributes: dict

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

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

默认为空字典。

map abstractmethod

map(run: Run) -> dict[str, str]

将运行映射到一个字典。

__call__

__call__(run: Run) -> dict[str, str]

将运行映射到一个字典。

__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

将对象序列化为 JSON。

引发 描述
ValueError

如果类有已弃用的属性。

返回 描述
SerializedConstructor | SerializedNotImplemented

一个可 JSON 序列化的对象或一个 SerializedNotImplemented 对象。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

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

返回 描述
SerializedNotImplemented

SerializedNotImplemented.

ToolStringRunMapper

基类: StringRunMapper

将输入映射到工具。

方法 描述
map

将运行映射到一个字典。

__init__
is_lc_serializable

这个类是否可序列化?

get_lc_namespace

获取 LangChain 对象的命名空间。

lc_id

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

to_json

将对象序列化为 JSON。

to_json_not_implemented

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

__call__

将运行映射到一个字典。

lc_secrets property

lc_secrets: dict[str, str]

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

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

lc_attributes property

lc_attributes: dict

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

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

默认为空字典。

output_keys property

output_keys: list[str]

要从运行中提取的键。

map

map(run: Run) -> dict[str, str]

将运行映射到一个字典。

__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

将对象序列化为 JSON。

引发 描述
ValueError

如果类有已弃用的属性。

返回 描述
SerializedConstructor | SerializedNotImplemented

一个可 JSON 序列化的对象或一个 SerializedNotImplemented 对象。

to_json_not_implemented

to_json_not_implemented() -> SerializedNotImplemented

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

返回 描述
SerializedNotImplemented

SerializedNotImplemented.

__call__

__call__(run: Run) -> dict[str, str]

将运行映射到一个字典。

langchain_classic.smith.evaluation.name_generation

函数 描述
random_name

生成一个随机名称。

random_name

random_name() -> str

生成一个随机名称。

© . This site is unofficial and not affiliated with LangChain, Inc.