`langchain_core.language_models.chat_models`.SimpleChatModel¶

class langchain_core.language_models.chat_models.SimpleChatModel[source]¶

Bases: BaseChatModel

简化的聊天模型实现，供继承使用。

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

param cache: Union[BaseCache, bool, None] = None¶

是否缓存响应。

如果为True，则将使用全局缓存。
如果为False，则不使用缓存。
如果为None，则如果设置了全局缓存，则使用全局缓存，否则不使用缓存。
如果为BaseCache的实例，则将使用提供的缓存。

目前不支持对模型的流式方法进行缓存。

param callback_manager: Optional[BaseCallbackManager] = None¶: [已弃用] 添加到运行跟踪的回调管理器。

param callbacks: Callbacks = None¶: 运行跟踪中需要添加的回调函数。

param custom_get_token_ids: Optional[Callable[[str], List[int]]] = None¶: 用于计算标记的可选编码器。

param metadata: Optional[Dict[str, Any]] = None¶: 要添加到运行跟踪中的元数据。

param tags: Optional[List[str]] = None¶: 要添加到运行跟踪中的标签。

param verbose: bool [Optional]¶: 是否打印响应文本。

__call__(messages: List[BaseMessage], stop: Optional[List[str]] = None, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, **kwargs: Any) → BaseMessage¶

[Deprecated]

Notes

Deprecated since version langchain-core==0.1.7: Use invoke instead.

Parameters

messages (List[BaseMessage]) –
stop (Optional[List[str]]) –
callbacks (Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]]) –
kwargs (Any) –

Return type

BaseMessage

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

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

对于IO绑定的可运行对象，默认的批处理实现效果很好。

如果子类可以更有效地批处理，应该重写这个方法；例如，如果底层可运行对象使用支持批处理模式的API。

Parameters

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

Return type

List[Output]

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

在并行上运行一组输入的 ainvoke，完成后产生结果。

Parameters

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

Return type

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

async agenerate(messages: List[List[BaseMessage]], stop: Optional[List[str]] = None, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, run_id: Optional[UUID] = None, **kwargs: Any) → LLMResult¶

将一个序列的提示异步传递给模型，并返回生成的结果。

该方法应该利用暴露批处理API的模型进行批处理调用。

在以下情况下使用此方法：

利用批处理调用，
需要模型生成的输出不仅仅是顶部生成的值，
构建对底层语言模型类型不可知的链式结构（例如，纯文本完成模型与聊天模型）。

参数：

messages: 消息列表的列表。 stop: 生成时使用的停止词。模型输出在出现这些子字符串的第一次发生时被截断。 callbacks: 要传递的回调函数。用于在生成过程中执行额外的功能，如日志记录或流式传输。 **kwargs: 任意额外的关键字参数。这些通常传递给模型提供者的API调用。

返回：

一个LLMResult，其中包含每个输入提示的候选生成列表和额外的模型提供者特定输出。

Parameters

messages (List[List[BaseMessage]]) –
stop (Optional[List[str]]) –
callbacks (Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]]) –
tags (Optional[List[str]]) –
metadata (Optional[Dict[str, Any]]) –
run_name (Optional[str]) –
run_id (Optional[UUID]) –
kwargs (Any) –

Return type

LLMResult

async agenerate_prompt(prompts: List[PromptValue], stop: Optional[List[str]] = None, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, **kwargs: Any) → LLMResult¶

异步传递一系列提示并返回模型生成。

此方法应该利用暴露批处理API的模型进行批处理调用。

在以下情况下使用此方法：

利用批处理调用，
需要模型生成的更多输出而不仅仅是顶部生成的值，
构建对基础语言模型类型不可知的链式结构（例如，纯文本完成模型与聊天模型）。

参数：

