企业官网建设流程全解析

写 LangChain 模块系列的时候，我就在想第二篇一定要聊 Chains。原因很简单：大多数人刚开始接触 LangChain，第一个跑通的代码就是prompt | llm | parser这种一条管道，但真到复杂业务里，链的关键不是“把三步拼在一起”，而是把检索、判断、分支、校验、回退这些都塞进同一条工作流里。今天这篇，我围绕“Chains 工作流编排核心”这个主题，把 Runnable 协议、LCEL 表达式、并行分支、条件路由、生产排错，以及老链 API 迁移这些内容完整拆一遍。文章偏工程向，适合已经能写简单 LLM 应用、想把流程真正落到项目里的同学，当然刚入门的人把概念抓起来也不会太难。

1. 单次模型调用不够用：Chain 解决的是“流程编排”问题

1.1 业务里最常见的需求其实都是多步组合

先看个常见场景：你写了一个客户工单系统，用户提交一句“数据库连接超时，重启了也没用”。如果只把这句话丢给大模型，让模型回答“请重启数据库”，这条业务根本没法闭环。真实需求是：先判断工单类型（数据库、网络、权限还是存储），再抽取标题、紧急程度、涉及的设备信息，然后去知识库检索相似故障的操作手册，最后生成回复并校验格式。

这种“先分类、再提取、再检索、再生成”的流程，本质上是一串有依赖关系的步骤。上一步输出的数据要转换成下一步的输入，中间还要并几个分支。如果主代码里写 20 个if-else调用模型，每一处都要处理超时、重试、日志、参数格式化，维护成本会迅速失控。

LangChain 的 Chains 模块就是为这种场景出现的。它把每个步骤都抽象成可组合的组件，步骤之间通过统一接口传递数据。你不需要关心上一个组件内部怎么实现，只需要知道它接受什么、输出什么，然后把它们像拼管道一样接起来。这一步抽象解决的不只是“代码更短”，还有流程的复用、测试、并行和监控。

1.2 “链”和“流水线”的抽象

理解 Chains 最好的方式，是把它想成工厂流水线。一件产品从 A 工位走到 B 工位，每个工位只干一件事，干完后放到传送带上给下一个工位。LangChain早期的LLMChain就是“提示词工位 + 模型工位”的打包，而到了现在，整条链已经演变成更细粒度的流水线：检索器是工位，提示词模板是工位，模型是工位，输出解析器也是工位。

每个工位必须暴露同样的入口，这就是后面要展开说的 Runnable 协议。因为入口一致，你可以任意替换某个工位而不影响其他部分。比如把“向量检索”换成“倒排索引检索”，只要它同样接收一个查询字符串、返回文档列表，整条链的其余部分完全不用改。这种可替换性是 Chains 的价值核心，也是工作流编排的第一原则：组件之间靠约定通信，而不是靠硬编码顺序。

多数新人会忽略的一点是，链的编排并不仅限于串行。很多流程天然是并行的：一边检索知识库，一边分析用户情绪，一边检查参数合法性。这时链还需要支持“分叉再汇合”的结构。Chains 提供了RunnableParallel来处理这种场景，它和流水线的区别在于，传送带在某些节点可以分成多条，最终再汇入同一个提示词模板。

1.3 旧链对象与新式链式的设计差异

如果你看过老教程，会见到SimpleSequentialChain、SequentialChain、LLMChain这类对象。它们确实算“链”，但设计上有一个明显的别扭之处：每个链内部黑箱化太重，调用方式是.run()、.apply()，缺少统一的行为约定，调试时你很难看清楚中间那一步到底返回了什么。

新式链用 LCEL（LangChain Expression Language）重写了这套东西。所谓 LCEL，核心就是管道符|和一组 Runnable 组件。任何一个 Runnable 对象都可以通过|接到另一个 Runnable 后面，形成一条新链，新链本身又是一个 Runnable。这带来一个极大的工程优势：你构建出的链，在另一个地方可以直接作为组件继续被组合，形成链中链。

我在实际迁移项目的感受是，旧的SequentialChain最让人头大的是输入输出映射表。它需要你专门声明input_variables和output_variables，稍不留神就会把变量名写错。LCEL 下你用字典直接控制键的流向，哪里多了一个键、哪个键没传，代码一眼就能看穿。所以这篇我会把重心放在 LCEL 上，旧链 API 只作为迁移对照来讲。

注意：如果你正在维护老项目，先别急着删旧代码，但新功能建议不要再基于LLMChain写了，因为它在新版本中逐步走向废弃，维护成本只会越来越高。

