本地轻量 RAG 实战：Claude + sentence-transformers

Retrieval-Augmented Generation（RAG）是无需微调即可让 Claude 接入私有数据的最快方式。但完整的 RAG 流水线往往意味着要摆弄笨重的外部向量数据库或昂贵的 embedding API。

本文将使用 Python、sentence-transformers（本地向量嵌入）和 claude-opus-5（生成答案）构建一个轻量、极速的 RAG 系统。无需复杂的云基础设施——在查询模型之前，所有操作都在本地运行。

你需要 Python 3.10+ 和官方 Anthropic SDK，以及一些用于处理本地向量搜索的数据科学库：

pip install anthropic sentence-transformers numpy

确保你的 API key 已设置在环境变量中：

export ANTHROPIC_API_KEY="your-api-key-here"

步骤 1：生成本地 Embedding

我们首先将小的知识库文本片段通过 Hugging Face 的轻量模型转换为向量嵌入，并存储在内存中。

import numpy as np
from sentence_transformers import SentenceTransformer

# 1. Load a fast, lightweight local embedding model
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")

# 2. Define our local knowledge base
documents = [
    "Project Apollo uses Python 3.11 and FastAPI for the backend services.",
    "The deployment pipeline for staging is triggered automatically via GitHub Actions.",
    "Database migrations are managed using Alembic. Never run raw DDL in production.",
    "To reset the local Redis cache, run 'redis-cli flushall' in your terminal."
]

# 3. Generate embeddings for our documents
doc_embeddings = embedding_model.encode(documents)

步骤 2：检索相关上下文并查询 Claude

当用户提问时，我们将查询向量化，计算与文档嵌入的余弦相似度，提取最匹配的结果。然后将这些片段直接传给 claude-opus-5。

记住：当前的 Claude 模型完全依赖 prompt 引导（无 temperature、top_p 或 top_k），并使用 thinking={"type": "adaptive"} 来进行扩展推理。

import anthropic

client = anthropic.Anthropic()

def search_docs(query: str, top_k: int = 2):
    # Embed the incoming query
    query_vector = embedding_model.encode(query)

    # Compute cosine similarity
    similarities = np.dot(doc_embeddings, query_vector) / (
        np.linalg.norm(doc_embeddings, axis=1) * np.linalg.norm(query_vector)
    )

    # Get top_k indices sorted by highest similarity
    top_indices = np.argsort(similarities)[::-1][:top_k]
    return [documents[i] for i in top_indices]

def ask_rag(user_query: str):
    # Retrieve relevant context locally
    retrieved_chunks = search_docs(user_query)
    context = "\n---\n".join(retrieved_chunks)

    # Construct the grounded prompt
    prompt = f"""You are a helpful technical assistant. Answer the user's question using ONLY the provided context. If you don't know the answer based on the context, say so.

Context:
{context}

Question:
{user_query}"""

    # Call Claude using the current SDK and API rules
    response = client.messages.create(
        model="claude-opus-5",
        max_tokens=4000,
        thinking={"type": "adaptive"},
        output_config={"effort": "medium"},
        messages=[{"role": "user", "content": prompt}]
    )

    return response.content

# Test the RAG pipeline
answer = ask_rag("How do we handle database migrations?")
print(answer)

生产环境注意事项

虽然这种内存 NumPy 方案对原型和小代码库来说效果很好，但上规模时需要一些最佳实践：

切换到向量数据库：对于数千或数百万份文档，将 NumPy 数组替换为 Chroma、FAISS 或 pgvector 等持久化向量存储。

分块策略：不要用整句，而是将大型 Markdown 或 PDF 文件拆分为 500 token 的块，块间重叠 50 token，以保留上下文。

模型选型：如果运行高吞吐量的自动化问答流水线，吞吐量至关重要，此时可使用 claude-sonnet-5。

构建高效的 RAG 并不需要庞大的云技术栈。通过将本地 embedding 与 claude-opus-5 相结合，你可以获得一个安全、私密且极速的问答系统，直接运行在应用程序的工作流中。

你用过 Claude 构建过 RAG 流水线吗？欢迎在评论区分享你最喜欢的 embedding 模型或关于新 API 参数的问题！