Gemma3のFunction Callingを使用してみた

AI
ローカルLLMのOllamaとPythonで作るFunction Calling機能の実装方法3
OllamaのFunction Calling機能を活用し、PythonでOpen-Meteo APIとNominatimを連携した天気予報システムを構築する最新ガイド。天気情報、API連携、エラーハンドリングを分かりやすく解説します。
yanoasa-engineer.com
2025.03.04

前回はMistrial-smallを使用して地名から経度緯度を取得して、現在の天気を取得できるようにしていた。今回はGoogleが先日発表してたGemma3を使用して同じことを実施してみる。

Gemma3とは

  • 128K の大きなコンテキスト ウィンドウ
  • 140 を超える言語での多言語サポート
  • 画像入力にも対応

これがローカル環境で動かせるのすごい・・・
1年前はローカルで賢くはないけどテキストチャットができる！ぐらいだった認識なので、最近の生成AIの発展具合は凄まじい

Gemma3のダウンロード

OllamaのDockerコンテナが立ち上がっている状態で次のコマンドを実行し20分程度待機

docker exec -it ollama ollama run gemma3:12b

Ollamaのコンテナを立ち上げる手順はこちら

ローカルLLMのOllamaとPythonで作るFunction Calling機能の実装方法
ローカル環境でのAI機能実装！OllamaとPythonを使用したFunction Calling機能の完全解説。初心者でもわかるステップバイステップガイド
yanoasa-engineer.com
2025.02.14

実装

シーケンス図

sequenceDiagram
    participant User
    participant Main
    participant Gemma3 as Ollama (Gemma3)
    participant Mistrial-small as Ollama (Mistrial-small)
    participant OpenMeteo as Open-Meteo API
    participant Nominatim as Nominatim API

    User->>Main: 実行開始
    Main->>Main: create_payload()でリクエスト作成
    Main->>Gemma3: send_request()でリクエスト送信
    Gemma3 -->>Main: JSON形式で応答
    Main->>Main: handle_response()で応答処理
    Main->>OpenMeteo: get_coordinates_open_meteo()で位置情報取得
    OpenMeteo-->>Main: 成功時は緯度・経度を返却
    Main->>Main: 天気情報取得処理へ進む
    alt 失敗時
    　　OpenMeteo-->>Main: 失敗時はエラーを返却
    　　Main->>Mistrial-small: guess_prefecture_with_llm()で県名推定
    　　Mistrial-small-->>Main: 推定された県名を返却
　　    Main->>OpenMeteo: 推定された県名で再試行
        OpenMeteo-->>Main: 成功時は緯度・経度を返却
        OpenMeteo-->>Main: 再試行も失敗
        Main->>Nominatim: get_coordinates()でフォールバック
        Nominatim-->>Main: 緯度・経度を返却
    end
    Main->>Main: 天気情報取得処理へ進む
    Main->>OpenMeteo: 天気情報をリクエスト
    OpenMeteo-->>Main: 天気データを返却
    Main-->>User: 結果を表示

今回は地名の取得を行うモデルと、緯度、経度の取得に失敗したときに地名が何県になるかを判断するモデルで2つ使用した。

おそらく地名から県名を推定する場合はLLMよりは土地名のDBを使用したほうがいいとは思うが、せっかくなのでLLMで使用してみる。

地名推定

def guess_prefecture_with_llm(location: str) -> str:
    """
    LLMを使用して地名から県名を推定します。
    """
    messages = [
        {
            "role": "system",
            "content": (
                "あなたは地名から都道府県を推定する専門家です。"
                "入力された地名に対して、最も可能性の高い都道府県名を返してください。"
                "回答は必ず「都」「道」「府」「県」除いた県名で回答してください。。\n"
                "例: 入力「渋谷」→ 出力「東京」\n"
                "例: 入力「難波」→ 出力「大阪」"
            )
        },
        {"role": "user", "content": f"{location}はどの都道府県にありますか？"}
    ]
    
    payload = {
        "model": MISTRAL_MODEL_NAME,
        "messages": messages,
    }
    
    try:
        response = requests.post(OLLAMA_URL, json=payload)
        response.raise_for_status()
        result = response.json()
        # print(result)
        prefecture = result['choices'][0]['message']['content'].strip()
        # 都道府県名のパターンにマッチするか確認
        # if re.search(r'[都道府県]$', prefecture):
        #     print(f"LLMによる県名推定: {location} → {prefecture}")
        #     return prefecture
        # else:
        #     print(f"LLMの回答が不適切な形式でした: {prefecture}")
        #     return guess_prefecture(location)
        print(f"LLMによる県名推定: {location} → {prefecture}")
        return prefecture
    except Exception as e:
        print(f"LLMによる県名推定に失敗: {e}")
        return guess_prefecture(location)