2. 核心机制：Runnable 协议、LCEL 管道与四大组合件

2.1 所有链都遵循统一调用协议

Chains 能编排的核心，是所有对象都实现了 Runnable 协议。你可以把 Runnable 理解成一个“统一插头”，它有四个主要方法：

• invoke(input)：同步调用，传入一个输入，返回一个输出。
• batch(inputs)：批量调用，传入一组输入，返回一组输出。
• stream(input)：流式调用，一边生成一边产出 chunk。
• ainvoke/abatch/astream：对应的异步版本。

这套协议的意义在于标准化。无论你接的是模型、提示词模板、检索器、输出解析器，还是一个自定义 Python 函数，外部都只需要知道它有“插头接口”，不关心内部是调了 API 还是做了本地计算。我经常跟同事说，Runnable 就像 USB-C 接口，长得好不好看不重要，插上能用最重要。

实际写代码时，你可以把一条链当作一个整体来调用：

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # get_model 是你自己实现的工厂函数，返回任意兼容 LangChain 的对话模型实例 llm = get_model() prompt = ChatPromptTemplate.from_template( "把下面这句话翻译成英文：{text}" ) chain = prompt | llm | StrOutputParser() result = chain.invoke({"text": "今天写链写得很顺手"}) print(result)

这里prompt | llm | StrOutputParser()就是一条链，但它本身也是一个 Runnable，所以你可以继续对它调用chain.batch([{"text": "..."}, {"text": "..."}])，甚至async for chunk in chain.stream(...)。这是理解新式 Chains 最重要的心智模型：链是由 Runnable 组成的 Runnable。

2.2 LCEL 管道：用|组合一切

管道符|在 Python 语法里原本是针对集合的操作符，LangChain 重载了 Runnable 的二进制或运算，让代码写起来像 Unix 管道。数据从左流到右，前一个组件的输出成为后一个组件的输入。

用管道组合时，你必须想清楚一个关键问题：上一级输出的结构能不能被下一级接收？比如ChatPromptTemplate期望传入一个“变量字典”，那么它左边的组件就必须保证输出的是字典，或者你的管道起始处直接传入字典。反过来，StrOutputParser输出的是普通字符串，如果后面再接一个需要字典的组件，就会因为“键值不匹配”报错。

很多人第一次写链时遇到的TypeError/ValueError，基本都是这个原因。我自己的调试习惯是：先把链拆成单步，分别调用中间组件的invoke，观察每层返回的数据类型和内容，确认无误后再组起来跑。这一句话能省掉你大量排查时间。

为了适配结构，LangChain 提供了几个“形状转换”组件，最常用的就是RunnablePassthrough和RunnableParallel。

2.3 并行与旁路：RunnableParallel 和 RunnablePassthrough

RunnablePassthrough的字面意思是“原样传过去”。它有两个典型用法。第一种，输入是字典时，它会把字典原封不动地透传到下一步；第二种，配合RunnablePassthrough.assign(...)，可以在不丢掉现有字段的情况下添加新字段。

举个例子，一个客服工单流程需要同时知道“原始描述”和“已提取的故障类型”：

from langchain_core.runnables import RunnablePassthrough, RunnableParallel def extract_type(desc: str) -> str: # 这里可以调用分类模型或规则判断 return "数据库故障" chain = RunnableParallel( original= RunnablePassthrough(), fault_type=RunnableLambda(extract_type), )

RunnableParallel接收一个字典，每个值都是一个 Runnable 或普通可调用函数。调用一次invoke，就会并行执行这些分支，最后把每个分支的结果按字段名汇聚成一个新字典。这里的“并行”是逻辑并行，如果你的组件都是基于 IO 的，比如同时向两个模型服务发请求，那么整体耗时可以显著缩短。

注意，RunnableParallel的字段名决定了输出字典的键。这些键往往会作为提示词模板里的变量被引用。所以字段命名要跟模板变量保持一致，否则数据会断在链中间。

2.4 函数封装：RunnableLambda 的使用边界

业务代码里永远有大量自定义逻辑，比如格式化文档、校验日期、调用内部接口。要把这些普通函数放进链里，就必须用RunnableLambda包装。

from langchain_core.runnables import RunnableLambda def format_docs(docs): return "\n\n".join(doc.page_content for doc in docs)

为什么不能直接把format_docs放在|后面？因为管道符要求对象是 Runnable，而普通函数没有统一的调用协议。RunnableLambda的作用就是给函数“套上插头”，让它可以被invoke、被组合、被异常捕获统一管理。

