LangChain Runnable和LCEL

1.RunnableRunnable是langchain中的抽象基类是langchain中所有链的核心抽象接口旨在为所有的可执行组件提供通用的操作接口用于构建所有的“链”组件代表一个可以调用运行的流程单元使一切可执行组件都有一个统一的调用方式。ABC抽象基类必须被继承不能直接用Generic[Input, Output]泛型规定输入/输出类型class Runnable(ABC, Generic[Input, Output]):Runnable将多个组件按照特定顺序组合起来形成一个可以完成复杂任务的工作流或管道pipeline任何东西只要继承了Runnable就必须实现run()/invoke()方法只要实现了Runnable就可以像函数一样invoke()或使用|管道操作符组合。在langchain中统一调用规范的接口在langchain_core.runnables中主要有15种常见的四种RunnableSequence顺序链RunnableBranch分支链RunnableParallel并行链RunnableLambda函数链1.1 RunnableSequence顺序链按顺序连接多个可运行对象上个对象的输出作为下个对象的输入。import os from langchain.chat_models import init_chat_model from langchain_core.prompts import PromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableSequence prompt_template PromptTemplate( template做一个关于{topic}的小诗, input_variables[topic] ) llm init_chat_model( model deepseek-chat, model_provider openai, api_key os.getenv(DSKEY), base_url https://api.deepseek.com ) parser StrOutputParser() chain RunnableSequence(prompt_template, llm, parser) resp chain.invoke({topic: 霸道总裁爱上做保洁的我}) print(resp)1.2 RunnableBranch分支链使用条件分支判断对列表和默认分支进行初始化例实现一个输入语种把语种名称就地翻译成这个语种对应的外语默认翻译成英语import os from langchain.chat_models import init_chat_model from langchain_core.prompts import PromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableSequence, RunnableBranch english_template PromptTemplate( template直接翻译成英语 {topic}, input_variables[topic] ) japanese_template PromptTemplate( template直接翻译成日语 {topic}, input_variables[topic] ) korean_template PromptTemplate( template直接翻译成韩语 {topic}, input_variables[topic] ) llm init_chat_model( model deepseek-chat, model_provider openai, api_key os.getenv(DSKEY), base_url https://api.deepseek.com ) parser StrOutputParser() chain RunnableBranch( (lambda e: 韩语 e[topic], RunnableSequence(korean_template, llm, parser) ), (lambda e: 日语 e[topic], RunnableSequence(japanese_template, llm, parser)), (RunnableSequence(english_template, llm, parser)) ) resp chain.invoke({topic: 韩语}) print(resp) resp chain.invoke({topic: 日语}) print(resp) resp chain.invoke({topic: 法语}) print(resp)**한국어** (Hangugeo) 「日本語」 French1.3 RunnableParallel并行链创建多个子链并在它们都完成后汇总结果常用于同时问多个问题并聚合结果或者多模型同时工作取最优答案再或者多路径和多模态的处理。import os from langchain.chat_models import init_chat_model from langchain_core.prompts import PromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableSequence, RunnableParallel english_template PromptTemplate( template你是个翻译机器人中译英 {topic}, input_variables[topic] ) japanese_template PromptTemplate( template你是个翻译机器人中译日 {topic}, input_variables[topic] ) korean_template PromptTemplate( template你是个翻译机器人中译韩 {topic}, input_variables[topic] ) llm init_chat_model( model deepseek-chat, model_provider openai, api_key os.getenv(DSKEY), base_url https://api.deepseek.com ) parser StrOutputParser() chain RunnableParallel({ korean: RunnableSequence(korean_template, llm, parser), japanese: RunnableSequence(japanese_template, llm, parser), english: RunnableSequence(english_template, llm, parser), }) resp chain.invoke({topic: 欲穷千里目更上一层楼}) print(resp){korean: 欲窮千里目更上一層樓 \n→ 천 리 밖까지 바라보려면, 한 층 더 높은 곳에 올라가야 한다. \n(또는 시의 느낌을 살려) \n→ 더 넓은 세상을 보려면, 더 높이 올라가야 한다., japanese: 「千里の目を窮めんと欲すれば、更に一層の楼に上る」, english: Here\s the English translation of the Chinese poem:\n\nTo see a thousand miles further, you must ascend one more story.\n\nThis is a famous couplet from the poem *On the Stork Tower* (《登鹳雀楼》) by Wang Zhihuan (王之涣) of the Tang Dynasty.}1.4 RunnableLambda函数链RunnableLambda是一个langchain的包装器将普通python函数融入到Runnable中把普通的函数转换为一个可执行的“链”然后就可以像其他langchain组件一样使用。例from langchain_core.runnables import RunnableLambda, RunnableBranch, RunnableSequence step1 RunnableLambda(lambda x: x.upper()) step2 RunnableLambda(lambda x: x !!!) chain RunnableSequence(step1, step2) res chain.invoke(hello ai) print(res)HELLO AI!!!例在大模型调用中自定义一个RunnableLambda(word_count)用于统计生成内容字数import os from langchain.chat_models import init_chat_model from langchain_core.prompts import PromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda, RunnableSequence prompt_template PromptTemplate( template做一个关于{topic}的小诗, input_variables[topic] ) llm init_chat_model( model deepseek-chat, model_provider openai, api_key os.getenv(DSKEY), base_url https://api.deepseek.com ) parser StrOutputParser() def word_count(text: str) - int: print(----------word_count---------) return len(text) word_counter RunnableLambda(word_count) chain RunnableSequence(prompt_template, llm, parser, word_counter) resp chain.invoke({topic: 霸道总裁爱上做保洁的我}) print(resp)----------word_count--------- 11642.LCEL要理解LCEL首先要了解RunnableRunnablelangchain_core.runnables.base.Runnable是langchain中可以调用批处理流式输出转换和组合的工作单元是实现LCEL的基础通过重写__or__()方法实现了|运算符的重载实现了Runable的类对象之间便可以进行一些类似linux命令中的管道|操作。LCEL全称LangChain Express Language即LangChain表达式语言也是LangChain官方推荐的写法是一种从Runnable而来的声明式方法用于声明组合和执行各种组件模型提示词工具等如果要使用LCEL对应的组件必须实现Runnable使用LCEL创建的Runnable称之为链。例import os from langchain.chat_models import init_chat_model from langchain_core.prompts import PromptTemplate from langchain_core.output_parsers import StrOutputParser prompt_template PromptTemplate( template做一个关于{topic}的小诗, input_variables[topic] ) llm init_chat_model( model deepseek-chat, model_provider openai, api_key os.getenv(DSKEY), base_url https://api.deepseek.com ) parser StrOutputParser() # LCEL重写 chain prompt_template | llm | parser resp chain.invoke({topic: 霸道总裁爱上做保洁的我}) print(resp)还可以自定义一个word_count(text: str) - int函数通过langchain的RunnableLambda对象包装使得函数变为获得链式的执行能力的Runable对象拼入链中统计大模型回复的字数import os from langchain.chat_models import init_chat_model from langchain_core.prompts import PromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda prompt_template PromptTemplate( template做一个关于{topic}的小诗, input_variables[topic] ) llm init_chat_model( model deepseek-chat, model_provider openai, api_key os.getenv(DSKEY), base_url https://api.deepseek.com ) parser StrOutputParser() def word_count(text: str) - int: print(----------word_count---------) return len(text) word_counter RunnableLambda(word_count) # LCEL重写 chain prompt_template | llm | parser | word_counter resp chain.invoke({topic: 霸道总裁爱上做保洁的我}) print(resp)运行----------word_count--------- 232