prompts: PromptValues的列表。PromptValue是一个对象，可以转换为与任何语言模型的格式匹配（纯文本生成模型的字符串和聊天模型的BaseMessages）。 stop: 生成时要使用的停止词。模型输出在出现这些子字符串的第一次发生时被截断。 callbacks: 要传递的回调。用于执行生成过程中的额外功能，例如日志记录或流式传输。 **kwargs: 任意的额外关键字参数。这些通常传递给模型提供者的API调用。

返回：

一个LLMResult，其中包含每个输入提示的候选生成列表和额外的模型提供者特定输出。

Parameters

prompts (List[PromptValue]) –
stop (Optional[List[str]]) –
callbacks (Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]]) –
kwargs (Any) –

Return type

LLMResult

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

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

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

如果子类可以异步运行，应该重写这个方法。

Parameters

input (LanguageModelInput) –
config (Optional[RunnableConfig]) –
stop (Optional[List[str]]) –
kwargs (Any) –

Return type

BaseMessage

async apredict(text: str, *, stop: Optional[Sequence[str]] = None, **kwargs: Any) → str¶

[Deprecated]

Notes

Deprecated since version langchain-core==0.1.7: Use ainvoke instead.

Parameters

text (str) –
stop (Optional[Sequence[str]]) –
kwargs (Any) –

Return type

str

async apredict_messages(messages: List[BaseMessage], *, stop: Optional[Sequence[str]] = None, **kwargs: Any) → BaseMessage¶

[Deprecated]

Notes

Deprecated since version langchain-core==0.1.7: Use ainvoke instead.

Parameters

messages (List[BaseMessage]) –
stop (Optional[Sequence[str]]) –
kwargs (Any) –

Return type

BaseMessage

assign(**kwargs: Union[Runnable[Dict[str, Any], Any], Callable[[Dict[str, Any]], Any], Mapping[str, Union[Runnable[Dict[str, Any], Any], Callable[[Dict[str, Any]], Any]]]]) → RunnableSerializable[Any, Any]¶

为这个可运行程序的字典输出分配新的字段。返回一个新的可运行程序。

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}"
)
llm = FakeStreamingListLLM(responses=["foo-lish"])

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

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

print(chain_with_assign.input_schema.schema())
# {'title': 'PromptInput', 'type': 'object', 'properties':
{'question': {'title': 'Question', 'type': 'string'}}}
print(chain_with_assign.output_schema.schema()) #
{'title': 'RunnableSequenceOutput', 'type': 'object', 'properties':
{'str': {'title': 'Str',
'type': 'string'}, 'hello': {'title': 'Hello', 'type': 'string'}}}

Parameters: kwargs (Union[Runnable[Dict[str, Any], Any], Callable[[Dict[str, Any]], Any], Mapping[str, Union[Runnable[Dict[str, Any], Any], Callable[[Dict[str, Any]], Any]]]]) –
Return type: RunnableSerializable[Any, Any]

async astream(input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[List[str]] = None, **kwargs: Any) → AsyncIterator[BaseMessageChunk]¶

默认实现astream，调用ainvoke。如果子类支持流式输出，则应该重写此方法。

Parameters

input (LanguageModelInput) –
config (Optional[RunnableConfig]) –
stop (Optional[List[str]]) –
kwargs (Any) –

Return type

AsyncIterator[BaseMessageChunk]

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[StreamEvent]¶

[Beta] 生成事件流。

用于创建一个迭代器，遍历StreamEvents，提供有关可运行对象进展的实时信息，包括来自中间结果的StreamEvents。

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

event: str - 事件名称的格式为：on_[runnable_type]_(start|stream|end)。
name: str - 生成事件的可运行对象的名称。
run_id: str - 与生成事件的可运行对象的执行关联的随机生成的ID。作为父可运行对象执行的一部分调用的子可运行对象被分配其自己的唯一ID。
tags: Optional[List[str]] - 生成事件的可运行对象的标签。
metadata: Optional[Dict[str, Any]] - 生成事件的可运行对象的元数据。
data: Dict[str, Any]

