说一个特别真实的阶段：Embedding 模型、向量库、Prompt 模板、大模型接口，这些单个组件你都能跑通，但一旦要把它们串成一个真正能用的 RAG 应用，代码立刻变得又臭又长。我最早做问答系统时就是这么过来的——先查向量库，再拼 Prompt，再调模型，中间全靠临时变量传值，改一个环节就得连带改后面所有代码，调试一次恨不得把每个中间结果都 print 一遍。直到我把整条链路切到 LangChain 的 Chain 抽象上去，代码结构才算真正稳定。
这篇是 LangChain+RAG 实战宝典系列的第十篇，聚焦链（Chains）的构建与组合，目标是把端到端 RAG 闭环彻底打通。适合已经了解 LangChain 基础组件、想做完整可维护应用的开发者。你会搞清楚 Chain 到底解决了什么问题，以及如何把检索、生成、格式化输出平滑地组合成一条可调试、可扩展的流水线。文中所有代码都基于 LangChain 的 LCEL 表达式语言，这也是目前官方主推的链式写法。
1. 从单次调用到链式编排：Chain 到底解决了什么
1.1 一个 RAG 请求背后其实是一连串动作
很多人刚开始接触 RAG 时，脑子里只有一个印象：把用户问题丢给大模型，让它回答。可实际上，一个完整的 RAG 请求在内部要经历至少四步动作：
- 把用户问题转成向量（Embedding）。
- 拿这个向量去向量库检索，拿到最相关的文档片段。
- 把用户问题和检索结果拼进 Prompt 模板。
- 把完整 Prompt 交给大模型，拿到答案，再做输出解析。
如果再加上引用溯源、多轮对话记忆、重排序（Rerank），动作数量直接翻倍。用最原始的 Python 代码实现，大概是这样的：
# 传统写法的痛点：所有环节都耦合在函数里 def rag_answer(question: str): # 1. embedding query_vector = embed_model.embed_query(question) # 2. retrieve docs = vector_store.similarity_search_by_vector(query_vector, k=4) # 3. build prompt context = "\n\n".join([d.page_content for d in docs]) prompt_text = prompt_template.format(context=context, question=question) # 4. call LLM response = llm.invoke(prompt_text) return response.content这段代码的问题在哪儿？首先是每一步的输入输出全靠手工对接，类型一多就容易乱。其次是没法做流式输出——你只能等大模型全部生成完再一次性返回。最重要的是，一旦你想在中间插入一个重排序环节，或者想对检索结果做过滤，就得改动函数内部结构，整条链路都要重新测试。
1.2 Chain 抽象：把流水线变成一等公民
LangChain 的 Chain 概念，本质上就是把"数据处理流水线"做成了一等公民。你不必再关心每一步之间怎么传值，只需要声明每一步是什么，框架负责把它们拼接起来。
在早期版本的 LangChain 中，Chain 通常是类对象，比如LLMChain、SequentialChain。而到了 LangChain 0.2 之后，官方全力推 LCEL（LangChain Expression Language），用管道符|把多个 Runnable 组件串起来。这种写法更像 Unix 管道：前一个组件的输出，自动成为后一个组件的输入。
from langchain_core.runnables import RunnablePassthrough from langchain_core.prompts import ChatPromptTemplate prompt = ChatPromptTemplate.from_template( "根据以下资料回答问题：\n{context}\n\n问题：{question}" ) chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt | llm | output_parser )看到这个写法的第一眼，很多人会觉得"这不就是链式调用吗？"但背后有个关键区别：LCEL 的每个组件都实现了统一的Runnable接口，有invoke()、stream()、batch()这些标准化方法。这意味着你换掉任何一个环节，其他环节完全不用动。比如把llm从一个模型换成另一个模型，整条链的调用方式不变，这就是 Chain 抽象带来的最大价值。
提示：当年我纠结过"到底要不要用 Chain，直接手写不也能跑吗？"——如果你只是做一次性实验，手写勉强能忍；但只要你打算把 RAG 应用上线、持续迭代，Chain 带来的可维护性和可观测性回报是巨大的。
2. RAG 闭环的四个核心环节：先拆开，再组合
2.1 检索链路：Embedding、向量库与 Retriever 的组装
RAG 闭环的第一段是"检索链路"。它要做的事情很纯粹：把用户问题变成向量，然后找出最相关的文档片段。在 LangChain 里，这一段的终点是一个Retriever对象，它是一个"输入问题、输出文档列表"的 Runnable。
构建 Retriever 的标准流程分三步。第一步准备 Embedding 模型，也就是把文本变成向量的模型；第二步准备向量库，把文档写入并建立索引；第三步把向量库包装成 Retriever。
from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS from langchain_text_splitters import RecursiveCharacterTextSplitter embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 文档切分：这一步直接影响检索质量 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=80, separators=["\n\n", "\n", "。", "！", "？", " ", ""] ) # 假设 docs 是已经加载好的原始文档 split_docs = text_splitter.split_documents(docs) vector_store = FAISS.from_documents(split_docs, embeddings) retriever = vector_store.as_retriever(search_kwargs={"k": 4})这里有个特别容易忽略的点：chunk_size和chunk_overlap的选择会直接决定检索质量。chunk_size太大，单个片段里包含太多无关信息，embedding 的语义会被稀释；太小，则可能截断关键内容。我实测的一个经验值是：文档以段落为主时，500 到 800 的块大小、10% 到 15% 的重叠率是一个不错的起点。
另外要提醒一句，as_retriever(search_kwargs={"k": 4})里的k是召回数量，不是最终答案质量的全部保证。k太小可能漏掉关键文档，k太大则会把噪声带进 Prompt。实际项目中我通常会先设k=6，再根据测试结果调整。
2.2 生成链路：Prompt 模板、LLM 与输出解析器的串联
检索链路拿到文档之后，接下来是"生成链路"。这一段的核心是：把检索结果塞进 Prompt，让大模型基于这些资料回答。很多初学者以为生成链路就是"喂给大模型就算完"，但实际工程中至少有三个细节要处理。
第一个细节是 Prompt 模板的设计。模板必须明确告诉大模型：哪些是参考资料、哪些是用户问题、回答不出来时怎么办。举个典型的例子：
from langchain_core.prompts import ChatPromptTemplate prompt_template = ChatPromptTemplate.from_messages([ ("system", "你是一个严谨的问答助手。请严格依据给定的资料回答用户问题。" "如果资料中找不到答案，请明确回答'资料中未找到相关信息'，不要编造。"), ("human", "参考资料：\n{context}\n\n用户问题：{question}") ])第二个细节是输出解析器。大模型返回的是一个字符串，但你可能想要结构化输出，比如"答案 + 引用的文档来源"。此时需要StrOutputParser或者自定义的输出解析器。
第三个细节是 LLM 的接入。这里要注意 Base URL、API Key、模型名称的配置，在生产环境里建议全部走环境变量，不要硬编码在代码里。
from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2) output_parser = StrOutputParser() # 最小生成链路 gen_chain = prompt_template | llm | output_parser把检索链路和生成链路拼起来，就得到了一条最简 RAG 链：
rag_chain = ( {"context": retriever, "question": RunnablePassthrough()} | prompt_template | llm | output_parser ) result = rag_chain.invoke("什么是向量数据库？")LCEL 里的{"context": retriever, "question": RunnablePassthrough()}是一个字典映射，意思是同时构建两个输入：context来自 retriever 的检索结果，question直接透传用户原始问题。这一步是整个链式编排的精华，理解了它，后面的并行和路由就顺理成章了。
3. LCEL 组合方式：管道符、并行与条件路由
3.1 Runnable 协议：为什么每个组件都能用管道符连接
LCEL 之所以能"无脑"用|连接，是因为所有组件都遵循同一个Runnable协议。这个协议规定了四个核心方法：invoke（同步调用）、ainvoke（异步调用）、stream（流式调用）、batch（批量调用）。
拿最典型的管道符来说：
chain = prompt_template | llm | output_parser这行代码背后发生了什么？Python 的|运算符被重载了，等价于chain = prompt_template.__or__(llm).__or__(output_parser)。LangChain 会把这些 Runnable 按顺序组成一个RunnableSequence。当你对这个组合链调用.invoke()时，框架自动把上一个组件的输出传给下一个组件。
这种设计的妙处在于：你不需要关心每个组件内部怎么实现，只要它实现了 Runnable 协议，就能参与组合。自己写一个自定义组件也只需要继承Runnable并实现invoke方法。
3.2 三种组合模式：顺序、并行与路由
顺序组合是最常见的，上面已经演示过。但实际项目中，经常遇到"需要同时做几件事"的情况，比如既要检索向量库，又要查结构化数据库，还要做关键词检索。如果串行做，每一步都要等前一步完成，太慢。这时候要用并行组合。
LCEL 提供RunnableParallel来支持并行执行。最直观的写法是用字典：
from langchain_core.runnables import RunnableParallel def query_sql_db(question: str): # 模拟查询结构化数据库 return f"SQL查询结果：{question} 对应的库存信息" parallel_chain = RunnableParallel( vector_context=retriever, sql_context=query_sql_db, question=RunnablePassthrough() )调用这个并行链时，vector_context和sql_context会同时执行，最后合并成一个字典输出。合并后的字典可以直接作为后续 Prompt 的输入。这种模式在"多路召回"场景下特别有用。
第三种组合模式是路由（RunnableBranch）。路由解决的是"不同问题走不同链路"的问题。举个例子：用户问的是"价格政策"，你应该走条款检索链路；用户问的是"售后流程"，你应该走另一个 FAQ 检索链路。用一个意图分类器决定走哪条路。
from langchain_core.runnables import RunnableBranch route_chain = RunnableBranch( # (condition, runnable) 按顺序匹配 (lambda x: "价格" in x["question"], price_chain), (lambda x: "售后" in x["question"], after_sale_chain), default_chain # 兜底 )路由模式的关键是条件函数的判断速度要快，因为它每个请求都会执行。条件里如果放一个 LLM 调用，响应延迟会显著上升，所以通常会先用规则或一个轻量分类模型做判断。
提示：我见过不少项目一上来就用 LLM 做路由判断，结果每个请求多出 2 到 3 秒的额外延迟。能用关键词或正则解决的，尽量别用模型。只有规则实在覆盖不了、而且对延迟不敏感的场景，才值得上 LLM 路由。
4. 端到端 RAG 闭环：带引用溯源与多轮记忆的完整实现
4.1 完整代码实现与逐段拆解
讲完组合模式，来一个能直接用的完整闭环。这个例子的场景是"某企业内部文档问答系统"，需求有三条：回答用户问题、给出引用来源、支持多轮对话。我把完整代码拆成三段来讲。
第一段：准备检索组件。
from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_community.vectorstores import FAISS from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough, RunnableParallel from langchain_core.documents import Document from langchain.memory import ConversationBufferMemory from langchain_core.chat_history import BaseChatMessageHistory # 假设已有 document_list: List[Document] text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=80 ) split_docs = text_splitter.split_documents(document_list) embeddings = OpenAIEmbeddings(model="text-embedding-3-small") vector_store = FAISS.from_documents(split_docs, embeddings) retriever = vector_store.as_retriever(search_kwargs={"k": 4})第二段：设计 Prompt 和自定义输出解析逻辑。这里不仅要普通文本答案，还要把引用的文档源信息单独提取出来。做法是让 LLM 按 JSON 格式输出，再交给解析函数处理。
from langchain_core.pydantic_v1 import BaseModel, Field from langchain_core.output_parsers import JsonOutputParser class RAGAnswer(BaseModel): answer: str = Field(description="对用户问题的回答") sources: list = Field(description="回答中引用的文档来源ID列表") prompt_with_sources = ChatPromptTemplate.from_messages([ ("system", "你是一个严谨的问答助手。请根据给定的资料回答问题。" "你的输出必须是JSON格式，包含字段：answer（回答内容）和sources（引用来源ID列表，来源ID在资料中用[ID:x]标记）。" "如果资料中找不到答案，answer字段返回'资料中未找到相关信息'，sources返回空列表。"), ("human", "参考资料：\n{context}\n\n问题：{question}") ]) json_parser = JsonOutputParser(pydantic_object=RAGAnswer)这里有个细节：为了让模型能输出来源 ID，我在写入向量库之前就给每篇文档加上了[ID:x]的前缀标记。具体做法是切分完文档后，对每个 chunk 的 page_content 做一次预处理。
第三段：组装完整链路。
retrieval_dict = RunnableParallel( context=retriever, question=RunnablePassthrough() ) rag_chain = retrieval_dict | prompt_with_sources | llm | json_parser # 调用示例 result = rag_chain.invoke("如何申请年假？") print(result["answer"]) print(result["sources"])可以看出来，整条链从"输入问题"到"输出 JSON 结构"只用了四段组件。如果后续需要加 Rerank，只需要在retriever后面再插一个 reranker Runnable，其他组件完全不用动。
4.2 流式输出：让用户体验不再卡顿
上面的链是完整调用的，但大模型生成本来就需要几秒，如果等全部生成完再返回，用户体验会非常差。LangChain 的 Runnable 协议提供了stream()方法，可以边生成边吐字。
但问题来了：json_parser 是一个"需要完整 JSON 才能解析"的组件，没法直接对部分文本做解析。所以流式输出通常要拆成两条线：一条是"结构化答案"走通用解析；另一条是"纯文本流式输出"直接吐字。
# 流式输出链：分开处理 stream_chain = retrieval_dict | prompt_with_sources | llm | StrOutputParser() for chunk in stream_chain.stream({"question": "如何申请年假？"}): print(chunk, end="", flush=True)这里要记住一个原则：流式输出和结构化解析是互斥的，至少不能放在同一条链里同时做。工程上的做法是：界面上先用流式链展示答案文本，后台再用结构化链拿到带来源的完整答案。两条链共享相同的检索组件和 Prompt，只是最后挂的解析器不同。
4.3 多轮对话：把历史记录接进链里
多轮对话是 RAG 应用的另一个刚需。用户可能会追问上一轮的细节，比如先问"年假有多少天"，再问"那可以分开休吗？"——第二问如果不结合第一轮的上下文，检索就会很迷茫。
LangChain 处理多轮对话的经典方案是用ConversationBufferMemory或者新版的消息历史对象。核心思路是：把历史消息和当前问题一起传给 Prompt，甚至把"历史上下文归纳后的重写问题"用来做检索。
from langchain.memory import ConversationBufferMemory from langchain_core.runnables import RunnableLambda memory = ConversationBufferMemory( return_messages=True, memory_key="chat_history" ) def process_question_with_history(question: str): # 读取历史记忆，拼进 Prompt；同时让 LLM 把当前问题改写为独立问题 history = memory.load_memory_variables({})["chat_history"] history_text = "\n".join([m.content for m in history]) # 用改写后的独立问题去检索，效果比直接拿原始问题检索好很多 rewritten = llm.invoke( f"根据对话历史改写当前问题，使其成为无需上下文即可理解的独立问题。\n" f"历史：{history_text}\n当前问题：{question}\n改写后的问题：" ).content memory.save_context({"input": question}, {"output": rewritten}) return rewritten final_chain = ( RunnableLambda(process_question_with_history) | retrieval_dict | prompt_with_sources | llm | json_parser )这里有个实战心得很重要：多轮对话的 RAG 里，检索阶段用的是"改写后的独立问题"，而不是原始问题。原因是向量检索本身没有上下文理解能力，你丢给它一句"那可以分开休吗"，它很难知道"那"指代什么。先把问题通过 LLM 改写为完整语义，再去做 embedding，检索质量会明显提升。
5. 实战中的坑与性能优化：我踩过的那些问题
5.1 检索质量瓶颈排查：从召回数量到切分策略
链跑通之后，最常遇到的现象是"回答看似合理，但总感觉缺了关键信息"。这种问题十有八九出在检索环节，而不是大模型环节。我排查这类问题有一套固定流程。
第一步，把检索结果直接打印出来看。用retriever.invoke("某个具体问题")单独观察返回的文档，看看返回的片段是不是真的包含关键信息。如果返回的文档明显偏题，先调整k值和切分策略。第二步，确认切分是否把关键语义切断了。比如一个完整的政策条款被从中间切开，两个 chunk 各自都不完整，检索自然就出问题。第三步，检查 Embedding 模型是否适合你的领域。通用模型在专业术语密集的场景下效果偏差，必要时换成领域微调过的模型。
# 检索环节单测：先于整条链调试 test_docs = retriever.invoke("如何申请年假？") for i, doc in enumerate(test_docs): print(f"[{i}] {doc.page_content[:100]}")这个调试习惯帮我省了大量时间。很多开发者遇到回答质量问题，第一反应是换 Prompt，其实问题根源根本不在生成链路，在检索链路就已经丢了信息。
5.2 链过长时的调试技巧：逐段观察与中间变量注入
链越长，越难定位问题出在哪个环节。LCEL 虽然让代码变得简洁，但也让中间状态变得不可见——你没法直接看到retriever输出了什么、prompt格式化成了什么。
我的调试套路是"分段独立 invoke"。把长链拆成几段，每段单独调用：
# 第一步：单独看检索 docs = retriever.invoke("如何申请年假？") # 第二步：单独看 Prompt 格式化结果 formatted_prompt = prompt_with_sources.invoke({ "context": "\n\n".join([d.page_content for d in docs]), "question": "如何申请年假？" }) # 第三步：单独看 LLM 输出 raw_output = llm.invoke(formatted_prompt) # 第四步：最后再看解析器 parsed = json_parser.invoke(raw_output.content)这种写法虽然繁琐，但胜在每一段的可观测性都极高。如果哪一段输出不符合预期，问题就锁定在那一段。另外，LangChain 生态里的 LangSmith 工具可以在线跟踪每一步的输入输出，如果你在用官方生态，建议从一开始就挂上监控，省去自己 print 的麻烦。
5.3 性能优化：缓存、并发与重排序的取舍
最后一个实战话题是性能。RAG 链的延迟主要由四个部分构成：Embedding 耗时、向量库检索耗时、LLM 生成耗时、可能的 Rerank 耗时。其中 LLM 生成是大头，检索其次。
优化手段里性价比最高的是缓存。对高频相似问题，直接用缓存命中，跳过整个链路。LangChain 提供了RunnableLambda包装缓存逻辑的方式，或者直接在外部加一层 Redis 缓存。缓存 key 可以用"改写后的问题 + 检索结果摘要"的组合，这样既能命中重复问答，又能避免同样的文档被反复 embedding。
import hashlib from langchain_core.runnables import RunnableLambda def cache_wrapper(inputs: dict): question = inputs["question"] cache_key = hashlib.md5(question.encode()).hexdigest() cached = cache_get(cache_key) # 伪代码 if cached: return cached result = full_chain.invoke(inputs) cache_set(cache_key, result) return result cached_chain = RunnableLambda(cache_wrapper)另一个优化点是检索并发。如果你用了多路召回（向量检索 + 关键词检索并行），RunnableParallel天然支持并发，比串行快很多。但要注意：并发过高可能把向量库的连接数打满，需要做限流。
至于重排序（Rerank），我的建议是"按需引入"。如果检索结果 Top 4 的准确率已经够用，不需要额外引入 Rerank 环节，毕竟每次 Rerank 也是一次模型调用，会带来延迟。只有当你发现k调大到 10 甚至 20 以后，答案质量仍不稳定时，才值得加一个 Reranker 把候选集重新排序，再取 Top 4 进 Prompt。
6. Chain 的边界与架构选择：一个过来人的体会
文章写到最后，我想聊聊 Chain 的边界问题。很多初学者容易走到两个极端：要么把所有逻辑都塞进一条超级长的链里，要么完全不用链、全部手工编排。我个人的体会是，Chain 适合"流程固定、组件边界清晰"的场景，比如 RAG 的标准四段式；一旦你的分支逻辑特别复杂、状态特别多，Link 式的线性编排会变得难以维护，这时候反而应该拆成多个小型链，再用业务代码做组合调度。
举个实际例子：我做过的"某企业知识库智能客服"项目，最终并没有把意图识别、多轮改写、检索、生成全塞进一条链。而是拆成了三条独立链——意图路由链、问题改写链、检索生成链，由外层一个轻量调度函数负责按顺序调用。这样做的原因是三条链的变更频率不同：检索生成链几乎每周都要调 Prompt 或切模型，但意图路由链很少动。拆开之后，改其中一条链完全不影响另外两条，CI 测试也好写。
从维护性的角度说，链式编排最大的价值不是让代码变短，而是让"数据流向"变得明确。你随时能回答"用户的输入经过哪些环节、每个环节处理了什么"，这就够了。至于每一步的细节，那是组件内部的事，跟你编排全局的代码无关。
我经历过从最原始的手工拼接，到用类式 Chain，再到 LCEL 的重构过程。如果你正准备把一个 RAG 概念验证项目推向生产环境，我建议直接以 LCEL 为骨架来组织代码。遇到问题的时候，先拆检索链路做单测，再排查 Prompt 格式化，最后才怀疑大模型本身——按这个顺序排查，大部分问题都能在半小时内定位。
最后留一个可以继续往下走的扩展方向：在现有链路里插入"文档更新检测"环节，让知识库能感知源文档的变更，自动重切分、重写入。这一步做完，端到端 RAG 闭环才算真正具备了生产级的自我维护能力。
