knowledge-base-rag

ナレッジベース RAG

googleadsagent.ai™ による Agent Skills™ の一部です。

説明

ナレッジベース RAG は、ドキュメントの取り込み、インテリジェントなチャンク分割、埋め込みの生成、ベクトルストアへのインデックス作成、セマンティック検索、根拠に基づく応答生成まで、検索拡張生成のパイプライン全体を実装します。エージェントは、出典を引用し、ハルシネーションを減らしながら、プライベートなナレッジベースから質問に回答する RAG システムを構築します。

RAG は、大規模言語モデルの根本的な制約を解決します。大規模言語モデルは、学習のカットオフ以降に作成された情報や、学習時に使用されなかった独自の情報にアクセスできません。ベクトルストアから関連ドキュメントを検索してプロンプトのコンテキストに挿入することで、RAG はモデルの応答を、事実に基づいた最新の組織固有の知識に基づかせます。

RAG システムの品質は、モデルの選択よりもチャンク分割の戦略に左右されます。このスキルには、本番環境で検証済みのチャンク分割方法が組み込まれています。段落のまとまりを保つセマンティック分割、ドキュメント構造（見出し、コードブロック、表）を尊重する再帰的分割、チャンク境界を越えてコンテキストを維持する重複ウィンドウです。最適な検索品質を実現するため、各戦略をドキュメントの種類に合わせて使用します。

使用する場面

• プライベートなドキュメントを対象とする質問応答システムを構築する場合
• ドキュメント、Wiki、PDF から検索可能なナレッジベースを作成する場合
• 検索した事実に LLM の応答を基づかせて、ハルシネーションを減らす場合
• 大規模なドキュメントコレクション全体にわたるセマンティック検索を実装する場合
• 製品固有の知識を備えたカスタマーサポートボットを構築する場合
• ユーザーが RAG、ベクトル検索、またはドキュメント埋め込みについて尋ねた場合

仕組み

graph TD
    A[Documents: PDF, MD, HTML] --> B[Ingestion Pipeline]
    B --> C[Extract Text + Metadata]
    C --> D[Intelligent Chunking]
    D --> E[Generate Embeddings]
    E --> F[Index in Vector Store]
    G[User Query] --> H[Embed Query]
    H --> I[Semantic Search: Top-K]
    I --> J[Re-rank Results]
    J --> K[Construct Prompt with Context]
    K --> L[LLM Generation]
    L --> M[Response with Citations]

このパイプラインは、オフラインの取り込み（ドキュメントからベクトルへの変換）とオンラインの検索（クエリから回答の生成）の 2 つのフェーズで構成されます。再ランキングのステップでは、クロスエンコーダーを適用して初期のベクトル検索結果を調整し、生成の前に精度を高めます。

実装

from dataclasses import dataclass
import hashlib

@dataclass
class Chunk:
    text: str
    metadata: dict
    embedding: list[float] | None = None

    @property
    def id(self) -> str:
        return hashlib.sha256(self.text.encode()).hexdigest()[:16]

class RecursiveChunker:
    def __init__(self, max_tokens: int = 512, overlap: int = 64):
        self.max_tokens = max_tokens
        self.overlap = overlap
        self.separators = ["\n## ", "\n### ", "\n\n", "\n", ". ", " "]

    def chunk(self, text: str, metadata: dict) -> list[Chunk]:
        chunks = self._split(text, self.separators)
        return [
            Chunk(text=c.strip(), metadata={**metadata, "chunk_index": i})
            for i, c in enumerate(chunks) if c.strip()
        ]

    def _split(self, text: str, separators: list[str]) -> list[str]:
        if not separators or self._token_count(text) <= self.max_tokens:
            return [text]

        sep = separators[0]
        parts = text.split(sep)
        chunks, current = [], ""

        for part in parts:
            candidate = current + sep + part if current else part
            if self._token_count(candidate) > self.max_tokens and current:
                chunks.append(current)
                overlap_text = current[-self.overlap * 4:]
                current = overlap_text + sep + part
            else:
                current = candidate

        if current:
            chunks.append(current)

        result = []
        for chunk in chunks:
            if self._token_count(chunk) > self.max_tokens:
                result.extend(self._split(chunk, separators[1:]))
            else:
                result.append(chunk)
        return result

    def _token_count(self, text: str) -> int:
        return len(text) // 4

class RAGPipeline:
    def __init__(self, embedder, vector_store, llm):
        self.embedder = embedder
        self.store = vector_store
        self.llm = llm
        self.chunker = RecursiveChunker()

    async def ingest(self, documents: list[dict]) -> int:
        all_chunks = []
        for doc in documents:
            chunks = self.chunker.chunk(doc["text"], doc["metadata"])
            for chunk in chunks:
                chunk.embedding = await self.embedder.embed(chunk.text)
            all_chunks.extend(chunks)

        await self.store.upsert(all_chunks)
        return len(all_chunks)

    async def query(self, question: str, top_k: int = 5) -> dict:
        query_embedding = await self.embedder.embed(question)
        results = await self.store.search(query_embedding, top_k=top_k)

        context = "\n\n".join(
            f"[Source: {r.metadata.get('source', 'unknown')}]\n{r.text}" for r in results
        )

        prompt = f"""Answer the question based on the provided context. Cite sources.
If the context does not contain the answer, say so explicitly.

Context:
{context}

Question: {question}"""

        response = await self.llm.generate(prompt)
        return {"answer": response, "sources": [r.metadata for r in results]}

ベストプラクティス

• 見出し、段落、コードブロックなどのドキュメント構造を尊重する再帰的なチャンク分割を使用します
• ほとんどのユースケースでは、チャンクサイズを 256～512 トークン、重複率を 10～15％に設定します
• ベクトル検索の結果を LLM に渡す前に、クロスエンコーダーで再ランキングします
• 引用を生成できるよう、すべてのチャンクに出典メタデータを含めます
• 検索ノイズを避けるため、インデックス作成前にコンテンツハッシュでチャンクの重複を排除します
• コンテキストに回答が含まれていない場合は「わかりません」と答えるよう LLM に指示します

プラットフォームの互換性

プラットフォーム	対応状況	備考
Cursor	完全対応	パイプラインのコード生成
VS Code	完全対応	Python/TS による RAG 実装
Windsurf	完全対応	RAG ワークフローに対応
Claude Code	完全対応	エンドツーエンドの RAG 構築
Cline	完全対応	ベクトルストアとの統合
aider	部分対応	コードレベルで対応

関連スキル

• AI Chat Studio
• Workflow Orchestration
• Knowledge Base Injection
• Entity Memory Management

キーワード

rag retrieval-augmented-generation vector-search embeddings chunking knowledge-base semantic-search document-qa

---

© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License