LangChain Core Introduction

LangChain Core定义了LangChain生态中最核心的最基础的抽象和LCEL原语(LangChain表达式语言)。

设计原则

遵循 模块化设计 和 极简风格

设计哲学：约定优于配置(Convention Over Configuration)

• 模块化设计: 各个抽象对象相互独立，不与任何特定LLM绑定。
  • 主要包括LM、Document Loader、Embedding、VectorStore、Retriever等抽象
• 极简风格: 接口设计简洁直观，避免过度封装。

优势: 任何provider都可以实现相关接口然后加入LangChain生态系统

核心抽象

核心抽象包括: 接口 + 协议 + 语法

三位一体的设计架构：

• 接口: 包括核心组件的接口(比如chat models, LLMs, vector stores, retrievers等等)
• 协议: 通用调用协议(Runnable抽象)
• 语法: LCEL (LangChain Expression Language), 用来声明编排各个组件

Runnable接口

• Runnable接口是LangChain设计的基石（类似Java中的Runnable接口）, 实现了与所有组件的标准化交互, 并通过LCEL实现各组件的组合
• 它代表一个可以被调用的、可并行化的、可流处理的、可变换的、可组合的工作单元

---

主要作用

• 统一的调用接口: 所有组件通过标准方法集(invoke/ainvoke, stream/astream, batch/abatch)提供一致的使用体验
• 并行化: 批处理操作batch
• 异步支持: a(async)开头的方法ainvoke, astream, abatch等
• 可组合性: 通过pipe语法实现各组件的组合

---

使用方式 Runnable接口的使用方式包括: 声明式 和 立即执行

• 声明式: 通过LCEL声明，lazy执行，类似spark的transform
• 立即执行: 调用invoke等方法，立即执行，类似spark的action

主要抽象类家族

LCEL

LCEL全称 LangChain Expression Language ，是一种声明式语言，用来组合LangChain Core的Runnable成各种序列或DAG，满足由LLM驱动的AI应用的常用组合模式的开发。 LangChain Core会将LCEL编译为优化的执行计划，并自动支持并行化、流式传输、过程跟踪和异步执行。

• LCEL通过以下机制实现高效执行

1. 自动优化：编译时合并操作符，减少调用开销
2. 透明并行：自动检测可并行执行的组件
3. 错误传播：异常处理链路贯穿整个工作流

主要组合原语是RunnableSequence和RunnableParallel。 RunnableSequence 按顺序依次调用一系列runnables。并且用上一个Runnable的输出作为下一个Runnable的输入。 构建方式

• 多个Runnable实例通过 | 运算符连接
• 通过传入多个Runnable实例列表给RunnableSequence直接创建。

RunnableSequence 实例

from langchain_core.runnables import RunnableLambda


# 示例1：创建简单计算流水线
# 通过 `|` 创建 RunnableSequence 实例
sequence = RunnableLambda(lambda x: x + 1) | RunnableLambda(lambda x: x * 2)
print(type(sequence))
# 输出: <class 'langchain_core.runnables.base.RunnableSequence'>

# 单个输入调用 invoke
invoke_result = sequence.invoke(1)
print(type(invoke_result), invoke_result)
# 输出: <class 'int'> 4

# 批处理 batch
batch_result = sequence.batch([1, 2, 3])
print(type(batch_result), batch_result)
# 输出: <class 'list'> [4, 6, 8]

RunnableParallel 并发调用一系列runnables。并且用同样的输入作为每一个Runnable的输入。 构建方式

• 多个Runnable实例通过 | 运算符连接
• 通过传入多个Runnable实例列表给RunnableSequence直接创建。

RunnableParallel实例

from langchain_core.runnables import RunnableLambda


# 示例2：创建并行计算分支
# 使用 dict 创建 RunnableParallel 实例
sequence = RunnableLambda(lambda x: x + 1) | {
    'mul_2': RunnableLambda(lambda x: x * 2),
    'mul_5': RunnableLambda(lambda x: x * 5),
}
print(type(sequence))
# 输出: <class 'langchain_core.runnables.base.RunnableSequence'>

# 单个输入调用 invoke
invoke_result = sequence.invoke(1)
print(type(invoke_result), invoke_result)
# 输出: <class 'dict'> {'mul_2': 4, 'mul_5': 10}

小提示

RunnableLambda是一个可以将一个python callable转换成一个Runnable实例的类。上述实例用来转换lambda函数。

Reference

• https://github.com/langchain-ai/langchain/blob/master/libs/core/README.md
• https://python.langchain.com/docs/concepts/lcel/
• langchain_core.runables.base.Runnable