【実装】LangGraphを使用した天気エージェント機能

概要

• LangGraphとモデルを使用して天気エージェントを作成しました。
• 最終にモデルからかえってきた回答をjson形式にして出力実装しています。

前置き

• 今回はgoogle AI studioからAPIkeyを取得して実装しています。（無料枠）

0.ライブラリーの読み込み

from typing import Annotated, Sequence, TypedDict, List
from langchain_core.messages import BaseMessage, AIMessage
from langgraph.graph.message import add_messages
from langchain_core.tools import tool
from geopy.geocoders import Nominatim
from pydantic import BaseModel, Field
from datetime import datetime
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, END
from langchain_core.messages import ToolMessage, AIMessage
import requests
import os
import json
import pandas as pd
from dotenv import load_dotenv

1.初期設定

• google AI studioからAPIkeyを「.env」ファイルに入力します。
• その値（APIKey）を読み込みます

##########################################################
# 1. 初期設定
##########################################################
load_dotenv()
api_key=os.environ['GOOGLE_AI_ST_API']

2.agentの状態管理を設定

• AgentState は「グラフ内で保持される状態（メッセージ履歴＋ステップ数）」を定義します。
  • messages → AI・ユーザー・Tool のすべてのメッセージ履歴
  • number_of_steps → 何ステップ実行したかのカウンター

##########################################################
# 2. agent状態管理（LangGraph）
##########################################################
class AgentState(TypedDict):
  messages: Annotated[Sequence[BaseMessage], add_messages]
  number_of_steps: int

3．レスポンスの型定義

• Pydanticを使用してレスポンスの型を定義します

##########################################################
# 3. Pydantic 出力モデル
##########################################################
class WeatherItem(BaseModel):
  date: str = Field(description="日時（YYYY-MM-DDTHH:MM）")
  temperature: str = Field(description="気温（℃）")
  weather: str = Field(description="天気")
  advice: str = Field(description="天候に応じたアドバイス")
class WeatherAnswer(BaseModel):
  answer: List[WeatherItem]

4.天気コードマップの定義

• Open-Meteoはオープンソースの天気APIで天気の情報を取得します。
• 天気コードを事前に設定します。

##########################################################
# 4. 天気コードマップ
##########################################################
WEATHER_CODE_MAP = {
  0: "快晴", 1: "晴れ", 2: "薄曇り", 3: "曇り",
  45: "霧", 48: "霧氷",
  51: "霧雨", 53: "霧雨", 55: "霧雨",
  61: "小雨", 63: "雨", 65: "強い雨",
  71: "小雪", 73: "雪", 75: "強い雪",
  80: "にわか雨", 81: "雨", 82: "激しいにわか雨",
  85: "にわか雪", 86: "大雪",
}

5.ツールの作成（天気情報取得ツール）

##########################################################
# 5. 天気情報取得ツール
##########################################################
## geopy の Nominatim（OpenStreetMap の地名検索 API）を使う為の初期化
geolocator = Nominatim(user_agent="weather-app")
## Tool に渡す 引数の型 を Pydantic で定義。
class SearchInput(BaseModel):
  location: str = Field(description="都道府県（例：東京）")
  date: str = Field(description="日付（yyyy-mm-dd）")
@tool("get_weather_forecast", args_schema=SearchInput, return_direct=True)
def get_weather_forecast(location: str, date: str):
    """Open-Meteo から天気予報を取得するツール"""
    ## 地名から緯度経度に変換
    loc = geolocator.geocode(location)
    if loc is None:
        return {"error": "Location not found"}
    ## Open-Meteo APIを呼び出す
    response = requests.get(
        "https://api.open-meteo.com/v1/forecast",
        params={
            "latitude": loc.latitude,
            "longitude": loc.longitude,
            "hourly": "temperature_2m,weathercode",
            "start_date": date,
            "end_date": date,
        }
    )
    ## Open-Meteo APIからのレスポンス処理
    data = response.json()
    times = data["hourly"]["time"]
    temps = data["hourly"]["temperature_2m"]
    codes = data["hourly"]["weathercode"]
    result = {}
    ## 天気コードより日本語に変換し整形
    for t, temp, code in zip(times, temps, codes):
        result[t] = {
            "temperature": temp,
            "weather": WEATHER_CODE_MAP.get(code, f"不明({code})"),
        }
    return result
## Graph 内でツールを使うための登録
## toolsリストとtool本体の辞書型を作成
tools = [get_weather_forecast]
tools_by_name = {tool.name: tool for tool in tools}

6.Geminiモデルの定義とツールのテスト

• Geminiモデルの設定を行います。
• 作成したToolsをモデルと関連付けます
• 実際にユーザーの問いに対して適切にToolに引数を渡せるかテスト実装を行います。

