企业官网建设流程全解析

写过一阵子 LangChain 的人，多少都经历过那种“明明只是想让大模型按步骤干活，代码却越写越绕”的崩溃瞬间。LCEL（LangChain Expression Language）和 Runnable 这套东西，就是官方为了终结这种混乱给出的答案。它不只是一种新语法，而是把 LangChain 里所有组件——模型、提示词模板、解析器、检索器、记忆模块——重新统一成一个抽象：一个 Runnable。想通这一点，LangChain 的学习曲线会平缓一大截。这篇就来聊聊 LCEL 和 Runnable 的底层逻辑、实战组合方式、以及我从传统风格迁移过来时踩过的坑。

1. 为什么需要LCEL：回想一下没有LCEL的日子

我第一次用 LangChain 写链路时，用的还是老式的LLMChain加各种Chain子类。那会儿代码写出来，短期项目倒是能跑，但一旦链路变复杂，维护起来真的是灾难。

1.1 早期链式调用的“面条代码”

举个例子，一个最简单的“根据用户问题生成答案并给出相关关键词”的功能，用旧写法差不多是这样：

from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain.llms import OpenAI # 第一部分：生成答案 prompt = PromptTemplate( template="请回答这个问题：{question}", input_variables=["question"] ) answer_chain = LLMChain(llm=llm, prompt=prompt) # 第二部分：提取关键词 keywords_prompt = PromptTemplate( template="从这段文字中提取关键词：{text}", input_variables=["text"] ) keywords_chain = LLMChain(llm=llm, prompt=keywords_prompt)

然后使用的时候，你得手动把第一个链的输出塞进第二个链的输入：

result1 = answer_chain.run(question="什么是Runnable？") keywords = keywords_chain.run(result1["text"])

看似没什么问题，但只要中间多几步——先查数据库、再调用外部API、再做一个条件判断、最后才叫模型——代码就开始变成一团乱麻。每一步的输入输出是哪个 key、哪个变量，全靠人脑记。更麻烦的是，你想给某个中间步骤单独加verbose、加回调、加超时控制，就得层层传递参数，改一处动全身。

这还不是最糟的。旧式Chain体系里，LLMChain、RetrievalQAChain、ConversationChain各有各的接口风格，有人用.run()，有人用.__call__()，有人用apply()，新手根本分不清。组件之间的组合靠的是继承和内部约定，而不是一个统一的行为协议。

1.2 LCEL的设计目标与核心承诺

LCEL 的定位就是把上面这堆麻烦事一次性解决。它的设计目标是：让所有组件都遵循同一个 Runnable 接口，然后用|管道符像拼积木一样把它们串起来。官方文档里的一句定位非常到位——“LCEL 是 LangChain 面向生产环境的底层基础设施”。它不是简单的新写法，而是一整套围绕可组合性、可流式、可异步、可批处理设计的执行体系。

核心承诺有几条：

1. 统一抽象：模型、提示、解析器、检索器、回调处理逻辑，全部是 Runnable。
2. 声明式组合：用|声明数据的流动方向，代码结构等于数据流图，一眼能看懂。
3. 执行能力内置：流式输出、异步调用、批处理，是所有 Runnable 的标配能力，不是你额外写一堆胶水代码去实现。
4. 运行时配置：通过RunnableConfig统一传递回调、超时、并发参数，不污染业务代码。

我当时真正被说服，不是因为官方文档吹得天花乱坠，而是把一段七步链路从旧式写法改成 LCEL 之后，代码从两百行缩到四十行，而且流式和异步几乎是白送的。这种体验，只有自己迁过一次才感受得到。

2. Runnable协议：所有组件都长成一个样子

LCEL 的根基就是 Runnable 协议。理解它的第一步，是忘掉“Chain”这个词，把所有东西都当成一个可执行对象。不管内部逻辑是大模型调用、函数计算、还是网络请求，对外暴露的行为都一致。

2.1 Runnable接口的五大核心方法

一个标准的 Runnable，实现了以下核心方法：

class Runnable(Generic[Input, Output]): def invoke(self, input: Input, config: RunnableConfig | None = None) -> Output: ... def batch(self, inputs: list[Input], config: RunnableConfig | None = None) -> list[Output]: ... def stream(self, input: Input, config: RunnableConfig | None = None) -> Iterator[Output]: ... async def ainvoke(self, input: Input, config: RunnableConfig | None = None) -> Output: ... async def abatch(self, inputs: list[Input], config: RunnableConfig | None = None) -> list[Output]: ... async def astream(self, input: Input, config: RunnableConfig | None = None) -> AsyncIterator[Output]: ...

