RAGチャットボットの参照元KB記事をCloud Logging→BigQuery→Data Studioで可視化する

Vertex AI RAG Engineを使ったGoogle Chatボットの構造化ログに参照元KBファイル情報を追加し、BigQueryビューとLooker Studioで「どのKB記事が回答に使われたか」を分析可能にするまでの実装を紹介します。Looker StudioのREPEATED型の罠やスキーマキャッシュの落とし穴も解説。

lin-yuchen

2026.07.16

 はじめにVertex AI RAG Engineを使ったGoogle Chatボットを運用していて、こんな疑問が出てきました。
「ユーザーの質問に対してボットがどのKB記事を参照して回答したか、ログから追えないのでは？」
構造化ログにはクエリテキスト、回答テキスト、RAGの類似度スコア、レイテンシなどは記録していましたが、参照元のKBファイル名が記録されていないことに気づきました。これでは低評価のフィードバックを受けても「どの記事を改善すべきか」がわかりません。
本記事では、RAGパイプラインの構造化ログに参照元ファイル情報を追加し、BigQueryビューとData Studioで分析可能にするまでの流れを紹介します。
 前提・環境Google Cloud Functions (2nd gen) + Cloud Run
Vertex AI RAG Engine（ナレッジベース検索）
Cloud Logging → BigQuery ログシンク（構造化ログの自動転送）
Data Studio（ダッシュボード）
Python 3.14
 全体構成
 現状の課題: 参照元ファイルが記録されていないRAGパイプラインの retrieve_context() 関数は、取得したチャンクごとに {text, score, source} を返していました。source には context.source_uri から取得したGCS URI（例: gs://bucket/kb_drive_export/gws_020.txt）が入っています。
しかし、usage eventのログ出力では以下の2つしか記録しておらず、source URIは完全に捨てていました:
rag_confidence: 最もスコアの高いチャンクのスコア
rag_sources_count: 取得チャンク数
bot/worker.py
# 変更前: sourceは未使用
best_score = contexts[0]["score"] if contexts else None
log_event(
    severity, "usage",
    rag_confidence=best_score,
    rag_sources_count=len(contexts),
    # ← rag_sources がない！
)
RAGオブザーバビリティのベストプラクティスでは、チャンクレベルの属性（chunk_id、similarity_score、source_url）をリトリーバルスパンに記録することが推奨されています。参照元の追跡がなければ、フィードバックと情報源を紐付けた品質改善ができません。
 実装: usage eventに参照元ファイルを追加 ログ出力の変更_emit_usage_event() に rag_sources フィールドを追加しました:
bot/worker.py
best_score = contexts[0]["score"] if contexts else None
rag_sources = list(dict.fromkeys(
    ctx["source"].rsplit("/", 1)[-1]
    for ctx in contexts if ctx.get("source")
))
log_event(
    severity, "usage",
    rag_confidence=best_score,
    rag_sources_count=len(contexts),
    rag_sources=rag_sources,  # ← 追加
    # ...
)
ポイントをいくつか解説します。
ファイル名のみ抽出
rsplit("/", 1)[-1] でGCS URIからファイル名だけを取り出しています。gs://your-bucket/kb_drive_export/foo_020.txt → foo_020.txt になります。フルパスでもいいのですが、バケット名やプレフィックスは固定なのでログが冗長になるだけでした。
重複排除
dict.fromkeys() で重複を排除しつつ、出現順（＝類似度スコア順）を保持しています。RAGでは同一ドキュメントから複数チャンクが返されることがよくあるため、重複排除は重要です。set() ではなく dict.fromkeys() を使うことで、最初に出現したもの（＝最も関連性が高い）の順序が保たれます。
全クエリで記録
is_answerable の値に関係なく、すべてのクエリでsource情報を記録しています。回答できなかったケース（fallback）でも「どの記事が取得されたが不十分だったか」がわかるため、KB改善の重要なシグナルになります。
 出力されるログの例{
  "severity": "INFO",
  "event_type": "usage",
  "query_text": "Claude Codeの利用申請をしたい",
  "rag_confidence": 0.193,
  "rag_sources_count": 5,
  "rag_sources": [
    "claude_020.txt",
    "boo_021.txt",
    "office_001.txt",
    "foo_035.txt",
    "workflow_022.txt"
  ],
  "is_fallback": false,
  "message_id": "spaces/xxx/messages/yyy"
}
 フィードバックとの紐付けフィードバックイベント（upvote/downvote）には message_id がJOINキーとして含まれています。usage eventにもsourceを追加したことで、BigQueryで以下のような分析が可能になりました:
低評価を受けた回答が参照していたKB記事の特定
fallbackクエリで取得されたが回答に至らなかった記事の特定
特定のKB記事が参照された頻度と評価の相関
 BigQueryビューの更新Cloud LoggingからBigQueryへのログシンクで、構造化ログは自動的にBigQueryテーブルに格納されます。rag_sources はJSONの配列なので、BigQueryでは REPEATED STRING 型として格納されます。
 usage_eventsビュー既存のビューに rag_sources を追加するだけ……と思いきや、ここで落とし穴がありました。