下面是一个表格，展示了各种链可能发出的一些事件。为简洁起见，表格中省略了元数据字段。表格后包含了链定义。

注意此参考表格适用于模式的V2版本。

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”}
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, …])

以下是与上述事件相关的声明：

format_docs:

def format_docs(docs: List[Document]) -> str:
    '''格式化文档。'''
    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")
]

# 将产生以下事件（为简洁起见，省略了run_id）：
[
    {
        "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": [],
    },
]

参数:: input: 可运行对象的输入。 config: 可运行对象的配置。 version: 要使用的模式版本，可以是`v2`或`v1`。用户应该使用`v2`。`v1`用于向后兼容，将在0.4.0中弃用。在API稳定之前不会分配默认值。 include_names: 仅包括与名称匹配的可运行对象的事件。 include_types: 仅包括与类型匹配的可运行对象的事件。 include_tags: 仅包括与标签匹配的可运行对象的事件。 exclude_names: 排除与名称匹配的可运行对象的事件。 exclude_types: 排除与类型匹配的可运行对象的事件。 exclude_tags: 排除与标签匹配的可运行对象的事件。 kwargs: 要传递给可运行对象的其他关键字参数。这些参数将传递给astream_log，因为此astream_events的实现是基于astream_log构建的。
返回:: 一个StreamEvents的异步流。

Notes

Parameters

input (Any) –
config (Optional[RunnableConfig]) –
version (Literal['v1', 'v2']) –
include_names (Optional[Sequence[str]]) –
include_types (Optional[Sequence[str]]) –
include_tags (Optional[Sequence[str]]) –
exclude_names (Optional[Sequence[str]]) –
exclude_types (Optional[Sequence[str]]) –
exclude_tags (Optional[Sequence[str]]) –
kwargs (Any) –

Return type

AsyncIterator[StreamEvent]

async astream_log(input: Any, config: Optional[RunnableConfig] = None, *, diff: bool = True, with_streamed_output_list: bool = True, 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) → Union[AsyncIterator[RunLogPatch], AsyncIterator[RunLog]]¶

从可运行对象中流式传输所有输出，如回调系统所报告的。这包括LLMs、检索器、工具等的所有内部运行。

输出被流式传输为Log对象，其中包括描述每个步骤中运行状态如何改变的jsonpatch操作列表，以及运行的最终状态。

jsonpatch操作可以按顺序应用以构建状态。

参数：: input：可运行对象的输入。 config：用于可运行对象的配置。 diff：是否在每个步骤之间生成差异，或者生成当前状态。 with_streamed_output_list：是否生成流式输出列表。 include_names：仅包括具有这些名称的日志。 include_types：仅包括具有这些类型的日志。 include_tags：仅包括具有这些标签的日志。 exclude_names：排除具有这些名称的日志。 exclude_types：排除具有这些类型的日志。 exclude_tags：排除具有这些标签的日志。

Parameters

input (Any) –
config (Optional[RunnableConfig]) –
diff (bool) –
with_streamed_output_list (bool) –
include_names (Optional[Sequence[str]]) –
include_types (Optional[Sequence[str]]) –
include_tags (Optional[Sequence[str]]) –
exclude_names (Optional[Sequence[str]]) –
exclude_types (Optional[Sequence[str]]) –
exclude_tags (Optional[Sequence[str]]) –
kwargs (Any) –

Return type

Union[AsyncIterator[RunLogPatch], AsyncIterator[RunLog]]

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

默认实现一个转换的方法，它会缓冲输入并调用流。如果子类可以在生成输入时开始产生输出，应该重写这个方法。

Parameters

input (AsyncIterator[Input]) –
config (Optional[RunnableConfig]) –
kwargs (Optional[Any]) –

Return type

AsyncIterator[Output]

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

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

默认的batch实现适用于IO绑定的可运行程序。

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

Parameters

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