##########################################################
# 6. Gemini モデルの定義
##########################################################
llm = ChatGoogleGenerativeAI(
    model="gemini-2.5-flash",
    google_api_key=api_key,
    temperature=0.8,
)
# ツールをモデルに関連付ける
model = llm.bind_tools(tools)
# 構造化出力用のモデル（ツール呼び出し後用）
structured_model = llm.with_structured_output(WeatherAnswer)
# モデルがツールに紐づいているかテスト
today_str = datetime.now().strftime("%Y-%m-%d %H") + ":00"
res=model.invoke(f"{today_str}の沖縄県の天気はどうですか?")
fc = res.additional_kwargs["function_call"]
print("Function name:", fc["name"])
# print("Arguments:", fc["arguments"])
# arguments を JSON としてパース
args = json.loads(fc["arguments"])
print("Arguments:", args)

• テスト実装の結果

Function name: get_weather_forecast
Arguments: {'location': '沖縄県', 'date': '2025-11-23'}

※ 本来はエージェントは下記の引数を渡しています。
※ 上記の実装結果は確認為に変換処理を加えています

Function name: get_weather_forecast
Arguments: {"location": "\u6c96\u7e04\u770c", "date": "2025-11-23"}

7.ツール結果をJSON形式に変換する関数を作成

• 天気ツール（get_weather_forecast）の返り値を、最終的に返すべきPydanticモデル（WeatherAnswer）に変換するための関数です。

##########################################################
# 7. ツール結果 → JSON の最終形式に変換
##########################################################
def convert_to_weather_answer(result: dict) -> WeatherAnswer:
  items = []
  for t, data in result.items():
    items.append(
      WeatherItem(
        date=t,
        temperature=f"{data['temperature']}℃",
        weather=data["weather"],
        advice="取得した天気のデータに応じてアドバイスをしてください",
      )
    )
  return WeatherAnswer(answer=items)

8.LangGraphノードの定義

以下の関数を定義します。

• call_tool：LLM からのtool_callを受け取ってツール実行する関数
• call_model：LLMを読んだりツール結果をJSON化にして出力する関数

##########################################################
# 8. Graph ノード定義
##########################################################
def call_tool(state: AgentState):
  ## LLM の最後のメッセージを確認
  last = state["messages"][-1]
  outputs = []
  ## LLMが要求しているtool_callを順に実行
  for tc in last.tool_calls:
    tool_result = tools_by_name[tc["name"]].invoke(tc["args"])
    ## LLM に返す「ツールの結果メッセージ」を生成
    outputs.append(
      ToolMessage(
        content=tool_result,
        name=tc["name"],
        tool_call_id=tc["id"]
      )
    )
  ## 次の状態としてmessagesを返す
  return {
      "messages": outputs,
      "number_of_steps": state["number_of_steps"] + 1
  }
def call_model(state: AgentState, config: RunnableConfig):
  # 状況に応じて処理を分岐処理を行います
  messages = state["messages"]
  # ▼ツール呼び出し（function_call）を含む → Gemini に渡す
  if any(hasattr(m, "tool_calls") and m.tool_calls for m in messages):
    response = model.invoke(messages, config)
    return {
      "messages": [response],
      "number_of_steps": state["number_of_steps"] + 1
    }
  # ▼ToolMessage（ツールの結果）→ ここが最終出力
  tool_messages = [m for m in messages if isinstance(m, ToolMessage)]
  if tool_messages:
    tool_content = tool_messages[-1].content
    answer = convert_to_weather_answer(tool_content)
    return {
      "messages": [AIMessage(content=json.dumps(answer.model_dump(), ensure_ascii=False))],
      "number_of_steps": state["number_of_steps"] + 1
    }
  # ▼初回（通常 LLM）
  response = model.invoke(messages, config)
  return {
    "messages": [response],
    "number_of_steps": state["number_of_steps"] + 1
  }

9.LangGraph の「分岐条件」(ルーティング条件) を定義

• グラフが次にどのノード（call_tool or 終了）へ進むかを決める“信号機（ルーター）” のような役割を定義します。

##########################################################
# 9. グラフ遷移条件
##########################################################
def should_continue(state: AgentState):
  last = state["messages"][-1]
  if hasattr(last, "tool_calls") and last.tool_calls:
    return "continue"
  return "end"

10LangGraphを構築

• 上記の定義した内容をLangGraphに構築処理を行います。

##########################################################
# 10. LangGraph 構築
##########################################################
## StateGraphのインスタンスを作成
workflow = StateGraph(AgentState)
## GrphにNodeを追加
workflow.add_node("llm", call_model)
workflow.add_node("tools", call_tool)
# 始点を定義
workflow.set_entry_point("llm")
# ループ処理を定義
workflow.add_conditional_edges(
  "llm",
  should_continue,
  {"continue": "tools", "end": END},
)
# Edgeを追加
workflow.add_edge("tools", "llm")
## NodeとEdgeで構成したGraphをコンパイル
graph = workflow.compile()
## Graphを可視化(mermaid)
img_bytes = graph.get_graph().draw_mermaid_png()
with open("./weather_tools.png", "wb") as f:
  f.write(img_bytes)

