Claude RAG完全ガイド｜仕組み・実装方法・精度向上テクニックを徹底解説

📝 RAGの動作フロー

具体的な動作フローは以下の通りです。

まず、ユーザーの質問を受け取ると、AIはその質問を理解し、関連する情報を外部の知識ベースから検索します。

次に、検索結果から最も関連性の高い情報を選び取り、それを元に回答を生成します。

このプロセスにより、AIは常に最新かつ正確な情報を提供できるようになります。

📝 事例1:大手製造業における技術文書検索システム

設計図や仕様書、過去の不具合履歴などを検索対象として登録し、エンジニアが技術的な質問をすると、関連する文書を瞬時に検索して回答を生成します。

これにより、設計変更の検討時間が50%削減されました。

📝 事例2:金融機関の規制対応サポート

金融商品取引法(e-Gov法令検索)や日本銀行の指針、社内コンプライアンスマニュアルなどをRAGシステムに組み込み、営業担当者が顧客対応で迷った際に、最新の規制情報に基づいた適切な回答を即座に提供します。

これにより、コンプライアンス違反のリスクが大幅に削減されました。

📝 事例3:教育機関の学習支援システム

教科書や参考資料、過去の試験問題を登録し、生徒が質問すると関連する学習内容を検索して、個別最適化された解説を生成します。

特に数学や理科といった教科で、理解度が30%向上したという報告があります。

📝 必要なライブラリのインストール

まず、必要なライブラリをインストールします:

pip install anthropic faiss-cpu sentence-transformers PyPDF2 python-dotenv

📝 Claude Projectsの特徴

最速で始められる反面、カスタマイズ性が限定的です。

小規模なPoCや個人利用、簡単な文書検索には最適ですが、本格運用には向きません。

ファイルサイズ制限(10MB×5ファイル)や、リアルタイム更新の不可、ブランドカスタマイズの制限などが課題です。

📝 Difyの特徴

ノーコードで高度なカスタマイズが可能で、中小規模の本格導入に適しています。

Web UIやAPIエンドポイントが自動生成される点、複数のLLMとの連携が容易な点、ロールベースのアクセス制御が可能な点が利点です。

ただし、セルフホストの場合はサーバー管理が必要であり、大規模な利用ではコストが課題になる可能性があります。

📝 Python + Claude APIの特徴

完全にカスタマイズ可能で、大規模かつ高性能なシステムに適しています。

独自のアルゴリズム実装、高度なセキュリティ制御、既存システムとの深い統合が可能です。

一方で、開発期間が長く、技術的な専門知識が必要です。

📝 1. API認証エラー(401 Unauthorized)

このエラーは、Anthropic APIキーの設定に問題がある場合に発生します。

まず、環境変数ANTHROPIC_API_KEYが正しく設定されているか確認します。

Linux/Macではecho $ANTHROPIC_API_KEY、Windowsではecho %ANTHROPIC_API_KEY%で確認できます。

APIキーが正しくない場合は、Anthropic Consoleで新しいキーを生成し、再設定します。

📝 2. コンテキスト長エラー(context_length_exceeded)

Claudeのコンテキストウィンドウ(最大200,000トークン)を超えた場合に発生します。

検索結果やプロンプトが長すぎることが原因です。

解決策として、チャンクサイズを小さくする(例:500文字→300文字)、検索結果の数を減らす(例:k=5→k=3)、プロンプトを簡潔にする、といった方法があります。

特に日本語の場合、1文字あたりのトークン数が多いため注意が必要です

📝 3. レート制限エラー(rate_limit_exceeded)

APIのレート制限に達した場合に発生します。

無料枠では1分あたり5リクエスト、有料枠でも1分あたり1000リクエストが上限です。

解決策は、リクエスト間に待機時間を設ける(例:time.sleep(1))、バッチ処理を使って効率化する、エクスポネンシャルバックオフを実装する、などがあります。

📝 4. JSONパースエラー(JSONDecodeError)

Claudeのレスポンスが期待しない形式で返ってきた場合に発生します。

特に日本語のレスポンスでは、文字エンコーディングの問題が原因となることがあります。

解決策として、レスポンスを文字列として扱い、正規表現で必要な部分を抽出する、try-except文で例外処理を実装する、レスポンスをログに記録して分析する、などがあります。

📝 5. ベクトル検索の精度が低い

検索結果が期待外れの場合、主に以下の原因が考えられます。

• Embeddingモデルの選定が不適切(日本語に対応していない)
• チャンクサイズが不適切(大きすぎるか小さすぎる)
• 類似度計算の方法が不適切

解決策として、日本語に最適化されたEmbeddingモデル(例:sentence-transformers/paraphrase-multilingual-mpnet-base-v2)を使用する、チャンクサイズを300-500文字に調整する、ハイブリッド検索(BM25 + ベクトル検索)を実装する、などがあります。

📝 前処理の重要なポイント

日本語テキストの前処理では、以下の点に注意が必要です:

import MeCab
import re

