09 · RAG 组装：用 LCEL 管道把零件串成一条链

本片目标：把前 8 篇的所有零件（模板、Retriever、模型、解析器）用 LCEL（|）串成一条完整的 RAG 链——输入问题，输出基于资料的答案。这是全课程的里程碑。 新增规定性：9（组合协议：Runnable 接口 + | 管道 + RunnablePassthrough） 数据字典：Runnable / RunnablePassthrough / RunnableSequence / RunnableParallel。 进程线程模型：链内并行分支（context 检索与 question 传递并行）在单线程内依次执行；批量调用走线程池（第 12 篇）。 网络模型：一次问答 = 检索(本地) + 1 次模型 POST。

1. 上集回顾

第 08 篇我们有了 Retriever（能召回最相关的块）。但还差最后一步：

检索到 3 块相关文本，然后呢？怎么把它变成答案？

完整的 RAG 是四步流水线：

① 检索：问题 → Retriever → 相关块
② 拼装：相关块 → 格式化成 context 文本
③ 生成：context + 问题 → prompt → 模型
④ 解析：AIMessage → 纯文本

旧教程的解法：RetrievalQA / create_retrieval_chain——但这两个已迁入 langchain-classic（弃用包），本系列不用。 1.x 的正解：LCEL 管道。我们第 04 篇已经见过 prompt | model | parser，第 08 篇知道 Retriever 是 Runnable。那么 retriever 也能接进管道——这就组成了 RAG 链。

2. 数据字典：Runnable 协议（规定性 9 的核心）

Runnable 是什么：一个实现了统一接口的”可执行单元”。所有能接进 | 的东西都是 Runnable。

方法	签名	说明
`invoke`	`invoke(input, config=None)`	同步执行
`batch`	`batch(inputs: list, config=None)`	批量执行
`stream`	`stream(input, config=None)`	流式执行
`ainvoke/abatch/astream`	异步版	第 12 篇

谁实现了 Runnable（本系列已见过的）：

对象	输入	输出
`ChatPromptTemplate`	`dict`	`PromptValue`
`ChatAnthropic`	`str/list[BaseMessage]/dict`	`AIMessage`
`StrOutputParser`	`AIMessage/str`	`TextAccessor`
Retriever	`str`	`list[Document]`
`RunnablePassthrough`	任意	原样返回
`prompt	model	parser`（组合链）	首段输入	末段输出

2.1 `RunnablePassthrough`（透传）

from langchain_core.runnables import RunnablePassthrough

RunnablePassthrough()        # 输入啥返回啥
RunnablePassthrough.assign() # 输入基础上加新键

用途：RAG 链里 question 走”旁路”原样传给模板，同时 context 走”检索→格式化”路径。两条路汇合。

2.2 `RunnableParallel`（并行 dict，`{...}` 语法糖）

# 这行：
{"context": retriever | format_docs, "question": RunnablePassthrough()}
# 等价于 RunnableParallel，两条管道并行执行，输出 dict 汇合

3. RAG 链完整代码（09 篇主角）

# ---------- 零件 ----------
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

def format_docs(docs):
    """list[Document] → 拼成一段带编号的文本"""
    return "\n\n".join(f"[资料{i+1}] {d.page_content}" for i, d in enumerate(docs))

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "你是公司新人培训手册问答助手。只能根据下面提供的【资料】回答员工的问题。"
     "资料里没有的信息，明确回答'手册里没有相关信息'，不要编造。\n\n【资料】\n{context}"),
    ("human", "{question}"),
])

# ---------- 组装 ----------
rag_chain = (
    {"context": retriever | format_docs,   # 左路：检索 + 拼资料
     "question": RunnablePassthrough()}    # 右路：问题原样传
    | prompt
    | model
    | StrOutputParser()
)

# ---------- 使用 ----------
answer = rag_chain.invoke("转正需要什么条件？")

类型流（每个环节的数据形态）：

"转正需要什么条件？"                          ← str
  → RunnableParallel:
      {"context": retriever|format_docs, "question": passthrough}
      → {"context": "[资料1] 转正要求：...\n\n[资料2] ...", "question": "转正需要什么条件？"}   ← dict
  → prompt.invoke(dict) → PromptValue → to_messages() → 消息列表
  → model.invoke → AIMessage
  → StrOutputParser → str（答案）

4. 关键方法

方法	签名	说明
`chain.invoke(question)`	`(str) -> str`	问一个问题得答案
`chain.batch([q1, q2])`	`-> list[str]`	批量问（第 12 篇）
`chain.stream(question)`	`-> Iterator[str]`	流式回答（打字机效果）
`chain.ainvoke(question)`	`-> str`	异步（第 11~12 篇）

链的自我检查（调试神器）：

rag_chain.get_graph().print_ascii()   # 打印管道结构图
rag_chain.input_schema / rag_chain.output_schema  # 输入输出 schema

5. 进程线程模型

