RAGで契約書検索が破綻する？複数文書の類似コンテンツを扱うための実践的チャンキング戦略

なぜRAGは「似たような文書」で迷子になるのか

「RAGシステムを構築したのに、全然使い物にならない」

この悩みは驚くほど多くの開発現場で聞かれる。特に深刻なのが、複数の類似文書を扱うケースだ。

例えば、こんなシナリオを想像してほしい：

• 50社との業務委託契約書を管理している
• どの契約も「第3条：報酬」「第8条：守秘義務」など構造が酷似
• 「報酬はいくらですか？」と質問すると、どの契約の話か分からない回答が返ってくる

これは、単純なチャンキング戦略の致命的な欠陥だ。契約書を100トークンの小さなチャンクに切り刻むと、「どの契約書の、どの部分か」という文脈情報が完全に失われる。

実際のチャンク例：

❌ 悪い例
"第3条 報酬は月額100万円とする。支払期日は毎月末日..."
→ どこの会社との契約？いつの契約？

本記事では、この問題を解決する4つの実践的なチャンキング戦略を、サンプルコード付きで解説する。それぞれの戦略が「どんなデータに適しているか」も明確にしていく。

パターン1：メタデータエンリッチメント方式

概要

最もシンプルで効果的な方法は、各チャンクにメタデータを付与することだ。これはベクトルDBの基本機能として多くのプラットフォームが対応している。

参考：Pinecone - Metadata Filtering

実装例

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings

# ドキュメントとメタデータ
contracts = [
    {
        "content": "第3条 報酬\n本契約における月額報酬は100万円とする...",
        "metadata": {
            "contract_id": "ABC-2024-001",
            "company_name": "株式会社サンプル",
            "contract_date": "2024-01-15",
            "contract_type": "業務委託契約",
            "section": "第3条",
            "category": "報酬"
        }
    }
]

# チャンキング
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)

# 各チャンクにメタデータを継承
chunks_with_metadata = []
for contract in contracts:
    chunks = text_splitter.split_text(contract["content"])
    for i, chunk in enumerate(chunks):
        chunks_with_metadata.append({
            "text": chunk,
            "metadata": {
                **contract["metadata"],
                "chunk_index": i
            }
        })

# ベクトルストアに保存
embeddings = OpenAIEmbeddings()
vectorstore = Pinecone.from_documents(
    chunks_with_metadata,
    embeddings,
    index_name="contracts"
)

# メタデータフィルタリング付き検索
results = vectorstore.similarity_search(
    "報酬について教えて",
    k=3,
    filter={"company_name": "株式会社サンプル"}
)

適したデータ

• ✅ 契約書、規約、マニュアルなど構造化された文書
• ✅ 発行日、部署、バージョンなど明確な属性を持つ文書
• ✅ 特定の条件で絞り込み検索が必要なケース

メリット・デメリット

メリット：

• 実装が簡単で、ほとんどのベクトルDBが対応
• 検索時の絞り込みが高速
• 既存のRAGパイプラインに容易に統合可能

デメリット：

• メタデータ設計が重要（後から変更しづらい）
• チャンク本文には文脈情報が含まれない
• メタデータが欠落すると検索精度が低下

参考：LangChain - Metadata

パターン2：コンテキストプレフィックス方式

概要

各チャンクのテキスト本文の先頭に文脈情報を埋め込む方法だ。メタデータに依存せず、チャンク自体が自己完結する。

実装例

def add_context_prefix(document, metadata):
    """
    チャンクに文脈プレフィックスを追加
    """
    prefix = f"""
【契約書情報】
契約先：{metadata['company_name']}
契約日：{metadata['contract_date']}
契約種別：{metadata['contract_type']}
セクション：{metadata['section']}

---
"""
    return prefix + document

# 適用例
original_chunk = "第3条 報酬\n月額報酬は100万円とし、毎月末日までに支払う。"

enriched_chunk = add_context_prefix(
    original_chunk,
    {
        "company_name": "株式会社サンプル",
        "contract_date": "2024-01-15",
        "contract_type": "業務委託契約",
        "section": "第3条"
    }
)

print(enriched_chunk)

出力：

【契約書情報】
契約先：株式会社サンプル
契約日：2024-01-15
契約種別：業務委託契約
セクション：第3条

---
第3条 報酬
月額報酬は100万円とし、毎月末日までに支払う。

DifyでのTips

Difyを使用している場合、カスタムセグメント分割ルールでこのパターンを適用できる：

# Dify Knowledge Base APIでの登録例
import requests