def preprocess_japanese_text(text):
    # 半角・全角の統一
    text = mojimoji.han_to_zen(text)
    
    # 不要な記号の除去
    text = re.sub(r'[!-/-@[-`{-~]', '', text)
    
    # 形態素解析による単語分割
    tagger = MeCab.Tagger('-Owakati')
    words = tagger.parse(text).strip()
    
    return words

📝 文字エンコーディングの問題

日本語ファイルを処理する際は、文字エンコーディングに注意が必要です:

def read_japanese_file(file_path):
    encodings = ['utf-8', 'shift_jis', 'euc_jp', 'cp932']
    
    for encoding in encodings:
        try:
            with open(file_path, 'r', encoding=encoding) as f:
                return f.read()
        except UnicodeDecodeError:
            continue
    
    raise Exception(f"ファイルを読み込めません: {file_path}")

📝 トークン数の最適化

プロンプトの最適化により、無駄なトークンを削減できます:

def optimize_prompt(query, context):
    # 重複する情報を除去
    unique_context = remove_duplicate_info(context)
    
    # 重要度に基づいてコンテキストをフィルタリング
    relevant_context = filter_by_relevance(unique_context, query)
    
    # 簡潔なプロンプトテンプレートを使用
    optimized_prompt = f"""
    質問: {query}
    
    参考情報:
    {relevant_context}
    
    上記の情報を基に、簡潔に回答してください。
    """
    return optimized_prompt

📝 プログレッシブな品質設定

用途に応じて、使用するモデルを切り替えます:

def select_model_by_complexity(query, context_length):
    # 簡単な質問や短いコンテキストには、軽量なモデルを使用
    if is_simple_query(query) and context_length < 1000:
        return "claude-3-haiku-20240307"  # 最速、低コスト
    
    # 複雑な質問や長いコンテキストには、高性能モデルを使用
    elif is_complex_query(query) or context_length > 5000:
        return "claude-3-opus-20240229"  # 最高性能
    
    # それ以外は中間モデル
    else:
        return "claude-3-sonnet-20240229"  # バランス型

📝 スマートなキャッシング戦略

Redisなどの外部キャッシュを活用することで、コストを大幅に削減できます:

import redis
import json
import hashlib

class SmartCache:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        self.cache_ttl = 3600  # 1時間
    
    def get_or_generate(self, query, context, generate_func):
        # キャッシュキーを生成
        cache_key = self._generate_cache_key(query, context)
        
        # キャッシュを確認
        cached_result = self.redis_client.get(cache_key)
        if cached_result:
            return json.loads(cached_result)
        
        # 新規生成
        result = generate_func(query, context)
        
        # キャッシュに保存
        self.redis_client.setex(
            cache_key, 
            self.cache_ttl, 
            json.dumps(result)
        )
        
        return result

📝 1. Python環境の準備

Python 3.8以上が必要です。

仮想環境を作成することを推奨します:

# 仮想環境の作成
python -m venv claude-rag-env

# 仮想環境のアクティベート
# Windows
claude-rag-env\Scripts\activate
# Mac/Linux
source claude-rag-env/bin/activate

# 必要なライブラリのインストール
pip install anthropic python-dotenv requests

📝 2. Claude APIキーの取得

Anthropic Consoleにアクセスして以下の手順でAPIキーを取得します:

• 「API Keys」メニューをクリック
• 「Create Key」をクリック
• キー名を入力(例:claude-rag-project)
• 生成されたキーをコピー(この時点でしか表示されません)

📝 3. 環境変数の設定

プロジェクトルートに.envファイルを作成します:

ANTHROPIC_API_KEY=sk-ant-api03-...your-api-key...

📝 4. 基本的なAPI接続テスト

以下のコードでAPI接続を確認します:

import os
from dotenv import load_dotenv
from anthropic import Anthropic

# 環境変数の読み込み
load_dotenv()

# Claude APIクライアントの初期化
client = Anthropic(
    api_key=os.getenv("ANTHROPIC_API_KEY")
)

# 接続テスト
try:
    response = client.messages.create(
        model="claude-3-sonnet-20240229",
        max_tokens=100,
        messages=[{
            "role": "user",
            "content": "Hello, Claude!"
        }]
    )
    print("API接続成功:", response.content[0].text)
except Exception as e:
    print("API接続エラー:", e)

📝 ドキュメント読み込み機能

PDF、テキスト、Markdownファイルに対応:

import os
import re
from pathlib import Path
from PyPDF2 import PdfReader