LLMでやるにしても日本語モデルを使用するのが妥当だと思うが、国産モデルはFunction Callingに対応しているモデルがなかった。自分で追加すればいいが、とりあえず動かしてみるために前回まで使用していたモデルを利用する。

結果

港区で実施

LLM からの応答:
{
  "id": "chatcmpl-81",
  "object": "chat.completion",
  "created": 1742479310,
  "model": "gemma3:12b",
  "system_fingerprint": "fp_ollama",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "{\n \"function_call\": {\n   \"name\": \"get_weather\",\n   \"arguments\": \"{\\\"location\\\": \\\"港区\\\"}\"\n }\n}"
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 119,
    "completion_tokens": 37,
    "total_tokens": 156
  }
}

LLM からの function_call: get_weather({'location': '港区'})

=== フェーズ1: 港区の位置情報取得を開始 ===
Open-Meteo で取得失敗: Open-Meteo で場所が見つかりません: 港区
LLMによる県名推定: 港区 → 東京
推定された県名を用いて再試行: 東京
Open-Meteo API: 東京 の位置情報取得成功 -> 緯度: 35.6895, 経度: 139.69171
位置情報取得完了: 緯度 35.6895, 経度 139.69171

=== フェーズ2: 天気情報の取得を開始 ===
Open-Meteo APIにリクエストを送信中...
天気APIからのレスポンス受信完了

=== フェーズ3: 天気情報の解析 ===
取得した気温: 5.7度
取得した天気コード: 0
天気コード 0 を '晴れ' に変換完了

=== 処理完了 ===
実行結果: 港区 の天気は 晴れ で、温度は 5.7 度です。

港区だと取得できなかったので、東京の気温として表示している。

存在しない地名百舌鳥田で実行してみる。

LLM からの応答:
{
  "id": "chatcmpl-287",
  "object": "chat.completion",
  "created": 1742479721,
  "model": "gemma3:12b",
  "system_fingerprint": "fp_ollama",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "{\n \"function_call\": {\n   \"name\": \"get_weather\",\n   \"arguments\": \"{\\\"location\\\": \\\"百舌鳥田\\\"}\"\n }\n}\n"
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 121,
    "completion_tokens": 40,
    "total_tokens": 161
  }
}

LLM からの function_call: get_weather({'location': '百舌鳥田'})

=== フェーズ1: 百舌鳥田の位置情報取得を開始 ===
Open-Meteo で取得失敗: Open-Meteo で場所が見つかりません: 百舌鳥田
LLMによる県名推定: 百舌鳥田 → 摩周
推定された県名を用いて再試行: 摩周
推定による取得も失敗: Open-Meteo で場所が見つかりません: 摩周 (Nominatimでフォールバックします)
Nominatim APIを使用して百舌鳥田の位置情報を検索中...
位置情報取得成功: 緯度 34.5583589, 経度 135.4889775
位置情報取得完了: 緯度 34.5583589, 経度 135.4889775

=== フェーズ2: 天気情報の取得を開始 ===
Open-Meteo APIにリクエストを送信中...
天気APIからのレスポンス受信完了

=== フェーズ3: 天気情報の解析 ===
取得した気温: 4.6度
取得した天気コード: 0
天気コード 0 を '晴れ' に変換完了

=== 処理完了 ===
実行結果: 百舌鳥田 の天気は 晴れ で、温度は 4.6 度です。

