langchain学习总结（1）LCEL

使用书籍为《 AI Agent 开发与应用》 我用的是25年4月第一次印刷的书籍，但是里面的代码完全不能用，langchain版本 1.0以上的和书里面的内容几本就不搭噶了。因此我使用各种模型去生成代码，然后找到一个我觉得最好的代码进行记录。同时也把我调试过程中遇到的问题进行积累记录。 1、LCEL （LangChain Expression Language） langchain当中所有的数据传递都是以字典的格式进行传递的。 （1）基础链 最经典的写法就是“三件套”：

# 一个基础链，通常包含提示词、模型和输出解析器chain=prompt|model|StrOutputParser()

然后通过 chain.invoke({“input”: “…”}) 来运行它。

完整示例：

fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_openaiimportChatOpenAI# 1. 定义组件prompt=ChatPromptTemplate.from_template("用一句话描述{topic}的魅力。")###prompt写法model=ChatOpenAI(model="gpt-4o-mini")####这里一般现在不这么调用，全部都是KEY+API网址。见下：#######################API_KEY="sk-xxxxxx"# ⚠️ 请替换为你自己的有效 keyBASE_URL="https://api.deepseek.com"#### 模型写法model=ChatOpenAI(model="deepseek-chat",openai_api_key=API_KEY,openai_api_base=BASE_URL,temperature=0.7)#######################parser=StrOutputParser()# 2. 构建链chain=prompt|model|parser# 3. 运行链result=chain.invoke({"topic":"LangChain"})print(result)

StrOutputParser()的作用是只获取AIMessage当中的content。并且可以处理类似\n这种转义符。

（2）顺序链
语法：A | B | C | D （管道符就是顺序）
多个基础链串起来即可。注意当中的参数传递：

# 第一步：提取问题关键词step1=ChatPromptTemplate.from_template("提取关键词：{question}")|model|StrOutputParser()# 第二步：用关键词回答step2=ChatPromptTemplate.from_template("根据关键词回答：{key_words}")|model|StrOutputParser()# 拼接成顺序链：自动按顺序执行seq_chain=step1|{"key_words":lambdax:x}|step2###{"key_words": lambda x: x} 这个代码的含义就是把step1的输出打包塞到key_words里。# 调用seq_chain.invoke({"question":"缅因猫掉毛怎么办"})

question我理解就是占位符，将“缅因猫掉毛怎么办”传入到step1当中，过LLM处理后解析格式，然后将输出结果传入到key_words，作为step2 的入参。

（3）并行链
语法：RunnableParallel({ key: 子链 })

fromlangchain_core.runnablesimportRunnableParallel# 定义两个并行任务chain1=ChatPromptTemplate.from_template("总结：{question}")|model|StrOutputParser()chain2=ChatPromptTemplate.from_template("提取关键词：{question}")|model|StrOutputParser()# 并行链：同时执行 chain1 和 chain2parallel_chain=RunnableParallel({"总结":chain1,"关键词":chain2})# 调用 → 同时返回两个结果result=parallel_chain.invoke({"question":"缅因猫掉毛怎么办"})# 从字典里拿 chain1 的结果print(result["总结"])print(result["关键词"])

注意，这里的并行，本质上是异步IO。一个线程，单线程并发。

（4）分支链
语法：RunnableBranch( (条件函数, 子链), 默认链 )

# 1. 定义条件函数：判断是不是动物问题defis_animal_question(input_dict):return"猫"ininput_dict["question"]or"狗"ininput_dict["question"]#返回bool值，并且下面的分支链必须以布尔值作为判断。# 2. 定义两个分支链animal_chain=ChatPromptTemplate.from_template("你是宠物专家：{question}")|model normal_chain=ChatPromptTemplate.from_template("正常回答：{question}")|model# 3. 分支链（条件判断）branch_chain=RunnableBranch((is_animal_question,animal_chain),# 满足条件走这里normal_chain# 否则走这里)|StrOutputParser()# 调用branch_chain.invoke({"question":"缅因猫掉毛怎么办"})

（5）数据转换链（插入自己的函数）

