LangChain 是当前最流行的 LLM（大语言模型）应用开发框架，其核心设计理念是 “可组合、可复用、可追踪”。而这一切的基石，正是它的 Runnable 接口体系。本文将从 Runnable 出发，深入浅出地介绍 LangChain 的核心思想与使用方式。

---

一、什么是 Runnable？—— LangChain 的“积木块”

在 LangChain 中，一切可执行的组件都是 Runnable。

Runnable 不是一个具体类，而是一个抽象接口（Protocol），定义了如何处理输入、产生输出，并支持链式调用、批处理、流式响应等高级功能。

Runnable 的核心能力

能力	方法	说明
同步调用	.invoke(input, config=None, **kwargs)	单次输入 → 单次输出，最常用
异步调用	.ainvoke(input, config=None, **kwargs)	异步版本，适用于 async/await 环境
批量处理	.batch(inputs, config=None, **kwargs)	多输入 → 多输出，支持并行（如多用户请求）
异步批量	.abatch(inputs, config=None, **kwargs)	异步批量处理
流式输出	.stream(input, config=None, **kwargs)	逐块（chunk）返回生成内容，用于实时打字机效果
异步流式	.astream(input, config=None, **kwargs)	异步流式生成
流式批处理	.stream_log(input, config=None, **kwargs)	流式返回执行日志（含中间步骤，用于调试）
异步流式日志	.astream_log(input, config=None, **kwargs)	异步版本的流式日志

这些方法让所有组件行为统一，无论你用的是模型、提示词、工具还是自定义函数。

---

二、常见的 Runnable 实例

1. 模型（Model）本身就是 Runnable

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
# llm 是一个 Runnable！

response = llm.invoke("你好！")  # 直接调用

• 输入：str 或 List[BaseMessage]
• 输出：AIMessage

---

2. PromptTemplate：将变量拼成消息

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}助手"),
    ("human", "{input}")
])
# prompt 也是 Runnable！

messages = prompt.invoke({"role": "客服", "input": "订单没到"})
# 输出: [SystemMessage(...), HumanMessage(...)]

• 输入：dict（键对应模板变量）
• 输出：List[BaseMessage]

---

3. Chain：用 | 拼接 Runnable（管道操作）

chain = prompt | llm  # 自动组合！

response = chain.invoke({"role": "诗人", "input": "写一首春天的诗"})
# 内部：dict → messages → AIMessage

这就是 LangChain 最强大的地方：像搭积木一样组合 AI 应用。

---

4. 自定义 Runnable：用函数包装

from langchain_core.runnables import RunnableLambda

def add_greeting(input_str: str) -> str:
    return f"你好！{input_str}"

greet_chain = RunnableLambda(add_greeting) | llm

response = greet_chain.invoke("今天天气如何？")

任何函数都可以变成 Runnable！

---

三、高级 Runnable：带状态与记忆

1. RunnableWithMessageHistory：会话记忆

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

def get_session_history(session_id: str):
    return InMemoryChatMessageHistory()

chain_with_memory = RunnableWithMessageHistory(
    chain,  # 底层 chain
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

config = {"configurable": {"session_id": "user123"}}
response = chain_with_memory.invoke(
    {"input": "我是小明"}, 
    config=config
)

• 自动加载历史、注入 prompt、保存新消息；
• 实现多轮对话记忆。

---

2. RunnablePassthrough：透传 + 增强

from langchain_core.runnables import RunnablePassthrough

# 在不改变输入的情况下，附加上下文
enriched_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
)

常用于 RAG（检索增强生成）场景。

---

四、Runnable 的底层机制

所有 Runnable 都继承自 RunnableSerializable，并实现 _invoke 等私有方法。但你无需关心这些，因为：

LangChain 的设计哲学是：关注“做什么”，而不是“怎么做”。

当你写 prompt | model 时，LangChain 自动：

1. 将 prompt 的输出（消息列表）作为 model 的输入；
2. 处理类型转换；
3. 支持异步、流式、批处理等。

---

五、为什么 Runnable 如此重要？

1. 统一接口

无论是模型、工具、数据库查询还是自定义逻辑，都遵循同一套调用规范。

2. 可组合性

通过 | 运算符，轻松构建复杂 pipeline：

retriever | prompt | llm | parser

3. 可观测性

配合 LangSmith，自动记录每一步的输入/输出、耗时、token 使用等。

4. 部署友好

整个 chain 可序列化为 JSON，便于部署到服务器或 Serverless 平台。

---

六、实战：构建一个带记忆的问答机器人

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.chat_history import InMemoryChatMessageHistory

# 1. 定义模型
llm = ChatOpenAI(model="gpt-4o-mini")

# 2. 定义带历史的 Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个 helpful AI 助手"),
    MessagesPlaceholder("history"),
    ("human", "{input}")
])

# 3. 构建 chain
chain = prompt | llm

# 4. 添加记忆
memory_store = {}

def get_history(session_id: str):
    if session_id not in memory_store:
        memory_store[session_id] = InMemoryChatMessageHistory()
    return memory_store[session_id]

chatbot = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 5. 使用
config = {"configurable": {"session_id": "abc123"}}
print(chatbot.invoke({"input": "你好！"}, config).content)
print(chatbot.invoke({"input": "我刚才说了什么？"}, config).content)

全程只调用 .invoke()，背后自动处理记忆、拼接、调用。

---

总结

LangChain 的灵魂，就是 Runnable。
它把 LLM 应用从“一次性脚本”提升为“可工程化系统”。理解 Runnable，就掌握了 LangChain 的钥匙。

• 想快速原型？用 RunnableLambda 包装函数。
• 想做 RAG？用 RunnablePassthrough 组合检索器。
• 想支持多轮对话？用 RunnableWithMessageHistory。