`langchain.evaluation.schema`.LLMEvalChain¶

注意

LLMEvalChain 实现了标准Runnable 接口。🏃

Runnable 接口具有在可执行对象中可用的附加方法，例如with_types、with_retry、assign、bind、get_graph等。

class langchain.evaluation.schema.LLMEvalChain[source]¶

基类：Chain

使用 LLM 的评估器的基类。

参数 callback_manager: Optional[BaseCallbackManager] = None¶: 已弃用。请使用 callbacks 代替。

参数 callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None¶: 可选的回调处理程序列表（或回调管理器）。默认为 None。回调处理程序在整个链的调用生命周期中被调用，从 on_chain_start 开始，以 on_chain_end 或 on_chain_error 结束。每个自定义链可以选择调用额外的回调方法，请参阅回调文档以获取完整详情。

参数 memory: Optional[BaseMemory] = None¶: 可选的内存对象。默认为 None。内存类在每条链的开始和结束时被调用。在开始时，内存加载变量并通过链传递。在结束时，它保存任何返回的变量。有许多不同的内存类型 - 请参阅内存文档以获取完整目录。

参数 metadata: Optional[Dict[str, Any]] = None¶: 与链关联的可选元数据。默认为 None。此元数据将与链的每次调用了关联，并将作为参数传递给在 callbacks 中定义的处理程序。您可以使用这些来识别特定链的使用案例。

参数 tags: Optional[List[str]] = None¶: 与链关联的可选标签列表。默认为 None。这些标签将关联到链的每次调用，并将其作为参数传递给在 callbacks 中定义的处理程序。您可以使用这些来识别特定链的使用案例。

参数 verbose: bool [可选]¶: 是否以详细模式运行。在详细模式下，将打印一些中间日志到控制台。默认为可全局访问的 verbose 值，通过 langchain.globals.get_verbose() 获取。

__call__(inputs: Union[Dict[str, Any], Any], return_only_outputs: bool = False, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = Noneinclude_run_info: bool = False) → Dict[str, Any]¶

自 langchain==0.1.0 版本以来已废止：请使用 invoke 代替。

执行链。

参数

inputs （Union[Dict[str, Any], Any]） – 输入字典，或者在链只需一个参数时为单个输入。应包含在 Chain.input_keys 中指定的所有输入，除链内存要设置的输入外。
返回输出的标志 (bool) – 是否在响应中只返回输出。如果为True，则只返回由该链创建的新键。如果为False，则返回输入键和由该链创建的新键。默认为False。
回调 (可选[联合[列表[BaseCallbackHandler], BaseCallbackManager]]) – 使用此链运行时使用的回调。这些回调将在构建链期间传递的回调之外被调用，但只有这些运行时回调将传播到对其他对象的调用。
标签 (可选[列表[str]]) – 传递给所有回调的字符串标签列表。这些标签将除构建链期间传递的标签外传递，但只有这些运行时标签将传播到对其他对象的调用。
元数据 (可选[Dict[str, Any]]) – 与链相关联的元数据。默认为None
是否包含运行信息 (bool) – 是否在响应中包含运行信息。默认为False。
运行名称 (可选[str]) –

返回值

一个命名输出的字典。应该包含在: Chain.output_keys中指定的所有输出.

返回类型

Dict[str, Any]

async abatch(inputs: List[Input], config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) → List[Output]¶

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

默认的批量实现对IO密集型runnable工作得很好。

子类如果可以更有效地批量处理，应重写此方法；例如，如果底层的Runnable使用支持批量模式的API。

参数

输入 (列表【Input】) – Runnable 接收的输入列表。
config (可选【联合【RunnableConfig，列表【RunnableConfig】】】) – 调用 Runnable 时使用的配置。配置支持标准密钥，如 'tags'（标签）、'metadata'（元数据）用于跟踪目的、'max_concurrency'（最大并发数）用于控制并行执行的工作量，以及其他密钥。请参阅 RunnableConfig 获取更多信息。默认值为 None。
return_exceptions (布尔) – 是否返回异常而不是抛出异常。默认值为 False。
kwargs (可选【任何】) – 传递给 Runnable 的附加关键字参数。

返回值

Runnable 的输出列表。

返回类型

列表[输出]

