この記事について

シリーズ目次

この記事で学べること

RAGシステムのパフォーマンス計測方法
チャンクサイズがRAG精度に与える影響
Streamlit UIでのパラメータ調整機能の実装
パフォーマンス最適化のベストプラクティス

前回までで基本的なRAGチャットボットが完成しました。今回はシステムのパフォーマンスを計測し、最適なパラメータを見つけるためのチューニング機能を実装します。

Phase 4の目標

Phase 4では以下の3つを実装します：

パフォーマンス計測機能: 検索時間・回答生成時間の可視化
パラメータ調整UI: チャンクサイズやtop_kをリアルタイムで変更
結果の比較機能: 異なるパラメータでの結果を比較

これにより、システムの挙動を理解し、最適な設定を見つけることができます。

1. パフォーマンス計測モジュールの実装

まずは処理時間を計測するユーティリティを実装します。

1.1 `utils/performance.py`の実装

"""
パフォーマンス計測ユーティリティ
処理時間の計測とレポート生成機能
"""

import time
from typing import Dict, Any
from contextlib import contextmanager


class PerformanceTracker:
    """パフォーマンス計測クラス"""

    def __init__(self):
        self.metrics = {}

    @contextmanager
    def track(self, metric_name: str):
        """
        コンテキストマネージャーで処理時間を計測

        使用例:
            tracker = PerformanceTracker()
            with tracker.track("search"):
                # 処理
                pass
        """
        start_time = time.time()
        try:
            yield
        finally:
            elapsed_time = time.time() - start_time
            self.add_metric(metric_name, elapsed_time)

    def add_metric(self, name: str, value: float):
        """メトリクスを追加"""
        if name not in self.metrics:
            self.metrics[name] = []
        self.metrics[name].append(value)

    def get_average(self, name: str) -> float:
        """メトリクスの平均値を取得"""
        if name not in self.metrics or not self.metrics[name]:
            return 0.0
        return sum(self.metrics[name]) / len(self.metrics[name])

    def format_time(self, seconds: float) -> str:
        """秒数を人間が読みやすい形式に変換"""
        if seconds < 0.001:
            return f"{seconds * 1000000:.2f} μs"
        elif seconds < 1:
            return f"{seconds * 1000:.2f} ms"
        else:
            return f"{seconds:.2f} s"


def format_performance_report(performance_data: Dict[str, Any]) -> str:
    """
    パフォーマンスデータを整形してレポート文字列を生成
    """
    tracker = PerformanceTracker()
    report_lines = ["## パフォーマンスレポート\n"]

    if 'search_time' in performance_data:
        search_time_str = tracker.format_time(performance_data['search_time'])
        report_lines.append(f"- 検索時間: {search_time_str}")

    if 'generation_time' in performance_data:
        gen_time_str = tracker.format_time(performance_data['generation_time'])
        report_lines.append(f"- 回答生成時間: {gen_time_str}")

    if 'total_time' in performance_data:
        total_time_str = tracker.format_time(performance_data['total_time'])
        report_lines.append(f"- 総処理時間: {total_time_str}")

    if 'num_results' in performance_data:
        report_lines.append(f"- 検索結果数: {performance_data['num_results']}件")

    return "\n".join(report_lines)

1.2 計測機能のポイント

コンテキストマネージャーの活用

@contextmanagerデコレータを使用することで、処理時間の計測を簡潔に記述できます：

with tracker.track("検索"):
    results = retriever.search(query)

時間単位の自動変換

format_timeメソッドは時間の大きさに応じて適切な単位（μs, ms, s）で表示します。これにより、ミリ秒単位の処理も見やすく表示できます。

2. QAチェーンへのパフォーマンス計測の統合

src/qa_chain.pyのanswer_questionメソッドに計測機能を追加します。

2.1 修正版`qa_chain.py`の該当箇所

