rag-infrastructure

RAG インフラストラクチャ

検索拡張生成（RAG）の本番環境向けインフラストラクチャです。ドキュメントを取り込み、埋め込みを生成してベクトルデータベースに保存し、根拠に基づく LLM の応答を提供します。

このスキルを使用する場面

次のような場合にこのスキルを使用します。

• 社内ドキュメントを対象にしたナレッジベースの Q&A システムを構築する
• 大規模なドキュメントコレクションを対象にしたセマンティック検索を実装する
• 検索したコンテキストを使って LLM のハルシネーションを減らす
• 埋め込みパイプラインとベクトルストアのインフラストラクチャをセットアップする
• ハイブリッド検索（密ベクトル＋疎ベクトル／BM25）をデプロイする

前提条件

• pip を使用できる Python 3.10 以降
• ベクトルデータベース（Qdrant、Weaviate、Pinecone、または pgvector）
• 埋め込みモデル（OpenAI、Cohere、または sentence-transformers を使ったローカルモデル）
• LLM エンドポイント（OpenAI API またはセルフホストの vLLM）
• ローカルでベクトル DB をデプロイするための Docker

アーキテクチャの概要

Documents → Chunker → Embedder → Vector Store
                                      ↓
User Query → Embedder → Vector Store (search) → Reranker → LLM → Answer

埋め込みパイプライン

from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid

# Local embedding model (no API cost)
model = SentenceTransformer("BAAI/bge-large-en-v1.5")

# Connect to Qdrant
client = QdrantClient("http://localhost:6333")

# Create collection
client.create_collection(
    collection_name="knowledge-base",
    vectors_config=VectorParams(size=1024, distance=Distance.COSINE),
)

def ingest_documents(docs: list[dict]):
    """Chunk, embed, and upsert documents."""
    points = []
    for doc in docs:
        chunks = chunk_text(doc["text"], chunk_size=512, overlap=50)
        embeddings = model.encode(chunks, batch_size=32, show_progress_bar=True)
        for chunk, embedding in zip(chunks, embeddings):
            points.append(PointStruct(
                id=str(uuid.uuid4()),
                vector=embedding.tolist(),
                payload={"text": chunk, "source": doc["source"], "title": doc["title"]},
            ))
    client.upsert(collection_name="knowledge-base", points=points)
    print(f"Ingested {len(points)} chunks")