class DocumentLoader:
    def __init__(self):
        self.supported_extensions = ['.pdf', '.txt', '.md']
    
    def load_document(self, file_path):
        """ドキュメントを読み込む"""
        file_path = Path(file_path)
        
        if not file_path.exists():
            raise FileNotFoundError(f"ファイルが見つかりません: {file_path}")
        
        extension = file_path.suffix.lower()
        
        if extension == '.pdf':
            return self._load_pdf(file_path)
        elif extension in ['.txt', '.md']:
            return self._load_text(file_path)
        else:
            raise ValueError(f"サポートされていないファイル形式: {extension}")
    
    def _load_pdf(self, file_path):
        """PDFファイルを読み込む"""
        reader = PdfReader(file_path)
        text = ""
        
        for page_num, page in enumerate(reader.pages, 1):
            try:
                page_text = page.extract_text()
                if page_text:
                    text += f"\n=== ページ {page_num} ===\n"
                    text += page_text
            except Exception as e:
                print(f"ページ {page_num} の読み取りエラー: {e}")
        
        return text
    
    def _load_text(self, file_path):
        """テキストファイルを読み込む"""
        encodings = ['utf-8', 'shift_jis', 'euc_jp', 'cp932']
        
        for encoding in encodings:
            try:
                with open(file_path, 'r', encoding=encoding) as f:
                    content = f.read()
                
                # 読み取り成功
                print(f"ファイルを読み込みました(エンコーディング: {encoding})")
                return content
            except UnicodeDecodeError:
                continue
        
        raise UnicodeDecodeError(
            f"ファイルを読み込めません: {file_path}"
        )
    
    def load_documents_from_directory(self, directory_path):
        """ディレクトリ内の全ドキュメントを読み込む"""
        directory = Path(directory_path)
        all_texts = []
        
        for file_path in directory.rglob('*'):
            if file_path.suffix.lower() in self.supported_extensions:
                try:
                    text = self.load_document(file_path)
                    all_texts.append({
                        'filename': file_path.name,
                        'content': text,
                        'path': str(file_path)
                    })
                except Exception as e:
                    print(f"エラー: {file_path} - {e}")
        
        return all_texts

📝 チャンク分割機能

文脈を保持しながら適切なサイズに分割:

