Gemini API × Google Workspace のパーソナル秘書AI — 認証・オーケストレーション・承認ゲートの本番設計

# gmail_agent.py
import base64
from datetime import datetime, timedelta
from workspace_auth import build_service
 
def fetch_recent_emails(max_results: int = 20, days_back: int = 1) -> list[dict]:
    """
    直近N日の重要メールを取得する。
    本文は取得せず、メタデータとスニペットのみ返す（コスト削減）。
    """
    service = build_service("gmail", "v1")
 
    # 日付フィルター（サーバーサイドで絞る）
    after_date = (datetime.now() - timedelta(days=days_back)).strftime("%Y/%m/%d")
    query = f"is:unread after:{after_date} -category:promotions -category:social"
 
    result = service.users().messages().list(
        userId="me",
        q=query,
        maxResults=max_results
    ).execute()
 
    messages = result.get("messages", [])
    email_data = []
 
    for msg in messages:
        # metadataフォーマットで取得（本文は含まない）
        detail = service.users().messages().get(
            userId="me",
            id=msg["id"],
            format="metadata",
            metadataHeaders=["Subject", "From", "Date"]
        ).execute()
 
        headers = {h["name"]: h["value"] for h in detail.get("payload", {}).get("headers", [])}
        email_data.append({
            "id": msg["id"],
            "subject": headers.get("Subject", "(件名なし)"),
            "from": headers.get("From", ""),
            "date": headers.get("Date", ""),
            "snippet": detail.get("snippet", ""),
        })
 
    return email_data
 
def get_email_body(message_id: str) -> str:
    """メール本文を取得する（優先度の高いメールのみ呼ぶ）"""
    service = build_service("gmail", "v1")
    detail = service.users().messages().get(
        userId="me",
        id=message_id,
        format="full"
    ).execute()
 
    def extract_text(payload: dict) -> str:
        """マルチパートメールから平文テキストを再帰的に取得"""
        if payload.get("mimeType") == "text/plain":
            data = payload.get("body", {}).get("data", "")
            if data:
                return base64.urlsafe_b64decode(data).decode("utf-8", errors="replace")
        for part in payload.get("parts", []):
            text = extract_text(part)
            if text:
                return text
        return ""
 
    body = extract_text(detail.get("payload", {}))
    # 長すぎる場合は最初の2000文字に制限（トークンコスト管理）
    return body[:2000] if len(body) > 2000 else body

# calendar_agent.py
from datetime import datetime
from zoneinfo import ZoneInfo
from workspace_auth import build_service
 
JST = ZoneInfo("Asia/Tokyo")
 
def fetch_todays_events() -> list[dict]:
    """
    今日の予定を取得する。
    タイムゾーンをJSTに揃えて返す。
    """
    service = build_service("calendar", "v3")
 
    # JST基準で今日の開始・終了時刻を計算
    now_jst = datetime.now(JST)
    day_start = now_jst.replace(hour=0, minute=0, second=0, microsecond=0)
    day_end = now_jst.replace(hour=23, minute=59, second=59, microsecond=0)
 
    events_result = service.events().list(
        calendarId="primary",
        timeMin=day_start.isoformat(),
        timeMax=day_end.isoformat(),
        singleEvents=True,
        orderBy="startTime",
        maxResults=20,
    ).execute()
 
    events = events_result.get("items", [])
    formatted = []
 
    for event in events:
        start_raw = event.get("start", {})
        end_raw = event.get("end", {})
 
        # 終日イベントとtimeイベントを区別する
        if "dateTime" in start_raw:
            start_dt = datetime.fromisoformat(start_raw["dateTime"]).astimezone(JST)
            end_dt = datetime.fromisoformat(end_raw["dateTime"]).astimezone(JST)
            time_str = f"{start_dt.strftime('%H:%M')}〜{end_dt.strftime('%H:%M')}"
        else:
            time_str = "終日"
 
        formatted.append({
            "title": event.get("summary", "(タイトルなし)"),
            "time": time_str,
            "location": event.get("location", ""),
            "description": event.get("description", "")[:300],
            "attendees": [a.get("email", "") for a in event.get("attendees", [])],
        })
 
    return formatted

# secretary_agent.py
import json
import google.generativeai as genai
from gmail_agent import fetch_recent_emails, get_email_body
from calendar_agent import fetch_todays_events
 
