尽管单个链条对独立任务有效,但大多数应用都涉及一系列操作。例如,要生成一篇技术博客文章,你可能需要先创建一个大纲,然后根据大纲撰写引言,最后生成社交媒体帖子来宣传。每一步都依赖于前一步的输出。LangChain允许你使用LangChain表达式语言 (LCEL) 将这些操作按顺序组合成管道。线性工作流连接操作最直接的方式是将其组合成一个线性序列,其中每一步都将其输出传递给下一步。考虑一个两步过程:生成标题: 给定一个主题,一个LLM为一部戏剧创建一个吸引人的标题。撰写剧情梗概: 给定戏剧标题,另一个LLM撰写一个简短的剧情梗概。数据流向为直线:主题 -> 标题 -> 剧情梗概。digraph G { rankdir=TB; graph [fontname="Arial", splines=ortho]; node [shape=box, style="rounded,filled", fontname="Arial", fillcolor="#a5d8ff", color="#4263eb"]; edge [fontname="Arial"]; input [label="输入\n(主题: '大萧条')", fillcolor="#e9ecef", color="#868e96"]; chain1 [label="链条1: 标题生成", shape=component, fillcolor="#bac8ff"]; intermediate [label="中间输出\n(标题)", shape=ellipse, style=dashed, color="#adb5bd"]; chain2 [label="链条2: 剧情梗概生成", shape=component, fillcolor="#bac8ff"]; output [label="最终输出\n(剧情梗概)", fillcolor="#e9ecef", color="#868e96"]; input -> chain1; chain1 -> intermediate [label="标题"]; intermediate -> chain2; chain2 -> output; }数据以线性序列流动。每个组件都将其输出提供给下一个。让我们来实现这一点。我们将使用提示词和一个聊天模型来设置两个链条。我们使用 StrOutputParser 来确保模型的输出是一个干净的字符串,以便进行下一步。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 假设 OPENAI_API_KEY 已在你的环境中设置 llm = ChatOpenAI(temperature=0.7, model_name="gpt-3.5-turbo") # 链条1: 为一部戏剧生成标题 prompt_title = ChatPromptTemplate.from_template( "你是一位剧作家。给定主题'{topic}',为一部戏剧撰写一个吸引人的标题。" ) chain_title = prompt_title | llm | StrOutputParser() # 链条2: 为这部戏剧生成剧情梗概 prompt_synopsis = ChatPromptTemplate.from_template( "你是一位剧评人。给定戏剧标题'{title}',撰写一个简短的、一段的剧情梗概。" ) chain_synopsis = prompt_synopsis | llm | StrOutputParser()为了连接这些,我们需要确保 chain_title 的输出与 chain_synopsis 的预期输入匹配。由于 chain_synopsis 预期一个带有 title 键的字典,而 chain_title 返回一个字符串,我们使用 RunnablePassthrough 通过一个字典明确地映射输出。# 使用管道操作符创建序列 overall_chain = ( chain_title | {"title": RunnablePassthrough()} | chain_synopsis ) # 使用初始主题运行链条 topic = "一个20世纪20年代爵士乐音乐家的兴衰" final_synopsis = overall_chain.invoke({"topic": topic}) print(final_synopsis)当你运行这个链条时,LangChain会执行 chain_title,获取结果字符串,将其映射到 title 键,并将其传递给 chain_synopsis。严格线性流的主要限制是,中间数据通常会丢失或无法供后续步骤使用,除非明确地传递。使用 RunnablePassthrough 管理状态对于更复杂的工作流,你通常需要在多个步骤中保持状态。这在以下情况中是必要的:早期链条产生多个后续链条所需的输出。后续链条需要访问中间输出 和 原始输入。RunnablePassthrough.assign() 方法通过向流经链条的字典添加新值来管理此过程,而不覆盖现有数据。它将工作流视为一个累积状态。让我们修改之前的例子。假设剧情梗概链条需要同时知道生成的 标题 和原始 主题,以添加更多上下文。digraph G { rankdir=TB; graph [fontname="Arial", splines=ortho]; node [shape=box, style="rounded,filled", fontname="Arial", fillcolor="#d0bfff", color="#7048e8"]; edge [fontname="Arial"]; subgraph cluster_state { label="链条状态"; style="dashed"; bgcolor="#f8f9fa"; state_topic [label="主题", shape=note, fillcolor="#ffec99", color="#f59f00"]; state_title [label="标题", shape=note, fillcolor="#ffec99", color="#f59f00"]; } chain1 [label="链条1: 标题生成", shape=component]; chain2 [label="链条2: 剧情梗概生成", shape=component]; chain1 -> state_title [label="添加"]; state_topic -> chain1 [label="读取"]; state_topic -> chain2 [label="读取"]; state_title -> chain2 [label="读取"]; }带有状态管理的数据流。assign 将输出添加到状态中,使其可用于后续步骤。为此,我们串联 assign 调用。每一步都会计算一个值并将其附加到状态。# 设置LLM和提示词 llm = ChatOpenAI(temperature=0.7, model_name="gpt-3.5-turbo") # 链条1定义: 仅获取标题的逻辑 prompt_title = ChatPromptTemplate.from_template( "你是一位剧作家。给定主题'{topic}',为一部戏剧撰写一个吸引人的标题。" ) chain_title = prompt_title | llm | StrOutputParser() # 链条2定义: 预期输入'title'和'topic' prompt_synopsis = ChatPromptTemplate.from_template( "为一部名为'{title}'、关于'{topic}'的戏剧撰写一个简短的一段式剧情梗概。" ) chain_synopsis = prompt_synopsis | llm | StrOutputParser()现在我们构建管道。我们使用 assign 捕获每个阶段的输出。# 创建带有状态管理的链条 overall_chain = ( # 步骤1: 计算标题并将其添加到数据流中。 # 数据流现在包含 {'topic': ..., 'title': ...} RunnablePassthrough.assign(title=chain_title) | # 步骤2: 计算剧情梗概。 # 它可以访问数据流中的'topic'和'title'。 RunnablePassthrough.assign(synopsis=chain_synopsis) ) # 运行链条 input_data = {"topic": "一个20世纪20年代爵士乐音乐家的兴衰"} result = overall_chain.invoke(input_data) print(result)此调用的输出是一个包含原始输入和所有已赋值变量的字典。{ "topic": "一个20世纪20年代爵士乐音乐家的兴衰", "title": "蓝调尾声", "synopsis": "以20世纪20年代充满活力、混乱的“咆哮的二十年代”为背景,《蓝调尾声》记录了萨克斯手利奥“国王”克里奥尔的迅速崛起和令人心碎的衰落……" }通过使用 RunnablePassthrough.assign,你能够对数据管道进行精细控制。这种结构允许你保存和重用序列中任何位置的信息,这对于精密的多步骤推理过程来说必不可少。