当前位置：首页 > news >正文

LangChain Runnable简介

news 来源：原创 2025/4/24 10:47:48

LangChain 是一个强大的语言模型开发框架，而 Runnable 接口则是其核心基础之一。通过 Runnable，可以将各种处理逻辑模块化、标准化，并灵活地组合成强大的链式处理流程。本文将详细介绍 Runnable 接口的原理、常用方法、典型用法，并给出具体代码示例，帮助你快速上手。

什么是 Runnable 接口？

Runnable 是一个用于定义“可运行组件”的抽象协议。它允许你将各类处理单元（如文本预处理、模型推理、输出解析等）标准化封装，统一调用，并在链式流程中灵活组合。

主要特性包括：

标准化：统一的调用和组合接口
组合性：可串联（顺序）、并联（并行）多个 Runnable
扩展性：支持同步、异步、批量、流式等多种执行方式

核心方法与用法

Runnable 接口通常包括以下核心方法：

invoke：对单个输入执行处理，返回输出
batch：对一组输入批量处理，返回输出列表
stream：对单个输入进行流式处理，边处理边输出
对应的异步方法：ainvoke、abatch、astream
input_schema、output_schema：可选，用于描述输入输出数据结构

代码示例

1. 自定义`Runnable` 类

假设我们要实现一个简单的文本处理组件，将输入文本转为大写：

from abc import ABC, abstractmethodclass Runnable(ABC):@abstractmethoddef run(self, input_data):pass@abstractmethoddef configure(self, **kwargs):pass@abstractmethoddef validate(self, input_data):passclass TextToUpperCase(Runnable):def configure(self, **kwargs):passdef validate(self, input_data):if not isinstance(input_data, str):raise ValueError("Input data must be a string")def run(self, input_data):self.validate(input_data)return input_data.upper()# 使用示例
processor = TextToUpperCase()
print(processor.run("hello, langchain"))  # 输出：HELLO, LANGCHAIN

该例中，TextToUpperCase 组件实现了 Runnable 接口，支持配置、校验和处理输入。

2. 使用 `LangChain` 内置 `RunnableLambda`

LangChain 提供了更简洁的方式，将任意函数包装为 Runnable：

from langchain_core.runnables import RunnableLambdadef greet(name):return f"Hello, {name}!"greet_runnable = RunnableLambda(greet)
result = greet_runnable.invoke("Alice")
print(result)  # 输出：Hello, Alice!

这样，任何 Python 函数都能被包装成标准 Runnable，便于链式组合和统一管理。

3. 组合 Runnables 构建链式流程

利用 | 运算符或 pipe 方法，可以将多个Runnable 顺序组合，形成复杂的处理链：

from langchain_core.runnables import RunnableLambdaadd_one = RunnableLambda(lambda x: x + 1)
mul_two = RunnableLambda(lambda x: x * 2)sequence = add_one | mul_two
print(sequence.invoke(1))  # 输出：4
print(sequence.batch([1, 2, 3]))  # 输出：[4, 6, 8]