RAG 实现指南——使用 Claude 和 Gemini 构建生产级检索增强生成

# Recursive character splitter with overlap — the boring choice that wins
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1200,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_text(document)
# 1200 chars ≈ 300-400 tokens — fits 5+ chunks in a Sonnet context window
# with room for system prompt + question + answer

# Hybrid retrieval: vector similarity + BM25 keyword scoring
# Pure vector misses literal keywords (product SKUs, IDs, dates)
def retrieve(question: str, k: int = 5) -> list[dict]:
    q_embed = embed([question])[0]
    vector_hits = vector_db.similarity_search(q_embed, k=k * 2)
    keyword_hits = bm25_search(question, k=k * 2)

    # Reciprocal rank fusion — simple, robust
    scores: dict[str, float] = {}
    for rank, hit in enumerate(vector_hits):
        scores[hit["id"]] = scores.get(hit["id"], 0) + 1 / (rank + 60)
    for rank, hit in enumerate(keyword_hits):
        scores[hit["id"]] = scores.get(hit["id"], 0) + 1 / (rank + 60)

    top_ids = sorted(scores, key=scores.get, reverse=True)[:k]
    return [chunk_by_id[i] for i in top_ids]

# Final RAG prompt structure — three sections, citations enforced
SYSTEM_PROMPT = """You answer based exclusively on the supplied Context.
- Cite the [doc_id] for each factual claim.
- If the context doesn't answer the question, say "I don't have that
  information" — do not extrapolate or use general knowledge.
- Be concise. No throat-clearing."""

def answer(question: str) -> dict:
    chunks = retrieve(question, k=5)
    context = "\n\n---\n\n".join(
        f"[doc:{c['id']}] {c['text']}" for c in chunks
    )
    resp = client.chat.completions.create(
        model="claude-sonnet-4-6",
        messages=[
            {"role": "system", "content": [{
                "type": "text",
                "text": SYSTEM_PROMPT,
                "cache_control": {"type": "ephemeral"},
            }]},
            {"role": "user", "content": f"# Context\n{context}\n\n# Question\n{question}"},
        ],
        max_tokens=600,
    )
    return {
        "text": resp.choices[0].message.content,
        "sources": [c["id"] for c in chunks],
    }