Gemini 2.5 Pro Function Calling实战：Tool Use自動化ワークフロー完全ガイド

こんにちは、HolySheep AI(今すぐ登録)のシニアAI統合エンジニア、田中です。本日はGemini 2.5 ProのFunction Calling機能を活用したツール呼び出しの自動化ワークフローについて、 실무で遭遇した具体的なエラーケースめながら詳しく解説します。

私は2024年末から複数のプロダクション環境でGemini 2.5 Proを運用していますが、Function Callingの設定で{" ConnectionError: timeout"}や{"401 Unauthorized"}に直面した経験が很多人的と思います。本記事ではこれらの厄介なエラーを解決しながら、高效な自動化ワークフローを構築する方法を紹介します。

Function Callingとは？なぜ重要か

Function Calling（関数呼び出し）は、LLMが外部APIやデータベースと安全に連携するための仕組みです。Gemini 2.5 Proでは{"tools"}パラメータを使用して、モデルが呼び出すべき関数を定義できます。これにより、...

リアルタイム天気情報の取得
データベースのクエリ実行
外部APIとの 안전한連携
複雑なワークフローの自动化

が可能になります。HolySheep AIのGemini 2.5 ProはTTOK辺り$3.50という競爭力のある 가격으로 제공되며、レートは¥1=$1（公式¥7.3=$1の85%節約）でご利用いただけます。

環境構築と前提条件

必要なライブラリ

# 必要なパッケージのインストール
pip install openai httpx python-dotenv

バージョン確認
python --version  # Python 3.9以上を推奨
pip list | grep -E "openai|httpx"

プロジェクト構造

project/
├── .env                 # APIキー管理
├── tools/
│   ├── __init__.py
│   ├── weather.py       # 天気情報取得ツール
│   ├── database.py      # データベース操作ツール
│   └── notification.py  # 通知ツール
├── workflow.py          # メイン自動化スクリプト
└── requirements.txt

実践的なFunction Calling実装

ここからは私が実際に運用しているワークフローを基に、段階的に実装方法を説明します。

Step 1: 基本設定とAPI接続

import os
from openai import OpenAI
from dotenv import load_dotenv

.envファイルからAPIキー読み込み
load_dotenv()

HolySheep AI設定
client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"  # 必ずこのエンドポイントを使用
)

接続確認
def test_connection():
    try:
        response = client.chat.completions.create(
            model="gemini-2.5-pro",
            messages=[{"role": "user", "content": "hello"}],
            max_tokens=10
        )
        print(f"✅ 接続成功: {response.choices[0].message.content}")
        return True
    except Exception as e:
        print(f"❌ 接続エラー: {type(e).__name__}: {e}")
        return False

if __name__ == "__main__":
    test_connection()

私は最初にここで{"ConnectionError: timeout"}に遭遇しました。原因是プロキシ設定の缺失です。

Step 2: 関数の定義（Tools宣言）

# tools/weather.py
"""天気情報取得ツール"""

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "指定した都市の現在、天気情報を取得します",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "都市名（例：東京、ニューヨーク）"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度の単位"
                    }
                },
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_forecast",
            "description": "指定した都市の5日間予報を取得します",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "都市名"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

関数実装
def get_weather(location: str, unit: str = "celsius") -> dict:
    """実際は外部API（OpenWeatherMap等）を呼び出す"""
    # モックデータ
    return {
        "location": location,
        "temperature": 22 if unit == "celsius" else 72,
        "condition": "晴れ",
        "humidity": 65,
        "timestamp": "2025-01-15T10:30:00Z"
    }

def get_forecast(location: str) -> dict:
    """5日間予報を取得"""
    return {
        "location": location,
        "forecast": [
            {"day": 1, "temp": 22, "condition": "晴れ"},
            {"day": 2, "temp": 18, "condition": "曇り"},
            {"day": 3, "temp": 15, "condition": "雨"},
            {"day": 4, "temp": 20, "condition": "晴れ"},
            {"day": 5, "temp": 24, "condition": "快晴"}
        ]
    }

Step 3: 完全自动化ワークフロー

import json
from tools.weather import TOOLS, get_weather, get_forecast

