Gemini API の Function Calling 実装 — 並列実行・ツールチェーン・Managed Agents の使い分け

import google.generativeai as genai
import json
import requests
 
genai.configure(api_key="YOUR_API_KEY")
 
# ツールの定義
tools = [
    {
        "function_declarations": [
            {
                "name": "get_weather",
                "description": "指定した都市の現在の天気情報を取得します",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "都市名（例: Tokyo, Osaka）"
                        },
                        "unit": {
                            "type": "string",
                            "enum": ["celsius", "fahrenheit"],
                            "description": "温度の単位"
                        }
                    },
                    "required": ["city"]
                }
            }
        ]
    }
]
 
# 実際の関数実装
def get_weather(city: str, unit: str = "celsius") -> dict:
    """実際のWeather APIを呼ぶ（ここではモック）"""
    # 実際には OpenWeatherMap 等のAPIを使う
    return {
        "city": city,
        "temperature": 22 if unit == "celsius" else 72,
        "condition": "晴れ",
        "humidity": 65,
        "unit": unit
    }
 
# Geminiとの対話ループ
def chat_with_tools(user_message: str) -> str:
    model = genai.GenerativeModel(
        model_name="gemini-2.0-flash",
        tools=tools
    )
 
    response = model.generate_content(user_message)
 
    # Function Calling が必要な場合
    if response.candidates[0].content.parts[0].function_call:
        fc = response.candidates[0].content.parts[0].function_call
        func_name = fc.name
        func_args = dict(fc.args)
 
        # 関数を実行
        if func_name == "get_weather":
            result = get_weather(**func_args)
 
        # 結果をGeminiに返す
        response = model.generate_content([
            user_message,
            response.candidates[0].content,
            {
                "role": "user",
                "parts": [{
                    "function_response": {
                        "name": func_name,
                        "response": result
                    }
                }]
            }
        ])
 
    return response.text
 
# 使用例
print(chat_with_tools("東京の今日の天気を教えてください"))

# 複数ツールの定義
tools = [
    {
        "function_declarations": [
            {
                "name": "search_products",
                "description": "商品データベースから商品を検索する",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "検索キーワード"},
                        "category": {"type": "string", "description": "商品カテゴリ"},
                        "max_price": {"type": "number", "description": "最大価格（円）"},
                        "limit": {"type": "integer", "description": "取得件数（デフォルト: 5）"}
                    },
                    "required": ["query"]
                }
            },
            {
                "name": "get_inventory",
                "description": "商品の在庫数を取得する",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "product_id": {"type": "string", "description": "商品ID"}
                    },
                    "required": ["product_id"]
                }
            },
            {
                "name": "create_order",
                "description": "注文を作成する",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "product_id": {"type": "string"},
                        "quantity": {"type": "integer"},
                        "user_id": {"type": "string"}
                    },
                    "required": ["product_id", "quantity", "user_id"]
                }
            }
        ]
    }
]

import asyncio
 
async def handle_parallel_tool_calls(response, available_functions: dict):
    """複数のFunction Callを並列処理する"""
    parts = response.candidates[0].content.parts
 
    # Function Callを全て抽出
    function_calls = [
        part.function_call
        for part in parts
        if hasattr(part, 'function_call') and part.function_call
    ]
 
    if not function_calls:
        return None
 
    # 並列実行
    async def execute_function(fc):
        func = available_functions.get(fc.name)
        if func:
            if asyncio.iscoroutinefunction(func):
                return await func(**dict(fc.args))
            else:
                return func(**dict(fc.args))
        return {"error": f"Function {fc.name} not found"}
 
    results = await asyncio.gather(*[execute_function(fc) for fc in function_calls])
 
    # 結果をFunction Responseとして整形
    function_responses = [
        {
            "function_response": {
                "name": fc.name,
                "response": result
            }
        }
        for fc, result in zip(function_calls, results)
    ]
 
    return function_responses
 
# 使用例
async def main():
    model = genai.GenerativeModel("gemini-2.0-flash", tools=tools)
 
    # 複数情報を同時に必要とする質問
    response = model.generate_content(
        "iPhone 15 Pro の在庫状況と、3万円以下のスマートフォンのおすすめを同時に教えてください"
    )
 
    available_functions = {
        "search_products": search_products,
        "get_inventory": get_inventory,
    }
 
    function_responses = await handle_parallel_tool_calls(response, available_functions)
 
    if function_responses:
        # 全ての結果をGeminiに返す
        final_response = model.generate_content([
            "iPhone 15 Pro の在庫状況と、3万円以下のスマートフォンのおすすめを同時に教えてください",
            response.candidates[0].content,
            {"role": "user", "parts": function_responses}
        ])
        print(final_response.text)
 
asyncio.run(main())

from dataclasses import dataclass
from typing import Any, Optional
import time
 
@dataclass
class ToolResult:
    success: bool
    data: Optional[Any] = None
    error: Optional[str] = None
 
