GeminiコンテキストキャッシュとLangfuseで実現するコスト監視

📊 Langfuse ダッシュボード表示例:
├── 入力トークン: 15 → $0.0000045
├── キャッシュトークン: 1,189 → $0.0000891
├── 出力トークン: 1,291 → $0.0032275
├── 合計コスト: $0.0033211
└── レイテンシ: 自動測定

google-generativeai
langfuse
python-dotenv

### .env
LANGFUSE_PUBLIC_KEY=[パブリックキー]
LANGFUSE_SECRET_KEY=[シークレットキー]
LANGFUSE_HOST=[langfuseのホストURL]
GOOGLE_CLOUD_PROJECT=[自身のGoogle Cloud Project ID]
GOOGLE_CLOUD_LOCATION=global

from google.genai.types import Content, CreateCachedContentConfig, Part

content_cache = client.caches.create(
    model="gemini-2.5-flash",
    config=CreateCachedContentConfig(
        contents=[Content(role="user", parts=[Part.from_text(text=long_context)])],
        ttl="3600s",  # 1時間有効
    ),
)
print(f"キャッシュID: {content_cache.name}")

# .env GEMINI_CACHE_ID=projects/xxx/locations/global/cachedContents/123

# キャッシュIDを使用してGemini呼び出し
cache_id = "projects/xxx/locations/global/cachedContents/123"
# ステップ2で取得

response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents="あなたの質問をここに入力",
    config=GenerateContentConfig(cached_content=cache_id)
)

# APIレスポンスから直接取得
usage_metadata = response.usage_metadata
cached_tokens = getattr(usage_metadata, "cached_content_token_count", 0)
input_tokens = getattr(usage_metadata, "prompt_token_count", 0) - cached_tokens
output_tokens = getattr(usage_metadata, "candidates_token_count", 0)

import os
from typing import Any
from dotenv import load_dotenv
from google import genai
from google.genai.types import GenerateContentConfig, HttpOptions
from langfuse import get_client, observe

load_dotenv()

def build_clients() -> tuple[genai.Client, Any]:
    _ = get_client()

    project_id = os.getenv("GOOGLE_CLOUD_PROJECT")
    location = os.getenv("GOOGLE_CLOUD_LOCATION")

    client = genai.Client(
        vertexai=True,
        project=project_id,
        location=location,
        http_options=HttpOptions(api_version="v1"),
    )
    return client, _

@observe(name="gemini-cached-call", as_type="generation")
def call_gemini_with_cache(
    client: genai.Client,
    cache_id: str,
    query: str,
    model_name: str = "gemini-2.5-flash",
) -> str:
    """Call Gemini with context cache and report usage to Langfuse."""
    response = client.models.generate_content(
        model=model_name,
        contents=query,
        config=GenerateContentConfig(cached_content=cache_id),
    )

    usage = response.usage_metadata
    cached_tokens = getattr(usage, "cached_content_token_count", 0)
    input_tokens = max(0, getattr(usage, "prompt_token_count", 0) - cached_tokens)
    output_tokens = getattr(usage, "candidates_token_count", 0)

    langfuse_client = get_client()
    langfuse_client.update_current_generation(
        model=model_name,
        input=query,
        output=(response.text or ""),
        usage_details={
            "input": input_tokens,
            "cached_tokens": cached_tokens,
            "output": output_tokens,
        },
        metadata={
            "cache_id": cache_id,
            "provider": "vertex-ai",
        },
    )

    return response.text or ""

def main() -> None:
    client, langfuse_client = build_clients()

    # Example placeholders for blog readers
    cache_id = os.getenv(
        "GEMINI_CACHE_ID",     "projects/[Project ID]/locations/global/cachedContents/[cache_number]",
    )
    query = "データサイエンスについて教えて"

    _ = call_gemini_with_cache(client, cache_id, query)

    # Ensure data is flushed to Langfuse backend in short-lived processes
    langfuse_client.flush()

if __name__ == "__main__":
    main()