摩周として推定していたが、緯度 34.5583589, 経度 135.4889775の場所というと

大阪にある百舌鳥駅を推定しているみたい。漢字は同じなので賢い？結果は得られてはいる？

摩周はどこなんだ・・・

まとめ

Gemma3がリリースされたので早速使いたくてとりあえずやってみたレベルで動かした。

Ollamaに対応していたこともあってコマンド一つで実行できるのは簡単で素晴らしいと思った。生成AIの混沌として開発環境がある程度は落ち着き始めているのかなと感じた。

知らないことでも回答するので、プロンプトで制御するかエラーハンドリングで避けるしかないのかな？

コード

import json
import re
import requests
import os
from geopy.geocoders import Nominatim
from geopy.exc import GeocoderTimedOut, GeocoderUnavailable

# 定数設定
OLLAMA_URL = "http://localhost:11434/v1/chat/completions"
MISTRAL_MODEL_NAME = "mistral-small"
# MODEL_NAME = "gemma3:27b"
MODEL_NAME = "gemma3:12b"
OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast"

def get_coordinates(location: str) -> tuple[float, float]:
    """
    Nominatim APIを使用して場所名から緯度経度を取得します。
    """
    print(f"Nominatim APIを使用して{location}の位置情報を検索中...")
    try:
        geolocator = Nominatim(user_agent="my_weather_app")
        location_data = geolocator.geocode(location)
        if location_data:
            print(f"位置情報取得成功: 緯度 {location_data.latitude}, 経度 {location_data.longitude}")
            return location_data.latitude, location_data.longitude
        else:
            print(f"エラー: {location}の位置情報が見つかりません")
            raise ValueError(f"場所が見つかりません: {location}")
    except (GeocoderTimedOut, GeocoderUnavailable) as e:
        print(f"エラー: 位置情報の取得に失敗しました - {e}")
        raise Exception(f"位置情報の取得に失敗しました: {e}")

def get_coordinates_open_meteo(location: str) -> tuple[float, float]:
    """
    Open-Meteo Geocoding API を使用し、場所名から緯度経度を取得します。
    APIドキュメントに従い、パラメータ 'language' を追加しています。
    """
    endpoint = "https://geocoding-api.open-meteo.com/v1/search"
    params = {"name": location, "count": 1, "language": "ja"}
    try:
        response = requests.get(endpoint, params=params)
        response.raise_for_status()
        data = response.json()
    except Exception as e:
        raise Exception(f"Open-Meteo API へのリクエストに失敗しました: {e}")
    
    results = data.get("results")
    if results and len(results) > 0:
        result = results[0]
        latitude = result.get("latitude")
        longitude = result.get("longitude")
        print(f"Open-Meteo API: {location} の位置情報取得成功 -> 緯度: {latitude}, 経度: {longitude}")
        return latitude, longitude
    else:
        raise ValueError(f"Open-Meteo で場所が見つかりません: {location}")

def get_weather(location: str) -> str:
    """
    Open-Meteo APIを用いて、指定された場所の天気と温度を返す関数です。
    """
    try:
        print(f"\n=== フェーズ1: {location}の位置情報取得を開始 ===")
        try:
            # まず Open-Meteo のジオコーディングAPI を試行
            latitude, longitude = get_coordinates_open_meteo(location)
        except ValueError as e:
            print(f"Open-Meteo で取得失敗: {e}")
            # LLMを使用して県名を推定
            inferred = guess_prefecture_with_llm(location)
            print(f"推定された県名を用いて再試行: {inferred}")
            try:
                latitude, longitude = get_coordinates_open_meteo(inferred)
            except Exception as e2:
                print(f"推定による取得も失敗: {e2} (Nominatimでフォールバックします)")
                latitude, longitude = get_coordinates(location)
        print(f"位置情報取得完了: 緯度 {latitude}, 経度 {longitude}")

        print(f"\n=== フェーズ2: 天気情報の取得を開始 ===")
        params = {
            "latitude": latitude,
            "longitude": longitude,
            "current": ["temperature_2m", "weather_code"],
            "timezone": "auto",
        }
        print("Open-Meteo APIにリクエストを送信中...")
        response = requests.get(OPEN_METEO_URL, params=params)
        response.raise_for_status()  # HTTPエラーをチェック
        data = response.json()
        print("天気APIからのレスポンス受信完了")     
        print(f"\n=== フェーズ3: 天気情報の解析 ===")
        temperature = data["current"]["temperature_2m"]
        weather_code = data["current"]["weather_code"]
        print(f"取得した気温: {temperature}度")
        print(f"取得した天気コード: {weather_code}")

        # 天気コードから天気の説明を取得する関数
        weather_description = get_weather_description(weather_code)
        print(f"天気コード {weather_code} を '{weather_description}' に変換完了")

        result = f"{location} の天気は {weather_description} で、温度は {temperature} 度です。"
        print("\n=== 処理完了 ===")
        return result
    except ValueError as e:
        return str(e)
    except Exception as e:
        return f"エラーが発生しました: {e}"

