Part 5: LCEL（LangChain Expression Language）
文章目录
- Part 5: LCEL（LangChain Expression Language）
- 5.1 LCEL 概述
- 5.2 Runnable 接口详解
- 所有方法表格
- 完整 Demo
- with_retry 和 with_fallbacks
- 5.3 管道操作符 |
- 数据流动图
- 5.4 RunnablePassthrough
- 5.5 RunnableParallel
- 5.6 RunnableLambda
- 5.7 RunnableBranch
- 5.8 配置与运行时修改
- 5.9 错误处理
- 5.10 LCEL 最佳实践
5.1 LCEL 概述
LCEL 是 LangChain 的核心创新，用声明式语法组合各种组件。
类比：LCEL 就像 Linux 管道
cat file | grep "error" | sort，在 LCEL 中：prompt | model | parser。
5.2 Runnable 接口详解
所有方法表格
| 方法 | 说明 | 同步/异步 |
|---|---|---|
invoke(input) | 单次调用 | 同步 |
batch(inputs) | 批量调用 | 同步 |
stream(input) | 流式调用 | 同步 |
ainvoke(input) | 单次调用 | 异步 |
abatch(inputs) | 批量调用 | 异步 |
astream(input) | 流式调用 | 异步 |
pick(keys) | 选择特定字段 | - |
assign(mapping) | 添加字段 | - |
map() | 对列表元素逐一应用 | - |
with_fallbacks(fallbacks) | 设置备用方案 | - |
with_retry() | 设置重试策略 | - |
with_config(config) | 设置运行时配置 | - |
完整 Demo
importasynciofromdotenvimportload_dotenv load_dotenv()fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnablePassthrough,RunnableParallel prompt=ChatPromptTemplate.from_template("用一句话解释{concept}")model=ChatOpenAI(model="gpt-4o-mini",temperature=0.7)chain=prompt|model|StrOutputParser()# 1. invokeresult=chain.invoke({"concept":"量子计算"})print(result)# 2. batchresults=chain.batch([{"concept":"机器学习"},{"concept":"深度学习"}])forrinresults:print(r)# 3. streamforchunkinchain.stream({"concept":"AI"}):print(chunk,end="",flush=True)print()# 4-6. 异步方法asyncdefasync_demo():result=awaitchain.ainvoke({"concept":"区块链"})print(result)results=awaitchain.abatch([{"concept":"云计算"}])asyncforchunkinchain.astream({"concept":"物联网"}):print(chunk,end="",flush=True)print()asyncio.run(async_demo())# 7. pickparallel=RunnableParallel(concept=RunnablePassthrough(),explanation=chain,)picked=parallel.pick("explanation")print(picked.invoke("Python"))# 8. assignassigned=parallel.assign(length=lambdax:len(x["explanation"]))result=assigned.invoke("Python")print(f"长度:{result['length']}")# 9. mapsimple=ChatPromptTemplate.from_template("用2个字概括{c}")|model|StrOutputParser()mapped=simple.map()results=mapped.invoke(["Python","Java","Go"])with_retry 和 with_fallbacks
fromdotenvimportload_dotenv load_dotenv()fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser chain=ChatPromptTemplate.from_template("解释{concept}")|ChatOpenAI(model="gpt-4o-mini")|StrOutputParser()# with_retry：自动重试reliable=chain.with_retry(stop_after_attempt=3,wait_exponential_multiplier=1,wait_exponential_max=10,)# with_fallbacks：备用方案backup=ChatPromptTemplate.from_template("简单解释{concept}")|ChatOpenAI(model="gpt-4o-mini",temperature=0.0)|StrOutputParser()fault_tolerant=chain.with_fallbacks([backup])result=fault_tolerant.invoke({"concept":"Python"})# with_config：运行时配置configured=chain.with_config(run_name="my_chain",tags=["tutorial"],metadata={"version":"1.0"},)5.3 管道操作符 |
数据流动图
fromdotenvimportload_dotenv load_dotenv()fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser prompt=ChatPromptTemplate.from_template("用一句话解释{concept}")model=ChatOpenAI(model="gpt-4o-mini",temperature=0.7)parser=StrOutputParser()# 管道组合chain=prompt|model|parser result=chain.invoke({"concept":"量子纠缠"})print(result)# 验证数据流step1=prompt.invoke({"concept":"量子纠缠"})# ChatPromptValuestep2=model.invoke(step1)# AIMessagestep3=parser.invoke(step2)# str5.4 RunnablePassthrough
fromlangchain_core.runnablesimportRunnablePassthrough,RunnableParallelfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser model=ChatOpenAI(model="gpt-4o-mini")# 基本用法：透传输入pt=RunnablePassthrough()print(pt.invoke("你好"))# "你好"# assign：添加字段enriched=RunnablePassthrough.assign(greeting=lambdax:f"你好，{x['name']}！",is_adult=lambdax:x["age"]>=18,)result=enriched.invoke({"name":"张三","age":25})print(result)# {'name': '张三', 'age': 25, 'greeting': '你好，张三！', 'is_adult': True}# 在 RunnableParallel 中保留原始输入analysis=ChatPromptTemplate.from_template("分析情感：{text}")|model|StrOutputParser()parallel=RunnableParallel(original=RunnablePassthrough(),sentiment=analysis,)result=parallel.invoke("今天天气真好！")print(f"原文:{result['original']}")print(f"情感:{result['sentiment']}")5.5 RunnableParallel
fromlangchain_core.runnablesimportRunnableParallelfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser model=ChatOpenAI(model="gpt-4o-mini")analysis=RunnableParallel(summary=ChatPromptTemplate.from_template("总结：{text}")|model|StrOutputParser(),sentiment=ChatPromptTemplate.from_template("情感分析：{text}")|model|StrOutputParser(),keywords=ChatPromptTemplate.from_template("提取3个关键词：{text}")|model|StrOutputParser(),)result=analysis.invoke({"text":"新版本性能提升50%，但安装有些复杂。"})print(f"摘要:{result['summary']}")print(f"情感:{result['sentiment']}")print(f"关键词:{result['keywords']}")# 多语言翻译parallel=RunnableParallel(chinese=ChatPromptTemplate.from_template("翻译成中文：{text}")|model|StrOutputParser(),japanese=ChatPromptTemplate.from_template("翻译成日文：{text}")|model|StrOutputParser(),)result=parallel.invoke({"text":"Hello, World!"})5.6 RunnableLambda
fromlangchain_core.runnablesimportRunnableLambdafromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser# 包装自定义函数defword_count(text:str)->int:returnlen(text)defadd_prefix(text:str)->str:returnf"[AI助手]{text}"count_runnable=RunnableLambda(word_count)print(count_runnable.invoke("Hello World"))# 11# 在链中使用model=ChatOpenAI(model="gpt-4o-mini")chain=(ChatPromptTemplate.from_template("解释{concept}")|model|StrOutputParser()|RunnableLambda(add_prefix)# 给输出添加前缀)result=chain.invoke({"concept":"Python"})print(result)# [AI助手] Python 是一种...5.7 RunnableBranch
fromlangchain_core.runnablesimportRunnableBranchfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser model=ChatOpenAI(model="gpt-4o-mini")short_chain=ChatPromptTemplate.from_template("简短回答：{text}")|model|StrOutputParser()long_chain=ChatPromptTemplate.from_template("详细回答：{text}")|model|StrOutputParser()branch=RunnableBranch((lambdax:len(x)<20,short_chain),# 短文本long_chain,# 默认（长文本）)print(branch.invoke("什么是AI？"))# 简短回答print(branch.invoke("请详细解释人工智能的发展历史、技术分支和应用场景")[:100])# 详细回答5.8 配置与运行时修改
fromlangchain_core.runnablesimportConfigurableFieldfromlangchain_openaiimportChatOpenAI# ConfigurableField：运行时动态修改参数model=ChatOpenAI(model="gpt-4o-mini",temperature=ConfigurableField(id="temperature",name="Temperature",description="控制随机性",),)result=model.invoke("你好")# 使用默认 temperatureresult=model.invoke("你好",config={"configurable":{"temperature":0.0}})# 运行时修改5.9 错误处理
fromdotenvimportload_dotenv load_dotenv()fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser chain=ChatPromptTemplate.from_template("解释{concept}")|ChatOpenAI(model="gpt-4o-mini")|StrOutputParser()# with_retry 参数reliable=chain.with_retry(stop_after_attempt=3,# 最大尝试次数retry_if_exception_type=(Exception,),# 重试的异常类型wait_exponential_multiplier=1,# 初始等待 1 秒wait_exponential_max=10,# 最大等待 10 秒)# with_fallbacks 参数backup=ChatPromptTemplate.from_template("简单解释{concept}")|ChatOpenAI(model="gpt-4o-mini",temperature=0.0)|StrOutputParser()fault_tolerant=chain.with_fallbacks([backup],# 备用链列表exceptions_to_handle=(Exception,),# 触发 fallback 的异常)5.10 LCEL 最佳实践
- 优先使用 LCEL：所有新代码都应该使用 LCEL 的管道语法
- 保持链简洁：每个链只做一件事，复杂逻辑通过组合实现
- 善用 RunnableParallel：独立的操作应该并行执行
- 添加错误处理：生产环境必须配置 retry 和 fallback
- 使用 with_config：添加 run_name 和 tags 便于调试和监控
- 类型标注：使用
with_types()标注输入输出类型
