使用书籍为《 AI Agent 开发与应用》 我用的是25年4月第一次印刷的书籍但是里面的代码完全不能用langchain版本 1.0以上的和书里面的内容几本就不搭噶了。因此我使用各种模型去生成代码然后找到一个我觉得最好的代码进行记录。同时也把我调试过程中遇到的问题进行积累记录。 1、LCEL LangChain Expression Language langchain当中所有的数据传递都是以字典的格式进行传递的。 1基础链 最经典的写法就是“三件套”# 一个基础链通常包含提示词、模型和输出解析器chainprompt|model|StrOutputParser()然后通过 chain.invoke({“input”: “…”}) 来运行它。完整示例fromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_openaiimportChatOpenAI# 1. 定义组件promptChatPromptTemplate.from_template(用一句话描述{topic}的魅力。)###prompt写法modelChatOpenAI(modelgpt-4o-mini)####这里一般现在不这么调用全部都是KEYAPI网址。见下#######################API_KEYsk-xxxxxx# ⚠️ 请替换为你自己的有效 keyBASE_URLhttps://api.deepseek.com#### 模型写法modelChatOpenAI(modeldeepseek-chat,openai_api_keyAPI_KEY,openai_api_baseBASE_URL,temperature0.7)#######################parserStrOutputParser()# 2. 构建链chainprompt|model|parser# 3. 运行链resultchain.invoke({topic:LangChain})print(result)StrOutputParser()的作用是只获取AIMessage当中的content。并且可以处理类似\n这种转义符。2顺序链语法A | B | C | D 管道符就是顺序多个基础链串起来即可。注意当中的参数传递# 第一步提取问题关键词step1ChatPromptTemplate.from_template(提取关键词{question})|model|StrOutputParser()# 第二步用关键词回答step2ChatPromptTemplate.from_template(根据关键词回答{key_words})|model|StrOutputParser()# 拼接成顺序链自动按顺序执行seq_chainstep1|{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# 定义两个并行任务chain1ChatPromptTemplate.from_template(总结{question})|model|StrOutputParser()chain2ChatPromptTemplate.from_template(提取关键词{question})|model|StrOutputParser()# 并行链同时执行 chain1 和 chain2parallel_chainRunnableParallel({总结:chain1,关键词:chain2})# 调用 → 同时返回两个结果resultparallel_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_chainChatPromptTemplate.from_template(你是宠物专家{question})|model normal_chainChatPromptTemplate.from_template(正常回答{question})|model# 3. 分支链条件判断branch_chainRunnableBranch((is_animal_question,animal_chain),# 满足条件走这里normal_chain# 否则走这里)|StrOutputParser()# 调用branch_chain.invoke({question:缅因猫掉毛怎么办})5数据转换链插入自己的函数fromlangchain_core.runnablesimportRunnableLambdafromlangchain_core.promptsimportChatPromptTemplatefromlangchain_openaiimportChatOpenAIfromlangchain_core.output_parsersimportStrOutputParser# 1. 大模型API_KEYsk-xxxxxx# ⚠️ 请替换为你自己的有效 keyBASE_URLhttps://api.deepseek.com#### 模型写法modelChatOpenAI(modeldeepseek-chat,openai_api_keyAPI_KEY,openai_api_baseBASE_URL,temperature0.7)# 2. 你的自定义函数字典进→字典出defmy_custom_func(input_dict):questioninput_dict[question]clean_questionquestion.replace(哎,).replace(啊,)return{clean_question:clean_question}# 3. 包装成LCEL组件data_transform_chainRunnableLambda(my_custom_func)# 4. 提示词模板核心填充字典数据 两种写法。第二种是带系统提示词的。promptChatPromptTemplate.from_template(回答问题{clean_question})# 不带系统提示词promptChatPromptTemplate.from_messages([# -------- 【系统提示词】固定人设/规则 --------(system,你是专业的宠物医生只回答科学靠谱的内容),# -------- 【用户提示词】用户的问题 --------(user,{clean_question})])# 带系统提示词。# 5. LCEL链式拼接全部用 | 串联full_chaindata_transform_chain|prompt|model|StrOutputParser()# 6. 调用输入字典resultfull_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_KEYsk-xxxxxx# ⚠️ 请替换为你自己的有效 keyBASE_URLhttps://api.deepseek.com#### 模型写法modelChatOpenAI(modeldeepseek-chat,openai_api_keyAPI_KEY,openai_api_baseBASE_URL,temperature0.7)# # 【模块1】自定义数据转换链清洗用户废话# defclean_user_question(input_dict:dict)-dict:清洗用户输入的废话提纯问题questioninput_dict[question]# 过滤冗余词useless_words[哎,啊,请问,我想知道,谢谢]forwordinuseless_words:questionquestion.replace(word,)return{clean_question:question.strip()}# 包装成LCEL组件clean_chainRunnableLambda(clean_user_question)# # 【模块2】分支链判断问题类型走不同逻辑# # 分支1判断是否是宠物问题defis_pet_question(input_dict:dict)-bool:return猫ininput_dict[clean_question]or狗ininput_dict[clean_question]# 宠物专家提示词系统用户pet_promptChatPromptTemplate.from_messages([(system,你是专业宠物医生回答科学、简洁、有耐心),(user,请回答宠物问题{clean_question})])# 通用问答提示词normal_promptChatPromptTemplate.from_messages([(system,你是通用AI助手礼貌回答所有问题),(user,请回答问题{clean_question})])# 构建分支链branch_chainRunnableBranch((is_pet_question,pet_prompt),# 宠物问题 → 走专家逻辑normal_prompt# 其他问题 → 走通用逻辑)# # 【模块3】并行链模拟RAG检索上下文同时传参检索# parallel_chainRunnableParallel({clean_question:RunnablePassthrough(),# 原样传递清洗后的问题# 模拟向量库检索上下文真实RAG替换为retrievercontext:lambdax:宠物护理通用知识掉毛可通过梳理、饮食改善})# # 【终极组合链】把所有模块串起来LCEL核心# final_chain(clean_chain# 第一步清洗数据自定义函数|branch_chain# 第二步分支判断|parallel_chain# 第三步并行获取问题上下文|model# 第四步大模型生成|StrOutputParser()# 第五步转纯文本)# # 测试调用# if__name____main__:# 测试1宠物问题带废话print(【测试1宠物问题】)res1final_chain.invoke({question:哎请问啊我家缅因猫掉毛怎么办})print(res1)print(\n-*50)# 测试2非宠物问题print(【测试2通用问题】)res2final_chain.invoke({question:Python是什么})print(res2)就像搭积木一样简单。