RAG 检索增强生成完全教程

从原理到生产，全面掌握 RAG 技术栈

目录

1. RAG 核心原理与架构
2. 文档加载器
3. 文档分割策略
4. Embedding 模型选择与使用
5. 向量数据库对比
6. 检索策略
7. 重排序 Reranking 优化
8. 查询改写与 HyDE
9. 上下文压缩与过滤
10. 多模态 RAG
11. GraphRAG 知识图谱增强
12. RAG 评估指标
13. 生产级 RAG 架构设计
14. 错误处理与降级策略
15. 成本优化与缓存
16. 实战项目：企业知识库问答系统

1. RAG 核心原理与架构

1.1 什么是 RAG

RAG（Retrieval-Augmented Generation，检索增强生成）是一种将外部知识检索与大语言模型生成相结合的技术架构。它解决了 LLM 的三大核心问题：

1.2 RAG 核心架构

离线索引阶段：
┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│ 文档加载  │ →  │ 文档分割  │ →  │ Embedding │ →  │ 向量数据库 │
│ Loading  │    │ Splitting│    │ 向量化    │    │ VectorDB │
└──────────┘    └──────────┘    └──────────┘    └──────────┘

在线查询阶段：
┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│ 用户查询  │ →  │ 查询改写  │ →  │ 向量检索  │ →  │ 重排序    │
│ Query    │    │ Rewrite  │    │ Retrieve │    │ Rerank   │
└──────────┘    └──────────┘    └──────────┘    └────┬─────┘
                                                      │
┌──────────┐    ┌──────────┐                          │
│ 回答输出  │ ←  │ LLM 生成  │ ← 上下文拼接 ←──────────┘
│ Output   │    │ Generate │
└──────────┘    └──────────┘

1.3 快速入门示例

"""最简 RAG 系统 - 5分钟上手"""
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 1. 加载文档
loader = TextLoader("knowledge.txt", encoding="utf-8")
docs = loader.load()

# 2. 分割
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(docs)

# 3. 向量化存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 4. 构建 RAG 链
prompt = ChatPromptTemplate.from_template("""
基于以下参考资料回答问题。如果资料中没有相关信息，请明确说明。

参考资料：
{context}

问题：{question}
""")

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

def format_docs(docs):
    return "\n---\n".join(f"[来源{i+1}] {doc.page_content}" for i, doc in enumerate(docs))

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

answer = rag_chain.invoke("公司的年假政策是什么？")
print(answer)

2. 文档加载器

2.1 PDF 文档加载

from langchain_community.document_loaders import PyPDFLoader, PyMuPDFLoader

# 基础 PDF 加载（适合简单 PDF）
loader = PyPDFLoader("report.pdf")
pages = loader.load()  # 每页一个 Document

# PyMuPDF 加载（更快，支持扫描件 OCR）
loader = PyMuPDFLoader("report.pdf")
pages = loader.load()

for page in pages:
    print(f"第 {page.metadata['page']} 页：{page.page_content[:100]}...")

2.2 Word 文档加载

from langchain_community.document_loaders import Docx2txtLoader, UnstructuredWordDocumentLoader

# 方式一：docx2txt（轻量）
loader = Docx2txtLoader("document.docx")
docs = loader.load()

# 方式二：Unstructured（功能更强，支持表格和图片描述）
loader = UnstructuredWordDocumentLoader("document.docx", mode="elements")
docs = loader.load()

for doc in docs:
    print(f"类型: {doc.metadata.get('category', 'N/A')}")
    print(f"内容: {doc.page_content[:200]}")

2.3 网页内容加载

from langchain_community.document_loaders import (
    WebBaseLoader,
    AsyncHtmlLoader,
    RecursiveUrlLoader,
)

# 单页面加载
loader = WebBaseLoader("https://example.com/docs/intro")
docs = loader.load()

# 批量加载多个 URL
urls = [
    "https://example.com/docs/api",
    "https://example.com/docs/guide",
    "https://example.com/docs/faq",
]
loader = WebBaseLoader(urls)
docs = loader.load()

# 递归爬取整个文档站
loader = RecursiveUrlLoader(
    "https://docs.example.com",
    max_depth=2,  # 爬取深度
    prevent_outside=True,  # 不跳出域名
)
docs = loader.load()

2.4 数据库数据加载

from langchain_community.document_loaders import SQLDatabaseLoader
from langchain_community.utilities import SQLDatabase

# 连接数据库
db = SQLDatabase.from_uri("sqlite:///company.db")

# 从 SQL 查询结果加载文档
loader = SQLDatabaseLoader(
    query="SELECT id, title, content, created_at FROM articles WHERE status='published'",
    database=db,
    page_content_columns=["title", "content"],  # 作为文档内容
    metadata_columns=["id", "created_at"],        # 作为元数据
)
docs = loader.load()

2.5 自定义文档加载器

from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document
import csv

class CSVKnowledgeLoader(BaseLoader):
    """自定义 CSV 知识库加载器"""

    def __init__(self, file_path: str, content_col: str, metadata_cols: list[str]):
        self.file_path = file_path
        self.content_col = content_col
        self.metadata_cols = metadata_cols

    def lazy_load(self):
        with open(self.file_path, "r", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for i, row in enumerate(reader):
                metadata = {col: row[col] for col in self.metadata_cols if col in row}
                metadata["row_number"] = i
                yield Document(
                    page_content=row[self.content_col],
                    metadata=metadata,
                )

# 使用
loader = CSVKnowledgeLoader("faq.csv", content_col="answer", metadata_cols=["category", "id"])
docs = list(loader.lazy_load())

3. 文档分割策略

3.1 RecursiveCharacterTextSplitter（推荐默认）

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每个 chunk 最大字符数
    chunk_overlap=50,      # 相邻 chunk 重叠字符数
    length_function=len,   # 长度计算函数
    separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""],
    # 分隔符优先级：先按段落，再按句子，再按词，最后按字符
)

