天氣預報與 GenCast 小型演示

天氣是一個複雜的系統，任何時刻的小變化都可能導致隨著時間的推移出現重大且有時難以預測的變化。但是，破解這個混亂的系統並不容易。幾個世紀以來，我們一直在嘗試預測天氣，例如聽蟋蟀的叫聲或仰望星空尋找答案。這樣的方法實用嗎？別擔心。如果我告訴你，科技可以提前10到15天預測何時需要帶傘或準備颶風，你會怎麼想？聽起來很棒吧？Google DeepMind的GenCast正在做到這一切。

科技讓我們在自然天氣模式的不確定性中佔據了優勢。隨著人工智慧的興起，我們已經遠遠超越了傳統的觀察動物行為、民間傳說或天體位置的方法。Google DeepMind推出的GenCast是一個人工智慧天氣模型，正在為天氣預測和風險評估設定新的標準。今天在《自然》雜誌上發表的這個先進模型旨在提供準確且詳細的天氣預測，包括對極端天氣條件的預測，最長可達15天。GenCast的概率方法和人工智慧驅動的架構標誌著天氣預測的一次重大飛躍，滿足了從日常生活規劃到災害管理和可再生能源生產等重要社會需求。

進階天氣預測的需求

天氣影響著人類生活的幾乎每一個方面。從家庭的日常決策到農業實踐，再到可再生能源的生產，理解和預測天氣模式至關重要。早期的天氣預測依賴於複雜的物理模型，這些模型需要巨大的計算能力來運行模擬。這些模型通常需要在超級計算機上運行幾個小時才能產生預測。此外，傳統的預測通常只提供一個確定的未來天氣條件估計，雖然有用，但往往不夠準確，無法應對不確定性或極端天氣事件。因此，進階的天氣預測變得至關重要。

Google DeepMind的GenCast：天氣預測的人工智慧革命

Google的GenCast採用概率集成預測方法來解決這些限制。與傳統模型提供單一預測不同，GenCast生成多個潛在情境——在某些情況下超過50種——以提供一系列可能的結果，並附上每種情境的概率。這種方法不僅提供了更準確的預測，還為決策者提供了潛在天氣結果的更全面畫面，包括相關的不確定性程度。

GenCast如何運作？

GenCast是一種擴散模型，這是一種機器學習模型，最近也推動了生成式人工智慧的進步，例如圖像、視頻和音樂生成。然而，與這些應用不同，GenCast專門調整以考慮地球的球面幾何，讓它能以全球相關的方式預測天氣模式。

GenCast的核心是從歷史天氣數據中學習，並根據這些學到的知識生成未來的天氣條件預測。該模型使用來自歐洲中期天氣預報中心 (ECMWF) 的40年天氣數據進行訓練，利用各種變量，如溫度、風速和不同高度的氣壓。這使得GenCast能夠以高解析度（0.25°）學習和建模全球天氣模式，顯著提升其預測能力。

GenCast是一個概率天氣模型，旨在生成高準確度的全球15天集成預測。它以0.25°的細緻解析度運行，在預測準確性方面超越了傳統的運行集成系統，如ECMWF的ENS。GenCast通過基於當前和過去的天氣條件建模未來天氣狀態的條件概率分佈來運作。

GenCast的主要特點

以下是GenCast的特點：

1. 預測解析度和速度：

全球覆蓋，解析度為0.25°：GenCast以0.25°的細緻解析度生成預測，提供詳細的全球天氣預測。

快速預測生成：每個15天的預測大約在8分鐘內生成，使用Cloud TPUv5設備。此外，這樣的預測集成可以並行生成。

2. 概率方法：

GenCast建模條件概率分佈以預測未來的天氣狀態。

預測軌跡是通過條件化初始和先前的天氣狀態生成的，模型迭代地考慮連續狀態的聯合分佈。

3. 模型表示：

天氣狀態表示：全球天氣狀態由六個表面變量和六個大氣變量表示，每個變量在0.25°的緯度-經度網格上定義於13個垂直壓力層。

預測範圍：GenCast生成15天的預測，每12小時進行一次預測，共產生30個時間步。

4. 擴散模型架構：

生成模型：GenCast作為條件擴散模型實現，這種模型通過隨機噪聲迭代地改進預測。這種類型的模型最近在生成式人工智慧中獲得了廣泛應用，特別是在涉及圖像、聲音和視頻的任務中。

自回歸過程：從初始噪聲樣本開始，GenCast逐步改進，根據大氣的前兩個狀態進行條件化，以生成預測。這一過程是自回歸的，意味著每一步都依賴於前一步的輸出。這使得整個預測軌跡的生成成為可能。

