LangChain Runnable和LCEL

本文最后更新于 2026年7月19日

1.Runnable

Runnable是langchain中的抽象基类，是langchain中所有链的核心抽象接口，旨在为所有的可执行组件提供通用的操作接口，用于构建所有的“链”组件，代表一个可以调用运行的流程单元，使一切可执行组件都有一个统一的调用方式。

ABC 抽象基类，必须被继承，不能直接用
Generic[Input, Output] 泛型，规定输入/输出类型
```
class Runnable(ABC, Generic[Input, Output]):
```

Runnable将多个组件按照特定顺序组合起来，形成一个可以完成复杂任务的工作流或管道（pipeline），任何东西只要继承了Runnable，就必须实现run()/invoke()方法，只要实现了Runnable，就可以像函数一样invoke()或使用|管道操作符组合。

在langchain中，统一调用规范的接口在langchain_core.runnables中，主要有15种，常见的四种：

RunnableSequence 顺序链
RunnableBranch 分支链
RunnableParallel 并行链
RunnableLambda 函数链

1.1 RunnableSequence

顺序链，按顺序连接多个可运行对象，上个对象的输出作为下个对象的输入。


import os

from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableSequence

prompt_template = PromptTemplate(
    template='做一个关于{topic}的小诗',
    input_variables=['topic']
)

llm = init_chat_model(
    model = 'deepseek-chat',
    model_provider = 'openai',
    api_key = os.getenv('DSKEY'),
    base_url = 'https://api.deepseek.com'
)

parser = StrOutputParser()

chain = RunnableSequence(prompt_template, llm, parser)

resp = chain.invoke({'topic': '霸道总裁爱上做保洁的我'})

print(resp)

1.2 RunnableBranch

分支链，使用条件分支判断对列表和默认分支进行初始化

例：实现一个输入语种，把语种名称就地翻译成这个语种对应的外语，默认翻译成英语


import os

from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableSequence, RunnableBranch

english_template = PromptTemplate(
    template='直接翻译成英语： {topic}',
    input_variables=['topic']
)

japanese_template = PromptTemplate(
    template='直接翻译成日语： {topic}',
    input_variables=['topic']
)

korean_template = PromptTemplate(
    template='直接翻译成韩语： {topic}',
    input_variables=['topic']
)

llm = init_chat_model(
    model = 'deepseek-chat',
    model_provider = 'openai',
    api_key = os.getenv('DSKEY'),
    base_url = 'https://api.deepseek.com'
)

parser = StrOutputParser()

chain = RunnableBranch(
    (lambda e: '韩语' == e['topic'], RunnableSequence(korean_template, llm, parser) ),
    (lambda e: '日语' == e['topic'], RunnableSequence(japanese_template, llm, parser)),
    (RunnableSequence(english_template, llm, parser))
)

resp = chain.invoke({'topic': '韩语'})

print(resp)

resp = chain.invoke({'topic': '日语'})

print(resp)

resp = chain.invoke({'topic': '法语'})

print(resp)

**한국어** (Hangugeo)
「日本語」
French

1.3 RunnableParallel

并行链，创建多个子链，并在它们都完成后，汇总结果，常用于同时问多个问题并聚合结果，或者多模型同时工作取最优答案，再或者多路径和多模态的处理。


import os

from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableSequence, RunnableParallel


english_template = PromptTemplate(
    template='你是个翻译机器人，中译英： {topic}',
    input_variables=['topic']
)

japanese_template = PromptTemplate(
    template='你是个翻译机器人，中译日： {topic}',
    input_variables=['topic']
)

korean_template = PromptTemplate(
    template='你是个翻译机器人，中译韩： {topic}',
    input_variables=['topic']
)

llm = init_chat_model(
    model = 'deepseek-chat',
    model_provider = 'openai',
    api_key = os.getenv('DSKEY'),
    base_url = 'https://api.deepseek.com'
)


parser = StrOutputParser()

chain = RunnableParallel({
    "korean": RunnableSequence(korean_template, llm, parser),
    "japanese": RunnableSequence(japanese_template, llm, parser),
    "english": RunnableSequence(english_template, llm, parser),
})