chunks = splitter.split_documents(docs)

# 查看分割结果
for i, chunk in enumerate(chunks[:3]):
    print(f"Chunk {i} (长度 {len(chunk.page_content)}):")
    print(chunk.page_content[:100])
    print("---")

3.2 语义分割（Semantic Splitting）

"""基于语义相似度的智能分割"""
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

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

# 语义分割器会自动检测语义边界
splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",  # 使用百分位阈值
    breakpoint_threshold_amount=75,           # 75% 分位数
)

chunks = splitter.split_documents(docs)
# 语义相近的内容会被分到同一个 chunk

3.3 代码分割

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    Language,
)

# 按编程语言的语法结构分割
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1000,
    chunk_overlap=100,
)

# 加载 Python 代码文件
from langchain_community.document_loaders import TextLoader
code_doc = TextLoader("main.py").load()
chunks = python_splitter.split_documents(code_doc)

# 支持的语言：PYTHON, JS, TS, JAVA, CPP, GO, RUST, CSHARP 等

3.4 分割策略选择指南

文档类型              推荐策略                    chunk_size
─────────────────────────────────────────────────────────
普通文本/文章          RecursiveCharacter         500-1000
技术文档/API 文档      RecursiveCharacter         300-500
法律/学术论文          SemanticChunker            800-1500
源代码                Language-specific           1000-2000
对话记录              按轮次分割                   500-800
表格数据              按行/按表分割                视情况而定

4. Embedding 模型选择与使用

4.1 主流 Embedding 模型对比

4.2 OpenAI Embedding

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=1024,  # 支持自定义维度（MRL 技术）
)

# 单条文本向量化
vector = embeddings.embed_query("什么是机器学习？")
print(f"向量维度: {len(vector)}")  # 1024

# 批量向量化（更高效）
texts = ["机器学习", "深度学习", "自然语言处理"]
vectors = embeddings.embed_documents(texts)
print(f"批量向量化 {len(vectors)} 条文本")

4.3 本地 Embedding 模型

from langchain_huggingface import HuggingFaceEmbeddings

# 使用本地 HuggingFace 模型
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",  # 中文小型模型
    model_kwargs={"device": "cpu"},         # 或 "cuda"
    encode_kwargs={"normalize_embeddings": True},  # 归一化
)

vector = embeddings.embed_query("如何使用RAG？")

4.4 Embedding 缓存优化

from langchain_community.embeddings import CacheBackedEmbeddings
from langchain_community.storage import LocalFileStore

# 基础 Embedding
underlying_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 包装缓存层
store = LocalFileStore("./embedding_cache/")
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings=underlying_embeddings,
    document_embedding_cache=store,
    namespace="openai_text-embedding-3-small",
)

# 第一次调用：实际调用 API
vector1 = cached_embeddings.embed_query("测试文本")

# 第二次调用：从缓存读取，零 API 成本
vector2 = cached_embeddings.embed_query("测试文本")

5. 向量数据库对比

5.1 主流向量数据库特性对比

5.2 Chroma（开发首选）

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

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

# 创建向量数据库
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_data",  # 持久化目录
    collection_name="knowledge_base",    # 集合名称
)

# 相似度检索
results = vectorstore.similarity_search("机器学习算法", k=3)

# 带分数的检索（越低越相似）
results_with_scores = vectorstore.similarity_search_with_score("机器学习算法", k=3)
for doc, score in results_with_scores:
    print(f"分数: {score:.4f} | 内容: {doc.page_content[:80]}")

# 带元数据过滤
results = vectorstore.similarity_search(
    "深度学习",
    filter={"source": "textbook.pdf"},  # 只搜索特定来源
    k=5,
)

# 加载已有数据库
vectorstore = Chroma(
    persist_directory="./chroma_data",
    embedding_function=embeddings,
)

5.3 Milvus（大规模生产）

from langchain_community.vectorstores import Milvus
from pymilvus import connections

# 连接 Milvus
connections.connect("default", host="localhost", port="19530")

vectorstore = Milvus.from_documents(
    documents=chunks,
    embedding=embeddings,
    collection_name="enterprise_kb",
    connection_args={"host": "localhost", "port": "19530},
    index_params={"metric_type": "COSINE", "index_type": "HNSW", "M": 16, "efConstruction": 256},
    search_params={"metric_type": "COSINE", "params": {"ef": 100}},
)

5.4 FAISS（高性能本地）

from langchain_community.vectorstores import FAISS

# 创建索引
vectorstore = FAISS.from_documents(chunks, embeddings)

# 保存索引
vectorstore.save_local("./faiss_index")

# 加载索引
vectorstore = FAISS.load_local(
    "./faiss_index",
    embeddings,
    allow_dangerous_deserialization=True,  # FAISS 需要此参数
)

# 添加新文档
vectorstore.add_documents(new_chunks)

# 检索
results = vectorstore.similarity_search("自然语言处理", k=3)

6. 检索策略

6.1 基础相似度检索

# 默认使用余弦相似度
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4},  # 返回 top 4
)

# 也可以使用最大边际相关性（MMR）
retriever_mmr = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.7},
    # fetch_k: 初始候选数量
    # lambda_mult: 相关性与多样性的平衡，1=纯相关，0=纯多样
)

6.2 MMR（最大边际相关性）

"""MMR 在保证相关性的同时增加结果多样性"""

# 直接调用
results = vectorstore.max_marginal_relevance_search(
    "Python 编程",
    k=5,            # 最终返回数量
    fetch_k=20,     # 候选池大小
    lambda_mult=0.5 # 多样性权重
)