5. 神經網絡架構：

編碼器-處理器-解碼器設計：GenCast應用了一種複雜的神經網絡架構，由三個主要組件組成：

編碼器：將輸入（嘈雜的候選狀態）和來自緯度-經度網格的條件信息映射到內部表示。

處理器：圖形變壓器處理編碼數據，圖中的每個節點都關注其在網格上的鄰居，幫助捕捉複雜的空間依賴性。

解碼器：將處理後的信息映射回緯度-經度網格上的精細預測狀態。

6. 訓練：

數據：GenCast使用來自歐洲中期天氣預報中心 (ECMWF) 的40年ERA5重分析數據（1979-2018）進行訓練。這些數據包括一組全面的大氣和表面變量，用於訓練模型。

擴散模型訓練：該模型使用標準的擴散模型去噪目標，幫助將嘈雜的預測樣本改進為更準確的預測。

7. 集成預測：

不確定性建模：在生成預測時，GenCast考慮了初始條件的不確定性。這是通過用ERA5數據同化的集成擾動來擾動初始ERA5重分析數據來實現的。這使得GenCast能夠生成多個預測軌跡，捕捉可能的未來天氣情境範圍。

8. 評估：

重分析初始化：為了進行評估，GenCast使用ERA5重分析數據以及來自EDA的擾動進行初始化，確保考慮到初始不確定性。這使得模型能夠生成穩健的預測集，提供對潛在未來天氣條件的更全面展望。

GenCast代表了天氣預測的一次重大飛躍，通過結合擴散模型、高級神經網絡和集成技術，生成高準確度的概率15天預測。其自回歸設計，加上對大氣和表面變量的詳細表示，使其能夠在全球範圍內生成現實且多樣的天氣預測。

人工智慧驅動的速度和準確性

GenCast最令人印象深刻的特點之一是其速度。如前所述，僅使用一個Google Cloud TPU v5芯片，GenCast可以在8分鐘內生成15天的預測。這比傳統的基於物理的模型有了顯著改善，後者通常需要大量的超級計算資源，並且需要幾個小時才能生成類似的預測。GenCast通過並行運行所有集成預測來實現這一點，提供快速、高解析度的預測，這是傳統模型無法比擬的。

在準確性方面，GenCast經過嚴格測試，與ECMWF的ENS（歐洲中期天氣預報中心的運行集成模型）進行比較，後者是最廣泛使用的預測系統之一。在97.2%的測試案例中，GenCast的表現超過了ENS，顯示出其優越的準確性，特別是在預測極端天氣事件時。

精確應對極端天氣

極端天氣事件——例如熱浪、寒潮和強風——是需要精確預測的最關鍵因素之一。GenCast在這方面表現出色，提供可靠的極端條件預測，使得及時的預防行動得以實施，保護生命、減少損失並節省成本。無論是準備應對熱浪還是強風，GenCast在提供準確預測方面始終超越傳統系統。

此外，GenCast在預測熱帶氣旋（颶風和台風）的路徑方面顯示出更高的準確性。該模型能以更大的信心預測風暴的軌跡，提供更早的警告，這可以顯著改善災害準備工作。

GenCast迷你演示

這裡有更多的閱讀鏈接：

要使用Google Cloud Compute運行其他模型，請參考gencast_demo_cloud_vm.ipynb。

注意：此包提供四個預訓練模型：

還可以閱讀：GenCast：基於擴散的中期天氣集成預測

我們使用的是：

GenCast 1p0deg Mini <2019，一個解析度為1度的GenCast模型，具有13個壓力層和4次細化的二十面體網格。它是基於1979年至2018年的ERA5數據進行訓練的，並可以在2019年及以後進行因果評估。

原因：這個模型的內存佔用最小，是提供的模型中唯一可以在Colab中免費提供的TPUv2-8配置下運行的模型。

要獲取包含的權重：

您可以訪問Google Deepmind提供的Cloud存儲。它包含GenCast的所有數據，包括統計數據和參數。

GenCast迷你實現

升級包（運行此單元後需要重啟內核）

# @title 升級包（運行此單元後需要重啟內核）。
%pip install -U importlib_metadata

安裝repo和依賴項

# @title 安裝repo和依賴項
%pip install –upgrade https://github.com/deepmind/graphcast/archive/master.zip

導入