invoke("转正需要什么条件？")
  ├─ RunnableParallel（两条子管道）
  │    ├─ 左: retriever.invoke("转正需要什么条件？")   → 3 个 Document（本地检索，~几ms）
  │    │      format_docs(docs)                  → 文本（纯内存）
  │    └─ 右: RunnablePassthrough()              → 原样返回（纯内存）
  ├─ prompt.invoke(dict)                         → 消息列表（纯内存）
  ├─ model.invoke(messages)                      → 网络阻塞 1~3s ← 全链唯一慢点
  └─ StrOutputParser.invoke(ai_msg)              → str（纯内存）

关键洞察：

只有 model.invoke 一次网络往返——其余环节全部本地毫秒级。
RunnableParallel 的同步 invoke 内部是顺序执行两条分支（线程池并发是 batch/异步的事，第 12 篇）。
所以 RAG 问答的延迟 ≈ 一次模型调用延迟（1~3s）+ 检索几毫秒。

6. 网络模型

rag_chain.invoke("转正需要什么条件？")
  ├─ 检索：本地 Chroma（0 网络请求）
  ├─ 模型：POST https://open.bigmodel.cn/api/anthropic/v1/messages
  │     Body: {"model":"glm-4.7","max_tokens":512,
  │            "messages":[{system 含资料}, {user: "转正需要什么条件？"}]}
  │     注意：context（资料）被拼进了 system 消息
  └─ 解析：本地

RAG 的本质网络模型：把”外部知识”通过 system 消息注入——模型本身不需要”记住”你的文档，每次问答把相关资料塞进上下文即可。

7. 验证：跑起来

配套代码 code/09_rag.py（沿用第 08 篇建好的 data/chroma_db）：

加载已建向量库 → retriever；
组装 RAG 链；
问 4 个问题（含一个手册里没有的，测试”不说谎”）；
解剖链中间产物：模型返回的 AIMessage 完整结构。

cd enterprise-rag-course\code
python 09_rag.py

预期输出（节选）：

问：转正需要什么条件？
答：根据手册，转正需试用期满、累计至少 60 篇日总结、答辩平均分 75 分以上...

问：公司食堂中午有什么菜？
答：手册里没有相关信息。   ← 资料里没有，明确说没有，不编造

===== 解剖 =====
链中间产物 AIMessage.response_metadata.stop_reason = 'end_turn'  ← 普通问答，直接回答完毕

8. 边界

k 值影响质量：k=3 召回 3 块。k 太小漏信息，k 太大塞进多余内容干扰模型。第 15 篇讲调优。
“资料里没有就说没有”靠提示词约束——不是硬保证。模型仍可能幻觉，第 10 篇重排序 + 第 15 篇引用溯源缓解。
不用 create_retrieval_chain——它在 langchain-classic。LCEL 管道完全等价且更透明（你看着每个环节）。
format_docs 是可换的——想保留来源信息（第 15 篇要溯源），改这个函数就行，链的其他部分不动。

9. 未完待续

RAG 能回答了。但两个质量问题：

检索不准：向量检索召回 3 块，可能第 3 块是”噪音”。怎么精排？
不能多轮追问：“转正需要什么条件？” → “那答辩要几位评委？” —— 第二个问题没有”转正”字样，检索会失准。怎么结合对话历史？

这就是第 10 篇：重排序与对话式 RAG。 → 10 · 重排序与对话式 RAG

RAG 教程

小林学 LangChain

09 · RAG 组装：用 LCEL 管道把零件串成一条链

09 · RAG 组装：用 LCEL 管道把零件串成一条链

1. 上集回顾

2. 数据字典：Runnable 协议（规定性 9 的核心）

2.1 `RunnablePassthrough`（透传）

2.2 `RunnableParallel`（并行 dict，`{...}` 语法糖）

3. RAG 链完整代码（09 篇主角）

4. 关键方法

5. 进程线程模型

6. 网络模型

7. 验证：跑起来

8. 边界

推荐资料（延伸阅读）

9. 未完待续

​09 · RAG 组装：用 LCEL 管道把零件串成一条链

​1. 上集回顾

​2. 数据字典：Runnable 协议（规定性 9 的核心）

​2.1 RunnablePassthrough（透传）

​2.2 RunnableParallel（并行 dict，{...} 语法糖）

​3. RAG 链完整代码（09 篇主角）

​4. 关键方法

​5. 进程线程模型

​6. 网络模型

​7. 验证：跑起来

​8. 边界

​推荐资料（延伸阅读）

​9. 未完待续

09 · RAG 组装：用 LCEL 管道把零件串成一条链

1. 上集回顾

2. 数据字典：Runnable 协议（规定性 9 的核心）

2.1 `RunnablePassthrough`（透传）

2.2 `RunnableParallel`（并行 dict，`{...}` 语法糖）

3. RAG 链完整代码（09 篇主角）

4. 关键方法

5. 进程线程模型

6. 网络模型

7. 验证：跑起来

8. 边界

推荐资料（延伸阅读）

9. 未完待续