# 场景对比：
# similarity_search → 可能返回5个几乎相同的结果
# MMR → 返回5个相关但角度不同的结果

6.3 多查询检索（Multi-Query）

"""自动生成多个查询角度，扩大召回范围"""
from langchain.retrievers import MultiQueryRetriever

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

retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    llm=llm,
    # LLM 会自动将原始问题改写为3-5个不同角度的查询
    # 例如："如何提高代码质量？"
    # → "代码质量管理最佳实践"
    # → "代码审查流程"
    # → "自动化测试策略"
)

results = retriever.invoke("如何提高代码质量？")

6.4 上下文压缩检索

"""只保留与查询相关的部分，去除冗余信息"""
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

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

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
)

results = compression_retriever.invoke("公司的退款政策")
# 每个结果只保留与退款相关的部分

7. 重排序 Reranking 优化

7.1 为什么需要重排序

初始检索（向量相似度）是粗排，可能存在：

• 语义相似但不相关的结果
• 缺少对查询意图的深层理解
• 关键信息被排到后面

重排序（Reranking）是精排，使用交叉编码器对 query-document 对重新打分。

7.2 Cohere Reranker

from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank

# 使用 Cohere 的重排序模型
reranker = CohereRerank(
    model="rerank-v3.5",
    top_n=3,  # 重排后取前3
)

retriever = vectorstore.as_retriever(search_kwargs={"k": 10})  # 先粗排取10个

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

results = compression_retriever.invoke("如何申请退款？")

7.3 本地 Reranker（BGE Reranker）

from langchain_community.cross_encoders import HuggingFaceCrossEncoder
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain.retrievers import ContextualCompressionRetriever

# 使用本地交叉编码器
cross_encoder = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base")
reranker = CrossEncoderReranker(model=cross_encoder, top_n=3)

retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
compression_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=retriever,
)

7.4 混合检索（Hybrid Search）

"""结合向量检索和关键词检索（BM25）"""
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# BM25 关键词检索器
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# 向量检索器
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 混合检索：加权集成
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6],  # BM25 权重 40%，向量权重 60%
)

results = ensemble_retriever.invoke("退款流程是什么？")

8. 查询改写与 HyDE

8.1 查询改写（Query Rewriting）

"""将用户的口语化问题改写为更精确的检索查询"""
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

rewrite_prompt = ChatPromptTemplate.from_template("""
你是一个搜索查询优化器。将用户的问题改写为更适合文档检索的形式。
保持原始意图，但使语言更精确、更适合匹配技术文档。

原始问题：{question}

改写后的查询（只输出改写结果，不要解释）：
""")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
rewriter = rewrite_prompt | llm | StrOutputParser()

# "这玩意儿咋用啊" → "产品使用方法和操作指南"
rewritten = rewriter.invoke({"question": "这玩意儿咋用啊"})

8.2 HyDE（假设性文档嵌入）

"""让 LLM 先生成一个假设性的答案文档，用这个文档去检索"""
from langchain_core.prompts import ChatPromptTemplate

hyde_prompt = ChatPromptTemplate.from_template("""
请根据以下问题，写一段可能出现在技术文档中的回答（约200字）。
不需要准确，只需要语言风格和内容方向接近。

问题：{question}

假设性文档：
""")

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

# 生成假设性文档
hyde_chain = hyde_prompt | llm | StrOutputParser()
hypothetical_doc = hyde_chain.invoke({"question": "如何优化数据库查询性能？"})

# 用假设性文档检索（通常比原始查询效果更好）
results = vectorstore.similarity_search(hypothetical_doc, k=5)

8.3 多步检索（Step-back Prompting）

"""将具体问题抽象为更宽泛的问题，分层检索"""
step_back_prompt = ChatPromptTemplate.from_template("""
你是一个问题抽象专家。将以下具体问题抽象为一个更宽泛、更通用的问题。

具体问题：{question}

抽象问题（只输出问题）：
""")

# 具体："Python 3.12 的 match 语句怎么写？"
# 抽象："Python 的模式匹配语法和使用方法"

step_back_chain = step_back_prompt | llm | StrOutputParser()

# 先检索宽泛问题，再检索具体问题，合并结果
abstract_q = step_back_chain.invoke({"question": "Python 3.12 的 match 语句怎么写？"})
broad_results = vectorstore.similarity_search(abstract_q, k=3)
specific_results = vectorstore.similarity_search("Python 3.12 的 match 语句怎么写？", k=3)

# 合并去重
all_results = broad_results + specific_results

9. 上下文压缩与过滤

9.1 文档级压缩

from langchain.retrievers.document_compressors import LLMChainFilter

# 只保留相关文档（整篇保留或丢弃）
filter_compressor = LLMChainFilter.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=filter_compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10}),
)

9.2 元数据过滤

"""基于文档元数据进行精确过滤"""

# 在检索时过滤
results = vectorstore.similarity_search(
    "人工智能",
    filter={
        "$and": [
            {"source": {"$in": ["textbook.pdf", "paper.pdf"]}},
            {"page": {"$gte": 10}},
        ]
    },
    k=5,
)

# LangChain 支持的过滤操作符：
# $eq, $ne, $gt, $gte, $lt, $lte
# $in, $nin
# $and, $or

9.3 长上下文压缩

"""当检索结果总 token 数超过模型限制时的处理"""
from langchain.retrievers.document_compressors import LongContextReorder

# 重排序：将最相关的内容放在开头和结尾（利用 primacy-recency 效应）
reorder = LongContextReorder()

retriever = vectorstore.as_retriever(search_kwargs={"k": 20})
compression_retriever = ContextualCompressionRetriever(
    base_compressor=reorder,
    base_retriever=retriever,
)

10. 多模态 RAG

10.1 图文混合检索架构

