构建高可用 RAG 大模型系统：从基础检索到上下文增强的实战指南

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", " ", ""],
    length_function=len,
)
chunks = splitter.split_documents(docs)

for chunk in chunks:
    chunk.metadata["source"] = file_path
    chunk.metadata["section"] = extract_section(chunk.page_content)

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-large-zh-v1.5")
embeddings = model.encode(chunk_text, normalize_embeddings=True)  # L2归一化

# docker-compose.yml
services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
    volumes:
      - ./qdrant_storage:/qdrant/storage

from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance

client = QdrantClient(host="localhost", port=6333)
client.recreate_collection(
    collection_name="knowledge_base",
    vectors_config=VectorParams(
        size=1024,  # 与 embedding 维度一致
        distance=Distance.COSINE,
    ),
)

from qdrant_client.models import Filter, SearchRequest

# 向量检索
vector_results = client.search(
    collection_name="knowledge_base",
    query_vector=query_embedding,
    limit=10,
)

# 关键词检索（需 Qdrant 开启全文索引）
keyword_results = client.search(
    collection_name="knowledge_base",
    query_filter=Filter(must=[...]),
    query="用户输入关键词",
    limit=10,
    with_payload=True,
)

# RRF 融合
def reciprocal_rank_fusion(results_list, k=60):
    scores = {}
    for results in results_list:
        for rank, hit in enumerate(results):
            id = hit.id
            score = 1.0 / (k + rank + 1)
            scores[id] = scores.get(id, 0) + score
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

from transformers import AutoModelForSequenceClassification, AutoTokenizer
import torch

tokenizer = AutoTokenizer.from_pretrained("BAAI/bge-reranker-v2-m3")
model = AutoModelForSequenceClassification.from_pretrained("BAAI/bge-reranker-v2-m3")

def rerank(query, passages, top_k=5):
    pairs = [[query, p] for p in passages]
    inputs = tokenizer(pairs, padding=True, truncation=True, return_tensors="pt", max_length=512)
    with torch.no_grad():
        scores = model(**inputs).logits.squeeze().tolist()
    scored = sorted(zip(passages, scores), key=lambda x: x[1], reverse=True)
    return [p for p, s in scored[:top_k]]

def build_context(docs, max_tokens=4000):
    context_parts = []
    total_tokens = 0
    for idx, (content, score) in enumerate(docs):
        prefix = f"[{idx+1}] {content[:100]}... (来源: {metadata.get('source')})\n"
        part = f"{prefix}{content}\n"
        tokens = len(part) // 3   # 粗略估算
        if total_tokens + tokens > max_tokens:
            break
        context_parts.append(part)
        total_tokens += tokens
    return "\n".join(context_parts), total_tokens

SYSTEM_PROMPT = """你是一个基于知识库的问答助手。你必须严格按照以下规则回答：
1. 只使用下方提供的上下文信息，不要添加任何外部知识。
2. 如果上下文不足以回答问题，请明确回答“根据当前知识库，无法回答此问题”。
3. 在回答末尾，用[1][2]等标注所引用的文档编号。
4. 回答应简洁、条理清晰，不超过200字。
"""

USER_PROMPT_TEMPLATE = """
上下文：
{context}

问题：{question}
回答：
"""

from openai import OpenAI
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def generate_answer(context, question):
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": USER_PROMPT_TEMPLATE.format(context=context, question=question)}
    ]
    response = client.chat.completions.create(
        model="gpt-4-turbo",
        messages=messages,
        temperature=0.1,
        max_tokens=500,
        stream=False
    )
    return response.choices[0].message.content

pip install ragas

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

dataset = {
    "question": [...],
    "answer": [...],
    "contexts": [[...], ...],
    "ground_truth": [...]  # 可选
}

result = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_relevancy])
print(result)

import optuna

def objective(trial):
    chunk_size = trial.suggest_int("chunk_size", 256, 1024, step=128)
    overlap = trial.suggest_int("overlap", 20, 100, step=10)
    weight_vector = trial.suggest_float("weight_vector", 0.3, 0.7)
    # 重新运行索引和评估，返回 faithfulness
    return -faithfulness_score  # 最小化负值

study = optuna.create_study()
study.optimize(objective, n_trials=30)