也就是说，不管里面是什么，使用者都可以用同一种方式调用它：同步单次调用invoke，同步批量batch，同步流式stream，以及对应的三种异步版本。写上层业务代码的人，根本不用关心组件内部是干什么的，只需要掌握这六个方法。

我第一次意识到这个设计多好用，是在写单元测试的时候。我想 mock 掉一个大模型调用，不需要去构造一个假的 LLM 对象，只需要写一个实现了invoke方法的小类，扔进链路里就能跑。这在旧式 Chain 体系里想都不敢想。

2.2 常用Runnable类型盘点

LangChain 里现成的 Runnable 类型，本质上有这么几类：

这里要特别注意RunnableParallel这个命名。它不是让你在同一时刻调用多个大模型“并行计算”，而是描述一种数据流的并行布局。真正底层是否并发执行，取决于你调的是invoke还是batch，以及跑在什么线程池上。这个坑我在后面专门讲。

2.3 RunnableConfig：运行时参数传递的秘密通道

invoke(input, config)里的config参数是另一个容易忽略的关键点，它是 Runnable 体系里的运行时上下文通道。

from langchain_core.runnables import RunnableConfig config = RunnableConfig( tags=["example-tag"], metadata={"request_id": "abc123", "topic": "LCEL"}, callbacks=[my_callback_handler], recursion_limit=5, ) result = chain.invoke({"question": "你好"}, config=config)

RunnableConfig里的tags、metadata会随着链路的传递，被注入到每个组件的内部。最常见的用途就是做全链路日志追踪：你只需要在入口处传入一个request_id，链路里任何一个组件（包括模型调用）触发回调时，你都能在回调事件里拿到这个request_id，然后把它和日志、监控系统关联起来。

我之前做过一个模拟项目，链路总共四层，线上偶尔会出现某个子步骤超时。用旧式写法排查超时来源，得在每个子步骤里手动加日志字段。换成 LCEL 之后，只要在入口统一注入metadata，借助回调日志一次性就能看到每个环节的耗时分布。这个能力在生产环境的价值，怎么强调都不过分。

3. 用LCEL组装复杂链路：语法糖背后的执行逻辑

觉得 LCEL 只是“用管道符把代码写短一点”，那是没抓住精髓。|背后是一套明确的数据流执行逻辑，搞清楚它，组装复杂链路才不至于两眼一抹黑。

3.1 管道符到底做了什么

在 LCEL 中，|运算符不是简单的函数嵌套，它的语义是：左侧 Runnable 的输出，作为右侧 Runnable 的输入。

from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt = ChatPromptTemplate.from_template("请用一句话解释：{topic}") model = ChatOpenAI(model="gpt-4o-mini", temperature=0) # LCEL 链 chain = prompt | model | StrOutputParser()

这段代码的执行逻辑非常清晰：先给prompt传入一个包含topic字段的字典，生成一条消息；把这条消息交给model，得到AIMessage对象；再交给StrOutputParser，提取其中的字符串内容。

那prompt本身是怎么接收输入的呢？当你调用chain.invoke({“topic”: “LCEL”})时，它是把字典丢给第一个组件。ChatPromptTemplate作为一个 Runnable，内部会自动对齐输入字典和模板变量。如果模板里有topic，输入里没有，就会立刻抛出异常，提示缺少字段。这种在入口处显式校验的设计，比中间静默传错字段要友好得多。

管道符还支持多个组件连成一串，比如：

chain = prompt | model | StrOutputParser() | some_lambda | another_parser

每一步的输出只要和下一步的输入类型兼容，就能继续往下串。也可以在任意一步接RunnableBranch做条件分流，接RunnableParallel做数据合并。

3.2 RunnableParallel：并行不是你想的那种并行

看名字，很多人以为RunnableParallel是让两个大模型同时在两张卡上跑，实际不是。它的作用是：把同一个输入，同时送给多个 Runnable，最后把各自的输出打包成一个字典。

from langchain_core.runnables import RunnableParallel def extract_keywords(text: str) -> list[str]: # 模拟一个本地关键词提取函数 return [word for word in text.split() if len(word) > 3] write_joke = ChatPromptTemplate.from_template("讲一个关于{topic}的冷笑话") | model | StrOutputParser() write_fact = ChatPromptTemplate.from_template("给出一个关于{topic}的冷知识") | model | StrOutputParser() parallel = RunnableParallel( joke=write_joke, fact=write_fact, keywords=lambda topic: extract_keywords(topic), ) result = parallel.invoke({"topic": "程序员"}) # result = {"joke": "...", "fact": "...", "keywords": [...]}

