LangChain Runnables核心库怎么用?链式调用、流式处理、并行执行全掌握

　　引言

　　痛点场景：当“会调API”变成“不会搭流程”

　　你可能已经能熟练地调用LangChain的各种组件——用ChatPromptTemplate写提示词，用ChatOpenAI调模型，用StrOutputParser处理输出。但当你试图将这些步骤串成一个完整的业务流水线时，代码开始变得臃肿不堪：到处都是if判断、手动传递变量、反复处理异步回调。你想让检索和生成并行执行来提速，却发现自己写着“面条代码”，调试起来痛苦不堪。你缺的不是对单个组件的了解，而是一套能优雅“编排”它们的方法论。

　　核心价值：从“堆砌代码”到“声明式编排”

　　LangChain的Runnables核心库提供了一套统一的接口和组合原语，让你能像搭积木一样声明式地构建复杂的AI任务流水线。无论你是需要顺序执行、并行加速，还是流式输出，Runnables都能帮你用最简洁的代码实现。本文将带你掌握三个最核心的能力：链式调用(RunnableSequence)、流式处理(Streaming)、并行执行(RunnableParallel)，让你从“会调组件”进阶到“会搭流程”。

　　提纲预览

　　本文将围绕LangChain Runnables库的三个核心功能展开：

　　理解Runnable接口：掌握统一执行范式和五种调用方式。

　　链式调用：用|管道符或RunnableSequence构建线性任务流。

　　并行执行：用RunnableParallel让独立任务同时跑，大幅提速。

　　流式处理：用stream方法实时输出，改善用户体验。

　　组合实战：结合RAG场景演示三者的综合运用。

　　前置准备

　　开始动手前，你需要准备好以下东西：

　　Python环境：Python 3.8以上，建议使用虚拟环境隔离依赖。

　　LangChain核心库：安装langchain-core，这是Runnables的所在地。如果要用具体模型或向量库，根据需要安装langchain-openai、langchain-community等。

　　一个可用的LLM API Key：如OpenAI、Anthropic或国内模型服务商的API密钥，用于实际测试链式调用。

　　对LangChain基础组件的了解：知道什么是PromptTemplate、ChatModel、OutputParser即可，不要求精通。

　　核心步骤

　　步骤1：理解Runnable——一切可执行单元的统一接口

　　Runnable是LangChain中所有可执行组件的“基类” 。提示词模板、大模型、输出解析器、甚至你自己写的函数，都可以被包装成Runnable。它的设计哲学是统一执行范式：无论组件内部多复杂，对外都暴露同一套方法 。

　　每个Runnable都支持五种核心调用方式 ：

　　invoke：同步单次调用，阻塞式返回完整结果。

　　ainvoke：异步单次调用，适合高并发场景。

　　stream：流式输出，逐块返回结果(如生成文本时逐token输出)。

　　batch：批量调用，输入一个列表，并行执行多次invoke。

　　abatch：异步批量调用。

　　关键洞察：只要一个组件实现了Runnable接口，它就能和任何其他Runnable自由组合。这就像所有电器都用统一的插座，你可以随意插拔、串联。

　　步骤2：链式调用——用“|”管道符串联任务流

　　链式调用是Runnables最基础也最常用的组合方式，核心类是RunnableSequence 。它做的事情很简单：把上一个Runnable的输出，作为下一个Runnable的输入 。LangChain提供了两种等价的写法：

　　方式一：管道符 |(最推荐)

　　python

　　f rom langchain_core.prompts import ChatPromptTemplate

　　f rom langchain_core.output_parsers import StrOutputParser

　　f rom langchain_openai import ChatOpenAI

　　model = ChatOpenAI()

　　prompt = ChatPromptTemplate.from_template("写一首关于{topic}的2行诗")

　　chain = prompt | model | StrOutputParser()

　　result = chain.invoke({"topic": "月亮"})

　　print(result)

　　这种写法最直观，从左到右读就是“提示词模板处理输入 -> 模型生成 -> 解析器提取文本”。

　　方式二：RunnableSequence.f rom()

　　python

　　f rom langchain_core.runnables import RunnableSequence

　　chain = RunnableSequence.f rom([prompt, model, StrOutputParser()])

　　两者完全等价，|运算符会被自动转换为RunnableSequence 。链式调用的最大价值在于自动处理输入输出的流转——你不需要手动把模型输出传给解析器，框架帮你做了 。

