第05章 Runnable 与 LCEL 基础这一章就是讲 Chain。1. 本章目标前面几章中我们已经使用过Prompt 模板DeepSeek 模型Output Parser但是这些组件一直是分开调用的promptChatPromptTemplate.from_template(...)modelget_model()parserStrOutputParser()# 手动逐步调用messagesprompt.invoke({topic:Git}).to_messages()replymodel.invoke(messages)textparser.invoke(reply)本章学习使用LCEL把这些组件组合成一条处理链。学完本章后你应该能够理解 Runnable 和 LCEL 的作用使用|组合 Prompt、模型和 Parser使用invoke处理单条输入使用batch批量处理输入使用stream获取流式输出使用RunnableLambda包装普通函数使用RunnableParallel并行执行多个任务完成评论分析流水线案例2. 什么是 RunnableRunnable是 LangChain 中可以被执行和组合的组件。前面学过的很多组件本身就是 RunnablePrompt 模板Chat ModelOutput ParserRetriever检索器Runnable 通常提供统一的调用方法方法作用invoke处理单条输入batch批量处理多条输入stream流式返回结果ainvoke异步处理单条输入abatch异步批量处理本章重点掌握前三个。3. 什么是 LCELLCEL全称是LangChain Expression LanguageLangChain 表达式语言。假如多个组件都是 Runnable 旗下的组件那么它们之间就可以使用|来连接连接起来的东西就叫做chainchainprompt|model|parser它是一种组合 LangChain 组件的方式。最常见的写法是chainprompt|model|parser resultchain.invoke({topic:Git})数据会按照下面的顺序流动{topic: Git} ↓ prompt → 生成 messages ↓ model → 生成 AIMessage ↓ parser → 提取出普通字符串这种写法比手动逐步调用更简洁也更容易复用。4. 第一个 LCEL Chain先看没有 LCEL 的写法# 逐步调用messagesprompt.invoke({topic:Git}).to_messages()ai_messagemodel.invoke(messages)resultparser.invoke(ai_message)print(result)使用 LCEL 后chainprompt|model|parser resultchain.invoke({topic:Git})print(result)两种写法完成的任务相同。LCEL 的优势是代码更清晰组件可以复用自动支持invoke自动支持batch自动支持stream5. 案例一技术概念解释链创建01_concept_explainer.pyfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromutils.model_factoryimportget_model promptChatPromptTemplate.from_template(请用通俗易懂的语言向编程初学者解释「{concept}」这个概念不超过 100 字。)modelget_model()parserStrOutputParser()chainprompt|model|parser resultchain.invoke({concept:Git 的分布式版本控制})print(result)运行python 01_concept_explainer.py这里的chain是一条完整处理链。输入是字典{concept:Git 的分布式版本控制}输出是普通字符串。6. invoke 单条调用invoke用于处理单条输入也是我们最常用的调用方式。chainprompt|model|parser# 单条输入resultchain.invoke({concept:Git 的分布式版本控制})print(result)要点invoke接收一条输入返回一个最终结果。输入通常是字典对应 Prompt 模板里的变量名。返回值类型取决于链的最后一步以StrOutputParser结尾就返回字符串以PydanticOutputParser结尾就返回对象。7. batch 批量处理batch用于一次性处理多条输入比在循环里反复invoke更高效底层会自动做并发调度。chainprompt|model|parser# 多条输入用列表传入resultschain.batch([{concept:Git 的分布式版本控制},{concept:Git 的暂存区},{concept:Git 的分支},])forrinresults:print(r)要点batch接收列表列表里每个元素都是一条独立的输入。返回值是与输入顺序一致的结果列表。适合对一批数据做同样的处理如批量生成文案、批量抽取信息。8. stream 流式输出stream用于逐块返回结果适合在 Web 页面或终端里实现打字机式的实时输出效果而不是等全部生成完才一次性显示。chainprompt|model|parser# 流式输出forchunkinchain.stream({concept:Git 的合并冲突}):print(chunk,end,flushTrue)print()要点stream返回一个生成器每次yield一小块文本。普通invoke是攒齐了再给你stream是生成一点就吐一点。只有最终落到字符串/文本流上的链才能逐字输出如果链以 Pydantic 解析器结尾返回对象stream通常意义不大更适合invoke/batch。9. RunnableLambda 包装普通函数链里的每一步都必须是 Runnable。如果你有一段普通 Python 函数想接进链里比如做格式转换、字段映射、后处理需要用RunnableLambda把它包成 Runnable。fromlangchain_core.runnablesimportRunnableLambdadefto_upper(text:str)-str:# 普通函数把结果转成大写returntext.upper()chainprompt|model|parser|RunnableLambda(to_upper)resultchain.invoke({concept:Git 的远程仓库})print(result)为什么需要它普通函数不能直接用|接在链后面会报错。RunnableLambda让普通函数变成 Runnable从而可以无缝串进 LCEL 链。常用于清洗模型输出、抽取某字段、拼装最终结构等。10. RunnableParallel 并行执行多个任务RunnableParallel可以同时跑多个 Runnable常用于一个输入、多个不同处理分支。fromlangchain_core.runnablesimportRunnableParallel# 两个独立的解释链explain_chainprompt|model|parser summary_chain(ChatPromptTemplate.from_template(用一句话总结「{concept}」)|model|parser)parallel_chainRunnableParallel({explain:explain_chain,summary:summary_chain,})resultparallel_chain.invoke({concept:Git 的分支})print(result[explain])print(result[summary])要点RunnableParallel接收一个字典key 是分支名value 是各分支的 Runnable。所有分支同时执行互相不等待比顺序串行更快。返回结果也是一个字典key 与输入对应。注意它与顺序链的区别顺序链是上一步的输出喂给下一步而RunnableParallel是同一个输入分多路同时处理。11. 案例二评论分析流水线把前面学到的串起来做一个多条评论 → 批量分析的流水线。创建02_review_pipeline.pyfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableLambdafromutils.model_factoryimportget_model# 1) 单条评论分析链promptChatPromptTemplate.from_template(你是电商客服分析师。请分析这条评论的情感倾向正面/负面/中性并一句话说明原因。评论{review})modelget_model()parserStrOutputParser()analyze_chainprompt|model|parser# 2) 用 RunnableLambda 包裹一个加前缀的后处理函数defadd_prefix(text:str)-str:return【分析结果】text# 3) 组装完整流水线pipelineanalyze_chain|RunnableLambda(add_prefix)# 4) 用 batch 批量处理多条评论reviews[物流很快包装也很用心好评,客服半天不回消息体验太差了。,东西还行没什么特别的感觉。,]resultspipeline.batch(reviews)forrinresults:print(r)运行python 02_review_pipeline.py这条流水线体现了本章核心prompt | model | parser组成基础链 →RunnableLambda做后处理 →batch完成批量分析。12. 本章重点Runnable是 LangChain 中所有可执行的组件Prompt、Model、Parser、Retriever 都是。LCEL用|把多个 Runnable 串成chain数据按箭头方向流动。invoke处理单条batch处理多条stream流式输出三者自动可用。RunnableLambda把普通函数变成 Runnable 接进链。RunnableParallel对同一输入并行执行多个分支。只要最后一步是字符串解析器|链就自动拥有invoke/batch/stream能力。13. 常见问题Q为什么我写了chain prompt | model | parser但chain.stream()没效果A检查链的最后一步是不是字符串类解析器。如果最后一步是 Pydantic 解析器返回对象stream没法逐字吐文本改用invoke/batch。Q普通函数能直接|接到链后面吗A不能会报错。必须用RunnableLambda(func)包一层把它变成 Runnable。Qbatch和for循环里反复invoke有区别吗A功能结果一致但batch内部会做并发调度处理大量数据时更高效、代码也更简洁。QRunnableParallel和顺序链有什么区别A顺序链是上一步输出喂给下一步数据串行流动RunnableParallel是同一个输入分多路同时处理各分支并行、互不依赖。Q链里每一步的输入输出类型对不上怎么办ALCEL 要求前一步的输出能匹配下一步的输入。常见做法是中间用RunnableLambda做字段映射/格式转换确保类型和结构衔接正确。下一章预告学完基础链后我们会进入RAG检索增强生成与Agent智能体看看如何让链去查资料和调用工具。