"""处理包含图片、表格等多模态内容的文档"""
from langchain_experimental.open_clip import OpenCLIPEmbeddings
from langchain_community.vectorstores import Chroma
import base64

# 1. 文本 Embedding
text_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
text_vectorstore = Chroma.from_documents(text_chunks, text_embeddings)

# 2. 图片 Embedding（使用 CLIP 模型）
clip_embeddings = OpenCLIPEmbeddings(model_name="ViT-B-32", checkpoint="laion2b_s34b_b79k")
image_vectorstore = Chroma.from_documents(image_docs, clip_embeddings)

# 3. 多模态检索
def multimodal_retrieve(query: str, k: int = 3):
    """同时检索文本和图片，合并结果"""
    text_results = text_vectorstore.similarity_search(query, k=k)
    image_results = image_vectorstore.similarity_search(query, k=k)

    all_results = []
    for doc in text_results:
        all_results.append({"type": "text", "content": doc.page_content, "source": doc.metadata["source"]})
    for doc in image_results:
        all_results.append({"type": "image", "content": doc.page_content, "source": doc.metadata["source"]})

    return all_results

10.2 图片描述增强

"""先用视觉模型描述图片，再将描述文本人库"""
from langchain_openai import ChatOpenAI
import base64

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

def describe_image(image_path: str) -> str:
    """使用视觉模型生成图片描述"""
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode()

    response = vision_llm.invoke([
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "详细描述这张图片的内容，包括图表中的数据、文字和关键信息。"},
                {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_data}"}},
            ],
        }
    ])
    return response.content

# 将图片描述存入向量数据库
for img_path in image_files:
    description = describe_image(img_path)
    doc = Document(
        page_content=description,
        metadata={"type": "image", "source": img_path},
    )
    vectorstore.add_documents([doc])

11. GraphRAG 知识图谱增强

11.1 什么是 GraphRAG

GraphRAG 将知识图谱与传统 RAG 结合，通过实体关系图谱增强检索的推理能力。

传统 RAG:  文档 → 向量 → 相似度检索 → 回答
GraphRAG:  文档 → 实体抽取 → 知识图谱 → 图查询+向量检索 → 回答

11.2 使用 NetworkX 构建简单知识图谱

import networkx as nx
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# 1. 实体关系抽取
extract_prompt = ChatPromptTemplate.from_template("""
从以下文本中抽取实体和关系。以JSON格式输出：
[{{"subject": "实体1", "relation": "关系", "object": "实体2"}}]

文本：{text}
""")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
extractor = extract_prompt | llm | JsonOutputParser()

# 2. 构建知识图谱
G = nx.DiGraph()

for doc in docs:
    triples = extractor.invoke({"text": doc.page_content})
    for triple in triples:
        G.add_edge(
            triple["subject"],
            triple["object"],
            relation=triple["relation"],
            source=doc.page_content,
        )

# 3. 图检索
def graph_retrieve(query: str, hops: int = 2) -> list[str]:
    """基于实体的图遍历检索"""
    # 先从查询中提取实体
    entity_prompt = ChatPromptTemplate.from_template(
        "从以下问题中提取关键实体（只输出实体名，逗号分隔）：\n{query}"
    )
    entities = (entity_prompt | llm | StrOutputParser()).invoke({"query": query})
    entity_list = [e.strip() for e in entities.split(",")]

    context = []
    for entity in entity_list:
        if entity in G:
            # BFS 遍历指定跳数
            for neighbor in nx.single_source_shortest_path(G, entity, cutoff=hops):
                for _, target, data in G.edges(neighbor, data=True):
                    context.append(
                        f"{neighbor} --[{data['relation']}]--> {target}\n"
                        f"来源: {data.get('source', '未知')}"
                    )

    return list(set(context))

11.3 Microsoft GraphRAG 集成

"""使用 Microsoft 的 GraphRAG 框架（概念示例）"""

# 安装：pip install graphrag

# 初始化项目
# graphrag init --root ./graphrag_project

# 索引文档
# graphrag index --root ./graphrag_project

# 查询（局部搜索：适合具体问题）
# graphrag query --root ./graphrag_project --method local "什么是RAG？"

# 查询（全局搜索：适合总结性问题）
# graphrag query --root ./graphrag_project --method global "总结公司的主要业务"

12. RAG 评估指标

12.1 RAGAS 评估框架

"""使用 RAGAS 评估 RAG 系统质量"""
from ragas import evaluate
from ragas.metrics import (
    faithfulness,        # 忠实度：答案是否基于检索到的上下文
    answer_relevancy,    # 答案相关性：答案是否回答了问题
    context_precision,   # 上下文精确度：检索结果是否精确相关
    context_recall,      # 上下文召回率：是否检索到了所有相关信息
)
from datasets import Dataset

# 准备评估数据
eval_data = {
    "question": [
        "公司的退换货政策是什么？",
        "如何申请退款？",
        "保修期是多久？",
    ],
    "answer": [
        "7天无理由退换，15天质量问题换货。",
        "请在APP内提交退款申请，3-5个工作日处理。",
        "所有产品享受1年质保。",
    ],
    "contexts": [
        ["退换货政策：7天无理由退换，15天质量问题换货。请保留原始包装。"],
        ["退款流程：在APP内提交申请，审核通过后3-5个工作日退款。"],
        ["保修条款：所有产品享受1年质保，Apple Care+可延长至2年。"],
    ],
    "ground_truth": [
        "7天无理由退换货，15天内质量问题可换货。",
        "通过APP提交退款申请，3-5个工作日处理完成。",
        "标准保修期为1年，可购买延长保修至2年。",
    ],
}

dataset = Dataset.from_dict(eval_data)