异步 abatch_as_completed(inputs: 序列[Input], config: 可选[联合[RunnableConfig，序列[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: 可选[任何]) → 异步迭代器[元组[int，联合[Output，Exception]]]¶

并行运行 ainvoke，并在任务完成时生成结果。

参数

inputs (序列【Input】) – Runnable 的输入列表。
config (可选【联合【RunnableConfig，序列【RunnableConfig】】】) – 调用 Runnable 时使用的配置。配置支持标准密钥，如 'tags'（标签）、'metadata'（元数据）用于跟踪目的、'max_concurrency'（最大并发数）用于控制并行执行的工作量，以及其他密钥。请参阅 RunnableConfig 获取更多信息。默认值为 None。
return_exceptions (布尔) – 是否返回异常而不是抛出异常。默认值为 False。
kwargs (可选【任何】) – 传递给 Runnable 的附加关键字参数。

生成

Runnable的输入输出索引元组。

返回类型

AsyncIterator[Tuple[int, Union[Output, Exception]]]

async acall(inputs: Union[Dict[str, Any], Any], return_only_outputs: bool = False, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]], *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, include_run_info: bool = False) → Dict[str, Any]¶

自版本langchain==0.1.0已弃用: 请使用 ainvoke 代替。

异步执行链。

参数

inputs （Union[Dict[str, Any], Any]） – 输入字典，或者在链只需一个参数时为单个输入。应包含在 Chain.input_keys 中指定的所有输入，除链内存要设置的输入外。
返回输出的标志 (bool) – 是否在响应中只返回输出。如果为True，则只返回由该链创建的新键。如果为False，则返回输入键和由该链创建的新键。默认为False。
回调 (可选[联合[列表[BaseCallbackHandler], BaseCallbackManager]]) – 使用此链运行时使用的回调。这些回调将在构建链期间传递的回调之外被调用，但只有这些运行时回调将传播到对其他对象的调用。
标签 (可选[列表[str]]) – 传递给所有回调的字符串标签列表。这些标签将除构建链期间传递的标签外传递，但只有这些运行时标签将传播到对其他对象的调用。
元数据 (可选[Dict[str, Any]]) – 与链相关联的元数据。默认为None
是否包含运行信息 (bool) – 是否在响应中包含运行信息。默认为False。
运行名称 (可选[str]) –

返回值

一个命名输出的字典。应该包含在: Chain.output_keys中指定的所有输出.

返回类型

Dict[str, Any]

async ainvoke(input: Dict[str, Any], config: Optional[RunnableConfig] = None, **kwargs: Any) → Dict[str, Any]¶

ainvoke 的默认实现，从线程中调用 invoke。

默认实现允许即使 Runnable 没有实现 invoke 的原生异步版本，也可以使用 async 代码。

如果类可以异步运行，子类应该覆盖此方法。

参数

input (Dict[str, Any]) –
config (Optional[RunnableConfig]) –
kwargs (Any) –

返回类型

Dict[str, Any]

apply(input_list: List[Dict[str, Any]], callbacks: Optional[Union[BaseCallbackHandler], BaseCallbackManager]] = None) → List[Dict[str, str]]¶

自 langchain==0.1.0 版本以来已弃用：请使用 batch 代替。

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

参数

input_list (List[Dict[str, Any]]) –
回调函数 (可选[类型联合[列表[BaseCallbackHandler], BaseCallbackManager]]) –

返回类型

列表[字典[字符串, 字符串]]

async aprep_inputs(inputs: 类型联合[字典[字符串, 任何类型], 任何类型]) → 字典[字符串, 字符串]¶

准备链输入，包括添加来自内存的输入。

参数: 输入 (类型联合[字典[字符串, 任何类型], 任何类型]) – 原始输入的字典，如果链只需要一个参数则为单个输入。应包含除由链内存设置的输入之外的所有在 Chain.input_keys 中指定的输入。
返回值: 包括链内存添加的所有输入的字典。
返回类型: 字典[字符串, 字符串]

async aprep_outputs(inputs: 字典[字符串, 字符串], outputs: 字典[字符串, 字符串], return_only_outputs: 布尔型 = False) → 字典[字符串, 字符串]¶

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

参数

输入 (字典[字符串, 字符串]) – 链输入的字典，包括链内存添加的任何输入。
输出 (字典[字符串, 字符串]) – 初始链输出的字典。
return_only_outputs (布尔型) – 是否只返回链输出。如果为False，输入也将添加到最终输出中。

返回值

最终链输出的字典。