def answer_question(
    self,
    question: str,
    top_k: int = None,
    score_threshold: float = None
) -> Dict[str, Any]:
    """
    質問に対して回答を生成（パフォーマンス計測付き）
    """
    try:
        # 1. 関連ドキュメントを検索
        search_start = time.time()
        search_results = self.retriever.search(
            query=question,
            top_k=top_k,
            score_threshold=score_threshold,
            return_scores=True
        )
        search_time = time.time() - search_start

        documents = search_results['documents']

        if not documents:
            return {
                'answer': '関連する情報が見つかりませんでした',
                'sources': [],
                'performance': {
                    'search_time': search_time,
                    'generation_time': 0.0,
                    'total_time': search_time
                }
            }

        # 2. コンテキストを生成
        context = self.retriever.get_context_for_llm(documents)

        # 3. LLMで回答を生成
        generation_start = time.time()
        response = self.chain.invoke({
            'context': context,
            'question': question
        })
        generation_time = time.time() - generation_start

        # 4. 結果を整形
        formatted_sources = self.retriever.format_results(search_results)
        total_time = search_time + generation_time

        return {
            'answer': response['text'],
            'sources': formatted_sources,
            'performance': {
                'search_time': search_time,
                'generation_time': generation_time,
                'total_time': total_time,
                'num_results': len(documents)
            }
        }

    except Exception as e:
        raise Exception(f"回答生成エラー: {str(e)}")

2.2 パフォーマンス計測の内訳

RAGシステムの処理時間は主に2つに分かれます：

検索時間（search_time）: ベクトル検索でドキュメントを取得する時間
生成時間（generation_time）: LLMが回答を生成する時間

これらを個別に計測することで、どこがボトルネックかを特定できます。

3. Streamlit UIでのパラメータ調整機能

3.1 サイドバーの拡張

app.pyのサイドバー部分を拡張し、各種パラメータを調整できるようにします。

# サイドバー
with st.sidebar:
    st.header("⚙️ 設定")

    # インデックス構築セクション
    st.subheader("📁 インデックス構築")

    data_dir = st.text_input(
        "ドキュメントディレクトリ",
        value="./data/raw",
        help="読み込むドキュメントが格納されているディレクトリ"
    )

    st.write("**チャンキング設定**")
    chunk_size = st.slider(
        "チャンクサイズ",
        min_value=CHUNKING_CONFIG["chunk_size"]["min"],
        max_value=CHUNKING_CONFIG["chunk_size"]["max"],
        value=CHUNKING_CONFIG["chunk_size"]["default"],
        step=CHUNKING_CONFIG["chunk_size"]["step"],
        help=CHUNKING_CONFIG["chunk_size"]["description"]
    )

    chunk_overlap = st.slider(
        "チャンクオーバーラップ",
        min_value=CHUNKING_CONFIG["chunk_overlap"]["min"],
        max_value=CHUNKING_CONFIG["chunk_overlap"]["max"],
        value=CHUNKING_CONFIG["chunk_overlap"]["default"],
        step=CHUNKING_CONFIG["chunk_overlap"]["step"],
        help=CHUNKING_CONFIG["chunk_overlap"]["description"]
    )

    if st.button("🔨 インデックス作成", type="primary"):
        build_index(data_dir, chunk_size, chunk_overlap)

    # 検索設定
    st.subheader("🔍 検索設定")
    top_k = st.slider(
        "取得ドキュメント数",
        min_value=RETRIEVAL_CONFIG["top_k"]["min"],
        max_value=RETRIEVAL_CONFIG["top_k"]["max"],
        value=RETRIEVAL_CONFIG["top_k"]["default"],
        help=RETRIEVAL_CONFIG["top_k"]["description"]
    )

    # LLM設定
    st.subheader("🤖 LLM設定")
    temperature = st.slider(
        "Temperature",
        min_value=LLM_CONFIG["temperature"]["min"],
        max_value=LLM_CONFIG["temperature"]["max"],
        value=LLM_CONFIG["temperature"]["default"],
        step=LLM_CONFIG["temperature"]["step"],
        help=LLM_CONFIG["temperature"]["description"]
    )

    # インデックス情報
    st.subheader("📊 インデックス情報")
    if st.session_state.vector_store:
        doc_count = st.session_state.vector_store.get_document_count()
        st.info(f"インデックス化済: {doc_count}件")
    else:
        st.info("インデックスは作成されていません")