# 运行评估
result = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(result)
# {'faithfulness': 0.95, 'answer_relevancy': 0.88, 'context_precision': 0.90, 'context_recall': 0.85}

12.2 自定义评估指标

"""自定义 RAG 评估指标"""
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

def evaluate_answer_quality(question: str, answer: str, context: str) -> dict:
    """自定义评估：准确性、完整性、引用质量"""
    eval_prompt = ChatPromptTemplate.from_template("""
    请评估以下 RAG 系统的输出质量，给出 1-10 的分数。

    问题：{question}
    回答：{answer}
    参考上下文：{context}

    评估维度（JSON格式）：
    {{
        "accuracy": <准确性分数>,
        "completeness": <完整性分数>,
        "relevance": <相关性分数>,
        "hallucination_risk": <幻觉风险分数，越低越好>,
        "overall": <总体分数>
    }}
    """)

    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    chain = eval_prompt | llm | JsonOutputParser()

    return chain.invoke({
        "question": question,
        "answer": answer,
        "context": context,
    })

13. 生产级 RAG 架构设计

13.1 分层检索架构

"""生产级 RAG：多层检索 + 重排序 + 生成"""
from dataclasses import dataclass

@dataclass
class RAGConfig:
    """RAG 系统配置"""
    # 检索配置
    initial_k: int = 20          # 粗排返回数量
    rerank_top_n: int = 5        # 精排后保留数量
    min_relevance_score: float = 0.3  # 最低相关性阈值

    # 生成配置
    model: str = "gpt-4o-mini"
    temperature: float = 0
    max_tokens: int = 2048

    # 缓存配置
    enable_cache: bool = True
    cache_ttl: int = 3600  # 缓存有效期（秒）

class ProductionRAG:
    """生产级 RAG 系统"""

    def __init__(self, config: RAGConfig):
        self.config = config
        self.llm = ChatOpenAI(model=config.model, temperature=config.temperature)
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.vectorstore = self._init_vectorstore()
        self.retriever = self._build_retriever()
        self.chain = self._build_chain()

    def _init_vectorstore(self):
        return Chroma(
            persist_directory="./production_chroma",
            embedding_function=self.embeddings,
        )

    def _build_retriever(self):
        """构建多层检索器"""
        # 第一层：向量检索
        base_retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": self.config.initial_k}
        )

        # 第二层：重排序
        reranker = CohereRerank(model="rerank-v3.5", top_n=self.config.rerank_top_n)

        return ContextualCompressionRetriever(
            base_compressor=reranker,
            base_retriever=base_retriever,
        )

    def _build_chain(self):
        prompt = ChatPromptTemplate.from_template("""
        你是企业知识库助手。基于参考资料回答用户问题。

        规则：
        1. 只基于参考资料回答，不要编造信息
        2. 如果资料不足，明确说明"根据现有资料无法确定"
        3. 引用来源编号

        参考资料：
        {context}

        问题：{question}
        """)

        return (
            {"context": self.retriever | self._format_docs, "question": RunnablePassthrough()}
            | prompt
            | self.llm
            | StrOutputParser()
        )

    def _format_docs(self, docs):
        return "\n---\n".join(
            f"[来源{i+1}] {doc.page_content}" for i, doc in enumerate(docs)
        )

    def query(self, question: str) -> dict:
        """查询入口，返回答案和元信息"""
        docs = self.retriever.invoke(question)
        answer = self.chain.invoke(question)

        return {
            "answer": answer,
            "sources": [{"content": d.page_content, "metadata": d.metadata} for d in docs],
            "num_sources": len(docs),
        }

13.2 增量索引更新

import hashlib
import json
from pathlib import Path

class IncrementalIndexer:
    """增量文档索引管理"""

    def __init__(self, vectorstore, state_file="./index_state.json"):
        self.vectorstore = vectorstore
        self.state_file = Path(state_file)
        self.state = self._load_state()

    def _load_state(self) -> dict:
        if self.state_file.exists():
            return json.loads(self.state_file.read_text())
        return {"indexed_files": {}}

    def _save_state(self):
        self.state_file.write_text(json.dumps(self.state, ensure_ascii=False, indent=2))

    def _file_hash(self, file_path: str) -> str:
        with open(file_path, "rb") as f:
            return hashlib.md5(f.read()).hexdigest()

    def index_directory(self, directory: str, glob_pattern: str = "**/*.pdf"):
        """增量索引目录中的文档"""
        from langchain_community.document_loaders import PyPDFLoader
        from langchain.text_splitter import RecursiveCharacterTextSplitter

        splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
        new_count = 0
        updated_count = 0

        for file_path in Path(directory).glob(glob_pattern):
            path_str = str(file_path)
            current_hash = self._file_hash(path_str)

            if path_str in self.state["indexed_files"]:
                if self.state["indexed_files"][path_str] == current_hash:
                    continue  # 文件未变化，跳过
                # 文件已更新，需要先删除旧数据
                self._remove_file_docs(path_str)
                updated_count += 1
            else:
                new_count += 1

            # 索引新文件
            loader = PyPDFLoader(path_str)
            docs = loader.load()
            chunks = splitter.split_documents(docs)
            self.vectorstore.add_documents(chunks)

            self.state["indexed_files"][path_str] = current_hash

        self._save_state()
        return {"new": new_count, "updated": updated_count}

    def _remove_file_docs(self, file_path: str):
        """删除特定文件的所有文档"""
        # Chroma 实现
        collection = self.vectorstore._collection
        collection.delete(where={"source": file_path})

14. 错误处理与降级策略

14.1 多层降级机制

import logging
from typing import Optional

logger = logging.getLogger(__name__)