RunnableParallel按命名字段把各个子 Runnable 包起来，调用时统一接收同一个输入。最终输出是把每个分支的结果拼成一个字典。这种模式特别适合：从同一个问题中提取多个维度的信息、做多路答案生成再汇总排序、检索与生成并行做。

但注意：如果你调用的是parallel.invoke(...)，这些子 Runnable 默认是在同一个线程里按顺序执行的，并不会真的并发。要真正的并发执行，需要走异步接口parallel.ainvoke(...)，或者在 batch 场景里让底层线程池调度。所以“并行”指数据流层面的并行，不是资源执行层面的并行。

3.3 RunnableLambda和RunnablePassthrough：塞进去自研逻辑

链路里不能只有模型调用，还得有本地函数、判断逻辑、数据清洗。RunnableLambda就是把普通函数转换成一个 Runnable 的最直接方式。

from langchain_core.runnables import RunnableLambda def clean_text(text: str) -> str: text = text.replace("\n", " ").strip() return text def route_by_length(text: str) -> str: return "long" if len(text) > 100 else "short" chain = prompt | model | StrOutputParser() | RunnableLambda(clean_text)

有个细节：RunnableLambda包装的函数，输入输出类型是可以随意定义的。它不像模型那样要遵守消息格式。如果你想接收字典输入，那就声明def func(data: dict) -> str，LCEL 会自动把上一步的输出解包传进去。假如函数只声明了一个参数，而输入是一个字典，它会把整个字典当作这个参数传给函数。想拆开来传，得用键名对应。

RunnablePassthrough则是用来透传数据的。它的典型场景是：你不想让中间某个组件的输出把原始输入顶掉，而是想让原始输入继续往下传。