返回类型

字典[字符串, 字符串]

async arun(*args: Any, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) → Any¶

自版本langchain==0.1.0已弃用: 请使用 ainvoke 代替。

执行链的便捷方法。

此方法与 Chain.__call__ 的主要区别在于，此方法期望输入直接作为位置参数或关键字参数传递，而 Chain.__call__ 则期望一个包含所有输入的单个输入字典。

参数

*args (Any) – 如果链期望单个输入，则可以将其作为单独的位置参数传递。
回调 (可选[联合[列表[BaseCallbackHandler], BaseCallbackManager]]) – 使用此链运行时使用的回调。这些回调将在构建链期间传递的回调之外被调用，但只有这些运行时回调将传播到对其他对象的调用。
标签 (可选[列表[str]]) – 传递给所有回调的字符串标签列表。这些标签将除构建链期间传递的标签外传递，但只有这些运行时标签将传播到对其他对象的调用。
**kwargs (Any) – 如果链期望多个输入，则可以直接将其作为关键字参数传递。
metadata (Optional[Dict[str, Any]]) –
**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..."

as_tool(args_schema: Optional[Type[BaseModel]] = None, *, name: Optional[str] = None, description: Optional[str] = None, arg_types: Optional[Dict[str, Type]] = None) → BaseTool¶

预览

此API目前处于预览状态，未来可能会发生变化。

从可运行对象创建BaseTool。

as_tool将从可运行对象中实例化一个BaseTool，该工具具有名称、描述和args_schema。尽可能从runnable.get_input_schema推断模式。作为替代方案（例如，如果可运行对象接受字典作为输入且特定的字典键未指定类型），可以直接使用args_schema指定模式。您还可以通过传递arg_types仅指定必需的参数及其类型。

参数

args_schema (可选[类型[BaseModel]]) – 工具的模式。默认为None。
name (可选[字符串]) – 工具的名称。默认为None。
description (可选[字符串]) – 工具的描述。默认为None。
arg_types (可选[字典[字符串, 类型]]) – 参数名称到类型的字典。默认为None。

返回值

BaseTool实例。

返回类型

BaseTool

类型化的字典输入

from typing import List
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, Dict, List
from langchain_core.pydantic_v1 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, Dict, List
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")

新增于版本 0.2.14。

async astream(input: Input, config: Optional[RunnableConfig] = None, **kwargs: Optional[Any]) → AsyncIterator[Output]¶

Certain implementation of astream，which calls ainvoke. Subclasses should override this method if they support streaming output.

参数

input (Input) – The input to the Runnable.
config (Optional[RunnableConfig]) – The config to use for the Runnable. Defaults to None.
kwargs (可选【任何】) – 传递给 Runnable 的附加关键字参数。

生成

The output of the Runnable.

返回类型

AsyncIterator[Output]

astream_events(input: Any, config: Optional[RunnableConfig] = None, *, version: Literal['v1', 'v2'], include_names: Optional[Sequence[str]] = None, include_types: Optional[Sequence[str]] = None, include_tags: Optional[Sequence[str]] = None, exclude_names: Optional[Sequence[str]] = None, exclude_types: Optional[Sequence[str]] = None, exclude_tags: Optional[Sequence[str]] = None, **kwargs: Any) → AsyncIterator[Union[StandardStreamEvent, CustomStreamEvent]]¶

预览

此API目前处于预览状态，未来可能会发生变化。

Generate a stream of events.

Use to create an iterator over StreamEvents that provide real-time information about the progress of the Runnable, including StreamEvents from intermediate results.

A StreamEvent is a dictionary with the following schema

event: str - Event names are of the
format: on_[runnable_type]_(start|stream|end).
name: str - The name of the Runnable that generated the event.
run_id: str - randomly generated ID associated with the given execution of
the Runnable that emitted the event. A child Runnable that gets invoked as part of the execution of a parent Runnable is assigned its own unique ID.
parent_ids: List[str] - The IDs of the parent runnables that
generated the event. The root Runnable will have an empty list. The order of the parent IDs is from the root to the immediate parent. Only available for v2 version of the API. The v1 version of the API will return an empty list.
tags: Optional[List[str]] - The tags of the Runnable that generated
the event.
metadata: Optional[Dict[str, Any]] - The metadata of the Runnable
that generated the event.
data: Dict[str, Any]

