Gemini API Grounding × Structured Output — Web検索結果をJSON構造化データに変換する

from google import genai
from google.genai import types
import json
from datetime import datetime
 
client = genai.Client()
 
PRICE_SCHEMA = {
    "type": "object",
    "properties": {
        "products": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "product_name": {"type": "string"},
                    "vendor": {"type": "string"},
                    "price": {"type": "number"},
                    "currency": {"type": "string"},
                    "in_stock": {"type": "boolean"},
                    "url": {"type": "string"}
                },
                "required": ["product_name", "vendor", "price", "currency"]
            }
        },
        "search_timestamp": {"type": "string"},
        "confidence_note": {"type": "string"}
    },
    "required": ["products", "search_timestamp"]
}
 
def fetch_competitor_prices(product_query: str) -> dict:
    """競合価格をリアルタイム検索して構造化データで返す"""
    try:
        response = client.models.generate_content(
            model="gemini-2.5-flash",
            contents=(
                f"以下の製品について、主要なオンラインストアでの"
                f"現在の販売価格を調べてください。"
                f"価格は税込み表示で、在庫状況も含めてください。\n\n"
                f"対象製品: {product_query}\n\n"
                f"検索対象ストア: Amazon.co.jp, 楽天市場, Yahoo\!ショッピング"
            ),
            config=types.GenerateContentConfig(
                tools=[types.Tool(google_search=types.GoogleSearch())],
                response_mime_type="application/json",
                response_schema=PRICE_SCHEMA,
                temperature=0.1  # 事実ベースのデータには低温度を推奨
            )
        )
 
        data = json.loads(response.text)
        data["search_timestamp"] = datetime.now().isoformat()
 
        # 出典URLをメタデータから補完
        if response.candidates[0].grounding_metadata:
            source_urls = [
                chunk.web.uri
                for chunk in response.candidates[0].grounding_metadata.grounding_chunks
            ]
            data["source_urls"] = source_urls
 
        return data
 
    except Exception as e:
        return {
            "products": [],
            "search_timestamp": datetime.now().isoformat(),
            "confidence_note": f"検索エラー: {str(e)}"
        }
 
# 実行
result = fetch_competitor_prices("Sony WH-1000XM6 ワイヤレスヘッドホン")
print(json.dumps(result, ensure_ascii=False, indent=2))
# 出力例:
# {
#   "products": [
#     {
#       "product_name": "Sony WH-1000XM6",
#       "vendor": "Amazon.co.jp",
#       "price": 49500,
#       "currency": "JPY",
#       "in_stock": true,
#       "url": "https://www.amazon.co.jp/dp/..."
#     }
#   ],
#   "search_timestamp": "2026-04-12T15:30:00",
#   "source_urls": ["https://..."]
# }

schema = {
    "type": "object",
    "properties": {
        "price": {
            "type": "number",
            "description": "税込み価格（日本円）。見つからない場合は0"
        },
        "last_confirmed": {
            "type": "string",
            "description": "価格が確認された日付。YYYY-MM-DD形式。不明ならnull"
        }
    }
}

response = client.models.generate_content(
    model="gemini-2.5-pro",
    contents="このページに掲載されているイベント情報を抽出してください",
    config=types.GenerateContentConfig(
        tools=[
            types.Tool(url_context=types.UrlContext(
                urls=["https://example.com/events/2026-spring"]
            )),
            types.Tool(google_search=types.GoogleSearch())
        ],
        response_mime_type="application/json",
        response_schema={
            "type": "object",
            "properties": {
                "events": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "event_name": {"type": "string"},
                            "date": {"type": "string"},
                            "location": {"type": "string"},
                            "ticket_price": {"type": "string"}
                        },
                        "required": ["event_name", "date"]
                    }
                }
            }
        }
    )
)

from google import genai
from google.genai import types
import json
 
client = genai.Client()
 
MIN_SUPPORT = 1          # 1件以上の出典が必要
CONF_THRESHOLD = 0.6     # 信頼度スコアの下限（提供される場合）
 
def extract_with_provenance(prompt: str, schema: dict):
    resp = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(
            tools=[types.Tool(google_search=types.GoogleSearch())],
            response_mime_type="application/json",
            response_schema=schema,
        ),
    )
    cand = resp.candidates[0]
    data = json.loads(cand.content.parts[0].text)
    meta = cand.grounding_metadata
 
    # 出典 URL の集合（空なら「検索が走らなかった」とみなす）
    sources = []
    if meta and meta.grounding_chunks:
        sources = [c.web.uri for c in meta.grounding_chunks if c.web]
 
    # 出力テキストのうち、出典に支えられた区間の信頼度を集計
    supports = meta.grounding_supports if (meta and meta.grounding_supports) else []
    backed = len(sources) >= MIN_SUPPORT
    confident = True
    for sup in supports:
        scores = getattr(sup, "confidence_scores", None) or []
        if scores and max(scores) < CONF_THRESHOLD:
            confident = False
            break
 
    verdict = "accepted" if (backed and confident) else "quarantined"
    return {
        "verdict": verdict,
        "data": data,
        "sources": sources,
        "reason": (
            "ok" if verdict == "accepted"
            else ("no_grounding_source" if not backed else "low_confidence")
        ),
    }
 
 
schema = {
    "type": "object",
    "properties": {
        "price": {"type": "number", "description": "税込み価格（日本円）。不明なら0"},
        "currency": {"type": "string"},
    },
    "required": ["price", "currency"],
}
 
result = extract_with_provenance(
    "現在の某クラウドストレージ 1TB プランの月額を調べてJSONで返してください",
    schema,
)
 
if result["verdict"] == "accepted":
    save_to_production(result["data"], result["sources"])   # 本番テーブルへ
else:
    save_to_review_queue(result)                            # 隔離キューへ
    print("隔離:", result["reason"])