def execute_workflow(user_query: str):
    """Function Callingを活用した自動化ワークフロー"""
    
    messages = [
        {"role": "system", "content": "あなたは有用的な天気アシスタントです。"},
        {"role": "user", "content": user_query}
    ]
    
    # 最初のリクエスト
    response = client.chat.completions.create(
        model="gemini-2.5-pro",
        messages=messages,
        tools=TOOLS,
        tool_choice="auto"
    )
    
    assistant_message = response.choices[0].message
    messages.append(assistant_message)
    
    # 関数呼び出しがある場合
    while assistant_message.tool_calls:
        for tool_call in assistant_message.tool_calls:
            function_name = tool_call.function.name
            arguments = json.loads(tool_call.function.arguments)
            
            print(f"🔧 関数呼び出し: {function_name}")
            print(f"   引数: {arguments}")
            
            # 関数の実行
            if function_name == "get_weather":
                result = get_weather(**arguments)
            elif function_name == "get_forecast":
                result = get_forecast(**arguments)
            else:
                result = {"error": "不明な関数"}
            
            # 結果をメッセージに追加
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })
        
        # 結果を受けて再度LLMに送信
        response = client.chat.completions.create(
            model="gemini-2.5-pro",
            messages=messages,
            tools=TOOLS
        )
        
        assistant_message = response.choices[0].message
        messages.append(assistant_message)
    
    return assistant_message.content

実行例
if __name__ == "__main__":
    result = execute_workflow("東京の今日の天気と明日の予報を教えてください")
    print(f"\n📝 最终回答:\n{result}")

実践事例：マルチツール連携ワークフロー

ここからは私が実際に構築した、より複雑なマルチツール連携の例を紹介します。

# tools/database.py
"""データベース操作ツール"""

DB_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "query_orders",
            "description": "注文データを検索します",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "status": {
                        "type": "string", 
                        "enum": ["pending", "shipped", "delivered"]
                    },
                    "limit": {"type": "integer", "default": 10}
                }
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "update_order_status",
            "description": "注文ステータスを更新します",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "new_status": {"type": "string"}
                },
                "required": ["order_id", "new_status"]
            }
        }
    }
]

def query_orders(customer_id: str = None, status: str = None, limit: int = 10):
    """注文データのクエリ（実際はSQL/ORMを使用）"""
    return {
        "orders": [
            {"order_id": "ORD-001", "customer": "佐藤太郎", "total": 15000, "status": "shipped"},
            {"order_id": "ORD-002", "customer": "鈴木花子", "total": 8500, "status": "pending"}
        ],
        "total_count": 2
    }

def update_order_status(order_id: str, new_status: str):
    """注文ステータス更新"""
    return {"success": True, "order_id": order_id, "new_status": new_status}

# complete_automation.py
"""完全自动化ワークフロー - 天気×注文管理×通知"""

from openai import OpenAI
import json
import os
from datetime import datetime

client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1"
)

全ツール定義
ALL_TOOLS = [
    # 天気ツール
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "都市の天気取得",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["location"]
            }
        }
    },
    # 注文管理ツール
    {
        "type": "function",
        "function": {
            "name": "query_orders",
            "description": "注文検索",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                    "status": {"type": "string", "enum": ["pending", "shipped", "delivered"]}
                }
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "send_notification",
            "description": "ユーザーに通知を送信",
            "parameters": {
                "type": "object",
                "properties": {
                    "user_id": {"type": "string"},
                    "message": {"type": "string"},
                    "channel": {"type": "string", "enum": ["email", "sms", "push"]}
                },
                "required": ["user_id", "message"]
            }
        }
    }
]

def send_notification(user_id: str, message: str, channel: str = "email"):
    """通知送信（実際はSendGrid/Twilio等を使用）"""
    print(f"📧 通知送信: {user_id} via {channel}")
    return {"success": True, "sent_at": datetime.now().isoformat()}

