# MoE の概念的な実装
import torch
import torch.nn as nn
 
class MoELayer(nn.Module):
    """Mixture of Experts レイヤーの概念実装"""
 
    def __init__(self, d_model: int, n_experts: int, top_k: int = 2):
        super().__init__()
        self.n_experts = n_experts
        self.top_k = top_k
 
        # 各 Expert は独立した FFN
        self.experts = nn.ModuleList([
            nn.Sequential(
                nn.Linear(d_model, d_model * 4),
                nn.GELU(),
                nn.Linear(d_model * 4, d_model),
            )
            for _ in range(n_experts)
        ])
 
        # Router: 各トークンをどの Expert に送るか決定
        self.router = nn.Linear(d_model, n_experts)
 
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # x: [batch, seq_len, d_model]
        batch, seq_len, d_model = x.shape
 
        # Router で Expert の確率分布を計算
        router_logits = self.router(x)  # [batch, seq_len, n_experts]
        router_probs = torch.softmax(router_logits, dim=-1)
 
        # Top-K Expert を選択
        top_k_probs, top_k_indices = torch.topk(router_probs, self.top_k, dim=-1)
 
        # 選択された Expert の出力を加重合算
        output = torch.zeros_like(x)
        for k in range(self.top_k):
            expert_idx = top_k_indices[:, :, k]  # [batch, seq_len]
            weight = top_k_probs[:, :, k]         # [batch, seq_len]
 
            for i in range(self.n_experts):
                mask = (expert_idx == i)
                if mask.any():
                    expert_input = x[mask]
                    expert_output = self.expertsi
                    output[mask] += weight[mask].unsqueeze(-1) * expert_output
 
        return output
 
 
# Gemini のスケール感
# 例: 16 Experts, Top-2 activation
# 総パラメータ: ~1T (推定)
# 活性パラメータ: ~1T / 8 ≈ 125B (推論時)
# → GPT-4o 級の性能を 1/8 の計算コストで実現

# コンテキスト長の比較
context_windows = {
    "GPT-4o":          128_000,    # 128K
    "Claude 3.5":      200_000,    # 200K
    "Gemini 1.5 Pro":  2_000_000,  # 2M (!)
    "Gemini 1.5 Flash": 1_000_000, # 1M
}
 
for model, tokens in context_windows.items():
    pages = tokens // 500  # 1ページ ≈ 500 トークン
    print(f"{model}: {tokens:>10,} tokens ≈ {pages:>5,} pages")

# 長コンテキストでの精度劣化パターンと対策
 
# 問題: 200万トークンのうち中間部分の情報検索精度が低下する
# (先頭と末尾の情報は記憶しやすいが、中間は記憶しにくい)
 