def upload_with_context(file_path, metadata):
    # 文書を読み込み
    with open(file_path, 'r') as f:
        content = f.read()

    # プレフィックス付きコンテンツ生成
    enriched_content = add_context_prefix(content, metadata)

    # Dify APIにアップロード
    response = requests.post(
        'https://your-dify-instance.com/api/datasets/documents',
        headers={'Authorization': 'Bearer YOUR_API_KEY'},
        json={
            'name': f"{metadata['company_name']}_契約書",
            'text': enriched_content,
            'indexing_technique': 'high_quality',
            'segmentation': {
                'mode': 'custom',
                'max_tokens': 500,
                'overlap': 50
            }
        }
    )
    return response.json()

参考：Dify Documentation - Knowledge Base

適したデータ

• ✅ シンプルなRAGシステム（メタデータフィルタリング未対応）
• ✅ チャンク単体で意味を成す必要があるケース
• ✅ LLMが文脈を理解して回答する必要がある場合

メリット・デメリット

メリット：

• チャンク単体で完結するため、どのRAGシステムでも動作
• LLMへの入力として自然で理解しやすい
• メタデータフィルタリング機能がなくても使える

デメリット：

• チャンクサイズが増大（トークン消費増）
• 埋め込みベクトルに文脈情報も含まれる（検索精度への影響は要検証）
• プレフィックスの設計が重要

パターン3：階層的チャンキング方式

概要

複数の粒度でチャンクを作成し、検索クエリに応じて適切な階層を使い分ける方法だ。親子関係を保持することで、文脈と詳細のバランスを取る。

参考：LlamaIndex - Hierarchical Node Parser

実装例

from typing import List, Dict
import hashlib

class HierarchicalChunker:
    def __init__(self):
        self.chunks = []

    def create_hierarchical_chunks(self, contract: Dict) -> List[Dict]:
        """
        3階層のチャンクを生成
        Level 1: 契約書サマリー
        Level 2: セクションレベル
        Level 3: 詳細チャンク
        """
        contract_id = contract['metadata']['contract_id']

        # Level 1: サマリー
        summary = self._create_summary(contract)
        summary_chunk = {
            "id": f"{contract_id}_summary",
            "level": 1,
            "type": "summary",
            "text": summary,
            "metadata": contract['metadata'],
            "parent_id": None,
            "children_ids": []
        }

        # Level 2: セクション
        sections = self._split_sections(contract['content'])
        section_chunks = []

        for i, section in enumerate(sections):
            section_id = f"{contract_id}_section_{i}"
            section_chunk = {
                "id": section_id,
                "level": 2,
                "type": "section",
                "text": section['text'],
                "metadata": {
                    **contract['metadata'],
                    "section": section['title']
                },
                "parent_id": summary_chunk['id'],
                "children_ids": []
            }
            section_chunks.append(section_chunk)
            summary_chunk['children_ids'].append(section_id)

            # Level 3: 詳細チャンク
            details = self._split_into_small_chunks(section['text'], 300)
            for j, detail in enumerate(details):
                detail_id = f"{section_id}_detail_{j}"
                detail_chunk = {
                    "id": detail_id,
                    "level": 3,
                    "type": "detail",
                    "text": detail,
                    "metadata": {
                        **contract['metadata'],
                        "section": section['title']
                    },
                    "parent_id": section_id,
                    "children_ids": []
                }
                self.chunks.append(detail_chunk)
                section_chunk['children_ids'].append(detail_id)

        self.chunks.extend([summary_chunk] + section_chunks)
        return self.chunks

    def _create_summary(self, contract: Dict) -> str:
        """契約書サマリーを生成（実際はLLMを使用）"""
        meta = contract['metadata']
        return f"""
【契約書サマリー】
契約先：{meta['company_name']}
契約ID：{meta['contract_id']}
契約日：{meta['contract_date']}
契約種別：{meta['contract_type']}

【主要条項】
• 業務範囲：第2条
• 報酬条件：第3条
• 支払条件：第4条
• 守秘義務：第8条
• 契約期間：第10条

【契約期間】
{meta.get('start_date', 'N/A')} 〜 {meta.get('end_date', 'N/A')}
"""

    def _split_sections(self, content: str) -> List[Dict]:
        """セクションに分割（簡易版）"""
        import re
        sections = []
        section_pattern = r'(第\d+条[^\n]+)\n(.*?)(?=第\d+条|$)'
        matches = re.finditer(section_pattern, content, re.DOTALL)

        for match in matches:
            sections.append({
                'title': match.group(1).strip(),
                'text': match.group(0).strip()
            })
        return sections

    def _split_into_small_chunks(self, text: str, chunk_size: int) -> List[str]:
        """小チャンクに分割"""
        words = text.split()
        chunks = []
        for i in range(0, len(words), chunk_size):
            chunks.append(' '.join(words[i:i+chunk_size]))
        return chunks

