> ## Documentation Index
> Fetch the complete documentation index at: https://www.yuan111.asia/doc/llms.txt
> Use this file to discover all available pages before exploring further.

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

# 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`（透传）

```python theme={null}
from langchain_core.runnables import RunnablePassthrough

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

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

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

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

***

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

```python theme={null}
# ---------- 零件 ----------
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 篇） |

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

```python theme={null}
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`）：

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

```powershell theme={null}
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 篇要溯源），改这个函数就行，链的其他部分不动。

***

## 推荐资料（延伸阅读）

* [LangChain 官方文档 · 检索](https://docs.langchain.com/oss/python/langchain/retrieval) —— 检索概念与检索器接口
* [LangChain 官方文档 · 知识库](https://docs.langchain.com/oss/python/langchain/knowledge-base) —— 知识库问答的官方教程
* [LangChain 官方文档 · 组件架构](https://docs.langchain.com/oss/python/langchain/component-architecture) —— Runnable 协议 / LCEL 的官方说明

***

## 9. 未完待续

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

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

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

→ [10 · 重排序与对话式 RAG](/doc/doc/enterprise-rag-course/10-重排序与对话式RAG)