从零构建高性能 RAG 系统：架构、优化与生产级实践

┌─────────────────────────────────────────────────────────────┐
│                     离线索引管道                            │
│  原始文档 → 文档解析/清洗 → 分块(Chunking) → 向量化 → 索引写入 │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                     在线查询管道                            │
│  用户Query → 查询改写/扩展 → 向量检索+关键词检索 → 重排序    │
│      → 上下文压缩/精选 → 构造Prompt → LLM生成 → 输出       │
└─────────────────────────────────────────────────────────────┘

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,          # 重叠避免边界信息丢失
    separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""],
    length_function=len,
)
chunks = splitter.split_text(long_document)

from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def semantic_chunk(sentences, emb_model, threshold=0.3):
    embs = emb_model.encode(sentences)
    diffs = [cosine_similarity([embs[i]], [embs[i+1]])[0][0] 
             for i in range(len(embs)-1)]
    breakpoints = [i+1 for i, d in enumerate(diffs) if d < threshold]
    # 按断点合并句子
    chunks = []
    start = 0
    for bp in breakpoints:
        chunks.append("".join(sentences[start:bp]))
        start = bp
    chunks.append("".join(sentences[start:]))
    return chunks

{
  "indexes": [{
    "type": "hnsw",
    "options": {
      "m": 32,               // 每层最大连接数，越大召回率↑但内存↑
      "ef_construct": 200,   // 构建时动态候选列表大小
      "ef_search": 150,      // 查询时搜索广度，可运行时调整
      "full_scan_threshold": 10000  // 低于该值直接暴力搜索
    }
  }]
}

def reciprocal_rank_fusion(results_list, k=60):
    """results_list: list of list of (doc_id, score) 按相关性降序"""
    scores = {}
    for rank_list in results_list:
        for rank, (doc_id, _) in enumerate(rank_list, 1):
            scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (rank + k)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

from sentence_transformers import CrossEncoder

model = CrossEncoder('BAAI/bge-reranker-v2-m3', max_length=512)
pairs = [[query, doc_text] for doc_text in top100_docs]
scores = model.predict(pairs)   # 返回相关性分数 (0~1)

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

from llmlingua import PromptCompressor

compressor = PromptCompressor(model_name="microsoft/llmlingua-2-bert-base-multilingual-cased")
compressed_prompt = compressor.compress_prompt(
    messages=[{"role": "user", "content": user_query}],
    context=retrieved_docs,
    rate=0.5,   # 压缩至50%
    use_sentence_level=True,
)

def truncate_context(docs, max_tokens=3000, tokenizer=None):
    sorted_docs = sorted(docs, key=lambda x: x['score'], reverse=True)
    accumulated = []
    total_tokens = 0
    for doc in sorted_docs:
        tokens = tokenizer.encode(doc['text'])
        if total_tokens + len(tokens) > max_tokens:
            # 尝试截断单个文档
            remain = max_tokens - total_tokens
            if remain > 50:
                accumulated.append(tokenizer.decode(tokens[:remain]))
            break
        accumulated.append(doc['text'])
        total_tokens += len(tokens)
    return "\n\n".join(accumulated)

<context>
<doc id="1">...</doc>
<doc id="2">...</doc>
</context>

<instruction>基于以上文档回答用户问题。如果文档中没有相关信息，请明确告知。回答中请引用文档编号。</instruction>

<question>{user_query}</question>

<answer>

def hyde_rewrite(query, llm):
    prompt = f"请根据以下问题，生成一段可能包含答案的虚构文档片段（仅内容，不要回答）：{query}"
    hypo_doc = llm.generate(prompt, max_tokens=200)
    return hypo_doc  # 作为检索向量

def expand_queries(query, llm):
    prompt = f"生成与'{query}'语义相似但表述不同的3个检索查询，每行一个。"
    resp = llm.generate(prompt)
    return resp.strip().split('\n')

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_relevancy

result = evaluate(
    dataset=test_dataset,   # 包含 question, answer, contexts, ground_truth
    metrics=[faithfulness, answer_relevancy, context_relevancy]
)
print(result)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import qdrant_client
from sentence_transformers import SentenceTransformer, CrossEncoder
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

app = FastAPI()

# 初始化组件
embed_model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
reranker = CrossEncoder('BAAI/bge-reranker-v2-m3')
qdrant = qdrant_client.QdrantClient(host='localhost', port=6333)
tokenizer = AutoTokenizer.from_pretrained('Qwen/Qwen2-7B-Instruct')
llm = AutoModelForCausalLM.from_pretrained('Qwen/Qwen2-7B-Instruct', device_map='auto')

class QueryRequest(BaseModel):
    query: str
    top_k: int = 5
    use_rerank: bool = True

@app.post("/rag/query")
def rag_query(req: QueryRequest):
    # 1. 向量检索
    q_vec = embed_model.encode(req.query).tolist()
    search_result = qdrant.search(
        collection_name="docs",
        query_vector=q_vec,
        limit=20,  # 粗排取20
    )
    doc_ids = [hit.id for hit in search_result]
    doc_texts = [hit.payload['text'] for hit in search_result]
    scores = [hit.score for hit in search_result]
    
    # 2. 重排序（可选）
    if req.use_rerank and len(doc_texts) > 1:
        pairs = [[req.query, t] for t in doc_texts]
        rerank_scores = reranker.predict(pairs)
        combined = sorted(zip(doc_texts, rerank_scores), key=lambda x: x[1], reverse=True)
        final_docs = [text for text, _ in combined[:req.top_k]]
    else:
        final_docs = doc_texts[:req.top_k]
    
    # 3. 上下文压缩（截断到2000 token）
    context = "\n\n".join(final_docs)
    tokens = tokenizer.encode(context)
    if len(tokens) > 2000:
        context = tokenizer.decode(tokens[:2000])
    
    # 4. 生成
    prompt = f"<context>{context}</context>\n<question>{req.query}</question>\n<answer>"
    inputs = tokenizer(prompt, return_tensors="pt").to(llm.device)
    outputs = llm.generate(**inputs, max_new_tokens=512, temperature=0.1)
    answer = tokenizer.decode(outputs[0], skip_special_tokens=True)
    
    return {"answer": answer, "sources": final_docs}