デプロイ前のログには rag_sources フィールドが存在しないため、ビューを更新しようとすると Field name rag_sources does not exist in STRUCT エラーが出ます。最低1件のログが書き込まれるまでビュー更新ができませんでした。
デプロイ後にテストクエリを1回送信し、ログがBigQueryに到達してからビューを更新しました:
SELECT
  timestamp,
  jsonPayload.user_id,
  jsonPayload.space_id,
  jsonPayload.message_id,
  jsonPayload.query_text,
  CAST(jsonPayload.query_length AS INT64) AS query_length,
  jsonPayload.response_text,
  CAST(jsonPayload.response_length AS INT64) AS response_length,
  jsonPayload.rag_confidence,
  CAST(jsonPayload.rag_sources_count AS INT64) AS rag_sources_count,
  ARRAY_TO_STRING(jsonPayload.rag_sources, ', ') AS rag_source_files,
  CAST(jsonPayload.latency_ms AS INT64) AS latency_ms,
  jsonPayload.is_fallback,
  severity
FROM `PROJECT_ID.chat_analytics.run_googleapis_com_stdout_*`
WHERE jsonPayload.event_type = "usage"
 feedback_with_contextビューusageとfeedbackを message_id でJOINするビューにも rag_source_files を追加:
SELECT
  u.timestamp AS query_timestamp,
  f.timestamp AS feedback_timestamp,
  u.user_id,
  u.message_id,
  u.query_text,
  u.response_text,
  u.rag_confidence,
  u.rag_source_files,
  u.latency_ms,
  u.is_fallback,
  f.vote,
  f.reason_category,
  f.comment_text
FROM `PROJECT_ID.chat_analytics.usage_events` u
JOIN `PROJECT_ID.chat_analytics.feedback_events` f
  ON u.message_id = f.message_id
これで「低評価の回答がどのKB記事を参照していたか」を一目で確認できます:
-- 低評価の回答と参照元KB記事を表示
SELECT query_text, response_text, rag_source_files, reason_category, comment_text
FROM `PROJECT_ID.chat_analytics.feedback_with_context`
WHERE vote = 'down'
ORDER BY feedback_timestamp DESC
 Data Studioとの互換性: REPEATED型の罠BigQueryビューを更新してData Studioでダッシュボードに追加しようとしたところ、以下のエラーが発生しました:
Values referenced in UNNEST must be arrays.
UNNEST contains expression of type STRING
 原因BigQueryの REPEATED STRING 型（配列）は、Data Studioのテーブルコンポーネントで直接表示できません。Data StudioはREPEATED型のカラムに対して自動的にUNNESTを試みますが、ビューで ARRAY_TO_STRING() によりSTRINGに変換済みの場合、型の不一致でエラーになります。
さらに厄介なのは、Data Studioがデータソースのスキーマをキャッシュすることです。ビューの定義を更新しても、Data Studio側で「フィールドの更新」を行わないと古いスキーマ（REPEATED型）のままクエリが生成されます。
 解決策最終的に2つの対処が必要でした:
1. カラム名を変更する
rag_sources（元のREPEATED型と同名）のままだと、フィールド更新してもキャッシュが残ることがありました。カラム名を rag_source_files に変更することで、Data Studioに「新しいフィールド」として認識させました。
-- NG: 元のREPEATED型と同名
ARRAY_TO_STRING(jsonPayload.rag_sources, ', ') AS rag_sources

-- OK: 別名にする
ARRAY_TO_STRING(jsonPayload.rag_sources, ', ') AS rag_source_files
2. Data Studioのフィールドを更新する
リソース → 追加済みのデータソースの管理

対象のデータソースの 編集 をクリック

左下の フィールドを更新 をクリック

完了
この操作はデータソースごとに必要です。usage_events と feedback_with_context の両方のデータソースで実行する必要があります。
 まとめRAGチャットボットの運用改善で重要なのは、どのKB記事が回答に使われたかを追跡できるようにすることです。今回の対応で得られた知見をまとめます:
RAGの参照元は必ずログに記録する: source_uri はリトリーバルの段階で取得できる。ログに含めないのは情報の損失
全クエリで記録する: fallbackクエリ（回答不能）でも参照元は重要なシグナル。「記事はあるが不十分」と「記事がない」を区別できる
重複排除は順序を保持する: dict.fromkeys() で最も関連性の高い順を維持
BigQueryのREPEATED型はData Studioと相性が悪い: ARRAY_TO_STRING() でSTRINGに変換し、カラム名も変えるのが確実
Data Studioのスキーマキャッシュに注意: ビュー変更後は必ず「フィールドの更新」が必要。データソースごとに実行