Return type

List[Output]

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

在输入列表上并行运行invoke，当它们完成时产生结果。

Parameters

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

Return type

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

bind(**kwargs: Any) → Runnable[Input, Output]¶

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

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

示例：

from langchain_community.chat_models import ChatOllama
from langchain_core.output_parsers import StrOutputParser

llm = ChatOllama(model='llama2')

# 没有绑定。
chain = (
    llm
    | StrOutputParser()
)

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# 输出为 'One two three four five.'

# 有绑定。
chain = (
    llm.bind(stop=["three"])
    | StrOutputParser()
)

chain.invoke("Repeat quoted words exactly: 'One two three four five.'")
# 输出为 'One two'

Parameters: kwargs (Any) –
Return type: Runnable[Input, Output]

bind_tools(tools: Sequence[Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool]], **kwargs: Any) → Runnable[LanguageModelInput, BaseMessage]¶

Parameters

tools (Sequence[Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool]]) –
kwargs (Any) –

Return type

Runnable[LanguageModelInput, BaseMessage]

call_as_llm(message: str, stop: Optional[List[str]] = None, **kwargs: Any) → str¶

[Deprecated]

Notes

Deprecated since version langchain-core==0.1.7: Use invoke instead.

Parameters

message (str) –
stop (Optional[List[str]]) –
kwargs (Any) –

Return type

str

config_schema(*, include: Optional[Sequence[str]] = None) → Type[BaseModel]¶

此可运行程序接受的配置类型，指定为一个pydantic模型。

要将字段标记为可配置，请参见`configurable_fields`和`configurable_alternatives`方法。

参数：: include: 要包含在配置模式中的字段列表。
返回：: 一个可以用于验证配置的pydantic模型。

Parameters: include (Optional[Sequence[str]]) –
Return type: Type[BaseModel]

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

为可以在运行时设置的可运行项配置备选项。

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

# 使用默认模型ChatAnthropic
print(model.invoke("which organization created you?").content)

# 使用ChatOpenAI
print(
    model.with_config(
        configurable={"llm": "openai"}
    ).invoke("which organization created you?").content
)

Parameters

which (ConfigurableField) –
default_key (str) –
prefix_keys (bool) –
kwargs (Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]]) –

Return type

RunnableSerializable[Input, Output]

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

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

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="输出中的最大标记数",
        description="输出中的最大标记数",
    )
)

# max_tokens = 20
print(
    "max_tokens_20: ",
    model.invoke("告诉我一些关于国际象棋的事情").content
)

# max_tokens = 200
print("max_tokens_200: ", model.with_config(
    configurable={"output_token_number": 200}
    ).invoke("告诉我一些关于国际象棋的事情").content
)

Parameters: kwargs (Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) –
Return type: RunnableSerializable[Input, Output]

classmethod construct(_fields_set: Optional[SetStr] = None, **values: Any) → Model¶

Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

Parameters

_fields_set (Optional[SetStr]) –
values (Any) –

Return type

Model

copy(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, update: Optional[DictStrAny] = None, deep: bool = False) → Model¶

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters

include (Optional[Union[AbstractSetIntStr, MappingIntStrAny]]) – fields to include in new model
exclude (Optional[Union[AbstractSetIntStr, MappingIntStrAny]]) – fields to exclude from new model, as with values this takes precedence over include
update (Optional[DictStrAny]) – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data
deep (bool) – set to True to make a deep copy of the model
self (Model) –

Returns

new model instance

Return type

Model

dict(**kwargs: Any) → Dict¶

返回LLM的字典。

Parameters: kwargs (Any) –
Return type: Dict

classmethod from_orm(obj: Any) → Model¶

Parameters: obj (Any) –
Return type: Model

generate(messages: List[List[BaseMessage]], stop: Optional[List[str]] = None, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, *, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, run_id: Optional[UUID] = None, **kwargs: Any) → LLMResult¶

