LCEL快速入门与实战

体验课时间有限，若想深度学习大模型技术，欢迎大家报名由我主讲的《2025大模型Agent智能体开发实战》：

《2025大模型Agent智能体开发实战》为【100+小时】体系大课，总共20大模块精讲精析，零基础直达大模型企业级应用！

《2025大模型Agent智能体开发实战》2025 年新春班上新特惠，早鸟价入学，直降2000，木羽老师直播间专属叠加优惠券1000，仅需2999报名即学，【仅限前10名】详细信息扫码添加助教，回复“大模型”，即可领取课程大纲&查看课程详情👇

《2025大模型智能体Agent开发实战》体验课

LangChain AI+ DeepSeek v3 企业 Agent 开发实战

Part 1. LCEL (LangChain Expression Language) 快速入门与实战

1. LCEL基本概念入门

LCEL(LangChain Expression Language)是 LangChain 中非常关键的概念，它指的是一种声明式的表达式语言，用于通过链式组合不同的模块。它将不同的组件通过统一的Runable接口连接起来，从而实现具体的流程或功能（比如RAG、Agent）。

因为每个模块都是LCEL对象，因此基于LCEL对象连接形成的链，本身也是一个 LCEL 对象，所以当通过一组通用的调用方法（invoke、 batch、stream、 ainvoke ）方法时，就能够做到定制化组合链、并行化组件、回退、动态配置链内部结构等标准化的操作。它的优势非常明显：

统一的接口：每个 LCEL 对象都实现Runnable接口，因此可以非常方便的连接到一起；
模块化操作：每个组件都可以独立开发和测试，处理好输入和输出就可以通过LCEL集成到一起；
良好扩展性：各个模块组件之间都实现了通用的调用方法（invoke、 batch、stream、 ainvoke ）方法，因此可以灵活组合使用不同的使用场景；

比如LangChain中抽象出来的最简单的 Model I/O 模块。

LangChain的Model I/O模块提供了标准的、可扩展的接口实现与大语言模型的外部集成。所谓的Model I/O，包括模型输入（Prompts）、模型输出（OutPuts）和模型本身（Models），简单理解就是通过该模块，我们可以快速与某个大模型进行对话交互，整个内部逻辑就相当于我们最熟悉的这个过程：输入Prompt，得到大模型针对该Prompt的推理结果。如下示例为OpenAI的 GPT 系列模型的API 调用规范：

FENCE0

在LangChain的Model I/O模块设计中，包含三个核心部分： Prompt Template（对应上图中的Format部分）， Model（对应上图中的Predict部分）和Output Parser（对应上图中的Parse部分）。

Format：即指代Prompts Template，通过模板化来管理大模型的输入；
Predict：即指代Models，使用通用接口调用不同的大语言模型；
Parse：即指代Output部分，用来从模型的推理中提取信息，并按照预先设定好的模版来规范化输出。
Format

对于Prompt Template第一部分，传统上我们创建提示词是通过手工编写来实现的，在这个过程中会利用各种提示工程技巧，如Few-Shot、链式推理（CoT）等方法，以提高大模型的推理性能。然而，在应用开发中，一个关键的考量是提示词不能是一成不变的。 其原因在于，应用开发需要适应多变的用户需求和场景。固定的提示词限制了模型的灵活性和适用范围。例如，如果我们正在开发一个天气查询应用，用户可能会以多种方式提出查询，如“今天的天气怎么样？”或“明天纽约的温度是多少度？”。如果提示词是固定的，它可能只能处理一种特定类型的查询，而无法适应这种多样性的需求。

而Prompt Template，就像ReAct模版，将API的使用、问题解答过程等复杂逻辑封装成了一套结构化的格式。我们只需准备具体的外部函数信息和用户查询，即可生成定制化的提示词，引导模型按照既定逻辑进行思考和回答，从而实现外部函数的调用过程，即： FENCE0

因此，引入Prompt Template可以支持变量和动态内容的插入，使得同一个应用可以根据不同的输入动态调整提示词，从而更好地响应用户的具体需求。LangChain通过这种方式来提高应用的通用性和用户体验。

Predict

