LangChain：文档链使用方式和示例

LangChain中的文档链是处理文档内容的核心组件，专门用于文档总结、问答、信息提取等任务。在LangChain 1.0+版本中，官方推荐使用LCEL（LangChain Expression Language）风格的链，它们更易于扩展和集成。

文档链概念

文档链是LangChain 1.0+中实现RAG（检索增强生成）的核心组件，通过将文档检索与大语言模型无缝集成，使AI能基于特定文档内容生成准确回答，有效解决模型”幻觉”问题。

文档链（Document Chain）本质上是将多个处理步骤串联的可执行流程，在LangChain 1.0+中，高效的文档链方案不再是单一的“链”，而是由更现代、更灵活的LCEL（LangChain Expression Language) 和高级封装函数来实现声明式组合。它涵盖了从简单拼接到高级检索增强生成（RAG，即Retrieval-Augmented Generation）的完整流程。

LangChain 1.0+ 版本取消了旧版的 Stuff、Refine、MapReduce 和重排链，只保持了兼容，但不再更新维护，建议使用新的LCEL风格链。

文档链使用方法

基础文档链构建流程

create_stuff_documents_chain：基础文档拼接，官方推荐。

这是构建RAG系统的基石链，专注于处理和管理文档上下文。它会将所有文档格式化为提示，然后传递给LLM。

• 定义与功能：它接收一组文档，将它们全部“拼”进一个提示（Prompt）中，然后传递给LLM。它本质上是旧版StuffDocumentsChain的现代化替代品。
• 注意事项：所有文档都必须能放入LLM的上下文窗口，需确保总长度不超过LLM的上下文窗口。
• 开发建议：官方强烈推荐使用此链，而非@langchain/classic。它易于扩展，且原生支持流式处理和批处理。

示例1：构建文档链

from langchain.chains import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document

# 1. 初始化模型
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

# 2. 准备文档
documents = [
    Document(page_content="太阳从东边升起。"),
    Document(page_content="太阳从西边落下。")
]

# 3. 创建提示模板 (注意这里的变量名 "context" 必须与调用时的 key 一致)
# prompt = ChatPromptTemplate.from_template("请根据以下内容回答问题：{context}")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业助手，请根据提供的文档内容回答用户问题。"),
    ("human", "文档内容：\n{context}\n\n问题：{input}")
])

# 4. 创建拼合文档链
chain = create_stuff_documents_chain(llm, prompt)

# 5. 调用链 (以 'context' 为键传入文档列表)
# result = chain.invoke({"context": documents})
response = document_chain.invoke({
    "input": "早上的太阳是从哪边升起？",
    "context": documents
})

print(result)

示例2：多文档问答系统

# 构建支持多轮对话的文档问答系统
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.prompts import MessagesPlaceholder

# 带历史记录的提示模板
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的文档助手，基于以下上下文回答问题：\n{context}"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{input}")
])

# 创建带记忆的文档链
document_chain = create_stuff_documents_chain(llm, chat_prompt)

# 使用示例
chat_history = [
    HumanMessage(content="LangChain是什么？"),
    AIMessage(content="LangChain是一个用于开发由语言模型驱动的应用程序的框架")
]

response = document_chain.invoke({
    "input": "它支持哪些文档处理方式？",
    "context": docs,
    "chat_history": chat_history
})

结合向量检索的完整RAG流程

create_retrieval_chain：是RAG流程的完整封装链，通常与create_stuff_documents_chain结合使用实现RAG。

• 定义与功能：它接收一个用户查询，先交给retriever检索相关文档；然后，它无缝地将检索到的文档和原始查询一起，交给负责“合并与生成”的链（通常是create_stuff_documents_chain的实例）处理，最后生成答案。
• 适用场景：当知识库超过单个上下文窗口，无法一次性提交给LLM时，就需要使用这种方法。

示例1：

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 1. 准备模型和提示模板
llm = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("根据以下上下文：{context} 回答：{input}")

# 2. 创建"拼合文档链"，它将处理检索到的文档并生成回答
combine_docs_chain = create_stuff_documents_chain(llm, prompt)