# 対策 1: 重要な情報を先頭・末尾に配置
def arrange_context_optimally(documents: list[str], query: str) -> str:
    """重要度順にソートし、先頭と末尾に重要な文書を配置"""
    # 関連度スコアで文書をランク付け
    ranked = rank_by_relevance(documents, query)
 
    # 交互配置: 1位→先頭, 2位→末尾, 3位→先頭寄り, ...
    arranged = []
    for i, doc in enumerate(ranked):
        if i % 2 == 0:
            arranged.insert(len(arranged) // 2, doc)
        else:
            arranged.append(doc)
 
    return "\n\n".join(arranged)
 
 
# 対策 2: 階層的な要約 (Map-Reduce パターン)
def hierarchical_analysis(documents: list[str], query: str) -> str:
    """大量文書を階層的に処理"""
    import google.generativeai as genai
 
    model = genai.GenerativeModel("gemini-1.5-flash")
 
    # Step 1: 各文書を個別に要約 (Map)
    summaries = []
    for doc in documents:
        summary = model.generate_content(
            f"以下の文書を{query}の観点から要約してください:\n\n{doc}"
        )
        summaries.append(summary.text)
 
    # Step 2: 要約を統合 (Reduce)
    pro_model = genai.GenerativeModel("gemini-1.5-pro")
    final = pro_model.generate_content(
        f"以下の要約群を統合し、{query}に対する包括的な回答を作成してください:\n\n"
        + "\n---\n".join(summaries)
    )
    return final.text
 
 
# 対策 3: チャンク分割 + 選択的投入
def selective_context(
    all_docs: list[str],
    query: str,
    max_tokens: int = 500_000,
) -> list[str]:
    """関連文書のみ選択してコンテキストに投入"""
    # Embedding ベースで関連文書をフィルタリング
    relevant = retrieve_relevant(query, all_docs, top_k=50)
 
    # トークン数を推定してカット
    selected = []
    total_tokens = 0
    for doc in relevant:
        doc_tokens = estimate_tokens(doc)
        if total_tokens + doc_tokens > max_tokens:
            break
        selected.append(doc)
        total_tokens += doc_tokens
 
    return selected

# Gemini の料金体系の詳細
gemini_pricing = {
    "gemini-1.5-pro": {
        "input_128k_below": 1.25,   # $/1M tokens (128K以下)
        "input_128k_above": 2.50,   # $/1M tokens (128K超)
        "output_128k_below": 5.00,
        "output_128k_above": 10.00,
        "context_caching_storage": 4.50,  # $/1M tokens/hour
        "context_caching_input": 0.3125,  # 75% 割引
    },
    "gemini-1.5-flash": {
        "input_128k_below": 0.075,
        "input_128k_above": 0.15,
        "output_128k_below": 0.30,
        "output_128k_above": 0.60,
        "context_caching_storage": 1.00,
        "context_caching_input": 0.01875,
    },
    "gemini-2.0-flash": {
        "input": 0.10,
        "output": 0.40,
    },
}
 
 
def calculate_gemini_cost(
    model: str,
    input_tokens: int,
    output_tokens: int,
    use_caching: bool = False,
    cached_tokens: int = 0,
    cache_duration_hours: float = 1.0,
) -> dict:
    """Gemini のコストを詳細に計算"""
    pricing = gemini_pricing[model]
 
    if model == "gemini-2.0-flash":
        input_cost = (input_tokens / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        return {
            "input_cost": input_cost,
            "output_cost": output_cost,
            "total": input_cost + output_cost,
        }
 
    # 128K 境界での料金切り替え
    threshold = 128_000
    if input_tokens <= threshold:
        input_rate = pricing["input_128k_below"]
    else:
        input_rate = pricing["input_128k_above"]
 
    output_rate = pricing["output_128k_below"] if input_tokens <= threshold \
        else pricing["output_128k_above"]
 
    if use_caching and cached_tokens > 0:
        # キャッシュ利用時の割引料金
        cache_input_cost = (cached_tokens / 1_000_000) * pricing["context_caching_input"]
        regular_input_cost = ((input_tokens - cached_tokens) / 1_000_000) * input_rate
        cache_storage_cost = (cached_tokens / 1_000_000) * pricing["context_caching_storage"] * cache_duration_hours
        input_cost = cache_input_cost + regular_input_cost + cache_storage_cost
    else:
        input_cost = (input_tokens / 1_000_000) * input_rate
 
    output_cost = (output_tokens / 1_000_000) * output_rate
 
    return {
        "input_cost": input_cost,
        "output_cost": output_cost,
        "total": input_cost + output_cost,
    }
 
 
# 使用例: 1M トークンのドキュメント処理
result = calculate_gemini_cost(
    model="gemini-1.5-pro",
    input_tokens=1_000_000,
    output_tokens=2_000,
)
print(f"1M tokens処理コスト: ${result['total']:.2f}")
 
# キャッシュ利用時 (同じドキュメントに複数クエリ)
result_cached = calculate_gemini_cost(
    model="gemini-1.5-pro",
    input_tokens=1_000_000,
    output_tokens=2_000,
    use_caching=True,
    cached_tokens=950_000,
    cache_duration_hours=1.0,
)
print(f"キャッシュ利用時コスト: ${result_cached['total']:.2f}")

import google.generativeai as genai
 
# API キー設定
genai.configure(api_key="YOUR_API_KEY")
 
# モデル初期化
model = genai.GenerativeModel("gemini-1.5-pro")
 
# テキスト生成
response = model.generate_content("量子コンピュータを小学生に説明して")
print(response.text)

import google.generativeai as genai
from pathlib import Path
 
genai.configure(api_key="YOUR_API_KEY")
model = genai.GenerativeModel("gemini-1.5-pro")
 
# 画像読み込み
image = genai.upload_file(Path("architecture_diagram.png"))
 
# 画像 + テキストのプロンプト
response = model.generate_content([
    image,
    "この図のアーキテクチャを解説し、改善点を3つ提案してください。"
])
print(response.text)

import google.generativeai as genai
 
genai.configure(api_key="YOUR_API_KEY")
model = genai.GenerativeModel("gemini-1.5-flash")
 
# ストリーミングで生成
response = model.generate_content(
    "Pythonの非同期処理について詳しく解説してください。",
    stream=True
)
 
for chunk in response:
    print(chunk.text, end="", flush=True)

import vertexai
from vertexai.generative_models import GenerativeModel
 
# Vertex AI 初期化
vertexai.init(project="my-project", location="us-central1")
 
model = GenerativeModel("gemini-1.5-pro")
 
response = model.generate_content(
    "当社の売上データを分析し、来期の予測を作成してください。",
    generation_config={
        "temperature": 0.2,
        "top_p": 0.8,
        "top_k": 40,
        "max_output_tokens": 8192,
    }
)
print(response.text)

import google.generativeai as genai
 
genai.configure(api_key="YOUR_API_KEY")
 
model = genai.GenerativeModel(
    model_name="gemini-1.5-pro",
    system_instruction="あなたは金融の専門家です。正確で慎重な回答を心がけてください。",
    safety_settings={
        "HARM_CATEGORY_HARASSMENT": "BLOCK_MEDIUM_AND_ABOVE",
        "HARM_CATEGORY_HATE_SPEECH": "BLOCK_MEDIUM_AND_ABOVE",
        "HARM_CATEGORY_SEXUALLY_EXPLICIT": "BLOCK_MEDIUM_AND_ABOVE",
        "HARM_CATEGORY_DANGEROUS_CONTENT": "BLOCK_MEDIUM_AND_ABOVE",
    }
)
 
# チャット形式
chat = model.start_chat()
response = chat.send_message("日経平均の今後の見通しについて教えてください")
print(response.text)
 
# 会話履歴の参照
for message in chat.history:
    print(f"{message.role}: {message.parts[0].text[:50]}...")

import google.generativeai as genai
import json
 
genai.configure(api_key="YOUR_API_KEY")
 
# ツール定義
get_weather = genai.protos.Tool(
    function_declarations=[
        genai.protos.FunctionDeclaration(
            name="get_weather",
            description="指定された都市の現在の天気を取得する",
            parameters=genai.protos.Schema(
                type=genai.protos.Type.OBJECT,
                properties={
                    "city": genai.protos.Schema(
                        type=genai.protos.Type.STRING,
                        description="都市名 (例: 東京, 大阪)",
                    ),
                    "unit": genai.protos.Schema(
                        type=genai.protos.Type.STRING,
                        enum=["celsius", "fahrenheit"],
                        description="温度の単位",
                    ),
                },
                required=["city"],
            ),
        ),
    ]
)
 
model = genai.GenerativeModel(
    "gemini-1.5-pro",
    tools=[get_weather],
)
 
chat = model.start_chat()
response = chat.send_message("東京の天気を教えてください")
 
# Function Call の処理
for part in response.parts:
    if fn := part.function_call:
        print(f"関数呼び出し: {fn.name}")
        print(f"引数: {dict(fn.args)}")
 
        # 実際の関数を呼び出し (ここではモック)
        weather_data = {"temperature": 22, "condition": "晴れ", "humidity": 55}
 
        # 結果をモデルに返す
        response = chat.send_message(
            genai.protos.Content(
                parts=[genai.protos.Part(
                    function_response=genai.protos.FunctionResponse(
                        name=fn.name,
                        response={"result": weather_data},
                    )
                )]
            )
        )
        print(response.text)

import google.generativeai as genai
from google.generativeai import caching
import datetime
 
genai.configure(api_key="YOUR_API_KEY")
 
# 大きなドキュメントをキャッシュに格納
with open("large_document.txt", "r") as f:
    document_content = f.read()
 
# キャッシュの作成 (最低 32,768 トークン以上が必要)
cache = caching.CachedContent.create(
    model="models/gemini-1.5-pro-002",
    display_name="my-document-cache",
    system_instruction="あなたは文書分析の専門家です。",
    contents=[document_content],
    ttl=datetime.timedelta(hours=2),  # 2時間有効
)
 
print(f"Cache ID: {cache.name}")
print(f"Token Count: {cache.usage_metadata.total_token_count}")
 
# キャッシュを使ったモデル生成 (75% 割引の入力トークン料金)
model = genai.GenerativeModel.from_cached_content(cache)
 
# 同じドキュメントに対して複数の質問をコスト効率良く実行
questions = [
    "この文書の主要な論点を3つ挙げてください",
    "著者の結論は何ですか？",
    "第3章の要約を作成してください",
    "この文書の矛盾点や弱点を指摘してください",
]
 
for q in questions:
    response = model.generate_content(q)
    print(f"\nQ: {q}")
    print(f"A: {response.text[:200]}...")
 
# キャッシュの削除
cache.delete()

import vertexai
from vertexai.generative_models import GenerativeModel, Tool
from vertexai.preview.generative_models import grounding
 
vertexai.init(project="my-project", location="us-central1")
 
model = GenerativeModel("gemini-1.5-pro")
 
# Google Search Grounding を使った回答
tool = Tool.from_google_search_retrieval(
    grounding.GoogleSearchRetrieval()
)
 
response = model.generate_content(
    "2025年のAI規制に関する最新の動向を教えてください",
    tools=[tool],
    generation_config={"temperature": 0.1},
)
 
print(response.text)
 
# Grounding のメタデータ (ソース情報)
if hasattr(response, 'candidates'):
    for candidate in response.candidates:
        if hasattr(candidate, 'grounding_metadata'):
            metadata = candidate.grounding_metadata
            print("\n検索ソース:")
            for chunk in metadata.grounding_chunks:
                print(f"  - {chunk.web.title}: {chunk.web.uri}")

import google.generativeai as genai
import time
 
genai.configure(api_key="YOUR_API_KEY")
 
# 動画のアップロード
video_file = genai.upload_file("meeting_recording.mp4")
 
# 処理完了待ち
while video_file.state.name == "PROCESSING":
    print("動画を処理中...")
    time.sleep(10)
    video_file = genai.get_file(video_file.name)
 
if video_file.state.name == "FAILED":
    raise ValueError("動画のアップロードに失敗しました")
 
print(f"動画準備完了: {video_file.uri}")
 
model = genai.GenerativeModel("gemini-1.5-pro")
 
# タイムスタンプ付きの要約
response = model.generate_content([
    video_file,
    """この会議動画を分析してください:
    1. 参加者の一覧と発言頻度
    2. 議題ごとのタイムスタンプ (MM:SS - トピック)
    3. 各議題の結論と次のアクション
    4. 未解決の課題
    日本語で回答してください。"""
])
print(response.text)
 
# 特定の時間帯に関する質問
response = model.generate_content([
    video_file,
    "動画の5:00-10:00の区間で議論された技術的な課題を詳しく説明してください。"
])
print(response.text)

import google.generativeai as genai
import asyncio
from typing import Optional
 
genai.configure(api_key="YOUR_API_KEY")
 
 
async def batch_generate(
    prompts: list[str],
    model_name: str = "gemini-1.5-flash",
    max_concurrent: int = 5,
    temperature: float = 0.7,
) -> list[dict]:
    """Gemini API でバッチ処理を実行"""
    model = genai.GenerativeModel(model_name)
    semaphore = asyncio.Semaphore(max_concurrent)
    results = []
 
    async def process_single(prompt: str, index: int) -> dict:
        async with semaphore:
            try:
                response = await model.generate_content_async(
                    prompt,
                    generation_config={"temperature": temperature},
                )
                return {
                    "index": index,
                    "prompt": prompt[:100],
                    "response": response.text,
                    "status": "success",
                }
            except Exception as e:
                return {
                    "index": index,
                    "prompt": prompt[:100],
                    "error": str(e),
                    "status": "error",
                }
 
    tasks = [process_single(p, i) for i, p in enumerate(prompts)]
    results = await asyncio.gather(*tasks)
    return sorted(results, key=lambda x: x["index"])
 
 
# 使用例
prompts = [
    f"質問{i}: AIの応用分野{i}について100字で説明してください"
    for i in range(20)
]
 
results = asyncio.run(batch_generate(prompts, max_concurrent=10))
for r in results:
    if r["status"] == "success":
        print(f"[{r['index']}] {r['response'][:80]}...")
    else:
        print(f"[{r['index']}] ERROR: {r['error']}")

# Gemini Nano は Android の AICore API 経由で利用
# (Java/Kotlin の例)
 
"""
// Android での Gemini Nano 利用 (Kotlin)
import com.google.ai.client.generativeai.GenerativeModel
 
val generativeModel = GenerativeModel(
    modelName = "gemini-nano",
    // APIキー不要 - デバイス上で直接実行
)
 
// テキスト生成 (デバイス上で完結)
val response = generativeModel.generateContent("こんにちは")
println(response.text)
 
// Gemini Nano の利点:
// - ネットワーク不要 (飛行機モードでも動作)
// - データがデバイス外に出ない (完全なプライバシー)
// - レイテンシが極めて低い (ネットワーク遅延なし)
// - API 料金が発生しない
 
// 対応機能:
// - テキスト要約
// - スマートリプライ
// - 文法チェック
// - テキスト分類
"""
 
# Python からの Gemini Nano 利用 (Chrome の組み込み AI)
# Web 版の Gemini Nano は Chrome Canary で実験的に利用可能
"""
// JavaScript (Chrome Built-in AI)
const session = await ai.languageModel.create({
  systemPrompt: "あなたは親切なアシスタントです。"
});
 
const result = await session.prompt("明日の予定をリマインドして");
console.log(result);
 
// ストリーミング
const stream = session.promptStreaming("長い文章を生成してください");
for await (const chunk of stream) {
  console.log(chunk);
}
"""

import google.generativeai as genai
 
genai.configure(api_key="YOUR_API_KEY")
 
# Thinking Mode を使った高精度推論
model = genai.GenerativeModel("gemini-2.0-flash-thinking-exp")
 
response = model.generate_content(
    """
    以下の数学問題を段階的に解いてください:
 
    ある工場で、製品Aと製品Bを生産しています。
    - 製品Aは1個あたり3時間の加工と2時間の検査が必要
    - 製品Bは1個あたり2時間の加工と4時間の検査が必要
    - 1日の加工可能時間は18時間、検査可能時間は20時間
    - 製品Aの利益は500円、製品Bの利益は400円
 
    利益を最大化するための最適な生産量を求めてください。
    """
)
 
# Thinking Mode では思考過程と最終回答が分かれる
for part in response.parts:
    print(part.text)

import google.generativeai as genai
from google.api_core import exceptions
 
genai.configure(api_key="YOUR_API_KEY")
model = genai.GenerativeModel("gemini-1.5-pro")
 
 
# エラー 1: レート制限
def handle_rate_limit(prompt: str, max_retries: int = 5):
    """指数バックオフでレート制限に対応"""
    import time
    import random
 
    for attempt in range(max_retries):
        try:
            response = model.generate_content(prompt)
            return response.text
        except exceptions.ResourceExhausted as e:
            if attempt < max_retries - 1:
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"レート制限。{wait_time:.1f}秒待機... (試行 {attempt+1}/{max_retries})")
                time.sleep(wait_time)
            else:
                raise
 
 
# エラー 2: Safety フィルタによるブロック
def handle_safety_block(prompt: str):
    """Safety フィルタの診断と対応"""
    try:
        response = model.generate_content(prompt)
 
        # Safety ratings の確認
        if response.prompt_feedback.block_reason:
            print(f"ブロック理由: {response.prompt_feedback.block_reason}")
            for rating in response.prompt_feedback.safety_ratings:
                print(f"  {rating.category}: {rating.probability}")
            return None
 
        # 候補の Safety チェック
        for candidate in response.candidates:
            if candidate.finish_reason.name == "SAFETY":
                print("回答が Safety フィルタによりブロックされました")
                for rating in candidate.safety_ratings:
                    print(f"  {rating.category}: {rating.probability}")
                return None
 
        return response.text
 
    except exceptions.InvalidArgument as e:
        print(f"不正なリクエスト: {e}")
        return None
 
 
# エラー 3: トークン数超過
def check_token_count(content: str, model_name: str = "gemini-1.5-pro"):
    """事前にトークン数をチェック"""
    model = genai.GenerativeModel(model_name)
 
    token_count = model.count_tokens(content)
    print(f"トークン数: {token_count.total_tokens}")
 
    limits = {
        "gemini-1.5-pro": 2_000_000,
        "gemini-1.5-flash": 1_000_000,
        "gemini-2.0-flash": 1_000_000,
    }
 
    limit = limits.get(model_name, 128_000)
    if token_count.total_tokens > limit:
        print(f"警告: トークン数が上限 ({limit:,}) を超えています")
        return False
    else:
        remaining = limit - token_count.total_tokens
        print(f"残りトークン: {remaining:,}")
        return True
 
 
# エラー 4: ファイルアップロードの失敗
def safe_upload_file(file_path: str, max_retries: int = 3):
    """ファイルアップロードの堅牢な実装"""
    import os
    import time
 
    # ファイルサイズチェック (最大 2GB)
    file_size = os.path.getsize(file_path)
    if file_size > 2 * 1024 * 1024 * 1024:
        raise ValueError(f"ファイルサイズが上限を超えています: {file_size / (1024**3):.1f}GB")
 
    for attempt in range(max_retries):
        try:
            uploaded = genai.upload_file(file_path)
 
            # 動画の場合は処理完了を待機
            while uploaded.state.name == "PROCESSING":
                print(f"処理中... ({attempt+1}回目)")
                time.sleep(10)
                uploaded = genai.get_file(uploaded.name)
 
            if uploaded.state.name == "ACTIVE":
                return uploaded
            else:
                print(f"アップロード失敗: state={uploaded.state.name}")
 
        except Exception as e:
            print(f"アップロードエラー (試行 {attempt+1}): {e}")
            time.sleep(5)
 
    raise RuntimeError(f"ファイルアップロードが{max_retries}回失敗しました")

import google.generativeai as genai
import json
import time
from datetime import datetime
 
genai.configure(api_key="YOUR_API_KEY")
 
 
class GeminiMonitor:
    """Gemini API の利用状況モニタリング"""
 
    def __init__(self, model_name: str = "gemini-1.5-pro"):
        self.model = genai.GenerativeModel(model_name)
        self.model_name = model_name
        self.call_log = []
 
    def generate(self, prompt: str, **kwargs) -> str:
        """モニタリング付きの生成"""
        start_time = time.time()
 
        try:
            # トークン数の事前チェック
            token_count = self.model.count_tokens(prompt)
            input_tokens = token_count.total_tokens
 
            response = self.model.generate_content(prompt, **kwargs)
            latency = time.time() - start_time
 
            # 出力トークン数の取得
            output_tokens = self.model.count_tokens(response.text).total_tokens
 
            log_entry = {
                "timestamp": datetime.now().isoformat(),
                "model": self.model_name,
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
                "latency_seconds": round(latency, 3),
                "status": "success",
                "finish_reason": response.candidates[0].finish_reason.name,
            }
            self.call_log.append(log_entry)
 
            return response.text
 
        except Exception as e:
            latency = time.time() - start_time
            log_entry = {
                "timestamp": datetime.now().isoformat(),
                "model": self.model_name,
                "latency_seconds": round(latency, 3),
                "status": "error",
                "error": str(e),
            }
            self.call_log.append(log_entry)
            raise
 
    def get_statistics(self) -> dict:
        """利用統計を取得"""
        if not self.call_log:
            return {"message": "No calls recorded"}
 
        successful = [l for l in self.call_log if l["status"] == "success"]
        failed = [l for l in self.call_log if l["status"] == "error"]
 
        total_input = sum(l.get("input_tokens", 0) for l in successful)
        total_output = sum(l.get("output_tokens", 0) for l in successful)
        avg_latency = sum(l["latency_seconds"] for l in successful) / len(successful) if successful else 0
 
        return {
            "total_calls": len(self.call_log),
            "successful": len(successful),
            "failed": len(failed),
            "total_input_tokens": total_input,
            "total_output_tokens": total_output,
            "avg_latency_seconds": round(avg_latency, 3),
            "error_rate": round(len(failed) / len(self.call_log) * 100, 1),
        }
 
 
# 使用例
monitor = GeminiMonitor("gemini-1.5-flash")
 
prompts = [
    "AIの未来について100字で述べてください",
    "Pythonのデコレータを説明してください",
    "クイックソートのアルゴリズムを説明してください",
]
 
for p in prompts:
    result = monitor.generate(p)
    print(f"Response: {result[:80]}...\n")
 
stats = monitor.get_statistics()
print(f"\n=== 統計 ===")
print(json.dumps(stats, indent=2, ensure_ascii=False))

import google.generativeai as genai
import time
 
genai.configure(api_key="YOUR_API_KEY")
 
 
# 最適化 1: ストリーミングで TTFB を最小化
def optimized_streaming(prompt: str):
    """ストリーミングによるレイテンシ最適化"""
    model = genai.GenerativeModel("gemini-1.5-flash")
 
    start = time.time()
    first_token_time = None
 
    response = model.generate_content(prompt, stream=True)
    for chunk in response:
        if first_token_time is None:
            first_token_time = time.time()
            ttfb = first_token_time - start
            print(f"TTFB: {ttfb:.3f}s")
        print(chunk.text, end="", flush=True)
 
    total_time = time.time() - start
    print(f"\n総時間: {total_time:.3f}s")
 
 
# 最適化 2: 適切なモデル選択
# Flash は Pro の 3-5 倍高速
performance_comparison = {
    "gemini-1.5-pro": {
        "avg_ttfb": "1.5-3.0s",
        "tokens_per_second": "30-50",
        "best_for": "高精度タスク",
    },
    "gemini-1.5-flash": {
        "avg_ttfb": "0.3-0.8s",
        "tokens_per_second": "80-150",
        "best_for": "リアルタイム応答",
    },
    "gemini-2.0-flash": {
        "avg_ttfb": "0.2-0.6s",
        "tokens_per_second": "100-200",
        "best_for": "最新・最速",
    },
}
 
 
# 最適化 3: プロンプト最適化
def optimize_prompt(prompt: str) -> str:
    """トークン数を削減してコスト・速度を改善"""
    # 不要な空白の除去
    import re
    prompt = re.sub(r'\n{3,}', '\n\n', prompt)
    prompt = re.sub(r' {2,}', ' ', prompt)
 
    # 冗長な表現の除去
    replacements = {
        "できるだけ詳しく説明してください": "詳しく説明してください",
        "以下の内容について": "",
        "よろしくお願いします": "",
    }
    for old, new in replacements.items():
        prompt = prompt.replace(old, new)
 
    return prompt.strip()
 
 
# 最適化 4: Context Caching で繰り返しコストを削減
# (前述の 3.7 セクション参照)

# NG: 200万トークンあるからと全文書を無制限に投入
all_docs = load_all_company_documents()  # 300万トークン分
response = model.generate_content(all_docs + [query])
# → コンテキスト超過エラー or 精度低下 (中間部分の "Lost in the Middle" 問題)
 
# OK: 関連文書のみ抽出してから投入
relevant_docs = retrieve_relevant(query, all_docs, top_k=20)
response = model.generate_content(relevant_docs + [query])
# → 必要な情報に絞ることで精度向上 + コスト削減

# NG: 簡単な分類タスクに最高性能モデルを使用
model = genai.GenerativeModel("gemini-ultra")
for item in million_items:  # 100万件処理
    result = model.generate_content(f"カテゴリ分類: {item}")
# → 膨大なコスト発生
 
# OK: タスク難易度に応じたモデル選択
model = genai.GenerativeModel("gemini-1.5-flash")  # 分類タスクには Flash で十分
# さらに: バッチ API を使ってコスト削減

# NG: 安全性設定を全て無効化
safety = {cat: "BLOCK_NONE" for cat in all_categories}
# → 有害コンテンツ生成リスク、利用規約違反の可能性
 
# OK: ユースケースに応じた適切な設定
safety = {
    "HARM_CATEGORY_HARASSMENT": "BLOCK_MEDIUM_AND_ABOVE",
    "HARM_CATEGORY_DANGEROUS_CONTENT": "BLOCK_ONLY_HIGH",  # 医療用途など必要に応じて緩和
}

# NG: エラーハンドリングなしで本番運用
response = model.generate_content(prompt)
result = response.text  # Noneの場合にクラッシュ
 
# OK: 堅牢なエラーハンドリング
try:
    response = model.generate_content(prompt)
 
    # Safety ブロックの確認
    if not response.candidates:
        print("回答が生成されませんでした (Safety フィルタ)")
        result = fallback_response()
    elif response.candidates[0].finish_reason.name != "STOP":
        print(f"異常終了: {response.candidates[0].finish_reason.name}")
        result = fallback_response()
    else:
        result = response.text
 
except exceptions.ResourceExhausted:
    # レート制限 → リトライ
    result = retry_with_backoff(prompt)
except exceptions.InvalidArgument as e:
    # 不正なリクエスト → ログ記録
    log_error(e, prompt)
    result = fallback_response()
except Exception as e:
    # その他のエラー
    log_error(e, prompt)
    result = fallback_response()

# NG: 小さなコンテンツをキャッシュ (最低 32K tokens 必要)
cache = caching.CachedContent.create(
    model="models/gemini-1.5-pro-002",
    contents=["短いテキスト"],  # 100 tokens しかない
    ttl=datetime.timedelta(hours=24),
)
# → エラー: トークン数不足
 
# NG: TTL を長すぎに設定 (ストレージコストが発生)
cache = caching.CachedContent.create(
    model="models/gemini-1.5-pro-002",
    contents=[huge_document],
    ttl=datetime.timedelta(days=30),  # 30日 → 膨大なストレージ費用
)
 
# OK: 適切なサイズとTTLの設定
cache = caching.CachedContent.create(
    model="models/gemini-1.5-pro-002",
    contents=[document_over_32k_tokens],
    ttl=datetime.timedelta(hours=2),  # 必要な時間だけ
)
# クエリ完了後に明示的に削除
cache.delete()