# 使用例
chunker = HierarchicalChunker()
contract = {
    'content': "第1条 目的\n本契約は...\n第2条 業務範囲\n...",
    'metadata': {
        'contract_id': 'ABC-2024-001',
        'company_name': '株式会社サンプル',
        'contract_date': '2024-01-15',
        'contract_type': '業務委託契約'
    }
}

hierarchical_chunks = chunker.create_hierarchical_chunks(contract)

検索戦略

def hierarchical_search(query: str, vectorstore, top_k: int = 3):
    """
    階層的検索：まず上位階層で探し、必要に応じて下位へ
    """
    # Step 1: サマリーレベルで検索
    summary_results = vectorstore.similarity_search(
        query,
        k=top_k,
        filter={"level": 1}
    )

    relevant_contract_ids = [
        r.metadata['contract_id'] for r in summary_results
    ]

    # Step 2: 該当契約のセクションレベルで検索
    section_results = vectorstore.similarity_search(
        query,
        k=top_k * 2,
        filter={
            "level": 2,
            "contract_id": {"$in": relevant_contract_ids}
        }
    )

    # Step 3: 必要に応じて詳細チャンクを取得
    if needs_detail(section_results):
        detail_results = vectorstore.similarity_search(
            query,
            k=top_k * 3,
            filter={
                "level": 3,
                "parent_id": {"$in": [r.id for r in section_results]}
            }
        )
        return detail_results

    return section_results

適したデータ

• ✅ 長大な文書（100ページ以上の技術文書、法律文書）
• ✅ 明確な階層構造を持つ文書（章・節・項）
• ✅ 概要から詳細へ段階的に情報を深掘りするユースケース

メリット・デメリット

メリット：

• 文脈を保ちながら詳細情報にアクセス可能
• クエリの抽象度に応じた柔軟な検索
• トークン消費の最適化（必要な粒度のみ取得）

デメリット：

• 実装が複雑
• ストレージ容量が増大（同じ内容を複数粒度で保存）
• 階層設計が適切でないと逆効果

参考：Anthropic - Contextual Retrieval

パターン4：QAペア方式

概要

元文書から想定されるQ&Aペアを事前生成し、それ自体をチャンクとしてインデックス化する方法だ。ユーザーの質問と高い類似度を示すため、検索精度が劇的に向上する。

参考：OpenAI - Question Answering

実装例

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

class QAGenerator:
    def __init__(self, llm):
        self.llm = llm
        self.prompt = PromptTemplate(
            input_variables=["company_name", "section", "content"],
            template="""
以下の契約書セクションから、具体的なQ&Aペアを10個生成してください。

【重要】
- 各質問には必ず契約先名を含めること
- 回答は簡潔に（1-2文）
- 条項番号を明記すること

契約情報:
契約先: {company_name}
セクション: {section}

セクション内容:
{content}

形式:
Q1: [質問]
A1: [回答]（根拠：第X条Y項）

Q2: ...
"""
        )

    def generate_qa_pairs(self, contract_section: Dict) -> List[Dict]:
        """QAペアを生成"""
        # LLMでQA生成
        response = self.llm(
            self.prompt.format(
                company_name=contract_section['metadata']['company_name'],
                section=contract_section['metadata']['section'],
                content=contract_section['content']
            )
        )

        # パース（実際はより堅牢なパーサーを使用）
        qa_pairs = self._parse_qa_response(response)

        # メタデータ付与
        for qa in qa_pairs:
            qa['metadata'] = {
                **contract_section['metadata'],
                'type': 'qa_pair',
                'source_section': contract_section['metadata']['section']
            }

        return qa_pairs

    def _parse_qa_response(self, response: str) -> List[Dict]:
        """LLM出力をパース"""
        import re
        qa_pairs = []

        pattern = r'Q\d+:\s*(.+?)\s*A\d+:\s*(.+?)(?=Q\d+:|$)'
        matches = re.finditer(pattern, response, re.DOTALL)

        for match in matches:
            qa_pairs.append({
                'question': match.group(1).strip(),
                'answer': match.group(2).strip()
            })

        return qa_pairs

# 使用例
llm = OpenAI(temperature=0.3)
qa_generator = QAGenerator(llm)

contract_section = {
    'content': """
第3条（報酬）
1. 委託者は受託者に対し、本業務の対価として月額100万円を支払う。
2. 支払期日は毎月末日とし、受託者指定の銀行口座に振り込む。
3. 振込手数料は委託者の負担とする。
""",
    'metadata': {
        'company_name': '株式会社サンプル',
        'contract_id': 'ABC-2024-001',
        'section': '第3条'
    }
}

qa_pairs = qa_generator.generate_qa_pairs(contract_section)

