LangChain Chain 与 Runnable：把大模型调用串成流水线

模型会调了，提示词模板也会写了，但真正的魔法在于——把一堆组件串起来，让数据像流水线一样自动流转。这就是 LangChain 最核心的概念：Chain（链）和 Runnable（可运行接口）。本文带你从"一个组件一个组件手动调"进化到"一条管道全自动处理"，感受 LCEL 管道符的真正威力。

---

一、Chain 是什么？为什么需要它？

1.1 先看没有 Chain 的写法

假设你要做一个"翻译 + 摘要"的功能——先把中文翻译成英文，再对英文做摘要。没有 Chain 的话，你得这么写：

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

# 第一步：翻译
translate_prompt = ChatPromptTemplate.from_template(
    "将以下中文翻译成英文：{text}"
)
translate_messages = translate_prompt.format_messages(text="LangChain 是一个强大的大模型应用开发框架")
translate_result = llm.invoke(translate_messages)

# 第二步：摘要（需要手动拿上一步的输出当输入）
summary_prompt = ChatPromptTemplate.from_template(
    "用一句话概括以下英文内容：{english_text}"
)
summary_messages = summary_prompt.format_messages(english_text=translate_result.content)
summary_result = llm.invoke(summary_messages)

print(summary_result.content)

每一步的输出都要手动取出来，再塞进下一步的输入——两步就这样，十步呢？ 代码会变成一堆中间变量，像意大利面条一样缠在一起。

1.2 有了 Chain 之后

同样的逻辑，用 LCEL 管道符 | 串起来：

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser

llm = ChatOpenAI(model="gpt-4o-mini")

# 定义两条独立的链
translate_chain = (
    ChatPromptTemplate.from_template("将以下中文翻译成英文：{text}")
    | llm
    | StrOutputParser()
)

summary_chain = (
    ChatPromptTemplate.from_template("用一句话概括以下英文内容：{english_text}")
    | llm
    | StrOutputParser()
)

# 调用：数据自动流转
english = translate_chain.invoke({"text": "LangChain 是一个强大的大模型应用开发框架"})
summary = summary_chain.invoke({"english_text": english})

print(f"翻译：{english}")
print(f"摘要：{summary}")

Chain 的本质：把多个组件串成一条流水线，每个组件的输出自动变成下一个组件的输入。 你不用再手动管数据传递，只管"把什么拼到什么后面"。

---

二、管道符 | 的魔法：LCEL 深入

LCEL（LangChain Expression Language）的核心就是管道符 |。它的设计灵感来自 Unix 管道——cat file | grep key | sort，数据从左流到右。

2.1 管道符的底层：Runnable 接口

所有能用 | 拼接的组件，都实现了同一个接口：Runnable。这个接口定义了统一的调用方法：

方法	行为	适用场景
invoke(input)	同步调用，一次性返回	普通脚本、调试
ainvoke(input)	异步调用	FastAPI、异步 Web 框架
stream(input)	同步流式输出	实时展示生成内容
astream(input)	异步流式输出	异步 + 流式
batch(inputs)	批量处理多条数据	批量翻译、批量分析

2.2 一条链，四种调用方式

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser

prompt = ChatPromptTemplate.from_template("给一个关于{topic}的冷笑话")
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()

# 方式一：invoke（同步）
result = chain.invoke({"topic": "程序员"})

# 方式二：stream（流式，一个字一个字往外蹦）
for chunk in chain.stream({"topic": "程序员"}):
    print(chunk, end="", flush=True)

# 方式三：batch（批量处理）
results = chain.batch([
    {"topic": "程序员"},
    {"topic": "产品经理"},
    {"topic": "设计师"},
])

# 方式四：ainvoke（异步）
import asyncio
async def main():
    result = await chain.ainvoke({"topic": "程序员"})
    print(result)
asyncio.run(main())

一条链写好，四种调用方式全都有。 这就是 Runnable 接口的威力——你不需要为每种调用方式写不同的代码，接口层帮你全搞定了。

---

三、输出解析器：让模型输出"结构化"

默认情况下，模型返回的是纯文本。但很多时候你需要的是结构化的数据——JSON、列表、字典。输出解析器就是干这个的。

3.1 StrOutputParser：最简单的解析器

把模型输出的消息对象转换成人话字符串：

from langchain.schema.output_parser import StrOutputParser

chain = prompt | llm | StrOutputParser()

