Communityコーディング＆開発github.com

RAG実装パターン

ベクトルデータベースの選定、チャンク分割戦略、埋め込みモデル、検索最適化を含むRAGシステム実装の包括的なガイドです。RAGシステムの構築、セマンティック検索の実装、検索品質の最適化、RAGパフォーマンスの問題デバッグに使用します。

RAG実装パターン とは？

このスキルは、Retrieval-Augmented Generation (RAG) システムの実装に関する包括的なガイドを提供します。ベクトルデータベースの比較（Qdrant、Pinecone、Chroma、Weaviate、Milvus）、複数のチャンク分割戦略（固定サイズ、セマンティック、階層的、スライディングウィンドウ）、OpenAIやSentence Transformersなどの埋め込みモデルの選択、そしてハイブリッド検索やクエリ拡張などの検索最適化手法をカバーしています。主にQdrantとPineconeのセットアップ例をPythonコードで示し、LangChain、LlamaIndex、Haystackなどのフレームワークもサポートしています。また、コンテキスト管理と引用追跡に関するパターンも含まれており、プロトタイピングから本番環境までのあらゆる段階の開発者にとって有用です。

対応~Claude Code✓Codex CLI~Cursor
npx skills add https://github.com/applied-artificial-intelligence/claude-code-toolkit/tree/main/skills/rag-implementation

Installed? Explore more コーディング＆開発 skills: steipete/bluebubbles, steipete/eightctl, steipete/blucli · View all 6 →

オリジナルを見る→すべてのスキルを見る

お気に入りのAIに質問する

このエージェントスキルを事前に読み込んだ状態で新しいチャットを開きます。

OpenAIChatGPTClaudeClaudeGeminiGeminiGrokGrokPerplexityPerplexityDeepSeekDeepSeek

ドキュメント

RAG実装パターン は何をしますか？

Retrieval-Augmented Generation (RAG) システムの実装に関する包括的なガイドです。ベクトルデータベースの選定、チャンク分割戦略、埋め込みモデル、検索最適化、本番環境へのデプロイパターンを網羅しています。

クイックリファレンス

このスキルを使用するタイミング:

  • RAG/セマンティック検索システムの構築
  • ドキュメント検索パイプラインの実装
  • ベクトルデータベースのパフォーマンス最適化
  • 検索品質の問題デバッグ
  • ベクトルデータベースオプションの選択
  • チャンク分割戦略の設計
  • ハイブリッド検索の実装

カバーするテクノロジー:

  • ベクトルDB: Qdrant、Pinecone、Chroma、Weaviate、Milvus
  • 埋め込み: OpenAI、Sentence Transformers、Cohere
  • フレームワーク: LangChain、LlamaIndex、Haystack

パート1: ベクトルデータベースの選定

データベース比較マトリックス

データベース最適な用途デプロイパフォーマンスコスト
Qdrantセルフホスト、本番環境Docker/K8s優れている (Rust)無料 (セルフホスト)
Pineconeマネージド、迅速なプロトタイピングクラウド優れている従量課金制
Chromaローカル開発、組み込みインプロセス良好 (Python)無料
Weaviate複雑なスキーマ、GraphQLDocker/クラウド優れている (Go)無料 + クラウド
Milvus大規模、分散環境K8s優れている (C++)無料 (セルフホスト)

Qdrant セットアップ（本番環境推奨）

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

# クライアントの初期化（ローカルまたはクラウド）
client = QdrantClient(url="http://localhost:6333")  # またはクラウドURL

# コレクションの作成
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,  # OpenAI text-embedding-3-small の次元数
        distance=Distance.COSINE  # または DOT, EUCLID
    )
)

# ペイロード付きでベクトルを挿入
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],  # 1536次元
            payload={
                "text": "ドキュメントの内容",
                "source": "doc.pdf",
                "page": 1,
                "metadata": {...}
            }
        )
    ]
)

# 検索
results = client.search(
    collection_name="documents",
    query_vector=[0.1, 0.2, ...],
    limit=5,
    score_threshold=0.7  # 最小類似度
)

Pinecone セットアップ（マネージドサービス）

from pinecone import Pinecone, ServerlessSpec

# 初期化
pc = Pinecone(api_key="your-key")

# インデックスの作成
pc.create_index(
    name="documents",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1")
)

# インデックスの取得
index = pc.Index("documents")