チャンク分割の戦略

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_text(text: str, chunk_size: int = 512, overlap: int = 50) -> list[str]:
    """Recursive character splitter — best general-purpose strategy."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    return splitter.split_text(text)

# For code/markdown — use language-aware splitter
from langchain.text_splitter import MarkdownHeaderTextSplitter

headers = [("#", "H1"), ("##", "H2"), ("###", "H3")]
md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers)

ハイブリッド検索（密ベクトル＋疎ベクトル）

from qdrant_client.models import SparseVector, SparseVectorParams, NamedSparseVector
from fastembed import SparseTextEmbedding

# Qdrant hybrid collection (dense + BM25 sparse)
client.create_collection(
    collection_name="hybrid-kb",
    vectors_config={"dense": VectorParams(size=1024, distance=Distance.COSINE)},
    sparse_vectors_config={"sparse": SparseVectorParams()},
)

sparse_model = SparseTextEmbedding("prithivida/Splade_PP_en_v1")

def hybrid_search(query: str, top_k: int = 10) -> list[dict]:
    dense_vec = model.encode(query).tolist()
    sparse_vec = list(sparse_model.embed(query))[0]

    results = client.query_points(
        collection_name="hybrid-kb",
        prefetch=[
            {"query": dense_vec, "using": "dense", "limit": 20},
            {"query": SparseVector(indices=sparse_vec.indices.tolist(),
                                   values=sparse_vec.values.tolist()),
             "using": "sparse", "limit": 20},
        ],
        query={"fusion": "rrf"},   # Reciprocal Rank Fusion
        limit=top_k,
    )
    return [{"text": p.payload["text"], "score": p.score} for p in results.points]

リランキング

import cohere

co = cohere.Client("your-api-key")

def rerank(query: str, candidates: list[str], top_n: int = 5) -> list[str]:
    """Rerank retrieved chunks for relevance (improves RAG quality ~20-30%)."""
    response = co.rerank(
        model="rerank-english-v3.0",
        query=query,
        documents=candidates,
        top_n=top_n,
    )
    return [candidates[r.index] for r in response.results]

# Alternative: local reranker (no API cost)
from sentence_transformers import CrossEncoder
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def local_rerank(query: str, candidates: list[str], top_n: int = 5) -> list[str]:
    pairs = [[query, c] for c in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [text for text, _ in ranked[:top_n]]

RAG クエリパイプライン

from openai import OpenAI

llm = OpenAI(base_url="http://localhost:8000/v1", api_key="your-key")

def rag_query(user_question: str) -> str:
    # 1. Retrieve
    candidates = hybrid_search(user_question, top_k=20)
    texts = [c["text"] for c in candidates]

    # 2. Rerank
    top_chunks = local_rerank(user_question, texts, top_n=5)

    # 3. Generate
    context = "\n\n---\n\n".join(top_chunks)
    response = llm.chat.completions.create(
        model="meta-llama/Llama-3.1-8B-Instruct",
        messages=[
            {"role": "system", "content": (
                "Answer the question using only the provided context. "
                "If the answer isn't in the context, say so.\n\nContext:\n" + context
            )},
            {"role": "user", "content": user_question},
        ],
        temperature=0.1,
        max_tokens=1024,
    )
    return response.choices[0].message.content

Docker Compose：RAG スタック全体

services:
  qdrant:
    image: qdrant/qdrant:latest
    volumes:
      - qdrant-data:/qdrant/storage
    ports:
      - "6333:6333"
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    volumes:
      - redis-data:/data
    restart: unless-stopped

  ingestion-worker:
    build: ./ingestion
    environment:
      - QDRANT_URL=http://qdrant:6333
      - REDIS_URL=redis://redis:6379
    depends_on: [qdrant, redis]
    restart: unless-stopped

  rag-api:
    build: ./api
    ports:
      - "8080:8080"
    environment:
      - QDRANT_URL=http://qdrant:6333
      - LLM_BASE_URL=http://vllm:8000/v1
    depends_on: [qdrant]
    restart: unless-stopped

volumes:
  qdrant-data:
  redis-data:

よくある問題

問題	原因	対処法
検索品質が低い	チャンクサイズが大きすぎる	256～512 トークンを試し、オーバーラップを 10～15% にします
LLM が検索したコンテキストを無視する	コンテキストが長すぎる	リランキングを行い、上位 3～5 個のチャンクに絞ります
取り込みが遅い	埋め込みを逐次処理している	batch_size=64 と非同期アップサートを使用します
ドキュメントが古い	再取り込みパイプラインがない	doc_hash を記録し、変更時に再埋め込みします
埋め込みコストが高い	すべてのチャンクを再埋め込みしている	ハッシュベースの重複排除を使って埋め込みをキャッシュします

ベストプラクティス

• 高品質な無料の埋め込みには、BAAI/bge-large-en-v1.5 または nomic-embed-text を使用します。
• LLM に渡す前に必ずリランキングを行います。ノイズの多い 20 個より、精度の高い 5 個のチャンクが効果的です。
• 引用に使えるよう、ソースのメタデータ（URL、ページ、セクション）をベクトルのペイロードに保存します。
• マルチテナント RAG では、ベクトルストアで名前空間／テナントの分離を使用します。
• RAGAS の指標（忠実性、回答の関連性、コンテキストの適合率）を使って評価します。

関連スキル

• vector-database-ops - Qdrant／Weaviate の管理
• vllm-server - セルフホストの LLM エンドポイント
• ollama-stack - 開発用のローカル LLM
• ai-pipeline-orchestration - 取り込みパイプライン