Runnable与LCEL

Runnable 接口是使用 LangChain 组件的基础，它在许多组件中实现，例如语言模型、输出解析器、检索器、编译的 LangGraph 图等。

Runnable 接口强制要求所有 LCEL 组件实现一组标准方法：

LCEL 创建的 Runnable 为“链”，“链”本身就是 Runnable。

# 分步调用
prompt_text = prompt.invoke({"topic": "猫"})  # 方法1
model_out = model.invoke(prompt_text)  # 方法2
result = parser.invoke(model_out)  # 方法3

# LCEL管道式
chain = prompt | model | parser  # 用管道符组合
result = chain.invoke({"topic": "猫"})  # 所有组件统一用invoke

RunnableSequence 可运行序列

chain = runnable1 | runnable2
# 等同于
chain = RunnableSequence([runnable1, runnable2])

RunnableParallel 可运行并行

RunnableParallel 同时运行多个可运行对象，并为每个对象提供相同的输入。

对于同步执行，RunnableParallel 使用 ThreadPoolExecutor 来同时运行可运行对象。

对于异步执行，RunnableParallel 使用 asyncio.gather 来同时运行可运行对象。

在 LCEL 表达式中，字典会自动转换为 RunnableParallel。

mport os
from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_core.output_parsers import StrOutputParser

llm = init_chat_model(
    model="openai/gpt-oss-20b:free",
    model_provider="openai",
    base_url="https://openrouter.ai/api/v1",
    api_key=os.getenv("OPENROUTER_API_KEY"),
)

joke_chain = (
    PromptTemplate.from_template("讲一个关于{topic}的笑话") | llm | StrOutputParser()
)
poem_chain = (
    PromptTemplate.from_template("写一首关于{topic}的诗歌") | llm | StrOutputParser()
)

map_chain = RunnableParallel(joke=joke_chain, poem=poem_chain)

resp = map_chain.invoke({"topic": "人工智能"})
print(resp)

RunnableLambda 可运行

RunnableLambda 将 Python 可调用函数转换为 Runnable，使得函数可以在同步或异步上下文中使用。

from langchain_core.runnables import RunnableLambda

chain = {
    "text1": lambda x: x + " world",
    "text2": lambda x: x + ", how are you",
} | RunnableLambda(lambda x: len(x["text1"]) + len(x["text2"]))

result = chain.invoke("hello")
print(result)  # 29

RunnablePassthrough 可运行透传

RunnablePassthrough 接收输入并将其原样输出。RunnablePassthrough 是 LangChain LCEL 体系中的“无操作节点”，用于在流水线中透传输入或保留上下文，也可以用于向输出中添加键。

使用 assign() 向输出中添加键

from langchain_core.runnables import RunnablePassthrough

chain = {
    "text1": lambda x: x + " world",
    "text2": lambda x: x + ", how are you",
} | RunnablePassthrough.assign(word_count=lambda x: len(x["text1"] + x["text2"]))

result = chain.invoke("hello")
print(result)

# {'text1': 'hello world', 'text2': 'hello, how are you', 'word_count': 29}

RunnableBranch可运行分支

对输入进行操作时，选择第一个计算结果为 True 的条件，并在输入上运行相应的 Runnable。如果没有条件为 True，则在输入上运行默认分支。

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: isinstance(x, str), lambda x: x.upper()), # 如果x是字符串执行
    (lambda x: isinstance(x, int), lambda x: x + 1),   # 如果x是int执行
    (lambda x: isinstance(x, float), lambda x: x * 2),  #  如果x是浮点类型执行
    lambda x: "goodbye",  # 其他条件都不符合时执行
)

result = branch.invoke("hello")
print(result)  # HELLO

result = branch.invoke(None)
print(result)  # goodbye

RunnableWithFallbacks 可运行带回退

Runnable 失败后可以回退到其他 Runnable

可以直接在Runnable 上使用 with_fallbacks 方法。

import os
from langchain.chat_models import init_chat_model
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda

llm = init_chat_model(
    model="openai/gpt-oss-20b:free",
    model_provider="openai",
    base_url="https://openrouter.ai/api/v1",
    api_key=os.getenv("OPENROUTER_API_KEY"),
)

chain = PromptTemplate.from_template("hello") | llm
chain_with_fallback = chain.with_fallbacks([RunnableLambda(lambda x: "sorry")])

result = chain_with_fallback.invoke("1")  # 提示词模板中没有需要填充的变量，会报错
print(result)  # sorry