def run_automated_workflow(query: str):
    """自动化ワークフロー実行"""
    messages = [
        {"role": "system", "content": """あなたは注文管理アシスタントです。
        ユーザーの要求に応じて、ツールを呼び出してください。
        天気によって配送状況を判断し、適切な通知を行います。"""},
        {"role": "user", "content": query}
    ]
    
    max_iterations = 10
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        
        response = client.chat.completions.create(
            model="gemini-2.5-pro",
            messages=messages,
            tools=ALL_TOOLS,
            tool_choice="auto"
        )
        
        choice = response.choices[0].message
        messages.append(choice)
        
        if not choice.tool_calls:
            print(f"✅ ワークフロー完了（{iteration}イテレーション）")
            return choice.content
        
        # ツール呼び出し処理
        for tool_call in choice.tool_calls:
            fn_name = tool_call.function.name
            fn_args = json.loads(tool_call.function.arguments)
            
            print(f"🔧 実行: {fn_name}({fn_args})")
            
            # 関数実行
            if fn_name == "get_weather":
                result = {"temperature": 22, "condition": "晴れ", "location": fn_args.get("location")}
            elif fn_name == "query_orders":
                result = {"orders": [{"id": "123", "status": "shipped"}]}
            elif fn_name == "send_notification":
                result = send_notification(**fn_args)
            else:
                result = {"error": "Unknown function"}
            
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False)
            })

実行
if __name__ == "__main__":
    result = run_automated_workflow(
        "東京都墨田区在住の客户の保留中注文を確認し、"
        "天気が悪い場合は配送を延期する旨を通知してください"
    )
    print(f"\n📝 結果:\n{result}")

エラー處理とデバッグ技巧

レイテンシ測定ユーティリティ

import time
import httpx

def measure_latency(endpoint: str, api_key: str) -> dict:
    """APIレイテンシ測定"""
    latencies = []
    
    for i in range(5):
        start = time.time()
        try:
            with httpx.Client(timeout=30.0) as client:
                response = client.post(
                    f"{endpoint}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {api_key}",
                        "Content-Type": "application/json"
                    },
                    json={
                        "model": "gemini-2.5-pro",
                        "messages": [{"role": "user", "content": "ping"}],
                        "max_tokens": 5
                    }
                )
                elapsed = (time.time() - start) * 1000
                latencies.append(elapsed)
                print(f"  Request {i+1}: {elapsed:.1f}ms (status: {response.status_code})")
        except Exception as e:
            print(f"  Request {i+1}: ERROR - {e}")
    
    return {
        "avg": sum(latencies) / len(latencies),
        "min": min(latencies),
        "max": max(latencies)
    }

測定実行
metrics = measure_latency(
    endpoint="https://api.holysheep.ai/v1",
    api_key=os.getenv("HOLYSHEEP_API_KEY")
)
print(f"\n📊 レイテンシ結果:")
print(f"   平均: {metrics['avg']:.1f}ms")
print(f"   最小: {metrics['min']:.1f}ms")
print(f"   最大: {metrics['max']:.1f}ms")

よくあるエラーと対処法

エラー1: {"ConnectionError: timeout"}

症状：リクエスト送信時にタイムアウトエラーが発生する

原因：ネットワークプロキシの設定缺失、またはAPIエンドポイントへのルートが不安定

# 解決策：httpx Clientにタイムアウトとプロキシ設定を追加

from httpx import Client, Timeout, Proxy

カスタムクライアント設定
custom_client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1",
    http_client=Client(
        timeout=Timeout(60.0, connect=10.0),  # 合計60秒、接続10秒
        proxies={  # 必要に応じてプロキシ設定
            "http://": os.getenv("HTTP_PROXY"),
            "https://": os.getenv("HTTPS_PROXY")
        }
    )
)

リトライロジック付きリクエスト
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def robust_request(messages, tools):
    return custom_client.chat.completions.create(
        model="gemini-2.5-pro",
        messages=messages,
        tools=tools
    )

エラー2: {"401 Unauthorized"}

症状：認証エラーでAPI呼び出しが拒否される

原因：APIキーが無効または期限切れ、または環境変数の読み込み失敗

# 解決策：APIキーの検証と.env設定の確認

import os
from pathlib import Path