# ツール定義（Geminiに渡すFunction Schema）
WORKSPACE_TOOLS = [
    genai.protos.Tool(
        function_declarations=[
            genai.protos.FunctionDeclaration(
                name="get_todays_schedule",
                description="今日のGoogle Calendarの予定を取得する。日程確認・会議準備に使う。",
                parameters=genai.protos.Schema(
                    type=genai.protos.Type.OBJECT,
                    properties={},
                    required=[],
                ),
            ),
            genai.protos.FunctionDeclaration(
                name="get_unread_emails",
                description="未読メールのリストをメタデータ（件名・送信者・スニペット）付きで取得する。優先度判断に使う。",
                parameters=genai.protos.Schema(
                    type=genai.protos.Type.OBJECT,
                    properties={
                        "max_results": genai.protos.Schema(
                            type=genai.protos.Type.INTEGER,
                            description="取得するメール数。デフォルト15。",
                        ),
                    },
                    required=[],
                ),
            ),
            genai.protos.FunctionDeclaration(
                name="read_email_body",
                description="特定のメールの本文を取得する。優先度の高いメールの詳細確認に使う。",
                parameters=genai.protos.Schema(
                    type=genai.protos.Type.OBJECT,
                    properties={
                        "message_id": genai.protos.Schema(
                            type=genai.protos.Type.STRING,
                            description="取得するメールのID（get_unread_emailsで取得したid）",
                        ),
                    },
                    required=["message_id"],
                ),
            ),
        ]
    )
]
 
def handle_function_call(function_name: str, function_args: dict) -> str:
    """
    Geminiが呼び出したFunctionを実行して結果を返す。
    エラーが発生してもJSON形式で返すことでGeminiが適切に処理できる。
    """
    try:
        if function_name == "get_todays_schedule":
            events = fetch_todays_events()
            return json.dumps({"events": events, "count": len(events)}, ensure_ascii=False)
 
        elif function_name == "get_unread_emails":
            max_results = function_args.get("max_results", 15)
            emails = fetch_recent_emails(max_results=max_results)
            return json.dumps({"emails": emails, "count": len(emails)}, ensure_ascii=False)
 
        elif function_name == "read_email_body":
            message_id = function_args["message_id"]
            body = get_email_body(message_id)
            return json.dumps({"body": body}, ensure_ascii=False)
 
        else:
            return json.dumps({"error": f"未知の関数: {function_name}"}, ensure_ascii=False)
 
    except Exception as e:
        # エラーをGeminiに伝えることで、エラーを踏まえた回答が返ってくる
        return json.dumps({"error": str(e), "function": function_name}, ensure_ascii=False)
 
def run_secretary_agent(user_query: str) -> str:
    """
    パーソナル秘書AIエージェントのメインループ。
    Geminiが必要なツールを自律的に呼び出してユーザーの質問に答える。
    """
    genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
    model = genai.GenerativeModel(
        model_name="gemini-2.5-pro",
        tools=WORKSPACE_TOOLS,
        system_instruction="""あなたはユーザーの個人秘書AIです。
Gmail・Google Calendar・Google Driveのツールを使って、
ユーザーの質問に正確かつ簡潔に答えてください。
 
重要な指示:
- 複数のツールが必要な場合は順番に呼び出す
- メールの優先度は「返信が必要か」「締め切りがあるか」「重要人物からか」で判断する
- 個人情報に配慮した適切な要約を心がける
- 時刻は日本時間（JST）で表示する""",
    )
 
    chat = model.start_chat()
    response = chat.send_message(user_query)
 
    # Function Callingのループ処理
    max_iterations = 5  # 無限ループ防止
    iteration = 0
 
    while response.candidates[0].content.parts and iteration < max_iterations:
        # Function Callが含まれているか確認
        function_calls = [
            part for part in response.candidates[0].content.parts
            if hasattr(part, "function_call") and part.function_call.name
        ]
 
        if not function_calls:
            # テキスト応答のみ → 最終回答
            break
 
        # 各Function Callを実行して結果を返す
        function_responses = []
        for part in function_calls:
            fc = part.function_call
            result = handle_function_call(fc.name, dict(fc.args))
 
            function_responses.append(
                genai.protos.Part(
                    function_response=genai.protos.FunctionResponse(
                        name=fc.name,
                        response={"result": result},
                    )
                )
            )
 
        response = chat.send_message(function_responses)
        iteration += 1
 
    # 最終テキストを抽出
    final_text = ""
    for part in response.candidates[0].content.parts:
        if hasattr(part, "text"):
            final_text += part.text
 
    return final_text

# コスト追跡（secretary_agent.pyに追記）
import datetime
import json
from pathlib import Path
 
def log_usage(response, query: str):
    """
    API使用量をJSONLファイルに記録する。
    月次でコストを集計できるようにする。
 
    期待する出力例:
    {"timestamp": "2026-05-03T04:45:00", "query_preview": "今日の予定をまとめて",
     "prompt_tokens": 3240, "response_tokens": 512, "total_tokens": 3752}
    """
    usage = response.usage_metadata
    log_entry = {
        "timestamp": datetime.datetime.now().isoformat(),
        "query_preview": query[:50],
        "prompt_tokens": usage.prompt_token_count,
        "response_tokens": usage.candidates_token_count,
        "total_tokens": usage.total_token_count,
    }
 
    log_path = Path("~/.config/secretary-ai/usage.jsonl").expanduser()
    log_path.parent.mkdir(parents=True, exist_ok=True)
    with log_path.open("a") as f:
        f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
 