# 没有 StrOutputParser → 返回 AIMessage 对象
# 有 StrOutputParser → 返回纯字符串
result = chain.invoke({"topic": "Python"})
print(type(result))  # <class 'str'>

3.2 JsonOutputParser：让模型输出 JSON

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import JsonOutputParser
from pydantic import BaseModel, Field

# 定义期望的数据结构
class Movie(BaseModel):
    title: str = Field(description="电影名称")
    director: str = Field(description="导演")
    year: int = Field(description="上映年份")
    rating: float = Field(description="豆瓣评分，1-10分")

# 创建解析器
parser = JsonOutputParser(pydantic_object=Movie)

# 模板中注入格式说明
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个电影数据库。请按以下JSON格式输出：\n{format_instructions}"),
    ("human", "介绍一下电影《肖申克的救赎》"),
])

# 组装链
chain = prompt | llm | parser

result = chain.invoke({
    "format_instructions": parser.get_format_instructions(),
})

print(result)
# {'title': '肖申克的救赎', 'director': '弗兰克·德拉邦特', 'year': 1994, 'rating': 9.7}

3.3 常用解析器一览

解析器	输入	输出	典型场景
StrOutputParser	AIMessage	str	日常对话、文本生成
JsonOutputParser	AIMessage	dict	结构化数据提取
CommaSeparatedListOutputParser	AIMessage	list	提取关键词列表
DatetimeOutputParser	AIMessage	datetime	提取日期时间
PydanticOutputParser	AIMessage	Pydantic 对象	复杂结构化输出

---

四、RunnableLambda：把自定义函数塞进链里

有时候你想在链的中间插入一步自定义逻辑——比如对文本做预处理、过滤敏感词、记录日志。RunnableLambda 可以把任何 Python 函数变成一个"可拼接的组件"：

4.1 基础用法

from langchain.schema.runnable import RunnableLambda
from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser

# 定义一个普通函数
def to_uppercase(text: str) -> str:
    return text.upper()

# 包装成 Runnable
uppercase_runnable = RunnableLambda(to_uppercase)

# 塞进链里
prompt = ChatPromptTemplate.from_template("解释一下：{topic}")
llm = ChatOpenAI(model="gpt-4o-mini")

chain = prompt | llm | StrOutputParser() | uppercase_runnable

result = chain.invoke({"topic": "Python"})
print(result)  # 全部大写输出

4.2 实战：给输出加个"水印"

from langchain.schema.runnable import RunnableLambda

def add_watermark(text: str) -> str:
    return text + "\n\n---\n由 AI 生成，仅供参考"

chain = prompt | llm | StrOutputParser() | RunnableLambda(add_watermark)

result = chain.invoke({"topic": "机器学习"})
# 输出末尾自动加上水印

4.3 多步自定义处理

from langchain.schema.runnable import RunnableLambda

def remove_think_tags(text: str) -> str:
    """去掉模型的 思考过程 标签"""
    import re
    return re.sub(r'思考过程.*?/思考过程', '', text, flags=re.DOTALL)

def trim_whitespace(text: str) -> str:
    return text.strip()

# 串成处理链
cleanup_chain = (
    RunnableLambda(remove_think_tags)
    | RunnableLambda(trim_whitespace)
)

# 挂在主链后面
full_chain = prompt | llm | StrOutputParser() | cleanup_chain

RunnableLambda 的本质：把"链外的逻辑"变成"链内的组件"。 任何 Python 函数，只要输入输出类型对得上，就能无缝嵌入流水线。

---

五、RunnablePassthrough：数据"直通"与"分流"

RunnablePassthrough 是一个看似简单但极其有用的组件——它原样传递数据，不做任何修改。听起来没用？它的真正威力在于数据分流和并行。

5.1 基础：数据直通

from langchain.schema.runnable import RunnablePassthrough

passthrough = RunnablePassthrough()
result = passthrough.invoke("hello")
print(result)  # "hello" —— 原样通过

5.2 核心用法：数据分流

RAG 场景中最常见的模式——一个问题进来，同时传给检索器和提示词模板：

from langchain.schema.runnable import RunnablePassthrough, RunnableLambda
from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser

# 模拟检索器
def retrieve_docs(query: str) -> str:
    return "相关文档内容：LangChain 是一个 LLM 应用开发框架..."

