この記事について
シリーズ目次
- プロジェクト概要と環境構築
- Phase 1: ドキュメントローダーとチャンキング実装
- Phase 2&3: 検索と回答生成の実装
- Phase 4: パフォーマンスチューニング（本記事）
- Phase 5: デプロイと運用、総括
この記事で学べること
- 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でお気軽にどうぞ。