使用LCEL构建LangChain流程：线性、并行及分支结构。
学习如何使用pipe操作符、RunnableParallel和RunnableBranch，将提示词、模型及解析器整合为线性、多阶段、并行及条件化的LangChain流程。
来源: Chains in LangChain: Simple, Sequential, Parallel, and Conditional — Part 7
真正的大型语言模型功能几乎从不会仅停留在一次模型调用上。你需要先格式化提示词，再调用模型，处理返回的结果，将其作为新的提示词输入，有时还需同时运行多个请求，偶尔还会根据输入内容选择不同的处理路径。如果手动为每一步过渡编写代码，很快就会形成难以阅读且更难修改的冗长代码。本指南将介绍 LangChain 中基于 LangChain 表达语言（LCEL）构建的四种组合模式：直接流水线、多阶段流水线、使用 RunnableParallel 的分支处理，以及使用 RunnableBranch 的路由机制。每种模式都会在 Google Gemini 和 OpenAI 上各展示一次，这样你就能清楚地看到更换服务提供商时哪些部分会发生变化，哪些部分保持不变。
链式结构的真正含义
链是一种由 LangChain 组件构成的流水线，其中每个步骤都会接收前一步产生的输出。常见的构成要素包括提示词模板、聊天或完成模型以及输出解析器，但任何实现了 LangChain 的 Runnable 接口的组件都可以参与其中。最简单的实用流水线结构如下：
Prompt → LLM → Output Parser → Output
每个阶段都有其特定任务。提示词模板会接收输入变量并将其转换为模型可理解的指令。模型读取这些指令后生成响应，而输出解析器则将该响应转换为应用程序所需的形式，例如普通字符串或经过验证的对象。最终从解析器输出的内容就是整个流水线的结果。
抽象的价值在于无需再手动在各个阶段之间传输数据。只需声明一次顺序，LangChain就会负责为每个组件提供正确的输入。从概念上讲，这就好比将小块组件拼接成一个更大的整体：
Input
↓
Component 1
↓
Component 2
↓
Component 3
↓
Output
对于只有两步的脚本来说这点影响不大，但当应用程序发展到十几步时就变得至关重要。通过声明流程图，数据流可以集中显示在一个地方，而不会分散在各个辅助函数中。其思维模型就是不断循环：输入、处理、输出，然后再对输出结果进行处理，直到最终得到结果。
为何组合式设计优于手动连接
当每个中间值都需要显式传递时，代码会变得杂乱且脆弱。链式结构解决了三个实际问题。
无需粘合代码即可实现数据流
其核心优势在于，某一步的输出会自动成为下一步的输入。再以我们熟悉的流程为例：
Prompt → LLM → Output Parser → Output
如果手动实现，需要先格式化提示语，再用该格式化的值调用模型，从响应对象中提取文本并传递给解析器，每一步都需要一个变量来存储中间结果。而使用LCEL时，整个流程可以通过管道运算符整合为单个表达式：
chain = prompt | llm | parser
|运算符之所以能正常工作，是因为提示语、模型和解析器都实现了相同的Runnable接口，因此每个组件都知道如何接收前一个组件的输出。临时变量越少，出现错误传递的可能性就越低，而且整个处理流程会按照从左到右的顺序依次执行。
多步骤逻辑依然易于理解
在内容生成中，通常需要使用不同的提示词进行多次模型调用。常见的流程是先根据某个主题生成内容，再通过第二个提示词对中间结果进行后处理：
Topic
↓
Prompt 1
↓
LLM
↓
Output Parser
↓
Prompt 2
↓
LLM
↓
Output Parser
↓
Final Output
如果手动管理这种流程，随着步骤的增加会变得愈发复杂。而以链式结构表示时，各个阶段都包含在同一个声明中，且每个阶段会将结果传递给下一个阶段的规则由框架自动保障，无需额外约束。
组件可互换
由于每个阶段都是独立的组件，因此可以更换其中一个而不影响其他部分。假设当前的流程是在 Gemini 上运行的：
chain = prompt | gemini | parser
若要改用 OpenAI，只需更换模型组件即可：
chain = prompt | openai | parser
提示词和解析器保持不变。当需求发生变化、团队需要比较不同提供商的成本或质量，或者希望在开发阶段使用成本更低的模型而在生产环境使用性能更强的模型时，这种方法非常有用。
优势一览
- 更少的连接代码：输出会自动进入下一步流程。
- 结构化的多步骤工作流：多个提示词、模型和解析器都存在于同一个定义好的流程中。
- 更易于扩展：包括模型提供商在内的各个组件都可以被重复使用或替换，而无需重新设计整个流程。
四种组合模式
将 LangChain 流水线分为四种类型有助于理解其结构。严格来说，这些在现代 LangChain 中并非独立的类别，而是由相同的可执行基础组件构建而成的模式。这些名称仍可作为实用的术语：
- 简单链：从提示到解析器的单一直线流程。
- 顺序链：多条此类流程首尾相连。
- 并行链：多个独立的子链同时处理同一输入。
- 条件链：包含决策点，用于选择由哪个子链处理输入。
这四种类型涵盖了从一次性提示处理到需要分支决策的工作流等各种场景。下文将通过两个提供者的实际示例对每种类型进行说明。
简单链：单一流程
最简单的流程是将各个组件以直线方式连接起来。每个组件的输出就是下一个组件的输入，整个流程中没有分支或决策点：
Input
↓
Prompt Template
↓
LLM
↓
Output Parser
↓
Output
这个示例是一个学习辅助工具：给定一个主题，模型应输出一些易于理解的事实，而不需要深入复杂的技术细节。实现这一目标的流程尽可能简短：
prompt → llm → parser
用文字概括就是，topic值被输入到提示模板中，格式化后的提示发送给模型，模型的回复在返回之前会经过输出解析器。从技术上讲，解析器是可选的，但加上它可以让返回结果类型更加明确。
具体来说，该示例以板球作为主题，并要求提供五个有趣的事实。数据流动过程如下：
{'topic': 'cricket'}
↓
PromptTemplate
↓
LLM
↓
StrOutputParser
↓
Five interesting facts about cricket
Gemini版本
以下是通过 langchain_google_genai 集成使用 Google Gemini 的完整程序：
from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv
load_dotenv()
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
prompt = PromptTemplate(
template='Generate 5 interesting facts about {topic}',
input_variables=['topic']
)
parser = StrOutputParser()
chain = prompt | llm | parser
result = chain.invoke({'topic':'cricket'})
# print(result)
chain.get_graph().print_ascii()
该程序基于 gemini-2.5-flash 构建 GoogleGenerativeAI 模型，使用单变量提示词，并结合 StrOutputParser，再通过 prompt | llm | parser 的顺序将它们组合起来。最后一行会输出该处理链的 ASCII 图表，相关内容将在后文详细说明。注意，此版本中的 print(result) 行已被注释掉，因此除非重新启用，否则脚本仅会显示图表结构。
逐行解析 Gemini 代码
导入语句调用了脚本所需的四个组件：
from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv
GoogleGenerativeAI 是 Gemini 模型的封装类。PromptTemplate 定义了包含 {topic} 插槽的提示语，该插槽会在运行时被填充内容。StrOutputParser 负责将模型输出转换为普通的 Python 字符串。load_dotenv 用于读取本地的 .env 文件并将其内容设置为环境变量。
在创建任何客户端之前，调用 load_dotenv 可以确保 API 密钥以及 .env 文件中的其他设置能够被进程使用：
load_dotenv()
Gemini 集成会从环境变量中读取密钥，因此必须在模型实例化之前执行此操作。通常的做法是将密钥保存在 .env 文件中且不将其纳入版本控制。
随后会创建模型本身，指定其版本为 gemini-2.5-flash：
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
提示模板同时声明了文本及其所需的变量：
prompt = PromptTemplate(
template='Generate 5 interesting facts about {topic}',
input_variables=['topic']
)
在调用时，{topic}占位符会从输入字典中获取内容。传入的输入为：
{'topic': 'cricket'}
由此生成具体的提示内容：
Generate 5 interesting facts about cricket
解析器是在没有参数的情况下创建的：
parser = StrOutputParser()
它的任务是将模型的回答以字符串形式返回。对于像GoogleGenerativeAI这样已经直接返回文本的模型，其功能几乎相当于直接传递数据；而对于返回消息对象的聊天型模型，则需要提取其中的文本内容。无论哪种情况，后续接收方得到的都是普通的str类型数据。
接下来是定义整个处理流程的代码行：
chain = prompt | llm | parser
管道运算符将这三个组件连接起来，使得格式化后的提示语被输入模型，而模型的回复则被传递给解析器：
PromptTemplate → Gemini 2.5 Flash → StrOutputParser
最后，使用包含主题信息的字典来运行这个流程：
result = chain.invoke({'topic':'cricket'})
这单一次调用会按顺序触发各个阶段，代码中无需任何中间变量：
cricket
↓
PromptTemplate
↓
Gemini 2.5 Flash
↓
StrOutputParser
↓
Final Result
返回结果
返回值是一个字符串，其中包含关于板球的五条事实，内容大致如下：
1. Cricket originated in England and has been played for several centuries.
2. A cricket team consists of 11 players.
3. Test cricket matches can last up to five days.
4. The Cricket World Cup is one of the major international cricket tournaments.
5. Cricket is especially popular in countries such as India, Australia, England, and New Zeakand.
由于文本是动态生成的，其用词和具体事实每次运行都会不同，示例输出中甚至还存在拼写错误（“Zeakand”）。应将示例输出视为结构上的参考，而非固定结果。如果需要确定性的格式，那应该由更严格的解析器或结构化输出来实现，而非依赖模型重复相同输出。
在 OpenAI 上的相同流程
现在是在 OpenAI 上运行的完全相同的辅助工具。值得注意的是，其变化极为有限：
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv
load_dotenv()
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = PromptTemplate(
template='Generate 5 interesting facts about {topic}',
input_variables=['topic']
)
parser = StrOutputParser()
chain = prompt | llm | parser
result = chain.invoke({'topic':'cricket'})
print(result)
chain.get_graph().print_ascii()
现在的模型是 ChatOpenAI，基于 gpt-4o-mini。提示词、解析器以及组合方式都与之前完全一致，该版本还会输出结果及图表。
唯一的新元素是模型对象：
llm = ChatOpenAI(model="gpt-4o-mini")
langchain_openai 中的 ChatOpenAI 替代了原来的 GoogleGenerativeAI。它是一个聊天模型，因此返回的是消息对象而非原始字符串，这也正是保留链中解析器的重要原因。
提示词模板没有变化：
prompt = PromptTemplate(
template='Generate 5 interesting facts about {topic}',
input_variables=['topic']
)
解析器也没有变化：
parser = StrOutputParser()
组合方式同样没有变化：
chain = prompt | llm | parser
结构仍是三阶段流程，中间为OpenAI：
PromptTemplate → OpenAI → StrOutputParser
调用该链并打印结果的形式如下：
result = chain.invoke({'topic':'cricket'})
print(result)
OpenAI运行还会返回一条包含五个板球相关事实的字符串，例如：
1. Cricket is believed to have originated in England.
2. Each cricket team has 11 players.
3. A Test match can be played over a maximum of five days.
4. The Cricket World Cup is one of the most important international cricket tournaments.
5. Cricket has a particularly large following in India and several other countries.
同样，不同运行次数及不同模型下的表述会有所差异。
打印链的结构
两个脚本都以将处理流程以文本图表形式呈现的调用作为结尾：
chain.get_graph().print_ascii()
get_graph()会返回Runnable的图形表示，而print_ascii()则会在终端中将其绘制出来。对于Gemini链，输出效果大致如下：
PromptInput
*
*
PromptTemplate
*
*
GoogleGenerativeAI
*
*
StrOutputParser
*
*
StrOutputParserOutput
PromptInput和StrOutputParserOutput节点是源自架构的输入点与输出点；它们之间的三个节点即为你的组件。这是一种快速验证方式，用于确认节点连接关系是否符合预期。在大多数环境中，print_ascii()函数依赖于可选的grandalf包，如果调用时出现导入错误，请安装该包。
仅模型有所变化
将两个流程并排放置，差异仅在于一个节点。使用Gemini时：
Input
↓
PromptTemplate
↓
GoogleGenerativeAI
↓
StrOutputParser
↓
Output
使用OpenAI时：
Input
↓
PromptTemplate
↓
ChatOpenAI
↓
StrOutputParser
↓
Output
从代码层面来看，所有差异都取决于你构建的是哪个模型对象：
# Google Gemini
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
# OpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
其周围的各项都可以保持不变。这正是基于标准化组件进行开发的实际优势：提供商的选择变成了配置决策，而非需要重新编写代码。实际上仍有一些细微差别需要注意，比如聊天模型与完成模型、参数名称，以及各提供商处理系统指令的方式，但整个流程本身是可移植的。
完整的数据路径
大致来说，一个简单的流程链如下：
Input
↓
PromptTemplate
↓
LLM
↓
StrOutputParser
↓
Output
通过板球相关的输入进行追踪后，从字典到生成的提示语、原始模型响应以及最终字符串，每一个中间值都清晰可见：
{'topic': 'cricket'}
↓
PromptTemplate
↓
"Generate 5 interesting facts about cricket"
↓
LLM
↓
Raw LLM Response
↓
StrOutputParser
↓
Final String Output
这就是整个思路。每个组件负责完成一项任务，其输出便成为下一个组件的输入，如此依次传递，形成一条直线流程。
顺序式链：将一个模型调用结果传递给下一个
顺序链将同一理念贯穿多个阶段。一个简单的链通常只包含一个此类结构：
Prompt → LLM → Output Parser
顺序链可以包含多个提示模板、多次模型调用以及多个解析器，它们都被连接在一起。核心规则始终不变：当前阶段产生的内容即为下一阶段所使用的输入。仍然不存在分支结构，各阶段始终按固定顺序执行。两阶段的示例如下：
Input
↓
Prompt Template 1
↓
LLM 1
↓
Output Parser
↓
Prompt Template 2
↓
LLM 2
↓
Output Parser
↓
Output
示例：先生成详细报告，再生成摘要
该示例任务分为两个阶段。首先，针对某个主题生成一份详细报告；其次，将这份报告浓缩为五点摘要。第二阶段永远不会直接接触原始主题，它仅依据第一阶段生成的报告来工作：
Input
↓
Explanation Prompt
↓
LLM
↓
StrOutputParser
↓
Detailed Report
↓
Summary Prompt
↓
LLM
↓
StrOutputParser
↓
5-Point Summary
有两个提示模板负责处理任务。explaination_prompt（代码中存在拼写错误，复制代码片段时请保留原样）用于请求详细的报告。summary_prompt则以该报告为输入，要求提取出五个总结要点。
Gemini版本
以下是在Gemini上运行的完整程序：
from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv
load_dotenv()
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
# Prompt 1 : Explanation
explaination_prompt = PromptTemplate(
template='Generate a detailed report on {topic}',
input_variables=['topic']
)
# Prompt 2 : Summary
summary_prompt = PromptTemplate(
template='Generate a 5 pointer summary from the following text \n {text}',
input_variables=['text']
)
parser = StrOutputParser()
chain = explaination_prompt | llm | parser | summary_prompt | llm | parser
result = chain.invoke({"topic": "Generative AI"})
# print(result)
chain.get_graph().print_ascii()
一个GoogleGenerativeAI实例负责处理这两个阶段，整个流程由六个组件构成的单一线性表达式实现：两个提示模板、两次模型调用以及两次解析器调用。
逐行分析Gemini代码
导入的模块与简单的流程结构相对应：
from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv
GoogleGenerativeAI用于模型操作，PromptTemplate用于两个可重复使用的提示模板，StrOutputParser用于将每个响应转换为字符串，而load_dotenv则用于加载配置文件。
环境变量会像之前一样首先被加载：
load_dotenv()
模型仅创建一次：
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
在两个阶段中重复使用同一个模型对象是完全没问题的。LangChain中的模型封装体对于单次调用而言是无状态的，因此可以在链中多次引用同一个实例。
第一个提示词会接收一个topic参数，并要求生成详细的报告：
explaination_prompt = PromptTemplate(
template='Generate a detailed report on {topic}',
input_variables=['topic']
)
给定如下输入：
{"topic": "Generative AI"}
其输出结果为：
Generate a detailed report on Generative AI
有趣的任务交接发生在第二个提示词阶段：
summary_prompt = PromptTemplate(
template='Generate a 5 pointer summary from the following text \n {text}',
input_variables=['text']
)
它期望有一个名为text的变量，但输入字典中并没有名为text的键。该值来自第一阶段：即第一次模型调用生成的报告会填充这个位置。
这得益于位于各阶段之间的解析器：
parser = StrOutputParser()
在首次调用模型后，解析器会生成一个普通字符串。当带有恰好一个输入变量的PromptTemplate接收到的是纯字符串而非字典时，LCEL会将该字符串映射到这个单一变量上。因此无需任何显式映射，报告就会以{text}的形式呈现。如果第二个提示包含两个变量，则需要自行构建字典，例如使用一个包含可执行对象的小型字典或RunnablePassthrough.assign方法。
整个处理流程只需一行代码：
chain = explaination_prompt | llm | parser | summary_prompt | llm | parser
从左到右阅读即可了解执行顺序：
Explanation Prompt
↓
Gemini
↓
StrOutputParser
↓
Summary Prompt
↓
Gemini
↓
StrOutputParser
↓
Final Output
每个组件都会接收前一个组件的输出。您的代码中没有任何部分会修改中间报告。
运行该流程只需一个主题：
result = chain.invoke({"topic": "Generative AI"})
在那个调用之后，两个阶段会依次执行：
"Generative AI"
↓
Explanation Prompt
↓
Gemini
↓
StrOutputParser
↓
Detailed Report
↓
Summary Prompt
↓
Gemini
↓
StrOutputParser
↓
Final 5-Point Summary
返回结果
最终得到的值是对生成式AI的五点总结，例如：
1. Generative AI is a branch of artificial intelligence that creates new content.
2. It can generate text, images, audio, video, and code.
3. Foundation models are an important part of modern Generative AI systems.
4. Generative AI is used across areas such as education, software development, and customer support.
5. It is enabling applications that can automate and simplify many creative and knowledge-based tasks.
需要注意的是，这个总结并非直接根据主题内容生成的。Gemini首先会撰写一份完整报告，只有这份报告才会被发送给总结提示词。这种两步法虽然能让第二次调用有更丰富的内容可供压缩，但代价是需要额外的模型调用以及随之产生的延迟和token消耗。与以往一样，具体表述会因每次运行而有所不同。
在OpenAI上的相同流程
OpenAI版本保持了完整的工作流程：
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv
load_dotenv()
llm = ChatOpenAI()
explaination_prompt = PromptTemplate(
template="Generate a detailed report on {topic}",
input_variables=["topic"]
)
summary_prompt = PromptTemplate(
template="Generate a 5 pointer summary from the following text \n {text}",
input_variables=["text"],
)
parser = StrOutputParser()
chain = explaination_prompt | llm | parser | summary_prompt | llm | parser
result = chain.invoke({"topic": "Generative AI"})
print(result)
chain.get_graph().print_ascii()
仍然是两阶段结构：报告提示词、模型、解析器，总结提示词、模型、解析器。该脚本还两次导入了load_dotenv，虽然不会造成问题，但可以优化。
这次构建模型时没有传入任何参数：
llm = ChatOpenAI()
由于没有提供model参数，ChatOpenAI会使用库的默认模型，而这个默认模型在langchain_openai的不同版本中会有所变化。在实际项目中应明确指定模型名称，以避免升级后导致功能或成本方面的隐性变化。
报告提示词与Gemini版本完全相同：
explaination_prompt = PromptTemplate(
template="Generate a detailed report on {topic}",
input_variables=["topic"]
)
总结提示词也是如此，它同样要求将报告内容以{text}格式呈现：
summary_prompt = PromptTemplate(
template="Generate a 5 pointer summary from the following text \n {text}",
input_variables=["text"],
)
这就是顺序处理模式的优势所在：第二个提示词会自动根据第一阶段的输出来生成。
解析器仍是相同的 StrOutputParser：
parser = StrOutputParser()
且字符级的组合方式也完全一致：
chain = explaination_prompt | llm | parser | summary_prompt | llm | parser
从流程上看，其读取方式如下：
Prompt 1
↓
OpenAI
↓
StrOutputParser
↓
Prompt 2
↓
OpenAI
↓
StrOutputParser
↓
Final Output
只有模型节点与 Gemini 版本不同。调用时会再次传递主题参数。注意，所示代码片段缺少结束括号；完整的程序中该行是正确的：
result = chain.invoke({"topic": "Generative AI"}
随后该链会分别独立运行这两个阶段。
返回结果
结果仍是基于中间报告生成的五点总结：
1. Generative AI creates new content by learning patterns from existing data.
2. It is powered by large foundation models trained on diverse datasets.
3. Generative AI can produce text, images, audio, video, and code.
4. It is being used in many industries and applications.
5. It is becoming an important part of modern AI-powered software.
不同运行次数的细节会有所差异。
打印链的结构
该图结构辅助工具也能处理更长的链：
chain.get_graph().print_ascii()
对于 Gemini 顺序链，图表显示两个阶段是垂直堆叠的：
PromptInput
*
*
PromptTemplate
*
*
GoogleGenerativeAI
*
*
StrOutputParser
*
*
PromptTemplate
*
*
GoogleGenerativeAI
*
*
StrOutputParser
*
*
StrOutputParserOutput
由于在处理流程中的两个位置使用了相同的模型对象，因此模型节点出现了两次，而图表反映的是位置而非唯一的对象。
两种实现的比较
两个版本都遵循相同的顺序结构；唯一的实际差异在于提供方式。Gemini处理流程为：
Input
↓
Explanation Prompt
↓
GoogleGenerativeAI
↓
StrOutputParser
↓
Summary Prompt
↓
GoogleGenerativeAI
↓
StrOutputParser
↓
Output
OpenAI处理流程为：
Input
↓
Explanation Prompt
↓
ChatOpenAI
↓
StrOutputParser
↓
Summary Prompt
↓
ChatOpenAI
↓
StrOutputParser
↓
Output
提示词、解析器以及链式结构都是共享的。唯一会发生变化的部分是模型的构建方式：
# Google Gemini
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
# OpenAI
llm = ChatOpenAI()
在不同提供方式下，组合后的链式结构本身保持不变：
chain = explaination_prompt | llm | parser | summary_prompt | llm | parser
抽象地说，顺序式链式结构看起来是这样的：
Prompt 1
↓
LLM 1
↓
Parser
↓
Prompt 2
↓
LLM 2
↓
Parser
↓
Output
它与简单链式结构的区别在于存在多个阶段，且从第二个阶段开始，每个阶段都依赖于前一个阶段。
完整的数据路径
一般来说，顺序链处理数据的方式如下：
Input
↓
PromptTemplate 1
↓
LLM 1
↓
StrOutputParser
↓
Detailed Report
↓
PromptTemplate 2
↓
LLM 2
↓
StrOutputParser
↓
Final Output
结合生成式 AI 的输入以及生成的提示语进行追踪：
{"topic": "Generative AI"}
↓
Explanation Prompt
↓
"Generate a detailed
report on Generative AI"
↓
LLM
↓
StrOutputParser
↓
Detailed Report
↓
Summary Prompt
↓
"Generate a 5 pointer
summary from the following text"
↓
LLM
↓
StrOutputParser
↓
5-Point Summary
第一次模型调用只是中间步骤而非最终结果，其输出会成为后续操作的原始数据。如果工作流需要更多处理阶段，可以继续叠加这些阶段：
Prompt
↓
LLM
↓
Parser
↓
Prompt
↓
LLM
↓
Parser
↓
Prompt
↓
LLM
↓
Output
无论流程有多长，各阶段的顺序都是固定的，每个阶段都需要等待前一个阶段完成。这也意味着延迟会逐次累积：连续三次模型调用所需的时间大致等于它们响应时间之和。此外还应当设置机制防止中间输出无限累积，因为过长的报告会直接进入下一个提示语，从而增加其token数量。
并行链：基于相同输入的分布式处理
并行链将多个子链并排运行。与顺序链需要等待每个阶段完成后再启动下一个阶段不同，并行链会同时启动各个独立的子链：
┌──→ Sub-Chain 1 ──→ Output 1
│
Input ──────────────┼──→ Sub-Chain 2 ──→ Output 2
│
└──→ Sub-Chain 3 ──→ Output 3
↓
Combined Output
由于这些子链是相互独立的，因此没有一个子链能够看到其他子链的输出。其工作原理非常简单：
- 每个子链接收相同的输入并生成自己的输出。
- 所有子链同时启动。
- 并行阶段会等待所有子链都完成后再继续。
- 然后将这些结果汇总为一个综合值。
这个综合值实际上是一个字典。每个键是您为某个子链指定的名称，而每个值则是该子链的输出：
{
"facts": "...",
"definition": "...",
"use_case": "..."
}
示例：同一主题的三种视图
以人工智能这样的主题为例，目标是从同一输入中生成三样内容：五个有趣的事实、一个简明的定义以及一个实际应用案例。这些内容彼此之间并无关联，因此依次处理只会浪费时间，应当并行执行：
Input
↓
RunnableParallel
/ | \
/ | \
↓ ↓ ↓
Facts Definition Use Case
↓ ↓ ↓
LLM LLM LLM
↓ ↓ ↓
Parser Parser Parser
\ | /
\ | /
↓ ↓ ↓
Combined Result
有三个子流程负责处理任务：facts用于生成五个事实，definition用于生成简短定义，use_case则用于处理实际应用场景。它们通过RunnableParallel结合在一起。第一种实现方式对这三项任务都使用Gemini；第二种则混合使用OpenAI和Anthropic，并增加了合并步骤。
为何要并行运行子流程
采用这种模式有两个原因。
首先是速度问题。模型调用主要受网络传输和生成时间的影响，因此独立的调用任务能从并发处理中获益良多。想象有三个各需约两秒完成的子任务，如果依次执行，总耗时就是这些时间之和：
Chain 1 → 2 seconds
Chain 2 → 2 seconds
Chain 3 → 2 seconds
Total ≈ 6 seconds
若同时执行，总耗时大致等于其中最慢的那个任务的耗时：
Chain 1 ─┐
Chain 2 ─┼→ execute together → ≈ 2 seconds
Chain 3 ─┘
实际节省的时间取决于服务提供商的速率限制以及任务完成情况的均匀性，但对于独立任务而言，改进效果通常相当显著。第二个原因是结构上的：这些任务虽然共享同一输入数据，但并不需要彼此的处理结果。例如，从同一份文档中，我们可能需要摘要、关键主题列表以及一组学习问题。每个任务都可以直接使用该文档：
Document
↓
┌───────┼───────┐
↓ ↓ ↓
Summary Topics Questions
摘要无需与主题提取或问题生成环节紧密关联。当任务之间存在此类关联时，强行将它们按顺序处理只会增加延迟，还会让实际上并无依赖关系的步骤相互绑定。而当这些任务确实存在依赖关系时，并行处理并非合适的方法，采用顺序处理才是准确的表达方式。
Gemini版本
以下是Gemini中的三路分支结构：
from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel
from dotenv import load_dotenv
load_dotenv()
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
parser = StrOutputParser()
# Prompt 1: Facts
fact_prompt = PromptTemplate(
input_variables=["topic"],
template="Give five interesting short facts about: {topic}"
)
# Prompt 2: Definition
definition_prompt = PromptTemplate(
input_variables=["topic"],
template="Give a concise definition of: {topic}"
)
# Prompt 3: Use Case
usecase_prompt = PromptTemplate(
input_variables=["topic"],
template="What is a real-world use case of: {topic}?"
)
# Run all chains in parallel
parallel_chain = RunnableParallel({
"facts": fact_prompt | llm | parser,
"definition": definition_prompt | llm | parser,
"use_case": usecase_prompt | llm | parser
})
# Invoke parallel chain
result = parallel_chain.invoke({"topic": "Artificial Intelligence"})
# print(type(result))
# print(result)
# print("\n📌 Facts:\n", result["facts"])
# print("\n📘 Definition:\n", result["definition"])
# print("\n🛠️ Use Case:\n", result["use_case"])
parallel_chain.get_graph().print_ascii()
三个提示模板、一个模型和一个解析器被整合为三条小型处理链，随后由RunnableParallel在输入内容“人工智能”上同时执行这些处理链。有多行print语句被注释掉了；启用它们可以显示结果类型及各个字段的内容。
逐行解析Gemini代码
这些导入语句新增了一个组件：
from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel
除了模型封装类、PromptTemplate和StrOutputParser之外，该脚本还从langchain_core.runnables中导入了RunnableParallel，它负责并行运行各个子链。
会创建一个Gemini 2.5 Flash模型：
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
三个子链都共享这个模型。在并行分支间共享一个模型对象是安全的，因为每次调用都是独立的。
解析器也由一个共享：
parser = StrOutputParser()
第一个提示要求提供五个简短的事实：
fact_prompt = PromptTemplate(
input_variables=["topic"],
template="Give five interesting short facts about: {topic}"
)
第二个提示要求给出一个简洁的定义：
definition_prompt = PromptTemplate(
input_variables=["topic"],
template="Give a concise definition of: {topic}"
)
第三个提示要求举出一个实际应用场景：
usecase_prompt = PromptTemplate(
input_variables=["topic"],
template="What is a real-world use case of: {topic}?"
)
这三个提示都使用相同的变量名：
{topic}
正是这个共享的名称使得单个输入字典能够为所有分支提供数据。而这些任务本身则完全相互独立。
现在来看核心部分——并行链：
parallel_chain = RunnableParallel({
"facts": fact_prompt | llm | parser,
"definition": definition_prompt | llm | parser,
"use_case": usecase_prompt | llm | parser
})
RunnableParallel 接收一个值类型为可运行对象的字典，这些键将成为输出结果的键。
facts
definition
use_case
每个值都是一个独立完整的微型链。
fact_prompt → LLM → parser
definition_prompt → LLM → parser
usecase_prompt → LLM → parser
由于它们彼此之间没有依赖关系，因此可以同时运行。使用同步的 invoke 方法时，LangChain会在线程池中执行这些分支；而使用 ainvoke 方法时，它们则作为并发的异步任务运行。
该链只会被调用一次：
result = parallel_chain.invoke({"topic": "Artificial Intelligence"})
使用单个输入：
{"topic": "Artificial Intelligence"}
该字典会被传递给每一个分支。从概念上来说：
Artificial Intelligence
↓
RunnableParallel
/ | \
↓ ↓ ↓
Facts Definition Use Case
↓ ↓ ↓
LLM LLM LLM
↓ ↓ ↓
Parser Parser Parser
\ | /
\ | /
Combined Result
返回结果是什么
最终得到的是一个字典，每个子链对应一个条目：
{
"facts": "1. Artificial Intelligence enables machines to perform tasks that normally require human intelligence.\n2. ...",
"definition": "Artificial Intelligence is the field of creating systems that can perform tasks requiring human-like intelligence.",
"use_case": "A real-world use case of AI is customer support chatbots that can understand and respond to user queries."
}
每次运行时每个字段中的内容会有所不同，但其结构是稳定的，而你的代码正是依赖这一稳定性：
result["facts"]
result["definition"]
result["use_case"]
每个键都会返回对应分支的输出，这样下游代码就能准确获取它需要的部分。
混合提供者并合并结果
第二种实现方式还体现了另外两个理念：不同的分支可以使用不同的提供者，而且并行处理步骤本身也可以成为更大顺序流程中的一个环节。在这里，一个分支使用OpenAI，另一个分支使用Anthropic，最后通过一次OpenAI调用将两个结果合并为一份文档：
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from dotenv import load_dotenv
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.schema.runnable import RunnableParallel
load_dotenv()
model1 = ChatOpenAI()
model2 = ChatAnthropic(model_name='claude-3-7-sonnet-20250219')
prompt1 = PromptTemplate(
template='Generate short and simple notes from the following text \n {text}',
input_variables=['text']
)
prompt2 = PromptTemplate(
template='Generate 5 short question answers from the following text \n {text}',
input_variables=['text']
)
prompt3 = PromptTemplate(
template='Merge the provided notes and quiz into a single document \n notes -> {notes} and quiz -> {quiz}',
input_variables=['notes', 'quiz']
)
parser = StrOutputParser()
parallel_chain = RunnableParallel({
'notes': prompt1 | model1 | parser,
'quiz': prompt2 | model2 | parser
})
merge_chain = prompt3 | model1 | parser
chain = parallel_chain | merge_chain
text = """
Support vector machines (SVMs) are a set of supervised learning methods used for classification, regression and outliers detection.
The advantages of support vector machines are:
Effective in high dimensional spaces.
Still effective in cases where number of dimensions is greater than the number of samples.
Uses a subset of training points in the decision function (called support vectors), so it is also memory efficient.
Versatile: different Kernel functions can be specified for the decision function. Common kernels are provided, but it is also possible to specify custom kernels.
The disadvantages of support vector machines include:
If the number of features is much greater than the number of samples, avoid over-fitting in choosing Kernel functions and regularization term is crucial.
SVMs do not directly provide probability estimates, these are calculated using an expensive five-fold cross-validation (see Scores and probabilities, below).
The support vector machines in scikit-learn support both dense (numpy.ndarray and convertible to that by numpy.asarray) and sparse (any scipy.sparse) sample vectors as input. However, to use an SVM to make predictions for sparse data, it must have been fit on such data. For optimal performance, use C-ordered numpy.ndarray (dense) or scipy.sparse.csr_matrix (sparse) with dtype=float64.
"""
result = chain.invoke({'text': text})
print(result)
chain.get_graph().print_ascii()
输入内容是一篇关于支持向量机的文章。其中一个分支会从中提取学习笔记，另一个分支则会生成包含五道题的测验，随后通过合并流程将笔记与测验整合为一份文档。
此示例中有两点需要特别说明。它从langchain.schema.runnable中导入RunnableParallel，这是langchain_core分离之前的旧路径；当前版本则将其置于langchain_core.runnables下，如Gemini示例所示，旧路径在 newer版本中可能已不存在。此外，它还指定了claude-3-7-sonnet-20250219这一过时的模型标识符；服务提供商会逐步淘汰旧版本模型，因此在重复使用前请查看最新的Anthropic模型列表。
会创建两个模型对象：
model1 = ChatOpenAI()
model2 = ChatAnthropic(model_name='claude-3-7-sonnet-20250219')
ChatOpenAI负责撰写笔记，随后进行合并操作。ChatAnthropic则负责生成测验题。在RunnableParallel中，并没有要求所有分支必须使用相同的模型，因此当某个服务提供商在特定子任务上表现更优或成本更低时，这种设计十分实用。
笔记生成提示会将输入文本转化为简短、简单的笔记：
prompt1 = PromptTemplate(
template='Generate short and simple notes from the following text \n {text}',
input_variables=['text']
)
测验生成提示会从同一文本里提取出五组简短的问答对：
prompt2 = PromptTemplate(
template='Generate 5 short question answers from the following text \n {text}',
input_variables=['text']
)
这两项任务互不需要对方的输出，因此属于天然的并行分支。
第三个提示则不同，它会在其他任务处理完成后运行，并且需要这两项任务的输出结果：
prompt3 = PromptTemplate(
template='Merge the provided notes and quiz into a single document \n notes -> {notes} and quiz -> {quiz}',
input_variables=['notes', 'quiz']
)
它声明了两个输入变量：
notes
quiz
这些名称并非随意指定，必须与并行步骤的键相匹配，因为fan-out操作生成的字典正是该提示所接收的内容。
并行步骤定义了两个分支：
parallel_chain = RunnableParallel({
'notes': prompt1 | model1 | parser,
'quiz': prompt2 | model2 | parser
})
它的结构是双向拆分后再合并为字典的形式：
Input Text
↓
RunnableParallel
/ \
↓ ↓
Notes Chain Quiz Chain
↓ ↓
OpenAI Anthropic
↓ ↓
Parser Parser
\ /
↓ ↓
Dictionary
从概念上讲，中间值的结构如下：
{
"notes": "...",
"quiz": "..."
}
合并链是一个普通的简单链结构：
merge_chain = prompt3 | model1 | parser
它从并行步骤获取字典，填充{notes}和{quiz}，然后请求OpenAI整合成一份文档。
最后，这两个阶段通过另一个管道连接起来：
chain = parallel_chain | merge_chain
最终得到的是一种混合工作流：开始时并行处理，结束时顺序处理：
Input Text
↓
RunnableParallel
/ \
↓ ↓
Notes Quiz
↓ ↓
OpenAI Anthropic
↓ ↓
Parser Parser
\ /
↓ ↓
Combined Dictionary
↓
Merge Prompt
↓
OpenAI
↓
Parser
↓
Final Output
关键的行为特点在于这两个分支会独立运行，但必须等到两者都完成之后才能开始合并阶段。因此，速度最慢的分支决定了整个并行处理的节奏。如果某个分支失败，平行处理就会中止，合并操作也永远不会执行，所以在这些情况至关重要的时候，应当用重试或备用方案来包装不可靠的提供者。
整个流程是通过以下代码调用的：
result = chain.invoke({'text': text})
笔记生成分支和测验生成分支会同时处理文本段落。当两者都完成后，它们的输出会被送入合并链，而合并链生成的字符串结果就是invoke函数返回的值。
打印合并链的结构
图形工具也能直观展示这种并行结构：
parallel_chain.get_graph().print_ascii()
对于三分支的Gemini链，图表展示了其分支分离与合并的过程：
Parallel<...>Input
/ | \
/ | \
PromptTemplate PromptTemplate PromptTemplate
↓ ↓ ↓
LLM LLM LLM
↓ ↓ ↓
StrOutputParser StrOutputParser StrOutputParser
\ | /
\ | /
Parallel<...>Output
Parallel<...>Input和Parallel<...>Output节点分别标识输入数据的分配位置以及结果汇总回字典的位置。
两种实现的比较
这两种版本都体现了相同的理念：同时运行的独立子流程。Gemini版本则是在单个模型上通过三个分支实现纯粹的扇出结构：
Input
↓
RunnableParallel
├── Facts → Gemini → Parser
├── Definition → Gemini → Parser
└── Use Case → Gemini → Parser
↓
Dictionary Output
混合提供者版本会在不同的提供者上分成两个分支，然后再添加一个合并阶段：
Input
↓
RunnableParallel
├── Notes → OpenAI → Parser
└── Quiz → Anthropic → Parser
↓
Dictionary Output
↓
Merge Chain
↓
OpenAI
↓
Output
在两者背后都存在着相同的结构：
RunnableParallel({
"chain_1": chain_1,
"chain_2": chain_2,
"chain_3": chain_3
})
键用于命名子链，值则是这些子链本身。作为快捷方式，当管道表达式中出现普通的可执行对象字典时，LCEL会自动将其转换为RunnableParallel，因此你经常会看到该字典没有显式写出对应的类名。
完整的数据路径
一般来说，并行链会分发一个输入并收集一个字典：
Input
↓
RunnableParallel
/ | \
↓ ↓ ↓
Sub-chain Sub-chain Sub-chain
↓ ↓ ↓
Output Output Output
\ | /
\ | /
↓ ↓ ↓
Dictionary
以 Gemini 上的人工智能示例为例：
{"topic": "Artificial Intelligence"}
↓
RunnableParallel
/ | \
↓ ↓ ↓
Facts Definition Use Case
↓ ↓ ↓
Gemini Gemini Gemini
↓ ↓ ↓
Parser Parser Parser
\ | /
\ | /
Dictionary
最终结果中每个分支对应一个键：
{
"facts": "...",
"definition": "...",
"use_case": "..."
}
这就是它与顺序链的本质区别。在顺序链中，每个阶段都会使用前一个阶段的输出：
A → B → C → D
在并行链中，各个独立的阶段会在相同的输入上同时运行，然后它们的结果会被合并：
┌→ A ─┐
Input ─┼→ B ─┼→ Combined Output
└→ C ─┘
对于那些输入相同但结果不同的任务，应使用并行处理；而对于真正需要基于前序步骤继续处理的任务，则应采用顺序处理方式。像笔记与测验示例中所做的那样将两者混用是很常见的，也是完全符合常规做法的。
条件链：将输入路由到正确的子链
条件链会在运行时选择下一步操作。顺序链总是按固定顺序执行每个阶段，而并行链则同时执行所有分支。条件链会针对输入评估某个条件，仅执行与之匹配的分支：
Input
↓
Condition Check
↓
┌───────────────┐
│ │
Yes No
↓ ↓
Chain 1 Chain 2
↓ ↓
Output 1 Output 2
\ /
\ /
Final Output
在 LangChain 中，这通过 RunnableBranch 来实现。你需要提供一系列条件与对应可执行操作的配对列表，最后再指定一个默认的可执行操作。系统会按顺序测试这些条件，第一个返回 True 的条件所对应的那项可执行操作就会被执行；如果没有任何条件满足，则由默认的可执行操作来处理输入。
示例：根据文本长度进行路由
该示例会根据文本的长度采取不同的处理方式：
- 如果文本的单词数超过 50 个，则生成简短摘要。
- 否则，则对文本中的内容进行详细阐述。
最终的工作流程如下：
Input Text
↓
Condition Check
↓
Is word count > 50?
↓
┌───────┴───────┐
Yes No
↓ ↓
Summary Extend
Chain Chain
↓ ↓
LLM LLM
↓ ↓
Parser Parser
\ /
\ /
Final Output
有两个提示用于引导分支生成：summary_prompt要求提供简短总结，而extend_prompt则要求模型在原有基础上进一步发展。有一个名为is_long_text()的小函数用于统计单词数，RunnableBranch则根据其返回结果来选择后续处理路径。与其他模式一样，这也存在Gemini版本和OpenAI版本。
从通用结构来看，这种双向条件流程如下：
User Input
↓
Condition Check
↓
┌───────┴───────┐
Yes No
↓ ↓
Prompt 1 Prompt 2
↓ ↓
LLM LLM
↓ ↓
Parser Parser
\ /
└──────┬───────┘
↓
Final Output
对于任意给定的输入，只会有一条分支被执行。
以具体输入为例
示例输入是一句关于机器学习的定义，将其描述为人工智能的一个分支，通过统计方法让计算机从数据中学习。
该句子共有22个单词，远低于50个单词的阈值，因此条件判断结果为False，于是选择扩展型处理路径：
Input Text
↓
Condition Check
↓
22 > 50?
↓
No
↓
Extend Chain
↓
Gemini 2.5 Flash
↓
StrOutputParser
↓
Extended Text
如果输入文本长度超过50个字，条件就会变为True，此时输入会被发送到摘要处理流程。
何时需要条件路由
当相同类型的输入因某些属性不同而需要不同处理方式时，路由功能非常有用：
- 长度：对长文本进行简化，对短文本进行详细阐述。
- 情感倾向：将正面和负面反馈分别发送到不同的响应流程。
- 请求类型：将各类请求引导至相应的专用处理流程。
关键在于无需为不需要的分支支付成本，只有选定的路径才会被执行：
Input
↓
Condition
↓
Choose one branch
↓
Execute selected chain
↓
Output
这使得工作流程具备固定顺序所没有的适应性，同时与逐一执行所有分支并丢弃结果相比，还能减少模型调用次数。
Gemini版本
以下是在Gemini上运行的完整路由程序：
from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch
from dotenv import load_dotenv
load_dotenv()
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
parser = StrOutputParser()
# Prompt 1 : Summary
summary_prompt = PromptTemplate(
input_variables=["text"],
template="Summarize the following text briefly:\n\n{text}"
)
# Prompt 2 : Extend
extend_prompt = PromptTemplate(
input_variables=["text"],
template="Extend the idea in the following long text:\n\n{text}"
)
# Chain 1 : Summary
summary_chain = summary_prompt | llm | parser
# Chain 2 : Extend
extend_chain = extend_prompt | llm | parser
def is_long_text(input: dict) -> bool:
return len(input["text"].split()) > 50
conditional_chain = RunnableBranch(
(lambda x: is_long_text(x), summary_chain),
extend_chain
)
text_input = {
"text": "Machine learning is a field of artificial intelligence that uses statistical techniques to give computer systems the ability to learn from data."
}
result = conditional_chain.invoke(text_input)
# print("\nOutput:\n")
# print(result)
conditional_chain.get_graph().print_ascii()
它定义了两个小型链结构summary_chain和extend_chain，以及一个字数统计谓词，然后让RunnableBranch根据该谓词在两个链结构之间进行选择。
逐行解析Gemini代码
所使用的模型依旧是Gemini 2.5 Flash：
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
StrOutputParser负责将每个响应转换为字符串：
parser = StrOutputParser()
摘要提示要求提供简短的总结内容：
summary_prompt = PromptTemplate(
input_variables=["text"],
template="Summarize the following text briefly:\n\n{text}"
)
扩展提示则用于另一条路径上：
extend_prompt = PromptTemplate(
input_variables=["text"],
template="Extend the idea in the following long text:\n\n{text}"
)
仔细看看其表述：它提到的是“以下长文本”，但实际上这个分支仅在输入较短时才会执行。这种不一致虽然不会导致任何问题，但会给模型传递错误的提示。在你的版本中，将其改写为“对以下文本中的观点进行扩展”这样的表述是个值得做的简单修正。
每个提示都会与模型及解析器结合，形成独立的处理链：
summary_chain = summary_prompt | llm | parser
extend_chain = extend_prompt | llm | parser
这样就产生了两条独立的路径：
Summary Prompt → Gemini → Parser
Extend Prompt → Gemini → Parser
条件判断步骤决定了哪条路径会被执行。
该条件是一个普通的Python函数：
def is_long_text(input: dict) -> bool:
return len(input["text"].split()) > 50
它会根据空白字符将input[“text”]拆分，并统计分割后的片段数量。其功能非常简单：
True → if word count > 50
False → otherwise
按空白字符分割只能粗略统计字数；标点符号会附着在单词上，连写的词也会被视为一个单元。这种方法适用于按长度进行路由判断，但若真正关心模型容量限制，则应统计标记数而非字数。另外需注意，参数名input与Python内置函数同名；虽然能使用，但用data这类名称会更清晰。
在所有组件准备就绪后，RunnableBranch会组装出路由器：
conditional_chain = RunnableBranch(
(lambda x: is_long_text(x), summary_chain),
extend_chain
)
第一个参数是一个作为规则的元组：
If is_long_text(x) is True
↓
Execute summary_chain
最后一个参数没有用元组包裹，它代表默认值：
Otherwise
↓
Execute extend_chain
无需使用lambda包装函数。RunnableBranch可以接受任何可调用对象作为条件，因此直接传入is_long_text同样可行。综合起来，路由逻辑如下所示：
Input
↓
is_long_text()
/ \
True False
↓ ↓
summary_chain extend_chain
↓ ↓
Gemini Gemini
↓ ↓
Parser Parser
\ /
\ /
Output
运行时需要输入字典：
result = conditional_chain.invoke(text_input)
首先会检查条件，只有符合条件的分支才会调用模型。
返回结果
对于22个单词的示例输入，该谓词会返回：
is_long_text() → False
因此由extend_chain来处理输入。从概念上讲：
Input Text
↓
Word Count Check
↓
False
↓
extend_chain
↓
Gemini 2.5 Flash
↓
StrOutputParser
↓
Extended Text
如果输入文本超过50个单词，该谓词会返回：
is_long_text() → True
这会导致路径变为：
Input Text
↓
Word Count Check
↓
True
↓
summary_chain
↓
Gemini 2.5 Flash
↓
StrOutputParser
↓
Short Summary
一个处理流程，两种可能的路径，由输入本身决定。
OpenAI上的相同路由器
OpenAI版本保留了路由逻辑，仅更换模型：
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch
from dotenv import load_dotenv
load_dotenv()
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
# Prompt 1 : Summary
summary_prompt = PromptTemplate(
input_variables=["text"],
template="Summarize the following text briefly:\n\n{text}"
)
# Prompt 2 : Extend
extend_prompt = PromptTemplate(
input_variables=["text"],
template="Extend the idea in the following long text:\n\n{text}"
)
# Chain 1 : Summary
summary_chain = summary_prompt | llm | parser
# Chain 2 : Extend
extend_chain = extend_prompt | llm | parser
def is_long_text(input: dict) -> bool:
return len(input["text"].split()) > 50
conditional_chain = RunnableBranch(
(lambda x: is_long_text(x), summary_chain),
extend_chain
)
text_input = {
"text": "Machine learning is a field of artificial intelligence that uses statistical techniques to give computer systems the ability to learn from data."
}
result = conditional_chain.invoke(text_input)
print(result)
conditional_chain.get_graph().print_ascii()
两条处理链，一个谓词，一个RunnableBranch，与Gemini的架构相同。
所使用的模型为gpt-4o-mini：
llm = ChatOpenAI(model="gpt-4o-mini")
两个提示模板保持不变，一个用于总结，另一个用于扩展：
summary_prompt = PromptTemplate(
input_variables=["text"],
template="Summarize the following text briefly:\n\n{text}"
)
extend_prompt = PromptTemplate(
input_variables=["text"],
template="Extend the idea in the following long text:\n\n{text}"
)
链的构成与之前完全相同：
summary_chain = summary_prompt | llm | parser
extend_chain = extend_prompt | llm | parser
现在两条路径都通过OpenAI处理：
Summary Prompt → OpenAI → StrOutputParser
Extend Prompt → OpenAI → StrOutputParser
判断条件仍是那50字的检查：
def is_long_text(input: dict) -> bool:
return len(input["text"].split()) > 50
路由器的构建方式也相同：
conditional_chain = RunnableBranch(
(lambda x: is_long_text(x), summary_chain),
extend_chain
)
如果结果为True，则将输入发送到summary_chain；否则发送到extend_chain。
随后调用该链并打印结果：
result = conditional_chain.invoke(text_input)
print(result)
对于示例句子，判断条件再次得出：
is_long_text() → False
因此会执行扩展链，返回输入内容的扩展解释，类似如下：
Machine learning enables computer systems to learn patterns
from data and improve their performance without being explicitly
programmed for every task. It is widely used in applications
such as recommendation systems, image recognition, and predictive
analytics.
具体文本取决于模型，在不同运行中会发生变化。
打印链的结构
同样的图表工具也适用于路由器：
conditional_chain.get_graph().print_ascii()
渲染后的图表中最重要的元素是Branch节点。它就是LangChain评估条件并将输入传递给某个候选链的决策点。
两个分支的区别
每个分支都有各自的职责。总结分支：
summary_prompt
↓
Gemini
↓
StrOutputParser
↓
Short Summary
扩展分支：
extend_prompt
↓
Gemini
↓
StrOutputParser
↓
Extended Text
谓词会在它们之间进行选择：
Input
↓
is_long_text()
/ \
True False
↓ ↓
Summary Chain Extend Chain
这与并行链的关键区别在于：各个分支不会同时运行。只有满足条件的路径才会被执行，这样成本和延迟就能与实际所需的工作量成正比。
基于模型自身分类的路由
条件并不仅限于字数统计这类简单的检查。一种更为常见且有效的模式是让第一个处理链对输入进行分类，然后再根据该分类结果进行路由。在情感分析的例子中，收到的反馈会被标记为两个值之一：
positive
negative
RunnableBranch则会将对应标签的反馈转发给相应的响应处理链：
Feedback
↓
Sentiment Classifier
↓
┌───────────────┐
│ │
Positive Negative
│ │
↓ ↓
Positive Negative
Response Response
│ │
└───────┬───────┘
↓
Output
构建分类器的可靠方法是让模型返回经过Pydantic模型验证的结构化输出，例如将字段限制为positive或negative。这样，分支条件只需检查该字段即可，而无需解析自由格式的文本，从而避免因模型以大写字母和句号的形式返回“Positive.”而产生的意外情况。始终要设置一个合理的默认分支，因为分类器最终很可能会返回一些你未曾预料到的结果。
这表明一个条件不仅可以依赖于原始输入，还可以依赖于之前流程产生的信息。对于更复杂的路由逻辑，当前的 LangChain 文档还介绍了使用包裹在 RunnableLambda 中的自定义函数来指定要运行的流程；建议查看文档中针对您所使用版本推荐的方案。
完整的数据路径
一般的条件判断模式为：
Input
↓
Condition Check
↓
┌───────┴───────┐
↓ ↓
Condition Condition
True False
↓ ↓
Chain 1 Chain 2
↓ ↓
LLM LLM
↓ ↓
Parser Parser
\ /
\ /
└─────┬───────┘
↓
Output
以文本长度为例：
Machine Learning Text
↓
is_long_text()
↓
Word Count > 50?
/ \
Yes No
↓ ↓
Summary Extend
Chain Chain
↓ ↓
Gemini Gemini
↓ ↓
Parser Parser
\ /
\ /
Output
较短输入走一条路径：
Word Count ≤ 50
↓
Extend Chain
较长输入走另一条路径：
Word Count > 50
↓
Summary Chain
总结一下，这四种形态并列如下：
Simple Chain
A → B → C
Sequential Chain
A → B → C → D → E
Parallel Chain
┌→ B ─┐
A ────┼→ C ─┼→ Output
└→ D ─┘
Conditional Chain
↓
A
↓
Condition
/ \
B C
条件流程让大型语言模型工作流能够根据输入或流程中早先计算出的值来做出决策，从而执行不同的逻辑。
发货前的实用注意事项
从示例过渡到实际代码时，有几点常见问题值得注意：
- 固定模型名称。没有模型参数的构造函数会继承已安装集成所使用的默认值，而带有日期标识的名称最终会被废弃。
- 保留解析器。聊天模型会返回消息对象；
StrOutputParser能为后续的每个处理步骤提供可预测的str类型数据。 - 注意数据传递过程。只有当提示语中恰好包含一个变量时，纯字符串才会自动填充该提示语。如需更多功能，请显式构建字典。
- 确保键与变量对应。在先拆分再合并的操作中，各个分支的键必须与合并后的提示语中的变量名一致。
langchain_core.runnables中获取可执行对象，而非使用旧版的langchain.schema路径。grandalf后，使用get_graph().print_ascii()是一种简单的方法，可确认复杂的链结构是否按预期连接。如需更全面的介绍，了解这些链如何与检索功能、本地模型运行的工具及智能体结合使用，请参阅LangChain 1.x实践：本地链、RAG、工具及智能体。
核心要点
- 链是一种已定义的流水线，其中每个组件的输出会传递给下一个组件；
|运算符可利用任何可执行对象来构建此类链。 - 简单链仅包含一次处理流程；而顺序链则由多轮处理堆叠而成，每增加一个阶段就会带来额外的延迟。
RunnableParallel会同时运行多个独立的子链，并返回一个以分支名称为键的字典，后续阶段可通过变量名来访问这些数据。RunnableBranch会按顺序评估条件，仅执行第一个符合条件的分支，若没有匹配项则使用默认分支。- 这些条件可以是针对输入的简单检查，也可以是基于早期模型结构化分类做出的决策。
- 由于该模型仅由一个组件构成，因此在 Gemini、OpenAI 或 Anthropic 之间切换通常只需修改一行代码，而提示词、解析器及结构则保持不变。
这四种模式都基于同一个基础，即 Runnable 接口，正是该接口使得提示词、模型、解析器以及用户自定义的函数能够被组合在一起。在熟悉了这些结构之后，直接理解该接口便是自然而然的下一步。