目录
模块 II：RAG
如果在应用中需要使用特定的数据（未包含在模型训练集中的数据），此时就需要使用本地知识，即检索增强生成（Retrieval Augmented Generation, RAG），包括获取外部数据并将其整合到语言模型的生成过程中。LangChain 提供了一套全面的工具和功能，以促进这一过程，适用于从简单到复杂的应用场景。
文档加载器
LangChain 中的文档加载器能够从各种来源提取数据。LangChain 提供了 100 多种加载器，支持多种文档类型、应用程序和数据源（例如私有 S3 存储桶、公共网站、数据库等）。你可以根据需求选择合适的文档加载器（官方文档列出了所有文档加载器）。
所有这些加载器都会将数据导入到 Document 类中。稍后将学习如何使用导入到 Document 类中的数据。
1. 文本文件加载器：将简单的 .txt 文件加载为文档。
from langchain.document_loaders import TextLoader
loader = TextLoader("./sample.txt")
document = loader.load()
2. CSV 加载器：将 CSV 文件加载为文档。
from langchain.document_loaders.csv_loader import CSVLoader
loader = CSVLoader(file_path='./example_data/sample.csv')
documents = loader.load()
可以选择通过指定字段名称来自定义解析过程：
loader = CSVLoader(file_path='./example_data/mlb_teams_2012.csv', csv_args={
'delimiter': ',',
'quotechar': '"',
'fieldnames': ['MLB Team', 'Payroll in millions', 'Wins']
})
documents = loader.load()
3. PDF 加载器：LangChain 中的 PDF 加载器提供了多种方法来解析和提取 PDF 文件中的内容。每个加载器适用于不同的需求，并使用不同的底层库。以下是几个示例。
（1）PyPDFLoader 用于基本的 PDF 解析。
from langchain.document_loaders import PyPDFLoader
loader = PyPDFLoader("example_data/layout-parser-paper.pdf")
pages = loader.load_and_split()
（2）MathPixLoader 非常适合提取数学内容和图表。
from langchain.document_loaders import MathpixPDFLoader
loader = MathpixPDFLoader("example_data/math-content.pdf")
data = loader.load()
（3）PyMuPDFLoader 速度快，并且支持详细的元数据提取。
from langchain.document_loaders import PyMuPDFLoader
loader = PyMuPDFLoader("example_data/layout-parser-paper.pdf")
data = loader.load()
# Optionally pass additional arguments for PyMuPDF's get_text() call
data = loader.load(option="text")
（4）PDFMiner Loader 用于对文本提取进行更精细的控制。
from langchain.document_loaders import PDFMinerLoader
loader = PDFMinerLoader("example_data/layout-parser-paper.pdf")
data = loader.load()
（5）AmazonTextractPDFParser 利用 AWS Textract 实现 OCR 和其他高级 PDF 解析功能。
from langchain.document_loaders import AmazonTextractPDFLoader
# Requires AWS account and configuration
loader = AmazonTextractPDFLoader("example_data/complex-layout.pdf")
documents = loader.load()
（6）PDFMinerPDFasHTMLLoader 将 PDF 转换为 HTML，以实现语义解析。
from langchain.document_loaders import PDFMinerPDFasHTMLLoader
loader = PDFMinerPDFasHTMLLoader("example_data/layout-parser-paper.pdf")
data = loader.load()
（7）PDFPlumberLoader 提供详细的元数据支持，并且可以实现每页生成一个文档。
from langchain.document_loaders import PDFPlumberLoader
loader = PDFPlumberLoader("example_data/layout-parser-paper.pdf")
data = loader.load()
（8）集成加载器：LangChain 提供了各种自定义加载器，可以直接从应用程序（如 Slack、Sigma、Notion、Confluence、Google Drive 等）和数据库中加载数据，并将其用于大语言模型（LLM）应用程序。
完整的加载器列表可以在此处查看。
文档转换器
LangChain 中的文档转换器是用于操作文档的重要工具，这些文档是我们之前小节中创建的。
它们用于诸如将长文档拆分为较小片段、合并和过滤等任务，这对于将文档适配到模型的上下文窗口或满足特定应用需求至关重要。
其中一个工具是 RecursiveCharacterTextSplitter，这是一款多功能的文本分割器，它使用字符列表来进行分割。它允许设置诸如块大小（chunk size）、重叠（overlap）和起始索引（starting index）等参数。以下是一个在 Python 中如何使用它的示例：
from langchain.text_splitter import RecursiveCharacterTextSplitter
state_of_the_union = "Your long text here..."
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=100,
chunk_overlap=20,
length_function=len,
add_start_index=True,
)
texts = text_splitter.create_documents([state_of_the_union])
print(texts[0])
print(texts[1])
另一个工具是 CharacterTextSplitter，它基于指定的字符来分割文本，并包含对块大小（chunk size）和重叠（overlap）的控制：
from langchain.text_splitter import CharacterTextSplitter
text_splitter = CharacterTextSplitter(
separator="\n\n",
chunk_size=1000,
chunk_overlap=200,
length_function=len,
is_separator_regex=False,
)
texts = text_splitter.create_documents([state_of_the_union])
print(texts[0])
HTMLHeaderTextSplitter 旨在根据标题标签（header tags）分割 HTML 内容，同时保留语义结构：
from langchain.text_splitter import HTMLHeaderTextSplitter
html_string = "Your HTML content here..."
headers_to_split_on = [("h1", "Header 1"), ("h2", "Header 2")]
html_splitter = HTMLHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
html_header_splits = html_splitter.split_text(html_string)
print(html_header_splits[0])
通过将 HTMLHeaderTextSplitter 与另一个分割器（如 Pipelined Splitter）结合使用，可以实现更复杂的操作：
from langchain.text_splitter import HTMLHeaderTextSplitter, RecursiveCharacterTextSplitter
url = "https://example.com"
headers_to_split_on = [("h1", "Header 1"), ("h2", "Header 2")]
html_splitter = HTMLHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
html_header_splits = html_splitter.split_text_from_url(url)
chunk_size = 500
text_splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size)
splits = text_splitter.split_documents(html_header_splits)
print(splits[0])
LangChain 还提供了针对不同编程语言的专用分割器，例如 Python 代码分割器和 JavaScript 代码分割器：
from langchain.text_splitter import RecursiveCharacterTextSplitter, Language
python_code = """
def hello_world():
print("Hello, World!")
hello_world()
"""
python_splitter = RecursiveCharacterTextSplitter.from_language(
language=Language.PYTHON, chunk_size=50
)
python_docs = python_splitter.create_documents([python_code])
print(python_docs[0])
js_code = """
function helloWorld() {
console.log("Hello, World!");
}
helloWorld();
"""
js_splitter = RecursiveCharacterTextSplitter.from_language(
language=Language.JS, chunk_size=60
)
js_docs = js_splitter.create_documents([js_code])
print(js_docs[0])
为了根据 token 数量分割文本（这对于有标记限制的语言模型非常有用），可以使用 TokenTextSplitter：
from langchain.text_splitter import TokenTextSplitter
text_splitter = TokenTextSplitter(chunk_size=10)
texts = text_splitter.split_text(state_of_the_union)
print(texts[0])
最后，LongContextReorder 会对文档进行重新排序，以防止因上下文过长而导致模型性能下降：
from langchain.document_transformers import LongContextReorder
reordering = LongContextReorder()
reordered_docs = reordering.transform_documents(docs)
print(reordered_docs[0])
这些工具展示了在 LangChain 中转换文档的各种方法，从简单的文本分割到复杂的重新排序和针对特定语言的分割。对于更深入和具体的用例，可以参考 LangChain 的文档和集成部分。
在我们的示例中，加载器已经创建了分块的文档，这部分内容已经处理完毕。
文本嵌入模型
关于文本嵌入的基本原理，请阅读：LLM(12)：创建 token 嵌入
LangChain 中的文本嵌入模型为各种嵌入模型提供商（如 OpenAI、Cohere 和 Hugging Face）提供了一个标准化的接口。这些模型将文本转换为向量表示，从而能够通过向量空间中的文本相似性执行诸如语义搜索等操作。
要使用文本嵌入模型，通常需要安装特定的软件包并设置 API 密钥。我们已经为 OpenAI 完成了这些步骤。
在 LangChain 中，embed_documents 方法用于嵌入多个文本，返回一个包含向量表示的列表。例如：
from langchain.embeddings import OpenAIEmbeddings
# Initialize the model
embeddings_model = OpenAIEmbeddings()
# Embed a list of texts
embeddings = embeddings_model.embed_documents(
["Hi there!", "Oh, hello!", "What's your name?", "My friends call me World", "Hello World!"]
)
print("Number of documents embedded:", len(embeddings))
print("Dimension of each embedding:", len(embeddings[0]))
对于嵌入单个文本（例如搜索查询），可以使用 embed_query 方法。这在将查询与一组文档嵌入进行比较时非常有用。例如：
from langchain.embeddings import OpenAIEmbeddings
# Initialize the model
embeddings_model = OpenAIEmbeddings()
# Embed a single query
embedded_query = embeddings_model.embed_query("What was the name mentioned in the conversation?")
print("First five dimensions of the embedded query:", embedded_query[:5])
理解这些嵌入的原理非常重要。每一段文本都会被转换为一个向量，其维度取决于所使用的模型。例如，OpenAI 模型通常生成 1536 维的向量。这些嵌入随后被用于检索相关信息。
LangChain 的嵌入功能并不局限于 OpenAI，而是设计为与各种提供商兼容。不同提供商的设置和使用方式可能略有差异，但将文本嵌入到向量空间的核心概念保持不变。有关高级配置以及与不同嵌入模型提供商集成的详细用法，请阅读 LangChain 文档中的 Embedding models。
向量存储
LangChain 中的向量存储支持对文本嵌入进行高效的存储和搜索。LangChain 集成了超过 50 种向量存储，并提供了一个标准化接口以便于使用。
示例：存储和搜索嵌入
在嵌入文本后，我们可以将它们存储到像 Chroma 这样的向量存储中，并执行相似性搜索：
from langchain.vectorstores import Chroma
db = Chroma.from_texts(embedded_texts)
similar_texts = db.similarity_search("search query")
换一种方式，使用 FAISS 向量存储 为文档创建索引。
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
pdfstore = FAISS.from_documents(pdfpages,
embedding=OpenAIEmbeddings())
airtablestore = FAISS.from_documents(airtabledocs,
embedding=OpenAIEmbeddings())
检索器
LangChain 中的检索器（Retrievers）是一种接口，用于响应非结构化查询并返回文档。它们比向量存储更通用，专注于检索而非存储。尽管向量存储可以用作检索器的基础，但也有其他类型的检索器。
要设置一个 Chroma 检索器，首先需要使用 pip install chromadb 安装它。然后，通过一系列 Python 命令加载、分割、嵌入并检索文档。以下是一个设置 Chroma 检索器的代码示例：
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import Chroma
full_text = open("state_of_the_union.txt", "r").read()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
texts = text_splitter.split_text(full_text)
embeddings = OpenAIEmbeddings()
db = Chroma.from_texts(texts, embeddings)
retriever = db.as_retriever()
retrieved_docs = retriever.invoke("What did the president say about Ketanji Brown Jackson?")
print(retrieved_docs[0].page_content)
MultiQueryRetriever 通过为用户输入的查询生成多个查询并合并结果，从而实现提示调优的自动化。以下是一个简单使用的示例：
from langchain.chat_models import ChatOpenAI
from langchain.retrievers.multi_query import MultiQueryRetriever
question = "What are the approaches to Task Decomposition?"
llm = ChatOpenAI(temperature=0)
retriever_from_llm = MultiQueryRetriever.from_llm(
retriever=db.as_retriever(), llm=llm
)
unique_docs = retriever_from_llm.get_relevant_documents(query=question)
print("Number of unique documents:", len(unique_docs))
LangChain 中的上下文压缩（Contextual Compression） 会利用查询的上下文对检索到的文档进行压缩，确保只返回相关的信息。这涉及内容缩减以及过滤掉不太相关的文档。以下代码示例展示了如何使用 ContextualCompressionRetriever：
from langchain.llms import OpenAI
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
llm = OpenAI(temperature=0)
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(base_compressor=compressor, base_retriever=retriever)
compressed_docs = compression_retriever.get_relevant_documents("What did the president say about Ketanji Jackson Brown")
print(compressed_docs[0].page_content)
EnsembleRetriever 通过结合不同的检索算法来实现更优的性能。以下代码展示了如何将 BM25 和 FAISS 检索器结合起来的示例：
from langchain.retrievers import BM25Retriever, EnsembleRetriever
from langchain.vectorstores import FAISS
bm25_retriever = BM25Retriever.from_texts(doc_list).set_k(2)
faiss_vectorstore = FAISS.from_texts(doc_list, OpenAIEmbeddings())
faiss_retriever = faiss_vectorstore.as_retriever(search_kwargs={"k": 2})
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, faiss_retriever], weights=[0.5, 0.5]
)
docs = ensemble_retriever.get_relevant_documents("apples")
print(docs[0].page_content)
LangChain 中的 MultiVectorRetriever 允许为每个文档使用多个向量进行查询，这有助于捕捉文档内的不同语义方面。创建多个向量的方法包括将文档拆分为更小的片段、生成摘要或提出假设性问题。以下 Python 代码展示了如何将文档拆分为更小的片段：
from langchain.retrievers.multi_vector import MultiVectorRetriever
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.storage import InMemoryStore
from langchain.document_loaders from TextLoader
import uuid
loaders = [TextLoader("file1.txt"), TextLoader("file2.txt")]
docs = [doc for loader in loaders for doc in loader.load()]
text_splitter = RecursiveCharacterTextSplitter(chunk_size=10000)
docs = text_splitter.split_documents(docs)
vectorstore = Chroma(collection_name="full_documents", embedding_function=OpenAIEmbeddings())
store = InMemoryStore()
id_key = "doc_id"
retriever = MultiVectorRetriever(vectorstore=vectorstore, docstore=store, id_key=id_key)
doc_ids = [str(uuid.uuid4()) for _ in docs]
child_text_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
sub_docs = [sub_doc for doc in docs for sub_doc in child_text_splitter.split_documents([doc])]
for sub_doc in sub_docs:
sub_doc.metadata[id_key] = doc_ids[sub_docs.index(sub_doc)]
retriever.vectorstore.add_documents(sub_docs)
retriever.docstore.mset(list(zip(doc_ids, docs)))
生成摘要以通过更聚焦的内容表示来改进检索是另一种方法。以下是一个生成摘要的示例：
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.document import Document
chain = (lambda x: x.page_content) | ChatPromptTemplate.from_template("Summarize the following document:\n\n{doc}") | ChatOpenAI(max_retries=0) | StrOutputParser()
summaries = chain.batch(docs, {"max_concurrency": 5})
summary_docs = [Document(page_content=s, metadata={id_key: doc_ids[i]}) for i, s in enumerate(summaries)]
retriever.vectorstore.add_documents(summary_docs)
retriever.docstore.mset(list(zip(doc_ids, docs)))
使用大语言模型（LLM）为每个文档生成相关的假设性问题是另一种方法。这可以通过以下代码实现：
functions = [{"name": "hypothetical_questions", "parameters": {"questions": {"type": "array", "items": {"type": "string"}}}}]
from langchain.output_parsers.openai_functions import JsonKeyOutputFunctionsParser
chain = (lambda x: x.page_content) | ChatPromptTemplate.from_template("Generate 3 hypothetical questions:\n\n{doc}") | ChatOpenAI(max_retries=0).bind(functions=functions, function_call={"name": "hypothetical_questions"}) | JsonKeyOutputFunctionsParser(key_name="questions")
hypothetical_questions = chain.batch(docs, {"max_concurrency": 5})
question_docs = [Document(page_content=q, metadata={id_key: doc_ids[i]}) for i, questions in enumerate(hypothetical_questions) for q in questions]
retriever.vectorstore.add_documents(question_docs)
retriever.docstore.mset(list(zip(doc_ids, docs)))
ParentDocumentRetriever 是另一种检索器，它通过存储小片段并检索其较大的父文档，在嵌入准确性与上下文保留之间取得平衡。其具体实现如下：
from langchain.retrievers import ParentDocumentRetriever
loaders = [TextLoader("file1.txt"), TextLoader("file2.txt")]
docs = [doc for loader in loaders for doc in loader.load()]
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
vectorstore = Chroma(collection_name="full_documents", embedding_function=OpenAIEmbeddings())
store = InMemoryStore()
retriever = ParentDocumentRetriever(vectorstore=vectorstore, docstore=store, child_splitter=child_splitter)
retriever.add_documents(docs, ids=None)
retrieved_docs = retriever.get_relevant_documents("query")
自查询检索器（SelfQueryRetriever）会从自然语言输入中构造结构化查询，并将其应用于底层的 VectorStore。其实现如下代码所示：
from langchain.chat_models from ChatOpenAI
from langchain.chains.query_constructor.base from AttributeInfo
from langchain.retrievers.self_query.base from SelfQueryRetriever
metadata_field_info = [AttributeInfo(name="genre", description="...", type="string"), ...]
document_content_description = "Brief summary of a movie"
llm = ChatOpenAI(temperature=0)
retriever = SelfQueryRetriever.from_llm(llm, vectorstore, document_content_description, metadata_field_info)
retrieved_docs = retriever.invoke("query")
WebResearchRetriever 根据给定的查询执行网络搜索：
from langchain.retrievers.web_research import WebResearchRetriever
# Initialize components
llm = ChatOpenAI(temperature=0)
search = GoogleSearchAPIWrapper()
vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
# Instantiate WebResearchRetriever
web_research_retriever = WebResearchRetriever.from_llm(vectorstore=vectorstore, llm=llm, search=search)
# Retrieve documents
docs = web_research_retriever.get_relevant_documents("query")
在示例中，我们也可以使用已经作为向量存储对象的一部分实现的标准检索器，如下所示：
pdfretriever = pdfstore.as_retriever()
airtableretriever = airtablestore.as_retriever()
现在可以通过检索器进行查询，输出将是与查询相关的文档对象。这些文档最终将被用于在后续部分中创建相关的响应。
扫一扫
被折叠的 条评论
为什么被折叠?
到【灌水乐园】发言