在Predict部分，实质上是处理模型从接收输入到执行推理的整个过程。考虑到存在两种主要类型的大模型——Base类模型和Chat类模型，LangChain在其Model I/O模块中对这两种模型都进行了抽象，分别归类为LLMs（Large Language Models）和Chat Models。我们还是以OpenAI 的 Completion 和 Chatcompletions方法为例：

FENCE0

LLMs是简化的大语言模型抽象，即基于给定的Prompt提供内容生成的功能。而Chat Models则专注于聊天API的抽象，需要维护上下文的记忆（聊天记录），呈现出更接近对话或聊天形式的交互。

Parse

我们知道，大模型的输出是不稳定的，同样的输入Prompt往往会得到不同形式的输出。在自然语言交互中，不同的语言表达方式通常不会造成理解上的障碍。但在应用开发中，大模型的输出可能是下一步逻辑处理的关键输入。因此，在这种情况下，规范化输出是必须要做的任务，以确保应用能够顺利进行后续的逻辑处理。

输出解析器 Output Parser就是一个帮助结构化语言模型响应的抽象，可以获取格式指令或者进行更深层次的解析。这我们会在后面的实践中直观的体验到。

整体而言，在Model I/O模块的抽象中，其一能够让开发者快速的接入不同的大模型，比如OpenAI、ChatGLM、Qwen等，按照既定规范执行模型推理。其二通过输入和输出的模板化处理，使其更贴合于应用开发的最佳实践。接下来，我们就逐步的介绍上述三个流程在LangChain下是如何进行集成和操作的。

2. DeepSeek v3 模型注册与使用

DeepSeek v3账号注册与API获取

DeepSeek官网：https://www.deepseek.com/

新用户注册即赠送10元额度，约500万token额度（如果发现没有赠送额度，那就是官方停止了活动。）

对比GPT4o价格，约降低90%以上：输入价格为GPT4o的6%，输出价格为GPT4o的3%。

而且其API调用不限速：

最关键的是，调用风格和OpenAI完全一致：Function calling、Json Output等功能完全相同：

3. 基于LCEL 实现 DeepSeep v3 的集成

LangChain中最基本和常见的用例是将提示模板和模型链接在一起。为了看看这是如何工作的，我们创建一个接受某个主题并生成响应的链：

# ! pip install -qU langchain langchain-openai

from langchain_openai import ChatOpenAI

# 使用DeepSeek-chat模型作为chatmodel
model = ChatOpenAI(model="deepseek-chat",
                   api_key='sk-6c960cb2a7',
                   base_url='https://api.deepseek.com')

使用 LCEL 将这些不同的组件拼凑成一个链。其中 | 符号类似于unix管道运算符，它将不同的组件链接在一起，将一个组件的输出作为下一个组件的输入。

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("请你介绍一下什么是 {topic}")

chain = prompt | model | StrOutputParser()

在此链中，用户输入传递到提示模板，然后提示模板输出传递到模型，然后模型输出传递到输出解析器。

chain.invoke({"topic": "人工智能"})

在这个过程中，Runnable 会动态生成输入的描述，显示的将其传入Pydantic 模型。我们可以对其调用.schema()来获取 JSONSchema 表示。

chain.input_schema.schema()

prompt.input_schema.schema()

prompt.output_schema.schema()

model.input_schema.schema()

流式输出

for chunk in chain.stream("大模型技术"):
    print(chunk, end="", flush=True)

整个过程其实并不难理解，在LCEL中，所有组件都实现了runable接口，而runable的接口协议会将每个组件的输入和输出做一个描述，传入pydantic模型做一个强检验，如果一切正常则会正常传递数据，否则程序则抛出异常。

3. 实现复杂RAG聊天机器人

接下来，我们进一步探讨 LangChain 和 DeepSeek v3模型如何构建一个复杂的 RAG 聊天机器人，能够处理复杂的查询，并且可以通过聊天历史记录维护上下文，并使用 LangChain 的 LCEL语法遵守严格的Guardrails（护栏）。

Guardrails(护栏)对于确保AI系统的安全性和可靠性是比较重要的。通过设定明确的界限，我们可以防止大模型生成有害或误导性的内容。拒绝机制使机器人能够礼貌地拒绝违反这些护栏的请求，例如与敏感主题或非法活动相关的请求。

这里我们创建一个智能HR聊天机器人助手，该机器人将能够利用私有知识库回答有关公司政策、程序和福利的问题。