但RunnableLambda也有它的边界。如果函数内部太复杂，比如异步调用、数据库事务、长时间重试，那就不适合硬塞进链里。你应该把这类逻辑抽象成独立的 Runnables，或者在外面包一层 like a “gate”。否则链变得黑箱，调试时你根本分不清问题出在模型还是自定义函数里。

3. 典型场景实操：RAG、结构化抽取与条件路由的完整链

3.1 RAG 问答链：把检索器变成链的一环

RAG（检索增强生成）是 Chains 最常见的落地场景。以前把检索逻辑写在主程序里，每次调用都要手动拼接上下文，现在检索器本身就可以作为一个 Runnable 放进管道。

一个最小可用的 RAG 链可以这样写：

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough, RunnableLambda # 假设 vector_store 已经建好，as_retriever 返回一个可调用的检索器 retriever = vector_store.as_retriever(search_kwargs={"k": 4}) def format_docs(docs): return "\n\n".join([doc.page_content for doc in docs]) prompt = ChatPromptTemplate.from_template( "根据下面资料回答问题。\n资料：\n{context}\n\n问题：{question}" ) rag_chain = ( { "context": retriever | RunnableLambda(format_docs), "question": RunnablePassthrough(), } | prompt | llm | StrOutputParser() ) answer = rag_chain.invoke("LangChain 的 Chain 是什么？")

这段代码的关键在输入构造：最外层字典有两个字段。context分支把用户问题传给retriever，检索结果经过format_docs变成纯文本；question分支则用RunnablePassthrough()把原始问题原样留出。此时两条分支汇聚成一个字典，正好填充提示词模板里的两个变量。

这个并行结构在实践中比串行写法快很多。因为“检索上下文”和“保留原始问题”互不依赖，不需要等前一步完成再走下一步。你甚至可以再多加一个分支，比如同步做一次敏感词检查，把结果也塞进提示词里，让模型在生成时把不安全内容剔除掉。

注意：检索器接收的输入和用户问题保持一致，这是默认设定。如果你的查询要经过改写才能进检索器，那就在context分支前再接一个RunnableLambda，把“改写查询”也做成链的一环。

3.2 结构化信息抽取链：保证模型输出符合预期

很多同学被“模型输出脏数据”折腾过：明明要求返回 JSON，结果模型先输出了一段解释文字再给 JSON，或者字段名不一致。单纯靠提示词“你只能返回 JSON”并不可靠，真正稳妥的做法是让链的输出走“结构化模式”。

LangChain 里，很多模型接入都支持with_structured_output()，它会让模型以结构化方式返回数据，直接对接一个校验模型。下面是一个工单信息抽取的例子：

from pydantic import BaseModel, Field class TicketInfo(BaseModel): category: str = Field(description="故障类型") title: str = Field(description="工单标题") urgency: int = Field(ge=1, le=5, description="紧急程度，1最低5最高") prompt = ChatPromptTemplate.from_template( "从以下用户描述中提取工单信息：{description}" ) extract_chain = prompt | llm.with_structured_output(TicketInfo) ticket = extract_chain.invoke({"description": "数据库连接超时，已经持续半小时，影响前台收银。"}) print(ticket.title, ticket.urgency)

这里的设计意图很明确：把“输出必须符合 schema”这件事从提示词层的“软约束”变成代码层的“硬约束”。即使模型偶尔多说了几个字，解析器也会按字段定义进行抽取，最终拿到的是一个带类型的、可校验的对象。

如果模型接入不支持with_structured_output，退而求其次的做法是让模型输出 JSON 文本，再接一个RunnableLambda做json.loads和字段校验。但这样做要注意两点：一是必须处理 JSON 解析异常，二是模型可能漏掉必填字段，校验逻辑要能给出友好错误信息，配合重试链重新生成一次。

3.3 分类路由链：分支判断决定走哪条子链

工作流不止一条路。比如智能客服里，“查文档类问题”和“问价格类问题”应该走不同的提示词模板，甚至接不同的知识库。这时候需要RunnableBranch，它像一个岔路口，根据条件把输入定向到不同子链。

from langchain_core.runnables import RunnableBranch doc_chain = doc_prompt | llm | StrOutputParser() price_chain = price_prompt | llm | StrOutputParser() default_chain = default_prompt | llm | StrOutputParser() def is_doc_query(x: dict) -> bool: return "文档" in x["question"] def is_price_query(x: dict) -> bool: return "价格" in x["question"] router = RunnableBranch( (is_doc_query, doc_chain), (is_price_query, price_chain), default_chain, ) full_chain = {"question": RunnablePassthrough()} | router

