RAG

请基于以下参考资料回答用户问题。
参考资料：{检索结果}
问题：{用户问题}

from langchain_core.documents import Document

# Document 的基本结构
doc = Document(
    page_content="这是文档的正文内容",
    metadata={"source": "example.txt", "page": 1}
)

from langchain_community.document_loaders import DirectoryLoader, TextLoader

loader = DirectoryLoader(
    path="./docs",                    # 目录路径
    glob="**/*.md",                   # glob 模式匹配文件
    loader_cls=TextLoader,            # 指定文件加载器类
    loader_kwargs={"encoding": "utf-8"},  # 传递给加载器的参数
    show_progress=True,               # 显示加载进度条
    use_multithreading=True,          # 多线程加载
    max_concurrency=4,                # 最大并发数
    silent_errors=True,               # 跳过加载失败的文件（不抛异常）
)
docs = loader.load()
print(f"共加载 {len(docs)} 个文件")

import pypdf
from langchain_core.documents import Document


def load_pdf_pages(file_path: str) -> list[Document]:
    reader = pypdf.PdfReader(file_path)
    return [
        Document(
            page_content=page.extract_text() or "",
            metadata={"source": file_path, "page": i},
        )
        for i, page in enumerate(reader.pages)
    ]

docs = load_pdf_pages("document.pdf")

from langchain_text_splitters import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,            # 每个片段的最大字符数
    chunk_overlap=200,          # 相邻片段的重叠字符数（保持上下文连贯）
    length_function=len,        # 长度计算函数（默认按字符数）
    is_separator_regex=False,   # 分隔符是否为正则表达式
    add_start_index=True,       # 在元数据中记录片段在原文的起始位置
)

# 切分 Document 对象
splits = text_splitter.split_documents(docs)
print(f"切分为 {len(splits)} 个片段")

# 也可以直接切分纯文本字符串
chunks = text_splitter.split_text("一段很长的文本...")

# 安装：pip install langchain-experimental
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",   # 断点阈值类型
    breakpoint_threshold_amount=95,            # 百分位阈值（百分位类型时使用）
    number_of_chunks=None,                    # 手动指定切分数量
    min_chunk_size=None,                      # 最小片段字符数
)
splits = splitter.create_documents([long_text])

from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

from langchain.embeddings import init_embeddings

# 格式："provider:model_name"
embeddings = init_embeddings("openai:text-embedding-3-small")
embeddings = init_embeddings("openai:text-embedding-3-large")

# 本地 HuggingFace 模型
embeddings = init_embeddings("huggingface:BAAI/bge-m3")
embeddings = init_embeddings("huggingface:sentence-transformers/all-mpnet-base-v2")

# Ollama 本地服务
embeddings = init_embeddings("ollama:nomic-embed-text")

# 典型 RAG 流程中的用法
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 1. 索引阶段：批量嵌入文档片段（推荐用 embed_documents）
doc_chunks = [doc.page_content for doc in splits]
doc_vectors = embeddings.embed_documents(doc_chunks)
print(f"已嵌入 {len(doc_vectors)} 个文档片段")

# 2. 查询阶段：嵌入用户问题（用 embed_query）
query_vector = embeddings.embed_query("什么是 RAG？")
print(f"查询向量维度: {len(query_vector)}")

# text-embedding-3 系列支持截断输出维度，质量损失可控
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large",
    dimensions=256,  # 从 3072 截断到 256，大幅节省存储和计算
)

# 安装：pip install langchain-classic
from langchain_classic.embeddings import CacheBackedEmbeddings
from langchain_classic.storage import LocalFileStore

underlying_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 使用本地文件系统缓存（开发环境）
store = LocalFileStore("./cache/")
# 生产环境建议使用 RedisStore / SQLStore 等持久化存储

cached_embedder = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings,
    store,
    namespace="text-embedding-3-small",  # 不同模型使用不同命名空间，避免冲突
)

# 首次调用会请求 API 并缓存结果
vector = cached_embedder.embed_query("什么是 RAG？")
# 二次调用直接从缓存读取，几乎无耗时

import numpy as np

def cosine_similarity(vec1, vec2):
    """计算两个向量的余弦相似度"""
    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

