如何链接可运行对象

先决条件

本指南假定读者熟悉以下概念：

关于LangChain表达式语言的一点是，任意两个可运行对象都可以“链接”在一起形成序列。前一个可运行对象的.invoke()调用的输出将作为输入传递给下一个可运行对象。这可以通过管道操作符(|)实现，也可以使用更明确的.pipe()方法，两者功能相同。

生成的 RunnableSequence 本身就是一个可运行对象，这意味着它可以像其他可运行对象一样被调用、流式处理或进一步链式组合。以这种方式链式组合可运行对象的优势在于高效的流式传输（一旦输出可用，序列就会立即开始流式输出），以及借助 LangSmith 等工具进行调试和追踪。

管道操作符：`|`

为了展示其工作原理，我们来看一个例子。我们将演示LangChain中一种常见模式：使用提示模板将输入格式化为聊天模型，最后通过输出解析器将聊天消息输出转换为字符串。

选择聊天模型:

pip install -qU "langchain[openai]"

import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain.chat_models import init_chat_model

model = init_chat_model("gpt-4o-mini", model_provider="openai")

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")

chain = prompt | model | StrOutputParser()

API 参考：StrOutputParser | ChatPromptTemplate

提示词和模型都是可运行的，且提示词调用的输出类型与聊天模型的输入类型相同，因此我们可以将它们串联起来。随后，我们可以像调用其他可运行对象一样调用生成的序列：

chain.invoke({"topic": "bears"})

"Here's a bear joke for you:\n\nWhy did the bear dissolve in water?\nBecause it was a polar bear!"

强制

我们甚至可以将此链与其他可运行对象结合，创建另一个链。这可能需要使用其他类型的可运行对象进行输入/输出格式化，具体取决于链组件所需的输入和输出。

例如，假设我们想要将生成笑话的链与另一个链组合起来，以评估生成的笑话是否有趣。

在将输入格式化到下一个链时，我们需要格外小心。在下面的示例中，链中的字典会自动解析并转换为 RunnableParallel，它会并行运行所有值，并返回包含结果的字典。

这恰好是下一个提示模板所期望的格式。以下是其实际应用：

from langchain_core.output_parsers import StrOutputParser

analysis_prompt = ChatPromptTemplate.from_template("is this a funny joke? {joke}")

composed_chain = {"joke": chain} | analysis_prompt | model | StrOutputParser()

composed_chain.invoke({"topic": "bears"})

API 参考：StrOutputParser

'Haha, that\'s a clever play on words! Using "polar" to imply the bear dissolved or became polar/polarized when put in water. Not the most hilarious joke ever, but it has a cute, groan-worthy pun that makes it mildly amusing. I appreciate a good pun or wordplay joke.'

函数也将被强制转换为可运行对象，因此你也可以向你的链中添加自定义逻辑。下面的链产生了与之前相同的逻辑流程：

composed_chain_with_lambda = (
    chain
    | (lambda input: {"joke": input})
    | analysis_prompt
    | model
    | StrOutputParser()
)

composed_chain_with_lambda.invoke({"topic": "beets"})

"Haha, that's a cute and punny joke! I like how it plays on the idea of beets blushing or turning red like someone blushing. Food puns can be quite amusing. While not a total knee-slapper, it's a light-hearted, groan-worthy dad joke that would make me chuckle and shake my head. Simple vegetable humor!"

但是，请记住，使用此类函数可能会干扰流式传输等操作。详细了解，请参阅本节。

`.pipe()` 方法

我们也可以使用 .pipe() 方法来组合相同的序列。这是它的样子：

from langchain_core.runnables import RunnableParallel

composed_chain_with_pipe = (
    RunnableParallel({"joke": chain})
    .pipe(analysis_prompt)
    .pipe(model)
    .pipe(StrOutputParser())
)

composed_chain_with_pipe.invoke({"topic": "battlestar galactica"})

API 参考：RunnableParallel

"I cannot reproduce any copyrighted material verbatim, but I can try to analyze the humor in the joke you provided without quoting it directly.\n\nThe joke plays on the idea that the Cylon raiders, who are the antagonists in the Battlestar Galactica universe, failed to locate the human survivors after attacking their home planets (the Twelve Colonies) due to using an outdated and poorly performing operating system (Windows Vista) for their targeting systems.\n\nThe humor stems from the juxtaposition of a futuristic science fiction setting with a relatable real-world frustration – the use of buggy, slow, or unreliable software or technology. It pokes fun at the perceived inadequacies of Windows Vista, which was widely criticized for its performance issues and other problems when it was released.\n\nBy attributing the Cylons' failure to locate the humans to their use of Vista, the joke creates an amusing and unexpected connection between a fictional advanced race of robots and a familiar technological annoyance experienced by many people in the real world.\n\nOverall, the joke relies on incongruity and relatability to generate humor, but without reproducing any copyrighted material directly."

或者简写为：

composed_chain_with_pipe = RunnableParallel({"joke": chain}).pipe(
    analysis_prompt, model, StrOutputParser()
)

流式传输: 查看流式传输指南，了解链的流式传输行为

管道操作符：|​

强制​

.pipe() 方法​

相关​

管道操作符：`|`

强制

`.pipe()` 方法

相关