Google Drive+GAS+スプシによるミニマルなデータパイプライン【東京都知事候補 #安野たかひろ #AIあんの @takahiroanno 】

2024年7月6日 13:01

AIあんのについて

こんにちは、チームあんの技術班の志水です。
安野チームではみなさんのマニフェストに対する疑問に24時間答えれるよう、AIあんのを運用しています。AIあんのとはYoutube Live上で安野の3Dアバターがチャットに来た質問を拾い、回答してくれるサービスです。

AI あんのの裏側についてはぜひ以下の記事や詳細記事をご覧ください。

ログの集計

私がチームに参入したタイミングでは、課題の1つとしてAIあんの Youtube Liveに来ていた質問に答えられていた数を数えてKPIとしたい、というものがありました。
7/6(土)までしか運用しない期間限定のプロジェクトということもあり、なるべく素早くかつ軽量にログを整形・集計・可視化するデータパイプラインを作るモチベーションがありました。
その結果Google Drive へのログの集約、Google Apps Script (以下 GAS) を用いたログの cleaning 、GAS + Google Spreadsheets (以下スプシ)で可視化を行いました。
期間限定の突貫工事とはいえ、新たなインフラを構築することなくデータパイプラインを組めることが驚きだったので、その全体像を共有いたします。

Google Drive へのログの送信

サーバー上のログを log 配下の特定のファイルに出力した後、Google Drive へのアップロードは以下のようなクラスで行うことができます。

# python
import pathlib
from typing import Literal

from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload

class GoogleDrive:
    """Google Drive API"""

    def __init__(self) -> None:
        credentials = service_account.Credentials.from_service_account_file(
            settings.GOOGLE_APPLICATION_CREDENTIALS, 
            scopes=["https://www.googleapis.com/auth/drive.file"],
        )

        self._service = build("drive", "v3", credentials=credentials)

    def upload(
        self,
        *,
        file_path: pathlib.Path,
        folder_id: str,
        mime_type: Literal["text/csv"] | Literal["application/json"],
    ) -> None:
        """Google Driveにファイルをアップロードする

        Args:
            file_path: アップロードするファイルのパス
            mime_type: ファイルのMIMEタイプ
        """
        media = MediaFileUpload(file_path, mimetype=mime_type)

        file_metadata = {
            "name": file_path.name,
            "parents": [folder_id],
        }

        file = self._service.files().create(
           body=file_metadata, media_body=media, fields="id").execute()

生ログが Google Drive に溜まることで、サーバーへのアクセスがないメンバーでも解析ができる準備が整います！

GAS を用いた一時加工

生ログには集計には不要な情報も含まれるため、人間に分かりやすい形になるよう一次処理をかけてあげます。例えば以下のような処理でログから情報を抽出し、別の加工済みスプシに反映するメインの処理を書くことができます。

# .js
// 処理のメイン関数
function startProcessingLogs(){
  // ファイル、スプシ,シートを定義する
  const files = getJsonLogFiles(RAW_LOG_FOLDER_ID);
  const spreadSheet = SpreadsheetApp.openById(SPREADSHEET_ID);
  const sheet = spreadSheet.getSheetByName("Logs");
  
  // ログをまとめる
  let allLogs = [];
  for(const logFile of files){
    const log = getLogContent(logFile);
    allLogs.push(...log);
  }

  // 重複排除などの処理
  allLogsDeduped = dedupe(allLogs);

  // スプシに値を詰める
  const values =     [
      ...allLogsDeduped.map(l=>l.getRow())
    ];
  console.log(allLogsDeduped.length);
  sheet.getRange(2,1,allLogsDeduped.length,LOG_COLUMN_SIZE).setValues(values)
}

興味のあるログ部分の抜き出しは、例えば以下のように書くことができます。

// ログの形を定義する
class AITuberLog{

  constructor(record){
    this.timestamp = new Date(record.timestamp);
    this.question = record.question;
    this.response = record.response;
    this.slide = record.metadata.image;
  }

  getRow(){
    const timestampStr = this.timestamp.toLocaleString('ja-JP',TIMEZONE_OPTIONS);
    return [timestampStr
    ,this.question,this.response,this.slide,
  }
}

// 指定したfile IDのカンマ区切りのJSONログを読み取りパースした後の配列を返す
function getLogContent(fileId){
  // ファイルの中身の取得
  const file = DriveApp.getFileById(fileId);
  const jsonData = file.getBlob().getDataAsString("utf-8");
  const jsonLines = jsonData.split("\n").filter(s => s.trim() != "");

  return jsonLines.map(jsonStr=>new AITuberLog(JSON.parse(jsonStr)));
}

この GAS に対してトリガーを設定することで、定時実行するデータパイプラインになります。

エクセルでの集計と可視化

一時加工の結果、以下のような整形されたログを得られます。

回答の中の「こんにちは、その質問には答えられません。」というのはAIあんのが回答できなかった時の定型分のため、回答できた質問と回答できなかった質問を別シートで管理をしておきます。
ここまで来れば、日付ごとの回答数は pivot table を用いて出すことができます。

また COUNTIF を使うことで1時間ごとの質問総数と回答数を集計し、そこから有効回答率(％)を算出することができます。

有効回答率の推移を追うことで、回答率が落ちていたら打ち手について考え始める契機になります。例えばプロンプトの調整をしたり、回答できない質問を分析して FAQ を充実させることが考えられます。

また、この整形化されたログ自体が、次のステップとして Talk To The City (TTTC) へのインプットにもなります。データを綺麗に持っておくことで、チームを跨ぎさまざまな形で活用できることがわかります。

まとめ

この記事では、Google Drive + GAS + スプシを用いた軽量なデータパイプラインについて紹介しました。
既存の技術やサービスを上手く組み立てることで、AIあんののサービス改善に繋がる示唆だしや打ち手に繋がることが、個人的にはすごく驚きでした！
これを短時間で用意したAIあんののメンバーたちと働けたことは個人的にも非常に大きな財産でした。

そして選挙活動は今日6日が最終日、明日が投票日となっています。チーム安野たかひろ一同、これまでの選挙の歴史を変えるような結果を実現したいと考えています！！本日有楽町でのマイク納め(19時〜生配信あり)まで各地での街頭演説（全日程が画像を参照ください）、そしてマイク納め後も選挙活動ができる7日0:00の直前まで生配信を行っています！ぜひ街頭演説やYouTube生配信でお会いしましょう！

#安野たかひろを都知事に

最新情報は、本人・事務所の公式X(Twitter)アカウントをフォローしてご覧ください！
安野たかひろ事務所(@annotakahiro24)
安野たかひろ本人(@takahiroanno)