LangChain 管道（Runnable）：核心用法详解

LangChain 管道是一个 由多个可运行组件（Runnable）组成的流水线 ，数据从左到右依次流动，每个组件处理后传递给下一个。

在上一篇中，我们已经通过：

python
Prompt | LLM | OutputParser

完成了一个最小可用的 LangChain 调用流程。从这一行代码开始，LangChain 实际上已经引入了一个非常核心的抽象：Runnable（可执行管道）。这一篇将围绕 Runnable 的实际使用方式展开，重点放在：

常见 Runnable 类型的用法
如何自己封装 RunnableLambda
如何构建分支（RunnableBranch）与并行（RunnableParallel）管道

1. Runnable 是什么

所有能接入管道的组件都实现了 Runnable 接口；在 LangChain 中，只要一个对象：

能接收输入
能返回输出
能被 invoke() / batch() / stream() 调用

它就可以被视为一个 Runnable。上一篇用到的PromptTemplate、ChatPromptTemplate、LLM、OutputParser其实都是 Runnable

Runnable 的核心价值，在工程上可以理解为

把“一次模型调用”拆成多个可组合、可复用的执行步骤

常见的 `Runnable` 类型

类型	说明
`PromptTemplate`	提示词模板
`ChatModel`	大模型（如 `ChatOpenAI`）
`OutputParser`	输出解析器
`RunnableLambda`	包装普通函数
`RunnableParallel`	并行执行
`ToolNode`	工具调用节点
`Retriever`	向量检索器

2.RunnableLambda：把普通函数变成管道节点

我们在实际开发中，通常需要在调用模型前后，加一点自己的逻辑比如：

清洗用户输入
拼接上下文
对模型输出做轻量处理

我们可以将自己定义的逻辑使用RunnableLambda，使其可以在管道中使用

基本用法：

python
from langchain_core.runnables import RunnableLambda

def normalize_question(text: str) -> str:
    return text.strip().replace("\n", " ")

normalize_runnable = RunnableLambda(normalize_question)

现在，这个普通函数已经是一个 Runnable 了。

与 Prompt / LLM 组合

python
from langchain_core.runnables import RunnableLambda

def add_prefix(text):
    return f"[前缀]{text}"

def to_upper(text):
    return text.upper()

# 接入管道
chain = (
    RunnableLambda(add_prefix)
    | RunnableLambda(to_upper)
    | llm
    | StrOutputParser()
)

result = chain.invoke("hello world")

这里的管道结构非常清晰：

python
add_prefix → to_upper → llm → StrOutputParser

每一步都只做一件事。

3.RunnableBranch：根据条件走不同分支

RunnableBranch可构建条件分支，比如：

不同输入走不同 Prompt
不同问题类型用不同模型
是否命中规则决定是否调用 LLM

示例：按问题类型选择 Prompt

python
from langchain_core.runnables import RunnableBranch

# 如果问题包含代码，返回True
def is_code_question(inputs: dict) -> bool:
    return "代码" in inputs["question"]

# 编程助手
code_prompt = PromptTemplate(
    input_variables=["question"],
    template="你是一个编程助手，请回答：{question}"
)

# 文案助手
copywriter_prompt = PromptTemplate(
    input_variables=["question"],
    template="你是一个文案编辑助手，请帮我编辑：{question}"
)

chat_prompt = PromptTemplate(
    input_variables=["question"],
    template="你是聊天机器人，请回答：{question}"
)


# 使用 lambda 和 函数两张方式做判断都可以
branch = RunnableBranch(
    (is_code_question, code_prompt), #如果包含代码，走编程助手
    (lambda x: "文案" in x["question"], copywriter_prompt ), #直接使用lambda，如果包含文案，走文案助手
    chat_prompt  # 默认分支，否则走聊天助手
)

chain = (
    branch
    | llm
    | StrOutputParser()
)

chain.invoke({"question": "这段代码为什么会报错？"})

4.RunnableParallel：并行执行多个子任务

RunnableParallel 在支持的执行环境中，会并发调度各子任务，而不是顺序执行。

并行在以下场景非常常见：

同一个问题，用多个 Prompt 角度分析
同时生成摘要 / 关键词 / 解释
多路召回后统一汇总

基本用法

python
from langchain_core.runnables import RunnableParallel

parallel = RunnableParallel({
    "summary": PromptTemplate(
        input_variables=["text"],
        template="请用一句话总结：{text}"
    ) | llm | StrOutputParser(),

    "keywords": PromptTemplate(
        input_variables=["text"],
        template="请提取关键词：{text}"
    ) | llm | StrOutputParser()
})

调用并查看结果

python
result = parallel.invoke({
    "text": "LangChain 是一个用于构建大模型应用的框架"
})

print(result)

返回结构是一个字典

python
{
  "summary": "...",
  "keywords": "..."
}

每个 key 对应一条独立的执行路径。

5. 组合示例：分支 + 并行 + LLM

Runnable 的强大之处在于可以无限嵌套组合：

python
chain = (
    # 1：从输入字典中取出 question 字段
    RunnableLambda(lambda x: x["question"])
    # 2：根据问题内容进行条件分支，如果问题中包含“分析”，则走分析型 Prompt
    | RunnableBranch(
        # 条件函数： 返回 True 表示命中该分支
        (lambda q: "分析" in q, analysis_prompt),
        # 默认分支（兜底分支），当所有条件都不满足时使用
        chat_prompt
    )
    # 3：RunnableParallel会将上一个节点的输出同时作为输入，传递给每一个子 Runnable。
    | RunnableParallel({
        # 子任务 1：生成主要回答，Prompt → LLM → 字符串解析
        "answer": llm | StrOutputParser(),
        # 子任务 2：生成评价信息，可以使用不同 Prompt，但复用同一个 LLM
        "confidence": confidence_prompt | llm | StrOutputParser()
    })
)

# 调用
chain.invoke({"question":"帮我分析这篇文章中的。。。。"})

最终 invoke() 的返回值结构为：

python
{
  "answer": "...",
  "confidence": "..."
}

这已经是一个完整的轻量级工作流了。

6.小结

这一篇我们没有从抽象定义开始，而是从真实使用场景出发，介绍了 LangChain 管道（Runnable）的核心用法：

如何用 RunnableLambda 把普通函数接入管道
如何用 RunnableBranch 实现条件分支
如何用 RunnableParallel 并行执行多个子任务
如何把多个 Runnable 组合成清晰、可维护的执行流程

Runnable 并不是“高级特性”，而是 LangChain 架构的基础单元。理解它之后，后面的 RAG、Agent、LangGraph，本质上都只是更复杂的 Runnable 组合。

目录