Local LLMで実装するFunction Callingで天気情報の取得方法

前回で外部の天気情報を取得できるAPIから天気情報を取得できたので、今度はAPIの入力に必要であった緯度経度をAPIで取得して天気情報を取得できるようにする。

概要
システム概要
Open-Meteo API仕様
1. 実装コード
実行結果
まとめ
コード全文

概要

OllamaのFunction Calling機能を活用し、ユーザーが指定した場所名から緯度・経度を取得、Open-Meteo APIで天気情報を取得するシステムの実装方法を解説します。これにより、実際に天気予報システムを構築する。

システム概要

フェーズ1 (ユーザー入力と関数呼び出し)
ユーザーが場所名を入力すると、OllamaがFunction Calling形式で適切な関数呼び出し情報を返します。

フェーズ2 (位置情報の取得)
指定された場所名を元に、Nominatim API（またはOpen-Meteo Geocoding）で緯度・経度を取得します。

フェーズ3 (天気情報の取得)
取得した座標を使用して、Open-Meteo APIから天気と温度データを取得します。天気コードは、日本語の天気表記に変換されます。

フェーズ4 (結果表示)
最終的な天気情報がユーザーに表示されます。

今回のシーケンス図

sequenceDiagram
    participant User
    participant Main
    participant LLM as Ollama (LLM)
    participant Weather as get_weather関数
    participant Geocoding as Open-Meteo Geocoding API
    participant Weather_API as Open-Meteo API

    rect rgb(191, 223, 255)
        Note over User,Main: フェーズ1: ユーザー入力と関数呼び出し情報生成
        User->>Main: 場所名の入力
        Main->>Main: create_payload()でペイロード生成
        Main->>LLM: send_request()でリクエスト送信
        LLM-->>Main: JSON形式でFunction Calling情報応答
        Main->>Main: handle_response()で呼び出し情報抽出
    end

    rect rgb(255, 224, 191)
        Note over Weather,Geocoding: フェーズ2: 位置情報の取得
        Main->>Weather: get_weather("場所名")実行
        Weather->>Geocoding: 場所名検索
        Geocoding-->>Weather: 緯度・経度を返却
    end

    rect rgb(191, 255, 191)
        Note over Weather,Weather_API: フェーズ3: 天気情報の取得と解析
        Weather->>Weather_API: 緯度経度で天気リクエスト
        Weather_API-->>Weather: 天気データ（温度・天気コード）返却
        Weather->>Weather: 天気コードを日本語に変換
    end

    rect rgb(255, 191, 191)
        Note over Weather,User: フェーズ4: 結果の返却と表示
        Weather-->>Main: 整形済み天気情報返却
        Main-->>User: 結果を表示
    end

シーケンス図を見るとほとんどLLMを使用していない気が。。。

Open-Meteo API仕様

エンドポイント
`https://api.open-meteo.com/v1/forecast`

主要パラメータ
latitude: 緯度
longitude: 経度
current: 現在のデータ項目(例: `[“temperature_2m”, “weather_code”]`)
timezone: “auto”または任意のタイムゾーン指定

レスポンス例

  {
      "latitude": 35.6895,
      "longitude": 139.6917,
      "timezone": "Asia/Tokyo",
      "current": {
          "time": "2024-03-20T12:00",
          "temperature_2m": 22.5,
          "weather_code": 1
      }
  }

都市名での直接検索は非対応なので、緯度経度による指定が必須

更新頻度やレート制限（1日あたり10,000リクエストなど）に注意する必要はあるが、今回は勉強用なので無視

実装コード

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}")

実行結果

LLM からの応答:
{
  "id": "chatcmpl-27",
  "object": "chat.completion",
  "created": 1741006764,
  "model": "mistral-small",
  "system_fingerprint": "fp_ollama",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "```json\n{\n  \"function_call\": {\n    \"name\": \"get_weather\",\n    \"arguments\": \"\\\"沖縄\\\"\"\n  }\n}\n```"
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 129,
    "completion_tokens": 35,
    "total_tokens": 164
  }
}
関数引数のパースに失敗: Expecting value: line 1 column 1 (char 0)

LLM からの function_call: get_weather({})
リクエスト中にエラーが発生しました (試行1): get_weather() missing 1 required positional argument: 'location'
LLM からの応答:
{
  "id": "chatcmpl-417",
  "object": "chat.completion",
  "created": 1741006780,
  "model": "mistral-small",
  "system_fingerprint": "fp_ollama",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "```json\n{\n  \"function_call\": {\n    \"name\": \"get_weather\",\n    \"arguments\": \"{\\\"location\\\": \\\"沖縄\\\"}\"\n  }\n}\n```"
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 129,
    "completion_tokens": 39,
    "total_tokens": 168
  }
}

LLM からの function_call: get_weather({'location': '沖縄'})

=== フェーズ1: 沖縄の位置情報取得を開始 ===
Open-Meteo API: 沖縄 の位置情報取得成功 -> 緯度: 26.33583, 経度: 127.80139
位置情報取得完了: 緯度 26.33583, 経度 127.80139

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

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

=== 処理完了 ===
実行結果: 沖縄 の天気は 曇り で、温度は 21.1 度です。

まとめ

固定の応答から発展して外部APIとLLMを利用して地名から天気情報が取得できることができた。

Function Callingの実装について言えば、LLMにユーザの要求と対応可能な道具一覧を同時に提供し、最適な道具を選択させるという単純な仕組みに過ぎない。そのため、実務での応用を見据えると、各道具の要件定義が非常に高度で複雑な課題となることは避けられない。

さらに、生成AIは流行の最先端技術として過大な期待が寄せられているが、実際には万能の魔法の杖ではなく、慎重な設計と運用が必須である。

重ねて申し上げるが、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"
MODEL_NAME = "mistral-small"
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} (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 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()

Parse error on line 1:

^
Expecting 'NEWLINE', 'SPACE', 'GRAPH', got 'EOF'