fromlangchain_core.runnablesimportRunnableLambdafromlangchain_core.promptsimportChatPromptTemplatefromlangchain_openaiimportChatOpenAIfromlangchain_core.output_parsersimportStrOutputParser# 1. 大模型API_KEY="sk-xxxxxx"# ⚠️ 请替换为你自己的有效 keyBASE_URL="https://api.deepseek.com"#### 模型写法model=ChatOpenAI(model="deepseek-chat",openai_api_key=API_KEY,openai_api_base=BASE_URL,temperature=0.7)# 2. 你的自定义函数（字典进→字典出）defmy_custom_func(input_dict):question=input_dict["question"]clean_question=question.replace("哎","").replace("啊","")return{"clean_question":clean_question}# 3. 包装成LCEL组件data_transform_chain=RunnableLambda(my_custom_func)# 4. 提示词模板（核心！填充字典数据） 两种写法。第二种是带系统提示词的。prompt=ChatPromptTemplate.from_template("回答问题：{clean_question}")# 不带系统提示词prompt=ChatPromptTemplate.from_messages([# -------- 【系统提示词】固定人设/规则 --------("system","你是专业的宠物医生，只回答科学靠谱的内容"),# -------- 【用户提示词】用户的问题 --------("user","{clean_question}")])# 带系统提示词。# 5. LCEL链式拼接（全部用 | 串联）full_chain=data_transform_chain|prompt|model|StrOutputParser()# 6. 调用（输入字典）result=full_chain.invoke({"question":"哎啊缅因猫掉毛怎么办"})print(result)

注意：1、自己定义的函数必须是字典进，字典出。
2、数据流入：question->data_transform_chain->prompt (“回答问题：{clean_question}”)->model->格式转换

（6）工业级别组合链。

# 1. 导入所有核心依赖fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_openaiimportChatOpenAIfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimport(RunnableLambda,RunnableParallel,RunnableBranch,RunnablePassthrough)# 2. 初始化大模型API_KEY="sk-xxxxxx"# ⚠️ 请替换为你自己的有效 keyBASE_URL="https://api.deepseek.com"#### 模型写法model=ChatOpenAI(model="deepseek-chat",openai_api_key=API_KEY,openai_api_base=BASE_URL,temperature=0.7)# ==============================================# 【模块1】自定义数据转换链（清洗用户废话）# ==============================================defclean_user_question(input_dict:dict)->dict:"""清洗用户输入的废话，提纯问题"""question=input_dict["question"]# 过滤冗余词useless_words=["哎","啊","请问","我想知道","谢谢"]forwordinuseless_words:question=question.replace(word,"")return{"clean_question":question.strip()}# 包装成LCEL组件clean_chain=RunnableLambda(clean_user_question)# ==============================================# 【模块2】分支链：判断问题类型，走不同逻辑# ==============================================# 分支1：判断是否是宠物问题defis_pet_question(input_dict:dict)->bool:return"猫"ininput_dict["clean_question"]or"狗"ininput_dict["clean_question"]# 宠物专家提示词（系统+用户）pet_prompt=ChatPromptTemplate.from_messages([("system","你是专业宠物医生，回答科学、简洁、有耐心"),("user","请回答宠物问题：{clean_question}")])# 通用问答提示词normal_prompt=ChatPromptTemplate.from_messages([("system","你是通用AI助手，礼貌回答所有问题"),("user","请回答问题：{clean_question}")])# 构建分支链branch_chain=RunnableBranch((is_pet_question,pet_prompt),# 宠物问题 → 走专家逻辑normal_prompt# 其他问题 → 走通用逻辑)# ==============================================# 【模块3】并行链：模拟RAG检索上下文（同时传参+检索）# ==============================================parallel_chain=RunnableParallel({"clean_question":RunnablePassthrough(),# 原样传递清洗后的问题# 模拟向量库检索上下文（真实RAG替换为retriever）"context":lambdax:"宠物护理通用知识：掉毛可通过梳理、饮食改善"})# ==============================================# 【终极组合链】把所有模块串起来（LCEL核心）# ==============================================final_chain=(clean_chain# 第一步：清洗数据（自定义函数）|branch_chain# 第二步：分支判断|parallel_chain# 第三步：并行获取问题+上下文|model# 第四步：大模型生成|StrOutputParser()# 第五步：转纯文本)# ==============================================# 测试调用# ==============================================if__name__=="__main__":# 测试1：宠物问题（带废话）print("【测试1：宠物问题】")res1=final_chain.invoke({"question":"哎请问啊，我家缅因猫掉毛怎么办？"})print(res1)print("\n"+"-"*50)# 测试2：非宠物问题print("【测试2：通用问题】")res2=final_chain.invoke({"question":"Python是什么？"})print(res2)

就像搭积木一样简单。