def execute_with_retry(
    func,
    args: dict,
    max_retries: int = 3,
    backoff_factor: float = 1.5
) -> ToolResult:
    """リトライ付きのツール実行"""
    last_error = None
 
    for attempt in range(max_retries):
        try:
            result = func(**args)
            return ToolResult(success=True, data=result)
 
        except ConnectionError as e:
            last_error = str(e)
            wait_time = backoff_factor ** attempt
            print(f"接続エラー (試行 {attempt + 1}/{max_retries}): {wait_time:.1f}秒後にリトライ")
            time.sleep(wait_time)
 
        except ValueError as e:
            # バリデーションエラーはリトライしない
            return ToolResult(success=False, error=f"引数エラー: {str(e)}")
 
        except Exception as e:
            last_error = str(e)
            if attempt == max_retries - 1:
                break
 
    return ToolResult(success=False, error=f"最大リトライ回数超過: {last_error}")
 
def safe_tool_handler(fc, available_functions: dict) -> dict:
    """エラーセーフなツールハンドラー"""
    func = available_functions.get(fc.name)
 
    if not func:
        return {"error": f"未定義のツール: {fc.name}"}
 
    result = execute_with_retry(func, dict(fc.args))
 
    if result.success:
        return result.data
    else:
        # エラーをGeminiに伝えて代替案を提案させる
        return {
            "error": result.error,
            "fallback_suggestion": "情報を取得できませんでした。別の方法で回答を試みます。"
        }

def run_tool_chain(initial_query: str, max_steps: int = 5) -> str:
    """
    複数ステップのツールチェーンを自動実行するエージェントループ
    """
    model = genai.GenerativeModel("gemini-2.0-flash", tools=tools)
    conversation = [{"role": "user", "parts": [initial_query]}]
 
    for step in range(max_steps):
        response = model.generate_content(conversation)
        parts = response.candidates[0].content.parts
 
        # テキスト応答の場合は終了
        if all(hasattr(p, 'text') for p in parts):
            return parts[0].text
 
        # Function Callsを処理
        tool_responses = []
        for part in parts:
            if hasattr(part, 'function_call') and part.function_call:
                fc = part.function_call
                result = safe_tool_handler(fc, available_functions)
                tool_responses.append({
                    "function_response": {
                        "name": fc.name,
                        "response": result
                    }
                })
 
        # 会話履歴に追加してループ継続
        conversation.append({
            "role": "model",
            "parts": parts
        })
        conversation.append({
            "role": "user",
            "parts": tool_responses
        })
 
    return "最大ステップ数に達しました"
 
# 使用例: 複数ステップが必要なタスク
result = run_tool_chain(
    "在庫が10個以上ある3万円以下のスマートフォンを探して、最安値のものを注文してください"
)
print(result)

import { GoogleGenerativeAI, FunctionDeclaration } from "@google/generative-ai";
 
const genAI = new GoogleGenerativeAI(process.env.GEMINI_API_KEY!);
 
// 型安全なツール定義
const searchProductsTool: FunctionDeclaration = {
  name: "search_products",
  description: "商品データベースから商品を検索する",
  parameters: {
    type: "object" as const,
    properties: {
      query: { type: "string", description: "検索キーワード" },
      maxPrice: { type: "number", description: "最大価格" },
    },
    required: ["query"],
  },
};
 
async function geminiWithTools(userMessage: string): Promise<string> {
  const model = genAI.getGenerativeModel({
    model: "gemini-2.0-flash",
    tools: [{ functionDeclarations: [searchProductsTool] }],
  });
 
  const chat = model.startChat();
  let response = await chat.sendMessage(userMessage);
 
  // ツール呼び出しループ
  while (true) {
    const functionCalls = response.response.functionCalls();
 
    if (!functionCalls || functionCalls.length === 0) {
      return response.response.text();
    }
 
    // 全てのFunction Callを処理
    const toolResults = await Promise.all(
      functionCalls.map(async (fc) => {
        let result: unknown;
 
        switch (fc.name) {
          case "search_products":
            result = await searchProducts(fc.args as { query: string; maxPrice?: number });
            break;
          default:
            result = { error: `Unknown tool: ${fc.name}` };
        }
 
        return {
          functionResponse: {
            name: fc.name,
            response: result,
          },
        };
      })
    );
 
    response = await chat.sendMessage(toolResults);
  }
}

model = genai.GenerativeModel("gemini-flash-latest", tools=tools)
response = model.generate_content(user_message)
# どのモデルが選ばれたかはログに残らない

import logging
 
PINNED_MODEL = "gemini-3.5-flash"  # GA 版を明示。世代交代時はここだけ変える
 
def call_with_tools(user_message: str):
    model = genai.GenerativeModel(PINNED_MODEL, tools=tools)
    response = model.generate_content(user_message)
 
    served = getattr(response, "model_version", None)
    if served and not served.startswith(PINNED_MODEL):
        # 要求 != 応答。既定差し替わり・自動アップグレードを検知
        logging.warning("model drift: requested=%s served=%s", PINNED_MODEL, served)
 
    return response