本文深入探讨了LangChain框架的核心概念，包括LangChain表达式（LCEL）和组件，以及它们在构建RAG应用中的应用。文章详细介绍了Runnable接口及其各种实现，如RunnableSequence、RunnableParallel、RunnablePassthrough、RunnableLambda和RunnableBranch，并提供了代码示例。此外，文章还概述了LangChain的核心组件，如提示词模板、大语言模型、链、记忆功能和Agent，帮助读者全面理解LangChain的运作机制。

💡 **LangChain的核心组成**：LangChain由两个关键概念构成：LangChain表达式（LCEL）和组件。组件是构建RAG应用的基本操作单元，而LCEL则负责串联这些组件，定义它们的执行顺序，从而实现复杂的功能。

🔗 **Runnable接口及其作用**：Runnable是LangChain中的一个抽象接口，所有实现了该接口的组件都代表一个可执行单元，它提供了统一的调用方式，便于LCEL的调用和组合，并支持同步、异步、流式、批量调用等多种场景。

⚙️ **Runnable的常见实现**：LangChain提供了多种Runnable的实现，包括RunnableSequence（用于顺序执行）、RunnableParallel（用于并行执行）、RunnablePassthrough（用于传递输入数据或上下文）、RunnableLambda（用于包装Python函数）和RunnableBranch（用于条件分支）。

🧩 **核心组件概览**：LangChain的核心组件包括提示词模板、大语言模型（LLMs）、链（Chains）、记忆功能（Memory）和Agent。这些组件共同协作，使得LangChain能够构建各种复杂的应用，例如问答系统、聊天机器人等。

LangChain有两个重要的概念：LangChain表达式（LCEL） 、组件。我们使用组件，构建整个RAG应用中各种点单的操作，然后使用LCEL将这些组件进行连接，按照一定顺序运行，最终实现我们的目的。相关概念可在这里进一步理解。

Runnable

LCEL通过简洁的形式调用各类组件，实现方式是使用|串联所有实现了Runnable接口的组件。简而言之，LangChain的所有组件均实现了Runnable接口，通过LCEL的方式进行相连，然后按照顺序执行每个组件，最终得到结果。

概念

Runnable是LangChain中的一个抽象接口，凡是实现这个接口的均代表一个可执行单元，可通过标准化方法执行，处理并生成输出。其涉及目标是：

统一接口，为不同的组件提供统一的调用方式；便于LCEL调用，组合完成复杂任务；支持同步、异步、流式、批量调用等适合各种场景。

核心方法

invoke：同步单个调用

ainvoke：异步单个调用，适合高并发场景

stream：流式调用

batch：同步批量调用

abatch：异步批量调用

LangChain提供的常见的Runnable实现

RunnableSequence

例子：

from langchain_core.runnables import RunnableSequencefrom langchain_core.output_parsers import StrOutputParser# chain_of_sequence_v1 =  RunnableSequence(prompt, llm, StrOutputParser())# template_chain = prompt | llm# chain_of_sequence_v2 = RunnableSequence(first=template_chain, last=StrOutputParser())chain = prompt | llm | StrOutputParser()chain.invoke({"input":"你好"})

RunnableSequence有三个参数：first，middle，last，可以赋值实现了Runnable的组件，或者一个链。RunnableSequence等价于常用符号|。

RunnableParallel

例子：

from langchain_core.runnables import RunnableParallelsystem_prompt_of_parallel = "你是一个非常棒的助手，能根据用户的问题给出精准的回答。。"prompt_1 = ChatPromptTemplate(    [        ("system", system_prompt_of_parallel),        ("human", "请讲一个关于{object}的冷笑话。"),    ])prompt_2 = ChatPromptTemplate(    [        ("system", system_prompt_of_parallel),        ("human", "请详细介绍{object}。"),    ])chain_1 = prompt_1 | llm | StrOutputParser()chain_2 = prompt_2 | llm | StrOutputParser()chain_of_parallel = RunnableParallel({"chain_1":chain_1, "chain_2":chain_2})print(chain_of_parallel.invoke({"object":"苹果"}))

RunnableParallel的作用是，将两条链并行执行，并根据每条链的key输出对应的回答。并行执行后，也可以按照key取到对应的值，并执行后续操作。如对chain_of_parallel进行改造如下：