# 出力例:
# [
#   {
#     'question': '株式会社サンプルとの契約で月額報酬はいくらですか？',
#     'answer': '月額100万円です（根拠：第3条1項）',
#     'metadata': {...}
#   },
#   ...
# ]

Difyでのバッチ処理

import requests
import time

def batch_generate_qa_for_dify(contracts: List[Dict], dify_api_url: str, api_key: str):
    """
    Dify APIを使ってQA生成とナレッジベース登録を自動化
    """
    for contract in contracts:
        # Step 1: DifyのワークフローAPIでQA生成
        workflow_response = requests.post(
            f"{dify_api_url}/workflows/run",
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "inputs": {
                    "contract_content": contract['content'],
                    "company_name": contract['metadata']['company_name']
                },
                "response_mode": "blocking"
            }
        )

        qa_pairs = workflow_response.json()['data']['outputs']['qa_pairs']

        # Step 2: QAペアをナレッジベースに登録
        for qa in qa_pairs:
            document_text = f"Q: {qa['question']}\nA: {qa['answer']}"

            requests.post(
                f"{dify_api_url}/datasets/documents",
                headers={"Authorization": f"Bearer {api_key}"},
                json={
                    "name": f"QA_{contract['metadata']['contract_id']}_{qa['id']}",
                    "text": document_text,
                    "indexing_technique": "high_quality"
                }
            )

        time.sleep(1)  # レート制限対策

適したデータ

• ✅ FAQが明確に想定できる文書（マニュアル、規約、契約書）
• ✅ ユーザーが質問形式でクエリを投げるシステム
• ✅ 検索精度が最重要なケース

メリット・デメリット

メリット：

• 検索精度が極めて高い（質問→質問のマッチング）
• 回答がそのまま使える（LLMの生成処理不要な場合も）
• ユーザーの質問パターンに最適化

デメリット：

• QA生成コストが高い（LLM API料金）
• 想定外の質問には対応できない
• QAの品質がシステム全体の品質を左右

参考：LangChain - Question Answering over Documents

パターン比較表

実践：複数パターンの組み合わせ

実際のプロダクションでは、複数パターンを組み合わせるのがベストプラクティスだ。

推奨構成例

# 契約書RAGシステムの実装例
class ContractRAGSystem:
    def __init__(self):
        self.vectorstore = self._init_vectorstore()

    def index_contract(self, contract: Dict):
        """契約書を複数方式でインデックス化"""

        # 1. 階層的チャンク（メタデータ付き）
        hierarchical_chunks = self._create_hierarchical_chunks(contract)

        # 2. QAペア生成
        qa_pairs = self._generate_qa_pairs(contract)

        # 3. すべてをベクトルストアに登録
        all_chunks = hierarchical_chunks + qa_pairs

        for chunk in all_chunks:
            # コンテキストプレフィックスも追加
            enriched_text = self._add_context_prefix(
                chunk['text'],
                chunk['metadata']
            )

            self.vectorstore.add_documents([{
                'text': enriched_text,
                'metadata': chunk['metadata']
            }])

    def search(self, query: str, company_name: str = None):
        """検索戦略：QA優先、次に階層検索"""

        # まずQAペアで検索
        qa_results = self.vectorstore.similarity_search(
            query,
            k=3,
            filter={
                "type": "qa_pair",
                "company_name": company_name if company_name else None
            }
        )

        if self._is_confident(qa_results):
            return qa_results

        # QAで見つからなければ階層検索
        return self._hierarchical_search(query, company_name)

まとめ：RAGの成功は「データを知ること」から

RAGシステムが失敗する最大の理由は、データの特性を無視した一律のチャンキングだ。

本記事で紹介した4つのパターンは、それぞれ異なるデータ特性に最適化されている：

• メタデータエンリッチメント：手軽に始めるならこれ
• コンテキストプレフィックス：チャンク単体で完結させたいなら
• 階層的チャンキング：大規模文書で文脈を保ちたいなら
• QAペア方式：検索精度を最大化したいなら

特に契約書管理システムのような、複数の類似文書を扱うケースでは、「階層的チャンキング + QAペア」の組み合わせが効果的だ。

RAGは「魔法の技術」ではなく、データエンジニアリングの延長線にある。データを深く理解し、適切なチャンキング戦略を選択することで、初めて実用的なシステムになる。

次のステップとして、実際にサンプルデータでプロトタイプを作り、各パターンの検索精度を比較してみることをお勧めする。

参考リンク

• LangChain Documentation
• Pinecone Vector Database
• Dify - Open Source LLM App Development Platform
• Anthropic Contextual Retrieval
• OpenAI Embeddings Guide
• LlamaIndex
• Chroma Vector Database
• Weaviate Vector Database