前回は天気情報を取得する際は固定値で返すようにしていたため、外部のAPIを使用して天気情報を得るようにする。
前回記事
はじめに
本記事では、OllamaのFunction Calling機能とOpen-Meteo APIを組み合わせた、実践的な天気予報システムの実装方法を解説します。無料で利用できるOpen-Meteo APIを活用し、リアルタイムな天気情報を取得する。
システム概要
アーキテクチャ
以下のシーケンス図は、システム全体の処理フローを示す。
実装の詳細解説
天気情報取得機能（get_weather関数）
get_weather関数は、Open-Meteo APIを使用して天気情報を取得します。
実装のポイント：
- 緯度経度による位置指定
- 天気コードの日本語への変換
- エラーハンドリングの実装
def get_weather(location: str) -> str:
"""
Open-Meteo APIを用いて、指定された場所の天気と温度を返す関数です。
"""
try:
# Open-Meteo APIは都市名での直接検索をサポートしていないため、
# 緯度・経度を取得する別のAPI (例: Nominatim) を使用するか、
# または固定の緯度・経度を使用する必要があります。
# ここでは簡略化のため、東京の緯度・経度を固定で使用します。
latitude = 35.6895
longitude = 139.6917
params = {
"latitude": latitude,
"longitude": longitude,
"current": ["temperature_2m", "weather_code"],
"hourly": "temperature_2m",
"timezone": "auto",
}
response = requests.get(OPEN_METEO_URL, params=params)
response.raise_for_status() # HTTPエラーをチェック
data = response.json()
temperature = data["current"]["temperature_2m"]
weather_code = data["current"]["weather_code"]
# 天気コードから天気の説明を取得する関数
weather_description = get_weather_description(weather_code)
return f"{location} の天気は {weather_description} で、温度は {temperature} 度です。"
except requests.exceptions.RequestException 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, "不明")3.2 システムプロンプトの設計（create_payload関数）
Function Callingを効果的に活用するためのプロンプトを設計します。
重要なポイント：
- 明確な役割の定義
- JSON形式の指定
- エラー時の対応指示
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 payloadAPIとの連携方法
Open-Meteo APIの概要と特徴
天気取得のためにOpen‐MetroのAPIを使用します。特徴は以下の通り。
- 無料で利用可能
- API制限が緩やか
- 豊富な気象データ
- APIキー不要
Open-Meteo APIの基本仕様
エンドポイント
https://api.open-meteo.com/v1/forecastパラメータ
| パラメータ | 説明 | 例 |
| latitude | 緯度 | 35.6895 (東京) |
| longitude | 経度 | 139.6917 (東京) |
| current | 現在の気象データ | [“temperature_2m”, “weather_code”] |
| hourly | 1時間ごとの予報 | [“temperature_2m”] |
| daily | 日単位の予報 | [“temperature_2m_max”, “temperature_2m_min”] |
| timezone | タイムゾーン | "auto" または "Asia/Tokyo" |
レスポンス例
{
"latitude": 35.6895,
"longitude": 139.6917,
"timezone": "Asia/Tokyo",
"current": {
"time": "2024-03-20T12:00",
"temperature_2m": 22.5,
"weather_code": 1
}
}天気コードの解析と変換
Open-Meteo APIは国際的な気象コードを使用しています。日本語での表示のため、以下のように変換を行います。
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: "大ひょうを伴う雷雨",
}APIの利用上の注意点
- 位置情報の指定
- 都市名での直接検索は非対応
- 緯度経度による指定が必要
- 必要に応じてジオコーディングサービスとの併用を推奨
- データの更新頻度
- 現在の気象データ：10分ごとに更新
- 予報データ：6時間ごとに更新
- エラーハンドリング
- 無効な座標の場合：400エラー
- サーバー負荷時：429エラー（レート制限）
- 推奨：エラーコードに応じた適切な処理の実装
- レート制限
- 1日あたり10,000リクエストまで無料
- IP単位でのカウント
- 商用利用の場合は別途ライセンスが必要
- パフォーマンス最適化
params = {
"latitude": latitude,
"longitude": longitude,
"current": ["temperature_2m", "weather_code"], # 必要な項目のみ指定
"timezone": "auto",
}エラーハンドリング
例外処理
- APIリクエストエラー
- JSONパースエラー
- 関数引数のバリデーション
- 未定義関数の呼び出し検出
エラー発生時のフォールバック
try:
response = requests.get(OPEN_METEO_URL, params=params)
response.raise_for_status()
except requests.exceptions.RequestException as e:
return f"エラーが発生しました: {e}"
実行結果と動作確認
LLM からの function_call: get_weather({'location': 'Tokyo'})
実行結果: Tokyo の天気は 晴れ で、温度は 11.6 度です。まとめ
OllamaのFunction CallingとOpen-Meteo APIを組み合わせた実践的な実装方法を紹介しました。この実装を基に他の外部APIとの連携を図りたいと思います。次は緯度経度を取得したいと思います。
コード全文
import json
import re
import requests
import os
# 定数設定
OLLAMA_URL = "http://localhost:11434/v1/chat/completions"
MODEL_NAME = "mistral-small"
OPEN_METEO_URL = "https://api.open-meteo.com/v1/forecast"
def get_weather(location: str) -> str:
"""
Open-Meteo APIを用いて、指定された場所の天気と温度を返す関数です。
"""
try:
# Open-Meteo APIは都市名での直接検索をサポートしていないため、
# 緯度・経度を取得する別のAPI (例: Nominatim) を使用するか、
# または固定の緯度・経度を使用する必要があります。
# ここでは簡略化のため、東京の緯度・経度を固定で使用します。
latitude = 35.6895
longitude = 139.6917
params = {
"latitude": latitude,
"longitude": longitude,
"current": ["temperature_2m", "weather_code"],
"hourly": "temperature_2m",
"timezone": "auto",
}
response = requests.get(OPEN_METEO_URL, params=params)
response.raise_for_status() # HTTPエラーをチェック
data = response.json()
temperature = data["current"]["temperature_2m"]
weather_code = data["current"]["weather_code"]
# 天気コードから天気の説明を取得する関数
weather_description = get_weather_description(weather_code)
return f"{location} の天気は {weather_description} で、温度は {temperature} 度です。"
except requests.exceptions.RequestException 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()
try:
result = send_request(payload)
handle_response(result)
except Exception as e:
print("リクエスト中にエラーが発生しました:", e)
if __name__ == "__main__":
main()
コメント