def estimate_monthly_cost(log_path: Path, price_per_1m_input: float = 1.25) -> dict:
    """
    usageログから今月の推定コストを計算する。
    price_per_1m_input: 入力トークン100万個あたりのUSD単価
    """
    current_month = datetime.datetime.now().strftime("%Y-%m")
    total_input = 0
    total_output = 0
    count = 0
 
    if log_path.exists():
        for line in log_path.read_text().splitlines():
            entry = json.loads(line)
            if entry["timestamp"].startswith(current_month):
                total_input += entry.get("prompt_tokens", 0)
                total_output += entry.get("response_tokens", 0)
                count += 1
 
    # 簡易計算（実際の料金はGemini API pricing pageを参照）
    estimated_usd = (total_input / 1_000_000) * price_per_1m_input
    return {
        "count": count,
        "total_input_tokens": total_input,
        "total_output_tokens": total_output,
        "estimated_usd": round(estimated_usd, 4),
        "estimated_jpy": round(estimated_usd * 150, 1),  # 概算
    }

import time
import random
from googleapiclient.errors import HttpError
 
def with_retry(func, max_retries: int = 3, base_delay: float = 1.0):
    """
    Google API呼び出しに指数バックオフリトライを適用する。
    429（レート制限）と503（一時的なサーバーエラー）のみリトライ対象とする。
    404（リソース不存在）や403（権限なし）はリトライしても無意味なので即エラーにする。
    """
    for attempt in range(max_retries):
        try:
            return func()
        except HttpError as e:
            if e.resp.status in (429, 503) and attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"レート制限 — {delay:.1f}秒後にリトライ ({attempt + 1}/{max_retries})")
                time.sleep(delay)
            else:
                raise
 
# 使用例（fetch_todays_events内で活用する）
events = with_retry(lambda: service.events().list(
    calendarId="primary",
    timeMin=day_start.isoformat(),
    timeMax=day_end.isoformat(),
    singleEvents=True,
    orderBy="startTime",
    maxResults=20,
).execute())

# config.py — モデル指定を1箇所に集約する
from dataclasses import dataclass
 
@dataclass(frozen=True)
class ModelConfig:
    """
    秘書AIが使うモデルの指定。
    定期実行では固定バージョンを使い、移行はこのファイルの1行変更で行う。
    """
    # 分類・優先度スコアリング（呼び出し回数が多い＝コストが効く）
    triage: str = "gemini-3.5-flash"
    # 本文要約・返信下書き（品質が効く場面のみ Pro を使う）
    compose: str = "gemini-3.5-pro"
    # 移行検証用。切り替え前にここだけ latest にして出力を並べて比較する
    canary: str = "gemini-flash-latest"
 
MODELS = ModelConfig()

# approval.py
import json
from enum import Enum
 
class RiskTier(str, Enum):
    """ツールの危険度。READ は自動実行、WRITE は確認を挟む。"""
    READ = "read"      # 読むだけ。取り消し不要
    WRITE = "write"    # 状態を変える。取り消しできる（下書き作成など）
    SEND = "send"      # 外部に出る。取り消せない（送信・共有など）
 
TOOL_RISK = {
    "get_todays_schedule": RiskTier.READ,
    "get_unread_emails": RiskTier.READ,
    "read_email_body": RiskTier.READ,
    "create_draft_reply": RiskTier.WRITE,
}
 
def requires_approval(function_name: str) -> bool:
    """READ 以外は確認を必要とする。未知のツールは安全側に倒す。"""
    return TOOL_RISK.get(function_name, RiskTier.SEND) != RiskTier.READ
 
def request_approval(function_name: str, function_args: dict) -> bool:
    """
    書き込み系ツールの実行前に確認を取る。
    cron 実行では標準入力がないため、承認待ちキューへ積む実装に差し替える。
    """
    print(f"\n[確認] {function_name} を実行しようとしています")
    print(json.dumps(function_args, ensure_ascii=False, indent=2))
    answer = input("実行しますか？ [y/N]: ").strip().lower()
    return answer == "y"

def handle_function_call(function_name: str, function_args: dict) -> str:
    if requires_approval(function_name):
        if not request_approval(function_name, function_args):
            # 拒否も「結果」として返す。例外にしない
            return json.dumps(
                {"status": "declined", "reason": "ユーザーが実行を承認しませんでした"},
                ensure_ascii=False,
            )
    # 以降は既存の分岐（get_todays_schedule / get_unread_emails / ...）