　　步骤3：并行执行——用RunnableParallel让任务同时跑

　　当你需要同时做两件互不依赖的事情时，顺序执行会浪费大量时间。比如同时生成一个笑话和一首诗，它们都只需要“topic”作为输入，完全可以并行 。这时就要用到RunnableParallel(也叫RunnableMap)。

　　核心原理：RunnableParallel接收一个字典，字典的每个值都是一个Runnable。它用同一个输入同时调用所有Runnable，然后返回一个字典，键名对应每个分支 。

　　python

　　f rom langchain_core.runnables import RunnableParallel

　　f rom langchain_core.prompts import ChatPromptTemplate

　　f rom langchain_openai import ChatOpenAI

　　model = ChatOpenAI()

　　# 定义两个独立的任务链

　　joke_chain = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话") | model | StrOutputParser()

　　poem_chain = ChatPromptTemplate.from_template("写一首关于{topic}的2行诗") | model | StrOutputParser()

　　# 并行执行

　　parallel_chain = RunnableParallel(

　　joke=joke_chain,

　　poem=poem_chain

　　)

　　result = parallel_chain.invoke({"topic": "熊"})

　　print(result["joke"]) # 笑话

　　print(result["poem"]) # 诗

　　性能收益：实测中，并行执行两个任务的总耗时约等于单个任务耗时，而不是两者之和 。当你在RAG场景中需要同时做“检索相关文档”和“处理用户问题”时，用RunnableParallel能显著缩短响应时间。

　　进阶用法：在链中直接写字典也会被自动转为RunnableParallel ：

　　python

　　chain = {"context": retriever, "question": RunnablePassthrough()} | prompt | model | parser

　　这里{"context": retriever, "question": RunnablePassthrough()}就是一个RunnableParallel，它会并行执行检索和透传用户问题。

　　步骤4：流式处理——用stream方法实时输出

　　对于生成式任务，用户等待完整响应时体验往往不佳。流式处理允许你逐token输出结果，让用户“边看边等”。Runnables原生支持stream方法 。

　　python

　　# 定义一个链

　　chain = prompt | model | StrOutputParser()

　　# 流式调用

　　for chunk in chain.stream({"topic": "人工智能"}):

　　print(chunk, end="", flush=True)

　　每次迭代会返回一个片段(可能是一个token或几个字符)，你可以实时推送给前端 。

　　注意事项：RunnableLambda(用函数包装的Runnable)默认不支持流式输出，因为函数是一块执行的 。如果你的链中包含RunnableLambda，流式输出会在该节点处“卡住”，等函数执行完才继续。解决办法是让函数返回生成器，或直接继承Runnable实现transform方法 。

　　步骤5：综合实战——构建一个RAG检索链

　　用一个典型的RAG场景把上述三个能力串起来：

　　用户提一个问题

　　同时做两件事：从向量库检索相关文档，并透传用户问题

　　把检索结果和问题一起塞给Prompt

　　模型生成答案，流式输出

　　python

　　f rom langchain_core.runnables import RunnablePassthrough, RunnableParallel

　　f rom langchain_core.prompts import ChatPromptTemplate

　　f rom langchain_core.output_parsers import StrOutputParser

　　f rom langchain_openai import ChatOpenAI, OpenAIEmbeddings

　　f rom langchain_community.vectorstores import FAISS

　　# 准备检索器

　　vectorstore = FAISS.from_texts(

　　["LangChain的Runnable接口是所有可执行组件的基类"],

　　OpenAIEmbeddings()

　　)

　　retriever = vectorstore.as_retriever()

　　# 提示词模板

　　template = """基于以下上下文回答问题：

　　{context}

　　问题：{question}

　　"""

　　prompt = ChatPromptTemplate.from_template(template)

　　model = ChatOpenAI()

　　# 构建链：并行检索 + 透传问题 -> 提示词 -> 模型 -> 解析器

　　retrieval_chain = (

　　RunnableParallel(

　　context=retriever,

　　question=RunnablePassthrough()

　　)

　　| prompt

　　| model

　　| StrOutputParser()

　　)

　　# 流式调用

　　for chunk in retrieval_chain.stream("Runnable是什么?"):

　　print(chunk, end="", flush=True)

　　这个例子中，RunnableParallel让检索和透传并行执行，|串联了五个步骤，stream实现了逐词输出。

　　常见问题与避坑指南

