LangChain RAG Chain Types 详解

本文介绍 ConversationalRetrievalChain 的三种文档处理策略：stuff、map_reduce、refine。

1. 为什么需要不同的 Chain Type？

将文档塞给 LLM 时，有不同的策略：

策略	适用场景	优缺点
Stuff	文档较少	简单快速，但超过上下文限制会失败
Map Reduce	文档多、文档大	可处理大量文档，但可能丢失跨文档关联
Refine	需要连贯上下文	答案更连贯，但速度较慢

---

2. 三种 Chain Type 详解

2.1 Stuff（填充式）

原理：将所有检索到的文档拼接到一个 prompt 中，一次性发给 LLM。

[文档1] + [文档2] + [文档3] + [问题] → LLM → 答案

代码：

qa = ConversationalRetrievalChain.from_llm(
    llm=model,
    retriever=retriever,
    memory=memory,
    chain_type="stuff"  # 默认方式
)

优点	缺点
简单快速	受上下文窗口限制
保留完整上下文	文档过多时会失败

2.2 Map Reduce（映射归纳式）

原理：

1. Map：每个文档单独调用 LLM 生成答案
2. Reduce：所有单独答案合并后再调用 LLM 总结

[文档1] → LLM → 答案1
[文档2] → LLM → 答案2    →  LLM(合并) → 最终答案
[文档3] → LLM → 答案3

代码：

qa = ConversationalRetrievalChain.from_llm(
    llm=model,
    retriever=retriever,
    memory=memory,
    chain_type="map_reduce"
)

优点	缺点
可处理大量文档	可能丢失文档间关联
每个文档并行处理	多次 API 调用，成本较高

2.3 Refine（精炼式）

原理：逐个文档迭代精炼答案，后一个文档基于前一个答案继续优化。

[文档1] → LLM → 答案1
              ↓
[文档2] + 答案1 → LLM → 答案2
                            ↓
[文档3] + 答案2 → LLM → 最终答案

代码：

qa = ConversationalRetrievalChain.from_llm(
    llm=model,
    retriever=retriever,
    memory=memory,
    chain_type="refine"
)

优点	缺点
答案连贯一致	速度最慢
保留上下文顺序	错误可能累积

---

3. 对比总结

特性	stuff	map_reduce	refine
处理方式	一次性塞入	分开处理再合并	迭代精炼
文档数量	少	多	多
上下文保留	完整	部分丢失	较好
速度	快	中等	慢
成本	低	较高	最高

3.1 如何选择？

场景	推荐
检索文档 < 3 个	stuff
检索文档 3-10 个	map_reduce 或 refine
检索文档 > 10 个	map_reduce
需要连贯答案/长文	refine

---

4. 代码示例

4.1 测试三种 Chain Type

chain_types = ["stuff", "map_reduce", "refine"]

for chain_type in chain_types:
    memory = ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True,
        output_key="answer"
    )

    qa = ConversationalRetrievalChain.from_llm(
        llm=model,
        retriever=retriever,
        memory=memory,
        chain_type=chain_type,
        verbose=False
    )

    result = qa.invoke({"question": "卢浮宫这个名字怎么来的？"})
    print(f"chain_type={chain_type}: {result['answer']}")

结果：

4.2 带上下文的连续对话

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="answer"
)

qa = ConversationalRetrievalChain.from_llm(
    llm=model,
    retriever=retriever,
    memory=memory,
    chain_type="stuff",
    verbose=False
)

# 第一轮
result1 = qa.invoke({"question": "卢浮宫是什么时候建的？"})
print(f"Q1: 卢浮宫是什么时候建的？")
print(f"A1: {result1['answer']}")

# 第二轮 - "它"指代上文中的"卢浮宫"
result2 = qa.invoke({"question": "它在哪里？"})
print(f"Q2: 它在哪里？")
print(f"A2: {result2['answer']}")

结果：

4.3 查看对话历史

print("=== 对话历史 ===")
for msg in memory.chat_memory.messages:
    print(f"{msg.type}: {msg.content}")

结果：

---

5. 进阶：自定义 Chain

5.1 自定义 combine_docs_chain

from langchain.chains import LLMChain, StuffDocumentsChain

# 自定义 stuff chain
llm_chain = LLMChain(llm=model, prompt=custom_prompt)
combine_docs_chain = StuffDocumentsChain(llm_chain=llm_chain)

qa = ConversationalRetrievalChain(
    llm=model,
    retriever=retriever,
    memory=memory,
    combine_docs_chain=combine_docs_chain
)

5.2 自定义 map 阶段的 prompt

from langchain.chains import AnalyzeDocumentChain

qa_chain = AnalyzeDocumentChain(
    combine_docs_chain=custom_combine,
    vectorstore=db
)

---

6. 常见问题

Q1: stuff 超过上下文限制怎么办？

使用 map_reduce 或 refine，它们会将文档分批处理。

Q2: refine 会累积错误吗？

有可能。如果某个文档答案错误，可能影响后续迭代结果。重要场景建议用 map_reduce 验证。

Q3: 连续对话中 chain_type 怎么选？

连续对话建议用 stuff，因为每次只检索少量相关文档，内存占用可控。

Q4: 如何查看 chain 执行的中间结果？

qa = ConversationalRetrievalChain.from_llm(
    ...
    verbose=True  # 开启后会打印详细执行过程
)

Q5: 文档数量少但答案不完整，用哪个？

用 refine，它会基于前一个答案和下一个文档迭代优化。