from langchain_core.runnables import RunnablePassthrough, RunnableParallel chain = ( {"question": RunnablePassthrough()} # 保留原始问题 | RunnableParallel( answer=prompt_for_answer | model | StrOutputParser(), keywords=prompt_for_keywords | model | StrOutputParser(), ) )

上面这行链路的第一个组件{"question": RunnablePassthrough()}，本身就是一个RunnableParallel的简写。它从输入字典里取出question字段，原样传给下一步。这种做法在实现“回答问题时顺便提取关键词”这类需求时非常顺手，因为你不用在函数里手动维护状态。

3.4 条件路由与异常降级

业务链路里常有“如果……就走分支 A，否则走分支 B”的需求。LCEL 里用RunnableBranch做规则路由，用.with_fallbacks()做异常降级。

from langchain_core.runnables import RunnableBranch branch = RunnableBranch( (lambda x: x["type"] == "tech", tech_chain), (lambda x: x["type"] == "daily", daily_chain), default_chain, )

它按顺序检查每个条件，第一个返回True的就执行对应的链，全部不满足则走 default。注意条件函数接收的是上一步的输出对象，所以类型问题要提前对齐。

降级则更实用。比如主模型调用可能因为限流报错，你可以给它配一个备用链路：

robust_chain = model_primary.with_fallbacks([model_backup])

调用主模型失败时，自动尝试备用模型。还可以配合RunnableLambda包装一层降级之后的日志记录，方便观察触发了多少次降级。这套机制在真实的线上小项目中，比你自己写 try-except 再手动切换要干净得多。

4. 迁移实战：把一段传统代码改造成LCEL风格

理论知识说得再多，不如拿一段真实代码改一遍来得直观。下面这段是我以前在模拟项目 X 里写的一段传统链路改造过程，业务需求很简单：接收用户问题，先做一个本地规则判断，再决定是直接查库回答，还是让大模型生成。

4.1 传统代码的痛点

用旧式风格，大概是这样的结构：

def answer_question(q: str, kb): # 第一步：本地规则判断 if "价格" in q: return kb.get_price_info() elif "联系方式" in q: return kb.get_contact_info() else: # 第二步：走大模型 prompt = PromptTemplate( template="你是客服助手，请回答：{question}", input_variables=["question"], ) chain = LLMChain(llm=llm, prompt=prompt) return chain.run(question=q)

功能没问题，但问题很多：业务判断和模型调用耦合在同一个函数里，没法复用；要想给模型调用部分加上流式输出，得把这个函数改成 generator，改完整个调用方的代码都得变；想并发行多个问题的时候，还得自己维护线程池。每加一个新需求，函数就膨胀一分。

4.2 改造过程逐步拆解

第一步，把判断逻辑做成RunnableLambda：

from langchain_core.runnables import RunnableLambda def local_rule(q: str): if "价格" in q: return {"answer": kb.get_price_info(), "source": "local"} if "联系方式" in q: return {"answer": kb.get_contact_info(), "source": "local"} return {"question": q, "source": "llm"}

返回一个字典而不是纯字符串，是为了下一步能识别来源。

第二步，把大模型通路做成一条子链：

llm_prompt = ChatPromptTemplate.from_template( "你是客服助手，请基于以下知识库内容回答用户问题。\n知识库：{kb_content}\n问题：{question}" ) llm_chain = llm_prompt | model | StrOutputParser()

第三步，用RunnableBranch把两条路合成：

from langchain_core.runnables import RunnableBranch def is_local_result(data: dict) -> bool: return data["source"] == "local" handle_local = RunnableLambda(lambda data: data["answer"]) handle_llm = ( RunnablePassthrough() | llm_chain | RunnableLambda(lambda answer: {"answer": answer, "source": "llm"}) ) final_chain = ( RunnableLambda(local_rule) | RunnableBranch( (is_local_result, handle_local), handle_llm ) )

第四步，调用方式统一：

result = final_chain.invoke("你们的产品多少钱") print(result)

这版代码比旧版更长一些，但每个环节的功能边界非常清楚：local_rule只负责规则判断，RunnableBranch只负责路由，handle_llm里每一步都可以单独替换或测试。将来如果需要给handle_llm加缓存、加统计、加备用模型，都不需要动外面的调用方。

4.3 改造后的收益分析

改造完最大的感受，是你能在链路的任意一个节点插入观测代码，而不破坏原有结构。比如我想统计“多少比例的问题走了本地规则、多少比例走了大模型”——旧版代码我得在else分支里手动埋点；新版我只需要在handle_llm的末尾加一个RunnableLambda把结果丢进计数器。想加缓存，就在local_rule外面包一层带 cache 的 Runnable；想压测，直接调用final_chain.batch([...])，接口层面就支持批量。

对一个长期维护的项目来说，这种结构化带来的收益远超过最初多写的那几行代码。代码是要给三个月后的自己看的，LCEL 的声明式写法几乎就是一张可读的数据流图，比读流程函数舒服太多。

5. 流式、异步与批处理：LCEL最容易被低估的三件套

LCEL 的实用价值不只是简化写法，更在于它内建了三种执行能力。这三个能力在真实项目里解决过很多让我头疼的问题。

5.1 流式输出的内部机制与实现

传统 LLMChain 想做大模型输出一个字一个字往外蹦的效果，得手动从底层 API 拿 stream 响应，再做回调。LCEL 里，只要整条链支持流式，chain.stream()直接就能用。

for chunk in chain.stream({"topic": "LCEL"}): print(chunk, end="", flush=True)

这里的chunk类型取决于链的最后一个组件的stream实现。如果最后是StrOutputParser，chunk就是字符串片段；如果最后是模型本身，chunk可能是AIMessageChunk。组装链的时候，最好在末尾加上解析器，这样流式输出的数据形态对前端更友好。

有个容易踩的坑：如果链路里某些步骤是RunnableLambda包装的普通函数，它本身不支持流式输出，LCEL 会自动退化成“先等这个函数执行完，再继续往下流”。也就是说，链路里插了一个大计算量的本地函数，流式实时性会被这个函数卡住，不是所有步骤都能做到逐字输出。理解这个机制，就不会在生产环境里对流式效果有不切实际的期待。

5.2 异步接口与真实使用场景

异步化在本地的收益不明显，在 Web 服务里却是刚需。用 FastAPI 接 LCEL 链时，同步的invoke会阻塞事件循环，导致接口并发能力大幅下降。正确做法是用异步接口：

from fastapi import FastAPI app = FastAPI() @app.post("/chat") async def chat(request: dict): question = request["question"] result = await chain.ainvoke({"question": question}) return {"answer": result}

我自己做过一个简单的流式聊天接口，用astream把 token 逐块推给前端：

@app.post("/chat/stream") async def chat_stream(request: dict): question = request["question"] async def generate(): async for chunk in chain.astream({"question": question}): yield f"data: {chunk}\n\n" return StreamingResponse(generate(), media_type="text/event-stream")

如果链路里有真正耗时的非模型操作（比如查数据库），建议在RunnableLambda里用asyncio.to_thread转成线程执行，避免阻塞事件循环。这是我在并发测试中实际踩到的性能瓶颈。

这里还要提醒一句：不要在一个异步函数里用chain.invoke()，它会阻塞事件循环，应该是async环境下调用对应ainvoke。很多刚切到 LCEL 的人会忽略这一点，导致网页并发一上去就卡死。

5.3 批处理的实测效果

batch方法适合一次处理多条同类型请求的场景，比如离线批量生成摘要、批量分类问题。

questions = [ "什么是 Runnable？", "LCEL 和 Chain 有什么区别？", "如何给大模型链路加缓存？", ] results = chain.batch([{"question": q} for q in questions])

底层值得关注：batch默认会执行一个内部线程池，把多个输入并行丢给各个 Runnable 组件。并发度默认是多少？LangChain 的默认行为是用一个线程池执行器，但不同组件类型的实现略有差异，不保证全部并发。如果你对吞吐有硬性要求，可以显式传max_concurrency来控制：

results = chain.batch( [{"question": q} for q in questions], config={"max_concurrency": 5}, )

这个参数控制同一时间最多跑几个请求。不是越高越好——很多模型 API 有速率限制，并发开到 20 会导致大量 429。实测下来，对单模型 API 的场景，max_concurrency设在 5 到 8 之间通常比较稳，再高就该触发限流了。这里的数值需要你自己压测调整，没有统一标准。

批处理里还要注意一个问题：RunnableParallel内部如果有多个子分支，各自内部可能会再次起线程池，导致整体并发数叠加。监控线上服务的线程数和 API 调用频率，是最直接的验证方式。

5.4 关于并行度的一个易踩坑点

这个坑值得单独拿出来讲。我在一次批量任务里，用了RunnableParallel做了两条模型调用分支，然后在外层调batch跑了 20 条问题。我预想的是 20 个输入 × 2 条分支 = 40 个并发请求，结果实际被限流了。

排查后才发现：batch的默认并发基数跟max_concurrency的语义在不同 Runnable 之间并不是简单的乘法关系。有的实现会把并发限制直接应用到所有嵌套 Runnable 的顶层，有的则是各自独立维护线程池。这导致一个很尴尬的现象——你预期并发是 40，实际可能是 10 或者 80，完全是撞运气。

后来我不再依赖嵌套并发，而是按“整体请求量”来控制：

1. 先在外层调用batch时，把max_concurrency设成目标并发值的一半。
2. RunnableParallel内部需要真正并行的地方，主动把分支数控制到最少。
3. 给模型调用加上一个自定义限流器，用一秒钟最多几个请求的令牌桶做兜底。

这样即使底层并发调度行为有差异，顶层的令牌桶也能把 API 请求速率压住，避免被限流打得措手不及。这个经验对任何一个要上生产的 LCEL 项目都值得提前备着。

6. 几块藏得比较深的绊脚石

内容不少了，最后集中把几块“看着简单但实际会卡人”的地方列出来，都是我实际写代码时亲测过的。

6.1 RunnableLambda 的参数绑定问题

RunnableLambda包装一个函数时，如果函数定义有多个参数，默认是不会自动从输入字典里拆分传给函数的。它把整个输入字典当作一个参数传进去。要传多个参数，得写成接收单个字典参数，内部自己解包，或者用RunnableLambda(partial(func, something))预绑一部分参数。我一开始在这里绕了很久。

6.2 别把状态存在组件内部

LCEL 链默认被认为是无状态的。要是你在RunnableLambda内部写了一个self.count += 1，并发 batch 时会出现意料之外的共享状态问题。想要计数、缓存、日志，正确的做法是把它放到外部存储，或者通过RunnableConfig的metadata传递状态。否则并发一上去，各种变量互相污染，排查起来非常痛苦。

6.3 默认模型绑定的是旧接口

很多老代码里用的是from langchain.llms import OpenAI，而 LCEL 更适合搭配新的ChatOpenAI风格的模型接口。两者的输入输出格式不同。如果混用，组装链时经常出现“消息格式不匹配”之类的异常。迁移时，统一把模型初始化代码换成新接口，可以在源头上避免很多奇怪问题。

6.4 调试时善用 get_graph

LCEL 提供了一个很实用的调试图方法：

chain.get_graph().print_ascii()

它会把你组装的链以 ASCII 图的方式打印出来，清晰地显示每一步的输入输出关系。排查链路组装错误（比如字段名不对、类型不匹配）时，这个图比盲猜管用太多。我刚学 LCEL 那段时间，几乎每次组装新链都会先print_ascii()确认数据流走向。

写在最后

LCEL 和 Runnable 表面上是语法更新，底层是 LangChain 组件设计的一次重构——从继承体系转向协议体系。对我这种习惯直接上手写业务代码的人，最大的好处是：能把更多精力放在业务逻辑本身，而不是一遍又一遍地写胶水代码。如果你正在学的路上卡住，我建议别急着背 API，先把“所有组件都是 Runnable”这个心智模型立起来，再回头看文档，视野会开阔很多。