RAG系统搭建：企业知识库最佳实践

用户提问
    ↓
[查询改写] ← 优化用户问题
    ↓
[向量检索] ← 从知识库中找到相关片段
    ↓
[重排序] ← 对检索结果精排
    ↓
[上下文组装] ← 把相关内容塞进 Prompt
    ↓
[大模型生成] ← 基于上下文生成回答
    ↓
返回答案 + 引用来源

# 方案1：按固定长度分块（简单但效果一般）
from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
    chunk_size=500,      # 每块500字符
    chunk_overlap=50,    # 块间重叠50字符
    separator="\n"       # 优先在换行处切分
)

# 方案2：递归分块（推荐）
from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "！", "？", ".", " "]
)

# 方案3：语义分块（效果最好但更慢）
# 使用 embedding 相似度来决定切分点

import chromadb
from sentence_transformers import SentenceTransformer

# 初始化
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection("company_docs")

# 加载 embedding 模型
model = SentenceTransformer("BAAI/bge-large-zh-v1.5")  # 中文推荐

# 添加文档
docs = ["产品A的使用说明...", "售后服务政策..."]
embeddings = model.encode(docs).tolist()

collection.add(
    documents=docs,
    embeddings=embeddings,
    ids=["doc_1", "doc_2"],
    metadatas=[{"source": "产品手册"}, {"source": "售后文档"}]
)

# 检索
query = "产品怎么用？"
query_emb = model.encode([query]).tolist()
results = collection.query(query_embeddings=query_emb, n_results=3)

# Docker 部署 Milvus
docker compose -f milvus-docker-compose.yml up -d

# 或使用 Milvus Lite（嵌入式，适合小规模）
pip install milvus

# 使用 BGE 模型
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-large-zh-v1.5")

# 注意：BGE 模型对 query 需要加前缀
query = "为这个句子生成表示以用于检索相关文档：产品使用方法"
doc_embedding = model.encode(["产品使用方法..."])
query_embedding = model.encode([query])

# 向量检索 + BM25 关键词检索
# 权重可调：alpha=0.7 表示向量占70%，关键词占30%
def hybrid_search(query, alpha=0.7):
    # 向量检索
    vector_results = vector_search(query, top_k=20)
    # BM25 关键词检索
    bm25_results = bm25_search(query, top_k=20)

    # 融合排序 (RRF - Reciprocal Rank Fusion)
    final_results = reciprocal_rank_fusion(
        vector_results, bm25_results, 
        k=60, alpha=alpha
    )
    return final_results[:10]

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-large")

# 对检索结果重排序
pairs = [[query, doc] for doc in retrieved_docs]
scores = reranker.predict(pairs)

# 按分数重排
ranked = sorted(zip(retrieved_docs, scores), key=lambda x: x[1], reverse=True)

RAG_PROMPT = """你是一个专业的AI助手。请基于以下参考资料回答用户的问题。

## 规则
1. 只基于提供的参考资料回答，不要编造信息
2. 如果参考资料中没有相关内容，请明确说"根据现有资料无法回答"
3. 回答时请引用来源，格式为 [来源: 文档名]

## 参考资料
{context}

## 用户问题
{question}

## 回答"""

from langchain.chains import RetrievalQA
from langchain_community.llms import Ollama
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceBgeEmbeddings

# 1. 初始化 Embedding
embeddings = HuggingFaceBgeEmbeddings(
    model_name="BAAI/bge-large-zh-v1.5",
    model_kwargs={'device': 'cuda'},
    encode_kwargs={'normalize_embeddings': True}
)

# 2. 加载向量数据库
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

# 3. 初始化本地大模型
llm = Ollama(model="llama3:8b-instruct-q5_1", temperature=0.3)

# 4. 创建 RAG 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    return_source_documents=True
)

# 5. 提问
result = qa_chain.invoke({"query": "我们的退货政策是什么？"})
print(result["result"])
print("来源:", [doc.metadata["source"] for doc in result["source_documents"]])