# 用 RunnablePassthrough 做数据分流
chain = (
    {
        "context": RunnableLambda(retrieve_docs),  # 走检索
        "question": RunnablePassthrough(),          # 原样直传
    }
    | ChatPromptTemplate.from_template("根据以下资料回答问题：\n{context}\n\n问题：{question}")
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

result = chain.invoke("LangChain 是什么？")
print(result)

5.3 数据流图解

输入："LangChain 是什么？"
        │
        ├──→ retrieve_docs() ──→ "相关文档内容：..."
        │                              │
        │                              ├──→ {context: "相关文档...", question: "LangChain 是什么？"}
        │                              │              │
        └──→ RunnablePassthrough() ────┘              │
                                                      ▼
                                               提示词模板 → 模型 → 输出

RunnablePassthrough 的精髓：它让你可以"复制"一份原始数据，同时保留原始数据继续往下传。 这在需要把同一份数据同时喂给多个组件时非常有用。

---

六、RunnableParallel：让多个任务同时跑

有些场景下，你需要同时调用多个模型或执行多个操作，然后汇总结果。RunnableParallel 就是做这个的——并行执行，统一汇总。

6.1 基础用法

from langchain.schema.runnable import RunnableParallel, RunnableLambda

def analyze_sentiment(text: str) -> str:
    return "正面"

def extract_keywords(text: str) -> str:
    return "Python, 框架, 大模型"

# 并行执行两个分析任务
parallel_chain = RunnableParallel(
    sentiment=RunnableLambda(analyze_sentiment),
    keywords=RunnableLambda(extract_keywords),
)

result = parallel_chain.invoke("LangChain 是一个优秀的 Python 大模型框架")
print(result)
# {'sentiment': '正面', 'keywords': 'Python, 框架, 大模型'}

6.2 实战：同时调用两个模型，对比结果

from langchain.schema.runnable import RunnableParallel
from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser

prompt = ChatPromptTemplate.from_template("用一句话解释：{concept}")

# 用两个不同的模型
gpt4_chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
qwen_chain = prompt | ChatOpenAI(
    model="qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key="your-key"
) | StrOutputParser()

# 并行调用
compare_chain = RunnableParallel(
    gpt4=gpt4_chain,
    qwen=qwen_chain,
)

result = compare_chain.invoke({"concept": "递归"})
print(f"GPT-4: {result['gpt4']}")
print(f"通义千问: {result['qwen']}")

---

七、Chain 的调试：看清每一步发生了什么

链越长，调试越难。LangChain 提供了几个调试利器：

7.1 全局调试开关

import langchain
langchain.debug = True  # 开启全局调试

# 之后每条链的每次调用都会打印详细的中间过程
chain.invoke({"topic": "AI"})

输出会显示每一步的输入、输出、耗时等详细信息。

7.2 单条链调试：用 RunnableLambda 打印中间结果

from langchain.schema.runnable import RunnableLambda

def debug_print(data):
    print(f"[DEBUG] 当前数据: {data}")
    return data  # 原样返回，不影响链

chain = (
    prompt
    | llm
    | RunnableLambda(debug_print)  # 插入调试节点
    | StrOutputParser()
)

7.3 分步执行：手动检查每一步

# 链 = 分步执行
prompt_value = prompt.invoke({"topic": "AI"})
print(f"提示词: {prompt_value}")

llm_result = llm.invoke(prompt_value)
print(f"模型输出: {llm_result}")

final = StrOutputParser().invoke(llm_result)
print(f"最终结果: {final}")

调试技巧：复杂链先分段跑通，再拼成一条完整链。 一段一段验证，比直接调一整条链然后猜哪里出错，效率高十倍。

---

八、总结

组件	作用	一句话
管道符 |	拼接组件	LangChain 的"胶水"，数据自动流转
Runnable 接口	统一调用规范	invoke/stream/batch/ainvoke 四种方式全支持
StrOutputParser	提取文本	把消息对象变成人话字符串
JsonOutputParser	提取 JSON	让模型输出结构化数据
RunnableLambda	自定义函数入链	任何 Python 函数秒变链内组件
RunnablePassthrough	数据直通	复制数据、分流，RAG 的标配
RunnableParallel	并行执行	多个任务同时跑，结果统一汇总

Chain 的精髓就四个字：组合优于继承。 你不用写一个"翻译+摘要+格式化"的大类，而是把三个小功能串起来——每个小功能独立可测，组合起来又能完成复杂任务。这就是 LangChain 最核心的设计哲学。

---

参考资料

• LangChain LCEL 官方文档
• LangChain Runnable 接口文档