LCEL 链式类型RunableSequence--顺序链 RunableBranch--分支链 RunableSerializable--串行链 RunableParallel--并行链 RunableLambda--函数链分支链 分支链 在LangChain中提供了类RunnableBranch来完成LCEL中的条件分支判断它可以根据输入的不同采用不同的处理逻辑 具体示例如下 会根据用户输入中是否包含英语、韩语等关键词来选择对应的提示词进行处理。根据判断结果 再执行不同的逻辑分支 fromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportChatPromptTemplatefromloguruimportloggerfromlangchain_core.runnablesimportRunnableBranchimportos# 构建提示词english_promptChatPromptTemplate.from_messages([(system,你是一个英语翻译专家你叫小英),(human,{query})])japanese_promptChatPromptTemplate.from_messages([(system,你是一个日语翻译专家你叫小日),(human,{query})])korean_promptChatPromptTemplate.from_messages([(system,你是一个韩语翻译专家你叫小韩),(human,{query})])defdetermine_language(inputs):判断语言种类queryinputs[query]if日语inquery:returnjapaneseelif韩语inquery:returnkoreanelse:returnenglish# 初始化模型modelinit_chat_model(modelqwen-plus,model_provideropenai,api_keyos.getenv(aliQwen-api),base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1)# 创建字符串输出解析器用于处理模型输出parserStrOutputParser()# 创建一个可运行的分支链根据输入文本的语言类型选择相应的处理流程# 返回值RunnableBranch对象可根据输入动态选择执行路径的可运行链chainRunnableBranch((lambdax:determine_language(x)japanese,japanese_prompt|model|parser),(lambdax:determine_language(x)korean,korean_prompt|model|parser),(english_prompt|model|parser))# 测试查询test_queries[{query:请你用韩语翻译这句话:见到你很高兴},{query:请你用日语翻译这句话:见到你很高兴},{query:请你用英语翻译这句话:见到你很高兴}]forquery_inputintest_queries:# 判断使用哪个提示词langdetermine_language(query_input)logger.info(f检测到语言类型:{lang})# 根据语言类型选择对应的提示词并格式化iflangjapanese:chatPromptTemplatejapanese_prompteliflangkorean:chatPromptTemplatekorean_promptelse:chatPromptTemplateenglish_prompt#print(query_input) # {query: 请你用英语翻译这句话:见到你很高兴}# 格式化提示词并打印formatted_messageschatPromptTemplate.format_messages(**query_input)logger.info(格式化后的提示词:)formsginformatted_messages:logger.info(f[{msg.type}]:{msg.content})# 执行链resultchain.invoke(query_input)logger.info(f输出结果:{result}\n)函数链 RunnableLambda-函数链 将普通Python函数融入Runnable流程. fromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.runnablesimportRunnableLambdafromloguruimportloggerimportos modelinit_chat_model(modelqwen-plus,model_provideropenai,api_keyos.getenv(aliQwen-api),temperature0.0,base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1)# 一个简单的打印函数调试用defdebug_print(x):logger.info(f中间结果:{x})return{input:x}# 子链1提示词prompt1ChatPromptTemplate.from_messages([(system,你是一个知识渊博的计算机专家请用中文简短回答),(human,请简短介绍什么是{topic})])# 子链1解析器parser1StrOutputParser()# 子链1生成内容chain1prompt1|model|parser1# 子链2提示词prompt2ChatPromptTemplate.from_messages([(system,你是一个翻译助手将用户输入内容翻译成英文),(human,{input})])# 子链2解析器parser2StrOutputParser()# 子链2翻译内容chain2prompt2|model|parser2# 创建一个可运行的调试节点用于打印中间结果debug_nodeRunnableLambda(debug_print)# 构建完整的处理链将chain1、调试打印和chain2串联起来full_chainchain1|debug_print|chain2# 调用复合链result1full_chain.invoke({topic:langchain})logger.info(f最终结果111:{result1})# 构建完整的处理链将chain1、调试打印和chain2串联起来full_chainchain1|debug_node|chain2# 调用复合链result2full_chain.invoke({topic:langchain})logger.info(f最终结果222:{result2})并行链 RunnableParallel-并行链 在 Langchain 中创建并行链Parallel Chains是指同时运行多个子链Chain并在它们都完成后汇总结果。 **作用**同时执行多个 Runnable合并结果 fromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.runnablesimportRunnableParallelfromloguruimportloggerimportos modelinit_chat_model(modelqwen-plus,model_provideropenai,api_keyos.getenv(aliQwen-api),base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1)# 并行链1提示词prompt1ChatPromptTemplate.from_messages([(system,你是一个知识渊博的计算机专家请用中文简短回答),(human,请简短介绍什么是{topic})])# 并行链1解析器parser1StrOutputParser()# 并行链1生成中文结果chain1prompt1|model|parser1# 并行链2提示词prompt2ChatPromptTemplate.from_messages([(system,你是一个知识渊博的计算机专家请用英文简短回答),(human,请简短介绍什么是{topic})])# 并行链2解析器parser2StrOutputParser()# 并行链2生成英文结果chain2prompt2|model|parser2# 创建并行链,用于同时执行多个语言处理链parallel_chainRunnableParallel({chinese:chain1,english:chain2})# 调用复合链resultparallel_chain.invoke({topic:langchain})logger.info(result)# 打印并行链的ASCII图形表示LangGraph提前预告不是本节知识点parallel_chain.get_graph().print_ascii()顺序链 RunnableParallel-并行链 在 Langchain 中创建并行链Parallel Chains是指同时运行多个子链Chain并在它们都完成后汇总结果。 **作用**同时执行多个 Runnable合并结果 fromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.runnablesimportRunnableParallelfromloguruimportloggerimportos modelinit_chat_model(modelqwen-plus,model_provideropenai,api_keyos.getenv(aliQwen-api),base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1)# 并行链1提示词prompt1ChatPromptTemplate.from_messages([(system,你是一个知识渊博的计算机专家请用中文简短回答),(human,请简短介绍什么是{topic})])# 并行链1解析器parser1StrOutputParser()# 并行链1生成中文结果chain1prompt1|model|parser1# 并行链2提示词prompt2ChatPromptTemplate.from_messages([(system,你是一个知识渊博的计算机专家请用英文简短回答),(human,请简短介绍什么是{topic})])# 并行链2解析器parser2StrOutputParser()# 并行链2生成英文结果chain2prompt2|model|parser2# 创建并行链,用于同时执行多个语言处理链parallel_chainRunnableParallel({chinese:chain1,english:chain2})# 调用复合链resultparallel_chain.invoke({topic:langchain})logger.info(result)# 打印并行链的ASCII图形表示LangGraph提前预告不是本节知识点parallel_chain.get_graph().print_ascii()串行链 RunnableSerializable-串行链 子链叠加串行假如我们需要多次调用大模型将多个步骤串联起来实现功能 fromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportChatPromptTemplatefromloguruimportloggerimportos modelinit_chat_model(modelqwen-plus,model_provideropenai,api_keyos.getenv(aliQwen-api),base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1)# 子链1提示词prompt1ChatPromptTemplate.from_messages([(system,你是一个知识渊博的计算机专家请用中文简短回答),(human,请简短介绍什么是{topic})])# 子链1解析器parser1StrOutputParser()# 子链1生成内容chain1prompt1|model|parser1 result1chain1.invoke({topic:langchain})logger.info(result1)# 子链2提示词prompt2ChatPromptTemplate.from_messages([(system,你是一个翻译助手将用户输入内容翻译成英文),(human,{input})])# 子链2解析器parser2StrOutputParser()# 子链2翻译内容chain2prompt2|model|parser2# 组合成一个复合 Chain使用 lambda 函数将chain1执行结果content内容添加input键作为参数传递给chain2full_chainchain1|(lambdacontent:{input:content})|chain2# 调用复合链resultfull_chain.invoke({topic:langchain})logger.info(result)