3.2 パラメータ調整のポイント

チャンクサイズ（chunk_size）

小さい値（100-300）: 情報が細かく分割され、精密な検索が可能だが、文脈が失われやすい
中程度の値（400-600）: バランスが良い（推奨）
大きい値（800-2000）: 広い文脈を保持できるが、ノイズが入りやすい

チャンクオーバーラップ（chunk_overlap）

チャンク間で重複させる文字数
0だとチャンク境界で情報が分断される可能性
50-100程度が推奨

取得ドキュメント数（top_k）

LLMに渡すチャンク数
多すぎるとノイズが増え、少なすぎると情報不足
3-7が一般的

Temperature

LLMの創造性を制御
0に近いほど確定的、1に近いほど創造的
RAGでは正確性が重要なので0.0-0.3を推奨

4. パフォーマンス表示機能の実装

4.1 メイン画面でのパフォーマンス表示

チャット応答にパフォーマンス情報を追加します。

# 回答生成（app.pyの該当部分）
with st.chat_message("assistant"):
    with st.spinner("回答を生成しています..."):
        try:
            # LLM設定の更新
            st.session_state.qa_chain.update_config(temperature=temperature)

            # 質問に対する回答
            result = st.session_state.qa_chain.answer_question(
                question=question,
                top_k=top_k
            )

            # 回答の表示
            st.markdown(result['answer'])

            # ソース情報の表示
            if result['sources']:
                with st.expander("📚 参照元"):
                    for source in result['sources']:
                        st.markdown(
                            f"**[{source['rank']}] {source['metadata'].get('source', '')}** "
                            f"(スコア: {source['score']:.4f})"
                        )
                        st.text(source['content'][:200] + "...")
                        st.divider()

            # パフォーマンス情報の表示
            with st.expander("⏱️ パフォーマンス"):
                st.markdown(format_performance_report(result['performance']))

            # チャット履歴に追加
            st.session_state.chat_history.append({
                "role": "assistant",
                "content": result['answer'],
                "sources": result['sources'],
                "performance": result['performance']
            })

        except Exception as e:
            st.error(f"エラーが発生しました: {str(e)}")

4.2 パフォーマンス表示の例

実際の表示は以下のようになります：

## パフォーマンスレポート

- 検索時間: 45.23 ms
- 回答生成時間: 1.87 s
- 総処理時間: 1.92 s
- 検索結果数: 5件

この情報から、回答生成がボトルネックであることがわかります。

5. チューニングのベストプラクティス

5.1 チャンクサイズの選び方

チャンクサイズは「情報の粒度」と「文脈の広さ」のトレードオフです。

実験手順

まず中間値（500）で試す
回答の精度が低い場合：
- 情報が不足している → chunk_sizeを大きく（800-1000）
- ノイズが多い → chunk_sizeを小さく（300-400）
複数のチャンクサイズで同じ質問をテストし、比較する

5.2 top_kの調整

取得するドキュメント数は「情報量」と「ノイズ」のトレードオフです。

推奨手順

top_k=3から開始
回答が不完全な場合はtop_kを増やす（5, 7）
回答に無関係な情報が含まれる場合はtop_kを減らす

5.3 実験結果の記録

異なるパラメータでの結果を記録しておくことをお勧めします：

| chunk_size | top_k | 検索時間 | 生成時間 | 総時間 | 精度（主観） |
|-----------|-------|---------|---------|--------|-------------|
| 300       | 3     | 42ms    | 1.8s    | 1.84s  | 中          |
| 500       | 5     | 45ms    | 1.9s    | 1.95s  | 高          |
| 800       | 5     | 48ms    | 2.1s    | 2.15s  | 中          |

6. 実際のチューニング例

6.1 ケーススタディ: 社内規則の質問応答

テストケース: 「有給休暇は何日もらえますか？」

実験1: chunk_size=300, top_k=3

検索時間: 42ms
生成時間: 1.8s
結果: 「入社6ヶ月後に10日」と回答されたが、勤続年数に応じた増加の情報が欠落

実験2: chunk_size=500, top_k=5