class ResilientRAG:
    """带降级策略的 RAG 系统"""

    def __init__(self, primary_rag, fallback_llm):
        self.primary_rag = primary_rag
        self.fallback_llm = fallback_llm

    def query(self, question: str) -> dict:
        # 策略1：完整 RAG 流程
        try:
            result = self.primary_rag.query(question)
            if result["num_sources"] > 0:
                return {"answer": result["answer"], "mode": "rag", "sources": result["sources"]}
        except Exception as e:
            logger.warning(f"RAG 查询失败: {e}")

        # 策略2：仅使用 LLM（无检索）
        try:
            answer = self.fallback_llm.invoke(f"请回答以下问题（注意：没有参考资料）：\n{question}")
            return {"answer": answer.content, "mode": "llm_only", "sources": []}
        except Exception as e:
            logger.warning(f"LLM 调用失败: {e}")

        # 策略3：返回预设回复
        return {
            "answer": "抱歉，系统暂时无法回答您的问题。请联系人工客服获取帮助。",
            "mode": "fallback",
            "sources": [],
        }

14.2 重试与超时

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class RobustEmbedder:
    """带重试机制的 Embedding 服务"""

    def __init__(self, embeddings, max_retries=3):
        self.embeddings = embeddings
        self.max_retries = max_retries

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        retry=retry_if_exception_type((TimeoutError, ConnectionError)),
    )
    def embed_query(self, text: str) -> list[float]:
        return self.embeddings.embed_query(text)

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
    )
    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        # 分批处理，避免单次请求过大
        batch_size = 100
        all_embeddings = []
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            embeddings = self.embeddings.embed_documents(batch)
            all_embeddings.extend(embeddings)
        return all_embeddings

15. 成本优化与缓存

15.1 多级缓存策略

import hashlib
import json
from datetime import datetime, timedelta

class RAGCache:
    """RAG 系统多级缓存"""

    def __init__(self, redis_client=None):
        self.local_cache = {}  # L1：内存缓存
        self.redis = redis_client  # L2：Redis 缓存
        self.ttl = timedelta(hours=1)

    def _make_key(self, question: str, filters: dict = None) -> str:
        """生成缓存键"""
        content = question
        if filters:
            content += json.dumps(filters, sort_keys=True)
        return hashlib.md5(content.encode()).hexdigest()

    def get(self, question: str, filters: dict = None) -> Optional[dict]:
        key = self._make_key(question, filters)

        # L1: 内存缓存
        if key in self.local_cache:
            entry = self.local_cache[key]
            if datetime.now() - entry["time"] < self.ttl:
                return entry["data"]
            del self.local_cache[key]

        # L2: Redis 缓存
        if self.redis:
            cached = self.redis.get(f"rag:{key}")
            if cached:
                data = json.loads(cached)
                self.local_cache[key] = {"data": data, "time": datetime.now()}
                return data

        return None

    def set(self, question: str, answer: dict, filters: dict = None):
        key = self._make_key(question, filters)
        self.local_cache[key] = {"data": answer, "time": datetime.now()}

        if self.redis:
            self.redis.setex(
                f"rag:{key}",
                int(self.ttl.total_seconds()),
                json.dumps(answer, ensure_ascii=False),
            )

# 使用示例
cache = RAGCache()

def cached_rag_query(question: str, rag_system) -> dict:
    # 先查缓存
    cached = cache.get(question)
    if cached:
        cached["from_cache"] = True
        return cached

    # 缓存未命中，执行 RAG
    result = rag_system.query(question)
    cache.set(question, result)
    result["from_cache"] = False
    return result

15.2 Embedding 成本优化

"""减少 Embedding API 调用次数的策略"""

# 1. 批量处理（减少 API 调用次数）
def batch_embed(texts: list[str], batch_size: int = 100):
    """批量向量化，减少 API 调用"""
    results = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        vectors = embeddings.embed_documents(batch)
        results.extend(vectors)
    return results

# 2. 去重处理（避免重复计算）
def deduplicate_and_embed(texts: list[str]):
    """去重后再向量化"""
    unique_texts = list(set(texts))
    unique_vectors = embeddings.embed_documents(unique_texts)
    text_to_vector = dict(zip(unique_texts, unique_vectors))
    return [text_to_vector[t] for t in texts]

# 3. 使用小模型快速筛选
# 先用小模型粗筛，再用大模型精排（对少量结果）

16. 实战项目：企业知识库问答系统

16.1 系统架构

┌──────────────────────────────────────────────────────────────┐
│                    企业知识库问答系统                           │
├──────────┬──────────┬──────────┬──────────┬──────────────────┤
│ 文档管理  │ 索引引擎  │ 检索引擎  │ 生成引擎  │   评估与监控     │
│          │          │          │          │                  │
│ PDF/Word │ 增量索引  │ 混合检索  │ Prompt   │ RAGAS 评估       │
│ Web/DB   │ 定时更新  │ 多路召回  │ 流式输出  │ 成本监控         │
│ API导入   │ 版本管理  │ 重排序   │ 引用标注  │ 用户反馈         │
└──────────┴──────────┴──────────┴──────────┴──────────────────┘

16.2 完整实现

"""
企业知识库问答系统 - 生产级实现
支持：多格式文档导入、混合检索、流式输出、引用溯源、用户反馈
"""
from typing import TypedDict, Annotated, Optional
from dataclasses import dataclass, field
from pathlib import Path
import hashlib
import json
import logging
import time

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
    PyPDFLoader, Docx2txtLoader, WebBaseLoader, TextLoader
)
from langchain.retrievers import EnsembleRetriever, ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.retrievers import BM25Retriever
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_core.documents import Document

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# ==================== 配置 ====================