import dataclasses
import datetime
import math
from google.cloud import storage
from typing import Optional
import haiku as hk
from IPython.display import HTML
from IPython import display
import ipywidgets as widgets
import jax
import matplotlib
import matplotlib.pyplot as plt
from matplotlib import animation
import numpy as np
import xarray
from graphcast import rollout
from graphcast import xarray_jax
from graphcast import normalization
from graphcast import checkpoint
from graphcast import data_utils
from graphcast import xarray_tree
from graphcast import gencast
from graphcast import denoiser
from graphcast import nan_cleaning

繪圖函數

def select(
data: xarray.Dataset,
variable: str,
level: Optional[int] = None,
max_steps: Optional[int] = None
) -> xarray.Dataset:
data = data[variable]
if “batch” in data.dims:
data = data.isel(batch=0)
if max_steps is not None and “time” in data.sizes and max_steps < data.sizes["time"]:
data = data.isel(time=range(0, max_steps))
if level is not None and “level” in data.coords:
data = data.sel(level=level)
return data

def scale(
data: xarray.Dataset,
center: Optional[float] = None,
robust: bool = False,
) -> tuple[xarray.Dataset, matplotlib.colors.Normalize, str]:
vmin = np.nanpercentile(data, (2 if robust else 0))
vmax = np.nanpercentile(data, (98 if robust else 100))
if center is not None:
diff = max(vmax – center, center – vmin)
vmin = center – diff
vmax = center + diff
return (data, matplotlib.colors.Normalize(vmin, vmax),
(“RdBu_r” if center is not None else “viridis”))

def plot_data(
data: dict[str, xarray.Dataset],
fig_title: str,
plot_size: float = 5,
robust: bool = False,
cols: int = 4
) -> tuple[xarray.Dataset, matplotlib.colors.Normalize, str]:
first_data = next(iter(data.values()))[0]
max_steps = first_data.sizes.get(“time”, 1)
assert all(max_steps == d.sizes.get(“time”, 1) for d, _, _ in data.values())
cols = min(cols, len(data))
rows = math.ceil(len(data) / cols)
figure = plt.figure(figsize=(plot_size * 2 * cols,
plot_size * rows))
figure.suptitle(fig_title, fontsize=16)
figure.subplots_adjust(wspace=0, hspace=0)
figure.tight_layout()
images = []
for i, (title, (plot_data, norm, cmap)) in enumerate(data.items()):
ax = figure.add_subplot(rows, cols, i+1)
ax.set_xticks([])
ax.set_yticks([])
ax.set_title(title)
im = ax.imshow(
plot_data.isel(time=0, missing_dims=”ignore”), norm=norm,
origin=”lower”, cmap=cmap)
plt.colorbar(
mappable=im,
ax=ax,
orientation=”vertical”,
pad=0.02,
aspect=16,
shrink=0.75,
cmap=cmap,
extend=(“both” if robust else “neither”))
images.append(im)

def update(frame):
if “time” in first_data.dims:
td = datetime.timedelta(microseconds=first_data[“time”][frame].item() / 1000)
figure.suptitle(f”fig_title, td”, fontsize=16)
else:
figure.suptitle(fig_title, fontsize=16)
for im, (plot_data, norm, cmap) in zip(images, data.values()):
im.set_data(plot_data.isel(time=frame, missing_dims=”ignore”))
ani = animation.FuncAnimation(
fig=figure, func=update, frames=max_steps, interval=250)
plt.close(figure.number)
return HTML(ani.to_jshtml())

加載數據並初始化模型

使用Google Cloud Storage進行身份驗證

# 給你一個經過身份驗證的客戶端，以防你想使用私有桶。
gcs_client = storage.Client.create_anonymous_client()
gcs_bucket = gcs_client.get_bucket(“dm_graphcast”)
dir_prefix = “gencast/”

加載模型參數

選擇獲取模型參數的兩種方式之一：

隨機：你將獲得隨機預測，但可以更改模型架構，這可能運行得更快或適合你的設備。

檢查點：你將獲得合理的預測，但限於其訓練的模型架構，這可能不適合你的設備。

選擇模型

params_file_options = [
name for blob in gcs_bucket.list_blobs(prefix=(dir_prefix+”params/”))
if (name := blob.name.removeprefix(dir_prefix+”params/”))] # 去掉空字符串。
latent_value_options = [int(2**i) for i in range(4, 10)]
random_latent_size = widgets.Dropdown(
options=latent_value_options, value=512,description=”潛在大小：”)
random_attention_type = widgets.Dropdown(
options=[“splash_mha”, “triblockdiag_mha”, “mha”], value=”splash_mha”, description=”注意：”)
random_mesh_size = widgets.IntSlider(
value=4, min=4, max=6, description=”網格大小：”)
random_num_heads = widgets.Dropdown(
options=[int(2**i) for i in range(0, 3)], value=4,description=”頭數：”)
random_attention_k_hop = widgets.Dropdown(
options=[int(2**i) for i in range(2, 5)], value=16,description=”注意k跳：”)