# ベクトルのアップサート
index.upsert(vectors=[
    ("doc1", [0.1, 0.2, ...], {"text": "...", "source": "..."})
])

# クエリ
results = index.query(
    vector=[0.1, 0.2, ...],
    top_k=5,
    include_metadata=True
)

パート2: チャンク分割戦略

戦略1: 固定サイズチャンク分割（シンプル、高速）

def fixed_size_chunking(text: str, chunk_size: int = 512, overlap: int = 50) -> list[str]:
    """
    テキストをオーバーラップ付きの固定サイズチャンクに分割します。

    長所: シンプル、チャンクサイズが予測可能
    短所: 文の途中で分割される可能性がある、意味的な境界が不十分
    """
    words = text.split()
    chunks = []

    for i in range(0, len(words), chunk_size - overlap):
        chunk = ' '.join(words[i:i + chunk_size])
        chunks.append(chunk)

    return chunks

# 使用法
chunks = fixed_size_chunking(document, chunk_size=512, overlap=50)

使用するタイミング:

  • シンプルなドキュメント（ログ、トランスクリプト）
  • プロトタイピング/MVP
  • 一貫したトークン予算が必要な場合

戦略2: セマンティックチャンク分割（高品質）

from langchain.text_splitter import RecursiveCharacterTextSplitter

