リトリーバルとは

リトリーバル（Retrieval）は、大規模なデータベース、文書コレクション、知識ベースから、ユーザーのクエリに関連する情報を効率的に検索・取得する技術です。従来のキーワードベース検索から発展し、意味的類似性、ベクトル表現、ニューラル検索などの手法により、文脈と意図を理解した高精度な情報取得を実現します。RAG（Retrieval-Augmented Generation）、質問応答、文書要約などのAIアプリケーションにおいて、外部知識の活用と応答品質の向上を可能にする重要な基盤技術です。

背景と重要性

現代のAIシステムは膨大な知識を持ちますが、訓練データに含まれない最新情報や専門知識には限界があります。また、すべての情報をモデルパラメータに埋め込むことは計算的に非効率です。リトリーバル技術により、必要に応じて外部の信頼できる情報源から関連情報を取得することで、これらの課題を解決できます。

リトリーバルは、

• AI システムの知識範囲の動的拡張
• 最新情報と専門知識の活用
• 計算効率性と拡張性の向上
• 回答の信頼性と透明性の確保

を実現することで、より実用的で信頼できるAIアプリケーションの構築を可能にします。情報爆発時代において、適切な情報を効率的に見つけ出すことは極めて重要な能力です。

主な構成要素

文書コーパス（Document Corpus）

検索対象となる文書や情報の集合です。

インデックス（Index）

高速検索を可能にするデータ構造です。

クエリエンコーダー（Query Encoder）

検索クエリをベクトル表現に変換するモデルです。

文書エンコーダー（Document Encoder）

文書をベクトル表現に変換するモデルです。

類似度計算（Similarity Computation）

クエリと文書の関連性を測定する手法です。

ランキングシステム（Ranking System）

検索結果を関連度順に並べるシステムです。

主な特徴

拡張性

大規模なデータベースに対応できます。

効率性

高速な検索とリアルタイム応答が可能です。

多様性

様々なデータ形式と検索要求に対応できます。

リトリーバル手法の詳細

密ベクトル検索（Dense Retrieval）

Dense Passage Retrieval（DPR）の実装：

import numpy as np
import torch
import torch.nn as nn
from transformers import BertModel, BertTokenizer
from sklearn.metrics.pairwise import cosine_similarity
import faiss

