本文详细介绍LangChain Core的基本原理、核心概念、常见用法及生态

一、LangChain Core简介 #

LangChain Core 是 LangChain 框架的核心模块，提供了链式编排、数据流、回调等底层能力，是构建大语言模型应用的基础。

二、安装方法 #

介绍如何安装LangChain Core及其依赖。

# 安装LangChain Core主库
pip install langchain-core

三、核心概念 #

本节介绍LangChain Core的主要组成部分。

Runnable：可运行的链式组件，支持同步、异步、流式等多种执行方式。
Schema：用于定义输入输出的数据结构和类型。
Callback：回调机制，可在链执行过程中插入自定义逻辑（如日志、监控等）。

四、常见用法 #

介绍LangChain Core的典型用法，包括自定义链、流式处理和回调机制。

1. 自定义Runnable链 #

展示如何自定义一个简单的Runnable链。 `python

导入Runnable相关基类 #

from langchain_core.runnables import Runnable

定义一个自定义的Runnable类，实现call方法 #

class MyAddRunnable(Runnable):

# 实现__call__方法，接收输入并返回结果
def __call__(self, input_data):
    # 对输入数据加1
    return input_data + 1

创建自定义链对象 #

add_chain = MyAddRunnable()

运行链，输入数字3，输出4 #

result = add_chain(3)

打印结果 #

print(result)


### 2. 使用Schema定义输入输出
> 展示如何用Schema约束链的输入输出类型。
```python
# 导入Schema相关模块
from langchain_core.schema import BaseSchema

# 定义输入Schema
class InputSchema(BaseSchema):
    # 输入字段a，类型为int
    a: int

# 定义输出Schema
class OutputSchema(BaseSchema):
    # 输出字段result，类型为int
    result: int

# 定义一个带Schema的Runnable
from langchain_core.runnables import Runnable
class AddOneRunnable(Runnable):
    def __call__(self, input_data: InputSchema) -> OutputSchema:
        # 对输入a加1
        return OutputSchema(result=input_data.a + 1)

# 创建输入对象
input_obj = InputSchema(a=5)
# 创建链对象
add_one_chain = AddOneRunnable()
# 运行链，输出为6
output_obj = add_one_chain(input_obj)
# 打印输出结果
print(output_obj)

3. 流式处理（Streaming） #

展示如何实现流式数据处理。 `python

导入Runnable和生成器相关模块 #

from langchain_core.runnables import Runnable

定义一个流式Runnable，逐步输出结果 #

class StreamingRunnable(Runnable): def call(self, input_data):

    # 逐步yield输出
    for i in range(input_data):
        # 每次输出i
        yield i

创建流式链对象 #

stream_chain = StreamingRunnable()

运行链，遍历输出0~2 #

for value in stream_chain(3):

# 打印每个输出值
print(value)


### 4. 回调机制（Callback）
> 展示如何在链执行过程中插入自定义回调。
```python
# 导入回调相关模块
from langchain_core.callbacks import CallbackManager, BaseCallbackHandler
from langchain_core.runnables import Runnable

# 定义自定义回调处理器
class PrintCallbackHandler(BaseCallbackHandler):
    # 在链开始时回调
    def on_chain_start(self, *args, **kwargs):
        print("链开始执行")
    # 在链结束时回调
    def on_chain_end(self, *args, **kwargs):
        print("链执行结束")

# 创建回调管理器
callback_manager = CallbackManager([PrintCallbackHandler()])

# 定义一个简单的Runnable
class SimpleRunnable(Runnable):
    def __call__(self, input_data):
        return input_data * 2

# 创建链对象
simple_chain = SimpleRunnable()

# 运行链并触发回调
result = callback_manager.run(simple_chain, 10)
# 打印结果
print(result)

五、生态与扩展 #

LangChain Core 可与LangChain主库、LangServe、LangSmith等生态组件无缝集成，支持多种大模型、数据库、工具等扩展。

LangChain：完整的链式开发框架，基于Core构建。
LangServe：链/Agent的API服务化部署。
LangSmith：链路调试与监控平台。

六、应用场景 #

LangChain Core 适用于以下典型场景：

自定义链式数据处理
复杂业务流程自动化
多模型/多工具协作
需要流式/异步/回调的高级场景

七、参考资料 #

推荐学习和查阅的LangChain Core相关资料。

官方文档：https://python.langchain.com.cn/docs/core/
LangChain GitHub：https://github.com/langchain-ai/langchain
LangChain 中文社区：https://langchain.com.cn/