LangChain 入门实战(五):掌握LCEL高效AI流水线开发

一、前言为什么一定要学 LCEL在前面的学习中我们已经掌握了 LangChain 三大基础组件PromptTemplate提示词模板ChatModel大模型调用OutputParser结果解析器但在之前的写法中所有组件都是分步、独立调用代码非常冗余# 旧的分步写法繁琐、难复用 prompt prompt_template.invoke(data) response model.invoke(prompt) result parser.invoke(response)如果链路变长加检索、加预处理、加分支、加并行代码会变得极其混乱、难以维护、无法复用。所以 LangChain 推出了 LCEL专门用来组件组装、链路编排、标准化开发 AI 流水线。学完本章你将彻底掌握理解 Runnable 统一可运行组件思想掌握 LCEL 管道语法 | 串联任意组件掌握三大调用模式invoke / batch / stream掌握函数包装RunnableLambda掌握并行任务RunnableParallel掌握数据透传与字段追加RunnablePassthrough能独立搭建完整 AI 流水线清洗→Prompt→模型→解析→多分支并行二、核心概念1什么是 RunnableRunnable 是 LangChain 所有可执行组件的统一父类/统一规范。只要是 Runnable就具备两个能力可以用 | 管道符串联组合自带一套统一调用方法我们学过的所有组件全部都是 RunnablePrompt 模板ChatModel / LLM 模型OutputParser 解析器Retriever 检索器自定义 RunnableLambda 函数所有 Runnable 共享统一方法体系重点掌握前3个方法作用适用场景invoke单次同步调用单条问答、单次生成batch批量多条同步调用批量文案、批量分类、批量提取stream流式逐段返回聊天对话、实时输出ainvoke / abatch异步调用高并发服务开发三、核心概念2什么是 LCELLCEL 全称LangChain Expression LanguageLangChain 表达式语言。一句话定义LCEL 是 LangChain 官方的链式组装语法通过 | 管道运算符将所有 Runnable 组件串联成一条完整执行链路Chain。数据流动规则从上一个 Runnable 的输出自动流入下一个 Runnable 的输入。经典标准链路chain prompt_template | model | parser数据流输入字典 → Prompt渲染 → 模型推理 → 结果解析 → 最终字符串LCEL 相比分步写法的核心优势代码极简、结构清晰、可读性极强一条链自动支持 invoke / batch / stream无需重复编写组件可任意插拔、复用、组合天然支持并行、透传、字段扩展适合复杂业务流水线四、基础实战1最简 LCEL 链路技术概念解释实现功能输入概念学习者水平AI 通俗解释技术名词。import os from dotenv import load_dotenv from langchain.chat_models import init_chat_model from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate load_dotenv() 1. 初始化模型 model init_chat_model( modeldeepseek-v4-flash, model_provideropenai, base_urlos.getenv(DEEPSEEK_BASE_URL), api_keyos.getenv(DEEPSEEK_API_KEY), temperature0.5 ) 2. 定义提示词模板 template ChatPromptTemplate.from_messages([ (system, 你是一名编程讲师擅长用简单语言解释技术概念。), (human, 请解释下面的技术概念 概念{topic} 学习者水平{level} 要求 先给一句话定义 再给一个简单例子 不超过 200 字) ]) 3. 定义输出解析器 parser StrOutputParser() 4. LCEL 组装链路 chain template | model | parser 5. 单次调用 result chain.invoke({ topic: 什么是RAG?, level: 编程小白 }) print(result)五、三大核心调用方式详解1. invoke 单次调用最常用用于单条任务处理单次问答、单次文案生成、单次抽取。result chain.invoke({ topic: Embedding, level: 初学者 })2. batch 批量调用一次性传入多条数据列表批量执行相同逻辑。适用批量商品卖点、批量评论分析、批量分类。question_list [ {topic: 什么是RAG?,level: 编程小白}, {topic: 什么是向量数据库?,level: 有langchain基础}, {topic: 什么是MCP服务?,level: 会java和python代码} ] results chain.batch(question_list) for res in results: print(res) print(- * 50)可以通过config{max_concurrency: 2}限制并发数防止请求超限。3. stream 流式输出边生成、边返回、边打印模拟 ChatGPT 逐字输出效果。适合对话问答、长文本生成、学习建议、报告生成。data {topic: 什么是RAG?,level: 编程小白} for chunk in chain.stream(data): print(chunk, end, flushTrue)关键点flushTrue强制刷新缓冲区否则文字会堆积不实时输出。stream 是单次流式不用于批量场景。六、高阶1RunnableLambda 包装普通函数默认只有 LangChain 组件可以进管道。如果想把自己写的普通 Python 函数加入 Chain需要用 RunnableLambda 包装。常用场景输入清洗、数据预处理、自定义逻辑加工。案例自动清洗用户输入去空格from langchain_core.runnables import RunnableLambda from langchain_core.output_parsers import StrOutputParser from utils.model_factory import get_deepSeek_model from utils.prompt_template import getPromptTemplate model get_deepSeek_model() template getPromptTemplate( 你是一名 Python 讲师请简洁回答学生问题。, 学生的问题是:{question} ) parser StrOutputParser() 自定义预处理函数 def clean_input(qdict): # 去除首尾空格清洗脏数据 question qdict.get(wenti).strip() return {question: question} 包装成 Runnable可加入管道 clean_chain RunnableLambda(clean_input) 完整链路清洗 - 提示词 - 模型 - 解析 chain clean_chain | template | model | parser result chain.invoke({ wenti: 什么是闭包 }) print(result)核心价值把业务逻辑和AI链路解耦所有预处理统一放进链路上。七、高阶2RunnableParallel 并行多任务RunnableParallel 作用同一个输入同时分给多条子链并行执行。不是串行执行是并发执行。总耗时 ≈ 最慢的那条子链耗时不会累加。经典场景一条评论同时做情感分析、关键词提取、客服回复。案例完整评论分析流水线from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableParallel from utils.model_factory import get_deepSeek_model from utils.prompt_template import getPromptTemplate model get_deepSeek_model() parser StrOutputParser() 三条独立子链 chain_sentiment getPromptTemplate( 你是评论情感助手只回答正面、中性、负面。, 评价{review} ) | model | parser chain_keyword getPromptTemplate( 提取3个关键词顿号分隔。, 评价{review} ) | model | parser chain_reply getPromptTemplate( 你是电商客服80字以内礼貌回复。, 评价{review} ) | model | parser 并行组装 all_chain RunnableParallel( sentimentchain_sentiment, keywordschain_keyword, replychain_reply ) 一次输入同时得到三个结果 result all_chain.invoke({ review: 鼠标手感不错也很安静但是滚轮用了两周就有异响。 }) print(情感, result[sentiment]) print(关键词, result[keywords]) print(客服回复, result[reply])输出为字典结构多任务结果互不干扰、同时产出。八、高阶3RunnablePassthrough 数据透传与字段追加这是 RAG 项目最核心、面试最高频的知识点。两种