　　问题一：管道符|左边或右边不是Runnable怎么办?

　　LangChain会自动进行类型转换。函数会被转为RunnableLambda，字典会被转为RunnableParallel 。但要注意：转换后的行为可能和预期不同，比如函数默认不支持流式。

　　问题二：并行执行时一个分支报错会怎样?

　　RunnableParallel会等待所有分支完成。如果一个分支抛出异常，整个invoke会抛出该异常。建议在每个分支内部做异常捕获和兜底处理，让并行任务“尽量成功”。

　　问题三：流式输出时如何区分多个并行分支的输出?

　　当你在RunnableParallel上调用stream时，每个分支的流式输出会以字典形式交错返回，键名对应分支名称 。你需要在UI层按key分别渲染。

　　问题四：batch和parallel有什么区别?

　　batch是在同一个Runnable上处理多个输入，本质是并行调用invoke;RunnableParallel是在多个Runnable上处理同一个输入 。两者解决的是不同维度的问题。

　　进阶技巧/额外提示

　　善用RunnablePassthrough保留原始输入

　　当你需要把原始输入“透传”到链的后续环节时，RunnablePassthrough非常有用。比如在RAG链中，你需要同时把“用户问题”传给检索器和提示词模板，前者需要向量化检索，后者需要拼接上下文 。RunnablePassthrough会原样传递输入，不进行任何转换。

　　用itemgetter优雅地提取字典字段

　　Python标准库的itemgetter可以和RunnableParallel配合使用，从输入字典中提取特定字段 ：

　　python

　　f rom operator import itemgetter

　　chain = (

　　{

　　"context": itemgetter("question") | retriever,

　　"question": itemgetter("question"),

　　"language": itemgetter("language"),

　　}

　　| prompt

　　| model

　　| parser

　　)

　　这样写比用RunnablePassthrough更清晰，特别是需要从输入中提取多个字段时。

　　总结

　　回顾全文，LangChain的Runnables核心库为我们提供了一套统一的“积木”系统。我们首先理解了Runnable接口的统一执行范式;然后用管道符|和RunnableSequence实现了链式调用，让数据在组件间自动流转;接着通过RunnableParallel让互不依赖的任务并行执行，大幅提升效率;再通过stream方法实现了流式输出，优化了用户体验;最后用一个RAG检索链的例子，展示了三者如何协同工作。

　　掌握Runnables，你就掌握了LangChain的“语法”。下一步，你可以深入学习RunnableBranch(条件分支，根据输入走不同链路)和RunnableWithFallbacks(为关键链路配置备选方案，提升鲁棒性)。这两者让你的流程控制更加灵活和健壮。

　　常见问答

　　问：Runnable和LCEL是什么关系?

　　答：LCEL(LangChain Expression Language)就是基于Runnables构建的声明式语法。你写的prompt | model | parser这种代码就是LCEL。Runnables是LCEL的“执行引擎”，LCEL是Runnables的“语法糖”。理解了Runnables，你就理解了LCEL的背后原理。

　　问：RunnableLambda包装的函数能访问外部变量吗?

　　答：可以，但要注意闭包捕获的是引用。如果外部变量在运行时变化，可能会影响执行结果。另外，RunnableLambda默认不支持流式，如果需要在lambda中实现流式，可以返回一个生成器(Generator)。

　　问：如何限制并行执行的最大并发数?

　　答：batch方法支持max_concurrency参数，可以控制同时执行的线程数。RunnableParallel的并行度由异步运行时调度，目前没有直接的控制参数，但可以通过配置RunnableConfig中的max_concurrency来间接控制。

　　问：在链中间如何插入日志打印或调试信息?

　　答：可以用RunnableLambda包装一个打印函数，插入到链的任意位置。也可以使用astream_log方法，它会返回包含中间步骤输出的日志流 ，是调试长链的利器。

　　如果你想用LangChain Runnables构建复杂的AI应用，但缺乏专业开发人力，不妨将需求发布到一品威客任务大厅。无论是搭建一个支持并行检索的RAG问答系统，还是构建能流式输出的大模型对话应用，平台上的百万服务商都能提供从架构设计到代码落地的全套服务。你可以在人才大厅直接对接精通LangChain的Python工程师，在服务大厅浏览各类成熟的AI解决方案。通过商铺案例参考评估服务商的专业交付能力，在雇主攻略中学习如何清晰描述技术需求、把控项目进度。