• Graphを可視化

11.LangGraph実行と結果を取得

##########################################################
# 11. 実行
##########################################################
today = datetime.now().strftime("%Y-%m-%d")
user_msg = ("user", f"{today} の 青森県 の天気は？")
result = graph.invoke({
  "messages": [user_msg],
  "number_of_steps": 0
})
# result は LangGraph の結果
messages = result["messages"]
ai_message = result["messages"][-1]

12.実行結果からjson形式に成形

• 残念ながらLangGraph実行結果からそのまま想定通りのJSON形式になりませんでした。
• そのため、想定しているJSON形式に整形する処理を行います。

【ToolMessage（天気データ）を抽出】

###########################################################
# 12． ToolMessage（天気データ）を抽出
###########################################################
tool_msg = next(m for m in messages if isinstance(m, ToolMessage))
# content が文字列（シングルクォート）なので eval で dict に変換
weather_raw = tool_msg.content
if isinstance(weather_raw, str):
  weather_raw = eval(weather_raw)

• whether_rowの出力例

// weather_raw の例：
{'2025-11-23T00:00': {'temperature': 20.8, 'weather': '快晴'}, ... }

【pandasのDataFrameを使ってデータを整形】

###########################################################
# 13． pandas DataFrame に変換
###########################################################
df = pd.DataFrame([
  {
    "datetime": dt,
    "temperature": data["temperature"],
    "weather": data["weather"]
  }
  for dt, data in weather_raw.items()
])

###########################################################
# 14. 最頻値（天気）・気温の平均・日付を取得
###########################################################
most_common_weather = df["weather"].mode()[0]
avg_temp = round(df["temperature"].mean(), 1)
date_str = df["datetime"].iloc[0].split("T")[0]

【最後のAIMessageのアドバイスを抽出】

###########################################################
# 15．AIMessage のアドバイスを抽出
###########################################################
def extract_text_from_message(msg):
  content = msg.content
  # string の場合はそのまま
  if isinstance(content, str):
    return content
  # dict の場合
  if isinstance(content, dict):
    return content.get("text", json.dumps(content, ensure_ascii=False))
  # list の場合
  if isinstance(content, list):
    texts = []
    for c in content:
      if isinstance(c, dict):
        if "text" in c:
          texts.append(c["text"])
    return "\n".join(texts)
  return str(content)
advice_text = extract_text_from_message(ai_message)

【最終形態のJSON形式にして出力】

###########################################################
# 16 最終 JSON を整形
###########################################################
final_json = {
    "answer": {
        "date": date_str,
        "weather": most_common_weather,
        "temperature": f"{avg_temp}℃",
        "advice": advice_text
    }
}
with open("./res_wether.json", "w", encoding="utf-8") as f:
    json.dump(final_json, f, ensure_ascii=False, indent=2)

• jsonの出力例

{
  "answer": {
    "date": "2025-11-23",
    "weather": "晴れ",
    "temperature": "9.0℃",
    "advice": "青森県の2025-11-23の天気は以下の通りです。\n\n* 00:00: 気温8.9℃、晴れ\n* 01:00: 気温10.2℃、晴れ\n* 02:00: 気温11.3℃、薄曇り\n* 03:00: 気温10.1℃、霧雨\n* 04:00: 気温10.9℃、小雨\n* 05:00: 気温11.6℃、霧雨\n* 06:00: 気温11.4℃、霧雨\n* 07:00: 気温10.4℃、薄曇り\n* 08:00: 気温9.1℃、晴れ\n* 09:00: 気温9.1℃、晴れ\n* 10:00: 気温9.1℃、晴れ\n* 11:00: 気温9.1℃、晴れ\n* 12:00: 気温9.4℃、晴れ\n* 13:00: 気温9.4℃、晴れ\n* 14:00: 気温8.2℃、晴れ\n* 15:00: 気温7.8℃、快晴\n* 16:00: 気温7.5℃、晴れ\n* 17:00: 気温7.4℃、晴れ\n* 18:00: 気温7.3℃、晴れ\n* 19:00: 気温7.3℃、晴れ\n* 20:00: 気温7.2℃、晴れ\n* 21:00: 気温7.2℃、薄曇り\n* 22:00: 気温7.4℃、薄曇り\n* 23:00: 気温8.1℃、薄曇り"
  }
}

全体のコード(GitHub)

後記

• 最後まで読んで頂きありがとうございます。
• LangGraphを使用して天気エージェントを実装しました。
• 基礎のサンプルコードを基にAIとバイブコーディングしながら実装を進めました。
• これを基礎に様々なAPI情報取得からモデルによるコメントや要約などの出力までのAIエージェントを作成していきたいと思います。