def update_latent_options(*args):
def _latent_valid_for_attn(attn, latent, heads):
head_dim, rem = divmod(latent, heads)
if rem != 0:
return False
# splash attn所需。
if head_dim % 128 != 0:
return attn != “splash_mha”
return True

attn = random_attention_type.value
heads = random_num_heads.value
random_latent_size.options = [
latent for latent in latent_value_options
if _latent_valid_for_attn(attn, latent, heads)]

# 觀察變化以僅允許有效的組合。
random_attention_type.observe(update_latent_options, “value”)
random_latent_size.observe(update_latent_options, “value”)
random_num_heads.observe(update_latent_options, “value”)
params_file = widgets.Dropdown(
options=[f for f in params_file_options if “Mini” in f],
description=”參數文件：”,
layout={“width”: “max-content”})

source_tab = widgets.Tab([
widgets.VBox([
random_attention_type,
random_mesh_size,
random_num_heads,
random_latent_size,
random_attention_k_hop
]),
params_file,
])
source_tab.set_title(0, “隨機”)
source_tab.set_title(1, “檢查點”)
widgets.VBox([
source_tab,
widgets.Label(value=”運行下一個單元以加載模型。重新運行此單元將清除你的選擇。”)
])

加載模型

source = source_tab.get_title(source_tab.selected_index)
if source == “隨機”:
params = None # 下面填寫
state =
task_config = gencast.TASK
# 使用默認值。
sampler_config = gencast.SamplerConfig()
noise_config = gencast.NoiseConfig()
noise_encoder_config = denoiser.NoiseEncoderConfig()
# 配置，否則使用默認值。
denoiser_architecture_config = denoiser.DenoiserArchitectureConfig(
sparse_transformer_config = denoiser.SparseTransformerConfig(
attention_k_hop=random_attention_k_hop.value,
attention_type=random_attention_type.value,
d_model=random_latent_size.value,
num_heads=random_num_heads.value
),
mesh_size=random_mesh_size.value,
latent_size=random_latent_size.value,
)

else:
assert source == “檢查點”
with gcs_bucket.blob(dir_prefix + f”params/params_file.value”).open(“rb”) as f:
ckpt = checkpoint.load(f, gencast.CheckPoint)
params = ckpt.params
state =
task_config = ckpt.task_config
sampler_config = ckpt.sampler_config
noise_config = ckpt.noise_config
noise_encoder_config = ckpt.noise_encoder_config
denoiser_architecture_config = ckpt.denoiser_architecture_config
print(“模型描述：\n”, ckpt.description, “\n”)
print(“模型許可證：\n”, ckpt.license, “\n”)

加載示例數據

示例ERA5數據集可用於0.25度和1度解析度。

示例HRES-fc0數據集可用於0.25度解析度。

一些變換來自基礎數據集：我們將12小時的降水量累積，而不是默認的1小時。

對於HRES-fc0海面溫度，我們將ERA5數據集中海面溫度為NaN的網格單元分配NaNs（這在所有時間內保持不變）。

數據解析度必須與加載的模型匹配。由於我們正在運行GenCast Mini，這將是1度。

獲取並過濾可用示例數據集的列表

dataset_file_options = [
name for blob in gcs_bucket.list_blobs(prefix=(dir_prefix + “dataset/”))
if (name := blob.name.removeprefix(dir_prefix+”dataset/”))] # 去掉空字符串。

def parse_file_parts(file_name):
return dict(part.split(“-“, 1) for part in file_name.split(“_”))

def data_valid_for_model(file_name: str, params_file_name: str):
“””檢查數據類型和解析度是否匹配。”””
if source == “隨機”:
return True
data_file_parts = parse_file_parts(file_name.removesuffix(“.nc”))
res_matches = data_file_parts[“res”].replace(“.”, “p”) in params_file_name.lower()
source_matches = “Operational” in params_file_name
if data_file_parts[“source”] == “era5”:
source_matches = not source_matches
return res_matches and source_matches

dataset_file = widgets.Dropdown(
options=[
(“, “.join([f”k: v” for k, v in parse_file_parts(option.removesuffix(“.nc”)).items()]), option)
for option in dataset_file_options
if data_valid_for_model(option, params_file.value)
],
description=”數據集文件：”,
layout={“width”: “max-content”})

widgets.VBox([
dataset_file,
widgets.Label(value=”運行下一個單元以加載數據集。重新運行此單元將清除你的選擇並重新過濾與模型匹配的數據集。”)
])