RunnableBranch接收一组“条件 + 链”的配对。它从上到下依次执行条件函数，第一个返回True的配对会接管后续调用。如果全部不匹配，就走最后的default_chain。这一点如同if-elif-else，但与手动写条件语句不同的是，你仍然保持整条链作为一个整体来调用和复用，还能方便地在分支前后插入其他组件。

一个容易被忽略的坑：条件函数的入参是整个链在分支位置的输入。如果你在RunnableBranch前面用RunnablePassthrough拼装了字典，那条件函数收到的就是字典，不是裸字符串。所以上面的示例里is_doc_query写成x["question"]，而不是直接判断"文档" in x。

如果你想让模型来做意图分类，而不是靠关键词，我建议把“意图识别”单独做成一个子链，先让模型输出一个intent字段，再用RunnableLambda读取该字段进行分支。这样意图判断和内容生成解耦，可调试性会好很多。

3.4 链中套链：把子流水线组合成大流程

链的价值在于可组合性。一条已经建立好的rag_chain，可以作为一个组件进入更大的工作流。比如完整的客服流程是“意图识别 → 工单抽取 → 知识库检索 → 生成回复 → 质检评分”，每一段都是独立子链，大链只是负责把它们串起来。

pipeline = ( intent_chain | RunnableLambda(lambda r: build_ticket_query(r)) | assign_ticket_chain | rag_chain | quality_review_chain )

这种链中套链的模式有一个明显好处：你可以单独训练、单独测试每个子链。比如intent_chain输出错误，不会牵连到rag_chain。线上出问题时，把日志里上一级链的输出拿出来，直接喂给下一级链单独复现，定位速度会快很多。

组合链的另一个好处是“资源隔离”。检索链需要并发能力强，生成链需要重试策略，质检链需要流式响应。你可以为每个子链配不同的参数，再统一在大链上做编排，代码结构还是清晰的。

4. 生产环境避坑：流式、并发、回退与调试

4.1 流式输出与解析器的配合陷阱

流式输出在生成型应用里是刚需，用户不喜欢等 10 秒才看到完整回答。LCEL 的stream方法理论上支持流式，但不是所有组件都能流式传输。

体验最明显的几点：

• StrOutputParser可以逐 token 吐出文本，适合聊天场景。
• JSON 解析器往往需要“等到完整 JSON 结束”才能解析，因为中间分片无法构成合法 JSON。如果你在一个需要快速流式展示的场景里用了结构化输出，大概率会感觉“好像完全没流式”。
• 自定义RunnableLambda如果做了全量计算，它天然无法分片返回，只能等函数跑完再吐结果。

我通常这样处理：生成回复的最终阶段用StrOutputParser，让用户先看到文字流；同时把结构化校验和入库操作放到一个异步的后置分支里，用RunnableLambda接收最终文本，入库或做质量检测。不要让“审计逻辑”阻塞“用户可见内容”。

4.2 批处理并发与限流：设置合理参数

batch方法能一次处理多个输入，底层使用线程池，可以大幅提升吞吐。但大模型服务一般都有限流，你一口气发 50 个请求，很容易触发 429，反而比串行更慢。

我在实战中会把批处理包装成带并发控制的函数，不要直接把所有输入丢给batch：

from concurrent.futures import ThreadPoolExecutor def limited_batch(chain, inputs, max_workers=4): with ThreadPoolExecutor(max_workers=max_workers) as pool: return list(pool.map(chain.invoke, inputs))

这只是一个降级方案。如果你的模型服务商支持批量接口，优先用服务商的批量能力；如果不支持，再自己控制并发。并发数的设置要看模型服务的 TPM/RPM 配额，一般从 4 到 8 开始试，稳定后再往上加。

4.3 失败处理：回退链和重试策略

生产环境没有“不会失败的链”。网络抖动、模型超时、JSON 解析失败都可能在任何一步出现。LCEL 提供了一些组合工具，其中最实用的是回退链。

primary_chain = prompt | llm | parser fallback_chain = simple_prompt | llm | StrOutputParser() robust_chain = primary_chain.with_fallbacks([fallback_chain])

这段代码的意思是：主链一旦抛异常，自动改用备用的简化链。备用的模型可以是不同供应商、不同参数，也可以是更保守的模板。通过回退，你不需要在业务代码里到处写try-except，链自己就能处理部分故障。