@dataclass
class KnowledgeBaseConfig:
    """系统配置"""
    # 模型配置
    llm_model: str = "gpt-4o-mini"
    embedding_model: str = "text-embedding-3-small"
    reranker_model: str = "BAAI/bge-reranker-base"

    # 分割配置
    chunk_size: int = 500
    chunk_overlap: int = 50

    # 检索配置
    retrieval_k: int = 20        # 粗排数量
    rerank_top_n: int = 5        # 精排数量
    bm25_weight: float = 0.3     # BM25 权重
    vector_weight: float = 0.7   # 向量检索权重

    # 存储配置
    chroma_dir: str = "./knowledge_chroma"
    state_file: str = "./kb_state.json"

# ==================== 文档处理 ====================

class DocumentProcessor:
    """文档处理器"""

    LOADER_MAP = {
        ".pdf": PyPDFLoader,
        ".docx": Docx2txtLoader,
        ".txt": TextLoader,
        ".md": TextLoader,
    }

    def __init__(self, config: KnowledgeBaseConfig):
        self.config = config
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=config.chunk_size,
            chunk_overlap=config.chunk_overlap,
            separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " "],
        )

    def load_file(self, file_path: str) -> list[Document]:
        """加载单个文件"""
        path = Path(file_path)
        suffix = path.suffix.lower()

        if suffix not in self.LOADER_MAP:
            raise ValueError(f"不支持的文件格式: {suffix}")

        loader_cls = self.LOADER_MAP[suffix]
        loader = loader_cls(str(path))

        if suffix == ".txt" or suffix == ".md":
            docs = loader.load()
        else:
            docs = loader.load()

        # 添加元数据
        for doc in docs:
            doc.metadata["source"] = str(path)
            doc.metadata["file_type"] = suffix
            doc.metadata["file_name"] = path.name

        return docs

    def load_directory(self, directory: str, glob: str = "**/*") -> list[Document]:
        """加载目录中所有支持的文件"""
        all_docs = []
        dir_path = Path(directory)

        for file_path in dir_path.glob(glob):
            if file_path.suffix.lower() in self.LOADER_MAP:
                try:
                    docs = self.load_file(str(file_path))
                    all_docs.extend(docs)
                    logger.info(f"已加载: {file_path.name} ({len(docs)} 页)")
                except Exception as e:
                    logger.warning(f"加载失败 {file_path.name}: {e}")

        return all_docs

    def load_web(self, urls: list[str]) -> list[Document]:
        """加载网页内容"""
        loader = WebBaseLoader(urls)
        docs = loader.load()
        for doc in docs:
            doc.metadata["file_type"] = "web"
        return docs

    def split_documents(self, docs: list[Document]) -> list[Document]:
        """分割文档"""
        chunks = self.splitter.split_documents(docs)
        logger.info(f"文档分割完成: {len(docs)} 文档 → {len(chunks)} 片段")
        return chunks

# ==================== 检索引擎 ====================

class HybridRetriever:
    """混合检索引擎"""

    def __init__(self, vectorstore, config: KnowledgeBaseConfig):
        self.config = config
        self.vectorstore = vectorstore
        self._build_retriever()

    def _build_retriever(self):
        """构建混合检索器"""
        # 向量检索
        self.vector_retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": self.config.retrieval_k}
        )

        # BM25 检索（需要文档列表）
        # 注意：BM25Retriever 需要在有文档数据后初始化
        self.bm25_retriever = None

        # 重排序
        try:
            cross_encoder = HuggingFaceCrossEncoder(model_name=self.config.reranker_model)
            self.reranker = CrossEncoderReranker(model=cross_encoder, top_n=self.config.rerank_top_n)
        except Exception as e:
            logger.warning(f"重排序模型加载失败: {e}，将跳过重排序")
            self.reranker = None

    def init_bm25(self, documents: list[Document]):
        """初始化 BM25 检索器"""
        self.bm25_retriever = BM25Retriever.from_documents(documents)
        self.bm25_retriever.k = self.config.retrieval_k

    def retrieve(self, query: str) -> list[Document]:
        """混合检索"""
        if self.bm25_retriever:
            # 混合检索
            ensemble = EnsembleRetriever(
                retrievers=[self.bm25_retriever, self.vector_retriever],
                weights=[self.config.bm25_weight, self.config.vector_weight],
            )
            docs = ensemble.invoke(query)
        else:
            docs = self.vector_retriever.invoke(query)

        # 重排序
        if self.reranker and len(docs) > self.config.rerank_top_n:
            docs = self.reranker.compress_documents(docs, query)

        return docs

# ==================== 问答系统 ====================

