🔍 RAG（检索增强生成）详解

# 安装必要的包
pip install -U langchain langchain-openai langchain-chroma
pip install langchain-community langchain-text-splitters
pip install bs4  # 用于网页加载

import bs4
from langchain_community.document_loaders import WebBaseLoader, PyPDFLoader, TextLoader

# 方式1：加载网页
web_loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs=dict(
        parse_only=bs4.SoupStrainer(
            class_=("post-content", "post-title", "post-header")
        )
    ),
)
web_docs = web_loader.load()

# 方式2：加载 PDF
pdf_loader = PyPDFLoader("document.pdf")
pdf_docs = pdf_loader.load()

# 方式3：加载文本文件
text_loader = TextLoader("document.txt", encoding="utf-8")
text_docs = text_loader.load()

print(f"加载了 {len(web_docs)} 个文档")
print(f"第一个文档内容长度: {len(web_docs[0].page_content)} 字符")

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 创建文本分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,        # 每个块的字符数
    chunk_overlap=200,      # 块之间的重叠字符数
    length_function=len,    # 计算长度的函数
    is_separator_regex=False,
)

# 分割文档
all_splits = text_splitter.split_documents(web_docs)

print(f"文档被分割成 {len(all_splits)} 个块")
print(f"第一个块长度: {len(all_splits[0].page_content)} 字符")
print(f"第一个块内容预览: {all_splits[0].page_content[:200]}...")

import os
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

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

# 初始化 Embeddings 模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 创建向量存储并添加文档
vector_store = Chroma.from_documents(
    documents=all_splits,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 持久化存储路径
)

print(f"向量存储创建完成，包含 {len(all_splits)} 个文档块")

from langchain.tools import tool

@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """从知识库检索相关信息来回答用户问题。

    Args:
        query: 用户的查询问题

    Returns:
        检索到的相关文档内容和原始文档对象
    """
    # 使用相似度搜索检索相关文档（返回最相关的 k=2 个文档）
    retrieved_docs = vector_store.similarity_search(query, k=2)

    # 格式化检索结果
    serialized = "\n\n".join(
        f"来源: {doc.metadata}\n内容: {doc.page_content}"
        for doc in retrieved_docs
    )

    return serialized, retrieved_docs

# 测试检索工具
test_query = "什么是 Agent？"
result, docs = retrieve_context.invoke({"query": test_query})
print(f"检索结果:\n{result}")

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

# 初始化模型
model = init_chat_model("gpt-4", model_provider="openai")

# 定义系统提示词
system_prompt = """你是一个智能问答助手，可以访问知识库来回答问题。

使用检索工具 retrieve_context 来查找相关信息。当用户提问时：
1. 先使用检索工具查找相关信息
2. 基于检索到的信息生成准确的答案
3. 如果检索不到相关信息，诚实地告知用户
4. 在回答中引用信息来源

始终保持专业、准确和有帮助。"""

# 创建 Agent
agent = create_agent(
    model=model,
    tools=[retrieve_context],
    system_prompt=system_prompt
)

# 使用 Agent 回答问题
response = agent.invoke({
    "messages": [{"role": "user", "content": "请介绍一下 LangChain 中的 Agent"}]
})

print(response["messages"][-1].content)

"""
完整的 RAG 系统实现示例
适用于 LangChain 1.0
"""
import os
import bs4
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

# ==================== 配置 ====================
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# ==================== 第1步：加载文档 ====================
print("正在加载文档...")
loader = WebBaseLoader(
    web_paths=("https://python.langchain.com/docs/tutorials/agents/",),
    bs_kwargs=dict(parse_only=bs4.SoupStrainer(class_=("main",)))
)
docs = loader.load()
print(f"✓ 加载了 {len(docs)} 个文档")

# ==================== 第2步：分割文本 ====================
print("正在分割文本...")
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
all_splits = text_splitter.split_documents(docs)
print(f"✓ 分割成 {len(all_splits)} 个文本块")

# ==================== 第3步：创建向量存储 ====================
print("正在创建向量存储...")
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = Chroma.from_documents(
    documents=all_splits,
    embedding=embeddings,
    persist_directory="./chroma_langchain_docs"
)
print(f"✓ 向量存储创建完成")

# ==================== 第4步：创建检索工具 ====================
@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """从 LangChain 文档中检索相关信息。"""
    retrieved_docs = vector_store.similarity_search(query, k=3)
    serialized = "\n\n".join(
        f"【文档 {i+1}】\n{doc.page_content}"
        for i, doc in enumerate(retrieved_docs)
    )
    return serialized, retrieved_docs

# ==================== 第5步：创建 RAG Agent ====================
print("正在创建 RAG Agent...")
model = init_chat_model("gpt-4o-mini", model_provider="openai")

system_prompt = """你是 LangChain 文档助手。使用 retrieve_context 工具查找相关信息。

回答要求：
- 先检索相关信息再回答
- 基于检索内容生成准确答案
- 如果信息不足，说明需要更多上下文
- 用中文回答，保持专业和友好"""

agent = create_agent(
    model=model,
    tools=[retrieve_context],
    system_prompt=system_prompt
)
print("✓ RAG Agent 创建完成\n")

# ==================== 第6步：交互式问答 ====================
def ask_question(question: str):
    """向 RAG Agent 提问"""
    print(f"\n{'='*60}")
    print(f"问题: {question}")
    print('='*60)

    response = agent.invoke({
        "messages": [{"role": "user", "content": question}]
    })

    answer = response["messages"][-1].content
    print(f"\n回答:\n{answer}\n")
    return answer

# 示例问答
if __name__ == "__main__":
    questions = [
        "什么是 LangChain Agent？",
        "如何创建一个 Agent？",
        "LangChain 支持哪些模型提供商？"
    ]

    for q in questions:
        ask_question(q)

# 使用 MMR（最大边际相关性）检索，避免结果重复
retrieved_docs = vector_store.max_marginal_relevance_search(
    query="用户问题",
    k=4,              # 返回4个结果
    fetch_k=20,       # 先获取20个候选
    lambda_mult=0.5   # 平衡相关性和多样性
)

# 使用相似度阈值过滤低质量结果
retrieved_docs = vector_store.similarity_search_with_score(
    query="用户问题",
    k=5
)
filtered_docs = [doc for doc, score in retrieved_docs if score > 0.7]

from functools import lru_cache

@lru_cache(maxsize=100)
def cached_retrieve(query: str, k: int = 3):
    """缓存检索结果"""
    return vector_store.similarity_search(query, k=k)

# 使用缓存的检索
results = cached_retrieve("什么是 Agent？")

# 使用支持多语言的 embeddings 模型
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large"  # 支持多语言
)

from langchain.retrievers import EnsembleRetriever
from langchain.retrievers import BM25Retriever

# 创建多个检索器
vector_retriever = vector_store.as_retriever(search_kwargs={"k": 3})
bm25_retriever = BM25Retriever.from_documents(all_splits)
bm25_retriever.k = 3

# 组合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4]  # 语义搜索60%，关键词搜索40%
)

# 使用混合检索
results = ensemble_retriever.get_relevant_documents("用户问题")

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

# 使用 Cohere 重排序模型
compressor = CohereRerank(model="rerank-english-v2.0")

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_retriever
)

# 获取重排序后的结果
compressed_docs = compression_retriever.get_relevant_documents("用户问题")