ツール結果が会話を静かに食いつぶす — Gemini Function Calling の巨大レスポンスをハンドル渡しで軽くする設計

# compactor.py — ツール結果をトークン予算内に縮約してから functionResponse に詰める
# 何を解決するか: 使わないフィールドの再送で入力トークンが複利増加する問題
import json
from google import genai
 
client = genai.Client()
MODEL = "gemini-flash-latest"
 
def count(text: str) -> int:
    return client.models.count_tokens(model=MODEL, contents=text).total_tokens
 
def compact_tool_result(rows: list[dict], keep_fields: list[str],
                        budget_tokens: int = 6000) -> dict:
    """優先フィールドのみ残し、予算を超えたら行数を切り詰める。
    切り詰めたことは必ずメタ情報としてモデルに伝える。"""
    slim = [{k: r.get(k) for k in keep_fields} for r in rows]
    kept = len(slim)
    while kept > 1 and count(json.dumps(slim[:kept], ensure_ascii=False)) > budget_tokens:
        kept = int(kept * 0.8)  # 2割ずつ削って予算内に収める
    return {
        "rows": slim[:kept],
        "total_rows": len(rows),
        "returned_rows": kept,
        "truncated": kept < len(rows),
        "note": "truncated=true の場合、全件が必要なら fetch_more を呼んでください",
    }
 
# 使用例: レビューは text と rating だけ残す
# compacted = compact_tool_result(reviews, keep_fields=["text", "rating"])
# → 312件 48,200トークン が 6,000トークン以内に収まる

# handle_passing.py — ツール結果の参照渡し（ハンドル渡し）
# 何を解決するか: 巨大な結果を履歴に置かず、必要な断片だけ後から取得する
import uuid
 
class ResultStore:
    """run 単位で生きる一時ストア。run を跨いだ参照は明示的にエラーにする"""
    def __init__(self):
        self._data: dict[str, list[dict]] = {}
 
    def put(self, rows: list[dict]) -> str:
        handle = f"res_{uuid.uuid4().hex[:8]}"
        self._data[handle] = rows
        return handle
 
    def slice(self, handle: str, offset: int = 0, limit: int = 40) -> dict:
        if handle not in self._data:
            return {"error": f"handle {handle} はこの実行内に存在しません"}
        rows = self._data[handle][offset:offset + limit]
        return {"rows": rows, "offset": offset,
                "total": len(self._data[handle])}
 
store = ResultStore()
 
def fetch_reviews(store_id: str) -> dict:
    """モデルに渡るのはこの戻り値だけ。全文は ResultStore に退避"""
    rows = load_reviews_from_db(store_id)  # 実データ取得（各自の実装）
    handle = store.put(rows)
    return {
        "handle": handle,
        "total_rows": len(rows),
        "schema": ["text", "rating", "date", "lang"],
        "digest": summarize_locally(rows),  # 星別件数など軽い統計のみ
        "note": "本文が必要な場合は fetch_rows(handle, offset, limit) を呼んでください",
    }
 
def fetch_rows(handle: str, offset: int, limit: int) -> dict:
    """モデルが詳細を要求したときだけ呼ばれる第二のツール"""
    return store.slice(handle, offset, min(limit, 40))