class KnowledgeBaseQA:
    """企业知识库问答系统"""

    QA_PROMPT = ChatPromptTemplate.from_template("""
你是企业知识库智能助手。请严格基于参考资料回答用户问题。

回答规则：
1. 只使用参考资料中的信息回答，不要编造
2. 如果参考资料不足以回答，请说明"根据现有资料无法确定"
3. 在回答中标注信息来源，格式：[来源N]
4. 使用清晰、专业的中文回答

参考资料：
{context}

用户问题：{question}

回答：
""")

    def __init__(self, config: KnowledgeBaseConfig = None):
        self.config = config or KnowledgeBaseConfig()
        self.llm = ChatOpenAI(model=self.config.llm_model, temperature=0)
        self.embeddings = OpenAIEmbeddings(model=self.config.embedding_model)
        self.processor = DocumentProcessor(self.config)
        self.vectorstore = self._init_vectorstore()
        self.retriever = HybridRetriever(self.vectorstore, self.config)
        self.chain = self._build_chain()

    def _init_vectorstore(self):
        return Chroma(
            persist_directory=self.config.chroma_dir,
            embedding_function=self.embeddings,
        )

    def _build_chain(self):
        def format_context(docs):
            return "\n---\n".join(
                f"[来源{i+1}] {doc.page_content}\n(文件: {doc.metadata.get('file_name', '未知')})"
                for i, doc in enumerate(docs)
            )

        return (
            {
                "context": RunnablePassthrough() | (lambda q: self.retriever.retrieve(q)) | format_context,
                "question": RunnablePassthrough(),
            }
            | self.QA_PROMPT
            | self.llm
            | StrOutputParser()
        )

    def ingest_file(self, file_path: str):
        """导入单个文件"""
        docs = self.processor.load_file(file_path)
        chunks = self.processor.split_documents(docs)
        self.vectorstore.add_documents(chunks)
        logger.info(f"已导入: {file_path} ({len(chunks)} 片段)")

    def ingest_directory(self, directory: str):
        """导入目录"""
        docs = self.processor.load_directory(directory)
        chunks = self.processor.split_documents(docs)

        # 分批写入，避免内存问题
        batch_size = 100
        for i in range(0, len(chunks), batch_size):
            batch = chunks[i:i + batch_size]
            self.vectorstore.add_documents(batch)
            logger.info(f"已写入 {min(i+batch_size, len(chunks))}/{len(chunks)} 片段")

        # 初始化 BM25
        self.retriever.init_bm25(chunks)
        logger.info(f"目录导入完成: {len(chunks)} 片段")

    def query(self, question: str) -> dict:
        """查询入口"""
        start_time = time.time()

        docs = self.retriever.retrieve(question)
        answer = self.chain.invoke(question)

        elapsed = time.time() - start_time

        return {
            "answer": answer,
            "sources": [
                {
                    "content": doc.page_content[:200],
                    "file": doc.metadata.get("file_name", "未知"),
                    "page": doc.metadata.get("page", "N/A"),
                }
                for doc in docs
            ],
            "num_sources": len(docs),
            "elapsed_seconds": round(elapsed, 2),
        }

    def query_stream(self, question: str):
        """流式查询"""
        docs = self.retriever.retrieve(question)

        def format_context(docs):
            return "\n---\n".join(
                f"[来源{i+1}] {doc.page_content}" for i, doc in enumerate(docs)
            )

        chain = self.QA_PROMPT | self.llm | StrOutputParser()

        for chunk in chain.stream({"context": format_context(docs), "question": question}):
            yield chunk

# ==================== FastAPI 服务 ====================

def create_api(kb: KnowledgeBaseQA):
    """创建 API 服务"""
    from fastapi import FastAPI, UploadFile, File
    from fastapi.responses import StreamingResponse
    from pydantic import BaseModel
    import shutil

    app = FastAPI(title="企业知识库问答系统")

    class QueryRequest(BaseModel):
        question: str

    class QueryResponse(BaseModel):
        answer: str
        sources: list[dict]
        num_sources: int
        elapsed_seconds: float

    @app.post("/query", response_model=QueryResponse)
    async def query_endpoint(request: QueryRequest):
        result = kb.query(request.question)
        return QueryResponse(**result)

    @app.post("/query/stream")
    async def query_stream_endpoint(request: QueryRequest):
        def generate():
            for chunk in kb.query_stream(request.question):
                yield f"data: {json.dumps({'content': chunk})}\n\n"
            yield "data: [DONE]\n\n"

        return StreamingResponse(generate(), media_type="text/event-stream")

    @app.post("/upload")
    async def upload_file(file: UploadFile = File(...)):
        upload_dir = Path("./uploads")
        upload_dir.mkdir(exist_ok=True)
        file_path = upload_dir / file.filename

        with open(file_path, "wb") as f:
            shutil.copyfileobj(file.file, f)

        try:
            kb.ingest_file(str(file_path))
            return {"status": "success", "file": file.filename}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    @app.get("/health")
    async def health():
        return {"status": "healthy"}

    return app

# ==================== 主程序 ====================

if __name__ == "__main__":
    import uvicorn

    # 初始化系统
    config = KnowledgeBaseConfig()
    kb = KnowledgeBaseQA(config)

    # 导入知识库文档
    docs_dir = "./knowledge_docs"
    if Path(docs_dir).exists():
        kb.ingest_directory(docs_dir)
        print(f"知识库导入完成")

    # 测试查询
    result = kb.query("公司的年假政策是什么？")
    print(f"回答: {result['answer']}")
    print(f"来源数: {result['num_sources']}")
    print(f"耗时: {result['elapsed_seconds']}s")

    # 启动 API 服务
    app = create_api(kb)
    uvicorn.run(app, host="0.0.0.0", port=8080)

16.3 使用说明

# 1. 安装依赖
pip install langchain langchain-openai langchain-community chromadb
pip install langchain-huggingface faiss-cpu
pip install fastapi uvicorn python-multipart
pip install ragas datasets

# 2. 准备文档
mkdir knowledge_docs
# 将 PDF、Word、TXT 文件放入 knowledge_docs 目录

# 3. 设置环境变量
export OPENAI_API_KEY="your-key"

# 4. 启动服务
python knowledge_base.py

# 5. 测试 API
curl -X POST http://localhost:8080/query \
  -H "Content-Type: application/json" \
  -d '{"question": "公司的年假政策是什么？"}'

# 6. 流式输出
curl -X POST http://localhost:8080/query/stream \
  -H "Content-Type: application/json" \
  -d '{"question": "介绍一下公司的产品线"}'

# 7. 上传文档
curl -X POST http://localhost:8080/upload \
  -F "file=@new_document.pdf"

总结

本教程系统性地覆盖了 RAG 技术栈的完整链路：

进阶方向：

1. Agentic RAG：用 Agent 动态决定检索策略
2. GraphRAG：知识图谱增强复杂推理
3. 多模态 RAG：图文混合检索
4. Self-RAG：模型自主决定何时检索
5. Corrective RAG：检索结果质量自检与纠正

📅 最后更新：2026年5月 | 📝 适用版本：LangChain ≥ 0.3