# 3. 创建向量存储和检索器
vector_store = FAISS.from_texts(
    ["LangChain 是一个用于开发 LLM 应用的框架。", "它极大地简化了 RAG 应用的构建。"],
    embedding=OpenAIEmbeddings()
)
retriever = vector_store.as_retriever()

# 4. 创建最终的 RAG 链
rag_chain = create_retrieval_chain(retriever, combine_docs_chain)

# 5. 执行查询
response = rag_chain.invoke({"input": "LangChain 主要用来做什么？"})
print(response["answer"])

示例2：

from langchain.chains import create_retrieval_chain
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1. 文档加载与分割
loader = TextLoader("knowledge_base.txt")
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
docs = text_splitter.split_documents(documents)

# 2. 向量化存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=embeddings,
    persist_directory="./vector_db"
)

# 3. 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}  # 返回3个最相关文档
)

# 4. 创建检索链
retrieval_chain = create_retrieval_chain(
    retriever=retriever,
    combine_docs_chain=create_stuff_documents_chain(llm, prompt)
)

# 5. 执行检索与回答
response = retrieval_chain.invoke({
    "input": "如何使用LangChain构建多轮对话系统？"
})
print("回答:", response["answer"])
print("引用文档:", response["context"])

支持多轮对话的文档链

注意：对于长对话，建议使用ConversationSummaryMemory替代ConversationBufferMemory，避免token超限。

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

# 创建对话记忆
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    input_key="question",
    output_key="answer"
)

# 构建支持多轮对话的RAG链
conversational_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=retriever,
    memory=memory,
    return_source_documents=True
)

# 第一轮对话
response1 = conversational_chain.invoke({
    "question": "LangChain是什么？"
})
print("AI:", response1["answer"])

# 第二轮对话（自动包含历史）
response2 = conversational_chain.invoke({
    "question": "它有哪些核心组件？"
})
print("AI:", response2["answer"])

文档链的优化策略

文档预处理优化

• 智能分块：根据文档结构（如章节、段落）进行语义分块，而非简单按字符数分割
• 元数据添加：为文档块添加来源、主题等元数据，提高检索精度
• 去噪处理：移除无关内容（如页眉页脚、广告等）

# 语义感知分块示例
from langchain_community.document_loaders import PDFPlumberLoader
from langchain_text_splitters import MarkdownHeaderTextSplitter

loader = PDFPlumberLoader("technical_manual.pdf")
documents = loader.load()

# 按Markdown标题结构分块
headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3")
]
splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
docs = splitter.split_documents(documents)

检索效果优化技巧

• 混合检索：结合关键词检索（如BM25）与向量检索，提高召回率
• 查询重写：使用LLM将用户问题改写为更利于检索的形式
• HyDE技术：先让模型生成假设性答案，再用该答案的向量进行检索

# 查询重写示例
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

rewrite_prompt = PromptTemplate.from_template(
    "请将以下用户问题改写为更清晰、更利于检索的形式：\n{query}"
)
query_rewrite_chain = LLMChain(llm=llm, prompt=rewrite_prompt)

original_query = "LangChain怎么用？"
rewritten_query = query_rewrite_chain.run(original_query)
print("改写后查询:", rewritten_query)

生成优化策略

• 上下文压缩：使用LLM提炼文档核心内容，减少冗余
• 结构化输出：定义明确的输出格式，提高答案一致性
• 自验证机制：让模型自我检查答案是否基于文档内容

# 结构化输出示例
from langchain.output_parsers import PydanticOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

class DocumentAnswer(BaseModel):
    answer: str = Field(description="对用户问题的回答")
    confidence: float = Field(description="回答的置信度，0-1之间")
    source_docs: list[str] = Field(description="引用的文档ID列表")

parser = PydanticOutputParser(pydantic_object=DocumentAnswer)

提升检索质量

create_reranker：提升检索质量

• 定义与功能：它是一个重排序器。在检索到大量文档后，Reranker会深度学习模型对每个文档和问题相关性进行精确打分，将最相关的文档排在最前面。
• 适用场景：在RAG流程中，向量搜索可能存在信息丢失的问题。先用向量搜索快速召回候选文档（如Top-20），再用Reranker对这些候选进行精排，以提高送入LLM的上下文质量。在LangChain 1.0中，create_reranker就是用来实现这一功能的。