def get_weather_description(weather_code: int) -> str:
    """
    天気コードに基づいて天気の説明を返します。
    """
    weather_descriptions = {
        0: "晴れ",
        1: "主に晴れ",
        2: "部分的に曇り",
        3: "曇り",
        45: "霧",
        48: "凍結霧",
        51: "小雨",
        53: "適度な雨",
        55: "激しい雨",
        56: "凍雨",
        57: "激しい凍雨",
        61: "弱い雨",
        63: "適度な雨",
        65: "強い雨",
        66: "軽い凍雨",
        67: "重い凍雨",
        71: "わずかな積雪",
        73: "適度な積雪",
        75: "激しい積雪",
        77: "雪の粒子",
        80: "弱い雨",
        81: "適度な雨",
        82: "激しい雨",
        85: "弱い雪",
        86: "激しい雪",
        95: "雷雨",
        96: "ひょうを伴う雷雨",
        99: "大ひょうを伴う雷雨",
    }
    return weather_descriptions.get(weather_code, "不明")

# 追加: 推定用ヘルパー関数
def guess_prefecture(location: str) -> str:
    """
    入力の地名に「県」「府」「都」「道」が含まれていなければ、末尾に「県」を追加して推定する。
    """
    if any(suffix in location for suffix in ["県", "府", "都", "道"]):
        return location
    return location + "県"

def guess_prefecture_with_llm(location: str) -> str:
    """
    LLMを使用して地名から県名を推定します。
    """
    messages = [
        {
            "role": "system",
            "content": (
                "あなたは地名から都道府県を推定する専門家です。"
                "入力された地名に対して、最も可能性の高い都道府県名を返してください。"
                "回答は必ず「都」「道」「府」「県」除いた県名で回答してください。。\n"
                "例: 入力「渋谷」→ 出力「東京」\n"
                "例: 入力「難波」→ 出力「大阪」"
            )
        },
        {"role": "user", "content": f"{location}はどの都道府県にありますか？"}
    ]
    
    payload = {
        "model": MISTRAL_MODEL_NAME,
        "messages": messages,
    }
    
    try:
        response = requests.post(OLLAMA_URL, json=payload)
        response.raise_for_status()
        result = response.json()
        # print(result)
        prefecture = result['choices'][0]['message']['content'].strip()
        # 都道府県名のパターンにマッチするか確認
        # if re.search(r'[都道府県]$', prefecture):
        #     print(f"LLMによる県名推定: {location} → {prefecture}")
        #     return prefecture
        # else:
        #     print(f"LLMの回答が不適切な形式でした: {prefecture}")
        #     return guess_prefecture(location)
        print(f"LLMによる県名推定: {location} → {prefecture}")
        return prefecture
    except Exception as e:
        print(f"LLMによる県名推定に失敗: {e}")
        return guess_prefecture(location)