resp = chain.invoke({'topic': '欲穷千里目，更上一层楼'})

print(resp)

{'korean': '欲窮千里目，更上一層樓  \n→ 천 리 밖까지 바라보려면, 한 층 더 높은 곳에 올라가야 한다.  \n(또는 시의 느낌을 살려)  \n→ 더 넓은 세상을 보려면, 더 높이 올라가야 한다.', 'japanese': '「千里の目を窮めんと欲すれば、更に一層の楼に上る」', 'english': 'Here\'s the English translation of the Chinese poem:\n\n"To see a thousand miles further, you must ascend one more story."\n\nThis is a famous couplet from the poem *On the Stork Tower* (《登鹳雀楼》) by Wang Zhihuan (王之涣) of the Tang Dynasty.'}

1.4 RunnableLambda

函数链，RunnableLambda是一个langchain的包装器，将普通python函数融入到Runnable中，把普通的函数转换为一个可执行的“链”，然后就可以像其他langchain组件一样使用。

例：

from langchain_core.runnables import RunnableLambda, RunnableBranch, RunnableSequence

step1 = RunnableLambda(lambda x: x.upper())
step2 = RunnableLambda(lambda x: x + '!!!')

chain = RunnableSequence(step1, step2)

res = chain.invoke('hello ai')
print(res)

HELLO AI!!!

例：在大模型调用中，自定义一个RunnableLambda(word_count)，用于统计生成内容字数


import os

from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda, RunnableSequence

prompt_template = PromptTemplate(
    template='做一个关于{topic}的小诗',
    input_variables=['topic']
)

llm = init_chat_model(
    model = 'deepseek-chat',
    model_provider = 'openai',
    api_key = os.getenv('DSKEY'),
    base_url = 'https://api.deepseek.com'
)

parser = StrOutputParser()


def word_count(text: str) -> int:
    print('----------word_count---------')
    return len(text)

word_counter = RunnableLambda(word_count)

chain = RunnableSequence(prompt_template, llm, parser, word_counter)

resp = chain.invoke({'topic': '霸道总裁爱上做保洁的我'})

print(resp)

----------word_count---------
1164

2.LCEL

要理解LCEL，首先要了解Runnable，Runnable（langchain_core.runnables.base.Runnable）是langchain中可以调用，批处理，流式输出，转换和组合的工作单元，是实现LCEL的基础，通过重写__or__()方法，实现了|运算符的重载，实现了Runable的类对象之间便可以进行一些类似linux命令中的管道（|）操作。

LCEL，全称LangChain Express Language，即LangChain表达式语言，也是LangChain官方推荐的写法，是一种从Runnable而来的声明式方法，用于声明，组合和执行各种组件（模型，提示词，工具等），如果要使用LCEL，对应的组件必须实现Runnable，使用LCEL创建的Runnable称之为链。

例：

import os

from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt_template = PromptTemplate(
    template='做一个关于{topic}的小诗',
    input_variables=['topic']
)

llm = init_chat_model(
    model = 'deepseek-chat',
    model_provider = 'openai',
    api_key = os.getenv('DSKEY'),
    base_url = 'https://api.deepseek.com'
)

parser = StrOutputParser()

# LCEL重写
chain = prompt_template | llm | parser

resp = chain.invoke({'topic': '霸道总裁爱上做保洁的我'})

print(resp)

还可以自定义一个word_count(text: str) -> int函数，通过langchain的RunnableLambda对象包装，使得函数变为获得链式的执行能力的Runable对象，拼入链中，统计大模型回复的字数


import os

from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

prompt_template = PromptTemplate(
    template='做一个关于{topic}的小诗',
    input_variables=['topic']
)

llm = init_chat_model(
    model = 'deepseek-chat',
    model_provider = 'openai',
    api_key = os.getenv('DSKEY'),
    base_url = 'https://api.deepseek.com'
)

parser = StrOutputParser()


def word_count(text: str) -> int:
    print('----------word_count---------')
    return len(text)

word_counter = RunnableLambda(word_count)

# LCEL重写
chain = prompt_template | llm | parser | word_counter

resp = chain.invoke({'topic': '霸道总裁爱上做保洁的我'})

print(resp)