検索時間: 45ms
生成時間: 1.9s
結果: 「入社6ヶ月後に10日、その後勤続年数に応じて最大20日まで増加」と完全な回答

実験3: chunk_size=800, top_k=5

検索時間: 48ms
生成時間: 2.1s
結果: 回答は正確だが、無関係な情報（特別休暇の説明）も含まれる

結論: このユースケースではchunk_size=500, top_k=5が最適

6.2 パフォーマンスのボトルネック分析

計測結果から、ほとんどの処理時間が「回答生成時間」に費やされていることがわかります：

検索時間: 40-50ms（全体の2-3%）
生成時間: 1.8-2.1s（全体の97-98%）

最適化の方向性

検索の最適化は優先度が低い: 検索は十分高速
LLMの最適化が重要:
- より小さいchunk_sizeでtop_kを減らす → LLMへの入力トークン数削減
- ただし精度とのトレードオフを考慮

7. 設定ファイルの更新

チューニングで見つけた最適値をデフォルト値としてconfig/settings.pyに反映します：

# チャンキング設定
CHUNKING_CONFIG = {
    "chunk_size": {
        "default": 500,  # チューニング結果に基づく推奨値
        "min": 100,
        "max": 2000,
        "step": 100,
        "description": "テキストを分割する文字数"
    },
    "chunk_overlap": {
        "default": 50,
        "min": 0,
        "max": 500,
        "step": 10,
        "description": "チャンク間の重複文字数"
    },
    "separators": ["\n\n", "\n", "。", "、", " ", ""]
}

# 検索設定
RETRIEVAL_CONFIG = {
    "top_k": {
        "default": 5,  # チューニング結果に基づく推奨値
        "min": 1,
        "max": 20,
        "description": "取得するドキュメント数"
    },
    "score_threshold": {
        "default": 0.0,
        "min": 0.0,
        "max": 1.0,
        "step": 0.1,
        "description": "類似度スコアの閾値"
    }
}

8. トラブルシューティング

8.1 検索時間が遅い場合

症状: 検索時間が100ms以上

原因と対策

インデックスサイズが大きすぎる
- 対策: 不要なドキュメントを削除してインデックス再構築
ChromaDBの永続化ファイルが肥大化
- 対策: chroma_db/ディレクトリを削除して再構築

8.2 回答生成時間が遅い場合

症状: 生成時間が3秒以上

原因と対策

top_kが大きすぎる
- 対策: top_kを5以下に設定
chunk_sizeが大きすぎる
- 対策: chunk_sizeを500-600に調整
Google AI StudioのAPI制限
- 対策: レート制限に引っかかっている可能性。時間をおいて再試行

8.3 精度が低い場合

症状: 回答が不正確、または「情報がありません」と返される

原因と対策

chunk_sizeが小さすぎる
- 対策: chunk_sizeを600-800に増やす
top_kが小さすぎる
- 対策: top_kを7-10に増やす
ドキュメントに情報がない
- 対策: 必要なドキュメントがインデックス化されているか確認

まとめ

Phase 4では以下を実装しました：

パフォーマンス計測機能: 検索時間と生成時間を個別に計測
パラメータ調整UI: Streamlitのサイドバーで各種パラメータを調整
チューニング手法: chunk_size、top_k、temperatureの最適化方法

これにより、RAGシステムの挙動を理解し、ユースケースに応じた最適なパラメータを見つけることができるようになりました。

重要なポイント

チャンクサイズは情報の粒度と文脈の広さのトレードオフ
top_kは情報量とノイズのトレードオフ
パフォーマンス計測により、ボトルネックを特定できる
実験結果を記録し、最適値を見つける

次回（Phase 5）では、完成したアプリケーションをStreamlit Community Cloudにデプロイし、運用上の注意点を解説します。

次回予告

第5回: Phase 5 - デプロイと運用、総括

Streamlit Community Cloudへのデプロイ手順
環境変数の設定方法
運用上の注意点（API制限、データ永続化の課題）
プロジェクトの総括と今後の改善案

GitHubリポジトリ

コード全体は以下のリポジトリで公開しています：

ご質問やフィードバックは、コメント欄またはGitHubのIssueでお気軽にどうぞ。