def create_payload() -> dict:
    """
    システムプロンプトとユーザーメッセージを含むリクエストペイロードを生成します。
    ※システムプロンプトで、出力は必ず JSON 形式の function_call として返すように明示しています。
    """
    messages = [
        {
            "role": "system",
            "content": (
                "あなたは天気情報を提供するエージェントです。"
                "場所名を受け取り、その場所の天気情報を返します。"
                "ユーザーからの質問に対しては、必ず下記の形式に沿った JSON 形式の function_call として回答してください。\n\n"
                "例:\n"
                "{\n"
                '  "function_call": {\n'
                '    "name": "get_weather",\n'
                '    "arguments": "{\\"location\\": \\"東京\\"}"\n'
                "  }\n"
                "}\n\n"
                "絶対に他の形式では回答しないでください。"
            ),
        },
        {"role": "user", "content": "百舌鳥田の天気と温度を教えてください。"},
    ]

    functions = [
        {
            "name": "get_weather",
            "description": "指定された場所の天気と温度を返す関数です。",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "天気情報を取得する場所の名前",
                    }
                },
                "required": ["location"],
            },
        }
    ]

    payload = {
        "model": MODEL_NAME,
        "messages": messages,
        "functions": functions,
        # 明示的に関数呼び出しを指定
        "function_call": {"name": "get_weather"},
    }
    return payload

def send_request(payload: dict) -> dict:
    """
    Ollama サーバーへリクエストを送信し、JSON レスポンスを返します。
    """
    response = requests.post(OLLAMA_URL, json=payload)
    response.raise_for_status()
    return response.json()

def extract_function_call_from_content(content: str) -> dict:
    """
    メッセージの content 内に含まれるコードブロックから JSON を抽出し、
    その中に含まれる function_call 情報を返します。
    """
    # 言語指定 "json" があってもなくてもマッチする正規表現
    pattern = r"```(?:json)?\s*(\{.*?\})\s*```"
    match = re.search(pattern, content, re.DOTALL)
    if match:
        json_str = match.group(1)
        try:
            parsed = json.loads(json_str)
            if "function_call" in parsed:
                return parsed["function_call"]
        except Exception as e:
            print("コードブロック内の JSON パースに失敗:", e)
    else:
        # もしコードブロックがない場合、内容全体が JSON であれば試す
        try:
            parsed = json.loads(content)
            if "function_call" in parsed:
                return parsed["function_call"]
        except Exception:
            pass
    return {}

def handle_response(result: dict):
    """
    LLM のレスポンスから function_call 情報を抽出し、
    対応するローカル関数を実行する処理を行います。
    """
    print("LLM からの応答:")
    print(json.dumps(result, indent=2, ensure_ascii=False))

    choices = result.get("choices", [])
    if not choices:
        print("応答形式が不正です。")
        return

    message = choices[0].get("message", {})
    # まず、直接 function_call フィールドがあるか確認
    function_call = message.get("function_call")
    if not function_call:
        # なければ、content 内のコードブロックから抽出を試みる
        content = message.get("content", "")
        function_call = extract_function_call_from_content(content)

    if function_call:
        function_name = function_call.get("name")
        raw_arguments = function_call.get("arguments", "{}")
        # 引数が余分な引用符で囲まれている場合、その除去を試みる
        if raw_arguments.startswith('"') and raw_arguments.endswith('"'):
            raw_arguments = raw_arguments[1:-1]

        try:
            arguments = json.loads(raw_arguments)
        except Exception as e:
            print("関数引数のパースに失敗:", e)
            arguments = {}

        print(f"\nLLM からの function_call: {function_name}({arguments})")
        if function_name == "get_weather":
            result_value = get_weather(**arguments)
            print("実行結果:", result_value)
        else:
            print(f"定義されていない関数が呼ばれました: {function_name}")
    else:
        print("LLM の応答内に function_call が見つかりませんでした。")
        print("返答内容:", message.get("content"))

def main():
    payload = create_payload()
    for attempt in range(1, 6):
        try:
            result = send_request(payload)
            handle_response(result)
            break  # 成功時はループを抜ける
        except Exception as e:
            print(f"リクエスト中にエラーが発生しました (試行{attempt}):", e)
            if attempt == 5:
                print("5回試行しましたが、全て失敗しました。")

if __name__ == "__main__":
    main()

タイトルとURLをコピーしました