LangChain 1.3 完全教程：从入门到精通 - Part 5: LCEL（LangChain Expression Language）

Part 5: LCELLangChain Expression Language文章目录Part 5: LCELLangChain Expression Language5.1 LCEL 概述5.2 Runnable 接口详解所有方法表格完整 Demowith_retry 和 with_fallbacks5.3 管道操作符 |数据流动图5.4 RunnablePassthrough5.5 RunnableParallel5.6 RunnableLambda5.7 RunnableBranch5.8 配置与运行时修改5.9 错误处理5.10 LCEL 最佳实践5.1 LCEL 概述LCEL 是 LangChain 的核心创新用声明式语法组合各种组件。类比LCEL 就像 Linux 管道cat file | grep error | sort在 LCEL 中prompt | model | parser。Runnable核心接口统一方法签名可组合组件invoke / batch / streamainvoke / abatch / astreamRunnablePassthroughRunnableLambdaRunnableParallelRunnableBranch5.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)设置运行时配置-完整 Demoimportasynciofromdotenvimportload_dotenv load_dotenv()fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnablePassthrough,RunnableParallel promptChatPromptTemplate.from_template(用一句话解释{concept})modelChatOpenAI(modelgpt-4o-mini,temperature0.7)chainprompt|model|StrOutputParser()# 1. invokeresultchain.invoke({concept:量子计算})print(result)# 2. batchresultschain.batch([{concept:机器学习},{concept:深度学习}])forrinresults:print(r)# 3. streamforchunkinchain.stream({concept:AI}):print(chunk,end,flushTrue)print()# 4-6. 异步方法asyncdefasync_demo():resultawaitchain.ainvoke({concept:区块链})print(result)resultsawaitchain.abatch([{concept:云计算}])asyncforchunkinchain.astream({concept:物联网}):print(chunk,end,flushTrue)print()asyncio.run(async_demo())# 7. pickparallelRunnableParallel(conceptRunnablePassthrough(),explanationchain,)pickedparallel.pick(explanation)print(picked.invoke(Python))# 8. assignassignedparallel.assign(lengthlambdax:len(x[explanation]))resultassigned.invoke(Python)print(f长度:{result[length]})# 9. mapsimpleChatPromptTemplate.from_template(用2个字概括{c})|model|StrOutputParser()mappedsimple.map()resultsmapped.invoke([Python,Java,Go])with_retry 和 with_fallbacksfromdotenvimportload_dotenv load_dotenv()fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser chainChatPromptTemplate.from_template(解释{concept})|ChatOpenAI(modelgpt-4o-mini)|StrOutputParser()# with_retry自动重试reliablechain.with_retry(stop_after_attempt3,wait_exponential_multiplier1,wait_exponential_max10,)# with_fallbacks备用方案backupChatPromptTemplate.from_template(简单解释{concept})|ChatOpenAI(modelgpt-4o-mini,temperature0.0)|StrOutputParser()fault_tolerantchain.with_fallbacks([backup])resultfault_tolerant.invoke({concept:Python})# with_config运行时配置configuredchain.with_config(run_namemy_chain,tags[tutorial],metadata{version:1.0},)5.3 管道操作符 |数据流动图最终结果OutputParserChatModelPromptTemplate用户输入最终结果OutputParserChatModelPromptTemplate用户输入填充模板变量调用 LLM提取文本{topic: AI}解释什么是AIAIMessageAI是...fromdotenvimportload_dotenv load_dotenv()fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser promptChatPromptTemplate.from_template(用一句话解释{concept})modelChatOpenAI(modelgpt-4o-mini,temperature0.7)parserStrOutputParser()# 管道组合chainprompt|model|parser resultchain.invoke({concept:量子纠缠})print(result)# 验证数据流step1prompt.invoke({concept:量子纠缠})# ChatPromptValuestep2model.invoke(step1)# AIMessagestep3parser.invoke(step2)# str5.4 RunnablePassthroughfromlangchain_core.runnablesimportRunnablePassthrough,RunnableParallelfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser modelChatOpenAI(modelgpt-4o-mini)# 基本用法透传输入ptRunnablePassthrough()print(pt.invoke(你好))# 你好# assign添加字段enrichedRunnablePassthrough.assign(greetinglambdax:f你好{x[name]},is_adultlambdax:x[age]18,)resultenriched.invoke({name:张三,age:25})print(result)# {name: 张三, age: 25, greeting: 你好张三, is_adult: True}# 在 RunnableParallel 中保留原始输入analysisChatPromptTemplate.from_template(分析情感{text})|model|StrOutputParser()parallelRunnableParallel(originalRunnablePassthrough(),sentimentanalysis,)resultparallel.invoke(今天天气真好)print(f原文:{result[original]})print(f情感:{result[sentiment]})5.5 RunnableParallelfromlangchain_core.runnablesimportRunnableParallelfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser modelChatOpenAI(modelgpt-4o-mini)analysisRunnableParallel(summaryChatPromptTemplate.from_template(总结{text})|model|StrOutputParser(),sentimentChatPromptTemplate.from_template(情感分析{text})|model|StrOutputParser(),keywordsChatPromptTemplate.from_template(提取3个关键词{text})|model|StrOutputParser(),)resultanalysis.invoke({text:新版本性能提升50%但安装有些复杂。})print(f摘要:{result[summary]})print(f情感:{result[sentiment]})print(f关键词:{result[keywords]})# 多语言翻译parallelRunnableParallel(chineseChatPromptTemplate.from_template(翻译成中文{text})|model|StrOutputParser(),japaneseChatPromptTemplate.from_template(翻译成日文{text})|model|StrOutputParser(),)resultparallel.invoke({text:Hello, World!})5.6 RunnableLambdafromlangchain_core.runnablesimportRunnableLambdafromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser# 包装自定义函数defword_count(text:str)-int:returnlen(text)defadd_prefix(text:str)-str:returnf[AI助手]{text}count_runnableRunnableLambda(word_count)print(count_runnable.invoke(Hello World))# 11# 在链中使用modelChatOpenAI(modelgpt-4o-mini)chain(ChatPromptTemplate.from_template(解释{concept})|model|StrOutputParser()|RunnableLambda(add_prefix)# 给输出添加前缀)resultchain.invoke({concept:Python})print(result)# [AI助手] Python 是一种...5.7 RunnableBranchfromlangchain_core.runnablesimportRunnableBranchfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser modelChatOpenAI(modelgpt-4o-mini)short_chainChatPromptTemplate.from_template(简短回答{text})|model|StrOutputParser()long_chainChatPromptTemplate.from_template(详细回答{text})|model|StrOutputParser()branchRunnableBranch((lambdax:len(x)20,short_chain),# 短文本long_chain,# 默认长文本)print(branch.invoke(什么是AI))# 简短回答print(branch.invoke(请详细解释人工智能的发展历史、技术分支和应用场景)[:100])# 详细回答5.8 配置与运行时修改fromlangchain_core.runnablesimportConfigurableFieldfromlangchain_openaiimportChatOpenAI# ConfigurableField运行时动态修改参数modelChatOpenAI(modelgpt-4o-mini,temperatureConfigurableField(idtemperature,nameTemperature,description控制随机性,),)resultmodel.invoke(你好)# 使用默认 temperatureresultmodel.invoke(你好,config{configurable:{temperature:0.0}})# 运行时修改5.9 错误处理fromdotenvimportload_dotenv load_dotenv()fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser chainChatPromptTemplate.from_template(解释{concept})|ChatOpenAI(modelgpt-4o-mini)|StrOutputParser()# with_retry 参数reliablechain.with_retry(stop_after_attempt3,# 最大尝试次数retry_if_exception_type(Exception,),# 重试的异常类型wait_exponential_multiplier1,# 初始等待 1 秒wait_exponential_max10,# 最大等待 10 秒)# with_fallbacks 参数backupChatPromptTemplate.from_template(简单解释{concept})|ChatOpenAI(modelgpt-4o-mini,temperature0.0)|StrOutputParser()fault_tolerantchain.with_fallbacks([backup],# 备用链列表exceptions_to_handle(Exception,),# 触发 fallback 的异常)5.10 LCEL 最佳实践优先使用 LCEL所有新代码都应该使用 LCEL 的管道语法保持链简洁每个链只做一件事复杂逻辑通过组合实现善用 RunnableParallel独立的操作应该并行执行添加错误处理生产环境必须配置 retry 和 fallback使用 with_config添加 run_name 和 tags 便于调试和监控类型标注使用with_types()标注输入输出类型