ベクトルDBなしでRAG構築!Gemini File SearchでWikipedia検索システムを作る方法【ソースコード付】

この記事は以下のエントリの継続的な記事になっています。

WikipediaAPIから情報を取得する方法

uepon.hatenadiary.com

ChromaDB+GeminiAPIでRAGシステムを構築する方法

uepon.hatenadiary.com


2025年11月にリリースされたGoogle Gemini APIFile Search Tool(以降Gemini File Searchとします)を使用して、先日作成した「Wikipedia記事を検索できるRAGシステムの改良」を行ってみました。

参考

blog.google

⚠️ 本記事の情報について

この記事は2025年11月時点での情報に基づいていますが、Gemini APIの機能・料金・制限は予告なく変更される可能性があります。最新の公式ドキュメントもご確認ください。


Gemini File Searchの特徴

Gemini File Searchには以下のような特徴があります。

特徴 説明
高度な自動化 データ取り込みから検索結果の注入まで多くを自動化😊
自動インデックス作成 埋め込み生成とチャンク分割が自動で実行される
シンプルな料金体系 インデックス作成時に料金が発生
永続ストレージ対応 File Search Storeを使用することでデータを永続化可能
自動引用機能 回答に使用されたドキュメントの引用元を自動表示🤩

これだけみてもわかりにくいと思うので従来のRAGシステムを比較してみると以下のようなところですかね。

  • テキストのチャンク分割
  • ベクトルデータベースの構築(ChromaDB等)
  • Embedding生成の実装
  • 検索ロジックの実装

上記の処理が不要となり、File Search StoreにファイルをアップロードするだけでRAGシステムが構築できる点が最大のメリットとなります。では、具体的にWikipedia記事を使ったRAGシステムを構築してみようと思いますが、まずは簡単なサンプルをつかって内容を確認していきます。

Gemini File Searchを使用した基本的なRAGシステムの構築

まずは使い方とみるということでシンプルな例を作成してみます

プロジェクトの仮想環境の構築

方法1: uv を使用した環境構築(推奨)

uvがインストールされていること (インストール方法)を確認してください。

セットアップ手順

# プロジェクトディレクトリの作成と移動
$ mkdir -p wikipedia-rag-filesearch/sample
$ cd wikipedia-rag-filesearch

# uvプロジェクトの初期化
$ uv init

# 仮想環境の作成
$ uv venv

# 仮想環境の有効化
$ source .venv/bin/activate  # WSL/Linux/Mac

# 必要なパッケージのインストール
$ uv pip install google-genai python-dotenv tqdm

# サンプル用のディレクトリに移動
$ cd sample 

方法2: 標準のpipを使用した環境構築

セットアップ手順

# プロジェクトディレクトリの作成と移動
$ mkdir -p wikipedia-rag-filesearch/sample
$ cd wikipedia-rag-filesearch

# 仮想環境の作成
$ python -m venv .venv

# 仮想環境の有効化
$ source .venv/bin/activate  # WSL/Linux/Mac

# 必要なパッケージのインストール
$ pip install google-genai python-dotenv tqdm

# サンプル用のディレクトリに移動
$ cd sample

環境変数の設定

今回はAPIキーを.envファイルで管理します。

Githubなどでは.envファイルをコミットしないように注意してください。.gitignoreに追加することを強く推奨します。

1. .envを作成

GithubリポジトリからCloneした場合は.env.sample以下でコピーしてください。新規作成している場合は そのまま 2. の手順へ進んでください。

$ cp .env.sample .env

2. .envファイルを編集

お好みのエディタで.envファイルを開き、実際のAPIキーを設定してください。

GOOGLE_API_KEY=実際のAPIキーをここに入力

APIキーの取得方法については以下を参照してください。

Google AI StudioAPIキーを生成


システムの基本構成

Gemini File Search RAG システムサンプル
sample
├── .env                               # APIキーやモデル名などの格納
├── file_search_sample.py   # RAGシステムのコア
└── sample.md                    # Wikipedia記事(Markdown)

シンプルな機能実装

まずは、機能の簡単な実装例を示してみようと思います。 各ステップを結合したプログラムは、まとめたものを使用してください。

【ステップ1】File Search Storeの作成(抜粋)

File Search Storeは、アップロードしたファイルを保存・管理するためのストレージ領域となります。 以下のコードはStoreを作成します。

# Storeの作成
print("Creating file search store...")
store = client.file_search_stores.create(
    config={'display_name': 'wikipedia-knowledge-base'}
)
print(f"Store created: {store.name}")

【ステップ2】 ファイルのアップロード(抜粋)

Storeが作成できたら、ファイルをアップロードします。以下のコードはMarkdownファイルをアップロードしています。

# sample.mdをアップロード
file_path = "sample.md"

if os.path.exists(file_path):
    print(f"\nUploading {file_path}...")
    
    # ファイルをアップロード
    operation = client.file_search_stores.upload_to_file_search_store(
        file_search_store_name=store.name,
        file=file_path
    )
    
    # 完了待機
    while not operation.done:
        print("Waiting for upload to complete...")
        time.sleep(1)
        operation = client.operations.get(operation)
    
    print(f"Upload completed: {operation.result()}")

