LangChain基础

LangChain是一个强大的框架，用于构建基于大语言模型的应用程序，特别适合实现RAG系统。本章节将详细介绍LangChain 1.2版本的核心概念、组件和使用方法，帮助你快速上手构建RAG应用。

1. LangChain核心概念

组件（Components）

• LLMs：大语言模型，如OpenAI GPT、Claude等
• Embeddings：嵌入模型，用于将文本转换为向量
• Document Loaders：文档加载器，用于从各种源加载文档
• Text Splitters：文本分割器，用于将长文本分割成小块
• Vector Stores：向量存储，用于存储和检索向量
• Retrievers：检索器，用于从向量存储中检索相关文档
• Chains：链，用于组合多个组件
• Agents：代理，用于基于用户输入做出决策

工作流（Workflows）

1. 数据加载：使用Document Loaders加载文档
2. 文本分割：使用Text Splitters分割文本
3. 向量存储：使用Embeddings将文本转换为向量并存储
4. 检索：使用Retrievers检索相关文档
5. 生成：使用LLMs基于检索结果生成回答

2. 安装与配置

安装LangChain

pip install langchain==1.2.0

安装依赖

# 安装OpenAI依赖（如果使用OpenAI模型）
pip install openai

# 安装Hugging Face依赖（如果使用Hugging Face模型）
pip install transformers sentence-transformers

# 安装向量数据库依赖（根据选择）
pip install chromadb  # 轻量级向量数据库
# 或 pip install pinecone-client  # 托管向量数据库

# 安装文档加载器依赖
pip install pypdf python-docx

# 安装环境变量管理
pip install python-dotenv

配置API密钥

from dotenv import load_dotenv
import os

# 加载.env文件
load_dotenv()

# 设置API密钥
os.environ["OPENAI_API_KEY"] = "your-api-key"

3. 文档处理

加载文档

from langchain.document_loaders import TextLoader, PyPDFLoader, DirectoryLoader

# 加载文本文件
loader = TextLoader("document.txt")
documents = loader.load()

# 加载PDF文件
pdf_loader = PyPDFLoader("document.pdf")
pdf_documents = pdf_loader.load()

# 加载整个目录
dir_loader = DirectoryLoader(
    "./docs",
    glob="**/*.txt",
    loader_cls=TextLoader
)
dir_documents = dir_loader.load()

分割文档

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 创建文本分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,  # 每个块的大小
    chunk_overlap=200,  # 重叠大小
    length_function=len,
)

# 分割文档
chunks = text_splitter.split_documents(documents)
print(f"分割成 {len(chunks)} 个块")

4. 向量存储与检索

创建向量存储

from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings
from langchain.vectorstores import Chroma, FAISS

# 选择嵌入模型
# embeddings = OpenAIEmbeddings()
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

# 创建Chroma向量存储
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 或使用FAISS
# vectorstore = FAISS.from_documents(documents=chunks, embedding=embeddings)

相似度搜索

# 直接搜索
results = vectorstore.similarity_search("什么是RAG？", k=3)
for doc in results:
    print(doc.page_content)
    print("---")

# 带分数的搜索
results_with_scores = vectorstore.similarity_search_with_score("什么是RAG？", k=3)
for doc, score in results_with_scores:
    print(f"分数: {score}")
    print(doc.page_content)
    print("---")

5. 构建RAG链

基础RAG链

from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

# 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 创建RAG链
qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(temperature=0),
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 执行查询
result = qa_chain({"query": "什么是RAG技术？"})
print(result["result"])
print("\n来源文档:")
for doc in result["source_documents"]:
    print(f"- {doc.metadata.get('source', 'Unknown')}")

自定义提示模板

from langchain.prompts import PromptTemplate

# 创建自定义提示模板
custom_prompt = PromptTemplate(
    template="""基于以下上下文回答问题。如果无法从上下文中找到答案，请说"我不知道"。

上下文：
{context}

问题：{question}

请提供详细且准确的回答：""",
    input_variables=["context", "question"]
)

# 使用自定义提示创建链
qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(temperature=0),
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": custom_prompt}
)

6. 高级链类型

Map-Reduce链

适用于大量文档的场景：

qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(temperature=0),
    chain_type="map_reduce",
    retriever=retriever
)

Refine链

迭代优化回答：

qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(temperature=0),
    chain_type="refine",
    retriever=retriever
)

7. 完整示例

from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

class RAGSystem:
    def __init__(self):
        self.embeddings = HuggingFaceEmbeddings()
        self.vectorstore = None
        self.qa_chain = None
    
    def load_documents(self, file_path):
        """加载文档"""
        loader = TextLoader(file_path)
        documents = loader.load()
        
        # 分割文档
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        chunks = text_splitter.split_documents(documents)
        
        # 创建向量存储
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings
        )
        
        # 创建RAG链
        retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=OpenAI(temperature=0),
            chain_type="stuff",
            retriever=retriever,
            return_source_documents=True
        )
        
        print(f"已加载 {len(chunks)} 个文档块")
    
    def query(self, question):
        """查询"""
        if not self.qa_chain:
            raise ValueError("请先加载文档")
        
        result = self.qa_chain({"query": question})
        return {
            "answer": result["result"],
            "sources": [doc.metadata.get('source', 'Unknown') 
                       for doc in result["source_documents"]]
        }

# 使用示例
rag = RAGSystem()
rag.load_documents("knowledge_base.txt")
response = rag.query("什么是RAG技术？")
print(response["answer"])

8. 调试与监控

查看检索结果

# 启用详细输出
qa_chain.verbose = True

# 或者手动检查检索结果
retriever = vectorstore.as_retriever()
retrieved_docs = retriever.get_relevant_documents("查询")
for i, doc in enumerate(retrieved_docs):
    print(f"文档 {i+1}:")
    print(doc.page_content[:200])
    print("---")

回调函数

from langchain.callbacks import StdOutCallbackHandler

qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(temperature=0),
    chain_type="stuff",
    retriever=retriever,
    callbacks=[StdOutCallbackHandler()]
)