def validate_api_key():
    """APIキーの妥当性チェック"""
    
    # 1. 環境変数または.envファイルからキー取得
    api_key = os.getenv("HOLYSHEEP_API_KEY")
    
    if not api_key:
        # .envファイルが見つからない場合
        env_path = Path(__file__).parent / ".env"
        if env_path.exists():
            from dotenv import load_dotenv
            load_dotenv(env_path)
            api_key = os.getenv("HOLYSHEEP_API_KEY")
    
    if not api_key:
        raise ValueError("""
        ❌ HOLYSHEEP_API_KEYが設定されていません！
        
        設定手順：
        1. https://www.holysheep.ai/register でアカウント作成
        2. ダッシュボードからAPIキーを取得
        3. .envファイルに HOLYSHEEP_API_KEY=your_key を追加
        """)
    
    # 2. キー形式チェック（sk-で始まる必要がある）
    if not api_key.startswith("sk-"):
        raise ValueError(f"❌ APIキー形式が無効です: {api_key[:10]}...")
    
    # 3. 接続テスト
    client = OpenAI(api_key=api_key, base_url="https://api.holysheep.ai/v1")
    try:
        client.chat.completions.create(
            model="gemini-2.5-pro",
            messages=[{"role": "user", "content": "test"}],
            max_tokens=5
        )
        print("✅ APIキー認証成功")
    except Exception as e:
        raise ValueError(f"❌ 認証失敗: {e}")
    
    return api_key

エラー3: {"InvalidRequestError: tools parameter format error"}

症状：Function Calling使用時にツール定義の形式エラー

原因：toolsパラメータの構造がGeminiの要件を満たしていない

# 解決策：ツール定義の正しい形式を定義

def create_valid_tools():
    """Gemini 2.5 Pro互換のツール定義"""
    
    # ❌ 間違い：typeフィールドがない
    bad_tools = [{"function": {"name": "test", ...}}]
    
    # ✅ 正しい形式
    good_tools = [
        {
            "type": "function",  # ← 必ず必要
            "function": {
                "name": "get_weather",
                "description": "指定した都市の天気を取得します。",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {
                            "type": "string",
                            "description": "都市名（例：東京）"
                        }
                    },
                    "required": ["location"]
                }
            }
        }
    ]
    
    # ツール定義のバリデーション
    def validate_tool(tool):
        assert "type" in tool, "toolにはtypeフィールドが必要です"
        assert tool["type"] == "function", "typeは'function'である必要があります"
        assert "function" in tool, "function定義がありません"
        fn = tool["function"]
        assert "name" in fn, "関数名が必要です"
        assert "parameters" in fn, "parameters定義が必要です"
        assert fn["parameters"]["type"] == "object", "parametersのtypeは'object'"
        return True
    
    for tool in good_tools:
        validate_tool(tool)
    
    print("✅ ツール定義バリデーション通過")
    return good_tools

エラー4: {"RateLimitError: rate limit exceeded"}

症状：リクエスト頻度制限に到達し、しばらく利用不可

原因：短時間での大量リクエスト

# 解決策：レート制限への対応とリクエスト間隔制御

import time
import asyncio
from collections import deque

class RateLimiter:
    """トークンベースのレ이트リミッター"""
    
    def __init__(self, max_requests: int = 60, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests = deque()
    
    def wait_if_needed(self):
        """必要に応じて待機"""
関連リソース
📚 AI API 記事一覧
💰 料金を見る
📖 開発者ドキュメント
🚀 無料登録
関連記事
機械学習で株価予測：LSTM/TransformerモデルのAPI構築とデプロイ
AI投資ポートフォリオ最適化：多目的遺伝的アルゴリズムとLLMの融合
量化回测框架：Backtrader集成AI信号回测教程

Function Callingとは？なぜ重要か

環境構築と前提条件

必要なライブラリ

バージョン確認

プロジェクト構造

実践的なFunction Calling実装

Step 1: 基本設定とAPI接続

.envファイルからAPIキー読み込み

HolySheep AI設定

接続確認

Step 2: 関数の定義（Tools宣言）

関数実装

Step 3: 完全自动化ワークフロー

実行例

実践事例：マルチツール連携ワークフロー

全ツール定義

実行

エラー處理とデバッグ技巧

レイテンシ測定ユーティリティ

測定実行

よくあるエラーと対処法

エラー1: {"ConnectionError: timeout"}

カスタムクライアント設定

リトライロジック付きリクエスト

エラー2: {"401 Unauthorized"}

エラー3: {"InvalidRequestError: tools parameter format error"}

エラー4: {"RateLimitError: rate limit exceeded"}

関連リソース

関連記事

🔥 HolySheep AIを使ってみる