CNDBA社区
在大语言模型（LLM）应用开发框架（特别是 LangChain）中，Runnable 是一个核心抽象概念，它统一了所有可以被“调用”或“执行”的组件的接口，使得链式调用、组合、并行、日志、流式传输等操作成为可能。
🧩 一、Runnable 是什么？
Runnable 是 LangChain 中的一个协议（Protocol）或接口（Interface），它定义了对象必须实现的方法，最核心的是：
invoke(input, config)：同步调用ainvoke(input, config)：异步调用stream(input, config)：流式输出astream(input, config)：异步流式输出batch(inputs, config)：批量调用abatch(inputs, config)：异步批量调用
✅ 任何实现了这些方法的对象，都可以被称为 Runnable，并可以被组合、嵌套、装饰。
🎯 二、Runnable 的作用
统一接口
无论是 PromptTemplate、LLM、Chain、Agent、Tool 还是自定义函数，只要包装成 Runnable，就可以用相同方式调用。链式组合（Chaining）
使用|操作符将多个 Runnable 串联起来，构建复杂流程：chain = prompt | llm | output_parser并行执行（Parallel）
使用RunnableParallel或字典方式并行执行多个分支：from langchain_core.runnables import RunnableParallel chain = RunnableParallel({ "summary": prompt1 | llm, "translation": prompt2 | llm })流式支持（Streaming）
支持逐 Token 输出，适用于聊天机器人、实时响应等场景。配置与上下文传递（config）
支持传入config参数，用于传递回调函数（callbacks）、元数据（metadata）、运行 ID（run_id）等，便于日志、监控、追踪。批处理（Batch）
支持一次处理多个输入，提高吞吐量。中间件与装饰（RunnableLambda, RunnablePassthrough 等）
可插入自定义逻辑、数据透传、条件分支等。
🛠️ 三、如何使用 Runnable
1. 基础使用：调用 LLM
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4")
# invoke 同步调用
result = llm.invoke("你好，介绍一下你自己")
print(result.content)
# ainvoke 异步调用
import asyncio
async def main():
result = await llm.ainvoke("你好")
print(result.content)
asyncio.run(main())
2. 链式调用（Chaining）
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_template("请写一首关于{topic}的诗")
llm = ChatOpenAI()
output_parser = StrOutputParser()
# 组合成链
chain = prompt | llm | output_parser
# 调用
poem = chain.invoke({"topic": "春天"})
print(poem)
3. 并行调用（Parallel）
from langchain_core.runnables import RunnableParallel
chain = RunnableParallel({
"poem": prompt | llm | output_parser,
"fact": lambda x: f"关于{x['topic']}的冷知识：它很美"
})
result = chain.invoke({"topic": "海洋"})
print(result)
# 输出: {'poem': '...', 'fact': '关于海洋的冷知识：它很美'}
4. 自定义 Runnable（使用 RunnableLambda）
from langchain_core.runnables import RunnableLambda
def reverse_text(input_str):
return input_str[::-1]
reverse_runnable = RunnableLambda(reverse_text)
result = reverse_runnable.invoke("hello")
print(result) # 输出: "olleh"
5. 流式输出（Streaming）
for chunk in llm.stream("讲个笑话"):
print(chunk.content, end="", flush=True)
6. 批量处理（Batch）
inputs = ["讲个笑话", "写首诗", "推荐电影"]
results = llm.batch(inputs)
for res in results:
print(res.content)
7. 使用 RunnablePassthrough 透传数据
from langchain_core.runnables import RunnablePassthrough
chain = (
{"topic": RunnablePassthrough()}
| prompt
| llm
| output_parser
)
result = chain.invoke("秋天")
print(result)
📦 四、哪些对象是 Runnable？
LangChain 中几乎所有组件都是 Runnable：
| 类型 | 示例 |
|---|---|
| LLM / ChatModel | ChatOpenAI, ChatAnthropic |
| PromptTemplate | ChatPromptTemplate |
| OutputParser | StrOutputParser |
| Chains | LLMChain, SequentialChain |
| Tools / Agents | Tool, AgentExecutor |
| 自定义函数 | RunnableLambda |
| 数据透传 | RunnablePassthrough |
| 并行/分支 | RunnableParallel |
📈 五、Runnable 的优势总结
| 特性 | 说明 | |
|---|---|---|
| 统一接口 | 所有组件调用方式一致 | |
| 可组合 | 使用 ` | ` 轻松构建复杂链 |
| 可并行 | 多任务同时执行 | |
| 可流式 | 支持逐 Token 输出 | |
| 可批处理 | 高效处理多个输入 | |
| 可配置 | 支持 callbacks、metadata 等 | |
| 可扩展 | 易于包装自定义逻辑 |
✅ 六、最佳实践建议
- 优先使用
|构建链，而不是旧式LLMChain - 使用
RunnableLambda包装自定义函数 - 利用
RunnableParallel提高并行效率 - 流式输出提升用户体验（如 Web 聊天）
- 配置
config实现日志、追踪、监控
📚 参考文档
- LangChain Runnable 文档：https://python.langchain.com/docs/expression_language/
- Runnable API 参考：https://api.python.langchain.com/en/latest/runnables/langchain_core.runnables.base.Runnable.html
✅ 总结：Runnable 是 LangChain 2.0+ 的核心抽象，它让 LLM 应用开发像搭积木一样灵活、可组合、可扩展。掌握 Runnable，是构建现代 LLM 应用的关键一步。