使用DeepSeek v3模型作为对话模型。

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="deepseek-chat",
                   api_key='sk-6c960cb2a3a24fa6b92f6993d5942777',
                   base_url='https://api.deepseek.com')

使用OpenAI的Embeddings模型将自然语言转化成词向量的表示。

from langchain_openai import OpenAIEmbeddings

embed = OpenAIEmbeddings(
    api_key='sk-proj-Vs1hvbqAjlKBnTS9yTrNygLSq16ElNdDuTV1ADDgfezt3b4B5oA',
    base_url='https://ai.devtool.tech/proxy/v1',     # 国内中转地址
    model="text-embedding-3-large" 
)

# ! pip install faiss-cpu

接下来，我们使用FAISS作为矢量数据库。 FAISS 是 Facebook AI Research 开发的一个库，用于高效相似性搜索和密集向量聚类。LangChain在第三方集成模块（Langchain_community）中已经接入了FAISS向量数据库，所以我们就可以直接使用。

from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document


# 加载一些模拟的假数据
doc1 = Document(page_content="员工每年享有一定数量的病假。有关资格和具体细节可以在员工手册中找到。")
doc2 = Document(page_content="员工请病假时，必须首先通知其主管关于病情和预计缺勤时间。员工需填写病假申请表，并提交给人力资源部门或主管。")
doc3 = Document(page_content="病假申请表可以在公司内部网找到。表格需要填写员工姓名、部门、缺勤日期和缺勤原因等信息。")


# 创建 Faiss 向量存储
vector_store = FAISS.from_documents([doc1, doc2, doc3], embed)

# 将文件保存到本地，包括：向量数据、索引文件和元数据文件
vector_store.save_local(folder_path='.')

创建矢量数据库后，我们可以进行测试：

# 加载本地的Faiss向量文件，allow_dangerous_deserialization 用于控制是否允许在加载向量存储时进行潜在的危险反序列化操作。
vector_store = FAISS.load_local(embeddings=embed, folder_path='.',allow_dangerous_deserialization=True)
 
# 将 FAISS 向量存储转换为一个 retriever（检索器），并为该检索器设置一些搜索相关的参数。k=1 表示检索时返回 最相似的 1 个文档
retriever = vector_store.as_retriever(search_kwargs={'k': 1})

# 执行相似度搜素
query = "请问我们公司有没有病假？"
results = retriever.invoke(query)

for doc in results:
    print(f"Content: {doc.page_content}")

我们从一个最简单的链开始，只接受用户问题，在提示中格式化它并输出该问题的答案（不检索）。这里使用 Langchain 的PromptTemplate并使用LCEL对其进行管道传输。

from langchain.prompts import PromptTemplate
from langchain.schema.output_parser import StrOutputParser

# 定义提示模板
prompt = PromptTemplate(
  input_variables = ["question"],
  template = "你是一个乐于助人的智能小助理。擅长根据用户输入的问题给出一个简短的回答：: {question}"
)


# 构建Chains
chain = (
  prompt
  | model
  | StrOutputParser()
)
print(chain.invoke({"question": "请问什么是人工智能？"}))

在这个过程中，会将带有question键的字典被传递到提示模板中，其中question值被提取并在模板中格式化，然后作为输入传递到model，最后将结果提取为使用StrOutputPaser()最终输出字符串。

接下来，因为最终我们想要构建一个聊天机器人，所以需要让它支持聊天历史记录，作为RAG系统的一个基础组件。当调用链时，以列表的形式传递历史记录，指定每条消息是由用户还是助手发送的。例如：

FENCE0

然后创建链组件，将此输入转换为传递给prompt_with_history的输入。与上面的代码类似，但在这里我们需要创建一个 RunnableLambda，它用来获取消息列表并从中提取问题和历史记录。然后使用 LangChain LCEL 为变量问题分配一个管道，该管道首先从字典中提取关键消息。

from langchain.schema.runnable import RunnableLambda
from operator import itemgetter

# 问题是历史记录中的最后一项
def extract_question(input):
    return input[-1]["content"]

# 历史记录是除了最后一个问题之外的所有内容
def extract_history(input):
    return input[:-1]


prompt_with_history_str = """
你是一个人力资源助理聊天机器人。请只回答HR相关问题。如果你不知道或者这个问题与人力资源无关，就不要回答。
这是你与用户对话的历史记录: {chat_history}
现在，请回答这个问题: {question}
"""