【ステップ3】 質問応答(抜粋)

アップロードが完了したら、実際に質問応答を試してみます。以下のコードはFile Searchを使った質問応答の例です。変数であるqueryの文字列が質問文となります。

# File Searchを使った質問応答
query = "作家がAnthropicを提訴した訴訟の判決内容を教えてください"
print(f"\nQuestion: {query}")
print("\nGenerating answer...")

response = client.models.generate_content(
    model=model_name,
    contents=query,
    config=types.GenerateContentConfig(
        tools=[
            types.Tool(
                file_search=types.FileSearch(
                    file_search_store_names=[store.name]
                )
            )
        ],
        temperature=0.7
    )
)

print(f"\nAnswer:\n{response.text}")

各ステップを組み合わせた完全なコード例

今回の例の場合、上記の各ステップを組み合わせた完全なコードは以下のようになります。 ※抜粋のコードでは、ファイルのロードをするとStoreが毎回新規作成されてしまうため、コストを下げるために初回の作成したStoreを使うように変更しました。

完全なサンプルコード

.env.sample

# Google Gemini APIキー
# APIキーの取得先: https://aistudio.google.com/app/apikey
GOOGLE_API_KEY=your_google_api_key_here

# Geminiモデル名
GEMINI_MODEL=models/gemini-2.5-pro

# File Search Store名(オプション - 初回実行時は空のまま)
# 初回実行後、ここにStore名をコピーすると既存Storeを再利用してコストを節約できます
# 初回実行時に表示されるStore名をコピーしてください
# 例)STORE_NAME=fileSearchStores/wikipediaknowledgebase-abc123xyz456
STORE_NAME=

表示の例

file_search_sample.py

import os
import time
from dotenv import load_dotenv
from google import genai
from google.genai import types

# .envファイルから環境変数を読み込む
load_dotenv()

# クライアントの作成
client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY"))
model_name = os.getenv("GEMINI_MODEL", "models/gemini-2.5-pro")
store_name = os.getenv("STORE_NAME")  # 既存のStore名(あれば)

# 既存Storeを使用するか、新規作成するか
if store_name:
    print(f"Using existing store: {store_name}")
    # 既存Storeを使用(Store objectを取得)
    # Note: 既存Store名を使ってそのまま利用
    class ExistingStore:
        def __init__(self, name):
            self.name = name
    
    store = ExistingStore(store_name)
else:
    # 新しいStoreを作成
    print("Creating new file search store...")
    store = client.file_search_stores.create(
        config={'display_name': 'wikipedia-knowledge-base'}
    )
    print(f"Store created: {store.name}")
    print("\n" + "="*70)
    print("To reuse this store and save costs, add this to your .env file:")
    print(f"STORE_NAME={store.name}")
    print("="*70 + "\n")

# sample.mdをアップロード
file_path = "sample.md"

if os.path.exists(file_path):
    print(f"\nUploading {file_path}...")
    
    # ファイルをアップロード
    operation = client.file_search_stores.upload_to_file_search_store(
        file_search_store_name=store.name,
        file=file_path
    )
    
    # 完了待機
    while not operation.done:
        print("Waiting for upload to complete...")
        time.sleep(1)
        operation = client.operations.get(operation)
    
    print("Upload completed successfully!")
else:
    print(f"Error: {file_path} not found in current directory")
    exit(1)

# File Searchを使った質問応答
print("\n" + "="*50)
print("File Search Question Answering Demo")
print("="*50)

query = "作家がAnthropicを提訴した訴訟の判決内容を教えてください"
print(f"\nQuestion: {query}")
print("\nGenerating answer...")

response = client.models.generate_content(
    model=model_name,
    contents=query,
    config=types.GenerateContentConfig(
        tools=[
            types.Tool(
                file_search=types.FileSearch(
                    file_search_store_names=[store.name]
                )
            )
        ],
        temperature=0.7
    )
)

print(f"\nAnswer:\n{response.text}")

サンプルコードの実行結果

$ python file_search_sample.py
Using existing store: fileSearchStores/wikipediaknowledgebase-0kfw8lcj92bh

Uploading sample.md...
Waiting for upload to complete...
Upload completed successfully!

==================================================
File Search Question Answering Demo
==================================================

Question: 作家がAnthropicを提訴した訴訟の判決内容を教えてください

Generating answer...

Answer:
作家のアンドレア・バーツ、チャールズ・グレーバー、カーク・ジョンソンが、AI開発会社Anthropicを相手取って起こした著作権侵害訴訟において、カリフォルニア州北部地区連邦地方裁判所は一部の行為についてフェアユースを認め、一部については認めないという判決を下しました。

**訴訟の背景**
作家たちは、AnthropicがAIモデル「Claude」を訓練する際に、海賊版サイトであるLibGenやBooks3のデータ、および物理的な書籍をスキャンしたデータを使用したことが著作権侵害にあたると主張しました。 これに対しAnthropicは、海賊版サイトの使用や数百万冊の書籍をデジタル化したデータの使用を認めた上で、これらの行為はフェアユースに該当すると反論しました。