Below is a table that illustrates some events that might be emitted by various chains. Metadata fields have been omitted from the table for brevity. Chain definitions have been included after the table.

ATTENTION This reference table is for the V2 version of the schema.

event	name	chunk	input	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]	‘Hello’
on_llm_end	[model name]		‘Hello, human!’
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”}
工具结束事件	some_tool			{“x”: 1, “y”: “2”}
检索器开始事件	【检索器名称】		{“query”: “hello”}
检索器结束事件	【检索器名称】		{“query”: “hello”}	【文档(…), ..】
提示开始事件	【模板名称】		{“question”: “hello”}
提示结束事件	【模板名称】		{“question”: “hello”}	ChatPromptValue(messages: [SystemMessage, …])

除了标准事件外，用户还可以分发自定义事件（以下示例）。

自定义事件仅在API的版本2 中可用！

自定义事件的格式如下

属性	类型	描述
name	str	用户定义的事件名称。
data	Any	与事件关联的数据。这可以是任何东西，尽管我们建议使其可序列化为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}

提示:

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)

参数

input (Any) – Runnable的输入。
config (Optional[RunnableConfig]) – 为Runnable使用的配置。
version (Literal['v1', 'v2']) – 要使用的模式版本，可以是版本2 或版本1。用户应使用版本2。版本1 适用于向后兼容，将在 0.4.0 中弃用。在API稳定之前不会分配默认值。自定义事件仅在版本2 中可用。
include_names (Optional[Sequence[str]]) – 只包含具有匹配名称的runnables的事件。
include_types (Optional[Sequence[str]]) – 只包含具有匹配类型的runnables的事件。
include_tags (Optional[Sequence[str]]) – 只包含具有匹配标签的runnables的事件。
exclude_names (Optional[Sequence[str]]) – 排除具有匹配名称的runnables的事件。
exclude_types (Optional[Sequence[str]]) – 排除具有匹配类型的runnables的事件。
exclude_tags (Optional[Sequence[str]]) – 排除具有匹配标签的runnables的事件。
kwargs (Any) – 传递给Runnable的附加关键字参数。这些将被传递给astream_log，因为此astream_events实现建立在astream_log之上。

生成

异步流中的StreamEvents。

异常

NotImplementedError – 如果版本不是版本1 或版本2。

返回类型

AsyncIterator[Union[StandardStreamEvent, CustomStreamEvent]]

batch(inputs: List[Input], config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) → List[Output]¶

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

默认的批量实现对IO密集型runnable工作得很好。

子类如果可以更有效地批量处理，应重写此方法；例如，如果底层的Runnable使用支持批量模式的API。

参数

inputs (List[Input]) –
config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) –
return_exceptions (bool) –
kwargs (Optional[Any]) –

返回类型

列表[输出]

batch_as_completed(inputs: Sequence[输入], config: Optional[Union[RunnableConfig, Sequence[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) → Iterator[Tuple[int, Union[输出, Exception]]]¶

在输入列表上并行运行调用，按完成顺序产生结果。

参数

inputs (Sequence[输入]) –
config (Optional[Union[RunnableConfig, Sequence[RunnableConfig]]]) –
return_exceptions (bool) –
kwargs (Optional[Any]) –

返回类型

Iterator[Tuple[int, Union[Output, Exception]]]

configurable_alternatives(which: ConfigurableField, *, default_key: str = 'default', prefix_keys: bool = False, **kwargs: Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]]) → RunnableSerializable[Input, Output]¶

配置可在运行时设置的 Runnables 的替代选项。

参数

which (ConfigurableField) – 用于选择替代方案的 ConfigurableField 实例。
default_key (字符串) – 如果未选择替代方案，则使用的默认键。默认为“default”。
prefix_keys (布尔值) – 是否使用 ConfigurableField id作为键的前缀。默认为False。
**kwargs (Union[Runnable[输入, 输出], 可调用函数[[], Runnable[输入, 输出]]]) – 键到 Runnable 实例或返回 Runnable 实例的可调用函数的字典。

返回值

配置了替代选项的新 Runnable。

返回类型

RunnableSerializable[输入, 输出]

from langchain_anthropic import ChatAnthropic
from langchain_core.runnables.utils import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatAnthropic(
    model_name="claude-3-sonnet-20240229"
).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
)

configurable_fields(**kwargs: Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) → RunnableSerializable[Input, Output]¶

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