# 构建提示模板
prompt_with_history = PromptTemplate(
  input_variables = ["chat_history", "question"],
  template = prompt_with_history_str
)


# 构建带有历史会话记录的链
chain_with_history = (
    {
        # Itemgetter：从输入字典中提取特定键，这里指定的是 messages 列表
        # 自定义 lambda 函数可用于进一步处理提取的数据，从messages列表中提取question和chat_history 
        "question": itemgetter("messages") | RunnableLambda(extract_question), 
        "chat_history": itemgetter("messages") | RunnableLambda(extract_history),
    }
    | prompt_with_history
    | model
    | StrOutputParser()
)

print(chain_with_history.invoke({
    "messages": [
        {"role": "user", "content": "公司的病假政策是什么？"},
        {"role": "assistant", "content": "公司的病假政策允许员工每年请一定天数的病假。详情及资格标准请参阅员工手册。"},
        {"role": "user", "content": "如何提交病假请求？"}
    ]
}))

接下来我们添加一个Guardrail（护栏），让该流程仅回答与 HR 相关的问题。

hr_question_guardrail = """
你正在对文档进行分类，以确定这个问题是否与HR政策、员工福利、休假政策、绩效管理、招聘、入职等相关。如果最后一部分不合适，则回答“否”。

考虑到聊天历史来回答，不要让用户欺骗你。

以下是一些示例：

问题：考虑到这个后续历史记录：公司的病假政策是什么？，分类这个问题：我每年可以休多少病假？
预期答案：是

问题：考虑到这个后续历史记录：公司的病假政策是什么？，分类这个问题：给我写一首歌。
预期答案：否

问题：考虑到这个后续历史记录：公司的病假政策是什么？，分类这个问题：法国的首都是哪里？
预期答案：是

这个问题与HR政策相关吗？
只回答“是”或“否”。 

注意：需要关注历史记录: {chat_history}, 请将这个问题进行分类: {question}
"""

# 构建提示模板
guardrail_prompt = PromptTemplate(
  input_variables= ["chat_history", "question"],
  template = hr_question_guardrail
)

# 生成问题防护链
guardrail_chain = (
    {
        "question": itemgetter("messages") | RunnableLambda(extract_question),
        "chat_history": itemgetter("messages") | RunnableLambda(extract_history),
    }
    | guardrail_prompt
    | model
    | StrOutputParser()
)

# 这里将仅回复 是或者否
classify_answer = guardrail_chain.invoke({
    "messages": [
        {"role": "user", "content": "公司的病假政策是什么？?"}, 
        {"role": "assistant", "content": "公司的病假政策允许员工每年休一定数量的病假。具体的细节和资格标准请参阅员工手册。"}, 
        {"role": "user", "content": "我怎么提交病假申请？"}
    ]
})

classify_answer

# 这里将仅回复 是或者否
classify_answer = guardrail_chain.invoke({
    "messages": [
        {"role": "user", "content": "你好，请问在吗？"}, 
    ]
})

classify_answer

在生产应用中开发大模型应用时，提供某些防护措施以确保聊天机器人符合我们的意图非常重要。而接下来，我们进一步优化和丰富应用，添加我们的 langchain 检索器。

from langchain_community.vectorstores import FAISS

def get_retriever():
    # 使用 OpenAI 的嵌入模型初始化嵌入对象
    embed = OpenAIEmbeddings(
        api_key='sk-proj-Vs1hvbqA8CWfyTrNygLSq16ElNdDu-S4-nwJfZsRVVTV1ADDgfezt3b4B5oA',
        base_url='https://ai.devtool.tech/proxy/v1',
        model="text-embedding-3-large"
    )
    
    # 从本地加载 FAISS 向量存储，并且指定嵌入对象
    vector_store = FAISS.load_local(embeddings=embed, folder_path='.',allow_dangerous_deserialization=True)
     
    # 配置文档检索，返回最相关的 1 个文档
    retriever = vector_store.as_retriever(search_kwargs={'k': 1})
    return retriever

# 构建检索器实例
retriever = get_retriever()

# 生成检索链
retrieve_document_chain = (
    itemgetter("messages") 
    | RunnableLambda(extract_question)
    | retriever
)