class TextChunker:
    def __init__(self, chunk_size=500, chunk_overlap=50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    def chunk_text(self, text, filename=""):
        """テキストをチャンクに分割する"""
        # 文の境界を考慮した分割
        sentences = self._split_into_sentences(text)
        chunks = []
        current_chunk = []
        current_length = 0
        
        for sentence in sentences:
            sentence_length = len(sentence)
            
            # チャンクサイズを超える場合、新しいチャンクを開始
            if current_length + sentence_length > self.chunk_size and current_chunk:
                chunk_text = " ".join(current_chunk)
                chunks.append({
                    'text': chunk_text,
                    'source': filename,
                    'chunk_id': len(chunks),
                    'length': len(chunk_text)
                })
                
                # オーバーラップ処理
                overlap_text = self._get_overlap_text(current_chunk)
                current_chunk = [overlap_text] if overlap_text else []
                current_length = len(overlap_text) if overlap_text else 0
            
            current_chunk.append(sentence)
            current_length += sentence_length
        
        # 最後のチャンクを追加
        if current_chunk:
            chunk_text = " ".join(current_chunk)
            chunks.append({
                'text': chunk_text,
                'source': filename,
                'chunk_id': len(chunks),
                'length': len(chunk_text)
            })
        
        return chunks
    
    def _split_into_sentences(self, text):
        """日本語・英語の文章を文に分割する"""
        # 日本語の句読点と英語の句読点に対応
        sentence_patterns = [
            r'[。！？…]+',  # 日本語
            r'[.!?]+',     # 英語
            r'\n+'         # 改行
        ]
        
        # 複数のパターンで分割
        combined_pattern = '|'.join(sentence_patterns)
        sentences = re.split(combined_pattern, text)
        
        # 空の文を除去
        sentences = [s.strip() for s in sentences if s.strip()]
        
        return sentences
    
    def _get_overlap_text(self, sentences):
        """オーバーラップ用のテキストを取得"""
        overlap_length = 0
        overlap_sentences = []
        
        for sentence in reversed(sentences):
            sentence_length = len(sentence)
            if overlap_length + sentence_length <= self.chunk_overlap:
                overlap_sentences.insert(0, sentence)
                overlap_length += sentence_length
            else:
                break
        
        return " ".join(overlap_sentences) if overlap_sentences else ""

📝 統合使用例

# ドキュメント読み込みとチャンク分割の統合
def prepare_documents(docs_directory):
    loader = DocumentLoader()
    chunker = TextChunker(chunk_size=500, chunk_overlap=50)
    
    # ドキュメントを読み込む
    documents = loader.load_documents_from_directory(docs_directory)
    
    all_chunks = []
    for doc in documents:
        print(f"処理中: {doc['filename']}")
        
        # チャンク分割
        chunks = chunker.chunk_text(
            doc['content'], 
            filename=doc['filename']
        )
        
        for chunk in chunks:
            chunk['source_file'] = doc['path']
            all_chunks.append(chunk)
    
    print(f"総チャンク数: {len(all_chunks)}")
    return all_chunks

# 使用例
chunks = prepare_documents("documents/")

📝 埋め込みモデルの設定

日本語に対応した多言語モデルを使用:

from sentence_transformers import SentenceTransformer
import numpy as np

class EmbeddingModel:
    def __init__(self, model_name='sentence-transformers/paraphrase-multilingual-mpnet-base-v2'):
        print(f"Embeddingモデルを読み込み中: {model_name}")
        self.model = SentenceTransformer(model_name)
        self.embedding_dimension = self.model.get_sentence_embedding_dimension()
    
    def encode_texts(self, texts, batch_size=32):
        """テキストのリストをベクトルに変換"""
        embeddings = self.model.encode(
            texts,
            batch_size=batch_size,
            show_progress_bar=True,
            convert_to_numpy=True
        )
        return embeddings
    
    def encode_single_text(self, text):
        """単一のテキストをベクトルに変換"""
        embedding = self.model.encode(text, convert_to_numpy=True)
        return embedding

📝 ベクトルデータベースの構築

FAISSを使用した高速検索インデックス:

import faiss
import pickle
import os
from typing import List, Dict
class VectorDatabase:
    def __init__(self, embedding_dimension: int, index_file: str = "vector_index.faiss"):
        self.embedding_dimension = embedding_dimension
        self.index_file = index_file
        self.metadata_file = index_file.replace(".faiss", "_metadata.pkl")
        
        # FAISSインデックスの初期化
        self.index = faiss.IndexFlatL2(embedding_dimension)
        self.metadata = []  # チャンクのメタデータを保存
    
    def add_vectors(self, embeddings: np.ndarray, metadata_list: List[Dict]):
        """ベクトルとメタデータを追加"""
        # FAISSインデックスにベクトルを追加
        self.index.add(embeddings.astype('float32'))
        
        # メタデータを保存
        self.metadata.extend(metadata_list)
        
        print(f"追加されたベクトル数: {len(embeddings)}")
        print(f"総ベクトル数: {self.index.ntotal}")
    
    def search(self, query_embedding: np.ndarray, k: int = 5) -> List[Dict]:
        """類似ベクトルを検索"""
        # クエリベクトルの正規化
        query_embedding = query_embedding.astype('float32')
        if len(query_embedding.shape) == 1:
            query_embedding = query_embedding.reshape(1, -1)
        
        # 検索実行
        distances, indices = self.index.search(query_embedding, k)
        
        results = []
        for i, (distance, idx) in enumerate(zip(distances[0], indices[0])):
            if idx < len(self.metadata):
                result = self.metadata[idx].copy()
                result['similarity_score'] = 1.0 / (1.0 + distance)  # 類似度スコア
                result['rank'] = i + 1
                results.append(result)
        
        return results
    
    def save(self):
        """インデックスとメタデータを保存"""
        # FAISSインデックスを保存
        faiss.write_index(self.index, self.index_file)
        
        # メタデータを保存
        with open(self.metadata_file, 'wb') as f:
            pickle.dump(self.metadata, f)
        
        print(f"インデックスを保存しました: {self.index_file}")
        print(f"メタデータを保存しました: {self.metadata_file}")
    
    def load(self):
        """インデックスとメタデータを読み込む"""
        if os.path.exists(self.index_file):
            self.index = faiss.read_index(self.index_file)
            print(f"インデックスを読み込みました: {self.index_file}")
        
        if os.path.exists(self.metadata_file):
            with open(self.metadata_file, 'rb') as f:
                self.metadata = pickle.load(f)
            print(f"メタデータを読み込みました: {self.metadata_file}")

📝 Claude API連携クラス

from anthropic import Anthropic
from typing import List, Dict

class ClaudeRAG:
    def __init__(self, api_key: str, model_name: str = "claude-3-sonnet-20240229"):
        self.client = Anthropic(api_key=api_key)
        self.model_name = model_name
        self.max_tokens = 1000
        self.temperature = 0.7
        
        # システムプロンプトの設定
        self.system_prompt = """あなたは高度なAIアシスタントです。提供された文脈情報を基に、ユーザーの質問に正確で役立つ回答を生成してください。

重要なガイドライン:
1. 提供された文脈情報のみを使用して回答してください
2. 推測や憶測を避け、文脈に基づく事実のみを述べてください
3. 回答は簡潔で理解しやすく、必要に応じて構造化してください
4. 日本語で自然な回答を提供してください
5. 文脈に情報が不足している場合は、その旨を明確に伝えてください"""
    
    def generate_response(self, query: str, search_results: List[Dict]) -> Dict:
        """検索結果を基にClaudeで回答を生成"""
        
        if not search_results:
            return {
                'answer': '申し訳ございません。質問に関連する情報が見つかりませんでした。',
                'sources': [],
                'confidence': 0.0
            }
        
        # コンテキストの構築
        context_parts = []
        sources = []
        
        for i, result in enumerate(search_results, 1):
            context_parts.append(f"[文脈{i}] {result['text']}")
            sources.append({
                'source': result.get('source', 'unknown'),
                'similarity_score': result.get('similarity_score', 0.0),
                'chunk_id': result.get('chunk_id', None)
            })
        
        context = "\n\n".join(context_parts)
        
        # プロンプトの構築
        prompt = f"""
以下の文脈情報を基に、ユーザーの質問に回答してください。

{context}

質問: {query}

回答の際は以下の形式に従ってください:
1. 直接的な回答
2. 必要に応じて補足説明
3. 情報源の明示(可能な場合)
"""
        
        try:
            # Claude APIを呼び出す
            response = self.client.messages.create(
                model=self.model_name,
                max_tokens=self.max_tokens,
                temperature=self.temperature,
                system=self.system_prompt,
                messages=[{
                    "role": "user",
                    "content": prompt
                }]
            )
            
            answer = response.content[0].text
            
            # 自信度の計算(類似度スコアの平均)
            confidence = sum(r.get('similarity_score', 0.0) for r in search_results) / len(search_results)
            
            return {
                'answer': answer,
                'sources': sources,
                'confidence': confidence,
                'model': self.model_name,
                'tokens_used': response.usage.input_tokens + response.usage.output_tokens
            }
            
        except Exception as e:
            return {
                'answer': f'回答の生成中にエラーが発生しました: {str(e)}',
                'sources': [],
                'confidence': 0.0,
                'error': str(e)
            }

📝 RAGパイプラインの実装

class ClaudeRAGPipeline:
    def __init__(self, api_key: str, vector_system):
        self.vector_system = vector_system
        self.claude_rag = ClaudeRAG(api_key)
        self.conversation_history = []
        self.max_history = 10
    
    def ask(self, question: str, k: int = 5) -> Dict:
        """質問に対して回答を生成"""
        
        # 検索実行
        search_results = self.vector_system.search(question, k=k)
        
        if not search_results:
            return {
                'answer': '申し訳ございません。質問に関連する情報が見つかりませんでした。',
                'sources': [],
                'confidence': 0.0
            }
        
        # Claude APIで回答生成
        response = self.claude_rag.generate_response(question, search_results)
        
        # 会話履歴に追加
        self.conversation_history.append({
            'question': question,
            'response': response,
            'timestamp': time.time()
        })
        
        # 履歴の上限管理
        if len(self.conversation_history) > self.max_history:
            self.conversation_history.pop(0)
        
        return response

📝 統合したベクトル化システム

class VectorizationSystem:
    def __init__(self, embedding_model_name='sentence-transformers/paraphrase-multilingual-mpnet-base-v2'):
        self.embedding_model = EmbeddingModel(embedding_model_name)
        self.vector_db = VectorDatabase(
            embedding_dimension=self.embedding_model.embedding_dimension
        )
        self.is_initialized = False
    
    def initialize_from_chunks(self, chunks: List[Dict]):
        """チャンクからベクトルデータベースを構築"""
        if not chunks:
            raise ValueError("チャンクが提供されていません")
        
        print("チャンクをベクトル化中...")
        
        # テキストを抽出
        texts = [chunk['text'] for chunk in chunks]
        
        # ベクトル化
        embeddings = self.embedding_model.encode_texts(texts)
        
        # メタデータの準備
        metadata_list = []
        for i, chunk in enumerate(chunks):
            metadata = {
                'chunk_id': chunk.get('chunk_id', i),
                'text': chunk['text'],
                'source': chunk.get('source', 'unknown'),
                'source_file': chunk.get('source_file', 'unknown'),
                'length': len(chunk['text'])
            }
            metadata_list.append(metadata)
        
        # ベクトルデータベースに追加
        self.vector_db.add_vectors(embeddings, metadata_list)
        
        self.is_initialized = True
        print(f"ベクトル化完了: {len(chunks)} チャンク")
    
    def search(self, query: str, k: int = 5) -> List[Dict]:
        """クエリに類似するチャンクを検索"""
        if not self.is_initialized:
            raise RuntimeError("システムが初期化されていません")
        
        # クエリをベクトル化
        query_embedding = self.embedding_model.encode_single_text(query)
        
        # 検索実行
        search_results = self.vector_db.search(query_embedding, k)
        
        return search_results
    
    def save_system(self, save_dir: str = "vector_system"):
        """システムを保存"""
        os.makedirs(save_dir, exist_ok=True)
        
        # ベクトルデータベースを保存
        index_path = os.path.join(save_dir, "vector_index.faiss")
        self.vector_db.index_file = index_path
        self.vector_db.save()
        
        print(f"システムを保存しました: {save_dir}")

📝 メインプログラム

import os
from dotenv import load_dotenv

def main():
    # 環境変数の読み込み
    load_dotenv()
    api_key = os.getenv("ANTHROPIC_API_KEY")
    
    # ステップ1: ドキュメントの読み込みとチャンク分割
    print("=== ステップ1: ドキュメント読み込み ===")
    chunks = prepare_documents("documents/")
    
    # ステップ2: ベクトルシステムの初期化
    print("\n=== ステップ2: ベクトル化 ===")
    vector_system = VectorizationSystem()
    vector_system.initialize_from_chunks(chunks)
    
    # ステップ3: システムを保存
    print("\n=== ステップ3: システム保存 ===")
    vector_system.save_system("my_rag_system")
    
    # ステップ4: RAGパイプラインの初期化
    print("\n=== ステップ4: RAGパイプライン初期化 ===")
    rag_pipeline = ClaudeRAGPipeline(api_key, vector_system)
    
    # ステップ5: インタラクティブな質問応答
    print("\n=== ステップ5: 質問応答開始 ===")
    print("質問を入力してください(終了: q)\n")
    
    while True:
        question = input("質問: ")
        
        if question.lower() == 'q':
            break
        
        # 回答を生成
        result = rag_pipeline.ask(question, k=3)
        
        # 結果を表示
        print(f"\n回答: {result['answer']}")
        print(f"\n自信度: {result['confidence']:.2%}")
        print(f"使用トークン数: {result.get('tokens_used', 'N/A')}")
        
        if result['sources']:
            print("\n情報源:")
            for i, source in enumerate(result['sources'], 1):
                print(f"  {i}. {source['source']} (類似度: {source['similarity_score']:.2%})")
        
        print("\n" + "="*50 + "\n")

if __name__ == "__main__":
    main()

📝 従来の問題点と解決策

従来のRAGでは、チャンク単位でテキストを分割する際、各チャンクが元の文脈から切り離されてしまいます。

例えば、「この製品の価格は変更されました」という文だけを見た場合、どの製品なのか、いつ変更されたのか、新しい価格はいくらなのかが不明です。

📝 Contextual Retrievalの実装

class ContextualChunker:
    def __init__(self, context_window=200):
        self.context_window = context_window
    
    def create_contextual_chunks(self, text, filename=""):
        """文脈を考慮したチャンクを作成"""
        lines = text.split('\n')
        chunks = []
        
        for i, line in enumerate(lines):
            if self._is_semantically_significant(line):
                # 前後の文脈を取得
                context = self._get_surrounding_context(lines, i)
                
                # 文脈情報を付加
                contextual_text = self._add_contextual_information(
                    line, context, filename
                )
                
                chunks.append({
                    'id': len(chunks),
                    'text': line,
                    'contextual_text': contextual_text,
                    'context': context,
                    'position': i,
                    'metadata': {
                        'type': self._identify_content_type(line),
                        'relevance_score': self._calculate_relevance_score(line)
                    }
                })
        
        return chunks
    
    def _get_surrounding_context(self, lines, index):
        """周囲の文脈を取得"""
        start = max(0, index - self.context_window)
        end = min(len(lines), index + self.context_window + 1)
        
        return {
            'before': lines[start:index],
            'after': lines[index+1:end],
            'section': self._identify_section(lines, index)
        }
    
    def _add_contextual_information(self, text, context, filename):
        """文脈情報を付加"""
        contextual_info = []
        
        # ファイル情報
        if filename:
            contextual_info.append(f"ファイル: {filename}")
        
        # セクション情報
        if context['section']:
            contextual_info.append(f"セクション: {context['section']}")
        
        # 前後の参照情報
        if context['before']:
            contextual_info.append(f"前文脈: {' '.join(context['before'][-3:])}")
        
        if context['after']:
            contextual_info.append(f"後文脈: {' '.join(context['after'][:3])}")
        
        # 内容のタイプ
        content_type = self._identify_content_type(text)
        contextual_info.append(f"内容タイプ: {content_type}")
        
        # 全文を構築
        full_context = " | ".join(contextual_info)
        return f"{full_context}\n\n{text}"

📝 ハイブリッド検索の実装

from rank_bm25 import BM25Okapi
import re
from typing import List, Dict

class HybridSearch:
    def __init__(self, alpha: float = 0.7):
        self.alpha = alpha  # セマンティック検索の重み
        self.bm25 = None
        self.corpus = []
        self.vector_system = None
    
    def initialize_bm25(self, texts: List[str]):
        """BM25検索エンジンの初期化"""
        # テキストをトークン化
        tokenized_corpus = [self._tokenize(text) for text in texts]
        
        # BM25インデックスを構築
        self.bm25 = BM25Okapi(tokenized_corpus)
        self.corpus = texts
    
    def _tokenize(self, text: str) -> List[str]:
        """日本語テキストをトークン化"""
        # 全角・半角の統一
        text = text.lower().strip()
        
        # 記号を除去
        text = re.sub(r'[^\w\s]', '', text)
        
        # スペースで分割
        tokens = text.split()
        
        # ストップワードを除去
        stop_words = ['の', 'に', 'は', 'を', 'が', 'で', 'て', 'と', 'し', 'れ', 'さ']
        tokens = [token for token in tokens if token not in stop_words]
        
        return tokens
    
    def search(self, query: str, k: int = 10) -> List[Dict]:
        """ハイブリッド検索を実行"""
        
        # セマンティック検索
        semantic_results = self.vector_system.search(query, k=k*2)
        
        # BM25検索
        bm25_results = self._search_bm25(query, k=k*2)
        
        # 結果を統合
        combined_results = self._merge_results(
            semantic_results, bm25_results, k=k
        )
        
        return combined_results
    
    def _search_bm25(self, query: str, k: int) -> List[Dict]:
        """BM25検索を実行"""
        if not self.bm25:
            return []
        
        # クエリをトークン化
        query_tokens = self._tokenize(query)
        
        # BM25スコアを計算
        scores = self.bm25.get_scores(query_tokens)
        
        # 上位k件を取得
        top_indices = sorted(
            range(len(scores)), 
            key=lambda i: scores[i], 
            reverse=True
        )[:k]
        
        # 結果をフォーマット
        results = []
        for rank, idx in enumerate(top_indices, 1):
            results.append({
                'text': self.corpus[idx],
                'score': scores[idx],
                'rank': rank,
                'type': 'bm25',
                'index': idx
            })
        
        return results
    
    def _merge_results(self, semantic_results: List[Dict], 
                      bm25_results: List[Dict], k: int) -> List[Dict]:
        """検索結果を統合"""
        
        # スコアを正規化
        semantic_scores = self._normalize_scores(
            [r['similarity_score'] for r in semantic_results]
        )
        bm25_scores = self._normalize_scores(
            [r['score'] for r in bm25_results]
        )
        
        # 統合スコアを計算
        combined_scores = {}
        
        # セマンティック検索結果
        for i, result in enumerate(semantic_results):
            key = result['text'][:100]  # テキストのハッシュ
            combined_scores[key] = {
                'data': result,
                'semantic_score': semantic_scores[i] * self.alpha
            }
        
        # BM25検索結果
        for i, result in enumerate(bm25_results):
            key = result['text'][:100]
            if key in combined_scores:
                combined_scores[key]['bm25_score'] = bm25_scores[i] * (1 - self.alpha)
            else:
                combined_scores[key] = {
                    'data': result,
                    'bm25_score': bm25_scores[i] * (1 - self.alpha)
                }
        
        # 最終スコアを計算
        final_results = []
        for key, scores in combined_scores.items():
            total_score = scores.get('semantic_score', 0.0) + scores.get('bm25_score', 0.0)
            result = scores['data'].copy()
            result['combined_score'] = total_score
            result['search_type'] = 'hybrid'
            final_results.append(result)
        
        # スコアでソート
        final_results.sort(key=lambda x: x['combined_score'], reverse=True)
        
        return final_results[:k]
    
    def _normalize_scores(self, scores: List[float]) -> List[float]:
        """スコアを0-1に正規化"""
        if not scores:
            return []
        
        min_score = min(scores)
        max_score = max(scores)
        
        if max_score == min_score:
            return [1.0] * len(scores)
        
        return [(s - min_score) / (max_score - min_score) for s in scores]

📝 Cross-Encoderを使用したReranking

from sentence_transformers import CrossEncoder
import numpy as np
from typing import List, Dict

class Reranker:
    def __init__(self, model_name: str = 'cross-encoder/ms-marco-MiniLM-L-6-v2'):
        self.model = CrossEncoder(model_name)
        self.rerank_threshold = 0.3
    
    def rerank(self, query: str, documents: List[str], 
               top_k: int = 5) -> List[Dict]:
        """検索結果を再ランキング"""
        
        if not documents:
            return []
        
        # Query-Documentペアを作成
        pairs = [[query, doc] for doc in documents]
        
        # スコアを計算
        scores = self.model.predict(pairs)
        
        # 結果をフォーマット
        reranked_results = []
        for i, (doc, score) in enumerate(zip(documents, scores)):
            reranked_results.append({
                'text': doc,
                'rerank_score': float(score),
                'original_rank': i + 1,
                'significant': score > self.rerank_threshold
            })
        
        # スコアでソート
        reranked_results.sort(key=lambda x: x['rerank_score'], reverse=True)
        
        return reranked_results[:top_k]
    
    def rerank_with_diversity(self, query: str, documents: List[str], 
                              top_k: int = 5, diversity_weight: float = 0.2) -> List[Dict]:
        """多様性を考慮したReranking"""
        
        # 通常のReranking
        base_results = self.rerank(query, documents, top_k=len(documents))
        
        # 多様性スコアを計算
        for i, result in enumerate(base_results):
            diversity_score = self._calculate_diversity_score(
                result, base_results[:i]
            )
            
            # 最終スコアを計算
            result['final_score'] = (
                (1 - diversity_weight) * result['rerank_score'] +
                diversity_weight * diversity_score
            )
        
        # 最終スコアでソート
        base_results.sort(key=lambda x: x['final_score'], reverse=True)
        
        return base_results[:top_k]

📝 LLMベースのReranking

より高度なRerankingには、Claude APIを使用します:

class LLMReranker:
    def __init__(self, claude_rag_instance):
        self.claude_rag = claude_rag_instance
    
    def rerank_with_llm(self, query: str, documents: List[str], 
                       top_k: int = 5) -> List[Dict]:
        """LLMを使用して結果を再ランキング"""
        
        reranked_results = []
        
        for i, doc in enumerate(documents):
            # 各ドキュメントの関連性を評価
            evaluation_prompt = f"""
            以下の質問とドキュメントを分析し、関連性を0から10のスケールで評価してください。

            質問: {query}

            ドキュメント: {doc[:500]}...  # 最初の500文字を使用

            評価基準:
            - 10: 質問に対する直接的で包括的な回答
            - 7-9: 質問に関連する有用な情報を含む
            - 4-6: 部分的に関連する情報
            - 1-3: ほとんど関連しない
            - 0: 完全に無関係

            回答は「関連性スコア: [数字]」の形式で提供してください。
            """
            
            try:
                response = self.claude_rag.client.messages.create(
                    model="claude-3-haiku-20240307",  # 高速なモデルを使用
                    max_tokens=50,
                    messages=[{"role": "user", "content": evaluation_prompt}]
                )
                
                # スコアを抽出
                score_text = response.content[0].text
                score = self._extract_score(score_text)
                
                reranked_results.append({
                    'text': doc,
                    'llm_score': score,
                    'original_rank': i + 1,
                    'evaluation': score_text
                })
                
            except Exception as e:
                print(f"LLM評価エラー: {e}")
                # エラー時は中程度のスコアを割り当て
                reranked_results.append({
                    'text': doc,
                    'llm_score': 5.0,
                    'original_rank': i + 1,
                    'evaluation': '評価エラー'
                })
        
        # LLMスコアでソート
        reranked_results.sort(key=lambda x: x['llm_score'], reverse=True)
        
        return reranked_results[:top_k]

📝 動的チャンクサイズの実装

状況に応じてチャンクサイズを調整する高度な実装:

class AdaptiveChunker:
    def __init__(self, base_size=500, min_size=200, max_size=1000):
        self.base_size = base_size
        self.min_size = min_size
        self.max_size = max_size
    
    def chunk_by_content_type(self, text: str, content_type: str) -> List[Dict]:
        """コンテンツタイプに応じて動的にチャンクサイズを調整"""
        
        if content_type == 'technical':
            chunk_size = self.base_size  # 技術文書は標準サイズ
        elif content_type == 'legal':
            chunk_size = self.base_size * 0.7  # 法律文書は小さいサイズ
        elif content_type == 'manual':
            chunk_size = self.base_size * 1.2  # マニュアルは大きいサイズ
        elif content_type == 'news':
            chunk_size = self.base_size * 0.8  # ニュースは中程度
        else:
            chunk_size = self.base_size
        
        # 範囲内に制限
        chunk_size = max(self.min_size, min(self.max_size, int(chunk_size)))
        
        return self._chunk_with_size(text, chunk_size)

📝 Recall@Kの実装

class EvaluationMetrics:
    def __init__(self):
        self.results = []
    
    def calculate_recall_at_k(self, retrieved: List[str], 
                            relevant: List[str], k: int) -> float:
        """Recall@Kを計算"""
        
        if not relevant:
            return 0.0
        
        # 上位K件を取得
        retrieved_at_k = retrieved[:k]
        
        # 関連アイテムの数を数える
        relevant_retrieved = sum(1 for item in retrieved_at_k 
                               if item in relevant)
        
        recall = relevant_retrieved / len(relevant)
        
        return recall
    
    def calculate_precision_at_k(self, retrieved: List[str], 
                               relevant: List[str], k: int) -> float:
        """Precision@Kを計算"""
        
        if k == 0:
            return 0.0
        
        retrieved_at_k = retrieved[:k]
        relevant_retrieved = sum(1 for item in retrieved_at_k 
                               if item in relevant)
        
        precision = relevant_retrieved / k
        
        return precision
    
    def calculate_f1_at_k(self, retrieved: List[str], 
                         relevant: List[str], k: int) -> float:
        """F1@Kを計算"""
        
        recall = self.calculate_recall_at_k(retrieved, relevant, k)
        precision = self.calculate_precision_at_k(retrieved, relevant, k)
        
        if (recall + precision) == 0:
            return 0.0
        
        f1 = 2 * (recall * precision) / (recall + precision)
        
        return f1
    
    def calculate_mean_reciprocal_rank(self, 
                                     query_results: List[Tuple[str, List[str]]],
                                     relevance_judgments: Dict[str, List[str]]) -> float:
        """Mean Reciprocal Rank (MRR)を計算"""
        
        reciprocal_ranks = []
        
        for query, retrieved in query_results:
            relevant_docs = relevance_judgments.get(query, [])
            
            if not relevant_docs:
                continue
            
            # 最初の関連アイテムの位置を見つける
            for rank, item in enumerate(retrieved, 1):
                if item in relevant_docs:
                    reciprocal_ranks.append(1.0 / rank)
                    break
            else:
                # 関連アイテムが見つからない場合
                reciprocal_ranks.append(0.0)
        
        if not reciprocal_ranks:
            return 0.0
        
        mrr = sum(reciprocal_ranks) / len(reciprocal_ranks)
        
        return mrr