from langchain_core.runnables import RunnableParallel,RunnableLambdasystem_prompt_of_parallel = "你是一个非常棒的助手，能根据用户的问题给出精准的回答。。"prompt_1 = ChatPromptTemplate(    [        ("system", system_prompt_of_parallel),        ("human", "请讲一个关于{object}的冷笑话。"),    ])prompt_2 = ChatPromptTemplate(    [        ("system", system_prompt_of_parallel),        ("human", "请详细介绍{object}。"),    ])prompt_3 = ChatPromptTemplate(    [        ("system", system_prompt_of_parallel),        ("human", "评价一下 \n\n {chain_1} \n\n这个冷笑话怎么样,并且判断一下 \n\n {chain_2} \n\n 描述是否准确"),    ])chain_1 = prompt_1 | llm | StrOutputParser()chain_2 = prompt_2 | llm | StrOutputParser()chain_3 = prompt_3 | llm | StrOutputParser()chain_of_parallel = RunnableParallel({"chain_1":chain_1, "chain_2":chain_2}) | RunnableLambda(lambda x: {"chain_1":x["chain_1"], "chain_2":x["chain_2"]}) | chain_3print(chain_of_parallel.invoke({"object":"苹果"}))

上述代码可实现，在并行执行的链后，提取执行结果，并进一步对执行结果处理。

RunnablePassthrough

RunnablePassthrough是构建链式任务时直接传递输入数据或上下文，同时支持附加额外数据以增强后续处理能力。

例子：

from langchain_core.runnables import RunnablePassthroughsystem_prompt_of_passthrough = "你是一个非常棒的助手，能根据用户的问题给出精准的回答。。"prompt_of_passthrough = ChatPromptTemplate(    [        ("system", system_prompt_of_parallel),        ("human", "请讲一个关于{object}的冷笑话。"),    ])chain_of_passthrough = RunnablePassthrough() | llm | StrOutputParser()# chain_of_passthrough_1 = RunnablePassthrough.assign(object=RunnableLambda(lambda x:"西瓜"))| prompt_of_passthrough | StrOutputParser()chain_of_passthrough.invoke({"object":"苹果"})

需要注意的是，RunnablePassthrough.assign()可以修改后添加参数，如chain_of_passthrough_1所示，其中assign函数的参数为key=func，key为需要修改或新增参数的key，func为修改或新增的参数的函数。

RunnableLambda

可以将Python函数包装成为Runnable的实现，应用于链中。

例子：

from langchain_core.runnables import RunnableLambdachain_of_lambda = RunnableLambda(lambda x:x["object"])| StrOutputParser()# chain_of_lambda_v2 =  (lambda x:x["object"]) | StrOutputParser()chain_of_lambda_v2.invoke({"object":"苹果"})

此处使用lambda函数作为例子，也可以使用def定义的python函数。另外需要说明的，在链中出现的函数，即使是没有使用RunnableLambda进行包装，链的底层也会自动将其进行包装。

RunnableBranch

例子：

from langchain_core.runnables import RunnableBranchdefault_branch = (lambda x:"这个是default_branch") | StrOutputParser()chain_of_branch_1 = (lambda x:"这个是chain_of_branch_1") | StrOutputParser()chain_of_branch_2 = (lambda x:"这个是chain_of_branch_2") | StrOutputParser()chain_of_branch = RunnableBranch(    (lambda x:x == 1,chain_of_branch_1),    (lambda x:x == 2,chain_of_branch_2),    default_branch)chain_of_branch.invoke({"object":"苹果"})

根据条件选择执行不同的 Runnable 分支。

RunnableWithMessageHistory

添加对话历史管理和多轮对话跟踪，具体用法详见：这里

核心组件

提示词模板（Prompt Templates）：用于动态生成提示词；大语言模型（LLMs）：提供统一的对外访问API链（Chains）：将多个简单操作组合执行，实现复杂功能记忆功能（Memory）：记录对话内容，实现多轮对话Agent：大模型自主决策执行、动态调用工具

除了上述核心组件外，还有许多其他组件，如：检索器（Retrievers）、向量存储（Vector Store）、工具（tools）等，不一一列举，均可在这里找到。

原文地址：https://www.cnblogs.com/AfroNicky/p/18922644