LCEL实现Map-Reduce

高级模式：LCEL实现Map-Reduce（替代旧版MapReduceDocumentsChain）

• 定义与功能：用于处理单个大文档，而非传统“多文档”场景。它通过将一个大文档切分，先并行处理每个小段（Map），再将所有结果合并（Reduce），生成最终摘要。
• 适用场景：文本摘要任务中，源文档非常长。
• 实现方式：官方认为，通过LCEL可以完全重现旧版MapReduceDocumentsChain的功能，且支持批处理、异步等特性。

快速选型指南

• 直接使用 create_stuff_documents_chain
  • 文档数量：少（几个到几十个）
  • 文档长度：短
  • 场景：总长度在上下文窗口内
• 使用 create_retrieval_chain
  • 文档数量：多
  • 文档长度：不定
  • 场景：知识库很大，或需要问答/检索增强生成（RAG）
• 使用 Map-Reduce 模式（LCEL）
  • 文档数量：1个（但内容很长）
  • 文档长度：长
  • 场景：对一个超长的单文档进行摘要或分析
• 增加 create_reranker 增强
  • 文档数量：多
  • 文档长度：任意
  • 场景：对检索结果的准确性有极高要求，希望提升RAG质量

总的来说，LangChain 1.0+ 的方案更偏重于组合（LCEL） 而非继承（旧链），用更少的代码，封装更强大的功能。

实用建议与最佳实践

1. 文档质量优先：确保输入文档准确、完整、格式规范，垃圾输入导致垃圾输出
2. 分块策略：中文文档建议分块大小200-500字，重叠50-100字，避免语义断裂
3. 嵌入模型选择：中文场景推荐使用BAAI/bge-small-zh-v1.5等中文优化模型
4. 检索数量：通常返回3-5个最相关文档，过多会降低答案质量
5. 流式处理：对大量文档使用stream()方法而非invoke()，避免内存溢出
6. 缓存机制：对重复查询实现缓存，提高响应速度并降低成本
7. 文档检索不准确：优化分块策略，添加文档元数据，使用查询重写

1. 文档链性能优化

• 向量数据库优化：使用search_kwargs={"k": 3}限制返回结果数量
• 批量处理：利用chain.batch()方法处理多个查询，配置max_concurrency控制并发
• 缓存机制：对重复查询启用缓存，减少API调用

2. 处理文档链中的”幻觉”

• 严格引用检查：设置return_source_documents=True验证答案来源
• 置信度阈值：对低置信度结果返回”无法确定”而非猜测
• 多文档交叉验证：要求关键信息至少出现在两个文档中

3. 调试与监控

• 启用详细日志：设置verbose=True查看完整处理流程
• 使用LangSmith：集成LangChain官方监控平台，跟踪链执行情况
• 回调系统：添加FileCallbackHandler记录关键步骤

建议的进一步问题

基于LangChain 1.0+的文档链开发，以下问题值得深入探讨：

1. 性能优化：在LangChain 1.0+中，如何通过LCEL的batch和stream方法优化create_stuff_documents_chain处理大量文档时的性能？

2. 错误处理：当文档链处理过程中遇到格式异常或内容缺失时，LangChain 1.0+提供了哪些新的错误恢复和重试机制？

3. 自定义扩展：如何基于Runnable接口自定义文档链，以支持特定格式的文档处理（如表格提取、代码片段分析、多语言混合内容）？

4. 评估监控：使用LangSmith在LangChain 1.0+中如何监控和评估文档链的准确率、响应时间和资源消耗？

5. 生产部署：将基于create_retrieval_chain的RAG系统部署到生产环境时，需要注意哪些新的性能瓶颈和扩展性问题？

6. 版本迁移：从LangChain 0.x迁移到1.0+时，文档链相关的代码需要做哪些主要改动？特别是StuffDocumentsChain到create_stuff_documents_chain的迁移路径。

7. 混合链设计：如何结合create_stuff_documents_chain与其他链（如工具调用链、路由链）构建复杂的多步骤文档处理流水线？

8. 上下文管理：在LangChain 1.0+中，如何处理超长文档的分块和上下文窗口限制，特别是与新的RunnableWithMessageHistory组件的集成？