query_vec = embeddings.embed_query("什么是 RAG？")
doc_vec = embeddings.embed_documents(["RAG 是检索增强生成技术"])[0]
sim = cosine_similarity(query_vec, doc_vec)
print(f"相似度: {sim:.4f}")  # 越接近 1 越相似

from langchain_huggingface import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="intfloat/e5-large-v2",
    encode_kwargs={"prompt": "passage: "},          # 文档端前缀
    query_encode_kwargs={"prompt": "query: "},     # 查询端前缀（部分版本支持）
)

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

# 创建向量库并存储文档
embeddings = OpenAIEmbeddings()
vectorstore = InMemoryVectorStore.from_documents(
    documents=splits,
    embedding=embeddings,
)

# 也可以分步操作
vectorstore = InMemoryVectorStore(embeddings)
vectorstore.add_documents(splits)

# 将向量库转换为检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",  # 检索类型
    search_kwargs={"k": 4},    # 返回前 4 个最相关的结果
)

# 使用检索器
docs = retriever.invoke("什么是 RAG？")
for doc in docs:
    print(doc.page_content[:100], "...")

# MMR 检索：避免返回过于相似的结果，提高多样性
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.5},
)

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import bs4

# 1. 加载文档
loader = WebBaseLoader(
    web_paths=["https://example.com/article"],
    bs_kwargs={"parse_only": bs4.SoupStrainer(class_=("post-content",))},
)
docs = loader.load()

# 2. 切分文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    add_start_index=True,
)
splits = text_splitter.split_documents(docs)

# 3. 创建向量库并存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = InMemoryVectorStore.from_documents(
    documents=splits,
    embedding=embeddings,
)

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

# 5. 构建 RAG 链
llm = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题。如果上下文中没有相关信息，请说明你不确定。

上下文：
{context}

问题：{question}
""")


def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)


rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 6. 提问
answer = rag_chain.invoke("这篇文章的主要观点是什么？")
print(answer)

# 上面的 RAG 链可以用更简洁的方式表达
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 流式输出
for chunk in rag_chain.stream("解释一下 RAG 的工作原理"):
    print(chunk, end="", flush=True)

import bs4
from langchain_community.document_loaders import WebBaseLoader

# 使用 BeautifulSoup 过滤器只提取关键内容
loader = WebBaseLoader(
    web_paths=["https://example.com/article"],
    bs_kwargs={
        "parse_only": bs4.SoupStrainer(
            class_=("post-title", "post-content", "post-meta")
        )
    },
)
docs = loader.load()

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")

multi_query_prompt = ChatPromptTemplate.from_template(
    "你是一个 AI 助手。请为以下问题生成 3 个不同角度的检索查询，每行一个：\n"
    "问题：{question}"
)

# 生成多个查询
chain = multi_query_prompt | llm | StrOutputParser()
queries = chain.invoke({"question": "RAG 的优势是什么？"}).strip().split("\n")

# 用所有查询进行检索，合并去重
all_docs = []
seen = set()
for query in queries:
    for doc in retriever.invoke(query):
        if doc.page_content not in seen:
            seen.add(doc.page_content)
            all_docs.append(doc)

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
compressor = LLMChainExtractor.from_llm(llm)

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

# 返回的文档只包含与问题相关的内容
compressed_docs = compression_retriever.invoke("RAG 的核心组件有哪些？")

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings

# 假设已有 vectorstore
retriever = vectorstore.as_retriever()


@tool
def search_knowledge_base(query: str) -> str:
    """搜索知识库，获取与查询相关的文档内容。当需要查找内部知识时使用此工具。"""
    docs = retriever.invoke(query)
    return "\n\n".join(doc.page_content for doc in docs)


# 创建带检索工具的 Agent
agent = create_agent(
    model="gpt-4o-mini",
    tools=[search_knowledge_base],
    system_prompt=(
        "你是一个专业的问答助手。你可以使用 search_knowledge_base 工具"
        "来搜索知识库获取信息。请基于检索到的信息回答问题。"
    ),
)

# Agent 会自主决定是否需要检索
response = agent.invoke({
    "messages": [("human", "RAG 和微调有什么区别？")]
})