更细粒度的重试可以用with_retry()，指定重试次数和退避策略。但要注意，重试只适合“可重试”的异常，比如超时、限流。模型返回内容不合法这类问题，通常重试几次也一样，须得换模板或加解析兜底。

4.4 排查手段与常见问题速查

调试链，我依赖两个手段：第一是拆链单步执行，第二是看图。LangChain 的链对象上有get_graph()，可以输出整个链的结构，帮助确认连接关系。还有一个笨但有效的办法：在每个RunnableLambda里加打印，把输入输出打到日志里。

常见的几个坑，我整理成了一张速查表，方便你直接对照：

注意：确认链的“形状”比确认链的“效果”更重要。一个字段名不匹配的输入，模型再强大也没用。每当你准备在链上再加一层组件时，先问自己：这一层接收什么？吐什么？下一层能不能消化？想清楚了再动手，这就是最省钱的经验。

5. 从旧版 Chain 迁移到 LCEL，以及和 Agent 的边界

5.1 旧版链被弃用的根本原因

老版本里的LLMChain、SimpleSequentialChain、SequentialChain在概念上没有任何错误，但在工程实践里很别扭。

以SequentialChain为例，它要把多条链串起来，必须手动声明输入输出映射，代码冗长。更多时候，你只能通过.run()传入一个字符串，一旦要多参数输入，就得绕回apply。并行、流式、异常回退这些能力几乎为零。随着 LangChain 的组件越来越多，旧链的代码只能在内部继续叠加特例，维护成本越来越高，所以官方方向是全面转向 LCEL。

我理解这次“链的演进”的主线是：从“大对象封装”走向“细粒度组合”。LLMChain本质上就是把 prompt 和 llm 封死在一起，等于把一个 USB 口焊死在电脑上；LCEL 则是给你一根标准线，你要接显示器还是接音箱随意。对个人项目影响不大，但对跨团队、跨项目的复用是质的差别。

5.2 迁移映射：老 API 到新管道的对照

迁移的时候，不需要在“对象名”上一一对应，而要按“行为”去对应。下面是常用对照：

迁移的核心动作是取消“内部状态”。旧链里memory经常从一个LLMChain传给另一个，导致我不知道状态从哪来。LCEL 的做法是把上下文作为输入字典的字段显式传递，哪个步骤需要历史，哪个步骤生成新历史，都一目了然。这种“少一点魔法，多一点明确”的设计，长期维护时你会明显感受到差异。

5.3 什么时候该上 Agent/有状态编排，什么时候该用 Chain

字符串“Chain”和“Agent”经常被混为一谈，但它们解决的问题边界不同。Chain 适合处理“静态、可预知”的工作流：步骤固定，分支条件明确，不需要模型自己决定下一步做什么。比如 RAG、工单抽取、文本分类，用 Chain 是最稳的，结果可预测、速度也快。

Agent 则适合“动态、不可预知”的任务：模型需要自己规划工具调用顺序，可能需要多轮试探。典型场景是“帮用户订机票”，你需要查询航班、比价、下单、确认，过程中随时可能改变计划。这种场景如果用 Chain，你就得把所有可能分支写死，逻辑会非常庞大。

另一个容易混淆的是 LangGraph 这类有状态图编排。它把每个节点之间的状态显式维护，支持循环和回退，适合接近“状态机”的任务。Chain 则是无环的静态图为主，或者说是“可以嵌套分支但有方向”的流。我的判断标准很简单：如果你的流程里包括“根据结果反复修改输入”的循环，优先考虑图编排；如果流程是一条单向流水线加几个条件分支，Chain 仍然是最合适的选择。

5.4 我的取舍建议

最后分享一点个人经验。过去踩过最大的坑，就是把可以简单组合的链写成“一个巨大的 Agent 图”。结果是，模型经常猜错下一步该做什么，反而弄得不如静态链稳定。现在我做任何模型工作流，都会先画出简陋的主流程，再判断里面有没有明显的循环、有没有模型自主决策需求。如果没有，就直接用 Chain 组件化；如果只有 1 个循环点，我也会先尝试用链加回退模拟，而不是立刻上整体 Agent。

给新手的建议是：先把你目前所有工作流拆成最小可测试的 Runnable，宁可多拆几层，也不要组装一个“一步到位”的巨链。每次调试都能精确定位到具体组件，这才是提高效率的正道。等你对链的边界把握更熟了，再决定哪些环节要换成 Agent，哪些环节要坚持用 Chain——这种选择的底气，来自于你对“流程本身”的解释力，而不是对某个 API 的熟练度。