参数: **kwargs (Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) – 需要配置的 ConfigurableField 实例的字典。
返回值: 配置了字段的新可运行对象。
返回类型: RunnableSerializable[输入, 输出]

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
)

abstract classmethod from_llm(llm: BaseLanguageModel, **kwargs: Any) → LLMEvalChain[源代码]¶

从LLM创建一个新的评估器。

参数

llm (BaseLanguageModel) –
kwargs (Any) –

返回类型

LLMEvalChain

invoke(input: Dict[str, Any], config: Optional[RunnableConfig] = None, **kwargs: Any) → Dict[str, Any]¶

将单个输入转换为输出。重写以实现。

参数

input (Dict[str, Any]) –可供运行体使用的输入。
config (Optional[RunnableConfig]) –在调用运行体时使用的配置。该配置支持如'tags'、'metadata'之类的标准键，用于跟踪目的，'max_concurrency'用于控制并行工作负载，以及其他键。请参阅RunnableConfig以获取更多详细信息。
kwargs (Any) –

返回值

The output of the Runnable.

返回类型

Dict[str, Any]

prep_inputs(inputs: Union[Dict[str, Any], Any]) → Dict[str, str]¶

准备链输入，包括添加来自内存的输入。

参数: 输入 (类型联合[字典[字符串, 任何类型], 任何类型]) – 原始输入的字典，如果链只需要一个参数则为单个输入。应包含除由链内存设置的输入之外的所有在 Chain.input_keys 中指定的输入。
返回值: 包括链内存添加的所有输入的字典。
返回类型: 字典[字符串, 字符串]

prep_outputs(inputs: Dict[str, str], outputs: Dict[str, str], return_only_outputs: bool = False) → Dict[str, str]¶

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

参数

输入 (字典[字符串, 字符串]) – 链输入的字典，包括链内存添加的任何输入。
输出 (字典[字符串, 字符串]) – 初始链输出的字典。
return_only_outputs (布尔型) – 是否只返回链输出。如果为False，输入也将添加到最终输出中。

返回值

最终链输出的字典。

返回类型

字典[字符串, 字符串]

run(*args: Any, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs: Any) → Any¶

自 langchain==0.1.0 版本以来已废止：请使用 invoke 代替。

执行链的便捷方法。

此方法与 Chain.__call__ 的主要区别在于，此方法期望输入直接作为位置参数或关键字参数传递，而 Chain.__call__ 则期望一个包含所有输入的单个输入字典。

参数

*args (Any) – 如果链期望单个输入，则可以将其作为单独的位置参数传递。
回调 (可选[联合[列表[BaseCallbackHandler], BaseCallbackManager]]) – 使用此链运行时使用的回调。这些回调将在构建链期间传递的回调之外被调用，但只有这些运行时回调将传播到对其他对象的调用。
标签 (可选[列表[str]]) – 传递给所有回调的字符串标签列表。这些标签将除构建链期间传递的标签外传递，但只有这些运行时标签将传播到对其他对象的调用。
**kwargs (Any) – 如果链期望多个输入，则可以直接将其作为关键字参数传递。
metadata (Optional[Dict[str, Any]]) –
**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..."

save(file_path: Union[Path, str]) → None¶

保存链。

期望实现 Chain._chain_type 属性，并确保内存为空。: None

参数: file_path (Union[Path, str]) – 保存链的文件路径。
返回类型: None

示例

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

stream(input: Input, config: Optional[RunnableConfig] = None, **kwargs: Optional[Any]) → Iterator[Output]¶

stream的默认实现，调用invoke。如果子类支持流式输出，则应覆盖此方法。

参数

input (Input) – The input to the Runnable.
config (Optional[RunnableConfig]) – The config to use for the Runnable. Defaults to None.
kwargs (可选【任何】) – 传递给 Runnable 的附加关键字参数。

生成

The output of the Runnable.

返回类型

Iterator[Output]

to_json() → Union[SerializedConstructor, SerializedNotImplemented]¶

将Runnable序列化为JSON。

返回值: Runnable的JSON序列化表示。
返回类型: Union[SerializedConstructor, SerializedNotImplemented]

abstract property input_keys: List[str]¶: 预计将在链输入中存在的键。

abstract property output_keys: List[str]¶: 预计将在链输出中存在的键。

langchain.evaluation.schema.LLMEvalChain¶

`langchain.evaluation.schema`.LLMEvalChain¶