加載天氣數據

with gcs_bucket.blob(dir_prefix+f”dataset/dataset_file.value”).open(“rb”) as f:
example_batch = xarray.load_dataset(f).compute()
assert example_batch.dims[“time”] >= 3 # 2個輸入，>=1個目標
print(“, “.join([f”k: v” for k, v in parse_file_parts(dataset_file.value.removesuffix(“.nc”)).items()]))
example_batch

選擇要繪製的數據

plot_example_variable = widgets.Dropdown(
options=example_batch.data_vars.keys(),
value=”2m_temperature”,
description=”變量”)
plot_example_level = widgets.Dropdown(
options=example_batch.coords[“level”].values,
value=500,
description=”層級”)
plot_example_robust = widgets.Checkbox(value=True, description=”穩健”)
plot_example_max_steps = widgets.IntSlider(
min=1, max=example_batch.dims[“time”], value=example_batch.dims[“time”],
description=”最大步數”)
widgets.VBox([
plot_example_variable,
plot_example_level,
plot_example_robust,
plot_example_max_steps,
widgets.Label(value=”運行下一個單元以繪製數據。重新運行此單元將清除你的選擇。”)
])

繪製示例數據

plot_size = 7
data =
” “: scale(select(example_batch, plot_example_variable.value, plot_example_level.value, plot_example_max_steps.value),
robust=plot_example_robust.value),

fig_title = plot_example_variable.value
if “level” in example_batch[plot_example_variable.value].coords:
fig_title += f” at plot_example_level.value hPa”
plot_data(data, fig_title, plot_size, plot_example_robust.value)

這個輸出是以開爾文為單位！

此外，關於提取訓練和評估數據、加載正規化數據、構建jitted函數、可能初始化隨機權重、運行模型（自回歸回放（Python中的循環））、繪製預測樣本和差異、繪製集成均值和CRPS、訓練模型、損失計算和梯度計算的更多信息，請查看此repo：gencast_mini_demo.ipynb

現實世界的應用和好處

GenCast的能力超越了災害管理。其高準確度的預測可以幫助社會的各個其他部門，最顯著的是可再生能源。例如，更好的風能發電預測可以提高風能的可靠性，這是可再生能源轉型的關鍵部分。一項原理測試顯示，GenCast在預測全球風力發電場的總風能產量方面比ENS更準確。這為更有效的能源規劃開辟了大門，並可能加速可再生能源的採用。

GenCast增強的天氣預測還將在食品安全、農業和公共安全方面發揮作用，準確的天氣預測對於決策至關重要。無論是計劃作物種植還是災害應對，GenCast的預測都可以幫助指導關鍵行動。

還可以閱讀：GenCast：我們的新人工智慧模型提供更準確的天氣結果，更快。

推進氣候理解

GenCast是Google更大願景的一部分，旨在實現人工智慧驅動的天氣預測，包括Google DeepMind和Google Research開發的其他模型，如NeuralGCM、SEEDS，以及預測洪水和野火。這些模型旨在提供更詳細的天氣預測，涵蓋更廣泛的環境因素，包括降水和極端高溫。

通過與氣象機構的合作，Google打算繼續推進基於人工智慧的天氣模型，同時確保傳統模型在預測中仍然不可或缺。傳統模型為像GenCast這樣的系統提供了必要的訓練數據和初始天氣條件，確保人工智慧和傳統氣象學相輔相成，達到最佳效果。

為了促進進一步的研究和發展，Google決定向更廣泛的社區發布GenCast的模型代碼、權重和預測。這一決定旨在使氣象學家、數據科學家和研究人員能夠將GenCast的先進預測能力整合到他們自己的模型和研究工作流程中。

通過開放GenCast，Google希望鼓勵天氣和氣候科學社區之間的合作，包括與學術研究人員、可再生能源公司、災害應對組織和專注於食品安全的機構的夥伴關係。GenCast的開放發布將推動天氣預測技術的更快進步，幫助提高對氣候變化和極端天氣事件的韌性。

結論

GenCast代表了天氣預測的新前沿：結合人工智慧和傳統氣象學，提供更快、更準確和概率性的預測。憑藉其開源模型、快速處理和卓越的預測能力，GenCast有望改變我們對天氣預測、風險管理和氣候適應的看法。

隨著我們的前進，像GenCast這樣的人工智慧模型與傳統預測系統之間的持續合作將在提高天氣預測的準確性和速度方面至關重要。這種合作無疑將使全球數百萬人受益，幫助社會更好地為極端天氣事件和氣候變化帶來的挑戰做好準備。