将一系列提示传递给模型，并返回模型生成的结果。

对于公开批处理API的模型，此方法应该使用批处理调用。

在以下情况下使用此方法：

利用批处理调用，
需要从模型中获取更多输出，而不仅仅是顶部生成的值，
正在构建对基础语言模型类型不可知的链式结构（例如，纯文本完成模型与聊天模型）。

参数：

messages：消息列表的列表。 stop：在生成时要使用的停止词。模型输出在出现这些子字符串的第一次出现时被截断。 callbacks：要传递的回调。用于在生成过程中执行额外功能，如日志记录或流式传输。 **kwargs：任意额外的关键字参数。通常这些参数会传递给模型提供者的API调用。

返回：

一个LLMResult，其中包含每个输入提示的候选生成列表以及额外的模型提供者特定输出。

Parameters

messages (List[List[BaseMessage]]) –
stop (Optional[List[str]]) –
callbacks (Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]]) –
tags (Optional[List[str]]) –
metadata (Optional[Dict[str, Any]]) –
run_name (Optional[str]) –
run_id (Optional[UUID]) –
kwargs (Any) –

Return type

LLMResult

generate_prompt(prompts: List[PromptValue], stop: Optional[List[str]] = None, callbacks: Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]] = None, **kwargs: Any) → LLMResult¶

将一系列提示传递给模型，并返回模型生成的结果。

对于暴露批处理API的模型，此方法应该使用批处理调用。

在以下情况下使用此方法：

利用批处理调用，
需要从模型获得更多输出而不仅仅是顶部生成的值，
正在构建对基础语言模型类型不可知的链（例如，纯文本完成模型与聊天模型）。

参数：

prompts: PromptValues列表。PromptValue是一个对象，可以转换为与任何语言模型匹配的格式（纯文本生成模型的字符串和聊天模型的BaseMessages）。 stop: 生成时要使用的停止词。模型输出在出现这些子字符串的第一次截断。 callbacks: 要传递的回调。用于在生成过程中执行额外功能，如日志记录或流式处理。 **kwargs: 任意额外的关键字参数。这些通常传递给模型提供者API调用。

返回：

一个LLMResult，其中包含每个输入提示的候选生成列表和额外的模型提供者特定输出。

Parameters

prompts (List[PromptValue]) –
stop (Optional[List[str]]) –
callbacks (Optional[Union[List[BaseCallbackHandler], BaseCallbackManager]]) –
kwargs (Any) –

Return type

LLMResult

get_graph(config: Optional[RunnableConfig] = None) → Graph¶

返回此可运行对象的图形表示。

Parameters: config (Optional[RunnableConfig]) –
Return type: Graph

get_input_schema(config: Optional[RunnableConfig] = None) → Type[BaseModel]¶

获取一个Pydantic模型，用于验证可运行对象的输入。

利用configurable_fields和configurable_alternatives方法的可运行对象将具有动态输入模式，取决于调用可运行对象时使用的配置。

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

参数：: config：生成模式时要使用的配置。
返回：: 可用于验证输入的Pydantic模型。

Parameters: config (Optional[RunnableConfig]) –
Return type: Type[BaseModel]

classmethod get_lc_namespace() → List[str]¶

获取langchain对象的命名空间。

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

Return type: List[str]

get_name(suffix: Optional[str] = None, *, name: Optional[str] = None) → str¶

获取可运行对象的名称。

Parameters

suffix (Optional[str]) –
name (Optional[str]) –

Return type

str

get_num_tokens(text: str) → int¶

获取文本中存在的标记数量。

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

参数：: text：要标记化的字符串输入。
返回：: 文本中标记的整数数量。

Parameters: text (str) –
Return type: int

get_num_tokens_from_messages(messages: List[BaseMessage]) → int¶

获取消息中的令牌数量。

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

参数：: messages：要进行标记化的消息输入。
返回：: 跨消息中令牌数量的总和。