print(retrieve_document_chain.invoke({"messages": [{"role": "user", "content": "病假的HR政策是什么？"}]}))

最后，我们实现完整的链来连接检索器。

from langchain.schema.runnable import RunnableBranch, RunnablePassthrough

question_with_history_and_context_str = """
你是一个可信赖的 HR 政策助手。你将回答有关员工福利、休假政策、绩效管理、招聘、入职以及其他与 HR 相关的话题。如果你不知道问题的答案，你会诚实地说你不知道。
阅读讨论以获取之前对话的上下文。在聊天讨论中，你被称为“系统”，用户被称为“用户”。

历史记录: {chat_history}

以下是一些可能帮助你回答问题的上下文： {context}

请直接回答，不要重复问题，不要以“问题的答案是”之类的开头，不要在答案前加上“AI”，不要说“这是答案”，不要提及上下文或问题。

根据这个历史和上下文，回答这个问题： {question}
"""

question_with_history_and_context_prompt = PromptTemplate(
  input_variables= ["chat_history", "context", "question"],
  template = question_with_history_and_context_str
)

def format_context(docs):
    return "\n\n".join([d.page_content for d in docs])


# 定义不相关的链
irrelevant_question_chain = (
  RunnableLambda(lambda x: {"result": '我不能回答与 HR 政策无关的问题。'})
)

# 定义相关的链
relevant_question_chain = (
  RunnablePassthrough() 
  |
  {
    "relevant_docs": prompt | model | StrOutputParser() | retriever,
    "chat_history": itemgetter("chat_history"), 
    "question": itemgetter("question")
  }
 |
  {
    "context": itemgetter("relevant_docs") | RunnableLambda(format_context),
    "chat_history": itemgetter("chat_history"), 
    "question": itemgetter("question")
  }
  |
  {
    "prompt": question_with_history_and_context_prompt,         
  }
  |
  {
    "result": itemgetter("prompt") | model | StrOutputParser(),
  }
)


# 定义分支
branch_node = RunnableBranch(
  (lambda x: "是" in x["question_is_relevant"].lower(), relevant_question_chain),
  (lambda x: "否" in x["question_is_relevant"].lower(), irrelevant_question_chain),
  irrelevant_question_chain
)

full_chain = (
  {
    "question_is_relevant": guardrail_chain,
    "question": itemgetter("messages") | RunnableLambda(extract_question),
    "chat_history": itemgetter("messages") | RunnableLambda(extract_history),    
  }
  | branch_node
)

import json

non_relevant_dialog = {
    "messages": [
        {"role": "user", "content": "公司的病假政策是什么？"},
        {"role": "assistant", "content": "公司的病假政策允许员工每年休一定数量的病假。具体的细节和资格标准请参阅员工手册。"},
        {"role": "user", "content": "你好，请你介绍一下你自己呀。"}
    ]
}

print(f'用不相关的问题测试')
response = full_chain.invoke(non_relevant_dialog)

response

dialog = {
    "messages": [
        {"role": "user", "content": "公司的病假政策是什么？"},
        {"role": "assistant", "content": "公司的病假政策允许员工每年休一定数量的病假。具体的细节和资格标准请参阅员工手册。"},
        {"role": "user", "content": "我应该如何提交病假的申请？"}
    ]
}

print(retrieve_document_chain.invoke({"messages": [{"role": "user", "content": "我应该如何提交病假的申请？?"}]}))

print(f'用相关的问题测试')
response = full_chain.invoke(dialog)

response

体验课时间有限，若想深度学习大模型技术，欢迎大家报名由我主讲的《2025大模型Agent智能体开发实战》：

《2025大模型Agent智能体开发实战》为【100+小时】体系大课，总共20大模块精讲精析，零基础直达大模型企业级应用！

《2025大模型智能体Agent开发实战》体验课

LangChain AI+ DeepSeek v3 企业 Agent 开发实战​

Part 1. LCEL (LangChain Expression Language) 快速入门与实战​

1. LCEL基本概念入门

2. DeepSeek v3 模型注册与使用

3. 基于LCEL 实现 DeepSeep v3 的集成

3. 实现复杂RAG聊天机器人

LangChain AI+ DeepSeek v3 企业 Agent 开发实战

Part 1. LCEL (LangChain Expression Language) 快速入门与实战