大模型 LangChain 开发框架：Runable 与 LCEL 初探
一、引言
在大模型开发领域，LangChain 作为一款强大的开发框架，为开发者提供了丰富的工具和功能。其中，Runnable 接口和 LangChain 表达式语言（LCEL）是构建高效、灵活应用程序的关键要素。本文将初步探讨 Runnable 和 LCEL 的概念、功能以及如何在实际应用中运用它们来构建简单的工作流程。
（https://python.langchain.com/docs/how_to/lcel_cheatsheet/#invoke-a-runnable）参考LangChain 表达语言速查表
二、Runnable 接口详解
（一）接口定义与功能概述
Runnable 接口是 LangChain 众多组件以及 LangChain 表达语言构建的基础抽象。它定义了一系列标准方法，使得 Runnable 组件具备多种强大的功能：
- invoke 方法：该方法能够将单个输入转换为相应的输出，是实现基本功能的核心。例如，在一个简单的文本处理任务中，输入一段文本，通过 invoke 方法可以得到处理后的结果，如文本分类、摘要生成等。
- batched 方法：支持多个输入高效地转换为输出。这在处理批量数据时非常有用，能够显著提高处理效率。比如，同时对多篇文档进行关键词提取，使用 batched 方法可以一次性处理多个文档，减少处理时间。
- steramed 方法：实现输出在生成时以流式传输形式进行。在处理大文本或实时数据时，流式传输可以让用户及时获取部分结果，提升用户体验。例如，在实时翻译场景中，随着源文本的输入，翻译结果可以逐句或逐段地流式输出。
- Inspected 方法：提供了访问有关 Runnable 的输入、输出和配置的示意图信息的能力。这有助于开发者在调试和优化过程中深入了解组件的运行情况，快速定位问题。
- Composed 方法：允许组合多个 Runnable，通过 LangChain 表达语言（LCEL）协同工作，从而创建复杂的管道。这是构建复杂应用程序的关键特性，能够将多个简单的组件组合成一个强大的系统。
（二）接口的重要性与应用场景
Runnable 接口的标准化使得不同的 LangChain 组件能够相互协作，实现更高级的功能。在自然语言处理任务中，如文档处理、问答系统、机器翻译等，都可以利用 Runnable 接口构建高效的处理流程。例如，在构建一个智能客服系统时，可以将语音识别、文本处理、意图识别、回答生成等多个 Runnable 组件组合起来，实现从用户提问到回答的完整流程。
三、LangChain 表达式语言（LCEL）解析
（一）LCEL 的概念与语法
LangChain 表达式语言（LCEL）是一种用于编排 LangChain 组件的语法。它提供了一种声明式的方法来组合 Runnable 组件，使得构建复杂的工作流程变得更加简单和直观。LCEL 的主要组合原语包括 RunnableSequence 和 RunnableParallel。
（二）RunnableSequence 的使用
RunnableSequence 用于按顺序调用一系列 Runnable，其中一个 Runnable 的输出作为下一个 Runnable 的输入。可以使用 “|” 运算符或将 Runnable 列表传递给 RunnableSequence 来构造。例如，在一个文档处理流程中，首先进行文本加载（loader_pdf），然后进行文本分割（text_splitter），接着进行向量化（embedding），最后进行存储（save），这些步骤可以通过 RunnableSequence 依次连接起来，形成一个完整的文档预处理管道。
（三）RunnableParallel 的使用
RunnableParallel 则同时调用可运行程序，为每个程序提供相同的输入。可以使用序列中的字典文字或通过将字典传递给 RunnableParallel 来构造它。例如，在一个搜索和翻译的应用中，同时进行文档搜索（search）和将搜索结果翻译为英文（translateToEnglish）的操作，这两个任务可以通过 RunnableParallel 并行执行，提高处理效率。
（四）LCEL 的优势与适用场景
LCEL 的优势在于它能够以优化的方式处理链的运行时执行，自动获得同步、异步、批处理和流式支持。对于简单的应用程序，LCEL 是一个理想的选择，能够快速构建起有效的工作流程。例如，在构建一个小型的文档搜索和翻译工具时，使用 LCEL 可以方便地组合各个组件，实现基本功能。
四、案例分析：PDF 内容搜索结果后翻译为英文
（一）需求分析
本案例旨在实现对 PDF 文档的内容搜索，并将搜索到的中文内容翻译为英文。具体步骤包括加载 PDF 文件、分词、向量化存储、问题查询、获取相似答案并进行翻译。
（二）代码实现与详细解释
- 组件导入与模型初始化
from langchain_community.embeddings import VolcanoEmbeddings, FastEmbedEmbeddings
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.runnables import RunnableLambda, RunnableParallel
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
model = ChatOpenAI(
model="deepseek-chat",
api_key='<api-key>',
base_url='https://api.deepseek.com'
)
这里导入了所需的各种组件，包括用于加载 PDF 的PyPDFLoader、文本分割器RecursiveCharacterTextSplitter、向量化工具FastEmbedEmbeddings、内存向量存储InMemoryVectorStore、Runnable 相关的类和函数、消息类、ChatOpenAI 模型以及提示模板类。同时，初始化了 ChatOpenAI 模型，指定了模型名称、API 密钥和基础 URL。
- 函数定义
def save(embeddings):
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(documents=splits)
return vector_store
def embedding(all_splits):
global splits
fastembed = FastEmbedEmbeddings()
for split in all_splits:
v = fastembed.embed_query(split.page_content)
splits = all_splits
return fastembed
def text_splitter(docs):
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=50, chunk_overlap=10, add_start_index=True
)
all_splits = text_splitter.split_documents(docs)
return all_splits
def loader_pdf(file_path):
loader = PyPDFLoader(file_path)
docs = loader.load()
return docs
def search(store, query):
results = store.similarity_search_with_score(query)
document_obj, score = results[0]
return document_obj.page_content
def translateToEnglish(store, query):
content = search(store, query)
system_template = "你是一名资深的翻译大师，把中文翻译为{language}"
prompt_template = ChatPromptTemplate.from_messages(
[("system", system_template), ("user", "{text}")]
)
prompt = prompt_template.invoke({"language": "英文", "text": content})
return model.invoke(prompt)
save函数：用于将向量化后的文档存储到内存向量存储中。它接受向量化结果作为输入，创建一个InMemoryVectorStore对象，并将文档添加到其中，最后返回存储对象。embedding函数：负责对文档进行向量化处理。它使用FastEmbedEmbeddings对文档的每个片段进行向量化，并将结果存储在全局变量splits中，最后返回向量化工具对象。text_splitter函数：将加载后的文档进行文本分割。使用RecursiveCharacterTextSplitter按照指定的块大小、重叠量和是否添加起始索引进行分割，返回分割后的文档片段列表。loader_pdf函数：实现加载 PDF 文件的功能。通过PyPDFLoader加载指定路径的 PDF 文件，并返回加载后的文档对象。search函数：在向量存储中进行相似性搜索。根据输入的查询，在向量存储中查找最相似的文档，并返回文档的内容。translateToEnglish函数：将搜索到的中文内容翻译为英文。它首先调用search函数获取中文内容，然后根据给定的系统模板和用户输入构建提示模板，通过模型进行翻译，并返回翻译结果。
- 链的构建与执行
if __name__ == '__main__':
q = "docker的配置？"
file_path = "./ragflow文档.pdf"
chain = (RunnableLambda(loader_pdf) | RunnableLambda(text_splitter) | RunnableLambda(embedding) | RunnableLambda(save) |
RunnableParallel(first=RunnableLambda(search).bind(query=q), second=RunnableLambda(translateToEnglish).bind(query=q)))
chain.get_graph().print_ascii() # 流程图
result = chain.invoke(file_path)
print(q)
print(result['first'])
print(result['second'].content)
在主函数中，首先定义了查询问题q和 PDF 文件路径file_path。然后，使用 RunnableLambda 和 RunnableParallel 构建了一个复杂的链。通过|运算符将loader_pdf、text_splitter、embedding和save按顺序连接起来，形成文档预处理的序列。接着，使用RunnableParallel并行执行search和translateToEnglish操作，并将查询问题q绑定到这两个操作上。最后，通过chain.invoke方法执行整个链，传入 PDF 文件路径作为输入，并获取结果。打印出查询问题、搜索到的文档内容以及翻译后的英文内容。
流程图示：
+------------------+
| loader_pdf_input |
+------------------+
*
*
*
+------------+
| loader_pdf |
+------------+
*
*
*
+---------------+
| text_splitter |
+---------------+
*
*
*
+-----------+
| embedding |
+-----------+
*
*
*
+------+
| save |
+------+
*
*
*
+-----------------------------+
| Parallel<first,second>Input |
+-----------------------------+
* *
** **
* *
+--------+ +------------+
| search | | ChatOpenAI |
+--------+ +------------+
* *
** **
* *
+------------------------------+
| Parallel<first,second>Output |
+------------------------------+
结果：
docker的配置？
根据官方文档启动后，访问主页 http://127.0.0.1
After starting according to the official documentation, access the homepage at http://127.0.0.1
五、LCEL 的指导原则与应用建议
（一）不同场景下的选择策略
- 单个 LLM 调用：如果仅进行单个大语言模型（LLM）调用，不需要使用 LCEL，直接调用底层聊天模型即可。这种情况下，使用 LCEL 会增加不必要的复杂性。
- 简单链构建：当有一个简单的链，如提示 + LLM + 解析器或简单的检索设置等，并且希望利用 LCEL 的优势，如自动获得多种执行方式的支持，那么 LCEL 是一个合理的选择。它可以快速构建起有效的工作流程，提高开发效率。
- 复杂链构建：对于构建复杂的链，如具有分支、循环、多个代理等情况，建议使用 LangGraph。LangGraph 能够更好地处理复杂的编排逻辑，提供更强大的状态管理和流程控制能力。同时，在 LangGraph 的各个节点内仍然可以使用 LCEL 来处理具体的任务。
（二）实际应用中的注意事项
- 组件兼容性：在组合 Runnable 组件时，要确保各个组件之间的兼容性。例如，输入和输出的数据类型要匹配，否则可能导致链的执行失败。
- 资源管理：注意资源的使用情况，特别是在处理大规模数据或长时间运行的任务时。合理配置模型参数、向量存储大小等资源，避免资源耗尽或性能下降。
- 可维护性：编写清晰、可读的代码，合理命名变量和函数，添加必要的注释，以便于后续的维护和扩展。特别是在构建复杂的链时，良好的代码结构和注释能够大大降低维护成本。
六、总结
本文介绍了 LangChain 开发框架中的 Runnable 接口和 LangChain 表达式语言（LCEL）。通过对 Runnable 接口功能的深入剖析以及对 LCEL 语法和组合原语的详细解释，结合 PDF 内容搜索与翻译的案例分析，展示了如何在实际应用中运用这些特性构建复杂的工作流程。同时，提供了 LCEL 的指导原则和应用建议，帮助开发者在不同场景下选择合适的工具和方法，提高大模型应用程序的开发效率和质量。在未来的开发中，开发者可以根据具体需求灵活运用 Runnable 和 LCEL，构建出更加强大、高效的大模型应用程序。