Parameters: messages (List[BaseMessage]) –
Return type: int

get_output_schema(config: Optional[RunnableConfig] = None) → Type[BaseModel]¶

获取一个Pydantic模型，用于验证可运行代码的输出。

利用configurable_fields和configurable_alternatives方法的可运行代码将具有动态输出模式，取决于调用可运行代码时使用的配置。

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

参数：: config: 生成模式时要使用的配置。
返回：: 一个Pydantic模型，可用于验证输出。

Parameters: config (Optional[RunnableConfig]) –
Return type: Type[BaseModel]

get_prompts(config: Optional[RunnableConfig] = None) → List[BasePromptTemplate]¶

Parameters: config (Optional[RunnableConfig]) –
Return type: List[BasePromptTemplate]

get_token_ids(text: str) → List[int]¶

返回文本中标记的令牌的有序id。

参数：: text：要进行标记化的字符串输入。
返回：: 返回一个id列表，对应于文本中的标记，按照它们在文本中出现的顺序。

Parameters: text (str) –
Return type: List[int]

invoke(input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[List[str]] = None, **kwargs: Any) → BaseMessage¶

将单个输入转换为输出。覆盖以实现。

参数：: input: 可运行的输入。 config: 在调用可运行对象时要使用的配置。

该配置支持标准键，如’tags’，’metadata’用于跟踪目的， ‘max_concurrency’用于控制并行执行的工作量，以及其他键。请参考RunnableConfig获取更多详细信息。
返回：: 可运行对象的输出。

Parameters

input (LanguageModelInput) –
config (Optional[RunnableConfig]) –
stop (Optional[List[str]]) –
kwargs (Any) –

Return type

BaseMessage

classmethod is_lc_serializable() → bool¶

这个类是否可序列化？

Return type: bool

json(*, include: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, exclude: Optional[Union[AbstractSetIntStr, MappingIntStrAny]] = None, by_alias: bool = False, skip_defaults: Optional[bool] = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Optional[Callable[[Any], Any]] = None, models_as_dict: bool = True, **dumps_kwargs: Any) → unicode¶

Generate a JSON representation of the model, include and exclude arguments as per dict().

encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().

Parameters

include (Optional[Union[AbstractSetIntStr, MappingIntStrAny]]) –
exclude (Optional[Union[AbstractSetIntStr, MappingIntStrAny]]) –
by_alias (bool) –
skip_defaults (Optional[bool]) –
exclude_unset (bool) –
exclude_defaults (bool) –
exclude_none (bool) –
encoder (Optional[Callable[[Any], Any]]) –
models_as_dict (bool) –
dumps_kwargs (Any) –

Return type

unicode

classmethod lc_id() → List[str]¶

用于序列化目的的此类的唯一标识符。

唯一标识符是一个描述对象路径的字符串列表。

Return type: List[str]

map() → Runnable[List[Input], List[Output]]¶

返回一个新的Runnable，它将一个输入列表映射到一个输出列表，通过对每个输入调用invoke()。

示例：

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]

Return type: Runnable[List[Input], List[Output]]

classmethod parse_file(path: Union[str, Path], *, content_type: unicode = None, encoding: unicode = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model¶

Parameters

path (Union[str, Path]) –
content_type (unicode) –
encoding (unicode) –
proto (Protocol) –
allow_pickle (bool) –

Return type

Model

classmethod parse_obj(obj: Any) → Model¶

Parameters: obj (Any) –
Return type: Model

classmethod parse_raw(b: Union[str, bytes], *, content_type: unicode = None, encoding: unicode = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model¶

Parameters

b (Union[str, bytes]) –
content_type (unicode) –
encoding (unicode) –
proto (Protocol) –
allow_pickle (bool) –

Return type

Model

pick(keys: Union[str, List[str]]) → RunnableSerializable[Any, Any]¶

从可运行的字典输出中选择键。

选择单个键：

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]"}

Parameters: keys (Union[str, List[str]]) –
Return type: RunnableSerializable[Any, Any]

pipe(*others: Union[Runnable[Any, Other], Callable[[Any], Other]], name: Optional[str] = None) → RunnableSerializable[Input, Other]¶

将这个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)
# 或者等价地：
# 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]

