rag-architect

RAG アーキテクト

コアワークフロー

1. 要件分析 — 検索のニーズ、レイテンシの制約、精度要件、規模を特定します
2. ベクトルストアの設計 — データベース、スキーマ設計、インデックス戦略、シャーディング方式を選択します
3. チャンク分割戦略 — ドキュメントの分割、オーバーラップ、意味的な境界、メタデータの拡充を決定します
4. 検索パイプライン — 埋め込みの選択、クエリ変換、ハイブリッド検索、再ランキングを行います
5. 評価と反復 — 指標の追跡、検索のデバッグ、継続的な最適化を行います

各ステップでは、次に進む前に検証してください（以下のチェックポイントを参照）。

リファレンスガイド

状況に応じて詳細なガイダンスを読み込みます。

トピック	リファレンス	読み込むタイミング
ベクトルデータベース	references/vector-databases.md	Pinecone、Weaviate、Chroma、pgvector、Qdrant を比較するとき
埋め込みモデル	references/embedding-models.md	埋め込みの選択、ファインチューニング、次元のトレードオフを検討するとき
チャンク分割戦略	references/chunking-strategies.md	ドキュメントの分割、オーバーラップ、意味ベースのチャンク分割を検討するとき
検索の最適化	references/retrieval-optimization.md	ハイブリッド検索、再ランキング、クエリ拡張、フィルタリングを検討するとき
RAG の評価	references/rag-evaluation.md	指標、評価フレームワーク、検索のデバッグを検討するとき

実装例

1. ドキュメントのチャンク分割

from langchain.text_splitter import RecursiveCharacterTextSplitter

# Evaluate chunk_size on your domain data — never use 512 blindly
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=100,
    separators=["\n\n", "\n", ". ", " "],
)

chunks = splitter.create_documents(
    texts=[doc.page_content for doc in raw_docs],
    metadatas=[{"source": doc.metadata["source"], "timestamp": doc.metadata.get("timestamp")} for doc in raw_docs],
)

チェックポイント： assert all(c.metadata.get("source") for c in chunks), "Missing source metadata"

2. 埋め込みの生成とインデックス作成

from openai import OpenAI
import qdrant_client
from qdrant_client.models import VectorParams, Distance, PointStruct

client = OpenAI()
qdrant = qdrant_client.QdrantClient("localhost", port=6333)

# Create collection
qdrant.recreate_collection(
    collection_name="knowledge_base",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

def embed_chunks(chunks: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    response = client.embeddings.create(input=chunks, model=model)
    return [r.embedding for r in response.data]

# Idempotent upsert with deduplication via deterministic IDs
import hashlib, uuid

points = []
for i, chunk in enumerate(chunks):
    doc_id = str(uuid.UUID(hashlib.md5(chunk.page_content.encode()).hexdigest()))
    embedding = embed_chunks([chunk.page_content])[0]
    points.append(PointStruct(id=doc_id, vector=embedding, payload=chunk.metadata))

qdrant.upsert(collection_name="knowledge_base", points=points)

チェックポイント： assert qdrant.count("knowledge_base").count == len(set(p.id for p in points)), "Deduplication failed"

3. ハイブリッド検索（ベクトル + BM25）

from qdrant_client.models import Filter, FieldCondition, MatchValue, SparseVector
from rank_bm25 import BM25Okapi

def hybrid_search(query: str, tenant_id: str, top_k: int = 20) -> list:
    # Dense retrieval
    query_embedding = embed_chunks([query])[0]
    tenant_filter = Filter(must=[FieldCondition(key="tenant_id", match=MatchValue(value=tenant_id))])
    dense_results = qdrant.search(
        collection_name="knowledge_base",
        query_vector=query_embedding,
        query_filter=tenant_filter,
        limit=top_k,
    )

    # Sparse retrieval (BM25)
    corpus = [r.payload.get("text", "") for r in dense_results]
    bm25 = BM25Okapi([doc.split() for doc in corpus])
    bm25_scores = bm25.get_scores(query.split())

    # Reciprocal Rank Fusion
    ranked = sorted(
        zip(dense_results, bm25_scores),
        key=lambda x: 0.6 * x[0].score + 0.4 * x[1],
        reverse=True,
    )
    return [r for r, _ in ranked[:top_k]]

チェックポイント： assert len(hybrid_search("test query", tenant_id="demo")) > 0, "Hybrid search returned no results"

4. 上位 K 件の結果の再ランキング

プロバイダーの API キーは環境変数またはシークレットマネージャーから読み込み、ソースコードにコミットしないでください。

import os

import cohere

co = cohere.Client(os.environ["COHERE_API_KEY"])

def rerank(query: str, results: list, top_n: int = 5) -> list:
    docs = [r.payload.get("text", "") for r in results]
    reranked = co.rerank(query=query, documents=docs, top_n=top_n, model="rerank-english-v3.0")
    return [results[r.index] for r in reranked.results]

5. 検索の評価

# Run precision@k and recall@k against a labeled evaluation set
# python evaluate.py --metrics precision@10 recall@10 mrr --collection knowledge_base

from ragas import evaluate
from ragas.metrics import context_precision, context_recall, faithfulness, answer_relevancy
from datasets import Dataset

eval_dataset = Dataset.from_dict({
    "question": questions,
    "contexts": retrieved_contexts,
    "answer": generated_answers,
    "ground_truth": ground_truth_answers,
})

results = evaluate(eval_dataset, metrics=[context_precision, context_recall, faithfulness, answer_relevancy])
print(results)

チェックポイント： LLM 統合に進む前に、context_precision >= 0.7 および context_recall >= 0.6 を目標にします。

制約

必ず行うこと

• 採用を決定する前に、ドメインデータで複数の埋め込みモデルを評価する
• 本番システムではハイブリッド検索（ベクトル + キーワード）を実装する
• マルチテナントまたはドメイン固有の検索にメタデータフィルターを追加する
• 検索指標（precision@k、recall@k、MRR、NDCG）を測定する
• コンテキストを LLM に渡す前に、上位 K 件の結果を再ランキングする
• 重複排除（決定的な ID）を備えた冪等な取り込みを実装する
• 検索のレイテンシと品質を継続的に監視する
• 埋め込みのバージョンを管理し、モデル移行を計画する

絶対に行わないこと

• ドメインデータで評価せずに、デフォルトのチャンクサイズ（512）を使用する
• メタデータの拡充（ソース、タイムスタンプ、セクション）を省略する
• LLM の出力品質だけを重視し、検索品質の指標を無視する
• 前処理やクリーニングをせずに生のドキュメントを保存する
• 複雑なマルチドメイン検索にコサイン類似度だけを使用する
• 本番環境に近いデータ量でテストせずにデプロイする
• エッジケース（結果が空の場合、不正なドキュメント）への対応を忘れる
• 埋め込みモデルをアプリケーションコードに密結合させる

出力テンプレート

RAG アーキテクチャを設計する際は、以下を提示します。

1. システムアーキテクチャ図（取り込み + 検索パイプライン）
2. ベクトルデータベースの選定とトレードオフ分析
3. 例と根拠を含むチャンク分割戦略
4. 検索パイプラインの設計（クエリ → 結果のフロー）
5. 指標、ベンチマーク、合否基準を含む評価計画

@jeffallan、Synergetic Solutions のプリンシパルコンサルタントが管理しています。

ドキュメント