:掌握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：什么是 Runnable？
Runnable 是 LangChain 所有可执行组件的统一父类/统一规范。只要是 Runnable，就具备两个能力：
- 可以用 | 管道符串联组合
- 自带一套统一调用方法
我们学过的所有组件，全部都是 Runnable：
- Prompt 模板
- ChatModel / LLM 模型
- OutputParser 解析器
- Retriever 检索器
- 自定义 RunnableLambda 函数
所有 Runnable 共享统一方法体系（重点掌握前3个）：
| 方法 | 作用 | 适用场景 |
|---|---|---|
| invoke | 单次同步调用 | 单条问答、单次生成 |
| batch | 批量多条同步调用 | 批量文案、批量分类、批量提取 |
| stream | 流式逐段返回 | 聊天对话、实时输出 |
| ainvoke / abatch | 异步调用 | 高并发服务开发 |
三、核心概念2：什么是 LCEL？
LCEL 全称：LangChain Expression Language（LangChain 表达式语言）。
一句话定义：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( model="deepseek-v4-flash", model_provider="openai", base_url=os.getenv("DEEPSEEK_BASE_URL"), api_key=os.getenv("DEEPSEEK_API_KEY"), temperature=0.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="", flush=True)关键点flush=True：强制刷新缓冲区，否则文字会堆积不实时输出。stream 是单次流式，不用于批量场景。
六、高阶1：RunnableLambda 包装普通函数
默认只有 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链路解耦，所有预处理统一放进链路上。
七、高阶2：RunnableParallel 并行多任务
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( sentiment=chain_sentiment, keywords=chain_keyword, reply=chain_reply ) 一次输入，同时得到三个结果 result = all_chain.invoke({ "review": "鼠标手感不错，也很安静，但是滚轮用了两周就有异响。" }) print("情感：", result["sentiment"]) print("关键词：", result["keywords"]) print("客服回复：", result["reply"])输出为字典结构，多任务结果互不干扰、同时产出。
八、高阶3：RunnablePassthrough 数据透传与字段追加
这是 RAG 项目最核心、面试最高频的知识点。两种