LangChain Runnable 完整深度解读适配LangChain 0.1.x/ 0.2.x 新版体系LCELLangChain Expression Language的底层基石前置认知所有你写的prompt | llm | parser | retriever每一环都是Runnable一、核心定义1. 什么是 RunnableRunnable是一个统一接口规范Protocol / 抽象基类Runnable[Input, Output]任意实现该接口的对象都支持管道运算符|串联并行执行RunnableParallel同步 / 异步调用、流式输出、批处理绑定参数、重试、回调、日志、可观测性泛型定义python运行from langchain_core.runnables import Runnable # Runnable[输入类型, 输出类型] class Runnable[Input, Output]: def invoke(self, input: Input, config: Optional[RunnableConfig]) - Output: ... async def ainvoke(self, input: Input, config: Optional[RunnableConfig]) - Output: ... def stream(self, input: Input, config: Optional[RunnableConfig]) - Iterator[Output]: ... async def astream(self, input: Input, config: Optional[RunnableConfig]) - AsyncIterator[Output]: ... def batch(self, inputs: List[Input], configs: ...) - List[Output]: ... async def abatch(self, inputs: List[Input], configs: ...) - List[Output]: ...所有 LangChain 核心组件都实现 RunnableChatPromptTemplate、ChatOpenAI、StrOutputParser、BaseRetriever、DocumentTransformer、自定义函数链等2. 四大标准执行方法重中之重表格方法作用使用场景.invoke(input)单次同步调用完整输入→完整输出常规问答获取完整结果.ainvoke(input)异步单次调用FastAPI、异步服务高并发.stream(input)同步流式迭代逐块返回输出前端打字机流式输出.astream(input)异步流式迭代异步服务流式返回.batch(inputs)批量执行一组输入批量问答、数据处理⚠️ 重要约束Runnable 管道中数据会沿着链路流转前一个的输出 后一个的输入。二、LCEL 管道|原理__or__运算符python运行chain r1 | r2 | r3等价于python运行from langchain_core.runnables import RunnableSequence chain RunnableSequence(r1, r2, r3)执行逻辑输入传入r1.invoke(input)r1 的输出 → 作为 r2 的输入r2 的输出 → 作为 r3 的输入类型流转示例经典 RAG 链plaintextstr(query) → RunnablePassthrough retriever → Dict[str, str | List[Document]] → ChatPromptTemplate → PromptValue → ChatOpenAI → AIMessage → StrOutputParser → str三、内置核心 Runnable 组件详解1.RunnablePassthrough透传器作用原样把输入往下传递不做修改。 最常见场景拆分字典字段供给 prompt。python运行from langchain_core.runnables import RunnablePassthrough # 输入字符串问题 { question: RunnablePassthrough(), context: retriever }流程 输入什么是Runnablequestion直接透传原字符串context原字符串送入 retriever 得到文档列表如果你不写RunnablePassthrough()无法把原始输入保留到下游。2.RunnableParallel并行执行别名RunnableMap接收字典并发执行每一条 value 对应的 Runnable最后合并字典输出。{...}语法糖本质就是RunnableParallel。python运行# 两种写法完全等价 chain RunnableParallel({ summary: llm1, key_points: llm2 }) # 语法糖写法 chain {summary: llm1, key_points: llm2}典型场景一条问题同时调用多条检索器同一个输入让 LLM 同时生成摘要 关键点RAG 中同时输出 answer 和 source_documents示例带来源文档的 RAG 链python运行rag_with_source RunnableParallel({ context: retriever, question: RunnablePassthrough() }).assign( answer (lambda x: { context: format_docs(x[context]), question: x[question] }) | prompt | llm | StrOutputParser() ).assign()在现有输出字典上新增字段非常常用。3.RunnableLambda封装普通函数把任意 Python 函数包装成 Runnable接入管道。python运行from langchain_core.runnables import RunnableLambda def format_docs(docs): return \n\n.join(d.page_content for d in docs) format_docs_runnable RunnableLambda(format_docs) # 直接放进管道 retriever | format_docs_runnable支持异步函数RunnableLambda(async_func)绝大多数数据预处理、结果格式化都靠它。4.RunnableBind—.bind()绑定固定参数对下游 Runnable 绑定静态参数最常用于 LLM 参数绑定。python运行base_llm ChatOpenAI(modelgpt-3.5-turbo) creative_llm base_llm.bind(temperature0.9) strict_llm base_llm.bind(temperature0.0)底层llm.bind(...)返回一个新 Runnable调用时自动带上参数。5.RunnableBranch分支路由条件判断类似if/else根据上游输出选择不同链路执行。python运行from langchain_core.runnables import RunnableBranch branch RunnableBranch( (lambda x: len(x[question]) 100, long_chain), (lambda x: 代码 in x[question], code_chain), default_chain # 默认分支 )适用路由到不同知识库、区分普通问答 / 代码问答。四、RunnableConfig全局配置对象所有invoke/stream都可以传入config用于控制链路行为python运行from langchain_core.runnables import RunnableConfig config { callbacks: [...], # 回调函数日志、追踪 tags: [rag-demo], # 链路标签用于LangSmith观测 metadata: {user_id: 123}, recursion_limit: 10 # 防止递归链无限循环 } chain.invoke(query, configconfig)LangSmith 追踪、限流、权限、用户身份都通过 config 传递。五、流式在 Runnable 中的传播机制容易踩坑一个关键规则管道中只要某一环支持 stream整条链支持.stream()数据以增量块chunk向下传递python运行chain prompt | llm | StrOutputParser() for chunk in chain.stream(你好): print(chunk, end)坑点RunnableLambda默认不支持流式会收集全部输入再处理如果需要自定义流式处理需要实现transform()。六、自定义 Runnable进阶当RunnableLambda不够用时可以继承Runnable实现完整接口python运行from langchain_core.runnables import Runnable from typing import Iterator class MyTransformRunnable(Runnable[str, str]): def invoke(self, input: str, configNone) - str: return input.upper() def stream(self, input: str, configNone) - Iterator[str]: for char in input: yield char.upper()七、数据流类型流转避坑高频错误输入类型不匹配python运行# 错误retriever 接收 str上游输出 dict chain RunnablePassthrough() | retriever # 正确使用lambda提取字段 chain RunnablePassthrough() | (lambda x: x[question]) | retriever混淆「值」和「Runnable」python运行# ❌ 错误写法 {question: 固定文本} # ✅ 正确固定常量使用 RunnableLambda {question: RunnableLambda(lambda x: 固定文本)}管道不能直接混用异步函数必须包 RunnableLambda八、Runnable vs 旧版 Chain历史区分旧版Chain强制输入输出为字典没有统一流式标准无法自由|拼接RetrievalQA基于老 Chain灵活性差新版Runnable输入可以是 str /dict/ 对象灵活原生支持同步 异步 流式 批处理LCEL 声明式构建链路官方主推所有新功能只迭代 Runnable结论新项目一律使用 Runnable / LCEL不要使用遗留 Chain。九、典型完整示例RAG 标准 Runnable 链路汇总python运行from langchain_core.runnables import RunnablePassthrough, RunnableParallel from langchain_core.output_parsers import StrOutputParser def format_docs(docs): return \n\n.join(d.page_content for d in docs) rag_chain ( RunnableParallel({ context: retriever | format_docs, question: RunnablePassthrough() }) | prompt | llm | StrOutputParser() ) # 调用 rag_chain.invoke(LangChain Runnable是什么)十、生产级拓展能力Runnable 生态.with_retry()自动重试python运行retry_llm llm.with_retry(max_attempts3).with_fallbacks()LLM 故障自动切换备用模型python运行safe_llm llm1.with_fallbacks([llm2]).pick()提取字典中某个字段python运行chain.pick(answer).map()批量处理等价循环调用python运行chain.map().invoke([问题1,问题2])十一、知识思维导图总结plaintextRunnable 统一接口 ├─ 五大执行原语invoke / ainvoke / stream / astream / batch ├─ 组合方式 │ ├─ | 管道 → RunnableSequence │ └─ {} 字典 → RunnableParallel ├─ 基础工具Runnable │ ├─ RunnablePassthrough透传输入 │ ├─ RunnableLambda包装普通函数 │ ├─ RunnableBranch条件分支 │ └─ .bind() 绑定参数 ├─ 控制层 │ ├─ RunnableConfig回调、标签、元数据 │ ├─ with_retry / with_fallbacks └─ 生态载体 ├─ Prompt、LLM、Parser、Retriever 全部实现Runnable