class DensePassageRetriever:
    def __init__(self, model_name='bert-base-uncased', max_length=512):
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.tokenizer = BertTokenizer.from_pretrained(model_name)
        self.max_length = max_length
        
        # エンコーダーモデル
        self.query_encoder = BertModel.from_pretrained(model_name).to(self.device)
        self.passage_encoder = BertModel.from_pretrained(model_name).to(self.device)
        
        # インデックス
        self.passage_embeddings = None
        self.passages = []
        self.faiss_index = None
        
    def encode_query(self, query):
        """クエリのエンコーディング"""
        encoded = self.tokenizer(
            query,
            truncation=True,
            padding=True,
            max_length=self.max_length,
            return_tensors='pt'
        ).to(self.device)
        
        with torch.no_grad():
            outputs = self.query_encoder(**encoded)
            # [CLS]トークンの埋め込みを使用
            query_embedding = outputs.last_hidden_state[:, 0, :].cpu().numpy()
        
        return query_embedding
    
    def encode_passages(self, passages):
        """パッセージのバッチエンコーディング"""
        embeddings = []
        batch_size = 16
        
        print(f"パッセージエンコーディング開始: {len(passages)} 件")
        
        for i in range(0, len(passages), batch_size):
            batch_passages = passages[i:i + batch_size]
            
            encoded = self.tokenizer(
                batch_passages,
                truncation=True,
                padding=True,
                max_length=self.max_length,
                return_tensors='pt'
            ).to(self.device)
            
            with torch.no_grad():
                outputs = self.passage_encoder(**encoded)
                batch_embeddings = outputs.last_hidden_state[:, 0, :].cpu().numpy()
                embeddings.append(batch_embeddings)
            
            if (i // batch_size + 1) % 10 == 0:
                print(f"進捗: {i + len(batch_passages)}/{len(passages)}")
        
        return np.vstack(embeddings)
    
    def build_index(self, passages):
        """FAISSインデックスの構築"""
        print("インデックス構築開始...")
        
        self.passages = passages
        
        # パッセージをエンコード
        self.passage_embeddings = self.encode_passages(passages)
        
        # FAISSインデックス構築
        dimension = self.passage_embeddings.shape[1]
        self.faiss_index = faiss.IndexFlatIP(dimension)  # 内積（コサイン類似度）
        
        # 正規化（コサイン類似度のため）
        faiss.normalize_L2(self.passage_embeddings)
        self.faiss_index.add(self.passage_embeddings)
        
        print(f"インデックス構築完了: {len(passages)} パッセージ, 次元数: {dimension}")
    
    def retrieve(self, query, top_k=10):
        """クエリに対する検索実行"""
        if self.faiss_index is None:
            raise ValueError("インデックスが構築されていません")
        
        # クエリエンコーディング
        query_embedding = self.encode_query(query)
        
        # 正規化
        faiss.normalize_L2(query_embedding)
        
        # 検索実行
        scores, indices = self.faiss_index.search(query_embedding, top_k)
        
        # 結果整理
        results = []
        for i, (score, idx) in enumerate(zip(scores[0], indices[0])):
            results.append({
                'rank': i + 1,
                'passage': self.passages[idx],
                'score': float(score),
                'index': int(idx)
            })
        
        return results
    
    def evaluate_retrieval(self, queries, relevant_passages_per_query, top_k=10):
        """検索性能の評価"""
        
        total_recall = 0
        total_precision = 0
        total_queries = len(queries)
        
        for i, query in enumerate(queries):
            relevant_passages = relevant_passages_per_query[i]
            retrieved_results = self.retrieve(query, top_k)
            
            # 取得されたパッセージのインデックス
            retrieved_indices = {result['index'] for result in retrieved_results}
            relevant_indices = set(relevant_passages)
            
            # Recall@k and Precision@k
            hits = len(retrieved_indices & relevant_indices)
            recall = hits / len(relevant_indices) if relevant_indices else 0
            precision = hits / len(retrieved_indices) if retrieved_indices else 0
            
            total_recall += recall
            total_precision += precision
            
            print(f"Query {i+1}: Recall@{top_k}={recall:.3f}, Precision@{top_k}={precision:.3f}")
        
        avg_recall = total_recall / total_queries
        avg_precision = total_precision / total_queries
        
        print(f"\n=== 評価結果 ===")
        print(f"平均 Recall@{top_k}: {avg_recall:.3f}")
        print(f"平均 Precision@{top_k}: {avg_precision:.3f}")
        
        return avg_recall, avg_precision

# ハイブリッド検索（BM25 + Dense）
class HybridRetriever:
    def __init__(self, dense_retriever, bm25_weight=0.3, dense_weight=0.7):
        self.dense_retriever = dense_retriever
        self.bm25_weight = bm25_weight
        self.dense_weight = dense_weight
        self.bm25_index = None
        
    def build_bm25_index(self, passages):
        """BM25インデックスの構築"""
        try:
            from rank_bm25 import BM25Okapi
        except ImportError:
            print("Warning: rank_bm25が利用できません。pip install rank_bm25を実行してください。")
            return
        
        # 簡単なトークナイゼーション
        tokenized_passages = [passage.lower().split() for passage in passages]
        self.bm25_index = BM25Okapi(tokenized_passages)
        print("BM25インデックス構築完了")
    
    def hybrid_retrieve(self, query, top_k=10, rerank_size=50):
        """ハイブリッド検索（BM25 + Dense）"""
        
        if self.bm25_index is None:
            print("BM25インデックスが利用できません。Dense検索のみ実行します。")
            return self.dense_retriever.retrieve(query, top_k)
        
        # 1. 両方の手法で多めに検索
        dense_results = self.dense_retriever.retrieve(query, rerank_size)
        
        # 2. BM25スコア計算
        query_tokens = query.lower().split()
        bm25_scores = self.bm25_index.get_scores(query_tokens)
        
        # 3. ハイブリッドスコア計算
        hybrid_results = []
        for result in dense_results:
            idx = result['index']
            dense_score = result['score']
            bm25_score = bm25_scores[idx] if idx < len(bm25_scores) else 0
            
            # 正規化とハイブリッドスコア
            hybrid_score = (self.dense_weight * dense_score + 
                          self.bm25_weight * bm25_score)
            
            hybrid_results.append({
                'rank': 0,  # 後で更新
                'passage': result['passage'],
                'dense_score': dense_score,
                'bm25_score': bm25_score,
                'hybrid_score': hybrid_score,
                'index': idx
            })
        
        # 4. ハイブリッドスコアでソート
        hybrid_results.sort(key=lambda x: x['hybrid_score'], reverse=True)
        
        # 5. ランク更新とtop_k選択
        for i, result in enumerate(hybrid_results[:top_k]):
            result['rank'] = i + 1
        
        return hybrid_results[:top_k]

def demonstrate_dense_retrieval():
    """Dense Retrievalのデモンストレーション"""
    
    print("=== Dense Retrieval デモ ===")
    
    # サンプル文書
    passages = [
        "機械学習は人工知能の一分野で、明示的にプログラムすることなく学習する能力をコンピューターに与える技術です。",
        "深層学習は機械学習の手法の一つで、人工ニューラルネットワークを使用してデータから学習します。",
        "自然言語処理は、人間の言語をコンピューターが理解し処理する技術分野です。",
        "コンピュータービジョンは、コンピューターが画像や動画から情報を抽出し理解する技術です。",
        "強化学習は、エージェントが環境との相互作用を通じて最適な行動を学習する手法です。",
        "ニューラルネットワークは、人間の脳神経系にヒントを得た計算モデルです。",
        "畳み込みニューラルネットワーク（CNN）は画像認識に特化したニューラルネットワークです。",
        "リカレントニューラルネットワーク（RNN）は時系列データの処理に適したニューラルネットワークです。",
        "トランスフォーマーは自然言語処理で広く使用される注意機構ベースのモデルです。",
        "BERT（Bidirectional Encoder Representations from Transformers）は事前学習済み言語モデルです。"
    ]
    
    # Dense Retrieverの初期化
    retriever = DensePassageRetriever()
    
    # インデックス構築
    retriever.build_index(passages)
    
    # 検索実行
    queries = [
        "深層学習について教えてください",
        "画像認識に使われる技術は何ですか",
        "自然言語を扱う技術について"
    ]
    
    for i, query in enumerate(queries):
        print(f"\n--- クエリ {i+1}: {query} ---")
        results = retriever.retrieve(query, top_k=3)
        
        for result in results:
            print(f"Rank {result['rank']}: Score {result['score']:.4f}")
            print(f"  {result['passage']}")
            print()
    
    # ハイブリッド検索のデモ
    print("\n=== ハイブリッド検索デモ ===")
    hybrid_retriever = HybridRetriever(retriever)
    hybrid_retriever.build_bm25_index(passages)
    
    query = "ニューラルネットワークの種類"
    print(f"クエリ: {query}")
    
    hybrid_results = hybrid_retriever.hybrid_retrieve(query, top_k=3)
    
    for result in hybrid_results:
        print(f"Rank {result['rank']}: "
              f"Hybrid {result['hybrid_score']:.4f} "
              f"(Dense: {result['dense_score']:.4f}, "
              f"BM25: {result['bm25_score']:.4f})")
        print(f"  {result['passage']}")
        print()

# demonstrate_dense_retrieval()

RAG（Retrieval-Augmented Generation）

RAGシステムの実装：

from transformers import T5ForConditionalGeneration, T5Tokenizer

class RAGSystem:
    def __init__(self, retriever, generator_model_name='t5-small'):
        self.retriever = retriever
        self.generator_tokenizer = T5Tokenizer.from_pretrained(generator_model_name)
        self.generator_model = T5ForConditionalGeneration.from_pretrained(generator_model_name)
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.generator_model.to(self.device)
        
    def generate_answer(self, question, top_k_passages=5, max_length=150):
        """RAGによる回答生成"""
        
        print(f"質問: {question}")
        print("関連文書を検索中...")
        
        # 1. 関連文書の検索
        retrieved_passages = self.retriever.retrieve(question, top_k=top_k_passages)
        
        # 2. コンテキストの構築
        context = self._build_context(retrieved_passages)
        
        # 3. 生成用プロンプトの作成
        prompt = self._create_generation_prompt(question, context)
        
        print(f"\n=== 検索された文書（上位{len(retrieved_passages)}件） ===")
        for i, passage in enumerate(retrieved_passages):
            print(f"{i+1}. (Score: {passage['score']:.4f}) {passage['passage'][:100]}...")
        
        # 4. 回答生成
        print("\n回答を生成中...")
        answer = self._generate_response(prompt, max_length)
        
        print(f"\n=== 生成された回答 ===")
        print(answer)
        
        return {
            'question': question,
            'answer': answer,
            'retrieved_passages': retrieved_passages,
            'context': context
        }
    
    def _build_context(self, passages):
        """検索された文書からコンテキストを構築"""
        context_parts = []
        for i, passage in enumerate(passages):
            context_parts.append(f"[文書{i+1}] {passage['passage']}")
        return " ".join(context_parts)
    
    def _create_generation_prompt(self, question, context):
        """生成用プロンプトの作成"""
        prompt = f"質問: {question}\n\n参考文書:\n{context}\n\n上記の文書を参考にして質問に答えてください:"
        return prompt
    
    def _generate_response(self, prompt, max_length):
        """T5モデルによる応答生成"""
        
        # 入力のトークン化
        inputs = self.generator_tokenizer(
            prompt,
            return_tensors='pt',
            max_length=512,
            truncation=True
        ).to(self.device)
        
        # 生成実行
        with torch.no_grad():
            outputs = self.generator_model.generate(
                **inputs,
                max_length=max_length,
                num_beams=4,
                early_stopping=True,
                no_repeat_ngram_size=2
            )
        
        # デコード
        answer = self.generator_tokenizer.decode(outputs[0], skip_special_tokens=True)
        return answer
    
    def multi_turn_qa(self, conversation_history, current_question, top_k=3):
        """複数ターンの質問応答"""
        
        # 会話履歴を考慮したクエリの拡張
        enhanced_query = self._enhance_query_with_history(conversation_history, current_question)
        
        # 検索と生成
        result = self.generate_answer(enhanced_query, top_k_passages=top_k)
        
        return result
    
    def _enhance_query_with_history(self, history, current_question):
        """会話履歴を考慮したクエリ拡張"""
        if not history:
            return current_question
        
        # 直前の質問と回答を考慮
        recent_context = history[-1] if history else ""
        enhanced_query = f"{recent_context} {current_question}"
        
        return enhanced_query

class AdvancedRAGSystem(RAGSystem):
    def __init__(self, retriever, generator_model_name='t5-small'):
        super().__init__(retriever, generator_model_name)
        self.query_rewriter = None
        
    def generate_answer_with_reranking(self, question, top_k_passages=10, 
                                     final_k=3, max_length=150):
        """再ランキング付きRAG"""
        
        # 1. 初期検索（多めに取得）
        initial_passages = self.retriever.retrieve(question, top_k=top_k_passages)
        
        # 2. 再ランキング
        reranked_passages = self._rerank_passages(question, initial_passages, final_k)
        
        # 3. 回答生成
        context = self._build_context(reranked_passages)
        prompt = self._create_generation_prompt(question, context)
        answer = self._generate_response(prompt, max_length)
        
        return {
            'question': question,
            'answer': answer,
            'initial_passages': initial_passages,
            'reranked_passages': reranked_passages
        }
    
    def _rerank_passages(self, question, passages, final_k):
        """Cross-encoder によるパッセージ再ランキング"""
        
        # 簡略化された再ランキング（実際にはCross-encoderを使用）
        # ここでは文字列の重複度で近似
        
        rerank_scores = []
        question_words = set(question.lower().split())
        
        for passage in passages:
            passage_words = set(passage['passage'].lower().split())
            overlap = len(question_words & passage_words)
            rerank_score = overlap / len(question_words) if question_words else 0
            
            rerank_scores.append({
                **passage,
                'rerank_score': rerank_score
            })
        
        # 再ランキングスコアでソート
        rerank_scores.sort(key=lambda x: x['rerank_score'], reverse=True)
        
        return rerank_scores[:final_k]
    
    def generate_with_citations(self, question, top_k_passages=3):
        """引用付き回答生成"""
        
        result = self.generate_answer(question, top_k_passages)
        
        # 引用を追加
        citations = []
        for i, passage in enumerate(result['retrieved_passages']):
            citations.append(f"[{i+1}] {passage['passage'][:50]}...")
        
        cited_answer = f"{result['answer']}\n\n参考文献:\n" + "\n".join(citations)
        
        return {
            **result,
            'cited_answer': cited_answer,
            'citations': citations
        }

def demonstrate_rag_system():
    """RAGシステムのデモンストレーション"""
    
    print("=== RAGシステム デモ ===")
    
    # 知識ベース
    knowledge_base = [
        "人工知能（AI）は、人間の知的な活動をコンピューターで模倣する技術の総称です。",
        "機械学習は、データから自動的にパターンを学習し、予測や判断を行う技術です。",
        "深層学習は、多層のニューラルネットワークを使用する機械学習の手法です。",
        "自然言語処理（NLP）は、人間の言語をコンピューターが理解し処理する技術です。",
        "コンピュータービジョンは、画像や動画からコンピューターが情報を抽出する技術です。",
        "強化学習は、試行錯誤を通じて最適な行動を学習するAI技術です。",
        "ニューラルネットワークは、人間の脳の神経回路を模倣した計算モデルです。",
        "畳み込みニューラルネットワーク（CNN）は画像認識に特化したモデルです。",
        "リカレントニューラルネットワーク（RNN）は時系列データの処理に適したモデルです。",
        "トランスフォーマーは注意機構を用いた強力な言語モデルアーキテクチャです。",
        "BERT は双方向エンコーダー表現を学習する事前学習済み言語モデルです。",
        "GPT は生成事前学習トランスフォーマーで、テキスト生成に優れたモデルです。",
        "エンベディングは、データを密ベクトル空間に変換する表現学習技術です。",
        "クラスタリングは、類似したデータポイントをグループ化する教師なし学習手法です。",
        "分類は、入力データを予め定義されたカテゴリに分ける教師あり学習タスクです。"
    ]
    
    # Retrieverの初期化
    retriever = DensePassageRetriever()
    retriever.build_index(knowledge_base)
    
    # RAGシステムの初期化
    rag_system = RAGSystem(retriever)
    
    # 質問応答のデモ
    questions = [
        "機械学習とは何ですか？",
        "画像認識に使われる技術について教えてください",
        "自然言語処理の応用例を教えてください"
    ]
    
    for question in questions:
        print(f"\n{'='*60}")
        result = rag_system.generate_answer(question, top_k_passages=3)
        print(f"{'='*60}\n")
    
    # 高度なRAGシステム
    print(f"\n{'='*60}")
    print("=== 高度なRAGシステム（再ランキング） ===")
    
    advanced_rag = AdvancedRAGSystem(retriever)
    result = advanced_rag.generate_answer_with_reranking(
        "深層学習の特徴について", top_k_passages=8, final_k=3
    )
    
    # 引用付き回答
    print(f"\n{'='*60}")
    print("=== 引用付き回答生成 ===")
    
    cited_result = advanced_rag.generate_with_citations("ニューラルネットワークの種類")
    print(cited_result['cited_answer'])

# demonstrate_rag_system()

多様な検索手法

コンテキスト適応型検索：

class ContextualRetriever:
    def __init__(self, base_retriever):
        self.base_retriever = base_retriever
        self.context_memory = []
        
    def contextual_retrieve(self, query, context_history=None, top_k=10):
        """コンテキストを考慮した検索"""
        
        # 1. クエリ拡張
        expanded_query = self._expand_query_with_context(query, context_history)
        
        # 2. 基本検索
        initial_results = self.base_retriever.retrieve(expanded_query, top_k * 2)
        
        # 3. コンテキスト適合性による再スコアリング
        contextual_results = self._rescore_with_context(
            initial_results, context_history, top_k
        )
        
        return contextual_results
    
    def _expand_query_with_context(self, query, context_history):
        """コンテキスト履歴を用いたクエリ拡張"""
        if not context_history:
            return query
        
        # 重要な用語を抽出
        context_terms = []
        for context in context_history[-3:]:  # 最近の3つのコンテキスト
            terms = context.split()[:5]  # 各コンテキストから5語
            context_terms.extend(terms)
        
        # クエリと組み合わせ
        expanded_query = f"{query} {' '.join(context_terms[:10])}"  # 上位10語
        return expanded_query
    
    def _rescore_with_context(self, results, context_history, top_k):
        """コンテキスト適合性による再スコアリング"""
        if not context_history:
            return results[:top_k]
        
        rescored_results = []
        
        for result in results:
            base_score = result['score']
            
            # コンテキスト適合性スコア
            context_score = self._compute_context_similarity(
                result['passage'], context_history
            )
            
            # 新しいスコア
            new_score = 0.7 * base_score + 0.3 * context_score
            
            rescored_results.append({
                **result,
                'original_score': base_score,
                'context_score': context_score,
                'final_score': new_score
            })
        
        # 新しいスコアでソート
        rescored_results.sort(key=lambda x: x['final_score'], reverse=True)
        return rescored_results[:top_k]
    
    def _compute_context_similarity(self, passage, context_history):
        """パッセージとコンテキスト履歴の類似度計算"""
        if not context_history:
            return 0
        
        passage_words = set(passage.lower().split())
        
        similarities = []
        for context in context_history:
            context_words = set(context.lower().split())
            intersection = len(passage_words & context_words)
            union = len(passage_words | context_words)
            similarity = intersection / union if union > 0 else 0
            similarities.append(similarity)
        
        # 平均類似度
        return np.mean(similarities)

class MultiModalRetriever:
    def __init__(self, text_retriever, image_retriever=None):
        self.text_retriever = text_retriever
        self.image_retriever = image_retriever
        
    def multimodal_retrieve(self, query, query_type='text', top_k=10):
        """マルチモーダル検索"""
        
        if query_type == 'text':
            return self.text_retriever.retrieve(query, top_k)
        elif query_type == 'image' and self.image_retriever:
            return self.image_retriever.retrieve(query, top_k)
        else:
            # テキストと画像の統合検索（概念的実装）
            text_results = self.text_retriever.retrieve(query, top_k // 2)
            # image_results = self.image_retriever.retrieve(query, top_k // 2)
            
            # 結果をマージ（実際にはより複雑な統合ロジックが必要）
            return text_results  # 簡略化

class FederatedRetriever:
    def __init__(self, retrievers):
        """複数の検索システムを統合"""
        self.retrievers = retrievers
        
    def federated_search(self, query, top_k_per_source=5, final_top_k=10):
        """フェデレーテッド検索"""
        
        all_results = []
        
        # 各検索システムから結果を取得
        for i, retriever in enumerate(self.retrievers):
            try:
                results = retriever.retrieve(query, top_k_per_source)
                
                # ソース情報を追加
                for result in results:
                    result['source'] = f'source_{i}'
                    result['source_rank'] = result['rank']
                
                all_results.extend(results)
                
            except Exception as e:
                print(f"Source {i} でエラー: {e}")
                continue
        
        # 統合ランキング
        integrated_results = self._integrate_rankings(all_results, final_top_k)
        
        return integrated_results
    
    def _integrate_rankings(self, all_results, final_top_k):
        """複数ソースからの結果統合"""
        
        # 簡略化された統合（実際にはより高度な手法を使用）
        # スコアの正規化と重み付け平均
        
        # ソース別にスコアを正規化
        sources = {}
        for result in all_results:
            source = result['source']
            if source not in sources:
                sources[source] = []
            sources[source].append(result)
        
        normalized_results = []
        for source, results in sources.items():
            scores = [r['score'] for r in results]
            max_score = max(scores) if scores else 1
            min_score = min(scores) if scores else 0
            
            for result in results:
                normalized_score = (result['score'] - min_score) / (max_score - min_score) if max_score > min_score else 0.5
                normalized_results.append({
                    **result,
                    'normalized_score': normalized_score
                })
        
        # 正規化スコアでソート
        normalized_results.sort(key=lambda x: x['normalized_score'], reverse=True)
        
        # ランク更新
        for i, result in enumerate(normalized_results[:final_top_k]):
            result['integrated_rank'] = i + 1
        
        return normalized_results[:final_top_k]

def demonstrate_advanced_retrieval():
    """高度な検索手法のデモンストレーション"""
    
    print("=== 高度な検索手法デモ ===")
    
    # 知識ベース
    tech_docs = [
        "Python は汎用プログラミング言語で、データサイエンスや機械学習に広く使用されています。",
        "JavaScript は主にWebブラウザで動作するスクリプト言語です。",
        "機械学習ライブラリとして scikit-learn、TensorFlow、PyTorch が人気です。",
        "データ分析には pandas、numpy、matplotlib がよく使用されます。",
        "Webフレームワークとして Django、Flask、FastAPI があります。",
        "自然言語処理には spaCy、NLTK、transformers ライブラリが利用されます。",
        "深層学習モデルの訓練には GPU が必要で、CUDA を使用します。",
        "Docker はアプリケーションのコンテナ化に使用される技術です。",
        "Git はバージョン管理システムで、ソフトウェア開発に不可欠です。",
        "API 開発では REST や GraphQL のアーキテクチャが使用されます。"
    ]
    
    # ベース検索器
    base_retriever = DensePassageRetriever()
    base_retriever.build_index(tech_docs)
    
    # 1. コンテキスト適応型検索
    print("\n=== コンテキスト適応型検索 ===")
    contextual_retriever = ContextualRetriever(base_retriever)
    
    # 最初の質問
    query1 = "プログラミング言語について"
    results1 = contextual_retriever.contextual_retrieve(query1, top_k=3)
    
    print(f"質問1: {query1}")
    for result in results1:
        print(f"  {result['passage'][:60]}... (Score: {result['score']:.3f})")
    
    # コンテキストを追加した質問
    context_history = ["Python プログラミング言語", "機械学習ライブラリ"]
    query2 = "おすすめのツール"
    results2 = contextual_retriever.contextual_retrieve(query2, context_history, top_k=3)
    
    print(f"\n質問2: {query2} (コンテキスト: {context_history})")
    for result in results2:
        print(f"  {result['passage'][:60]}... "
              f"(Final: {result['final_score']:.3f}, "
              f"Context: {result['context_score']:.3f})")
    
    # 2. フェデレーテッド検索（複数の検索システム統合）
    print(f"\n=== フェデレーテッド検索 ===")
    
    # 複数の専門検索器を模擬
    ml_docs = [
        "TensorFlow は Google が開発した機械学習フレームワークです。",
        "PyTorch は Facebook が開発した深層学習ライブラリです。",
        "scikit-learn は Python の機械学習ライブラリです。",
    ]
    
    web_docs = [
        "React は Facebook が開発したWebフロントエンドライブラリです。",
        "Node.js はサーバーサイド JavaScript 実行環境です。",
        "Express.js は Node.js 用の軽量 Web フレームワークです。",
    ]
    
    ml_retriever = DensePassageRetriever()
    ml_retriever.build_index(ml_docs)
    
    web_retriever = DensePassageRetriever()
    web_retriever.build_index(web_docs)
    
    federated_retriever = FederatedRetriever([base_retriever, ml_retriever, web_retriever])
    
    fed_query = "フレームワーク"
    fed_results = federated_retriever.federated_search(fed_query, top_k_per_source=2, final_top_k=5)
    
    print(f"統合検索クエリ: {fed_query}")
    for result in fed_results:
        print(f"Rank {result['integrated_rank']}: "
              f"({result['source']}) {result['passage'][:50]}... "
              f"(Score: {result['normalized_score']:.3f})")

# demonstrate_advanced_retrieval()

活用事例・ユースケース

リトリーバル技術は現代の情報システムにおいて広範囲に活用されています。

質問応答システム

カスタマーサポート、FAQ、技術文書からの情報検索。

文書検索・要約

大規模文書コレクションからの関連情報抽出と要約生成。

推薦システム

ユーザーの興味に基づくコンテンツや商品の推薦。

研究支援

学術論文検索、関連研究発見、文献レビュー支援。

企業内検索

社内文書、知識ベース、専門情報の効率的検索。

学ぶためのおすすめリソース

書籍

「Information Retrieval: Implementing and Evaluating Search Engines」（Stefan Büttcher）、「Modern Information Retrieval」（Ricardo Baeza-Yates）

オンラインコース

Stanford「CS276: Information Retrieval and Web Search」、CMU「11-442: Search Engines」

実装フレームワーク

Elasticsearch、Apache Solr、Faiss、Pinecone、Weaviate

論文

「Dense Passage Retrieval for Open-Domain Question Answering」、「Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks」

よくある質問（FAQ）

Q. Dense RetrievalとSparse Retrievalの違いは？
A. Dense Retrievalは密ベクトル表現を用いた意味的検索、Sparse Retrievalはキーワードベースの従来型検索です。多くの場合、両方を組み合わせたハイブリッド手法が効果的です。

Q. RAGシステムの精度を向上させる方法は？
A. 高品質な文書コーパス、適切なチャンク分割、再ランキング、クエリ拡張、文脈考慮などの手法により精度を向上できます。

Q. 大規模データに対するスケーラビリティは？
A. FAISSやPineconeなどのベクトル検索エンジン、分散インデックス、近似検索アルゴリズムにより、数十億件のデータに対してもスケールできます。

関連キーワード

情報検索、ベクトル検索、RAG、意味的類似性、Dense Retrieval、質問応答

まとめ

リトリーバルは、大規模な情報空間から関連する知識を効率的に取得する重要な技術です。Dense Retrieval、RAG、ハイブリッド検索など様々な手法により、従来のキーワード検索を超えた意味的で文脈適応的な情報取得が可能になります。質問応答、文書検索、推薦システムなど幅広いAIアプリケーションにおいて、外部知識の活用と応答品質の向上を実現します。今後も、より高精度で効率的な検索技術の発展により、AIシステムの知識範囲と実用性がさらに拡張されることが期待されます。