Parameters

others (Union[Runnable[Any, Other], Callable[[Any], Other]]) –
name (Optional[str]) –

Return type

RunnableSerializable[Input, Other]

predict(text: str, *, stop: Optional[Sequence[str]] = None, **kwargs: Any) → str¶

[Deprecated]

Notes

Deprecated since version langchain-core==0.1.7: Use invoke instead.

Parameters

text (str) –
stop (Optional[Sequence[str]]) –
kwargs (Any) –

Return type

str

predict_messages(messages: List[BaseMessage], *, stop: Optional[Sequence[str]] = None, **kwargs: Any) → BaseMessage¶

[Deprecated]

Notes

Deprecated since version langchain-core==0.1.7: Use invoke instead.

Parameters

messages (List[BaseMessage]) –
stop (Optional[Sequence[str]]) –
kwargs (Any) –

Return type

BaseMessage

classmethod schema(by_alias: bool = True, ref_template: unicode = '#/definitions/{model}') → DictStrAny¶

Parameters

by_alias (bool) –
ref_template (unicode) –

Return type

DictStrAny

classmethod schema_json(*, by_alias: bool = True, ref_template: unicode = '#/definitions/{model}', **dumps_kwargs: Any) → unicode¶

Parameters

by_alias (bool) –
ref_template (unicode) –
dumps_kwargs (Any) –

Return type

unicode

stream(input: LanguageModelInput, config: Optional[RunnableConfig] = None, *, stop: Optional[List[str]] = None, **kwargs: Any) → Iterator[BaseMessageChunk]¶

默认的流实现，调用invoke方法。如果子类支持流式输出，应该重写这个方法。

Parameters

input (LanguageModelInput) –
config (Optional[RunnableConfig]) –
stop (Optional[List[str]]) –
kwargs (Any) –

Return type

Iterator[BaseMessageChunk]

to_json() → Union[SerializedConstructor, SerializedNotImplemented]¶

将可运行对象序列化为JSON。

Return type: Union[SerializedConstructor, SerializedNotImplemented]

to_json_not_implemented() → SerializedNotImplemented¶

Return type: SerializedNotImplemented

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

transform的默认实现，它会缓冲输入然后调用stream。如果子类能够在生成输入的同时开始生成输出，那么应该重写这个方法。

Parameters

input (Iterator[Input]) –
config (Optional[RunnableConfig]) –
kwargs (Optional[Any]) –

Return type

Iterator[Output]

classmethod update_forward_refs(**localns: Any) → None¶

Try to update ForwardRefs on fields based on this Model, globalns and localns.

Parameters: localns (Any) –
Return type: None

classmethod validate(value: Any) → Model¶

Parameters: value (Any) –
Return type: Model

with_config(config: Optional[RunnableConfig] = None, **kwargs: Any) → Runnable[Input, Output]¶

将配置绑定到一个可运行对象，返回一个新的可运行对象。

Parameters

config (Optional[RunnableConfig]) –
kwargs (Any) –

Return type

Runnable[Input, Output]

with_fallbacks(fallbacks: Sequence[Runnable[Input, Output]], *, exceptions_to_handle: Tuple[Type[BaseException], ...] = (<class 'Exception'>,), exception_key: Optional[str] = None) → RunnableWithFallbacksT[Input, Output]¶

为可运行对象添加回退，返回一个新的可运行对象。

示例：

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

