企业官网建设流程全解析

LangChain 在 1.x 时代做的最关键一件事，不是又塞进来多少新组件，而是把"组件"这个概念本身统一成了一个 Runnable。这一个抽象，让 prompt 模板、聊天模型、向量检索器、输出解析器全变成了同一类东西。上层编排从此不需要认识底下到底是什么，只要它长得像 Runnable，就能往管道里塞。这篇文章想把 Runnable 协议和 LCEL 的底层机制讲透，不去罗列 API 怎么调，而是回答它为什么这么设计、凭什么让一条管道不用写额外代码就自动获得流式、并行和 trace 能力。

Runnable 协议：一切皆可调用，编排不必认识组件

Runnable 是 LangChain 1.x 的基石抽象，本质就是一个"可调用单元"的最小契约。它规定了任何组件只要实现了一组方法，就能被统一的编排器调度。

这个契约包含同步、批量、流式三套接口，以及一个对应的异步版本：

fromlangchain_core.runnablesimportRunnable# ChatPromptTemplate、ChatOpenAI、StrOutputParser 全都是 Runnableprompt=ChatPromptTemplate.from_messages([("system","你是一个架构师")])model=ChatOpenAI(model="gpt-4o")parser=StrOutputParser()# 它们共享同一组接口，调用方式完全一致prompt.invoke({"question":"什么是注意力机制？"})# RunnableSerializablemodel.invoke([HumanMessage(content="hi")])# BaseChatModelparser.invoke("原始模型输出文本")# BaseTransformOutputParser

• 统一接口降低耦合编排器只依赖 Runnable 这个抽象，不依赖任何具体类型。今天把 ChatOpenAI 换成本地 Ollama，明天在中间插一个检索器，编排代码一行都不用改。
• 批量与流式是协议的一部分batch和stream不是某个组件的特例，而是所有 Runnable 天生具备的能力。组件作者只要实现好最基本的invoke，框架能基于它推导出批量与异步。
• 异步是平行的镜像每个同步方法都有一个a前缀的异步版本（ainvoke、abatch、astream），签名一一对应，让同步写法和异步写法可以无缝切换。

关键点在于，这套协议不是给开发者看的文档约定，而是框架运行时真正依赖的类型边界。当一个函数接收到任何一个 Runnable，它不需要isinstance去判断"这是模型还是解析器"，它直接调.invoke()就行。

LCEL 管道：一个竖线串起 prompt、模型与解析

LCEL（LangChain Expression Language）的精髓就藏在|这个运算符里。当你写下prompt | model | parser，Python 会把它翻译成对Runnable.__or__的调用，返回的是一个RunnableSequence对象。

chain=prompt|model|parser# 上面这行，等价于显式构造：chain=RunnableSequence(first=prompt,middle=[model],last=parser)# 调用方式和单个 Runnable 没有任何区别answer=chain.invoke({"question":"Transformer 和 RNN 的核心区别是什么？"})

• 管道本身还是 RunnableRunnableSequence继承了 Runnable 契约，所以它能继续被|接上新的组件，也能被放进RunnableParallel、被with_retry包裹。组合是可以无限嵌套的。
• 数据沿管道单向流动上一步的输出自动成为下一步的输入。prompt渲染出 messages，model消费 messages 产出 completion，parser把 completion 文本转成结构化结果，全程没有一行胶水代码。
• 类型在组合时被推导LCEL 会在构造阶段校验上一步的输出类型能否喂给下一步。拼接时就报错，而不是等到运行时才发现类型对不上。

这里最容易误解的是以为|只是语法糖。它确实是语法糖，但这个糖的价值在于统一了"组合"这件事的写法。无论你串的是线性链、并行分支还是回退逻辑，你用的都是同一套|和同一套 Runnable 方法。

自动继承的能力：契约上的免费红利

为什么 LCEL 管道不用写额外代码就能流式输出、能异步、能上 trace？答案回到 Runnable 契约。因为RunnableSequence知道自己的每一步都是 Runnable，它就可以通过递归调用每一步的stream方法，把流式能力一级一级透传下去。

# 流式：链式调用链天然支持，无需任何额外封装forchunkinchain.stream({"question":"用一句话解释向量数据库"}):print(chunk,end="",flush=True)# 异步：换个方法名就能跑在事件循环里result=awaitchain.ainvoke({"question":"..."})# 批量：一组输入自动并发处理results=chain.batch([{"question":q}forqinquestions])

