LangChain Python版本 LCEL链式调用

LCEL 链式类型

RunableSequence--顺序链 RunableBranch--分支链 RunableSerializable--串行链 RunableParallel--并行链 RunableLambda--函数链

分支链

""" 分支链 在LangChain中提供了类RunnableBranch来完成LCEL中的条件分支判断，它可以根据输入的不同采用不同的处理逻辑， 具体示例如下 会根据用户输入中是否包含英语、韩语等关键词，来选择对应的提示词进行处理。根据判断结果， 再执行不同的逻辑分支 """fromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportChatPromptTemplatefromloguruimportloggerfromlangchain_core.runnablesimportRunnableBranchimportos# 构建提示词english_prompt=ChatPromptTemplate.from_messages([("system","你是一个英语翻译专家，你叫小英"),("human","{query}")])japanese_prompt=ChatPromptTemplate.from_messages([("system","你是一个日语翻译专家，你叫小日"),("human","{query}")])korean_prompt=ChatPromptTemplate.from_messages([("system","你是一个韩语翻译专家，你叫小韩"),("human","{query}")])defdetermine_language(inputs):"""判断语言种类"""query=inputs["query"]if"日语"inquery:return"japanese"elif"韩语"inquery:return"korean"else:return"english"# 初始化模型model=init_chat_model(model="qwen-plus",model_provider="openai",api_key=os.getenv("aliQwen-api"),base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")# 创建字符串输出解析器，用于处理模型输出parser=StrOutputParser()# 创建一个可运行的分支链，根据输入文本的语言类型选择相应的处理流程# 返回值：RunnableBranch对象，可根据输入动态选择执行路径的可运行链chain=RunnableBranch((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:# 判断使用哪个提示词lang=determine_language(query_input)logger.info(f"检测到语言类型:{lang}")# 根据语言类型选择对应的提示词并格式化iflang=="japanese":chatPromptTemplate=japanese_prompteliflang=="korean":chatPromptTemplate=korean_promptelse:chatPromptTemplate=english_prompt#print(query_input) # {'query': '请你用英语翻译这句话:"见到你很高兴"'}# 格式化提示词并打印formatted_messages=chatPromptTemplate.format_messages(**query_input)logger.info("格式化后的提示词:")formsginformatted_messages:logger.info(f"[{msg.type}]:{msg.content}")# 执行链result=chain.invoke(query_input)logger.info(f"输出结果:{result}\n")

函数链

""" RunnableLambda-函数链 将普通Python函数融入Runnable流程. """fromlangchain.chat_modelsimportinit_chat_modelfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.runnablesimportRunnableLambdafromloguruimportloggerimportos model=init_chat_model(model="qwen-plus",model_provider="openai",api_key=os.getenv("aliQwen-api"),temperature=0.0,base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")# 一个简单的打印函数，调试用defdebug_print(x):logger.info(f"中间结果:{x}")return{"input":x}# 子链1提示词prompt1=ChatPromptTemplate.from_messages([("system","你是一个知识渊博的计算机专家，请用中文简短回答"),("human","请简短介绍什么是{topic}")])# 子链1解析器parser1=StrOutputParser()# 子链1：生成内容chain1=prompt1|model|parser1# 子链2提示词prompt2=ChatPromptTemplate.from_messages([("system","你是一个翻译助手，将用户输入内容翻译成英文"),("human","{input}")])# 子链2解析器parser2=StrOutputParser()# 子链2：翻译内容chain2=prompt2|model|parser2# 创建一个可运行的调试节点，用于打印中间结果debug_node=RunnableLambda(debug_print)# 构建完整的处理链，将chain1、调试打印和chain2串联起来full_chain=chain1|debug_print|chain2# 调用复合链result1=full_chain.invoke({"topic":"langchain"})logger.info(f"最终结果111:{result1}")# 构建完整的处理链，将chain1、调试打印和chain2串联起来full_chain=chain1|debug_node|chain2# 调用复合链result2=full_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 model=init_chat_model(model="qwen-plus",model_provider="openai",api_key=os.getenv("aliQwen-api"),base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")# 并行链1提示词prompt1=ChatPromptTemplate.from_messages([("system","你是一个知识渊博的计算机专家，请用中文简短回答"),("human","请简短介绍什么是{topic}")])# 并行链1解析器parser1=StrOutputParser()# 并行链1：生成中文结果chain1=prompt1|model|parser1# 并行链2提示词prompt2=ChatPromptTemplate.from_messages([("system","你是一个知识渊博的计算机专家，请用英文简短回答"),("human","请简短介绍什么是{topic}")])# 并行链2解析器parser2=StrOutputParser()# 并行链2：生成英文结果chain2=prompt2|model|parser2# 创建并行链,用于同时执行多个语言处理链parallel_chain=RunnableParallel({"chinese":chain1,"english":chain2})# 调用复合链result=parallel_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 model=init_chat_model(model="qwen-plus",model_provider="openai",api_key=os.getenv("aliQwen-api"),base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")# 并行链1提示词prompt1=ChatPromptTemplate.from_messages([("system","你是一个知识渊博的计算机专家，请用中文简短回答"),("human","请简短介绍什么是{topic}")])# 并行链1解析器parser1=StrOutputParser()# 并行链1：生成中文结果chain1=prompt1|model|parser1# 并行链2提示词prompt2=ChatPromptTemplate.from_messages([("system","你是一个知识渊博的计算机专家，请用英文简短回答"),("human","请简短介绍什么是{topic}")])# 并行链2解析器parser2=StrOutputParser()# 并行链2：生成英文结果chain2=prompt2|model|parser2# 创建并行链,用于同时执行多个语言处理链parallel_chain=RunnableParallel({"chinese":chain1,"english":chain2})# 调用复合链result=parallel_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 model=init_chat_model(model="qwen-plus",model_provider="openai",api_key=os.getenv("aliQwen-api"),base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")# 子链1提示词prompt1=ChatPromptTemplate.from_messages([("system","你是一个知识渊博的计算机专家，请用中文简短回答"),("human","请简短介绍什么是{topic}")])# 子链1解析器parser1=StrOutputParser()# 子链1：生成内容chain1=prompt1|model|parser1 result1=chain1.invoke({"topic":"langchain"})logger.info(result1)# 子链2提示词prompt2=ChatPromptTemplate.from_messages([("system","你是一个翻译助手，将用户输入内容翻译成英文"),("human","{input}")])# 子链2解析器parser2=StrOutputParser()# 子链2：翻译内容chain2=prompt2|model|parser2# 组合成一个复合 Chain，使用 lambda 函数将chain1执行结果content内容添加input键作为参数传递给chain2full_chain=chain1|(lambdacontent:{"input":content})|chain2# 调用复合链result=full_chain.invoke({"topic":"langchain"})logger.info(result)