langchain学习总结（1）LCEL

通信HotdOg

46人浏览 · 2026-04-20 11:24:24

通信HotdOg · 2026-04-20 11:24:24 发布

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

（1）基础链
最经典的写法就是“三件套”：

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

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

完整示例：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
# 1. 定义组件
prompt = ChatPromptTemplate.from_template("用一句话描述{topic}的魅力。")  ###prompt写法
model = ChatOpenAI(model="gpt-4o-mini")####这里一般现在不这么调用，全部都是KEY+API网址。见下：
#######################
API_KEY = "sk-xxxxxx"   # ⚠️ 请替换为你自己的有效 key
BASE_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": lambda x: x} | step2 ###{"key_words": lambda x: x}  这个代码的含义就是把step1的输出打包塞到key_words里。
# 调用
seq_chain.invoke({"question": "缅因猫掉毛怎么办"})

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

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

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

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

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

# 1. 定义条件函数：判断是不是动物问题
def is_animal_question(input_dict):
    return "猫" in input_dict["question"] or "狗" in input_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）数据转换链（插入自己的函数）

from langchain_core.runnables import RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
# 1. 大模型
API_KEY = "sk-xxxxxx"   # ⚠️ 请替换为你自己的有效 key
BASE_URL = "https://api.deepseek.com"
#### 模型写法
model = ChatOpenAI(
    model="deepseek-chat",
    openai_api_key=API_KEY,
    openai_api_base=BASE_URL,
    temperature=0.7
)
# 2. 你的自定义函数（字典进→字典出）
def my_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. 导入所有核心依赖
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import (
    RunnableLambda,
    RunnableParallel,
    RunnableBranch,
    RunnablePassthrough
)

# 2. 初始化大模型
API_KEY = "sk-xxxxxx"   # ⚠️ 请替换为你自己的有效 key
BASE_URL = "https://api.deepseek.com"
#### 模型写法
model = ChatOpenAI(
    model="deepseek-chat",
    openai_api_key=API_KEY,
    openai_api_base=BASE_URL,
    temperature=0.7
)
# ==============================================
# 【模块1】自定义数据转换链（清洗用户废话）
# ==============================================
def clean_user_question(input_dict: dict) -> dict:
    """清洗用户输入的废话，提纯问题"""
    question = input_dict["question"]
    # 过滤冗余词
    useless_words = ["哎", "啊", "请问", "我想知道", "谢谢"]
    for word in useless_words:
        question = question.replace(word, "")
    return {"clean_question": question.strip()}

# 包装成LCEL组件
clean_chain = RunnableLambda(clean_user_question)
# ==============================================
# 【模块2】分支链：判断问题类型，走不同逻辑
# ==============================================
# 分支1：判断是否是宠物问题
def is_pet_question(input_dict: dict) -> bool:
    return "猫" in input_dict["clean_question"] or "狗" in input_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": lambda x: "宠物护理通用知识：掉毛可通过梳理、饮食改善"
})
# ==============================================
# 【终极组合链】把所有模块串起来（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)