• 流式是递归下沉的chain.stream调用prompt.stream，其输出喂给model.stream，model 的每个 token 再通过parser.stream吐出来。每一步只管实现自己的流式，组合层负责把它们接起来。
• tracing 是契约的副产品每个 Runnable 在执行时都会自动 emit 出结构化事件（run start / run end / token），LangSmith 这类观测工具只要接住这些事件，就能画出完整调用树。你没写过一行 trace 代码，但因为组件都遵循 Runnable，追踪信息天然就产生了。
• 能力来自协议而非实现如果一个组件不实现stream，它顶多退化为"一次性返回"，但绝不会让整条链崩溃。契约保证了能力的"渐进可用"。

这点极度重要：流式、异步、追踪这些横切能力，在传统写法里往往要每个组件单独手搓一遍。而在 LCEL 里，它们是契约的红利，你接入一个新组件，这些能力自动就位。

组合优于继承：分支、并行与回退都是 Runnable

Runnable 设计的哲学是组合优于继承。你不需要为了"一个输入要同时做检索和翻译"去写一个ParallelRetrievalAndTranslateChain子类，而是用RunnableParallel把多个 Runnable 并排拼起来。

fromlangchain_core.runnablesimportRunnableParallel,RunnablePassthrough# 一个输入扇出成两个分支，结果自动合并成 dictchain=(RunnableParallel({"context":retriever,# 分支一：检索知识"question":RunnablePassthrough(),# 分支二：原样透传问题})|prompt|model|parser)

• RunnableParallel 接收 dict 配置每个 key 是一个分支，value 是一个 Runnable。输入会被复制分发到每个分支，各分支独立执行，最终把结果组装成同 key 的 dict 交给下一步。
• RunnablePassthrough 是原样透传器当某个分支需要把原始输入原封不动地往下传（比如把用户问题直接塞进 prompt 的question字段），用RunnablePassthrough()占位即可，它本身就是个最朴素的 Runnable。
• 回退与重试同样是一层 Runnablewith_fallbacks和with_retry返回的都是包裹了原 Runnable 的新 Runnable，所以它能继续参与任意组合。下面的容错写法，本质上和prompt | model没有区别。

这种"万物皆可用|拼"的体验，根源就在于所有组合原语自身也是 Runnable。当你给RunnableSequence再|一个RunnableParallel，框架不会觉得你在干奇怪的事，因为它看到的全是同一类对象。

反例：手搓 streaming 和重试，是在为未来还债

讲完正路，看看不走 LCEL 的人是怎么把自己困住的。一个常见反模式是把"链路"写成一串硬编码的函数调用，每加一个环节就改一次主流程，流式、重试、兜底全靠手搓。

# 反模式：自己拼一条"链路"，每一步都手写作业defmy_bad_chain(question:str):# 1. 手搓模板渲染，没有统一接口prompt_text="问题："+question# 2. 手搓流式累加，缓冲逻辑散落在业务里full_response=""fortokenincall_llm_streaming(prompt_text):full_response+=tokenprint(token,end="")# 3. 手搓重试，try / except 包一层又一层forattemptinrange(3):try:returnparse_output(full_response)exceptParseError:ifattempt==2:raisetime.sleep(2**attempt)# 4. 手搓降级？根本没写，模型挂了整个接口 500

这段代码的麻烦不是"能跑不能跑"，而是每一项横切能力都要重复实现：

• 流式是复制粘贴出来的每接一个新的模型供应商，你都得重新写一遍 token 累加和缓冲逻辑。而 LCEL 里chain.stream()一行搞定，因为流式是协议层面的事。
• 重试和降级变成胶水上面那个for attempt循环，在 LCEL 里就是model.with_retry(stop_after_attempt=3)。降级更简洁，主模型挂了自动切备胎：

# 正路：重试和回退都是 Runnable 的一层修饰robust_chain=(model.with_retry(stop_after_attempt=3,backoff_strategy="exponential").with_fallbacks([backup_model])|parser)# 流式、异步、tracing 全部自动继承，无需多写一行forchunkinrobust_chain.stream({"question":"..."}):print(chunk,end="",flush=True)

• with_retry 自动重试遇到限流或网络抖动，框架按你配置的策略退避重试，调用方完全无感。
• with_fallbacks 自动降级主模型抛异常，框架按顺序切到备选 Runnable（可以是更便宜的模型、也可以是本地部署的兜底），链路不中断。
• 两者都是 Runnable 修饰器它们返回的新对象依然是 Runnable，所以可以无缝接进任意管道。反模式里那种手搓逻辑，永远只能服务于那一条特定的链路，换个场景又得重写。

组合优于继承这件事，在工程上的回报就在这里：当你把所有能力都表达成"Runnable 的组合"，横切关注点（流式、重试、追踪、降级）就不再是每个组件各自背负的负担，而成了框架基于契约免费发放的红利。