**判決内容**
裁判所は、訓練データの種類によって異なる判断を示しました。

*   **物理書籍のスキャンデータによる訓練:フェアユースを認定**
    物理的な書籍をスキャンして得たデータをAIの訓練に使用した点については、フェアユースであると認められました。 裁判所は、この訓練が全く新しい文章を生成するための統計的関係を学習する目的であり、AIの生成物が元の書籍のコピーや盗作をユーザーに提供するものではないと判断しました。

*   **海賊版データによるデータセット作成:フェアユースを認めず**
    一方で、海賊版サイトから入手したデータを用いてデータセットを作成した行為については、フェアユースとは認められませんでした。 この行為は有料のコピーを代替するものであり、変容的な利用ではないと判断されました。

Gemini File Searchを使用した応用的なRAGシステムの構築(以前作成したRAGの変更案)

簡単な実行例の作成が終わったので、以前作成したRAGシステムのコードを改良して、Gemini File Searchを使用するように変更しました。

参考

uepon.hatenadiary.com

File Search版への改良

先日作成した、ChromaDBを使用したRAGシステムは、ベクトルデータベースの構築や埋め込み生成など、多くの手動処理が必要でしたが、Gemini File Searchを使用して改良します。

主な機能の違い

項目 従来版(ChromaDB 改良版(File Search
ベクトルDB管理 手動でChromaDBを管理 File Search Storeが自動管理
埋め込み生成 手動で実装が必要 自動で実行される
チャンク分割 手動で設定が必要 自動で最適化される
引用機能 手動で実装が必要 自動で引用元を表示
データ永続化 ローカルのchroma_db/ File Search Storeクラウド
料金体系 無料(ローカル) インデックス作成時に課金
セットアップ Chromadb等の設定が必要 google-genaiのみ

File Search版の実装

システム構成

プロジェクトのファイル構造は大きく変えないようにしています。

wikipedia-rag-filesearch/
├── .env                                     # API設定ファイル
├── rag_system_filesearch.py    # File Search版RAGシステム
├── data_loader_filesearch.py   # File Search版データローダー
├── test_rag_filesearch.py         # テスト・デモ
├── data/
│   └── wikipedia/              # Wikipediaのmarkdownファイル
└── file_mappings.json          # ファイル名マッピング(自動生成)

主要な変更点

1. ベクトルDB管理の不要化

従来版では、ChromaDBのセットアップとコレクション管理が必要でしたが、File Search版ではFile Search Storeにファイルをアップロードするだけで完了します。

# 従来版(ChromaDB)
self.client = chromadb.PersistentClient(path="./chroma_db")
self.collection = self.client.get_or_create_collection(
    name="wikipedia_articles"
)

# 改良版(File Search)
store = client.file_search_stores.create(
    config={'display_name': 'wikipedia-knowledge-base'}
)

2. 日本語ファイル名の取り扱い

自作のWikipediaデータの取得ツールでは、全角記号などを含んだファイル名が生成されることもあるため、安全性のためにファイル名をASCII名に変換する仕組みを追加しました。

def safe_filename(filename):
    """日本語ファイル名を安全なASCII名に変換"""
    name, ext = os.path.splitext(filename)
    # ファイル名のハッシュ値を使用
    hash_name = hashlib.md5(name.encode('utf-8')).hexdigest()[:16]
    return f"wiki_{hash_name}{ext}"

3. ファイルマッピングの管理

元のファイル名と変換後のファイル名の対応をfile_mappings.jsonで管理します。 これは作成時にLLMに提案されました。

{
  "wiki_a1b2c3d4e5f6g7h8.md": {
    "original_filename": "機械学習.md",
    "title": "機械学習",
    "upload_date": "2024-11-15T10:30:00",
    "file_id": "files/abc123xyz"
  }
}

4. 質問応答の実装

File Searchを使った質問応答は、toolsパラメータで指定するだけで実現できます。

response = client.models.generate_content(
    model=model_name,
    contents=query,
    config=types.GenerateContentConfig(
        tools=[
            types.Tool(
                file_search=types.FileSearch(
                    file_search_store_names=[store_name]
                )
            )
        ],
        temperature=0.7
    )
)

環境構築(File Search版)

必要なパッケージ

# プロジェクトディレクトリの作成と移動
$ mkdir wikipedia-rag-filesearch
$ cd wikipedia-rag-filesearch

# uvプロジェクトの初期化
$ uv init
$ uv venv
$ source .venv/bin/activate

# 必要なパッケージのインストール
$ uv pip install google-genai python-dotenv tqdm

環境変数の設定

.envファイルの設定は先程のシンプルな機能実装と同じ構成になっています。ファイルをコピーすればOKです。

初回データアップロード後、コンソールに表示されるStore名を.envファイルのSTORE_NAMEに設定することで、2回目以降は既存のStoreを再利用でき、コストを削減できます。

# Google Gemini APIキー
GOOGLE_API_KEY=your_google_api_key_here
# Geminiモデル名
GEMINI_MODEL=models/gemini-2.5-pro
# File Search Store名(初回実行後に自動設定を推奨)
STORE_NAME=

ソースコード

data_loader_filesearch.py

# data_loader_filesearch.py
import os
import json
import time
import hashlib
import argparse
from pathlib import Path
from tqdm import tqdm
from google import genai
from google.genai import types
from dotenv import load_dotenv

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


def safe_filename(original_filename):
    """日本語ファイル名を安全なASCII名に変換
    
    Args:
        original_filename: 元のファイル名
        
    Returns:
        安全なASCII名のファイル名
    """
    name, ext = os.path.splitext(original_filename)
    # ファイル名のハッシュ値を使用
    hash_name = hashlib.md5(name.encode('utf-8')).hexdigest()[:16]
    return f"wiki_{hash_name}{ext}"


def load_file_mappings(mapping_file='file_mappings.json'):
    """ファイルマッピングを読み込み
    
    Args:
        mapping_file: マッピングファイルのパス
        
    Returns:
        マッピング辞書
    """
    if os.path.exists(mapping_file):
        with open(mapping_file, 'r', encoding='utf-8') as f:
            return json.load(f)
    return {}


def save_file_mappings(mappings, mapping_file='file_mappings.json'):
    """ファイルマッピングを保存
    
    Args:
        mappings: マッピング辞書
        mapping_file: マッピングファイルのパス
    """
    with open(mapping_file, 'w', encoding='utf-8') as f:
        json.dump(mappings, f, ensure_ascii=False, indent=2)


def get_or_create_store(client, store_name=None):
    """File Search Storeを取得または作成
    
    Args:
        client: Gemini APIクライアント
        store_name: 既存のStore名(Noneの場合は新規作成)
        
    Returns:
        Store object
    """
    if store_name:
        print(f"既存のStoreを使用: {store_name}")
        # 既存Storeを使用
        class ExistingStore:
            def __init__(self, name):
                self.name = name
        return ExistingStore(store_name)
    else:
        # 新しいStoreを作成
        print("新しいFile Search Storeを作成中...")
        store = client.file_search_stores.create(
            config={'display_name': 'wikipedia-knowledge-base'}
        )
        print(f"Store作成完了: {store.name}")
        print("\n" + "=" * 70)
        print("コスト削減のため、.envファイルに以下を追加してください:")
        print(f"STORE_NAME={store.name}")
        print("=" * 70 + "\n")
        return store


def delete_store_files(client, store_name):
    """Store内の全ファイルを削除
    
    Args:
        client: Gemini APIクライアント
        store_name: Store名
    """
    try:
        print("Store内のファイルを削除中...")
        files = client.file_search_stores.list_files(
            file_search_store_name=store_name
        )
        
        deleted_count = 0
        for file in files:
            try:
                client.files.delete(name=file.name)
                deleted_count += 1
            except Exception as e:
                print(f"ファイル削除エラー ({file.name}): {e}")
        
        print(f"{deleted_count}件のファイルを削除しました")
        
    except Exception as e:
        print(f"ファイル削除中にエラー: {e}")


def upload_wikipedia_data(data_dir, reset=False, mapping_file='file_mappings.json'):
    """WikipediaデータをFile Search Storeにアップロード
    
    Args:
        data_dir: データディレクトリ
        reset: 既存データをリセットするか
        mapping_file: マッピングファイルのパス
    """
    # クライアントの作成
    client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY"))
    store_name = os.getenv("STORE_NAME")
    
    # Storeの取得または作成
    if reset and store_name:
        confirm = input("既存のStoreをリセットしますか? (y/N): ")
        if confirm.lower() == 'y':
            # ファイルを削除
            delete_store_files(client, store_name)
            # マッピングもクリア
            save_file_mappings({}, mapping_file)
            print("データをリセットしました\n")
    
    store = get_or_create_store(client, store_name)
    
    # データディレクトリの存在確認
    data_path = Path(data_dir)
    if not data_path.exists():
        print(f"エラー: ディレクトリ '{data_dir}' が見つかりません")
        return
    
    # markdownファイルの読み込み
    md_files = list(data_path.glob("*.md"))
    
    if not md_files:
        print(f"{data_dir} にmarkdownファイルが見つかりません")
        return
    
    print(f"{len(md_files)}件のファイルをアップロードします...\n")
    
    # ファイルマッピングの読み込み
    mappings = load_file_mappings(mapping_file)
    
    # 各ファイルの処理
    success_count = 0
    error_count = 0
    
    for file_path in tqdm(md_files, desc="データアップロード中"):
        try:
            original_name = file_path.name
            ascii_name = safe_filename(original_name)
            
            # 一時ファイルとして保存(ASCII名)
            temp_path = file_path.parent / ascii_name
            
            # ファイルをコピー(ASCII名で)
            import shutil
            shutil.copy2(file_path, temp_path)
            
            try:
                # ファイルをアップロード
                operation = client.file_search_stores.upload_to_file_search_store(
                    file_search_store_name=store.name,
                    file=str(temp_path)
                )
                
                # 完了待機(タイムアウト: 60秒)
                timeout = 60
                start_time = time.time()
                while not operation.done:
                    if time.time() - start_time > timeout:
                        raise TimeoutError("アップロードがタイムアウトしました")
                    time.sleep(0.5)
                    operation = client.operations.get(operation)
                
                # マッピング情報を保存
                result = operation.result()
                mappings[ascii_name] = {
                    'original_filename': original_name,
                    'title': file_path.stem,
                    'upload_date': time.strftime('%Y-%m-%dT%H:%M:%S'),
                    'file_id': str(result) if result else 'N/A'
                }
                
                success_count += 1
                
            finally:
                # 一時ファイルを削除
                if temp_path.exists():
                    temp_path.unlink()
            
        except Exception as e:
            error_count += 1
            tqdm.write(f"\nエラー ({file_path.name}): {e}")
    
    # マッピングを保存
    save_file_mappings(mappings, mapping_file)
    
    # 結果の表示
    print(f"\n完了: {success_count}件のファイルをアップロードしました")
    if error_count > 0:
        print(f"エラー: {error_count}件のファイルで問題が発生しました")
    
    # Store内のファイル数を確認
    try:
        files = client.file_search_stores.list_files(
            file_search_store_name=store.name
        )
        file_count = sum(1 for _ in files)
        print(f"File Search Store総数: {file_count}件")
    except:
        print("File Search Store総数: 確認できませんでした")
    
    print(f"ファイルマッピングを保存しました: {mapping_file}")


def main():
    """メイン処理"""
    parser = argparse.ArgumentParser(
        description='Wikipedia記事をFile Search Storeにアップロード'
    )
    parser.add_argument(
        '--data-dir',
        default='./data/wikipedia',
        help='Wikipediaのmarkdownファイルが格納されているディレクトリ (デフォルト: ./data/wikipedia)'
    )
    parser.add_argument(
        '--reset',
        action='store_true',
        help='既存のデータをリセットしてからアップロード'
    )
    parser.add_argument(
        '--mapping-file',
        default='file_mappings.json',
        help='ファイルマッピングの保存先 (デフォルト: file_mappings.json)'
    )
    
    args = parser.parse_args()
    
    print("=== Wikipedia RAG File Search データローダー ===\n")
    print(f"データディレクトリ: {args.data_dir}\n")
    
    # データのアップロード
    upload_wikipedia_data(args.data_dir, args.reset, args.mapping_file)


if __name__ == "__main__":
    main()

rag_system_filesearch.py

# rag_system_filesearch.py
import os
from google import genai
from google.genai import types
from dotenv import load_dotenv

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

class WikipediaRAGFileSearch:
    """File Searchを使用したWikipedia RAGシステム"""
    
    def __init__(self, store_name=None):
        """RAGシステムの初期化
        
        Args:
            store_name: 既存のFile Search Store名(省略時は環境変数から取得)
        """
        # Gemini APIの設定
        self.client = genai.Client(api_key=os.getenv("GOOGLE_API_KEY"))
        self.model_name = os.getenv("GEMINI_MODEL", "models/gemini-2.5-pro")
        
        # Store名の取得(引数 > 環境変数)
        self.store_name = store_name or os.getenv("STORE_NAME")
        
        if not self.store_name:
            print("警告: STORE_NAMEが設定されていません。")
            print("data_loader_filesearch.pyでStoreを作成してください。")
    
    def generate_answer(self, query, temperature=0.7):
        """質問に対する回答を生成
        
        Args:
            query: 質問文
            temperature: 生成の創造性(0.0〜1.0)
            
        Returns:
            回答テキスト
        """
        if not self.store_name:
            return "エラー: File Search Storeが設定されていません。"
        
        try:
            # File Searchを使った回答生成
            response = self.client.models.generate_content(
                model=self.model_name,
                contents=query,
                config=types.GenerateContentConfig(
                    tools=[
                        types.Tool(
                            file_search=types.FileSearch(
                                file_search_store_names=[self.store_name]
                            )
                        )
                    ],
                    temperature=temperature
                )
            )
            
            # 引用情報の取得(あれば)
            answer_text = response.text
            
            # グラウンディングメタデータがあれば引用情報を追加
            if hasattr(response, 'candidates') and response.candidates:
                candidate = response.candidates[0]
                if hasattr(candidate, 'grounding_metadata'):
                    grounding = candidate.grounding_metadata
                    if hasattr(grounding, 'grounding_chunks') and grounding.grounding_chunks:
                        answer_text += "\n\n【引用元】\n"
                        for i, chunk in enumerate(grounding.grounding_chunks[:3], 1):
                            # チャンクから情報を抽出
                            if hasattr(chunk, 'web') and chunk.web:
                                answer_text += f"{i}. {chunk.web.uri}\n"
            
            return answer_text
            
        except Exception as e:
            return f"エラーが発生しました: {str(e)}"
    
    def get_store_info(self):
        """Store情報の取得
        
        Returns:
            Store情報の辞書
        """
        if not self.store_name:
            return {
                'store_name': None,
                'status': 'not_configured'
            }
        
        try:
            # Store情報を取得
            store = self.client.file_search_stores.get(self.store_name)
            
            return {
                'store_name': self.store_name,
                'display_name': store.display_name if hasattr(store, 'display_name') else 'N/A',
                'status': 'active'
            }
        except Exception as e:
            return {
                'store_name': self.store_name,
                'status': 'error',
                'error': str(e)
            }
    
    def list_files_in_store(self):
        """Store内のファイル一覧を取得
        
        Returns:
            ファイル情報のリスト
        """
        if not self.store_name:
            return []
        
        try:
            # Store内のファイルを一覧取得
            files = self.client.file_search_stores.list_files(
                file_search_store_name=self.store_name
            )
            
            file_list = []
            for file in files:
                file_list.append({
                    'name': file.name if hasattr(file, 'name') else 'N/A',
                    'display_name': file.display_name if hasattr(file, 'display_name') else 'N/A',
                    'size_bytes': file.size_bytes if hasattr(file, 'size_bytes') else 0,
                    'create_time': file.create_time if hasattr(file, 'create_time') else None
                })
            
            return file_list
            
        except Exception as e:
            print(f"ファイル一覧の取得中にエラー: {e}")
            return []


# 使用例
if __name__ == "__main__":
    # RAGシステムの初期化
    rag = WikipediaRAGFileSearch()
    
    # Store情報の表示
    print("=== Store情報 ===")
    store_info = rag.get_store_info()
    print(f"Store名: {store_info.get('store_name', 'N/A')}")
    print(f"表示名: {store_info.get('display_name', 'N/A')}")
    print(f"ステータス: {store_info.get('status', 'N/A')}")
    
    if store_info.get('status') == 'active':
        # ファイル一覧の表示
        print("\n=== Store内のファイル ===")
        files = rag.list_files_in_store()
        print(f"総ファイル数: {len(files)}件")
        
        # 質問応答のテスト
        print("\n=== 質問応答テスト ===")
        query = "作家がAnthropicを提訴した訴訟の判決内容を教えてください"
        print(f"質問: {query}")
        print("\n回答生成中...\n")
        
        answer = rag.generate_answer(query)
        print("【回答】")
        print(answer)
    else:
        print("\nStoreが設定されていないか、エラーが発生しています。")
        print("data_loader_filesearch.pyでデータをアップロードしてください。")

test_rag_filesearch.py

# test_rag_filesearch.py
import json
from rag_system_filesearch import WikipediaRAGFileSearch


def load_file_mappings(mapping_file='file_mappings.json'):
    """ファイルマッピングを読み込み
    
    Args:
        mapping_file: マッピングファイルのパス
        
    Returns:
        マッピング辞書
    """
    try:
        with open(mapping_file, 'r', encoding='utf-8') as f:
            return json.load(f)
    except FileNotFoundError:
        return {}


def test_qa():
    """質問応答のテスト"""
    rag = WikipediaRAGFileSearch()
    
    # Store情報の確認
    store_info = rag.get_store_info()
    if store_info.get('status') != 'active':
        print("\nエラー: File Search Storeが設定されていません")
        print("data_loader_filesearch.pyでデータをアップロードしてください")
        return
    
    query = input("\n質問を入力: ").strip()
    
    if not query:
        print("質問が入力されていません")
        return
    
    print(f"\n回答を生成中...\n")
    answer = rag.generate_answer(query)
    
    print("=" * 60)
    print("【回答】")
    print("=" * 60)
    print(answer)
    print("=" * 60)


def interactive_mode():
    """インタラクティブモード(連続質問)"""
    rag = WikipediaRAGFileSearch()
    
    # Store情報の確認
    store_info = rag.get_store_info()
    if store_info.get('status') != 'active':
        print("\nエラー: File Search Storeが設定されていません")
        print("data_loader_filesearch.pyでデータをアップロードしてください")
        return
    
    print("\nインタラクティブモードを開始します")
    print("終了するには 'quit' または 'exit' と入力してください")
    print("=" * 60)
    
    while True:
        query = input("\n質問: ").strip()
        
        if query.lower() in ['quit', 'exit', 'q']:
            print("終了します")
            break
        
        if not query:
            continue
        
        print("\n回答を生成中...\n")
        answer = rag.generate_answer(query)
        
        print("-" * 60)
        print(answer)
        print("-" * 60)


def show_statistics():
    """統計情報の表示"""
    rag = WikipediaRAGFileSearch()
    
    print("\n" + "=" * 60)
    print("統計情報")
    print("=" * 60)
    
    # Store情報
    store_info = rag.get_store_info()
    print(f"\n【Store情報】")
    print(f"  Store名: {store_info.get('store_name', 'N/A')}")
    print(f"  表示名: {store_info.get('display_name', 'N/A')}")
    print(f"  ステータス: {store_info.get('status', 'N/A')}")
    
    if store_info.get('status') == 'active':
        # ファイル一覧
        print(f"\n【Store内のファイル】")
        files = rag.list_files_in_store()
        print(f"  総ファイル数: {len(files)}件")
        
        if files:
            print(f"\n  最近のファイル(最大5件):")
            for i, file_info in enumerate(files[:5], 1):
                display_name = file_info.get('display_name', 'N/A')
                size_kb = file_info.get('size_bytes', 0) / 1024
                print(f"    {i}. {display_name} ({size_kb:.1f}KB)")
        
        # マッピング情報
        mappings = load_file_mappings()
        if mappings:
            print(f"\n【ファイルマッピング情報】")
            print(f"  マッピング総数: {len(mappings)}件")
            print(f"\n  マッピングサンプル(最大3件):")
            for i, (ascii_name, info) in enumerate(list(mappings.items())[:3], 1):
                original = info.get('original_filename', 'N/A')
                title = info.get('title', 'N/A')
                upload_date = info.get('upload_date', 'N/A')
                print(f"    {i}. {title}")
                print(f"       元ファイル: {original}")
                print(f"       アップロード日: {upload_date}")
    else:
        print("\nStoreが設定されていないか、エラーが発生しています")
        if store_info.get('error'):
            print(f"エラー詳細: {store_info.get('error')}")
    
    print("=" * 60)


def show_file_mappings():
    """ファイルマッピング一覧の表示"""
    print("\n" + "=" * 60)
    print("ファイルマッピング一覧")
    print("=" * 60)
    
    mappings = load_file_mappings()
    
    if not mappings:
        print("\nファイルマッピング情報が見つかりません")
        print("data_loader_filesearch.pyでデータをアップロードしてください")
        return
    
    print(f"\n総数: {len(mappings)}件\n")
    
    for i, (ascii_name, info) in enumerate(mappings.items(), 1):
        original = info.get('original_filename', 'N/A')
        title = info.get('title', 'N/A')
        upload_date = info.get('upload_date', 'N/A')
        
        print(f"{i}. {title}")
        print(f"   元ファイル名: {original}")
        print(f"   ASCII名: {ascii_name}")
        print(f"   アップロード日: {upload_date}")
        print()
    
    print("=" * 60)


def main():
    """メインメニュー"""
    while True:
        print("\n" + "=" * 60)
        print("Wikipedia RAG File Search システム - テストメニュー")
        print("=" * 60)
        print("1. 質問応答テスト")
        print("2. インタラクティブモード(連続質問)")
        print("3. 統計情報の表示")
        print("4. ファイルマッピング一覧")
        print("5. 終了")
        print("=" * 60)
        
        choice = input("\n選択 (1-5): ").strip()
        
        if choice == '1':
            test_qa()
        elif choice == '2':
            interactive_mode()
        elif choice == '3':
            show_statistics()
        elif choice == '4':
            show_file_mappings()
        elif choice == '5':
            print("\n終了します")
            break
        else:
            print("\n無効な選択です")


if __name__ == "__main__":
    # 初期確認
    rag = WikipediaRAGFileSearch()
    store_info = rag.get_store_info()
    
    if store_info.get('status') != 'active':
        print("\n" + "=" * 60)
        print("⚠️  注意")
        print("=" * 60)
        print("\nFile Search Storeが設定されていません")
        print("先にdata_loader_filesearch.pyを実行してください:")
        print("\n  $ python data_loader_filesearch.py")
        print("\n" + "=" * 60)
        
        proceed = input("\nそれでもテストメニューを起動しますか? (y/N): ").strip()
        if proceed.lower() != 'y':
            print("終了します")
            exit(0)
    
    # メインメニューを起動
    main()

実行方法

1. データのアップロード

# 基本的な実行(デフォルトディレクトリは./data/wikipedia)
$ python data_loader_filesearch.py

# データディレクトリを指定
$ python data_loader_filesearch.py --data-dir ./my_data/wiki

# 既存データをリセットして再アップロード
$ python data_loader_filesearch.py --reset

実行時の出力

$ python data_loader_filesearch.py --data-dir ./data/wikipedia
=== Wikipedia RAG File Search データローダー ===

データディレクトリ: ./data/wikipedia

新しいFile Search Storeを作成中...
Store作成完了: fileSearchStores/wikipediaknowledgebase-b41bw6479gvy

======================================================================
コスト削減のため、.envファイルに以下を追加してください:
STORE_NAME=fileSearchStores/wikipediaknowledgebase-b41bw6479gvy
======================================================================

2件のファイルをアップロードします...

アップロード中: 自然言語処理.md -> wiki_38bfa4204013f29a.md
  ✓ アップロード完了: 自然言語処理.md
アップロード中: 機械学習.md -> wiki_500932c40af10a90.md
  ✓ アップロード完了: 機械学習.md
データアップロード中: 100%|███████████████████████████████████████████████████████████████████| 2/2 [00:24<00:00, 12.26s/it]

完了: 2件のファイルをアップロードしました
File Search Store総ファイル数: 2件(マッピング情報より)

アップロード済みファイル:
  1. 自然言語処理 (自然言語処理.md)
  2. 機械学習 (機械学習.md)

✓ ファイルが正常にアップロードされました
  インデックス作成が完了するまで、数分かかる場合があります

ファイルマッピングを保存しました: file_mappings.json

実行時のこの部分を.envに格納します。

2. テストの実行

$ python test_rag_filesearch.py

メニューから以下の機能を試すことができます。

  1. 類似情報検索テスト - キーワードで関連記事を検索
  2. 質問応答テスト - 自然言語での質問に回答(引用付き)
  3. インタラクティブモード - 対話的に連続質問
  4. 統計情報の表示 - アップロードされているファイル数を確認

実行時の出力

$ python test_rag_filesearch.py

============================================================
Wikipedia RAG File Search システム - テストメニュー
============================================================
1. 質問応答テスト
2. インタラクティブモード(連続質問)
3. 統計情報の表示
4. ファイルマッピング一覧
5. 終了
============================================================

選択 (1-5): 1

質問を入力: 作家がAnthropicを提訴した訴訟の判決内容を教えてください
デバッグモードを有効にしますか? (y/N): n

回答を生成中...

============================================================
【回答】
============================================================
作家がAnthropicを著作権侵害で訴えた訴訟において、カリフォルニア州北部地区連邦地方裁判所は一部の申し立てについて判決を下しました。

この訴訟は、作家のアンドレア・バーツ氏、チャールズ・グレーバー氏、カーク・ジョンソン氏が、AnthropicがAIモデル「Claude」の訓練に際し、著作物を許可なく使用したとして提訴したものです。原告は、Anthropicが海賊版サイトのデータや物理的な書籍のスキャンデータを使用したと主張しました。

裁判所の判決内容は以下の通りです。

*   **物理書籍のスキャンデータによる訓練**: 裁判所は、物理書籍のスキャンデータを用いた訓練については、新しい文章を生成するための統計的関係を学ぶ目的であり、AIの生成物が元の本のコピーや盗作をユーザーに提供しているわけではないとして、「フェアユース」に該当すると認めました。
*   **海賊版サイトのデータによる訓練**: 一方で、海賊版サイトのデータからデータセットを作成したことについては、有料コピーの代替となり、変容的な利用ではないとして、「フェアユース」を認めませんでした。

この判決は、AIの学習データに著作物を使用する際の「フェアユース」の適用範囲について、そのデータの入手元によって異なる判断を示した点で注目されます。

なお、Anthropicは作家だけでなく、2023年10月にはユニバーサルミュージックグループ(UMG)からも、所属アーティストの著作権を巡り訴訟を起こされています。AIと著作権を巡る議論は現在も進行中であり、今後も様々な判例が出てくる可能性があります。

【引用元】
1. 機械学習
2. 引用: この訴訟は、作家のアンドレア・バーツ氏、チャールズ・グレーバー氏、カーク・ジョンソン氏が、AnthropicがAIモデル「Claude」の訓練に際し、著作物を許可なく使用したとして提訴したものです...
3. 引用: 原告は、Anthropicが海賊版サイトのデータや物理的な書籍のスキャンデータを使用したと主張しました...
4. 引用: *   **物理書籍のスキャンデータによる訓練**: 裁判所は、物理書籍のスキャンデータを用いた訓練については、新しい文章を生成するための統計的関係を学ぶ目的であり、AIの生成物が元の本のコピーや盗作...
5. 引用: *   **海賊版サイトのデータによる訓練**: 一方で、海賊版サイトのデータからデータセットを作成したことについては、有料コピーの代替となり、変容的な利用ではないとして、「フェアユース」を認めません...

これで、RAGシステムのFile Search版が完成しました😊

金管

File Searchの料金体系

  • インデックス作成 … ファイルサイズに応じて課金(サイズの制限 1GB/Store、100MB/ファイル)
  • 検索クエリ … 無料(モデル使用料は別途)

コストを抑えるためのポイント

  • 同じStoreを再利用する(.envSTORE_NAMEを設定)
  • 不要なファイルは削除する
  • 定期的にStore内のファイルを見直す

改良版の利点と欠点

利点

以下のような利点があります。

  1. セットアップが簡単 … ローカルのベクトルDB管理が不要
  2. コード量の削減 … コード削減・よりシンプルで理解しやすい
  3. 引用元の自動取得 … 回答に引用元が自動で付与されるため、情報の出所が明確
  4. メンテナンスが容易 … ローカルDBの管理不要・バックアップの心配不要
  5. スケーラビリティ … クラウドベースで拡張性が高い・大量のファイルにも対応可能

欠点

以下のような欠点も存在します。

  1. 料金が発生 … インデックス作成時に課金・大量のファイルでコスト増
  2. オフライン利用不可 … オフライン環境での利用不可で、ファイルの内容によってはセキュリティリスクも考慮
  3. 検索速度 … 従来版より若干遅い(1〜2秒)
  4. カスタマイズ性 … 埋め込みモデルの変更不可・チャンク分割の細かい制御不可

おわりに

Gemini File Searchを使用することで、RAGシステムの構築が大幅に簡素化された気がします。引用元も自動ででるのは結構たさうかりますね。

移行により、変わった点をまとめると以下の通りでしょうか。

  • セットアップの簡素化
  • コード量の削減
  • 引用元が自動で取得可能
  • メンテナンスの容易さ
  • スケーラビリティ

個人的には、プロトタイピングや小規模なプロジェクトではFile Search版大規模なプロダクションや完全なカスタマイズが必要な場合は従来版という使い分けになるのかなと思います。セキュリティを気にする場合には、ローカルで完結する従来版の方が良いかもしれません。

GitHubリポジトリ

今回のプログラムの完全なコードは以下のリポジトリで公開しています。

github.com

参考リンク