def semantic_chunking(text: str, chunk_size: int = 1000, overlap: int = 200) -> list[str]:
    """
    意味的な境界（段落、文）で分割します。

    長所: 意味を保持、検索品質が向上
    短所: チャンクサイズが可変、処理が遅い
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " ", ""],  # 優先順位
        length_function=len
    )

    return splitter.split_text(text)

# 使用法
chunks = semantic_chunking(document, chunk_size=1000, overlap=200)

使用するタイミング:

  • 長文ドキュメント（記事、書籍、レポート）
  • 品質 > 速度
  • 自然言語コンテンツ

戦略3: 階層的チャンク分割（構造化ドキュメントに最適）

def hierarchical_chunking(document: dict) -> list[dict]:
    """
    ドキュメント構造（セクション、サブセクション）に基づいてチャンク分割します。

    長所: 階層を保持、親子検索が可能
    短所: 構造化された入力が必要、より複雑
    """
    chunks = []

    for section in document['sections']:
        # 親チャンク（セクション要約）
        chunks.append({
            'text': section['title'] + '\n' + section['summary'],
            'type': 'parent',
            'section_id': section['id']
        })

        # 子チャンク（段落）
        for para in section['paragraphs']:
            chunks.append({
                'text': para,
                'type': 'child',
                'parent_id': section['id']
            })

    return chunks

使用するタイミング:

  • 技術文書
  • 目次付きの書籍
  • 法的文書
  • コンテキスト階層を保持する必要がある場合

戦略4: スライディングウィンドウ（最大のコンテキスト保存）

def sliding_window_chunking(text: str, window_size: int = 512, stride: int = 256) -> list[str]:
    """
    最大のコンテキストを得るためのオーバーラップウィンドウ。

    長所: 境界での情報損失がない
    短所: ストレージオーバーヘッド（コンテンツの重複）
    """
    words = text.split()
    chunks = []

    for i in range(0, len(words) - window_size + 1, stride):
        chunk = ' '.join(words[i:i + window_size])
        chunks.append(chunk)

    return chunks

使用するタイミング:

  • 重要な検索精度が必要な場合
  • 短いクエリに広いコンテキストが必要な場合
  • ストレージコストが問題にならない場合

パート3: 埋め込みモデル

モデル選択ガイド

モデル次元数速度品質コストユースケース
OpenAI text-embedding-3-small1536高速優れている$0.02/100万トークン本番環境、汎用
OpenAI text-embedding-3-large3072中程度最高$0.13/100万トークン高品質な検索
all-MiniLM-L6-v2384非常に高速良好無料セルフホスト、プロトタイピング
all-mpnet-base-v2768高速非常に良好無料セルフホスト、品質重視
Cohere embed-english-v3.01024高速優れている$0.10/100万トークンセマンティック検索特化

OpenAI 埋め込み（推奨）

from openai import OpenAI

client = OpenAI(api_key="your-key")

def get_embeddings(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    """
    OpenAI を使用して埋め込みを生成します。

    バッチサイズ: リクエストあたり最大2048入力
    レート制限: 階層制限を確認
    """
    response = client.embeddings.create(
        model=model,
        input=texts
    )

    return [item.embedding for item in response.data]

# 使用法
chunks = ["チャンク1", "チャンク2", ...]
embeddings = get_embeddings(chunks)

Sentence Transformers（セルフホスト）

from sentence_transformers import SentenceTransformer

# モデルのロード（初回ダウンロード後はキャッシュ）
model = SentenceTransformer('all-MiniLM-L6-v2')

def get_embeddings_local(texts: list[str]) -> list[list[float]]:
    """
    ローカルで埋め込みを生成します（APIコストなし）。

    バッチサイズ > 100 の場合はGPU推奨
    小規模バッチではCPUでも許容
    """
    return model.encode(texts, show_progress_bar=True).tolist()

# 使用法
embeddings = get_embeddings_local(chunks)

パート4: 検索最適化

テクニック1: ハイブリッド検索（密 + 疎）

from qdrant_client.models import Filter, FieldCondition, MatchValue

def hybrid_search(query: str, query_vector: list[float], top_k: int = 10):
    """
    密（ベクトル）検索と疎（キーワード）検索を組み合わせます。

    密: セマンティック類似度
    疎: 完全一致キーワード
    """
    # 密検索
    dense_results = client.search(
        collection_name="documents",
        query_vector=query_vector,
        limit=top_k * 2  # より多くの候補を取得
    )

    # 疎検索（メタデータによるBM25）
    sparse_results = client.search(
        collection_name="documents",
        query_filter=Filter(
            must=[
                FieldCondition(
                    key="text",
                    match=MatchValue(value=query)
                )
            ]
        ),
        limit=top_k * 2
    )

    # マージと再ランク付け
    combined = merge_results(dense_results, sparse_results, weights=(0.7, 0.3))
    return combined[:top_k]

テクニック2: クエリ拡張

def expand_query(query: str) -> list[str]:
    """
    より良い再現率を得るためにクエリのバリエーションを生成します。

    テクニック:
    - 同義語展開
    - 質問の言い換え
    - エンティティ抽出
    """
    from openai import OpenAI
    client = OpenAI()

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "ユーザーのクエリの言い換えを3つ生成してください。"
        }, {
            "role": "user",
            "content": query
        }]
    )

    expanded = [query] + response.choices[0].message.content.split('\n')
    return expanded

# 使用法
queries = expand_query("ニューラルネットワークの訓練方法は？")
# → ["ニューラルネットワークの訓練方法は？",
#    "ニューラルネットワークの訓練手法とは？",
#    "ニューラルネットワーク最適化手法",
#    "深層学習モデルの訓練"]

テクニック3: 再ランク付け

from sentence_transformers import CrossEncoder

# クロスエンコーダーのロード（再ランク付けにはバイエンコーダーより優れている）
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank_results(query: str, results: list[dict], top_k: int = 5) -> list[dict]:
    """
    クロスエンコーダーを使用して初期結果を再ランク付けします。

    最初の検索よりも正確だが遅い
    上位20～50件の候補にのみ使用
    """
    # 各クエリ-ドキュメントペアをスコアリング
    pairs = [(query, result['text']) for result in results]
    scores = reranker.predict(pairs)

    # スコアと結果を結合
    for result, score in zip(results, scores):
        result['rerank_score'] = float(score)

    # ソートして上位top_kを返す
    reranked = sorted(results, key=lambda x: x['rerank_score'], reverse=True)
    return reranked[:top_k]

テクニック4: メタデータフィルタリング

def filtered_search(
    query_vector: list[float],
    filters: dict,
    top_k: int = 5
):
    """
    メタデータ（日付、カテゴリ、作成者など）で検索をフィルタリングします。

    プリフィルター: 高速だが結果を見逃す可能性がある
    ポストフィルター: より多くの結果が得られるが低速
    """
    from qdrant_client.models import Filter, FieldCondition, Range

    # フィルター条件を構築
    conditions = []

    if 'date_range' in filters:
        conditions.append(
            FieldCondition(
                key="date",
                range=Range(
                    gte=filters['date_range']['start'],
                    lte=filters['date_range']['end']
                )
            )
        )

    if 'category' in filters:
        conditions.append(
            FieldCondition(
                key="category",
                match=MatchValue(value=filters['category'])
            )
        )

    # フィルター付きで検索
    results = client.search(
        collection_name="documents",
        query_vector=query_vector,
        query_filter=Filter(must=conditions) if conditions else None,
        limit=top_k
    )

    return results

パート5: コンテキスト管理

パターン1: 取得コンテキストの最適化

def optimize_context(query: str, retrieved_docs: list[str], max_tokens: int = 4000) -> str:
    """
    取得したコンテキストをLLMのコンテキストウィンドウに収まるように最適化します。

    戦略:
    1. 関連性に基づく切り詰め
    2. 抽出型要約
    3. 重複除去
    """
    # 関連性でソート
    sorted_docs = sorted(retrieved_docs, key=lambda d: d['score'], reverse=True)

    # トークン予算内でコンテキストを構築
    context_parts = []
    total_tokens = 0

    for doc in sorted_docs:
        doc_tokens = estimate_tokens(doc['text'])

        if total_tokens + doc_tokens <= max_tokens:
            context_parts.append(f"[ソース: {doc['source']}]\n{doc['text']}")
            total_tokens += doc_tokens
        else:
            # 収まるように最後のドキュメントを切り詰める
            remaining = max_tokens - total_tokens
            truncated = truncate_to_tokens(doc['text'], remaining)
            context_parts.append(f"[ソース: {doc['source']}]\n{truncated}")
            break

    return "\n\n---\n\n".join(context_parts)

パターン2: 引用追跡

def generate_with_citations(query: str, context: str, sources: list[dict]) -> dict:
    """
    引用追跡付きで回答を生成します。

    戻り値:
    - answer: 生成されたテキスト
    - citations: 使用されたソースドキュメントのリスト
    """
    from openai import OpenAI
    client = OpenAI()

    # ソースマップの作成
    source_map = {i+1: source for i, source in enumerate(sources)}
    numbered_context = "\n\n".join([
        f"[{i+1}] {source['text']}"
        for i, source in enumerate(sources)
    ])

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "提供されたソースを使用して回答してください。ソースは[1]、[2]などと引用してください。"
        }, {
            "role": "user",
            "content": f"コンテキスト:\n{numbered_context}\n\n質問: {query}"
        }]
    )

    answer = response.choices[0].message.content

    # 実際に使用されたソースを抽出
    used_sources = extract_cited_sources(answer)

    return {
        'answer': answer,
        'citations': [source_map[i] for i in used_sources if i in source_map]
    }

パターン3: 会話履歴管理

class ConversationMemory:
    """
    RAG会話の履歴を管理します。

    制限:
    - トークン予算に基づいて履歴をトリミング
    - 検索コンテキストとして関連する過去のやり取りを保持
    """

    def __init__(self, max_tokens: int = 2000):
        self.history = []
        self.max_tokens = max_tokens

    def add_interaction(self, query: str, answer: str, sources: list[dict]):
        self.history.append({
            'query': query,
            'answer': answer,
            'sources': sources
        })
        self._trim_history()

    def _trim_history(self):
        """
        トークン予算内に収まるように履歴をトリミングします。
        """
        while self._estimate_tokens() > self.max_tokens and len(self.history) > 1:
            self.history.pop(0)

    def _estimate_tokens(self) -> int:
        return sum(
            estimate_tokens(item['query'] + item['answer'])
            for item in self.history
        )

    def get_context(self, max_past_interactions: int = 5) -> str:
        """
        会話履歴をフォーマットされたコンテキストとして取得します。
        """
        recent = self.history[-max_past_interactions:]
        parts = []
        for i, item in enumerate(recent):
            parts.append(f"[過去のやり取り {i+1}]")
            parts.append(f"ユーザー: {item['query']}")
            parts.append(f"アシスタント: {item['answer']}")

        return "\n".join(parts)

パート6: 本番環境へのデプロイメント

推奨アーキテクチャ

[ユーザークエリ]
        ↓
[クエリ拡張]
        ↓
[エンベディング生成]
        ↓
[ハイブリッド検索] ←→ [ベクトルDB]
        ↓
[結果のマージ]
        ↓
[再ランク付け]
        ↓
[コンテキスト最適化]
        ↓
[LLM生成]
        ↓
[回答 + 引用]

パフォーマンスのベストプラクティス

  1. バッチ処理: 頻繁なラウンドトリップを避けるためにAPIコールをバッチ処理する
  2. キャッシング: 頻繁にアクセスされる埋め込みと検索結果
  3. 非同期処理: 複数の検索を並行して実行
  4. 監視: 検索品質メトリクス（MRR、NDCG@K）を追跡
  5. A/Bテスト: 本番環境で異なる戦略を比較

監視設定

def log_search_quality(query: str, results: list, relevant_docs: list[str]):
    """
    検索品質メトリクスを追跡します。

    メトリクス:
    - MRR (Mean Reciprocal Rank)
    - Recall@K
    - Precision@K
    """
    mrr = calculate_mrr(results, relevant_docs)
    recall = calculate_recall_at_k(results, relevant_docs, k=5)

    log_metric("search_mrr", mrr)
    log_metric("search_recall@5", recall)

トラブルシューティングガイド

問題1: 低い検索品質

考えられる原因:

  • 不適切なチャンクサイズ（大きすぎるか小さすぎる）
  • 間違った埋め込みモデル
  • 不十分なメタデータフィルタリング

解決策:

  1. ドキュメントの種類に基づいてチャンク戦略を調整する
  2. 異なる埋め込みモデルを比較する
  3. メタデータフィルターを実装する
  4. より良い結果を得るためにハイブリッド検索を使用する

問題2: 応答時間が遅い

考えられる原因:

  • 最適化されていないベクトルデータベース
  • 非効率なチャンク分割
  • LLMの長いコンテキスト処理

解決策:

  1. ベクトルDBのインデックス設定を最適化する
  2. APIコールをバッチ処理する
  3. 応答をキャッシュする
  4. コンテキストウィンドウサイズを制限する

問題3: 高いレイテンシー

考えられる原因:

  • ネットワークオーバーヘッド
  • レート制限
  • ボトルネックとなっているサービス

解決策:

  1. 可能な限りローカル処理を使用する
  2. 指数バックオフで再試行ロジックを実装する
  3. コンポーネントをプロファイリングしてボトルネックを特定する

付録: ユーティリティ関数

トークン見積もり

def estimate_tokens(text: str) -> int:
    """
    テキストのおおよそのトークン数を推定します。

    英語の場合は1単語あたり約1.3トークン
    正確な見積もりにはtiktokenを使用
    """
    return len(text.split()) * 4 // 3

切り詰め関数

def truncate_to_tokens(text: str, max_tokens: int) -> str:
    """
    テキストを指定されたトークン数に切り詰めます。
    """
    words = text.split()
    target_words = max_tokens * 3 // 4
    return ' '.join(words[:target_words])

結果のマージ

def merge_results(
    dense_results: list,
    sparse_results: list,
    weights: tuple = (0.7, 0.3)
) -> list:
    """
    密検索と疎検索の結果を重み付け付きでマージします。
    """
    from collections import defaultdict

    scores = defaultdict(float)

    for i, result in enumerate(dense_results):
        scores[result.id] += weights[0] * (1.0 / (i + 1))

    for i, result in enumerate(sparse_results):
        scores[result.id] += weights[1] * (1.0 / (i + 1))

    # スコアでソート
    sorted_results = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return [result for result_id, score in sorted_results]

Individual skills in this repo

This repo contains 2 individual skills — each has its own dedicated page.

huggingface-transformers

Hugging Face Transformers best practices including model loading, tokenization, fine-tuning workflows, and inference optimization. Use when working with transformer models, fine-tuning LLMs, implementing NLP tasks, or optimizing transformer inference.

llm-evaluation

LLM evaluation and testing patterns including prompt testing, hallucination detection, benchmark creation, and quality metrics. Use when testing LLM applications, validating prompt quality, implementing systematic evaluation, or measuring LLM performance.

関連スキル

steipete/bluebubbles

Send and manage iMessages via BlueBubbles, including attachments, tapbacks, edits, replies, and groups.

community

steipete/eightctl

Control Eight Sleep pods (status, temperature, alarms, schedules).

community

steipete/blucli

BluOS CLI (blu) for discovery, playback, grouping, and volume.

community

steipete/bear-notes

Create, search, and manage Bear notes via grizzly CLI.

community

steipete/camsnap

Capture frames or clips from RTSP/ONVIF cameras.

community

steipete/gifgrep

Search GIF providers with CLI/TUI, download results, and extract stills/sheets.

community
← More コーディング＆開発 skills