ChatPromptTemplate#
- class langchain_core.prompts.chat.ChatPromptTemplate[源代码]#
-
聊天模型的提示模板。
用于为聊天模型创建灵活的模板提示。
示例
在版本0.2.24中更改:您可以直接将
ChatPromptTemplate.from_messages()
支持的任何类似消息的格式传递给ChatPromptTemplate()
初始化。from langchain_core.prompts import ChatPromptTemplate template = ChatPromptTemplate([ ("system", "You are a helpful AI bot. Your name is {name}."), ("human", "Hello, how are you doing?"), ("ai", "I'm doing well, thanks!"), ("human", "{user_input}"), ]) prompt_value = template.invoke( { "name": "Bob", "user_input": "What is your name?" } ) # Output: # ChatPromptValue( # messages=[ # SystemMessage(content='You are a helpful AI bot. Your name is Bob.'), # HumanMessage(content='Hello, how are you doing?'), # AIMessage(content="I'm doing well, thanks!"), # HumanMessage(content='What is your name?') # ] #)
消息占位符:
# In addition to Human/AI/Tool/Function messages, # you can initialize the template with a MessagesPlaceholder # either using the class directly or with the shorthand tuple syntax: template = ChatPromptTemplate([ ("system", "You are a helpful AI bot."), # Means the template will receive an optional list of messages under # the "conversation" key ("placeholder", "{conversation}") # Equivalently: # MessagesPlaceholder(variable_name="conversation", optional=True) ]) prompt_value = template.invoke( { "conversation": [ ("human", "Hi!"), ("ai", "How can I assist you today?"), ("human", "Can you make me an ice cream sundae?"), ("ai", "No.") ] } ) # Output: # ChatPromptValue( # messages=[ # SystemMessage(content='You are a helpful AI bot.'), # HumanMessage(content='Hi!'), # AIMessage(content='How can I assist you today?'), # HumanMessage(content='Can you make me an ice cream sundae?'), # AIMessage(content='No.'), # ] #)
单变量模板:
如果你的提示只有一个输入变量(即只有一个“{variable_nams}”实例),并且你使用非字典对象调用模板,提示模板会将提供的参数注入到该变量位置。
from langchain_core.prompts import ChatPromptTemplate template = ChatPromptTemplate([ ("system", "You are a helpful AI bot. Your name is Carl."), ("human", "{user_input}"), ]) prompt_value = template.invoke("Hello, there!") # Equivalent to # prompt_value = template.invoke({"user_input": "Hello, there!"}) # Output: # ChatPromptValue( # messages=[ # SystemMessage(content='You are a helpful AI bot. Your name is Carl.'), # HumanMessage(content='Hello, there!'), # ] # )
从多种消息格式创建聊天提示模板。
- Parameters:
messages – 消息表示的序列。 消息可以使用以下格式表示: (1) BaseMessagePromptTemplate, (2) BaseMessage, (3) 2元组 (消息类型, 模板); 例如, (“human”, “{user_input}”), (4) 2元组 (消息类, 模板), (5) 一个字符串,它是 (“human”, 模板) 的简写; 例如, “{user_input}”.
template_format – 模板的格式。默认为“f-string”。
input_variables – 一个变量名称的列表,这些变量的值需要作为提示的输入。
optional_variables – 占位符变量名称的列表
inferred (或可选的MessagePlaceholder。这些变量是自动的)
它们。 (来自提示,用户无需提供)
partial_variables – 提示模板携带的部分变量的字典。部分变量填充模板,这样你就不需要在每次调用提示时传递它们。
validate_template – 是否验证模板。
input_types – 提示模板期望的变量类型的字典。如果未提供,则假定所有变量均为字符串。
- Returns:
聊天提示模板。
示例
从消息模板列表实例化:
template = ChatPromptTemplate([ ("human", "Hello, how are you?"), ("ai", "I'm doing well, thanks!"), ("human", "That's good to hear."), ])
从混合消息格式实例化:
template = ChatPromptTemplate([ SystemMessage(content="hello"), ("human", "Hello, how are you?"), ])
注意
ChatPromptTemplate 实现了标准的
Runnable Interface
。🏃Runnable Interface
接口在可运行对象上提供了额外的方法,例如with_types
,with_retry
,assign
,bind
,get_graph
, 等等。- param input_types: Dict[str, Any] [Optional]#
提示模板期望的变量类型的字典。 如果未提供,则假定所有变量均为字符串。
- param input_variables: list[str] [Required]#
需要作为提示输入的变量名称列表。
- param messages: Annotated[list[MessageLike], SkipValidation()] [Required]#
消息列表,包含消息提示模板或消息。
- param metadata: Dict[str, Any] | None = None#
用于追踪的元数据。
- param optional_variables: list[str] = []#
optional_variables: 占位符或MessagePlaceholder的可选变量名称列表。这些变量是从提示中自动推断出来的,用户无需提供它们。
- param output_parser: BaseOutputParser | None = None#
如何解析在此格式化提示上调用LLM的输出。
- param partial_variables: Mapping[str, Any] [Optional]#
提示模板携带的部分变量的字典。
部分变量填充模板,这样您就不需要在每次调用提示时传递它们。
- param tags: list[str] | None = None#
用于追踪的标签。
- param validate_template: bool = False#
是否尝试验证模板。
- 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绑定的runnables效果很好。
如果子类能够更高效地进行批处理,则应重写此方法; 例如,如果底层的Runnable使用支持批处理模式的API。
- Parameters:
inputs (list[Input]) – Runnable 的输入列表。
config (RunnableConfig | list[RunnableConfig] | None) – 调用Runnable时使用的配置。 该配置支持标准键,如用于跟踪目的的'tags'、'metadata',用于控制并行工作量的'max_concurrency',以及其他键。有关更多详细信息,请参阅RunnableConfig。默认为None。
return_exceptions (bool) – 是否返回异常而不是抛出它们。默认为 False。
kwargs (Any | None) – 传递给Runnable的额外关键字参数。
- Returns:
Runnable 的输出列表。
- Return type:
列表[输出]
- 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,在它们完成时产生结果。
- Parameters:
inputs (Sequence[Input]) – Runnable 的输入列表。
config (RunnableConfig | Sequence[RunnableConfig] | None) – 调用Runnable时使用的配置。 该配置支持标准键,如用于跟踪目的的'tags'、'metadata',用于控制并行工作量的'max_concurrency',以及其他键。有关更多详细信息,请参阅RunnableConfig。默认为None。默认为None。
return_exceptions (bool) – 是否返回异常而不是抛出它们。默认为 False。
kwargs (Any | None) – 传递给Runnable的额外关键字参数。
- Yields:
输入索引和Runnable输出的元组。
- Return type:
AsyncIterator[元组[int, Output | 异常]]
- async aformat(**kwargs: Any) str #
异步格式化聊天模板为字符串。
- Parameters:
**kwargs (Any) – 用于填充此聊天模板中所有模板消息中的模板变量的关键字参数。
- Returns:
格式化字符串。
- Return type:
字符串
- async aformat_messages(**kwargs: Any) list[BaseMessage] [source]#
异步格式化聊天模板为最终消息列表。
- Parameters:
**kwargs (Any) – 用于填充此聊天模板中所有模板消息中的模板变量的关键字参数。
- Returns:
格式化消息列表。
- Raises:
ValueError – 如果输入意外。
- Return type:
列表[BaseMessage]
- async aformat_prompt(**kwargs: Any) PromptValue #
异步格式提示。应返回一个PromptValue。
- Parameters:
**kwargs (Any) – 用于格式化的关键字参数。
- Returns:
提示值。
- Return type:
- async ainvoke(input: dict, config: RunnableConfig | None = None, **kwargs: Any) PromptValue #
异步调用提示。
- Parameters:
input (dict) – 字典,提示的输入。
config (RunnableConfig | None) – RunnableConfig,用于配置提示的配置。
kwargs (Any)
- Returns:
提示的输出。
- Return type:
- append(message: BaseMessagePromptTemplate | BaseMessage | BaseChatPromptTemplate | tuple[str | type, str | list[dict] | list[object]] | str) None [源代码]#
将消息附加到聊天模板的末尾。
- Parameters:
message (BaseMessagePromptTemplate | BaseMessage | BaseChatPromptTemplate | tuple[str | type, str | list[dict] | list[object]] | str) – 要附加的消息的表示。
- Return type:
无
- async astream(input: Input, config: RunnableConfig | None = None, **kwargs: Any | None) AsyncIterator[Output] #
astream的默认实现,调用ainvoke。 如果子类支持流式输出,则应重写此方法。
- Parameters:
input (Input) – Runnable 的输入。
config (RunnableConfig | None) – 用于Runnable的配置。默认为None。
kwargs (Any | None) – 传递给Runnable的额外关键字参数。
- Yields:
Runnable 的输出。
- Return type:
AsyncIterator[Output]
- async astream_events(input: Any, config: RunnableConfig | None = None, *, version: Literal['v1', '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[StandardStreamEvent | CustomStreamEvent] #
生成事件流。
用于创建一个迭代器,遍历提供实时信息的StreamEvents,包括来自中间结果的StreamEvents。
StreamEvent 是一个具有以下模式的字典:
event
: str - 事件名称的格式为格式: on_[runnable_type]_(start|stream|end).
name
: str - 生成事件的 Runnable 的名称。run_id
: str - 与给定执行相关联的随机生成的ID发出事件的Runnable。 作为父Runnable执行的一部分被调用的子Runnable会被分配其自己唯一的ID。
parent_ids
: List[str] - 生成事件的父可运行对象的ID。根可运行对象将有一个空列表。 父ID的顺序是从根到直接父对象。 仅适用于API的v2版本。API的v1版本将返回一个空列表。
tags
: Optional[List[str]] - 生成事件的Runnable的标签事件。
metadata
: Optional[Dict[str, Any]] - Runnable的元数据生成事件的元数据。
data
: Dict[str, Any]
下表展示了一些可能由不同链发出的事件。为了简洁起见,表中省略了元数据字段。链定义已包含在表后。
注意 此参考表适用于V2版本的架构。
事件
名称
块
输入
输出
on_chat_model_start
[模型名称]
{“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
[模型名称]
‘Hello’
on_llm_end
[model name]
‘你好,人类!’
链上开始
格式化文档
on_chain_stream
format_docs
“你好世界!,再见世界!”
on_chain_end
format_docs
[Document(…)]
“你好世界!,再见世界!”
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
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)
- Parameters:
input (Any) – Runnable 的输入。
config (RunnableConfig | None) – 用于Runnable的配置。
version (Literal['v1', 'v2']) – 使用的模式版本,可以是 v2 或 v1。 用户应使用 v2。 v1 是为了向后兼容,将在 0.4.0 版本中弃用。 在 API 稳定之前不会分配默认值。 自定义事件仅在 v2 中显示。
include_names (Sequence[str] | None) – 仅包含来自具有匹配名称的可运行对象的事件。
include_types (Sequence[str] | None) – 仅包含来自具有匹配类型的可运行对象的事件。
include_tags (Sequence[str] | None) – 仅包含具有匹配标签的可运行对象的事件。
exclude_names (Sequence[str] | None) – 排除具有匹配名称的可运行对象的事件。
exclude_types (Sequence[str] | None) – 排除具有匹配类型的可运行对象的事件。
exclude_tags (Sequence[str] | None) – 排除具有匹配标签的可运行对象的事件。
kwargs (Any) – 传递给 Runnable 的额外关键字参数。 这些参数将传递给 astream_log,因为 astream_events 的实现是基于 astream_log 的。
- Yields:
一个异步的StreamEvents流。
- Raises:
NotImplementedError – 如果版本不是v1或v2。
- Return type:
AsyncIterator[StandardStreamEvent | CustomStreamEvent]
- batch(inputs: list[Input], config: RunnableConfig | list[RunnableConfig] | None = None, *, return_exceptions: bool = False, **kwargs: Any | None) list[Output] #
默认实现使用线程池执行器并行运行invoke。
batch的默认实现对于IO绑定的runnables效果很好。
如果子类能够更高效地进行批处理,则应重写此方法; 例如,如果底层的Runnable使用支持批处理模式的API。
- Parameters:
inputs (列表[Input])
config (RunnableConfig | list[RunnableConfig] | None)
return_exceptions (bool)
kwargs (任意 | 无)
- Return type:
列表[输出]
- 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,在它们完成时产生结果。
- Parameters:
inputs (Sequence[Input])
config (RunnableConfig | Sequence[RunnableConfig] | None)
return_exceptions (bool)
kwargs (任意 | 无)
- Return type:
Iterator[元组[int, Output | 异常]]
- bind(**kwargs: Any) Runnable[Input, Output] #
将参数绑定到Runnable,返回一个新的Runnable。
当链中的Runnable需要一个不在前一个Runnable输出中或用户输入中的参数时,这很有用。
- Parameters:
kwargs (Any) – 绑定到Runnable的参数。
- Returns:
一个新的Runnable,参数已绑定。
- Return type:
Runnable[Input, Output]
示例:
from langchain_community.chat_models import ChatOllama from langchain_core.output_parsers import StrOutputParser llm = ChatOllama(model='llama2') # Without bind. chain = ( llm | StrOutputParser() ) chain.invoke("Repeat quoted words exactly: 'One two three four five.'") # Output is 'One two three four five.' # With bind. chain = ( llm.bind(stop=["three"]) | StrOutputParser() ) chain.invoke("Repeat quoted words exactly: 'One two three four five.'") # Output is 'One two'
- configurable_alternatives(which: ConfigurableField, *, default_key: str = 'default', prefix_keys: bool = False, **kwargs: Runnable[Input, Output] | Callable[[], Runnable[Input, Output]]) RunnableSerializable #
配置可以在运行时设置的Runnables的替代方案。
- Parameters:
which (ConfigurableField) – 将用于选择替代方案的ConfigurableField实例。
default_key (str) – 如果没有选择其他选项,则使用的默认键。 默认为“default”。
prefix_keys (bool) – 是否在键前加上 ConfigurableField 的 id。 默认为 False。
**kwargs (Runnable[Input, Output] | Callable[[], Runnable[Input, Output]]) – 一个键值对字典,键对应的是Runnable实例或返回Runnable实例的可调用对象。
- Returns:
一个新的Runnable,配置了替代方案。
- Return type:
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: ConfigurableField | ConfigurableFieldSingleOption | ConfigurableFieldMultiOption) RunnableSerializable #
在运行时配置特定的Runnable字段。
- Parameters:
**kwargs (ConfigurableField | ConfigurableFieldSingleOption | ConfigurableFieldMultiOption) – 一个包含 ConfigurableField 实例的字典,用于配置。
- Returns:
一个新的Runnable,其字段已配置。
- Return type:
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 )
- extend(messages: Sequence[BaseMessagePromptTemplate | BaseMessage | BaseChatPromptTemplate | tuple[str | type, str | list[dict] | list[object]] | str]) None [源代码]#
使用一系列消息扩展聊天模板。
- Parameters:
messages (Sequence[BaseMessagePromptTemplate | BaseMessage | BaseChatPromptTemplate | tuple[str | type, str | list[dict] | list[object]] | str]) – 要追加的消息表示的序列。
- Return type:
无
- format(**kwargs: Any) str #
将聊天模板格式化为字符串。
- Parameters:
**kwargs (Any) – 用于填充此聊天模板中所有模板消息中的模板变量的关键字参数。
- Returns:
格式化字符串。
- Return type:
字符串
- format_messages(**kwargs: Any) list[BaseMessage] [source]#
将聊天模板格式化为最终消息列表。
- Parameters:
**kwargs (Any) – 用于填充此聊天模板中所有模板消息中的模板变量的关键字参数。
- Returns:
格式化消息列表。
- Return type:
列表[BaseMessage]
- format_prompt(**kwargs: Any) PromptValue #
格式化提示。应返回一个PromptValue。
- Parameters:
**kwargs (Any) – 用于格式化的关键字参数。
- Returns:
提示值。
- Return type:
- classmethod from_messages(messages: Sequence[BaseMessagePromptTemplate | BaseMessage | BaseChatPromptTemplate | tuple[str | type, str | list[dict] | list[object]] | str], template_format: Literal['f-string', 'mustache', 'jinja2'] = 'f-string') ChatPromptTemplate [source]#
从多种消息格式创建聊天提示模板。
示例
从消息模板列表实例化:
template = ChatPromptTemplate.from_messages([ ("human", "Hello, how are you?"), ("ai", "I'm doing well, thanks!"), ("human", "That's good to hear."), ])
从混合消息格式实例化:
template = ChatPromptTemplate.from_messages([ SystemMessage(content="hello"), ("human", "Hello, how are you?"), ])
- Parameters:
messages (Sequence[BaseMessagePromptTemplate | BaseMessage | BaseChatPromptTemplate | tuple[str | type, str | list[dict] | list[object]] | str]) – 消息表示的序列。 消息可以使用以下格式表示: (1) BaseMessagePromptTemplate, (2) BaseMessage, (3) 2元组 (消息类型, 模板); 例如, (“human”, “{user_input}”), (4) 2元组 (消息类, 模板), (5) 字符串,它是 (“human”, 模板) 的简写; 例如, “{user_input}”.
template_format (Literal['f-string', 'mustache', 'jinja2']) – 模板的格式。默认为“f-string”。
- Returns:
一个聊天提示模板。
- Return type:
- classmethod from_role_strings(string_messages: list[tuple[str, str]]) ChatPromptTemplate [source]#
自版本0.0.1起已弃用:请改用
from_messages classmethod()
。从(角色,模板)元组列表中创建一个聊天提示模板。
- Parameters:
string_messages (list[tuple[str, str]]) – (角色, 模板) 元组的列表。
- Returns:
一个聊天提示模板。
- Return type:
- classmethod from_strings(string_messages: list[tuple[type[BaseMessagePromptTemplate], str]]) ChatPromptTemplate [来源]#
自版本 0.0.1 起已弃用:请改用
from_messages classmethod()
。从(角色类,模板)元组列表中创建一个聊天提示模板。
- Parameters:
string_messages (list[tuple[type[BaseMessagePromptTemplate], str]]) – (角色类, 模板) 元组的列表。
- Returns:
一个聊天提示模板。
- Return type:
- classmethod from_template(template: str, **kwargs: Any) ChatPromptTemplate [source]#
从模板字符串创建一个聊天提示模板。
创建一个聊天模板,该模板由一条假设来自人类的消息组成。
- Parameters:
template (str) – 模板字符串
**kwargs (Any) – 传递给构造函数的关键字参数。
- Returns:
该类的一个新实例。
- Return type:
- invoke(input: dict, config: RunnableConfig | None = None, **kwargs: Any) PromptValue #
调用提示。
- Parameters:
input (dict) – 字典,提示的输入。
config (RunnableConfig | None) – RunnableConfig,用于配置提示的配置。
kwargs (Any)
- Returns:
提示的输出。
- Return type:
- partial(**kwargs: Any) ChatPromptTemplate [source]#
获取一个新的ChatPromptTemplate,其中一些输入变量已经填充。
- Parameters:
**kwargs (Any) – 用于填充模板变量的关键字参数。应该是输入变量的子集。
- Returns:
一个新的ChatPromptTemplate。
- Return type:
示例
from langchain_core.prompts import ChatPromptTemplate template = ChatPromptTemplate.from_messages( [ ("system", "You are an AI assistant named {name}."), ("human", "Hi I'm {user}"), ("ai", "Hi there, {user}, I'm {name}."), ("human", "{input}"), ] ) template2 = template.partial(user="Lucy", name="R2D2") template2.format_messages(input="hello")
- pretty_print() None #
打印一个人类可读的表示。
- Return type:
无
- pretty_repr(html: bool = False) str [source]#
人类可读的表示。
- Parameters:
html (bool) – 是否格式化为HTML。默认为False。
- Returns:
人类可读的表示。
- Return type:
字符串
- save(file_path: Path | str) None [source]#
将提示保存到文件。
- Parameters:
file_path (Path | str) – 文件路径。
- Return type:
无
- stream(input: Input, config: RunnableConfig | None = None, **kwargs: Any | None) Iterator[Output] #
流的默认实现,调用invoke。 如果子类支持流输出,则应重写此方法。
- Parameters:
input (Input) – Runnable 的输入。
config (RunnableConfig | None) – 用于Runnable的配置。默认为None。
kwargs (Any | None) – 传递给Runnable的额外关键字参数。
- Yields:
Runnable 的输出。
- Return type:
迭代器[输出]
- with_alisteners(*, on_start: AsyncListener | None = None, on_end: AsyncListener | None = None, on_error: AsyncListener | None = None) Runnable[Input, Output] #
将异步生命周期监听器绑定到一个Runnable,返回一个新的Runnable。
on_start: 在Runnable开始运行之前异步调用。 on_end: 在Runnable完成运行之后异步调用。 on_error: 如果Runnable抛出错误,则异步调用。
Run对象包含有关运行的信息,包括其id、类型、输入、输出、错误、开始时间、结束时间以及添加到运行中的任何标签或元数据。
- Parameters:
on_start (Optional[AsyncListener]) – 在Runnable开始运行之前异步调用。 默认为None。
on_end (Optional[AsyncListener]) – 在Runnable运行结束后异步调用。 默认为None。
on_error (可选[AsyncListener]) – 如果Runnable抛出错误,则异步调用。 默认为None。
- Returns:
一个新的Runnable,绑定了监听器。
- Return type:
Runnable[Input, Output]
示例:
from langchain_core.runnables import RunnableLambda import time 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 2024-05-16T14:20:29.637053+00:00 on start callback starts at 2024-05-16T14:20:29.637150+00:00 on start callback ends at 2024-05-16T14:20:32.638305+00:00 on start callback ends at 2024-05-16T14:20:32.638383+00:00 Runnable[3s]: starts at 2024-05-16T14:20:32.638849+00:00 Runnable[5s]: starts at 2024-05-16T14:20:32.638999+00:00 Runnable[3s]: ends at 2024-05-16T14:20:35.640016+00:00 on end callback starts at 2024-05-16T14:20:35.640534+00:00 Runnable[5s]: ends at 2024-05-16T14:20:37.640169+00:00 on end callback starts at 2024-05-16T14:20:37.640574+00:00 on end callback ends at 2024-05-16T14:20:37.640654+00:00 on end callback ends at 2024-05-16T14:20:39.641751+00:00
- with_config(config: RunnableConfig | None = None, **kwargs: Any) Runnable[Input, Output] #
将配置绑定到一个可运行对象,返回一个新的可运行对象。
- Parameters:
config (RunnableConfig | None) – 绑定到Runnable的配置。
kwargs (Any) – 传递给Runnable的额外关键字参数。
- Returns:
一个新的Runnable,带有绑定的配置。
- 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] #
为Runnable添加回退,返回一个新的Runnable。
新的Runnable将尝试原始的Runnable,然后在失败时依次尝试每个回退。
- Parameters:
fallbacks (Sequence[Runnable[Input, Output]]) – 如果原始 Runnable 失败,将尝试的一系列 Runnable。
exceptions_to_handle (tuple[type[BaseException], ...]) – 要处理的异常类型的元组。 默认为 (Exception,)。
exception_key (Optional[str]) – 如果指定了字符串,则处理的异常将作为输入的一部分传递给后备函数,使用指定的键。如果为 None,异常将不会传递给后备函数。如果使用此参数,基础 Runnable 及其后备函数必须接受字典作为输入。默认为 None。
- Returns:
一个新的Runnable,它将在失败时尝试原始的Runnable,然后依次尝试每个回退。
- Return type:
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
- Parameters:
fallbacks (Sequence[Runnable[Input, Output]]) – 如果原始 Runnable 失败,将尝试的一系列 Runnable。
exceptions_to_handle (tuple[type[BaseException], ...]) – 要处理的异常类型的元组。
exception_key (Optional[str]) – 如果指定了字符串,则处理的异常将作为输入的一部分传递给后备函数,使用指定的键。如果为 None,异常将不会传递给后备函数。如果使用此参数,基础 Runnable 及其后备函数必须接受字典作为输入。
- Returns:
一个新的Runnable,它将在失败时尝试原始的Runnable,然后依次尝试每个回退。
- Return type:
RunnableWithFallbacksT[Input, Output]
- 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。
on_start: 在Runnable开始运行之前调用,带有Run对象。 on_end: 在Runnable完成运行之后调用,带有Run对象。 on_error: 如果Runnable抛出错误时调用,带有Run对象。
Run对象包含有关运行的信息,包括其id、类型、输入、输出、错误、开始时间、结束时间以及添加到运行中的任何标签或元数据。
- Parameters:
on_start (Optional[Union[Callable[[Run], None], Callable[[Run, RunnableConfig], None]]]) – 在Runnable开始运行之前调用。默认为None。
on_end (可选[联合[可调用[[运行], 无], 可调用[[运行, RunnableConfig], 无]]]) – 在Runnable运行结束后调用。默认为无。
on_error (可选[联合[可调用[[运行], 无], 可调用[[运行, RunnableConfig], 无]]]) – 如果Runnable抛出错误时调用。默认为无。
- Returns:
一个新的Runnable,绑定了监听器。
- Return type:
Runnable[Input, Output]
示例:
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_retry(*, retry_if_exception_type: tuple[type[BaseException], ...] = (<class 'Exception'>,), wait_exponential_jitter: bool = True, stop_after_attempt: int = 3) Runnable[Input, Output] #
创建一个新的Runnable,在异常时重试原始的Runnable。
- Parameters:
retry_if_exception_type (tuple[type[BaseException], ...]) – 一个异常类型的元组,用于重试。 默认值为 (Exception,)。
wait_exponential_jitter (bool) – 是否在重试之间的等待时间中添加抖动。默认为 True。
stop_after_attempt (int) – 在放弃之前尝试的最大次数。默认为3。
- Returns:
一个新的Runnable,在异常时重试原始的Runnable。
- Return type:
Runnable[Input, Output]
示例:
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)
- Parameters:
retry_if_exception_type (tuple[type[BaseException], ...]) – 一个异常类型的元组,用于在发生这些异常时重试
wait_exponential_jitter (bool) – 是否在重试之间为等待时间添加抖动
stop_after_attempt (int) – 在放弃之前尝试的最大次数
- Returns:
一个新的Runnable,在异常时重试原始的Runnable。
- Return type:
Runnable[Input, Output]
- with_types(*, input_type: type[Input] | None = None, output_type: type[Output] | None = None) Runnable[Input, Output] #
将输入和输出类型绑定到一个Runnable,返回一个新的Runnable。
- Parameters:
input_type (type[Input] | None) – 要绑定到Runnable的输入类型。默认为None。
output_type (type[Output] | None) – 要绑定到Runnable的输出类型。默认为None。
- Returns:
一个带有类型绑定的新Runnable。
- Return type:
Runnable[Input, Output]
使用 ChatPromptTemplate 的示例