参数：: fallbacks: 一系列可运行对象，如果原始可运行对象失败，则尝试。 exceptions_to_handle: 要处理的异常类型元组。 exception_key: 如果指定了字符串，则处理的异常将作为输入的一部分传递给回退，指定的键下。如果为None，则异常将不会传递给回退。如果使用，基本可运行对象及其回退必须接受字典作为输入。
返回：: 一个新的可运行对象，将尝试原始可运行对象，然后按顺序尝试每个回退，以处理失败情况。

Parameters

fallbacks (Sequence[Runnable[Input, Output]]) –
exceptions_to_handle (Tuple[Type[BaseException], ...]) –
exception_key (Optional[str]) –

Return type

RunnableWithFallbacksT[Input, Output]

with_listeners(*, on_start: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None, on_end: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None, on_error: Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]] = None) → Runnable[Input, Output]¶

将生命周期监听器绑定到一个Runnable，返回一个新的Runnable。

on_start: 在runnable开始运行之前调用，使用Run对象。 on_end: 在runnable完成运行后调用，使用Run对象。 on_error: 如果runnable抛出错误，则调用，使用Run对象。

Run对象包含有关运行的信息，包括其id、类型、输入、输出、错误、开始时间、结束时间以及添加到运行中的任何标签或元数据。

示例：

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)

Parameters

on_start (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) –
on_end (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) –
on_error (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) –

Return type

Runnable[Input, Output]

with_retry(*, retry_if_exception_type: ~typing.Tuple[~typing.Type[BaseException], ...] = (<class 'Exception'>,), wait_exponential_jitter: bool = True, stop_after_attempt: int = 3) → 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)

参数:: retry_if_exception_type: 重试的异常类型元组 wait_exponential_jitter: 是否在重试之间的等待时间中添加抖动 stop_after_attempt: 放弃之前尝试的最大次数
返回:: 一个新的Runnable，在发生异常时重试原始的runnable。

Parameters

retry_if_exception_type (Tuple[Type[BaseException], ...]) –
wait_exponential_jitter (bool) –
stop_after_attempt (int) –

Return type

Runnable[Input, Output]

with_structured_output(schema: Union[Dict, Type[BaseModel]], **kwargs: Any) → Runnable[Union[PromptValue, str, Sequence[Union[BaseMessage, List[str], Tuple[str, str], str, Dict[str, Any]]]], Union[Dict, BaseModel]]¶

在这个类中未实现。

Parameters

schema (Union[Dict, Type[BaseModel]]) –
kwargs (Any) –

Return type

Runnable[Union[PromptValue, str, Sequence[Union[BaseMessage, List[str], Tuple[str, str], str, Dict[str, Any]]]], Union[Dict, BaseModel]]

with_types(*, input_type: Optional[Type[Input]] = None, output_type: Optional[Type[Output]] = None) → Runnable[Input, Output]¶

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

Parameters

input_type (Optional[Type[Input]]) –
output_type (Optional[Type[Output]]) –

Return type

Runnable[Input, Output]

property InputType: TypeAlias¶: 获取此可运行对象的输入类型。

property OutputType: Any¶: 获取此可运行程序的输出类型。

property config_specs: List[langchain_core.runnables.utils.ConfigurableFieldSpec]¶: 列出此可运行程序的可配置字段。

property input_schema: Type[pydantic.main.BaseModel]¶: 此可运行程序接受的输入类型，由pydantic模型指定。

property lc_attributes: Dict¶

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

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

property lc_secrets: Dict[str, str]¶

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

例如，: {“openai_api_key”: “OPENAI_API_KEY”}

name: Optional[str] = None¶: 可运行对象的名称。用于调试和跟踪。

property output_schema: Type[pydantic.main.BaseModel]¶: 这个可运行程序产生的输出类型被指定为一个pydantic模型。

Examples using SimpleChatModel¶

custom_chat_model.md

langchain_core.language_models.chat_models.SimpleChatModel¶

Examples using SimpleChatModel¶

`langchain_core.language_models.chat_models`.SimpleChatModel¶