<!-- SPDX-FileCopyrightText: 2026 CoreWeave, Inc. SPDX-License-Identifier: Apache-2.0 SPDX-PackageName: skills -->

W&B Primary Skill

環境のデフォルト設定

• Python: python でスクリプトを実行、uv add でパッケージをインストール
• LLM: OpenAI gpt-5.4 (推論: 高、エンドポイント: responses)

---

クイックレシピ — これらを最初に使用

最も一般的なタスクをカバーしています。それぞれ単一のスクリプトです。コピーして、プレースホルダーを入力して、実行してください。

ランをカウント (正確で高速)

import wandb, os
api = wandb.Api(timeout=60)
path = f"{os.environ['WANDB_ENTITY']}/{os.environ['WANDB_PROJECT']}"
total = len(api.runs(path, per_page=1, include_sweeps=False, lazy=True))
finished = len(api.runs(path, filters={"state": "finished"}, per_page=1, include_sweeps=False, lazy=True))
crashed = len(api.runs(path, filters={"state": "crashed"}, per_page=1, include_sweeps=False, lazy=True))
running = len(api.runs(path, filters={"state": "running"}, per_page=1, include_sweeps=False, lazy=True))
print(f"Total: {total}  |  Finished: {finished}  |  Crashed: {crashed}  |  Running: {running}")

トレースをカウント (高速、サーバー側)

import weave, os, logging
logging.getLogger("weave").setLevel(logging.ERROR)
from weave.trace_server.trace_server_interface import CallsQueryStatsReq

entity = os.environ["WANDB_ENTITY"]
project = os.environ["WANDB_PROJECT"]
client = weave.init(f"{entity}/{project}")
pid = f"{entity}/{project}"

# Total root traces
stats = client.server.calls_query_stats(CallsQueryStatsReq(
    project_id=pid, filter={"trace_roots_only": True}
))
print(f"Root traces: {stats.count}")

# Count by op name
for op in ["Evaluation.evaluate", "my_op.turn"]:
    s = client.server.calls_query_stats(CallsQueryStatsReq(
        project_id=pid,
        filter={"op_names": [f"weave:///{entity}/{project}/op/{op}:*"]},
    ))
    print(f"  {op}: {s.count}")

プロジェクトをサマリー化 (ラン + トレース を1つのスクリプトで)

import wandb, weave, os, logging
logging.getLogger("weave").setLevel(logging.ERROR)
from weave.trace_server.trace_server_interface import CallsQueryStatsReq

entity = os.environ["WANDB_ENTITY"]
project = os.environ["WANDB_PROJECT"]
path = f"{entity}/{project}"

# --- Runs ---
api = wandb.Api(timeout=60)
total_runs = len(api.runs(path, per_page=1, include_sweeps=False, lazy=True))
finished = len(api.runs(path, filters={"state": "finished"}, per_page=1, include_sweeps=False, lazy=True))
recent = api.runs(path, order="-created_at", per_page=5)[:5]

print(f"=== Runs ({total_runs} total, {finished} finished) ===")
for r in recent:
    print(f"  {r.name} [{r.state}] {r.created_at[:10]}")

# --- Weave Traces ---
client = weave.init(path)
pid = f"{entity}/{project}"
root_stats = client.server.calls_query_stats(CallsQueryStatsReq(
    project_id=pid, filter={"trace_roots_only": True}
))
print(f"\n=== Weave Traces ({root_stats.count} root traces) ===")

recent_calls = list(client.get_calls(
    sort_by=[{"field": "started_at", "direction": "desc"}],
    limit=5,
    columns=["op_name", "started_at", "display_name"],
))
for c in recent_calls:
    name = c.display_name or c.op_name.split("/")[-1].split(":")[0]
    started = c.started_at.strftime("%Y-%m-%d %H:%M") if c.started_at else "?"
    print(f"  {name} @ {started}")

単一のランを検査

import wandb, os
api = wandb.Api(timeout=60)
path = f"{os.environ['WANDB_ENTITY']}/{os.environ['WANDB_PROJECT']}"

run = api.run(f"{path}/RUN_ID")
print(f"Name: {run.name}")
print(f"State: {run.state}")
print(f"Created: {run.created_at}")
print(f"Tags: {run.tags}")
print(f"Last step: {run.lastHistoryStep}")

# キーメトリクス (プローブまたはユーザーリクエストから実際のキーに置き換え)
for k in ["loss", "val_loss", "accuracy"]:
    v = run.summary_metrics.get(k)
    if v is not None:
        print(f"  {k}: {v}")

2つのランを比較

import wandb, os, sys
sys.path.insert(0, "skills/wandb-primary/scripts")
from wandb_helpers import get_api, compare_configs

api = get_api()
path = f"{os.environ['WANDB_ENTITY']}/{os.environ['WANDB_PROJECT']}"

run_a = api.run(f"{path}/RUN_A_ID")
run_b = api.run(f"{path}/RUN_B_ID")

# Config diff
diffs = compare_configs(run_a, run_b)
if diffs:
    print("Config differences:")
    for d in diffs:
        print(f"  {d['key']}: {d[run_a.name]} -> {d[run_b.name]}")
else:
    print("Configs are identical")

# Metric comparison
print("\nMetrics:")
for k in ["loss", "val_loss", "accuracy"]:
    a = run_a.summary_metrics.get(k, "N/A")
    b = run_b.summary_metrics.get(k, "N/A")
    print(f"  {k}: {a} vs {b}")

最新の評価をサマリー化

import weave, os, sys, logging
logging.getLogger("weave").setLevel(logging.ERROR)
from weave.trace.weave_client import CallsFilter
sys.path.insert(0, "skills/wandb-primary/scripts")
from weave_helpers import unwrap, eval_results_to_dicts, results_summary

entity = os.environ["WANDB_ENTITY"]
project = os.environ["WANDB_PROJECT"]
client = weave.init(f"{entity}/{project}")

# Get latest eval
op_ref = f"weave:///{entity}/{project}/op/Evaluation.evaluate:*"
evals = list(client.get_calls(
    filter=CallsFilter(op_names=[op_ref]),
    sort_by=[{"field": "started_at", "direction": "desc"}],
    limit=1,
))

if not evals:
    print("No evaluations found")
else:
    ec = evals[0]
    print(f"Eval: {ec.display_name or 'unnamed'} @ {ec.started_at}")

    # Get predict_and_score children
    pas_ref = f"weave:///{entity}/{project}/op/Evaluation.predict_and_score:*"
    pas = list(client.get_calls(
        filter=CallsFilter(op_names=[pas_ref], parent_ids=[ec.id])
    ))
    results = eval_results_to_dicts(pas, agent_name=ec.display_name or "agent")
    print(results_summary(results))

最近のトレースを検査

import weave, os, logging
logging.getLogger("weave").setLevel(logging.ERROR)
sys.path.insert(0, "skills/wandb-primary/scripts")
from weave_helpers import unwrap, get_token_usage

entity = os.environ["WANDB_ENTITY"]
project = os.environ["WANDB_PROJECT"]
client = weave.init(f"{entity}/{project}")

calls = list(client.get_calls(
    sort_by=[{"field": "started_at", "direction": "desc"}],
    limit=10,
))

for c in calls:
    name = c.display_name or c.op_name.split("/")[-1].split(":")[0]
    started = c.started_at.strftime("%Y-%m-%d %H:%M") if c.started_at else "?"
    duration = ""
    if c.started_at and c.ended_at:
        duration = f" ({(c.ended_at - c.started_at).total_seconds():.1f}s)"
    status = c.summary.get("weave", {}).get("status", "?") if c.summary else "?"
    tokens = get_token_usage(c)
    tok_str = f" [{tokens['total_tokens']} tok]" if tokens['total_tokens'] else ""
    print(f"  {name} [{status}] {started}{duration}{tok_str}")

W&B レポートを作成

import wandb, os
from wandb.apis import reports as wr

entity = os.environ["WANDB_ENTITY"]
project = os.environ["WANDB_PROJECT"]

runset = wr.Runset(entity=entity, project=project, name="All runs")
plots = wr.PanelGrid(
    runsets=[runset],
    panels=[
        wr.LinePlot(title="Loss", x="_step", y=["LOSS_KEY"]),
        wr.BarPlot(title="Accuracy", metrics=["ACC_KEY"], orientation="v"),
    ],
)

report = wr.Report(
    entity=entity,
    project=project,
    title="Project Analysis",
    description="Auto-generated summary",
    width="fixed",
    blocks=[
        wr.H1(text="Project Analysis"),
        wr.P(text="Auto-generated summary from W&B API."),
        plots,
    ],
)
report.save(draft=True)
print(f"Report saved: {report.url}")

Weave モニターを設定

import weave, os

entity = os.environ["WANDB_ENTITY"]
project = os.environ["WANDB_PROJECT"]
client = weave.init(f"{entity}/{project}")

# Define a scorer
@weave.op()
def my_scorer(output: dict) -> dict:
    """Score based on output quality."""
    # Replace with actual scoring logic
    passed = output.get("succeeded", False)
    return {"passed": passed, "score": 1.0 if passed else 0.0}

# Create monitor
monitor = weave.Monitor(
    entity=entity,
    project=project,
    name="quality-monitor",
    scorers=[my_scorer],
    # Filter which ops to monitor:
    # op_names=["my_agent.run"],
)
print(f"Monitor created: {monitor.name}")

ランを再起動 (1つのコマンド、キューを自動選択)

python skills/wandb-primary/scripts/launch_helpers.py relaunch \
  "https://wandb.ai/entity/project/runs/run_id" \
  --config '{"epochs": 100}'

ランを再起動 (Python)

import sys
sys.path.insert(0, "skills/wandb-primary/scripts")
from launch_helpers import parse_run_url, list_queues, relaunch_run

entity, project, run_id = parse_run_url("RUN_URL")
run_path = f"{entity}/{project}/{run_id}"

queues = list_queues(entity)
queue = queues[0]  # use the recommended queue

relaunch_run(
    run_path=run_path,
    queue_name=queue["name"],
    namespace=queue["namespace"],
    config={"lr": 0.001, "epochs": 20},
)

コードを変更して起動

import sys
sys.path.insert(0, "skills/wandb-primary/scripts")
from launch_helpers import get_job_artifact, download_code_artifact, list_queues, create_and_launch_modified_job

# Step 1: Download code
info = download_code_artifact("entity/project/job-name:latest")
# Edit files in info["code_dir"]...

# Step 2: Launch modified code
queues = list_queues("ENTITY")
queue = queues[0]
create_and_launch_modified_job(
    code_dir=info["code_dir"],
    entrypoint=info["entrypoint"],
    entity=info["entity"], project=info["project"],
    queue_name=queue["name"], namespace=queue["namespace"],
    job_name="my-modified-job",
    base_image=info["base_image"],
)

起動されたランのステータスを確認

python skills/wandb-primary/scripts/launch_helpers.py check \
  "entity" "project" "queue-name" "QUEUE_ITEM_ID"

---

Launch ルール

1. ターンを最小化。 単純な再起動で設定変更がある場合、1つのコマンドを使用: python skills/wandb-primary/scripts/launch_helpers.py relaunch <URL> --config '{"epochs": 100}'
2. CLI がすべてを自動処理。 relaunch はキューを自動発見し、最適なものを選択し、ジョブアーティファクトを見つけ、送信します。
3. launch_helpers.py を読まない。 このSKILL.mdで必要なすべてが記載されています。
4. Launch に WANDB_ENTITY/PROJECT 環境変数をチェックしない。 ラン URL にはエンティティとプロジェクトが含まれます。
5. wandb.init() で起動を偽物にしない。 relaunch_run() またはCLIを使用してください。
6. サンドボックスでトレーニングをローカルで実行しない。 GPU がありません。常に Launch を使用してください。
7. 設定変更 vs コード変更 — 最初に決定。

変更タイプ	例	起動方法
設定オーバーライド	epochs, lr, batch_size, wandb.config の任意の値	relaunch_run(..., config={"epochs": 100})
コード変更	モデルアーキテクチャ、損失関数、データ増強	コードをダウンロード → 編集 → create_and_launch_modified_job()

重要: ユーザーが設定フィールドではないものの変更を求めた場合 (例: "より多くの conv レイヤーを追加"、"オプティマイザーを変更")、コードを変更する必要があります。未知の設定キーを渡すと何もしません — トレーニングスクリプトがそれらを読みません。

Launch 意思決定ツリー

必要なこと	これを実行
ハイパーパラメータのみを変更	relaunch_run(run_path, queue_name, namespace, config={"epochs": 100})
コードを変更 (アーキテクチャ、ロジック)	download_code_artifact() -> ファイルを編集 -> create_and_launch_modified_job()
アーティファクトパスから起動	launch_job_artifact(artifact_path, queue_name)
新しいコードを送信 (既存ランなし)	submit_code_artifact_job(code_files, entrypoint, ...)
起動されたランを確認	check_launched_run(entity, project, queue_name, item_id)
キューを見つける	list_queues(entity) — 推奨されたものを使用
新しいキューを作成	create_queue(name, entity, gpus=1, cpu=8, memory="80Gi")

ステップバイステップ起動 (CLI ワンライナーが不十分な場合)

import sys
sys.path.insert(0, "skills/wandb-primary/scripts")
from launch_helpers import parse_run_url, list_queues, get_job_artifact, relaunch_run, submit_code_artifact_job

# 1. Parse the run URL
entity, project, run_id = parse_run_url("https://wandb.ai/entity/project/runs/run_id")
run_path = f"{entity}/{project}/{run_id}"

# 2. Find a queue (also gives namespace)
queues = list_queues(entity)
queue = queues[0]

# 3. Check for a job artifact
job_artifact = get_job_artifact(run_path)

# 4a. If job artifact exists -> relaunch with config overrides
if job_artifact:
    relaunch_run(run_path, queue["name"], queue["namespace"],
                 config={"lr": 0.001, "epochs": 20})

# 4b. If no job artifact -> submit code directly
else:
    submit_code_artifact_job(
        code_files=["train.py"], entrypoint="python train.py",
        entity=entity, project=project,
        queue_name=queue["name"], job_name="my-train-job",
        base_image="pytorch/pytorch:2.5.1-cuda12.4-cudnn9-runtime",
        requirements=["wandb"],
    )

コード変更ワークフロー — 1つではなく2つのスクリプト

スクリプト 1: コードをダウンロードして読む (編集対象を確認できるように):

import sys
sys.path.insert(0, "skills/wandb-primary/scripts")
from launch_helpers import parse_run_url, get_job_artifact, download_code_artifact
import os

entity, project, run_id = parse_run_url("RUN_URL")
art = get_job_artifact(f"{entity}/{project}/{run_id}")
info = download_code_artifact(f"{entity}/{project}/{art.name}")
# Prints: code_dir, files, entrypoint, base_image
for f in info["files"]:
    print(f"\n=== {f} ===")
    with open(os.path.join(info["code_dir"], f)) as fh:
        print(fh.read())

次に info["code_dir"] 内のファイルを write_file または apply_patch を使用して編集します。

スクリプト 2: 変更されたコードを起動:

import sys
sys.path.insert(0, "skills/wandb-primary/scripts")
from launch_helpers import list_queues, create_and_launch_modified_job

queues = list_queues("ENTITY")
queue = queues[0]
create_and_launch_modified_job(
    code_dir="CODE_DIR_FROM_STEP_1",
    entrypoint="python model.py",
    entity="ENTITY", project="PROJECT",
    queue_name=queue["name"], namespace=queue["namespace"],
    job_name="JOB_NAME", base_image="BASE_IMAGE",
)

Launch インフラストラクチャのうっかり

• 常に resource_args を明示的に渡す — キューのデフォルトはサーバーによってダブルネストされます
• キュー削除/再作成後にエージェントを再起動 — エージェントが登録を失います
• requirements.txt はコードディレクトリから読み込まれる — _create_job は venv を検査しません
• requirements.txt を最小限に保つ — ベースイメージに含まれていない deps のみ
• ベースイメージを linux/amd64 用に構築 — Mac のデフォルトではありません
• K8s シークレットは k8s_secrets パラメータで注入 — キューのデフォルト経由ではありません

---

クリティカル: 大規模プロジェクトのパフォーマンスルール

これらのルールは、10K+ ランまたは 1K+ メトリクスを持つランを持つプロジェクトで 502 エラー、タイムアウト、および複数分間のハングを防ぎます。これらのいずれかに違反すると、大規模プロジェクトで障害が発生します。

1. 常に wandb.Api(timeout=60) を使用 — デフォルトの 19 秒のタイムアウトは継続的な障害を引き起こします
2. 明示的な keys=[...] なしで history() または scan_history() を呼び出さない — 1K+ メトリクスを持つランは、すべての列を取得するときに 502 またはタイムアウトになります
3. リストタスクで api.runs() を呼び出すときは per_page=min(limit, 1000) を使用し、正確なカウントタスクでは per_page=1 を使用
4. サーバー側フィルターを優先 (summary_metrics.X: {$gt: Y}) — クライアント側の反復より
5. 正確なカウントでは、len(api.runs(..., per_page=1, include_sweeps=False, lazy=True)) を優先 — len(list(runs)) はしない
6. 10K+ 履歴ステップを持つランには beta_scan_history を使用 — GraphQL からではなくパーケットから読み込みます
7. すべての設定キーを反復しない — 明示的に必要な場合を除き、名前でキーにアクセスします
8. 読み取り専用取得タスクの場合、デフォルトで include_sweeps=False
9. トレース数をカウントするには calls_query_stats を使用 — カウントするだけで全呼び出しを実体化しない

---

何をいつ使用するか

必要なこと	使用
トレーニングラン、損失曲線、ハイパーパラメータをクエリ	W&B SDK (wandb.Api()) — references/WANDB_SDK.md を参照
GenAI トレース、呼び出し、評価をクエリ	Weave SDK (weave.init(), client.get_calls()) — references/WEAVE_SDK.md を参照
Weave ラッパータイプをプレーン Python に変換	weave_helpers.unwrap()
トレーニングランから DataFrame を構築	wandb_helpers.fetch_runs() (高速) または wandb_helpers.runs_to_dataframe()
分析用の評価結果を抽出	weave_helpers.eval_results_to_dicts()
トレースをフェッチせずにカウント	Weave サーバー API の calls_query_stats
低レベル Weave フィルタリング (CallsFilter, Query) が必要	Raw Weave SDK — references/WEAVE_SDK.md を参照
レポートを作成	wandb.apis.reports
本番モニタリングを設定	weave.Monitor
ランを再現/再起動	launch_helpers.relaunch_run() またはCLI
GPU/K8s でトレーニングジョブを起動	launch_helpers.submit_code_artifact_job()
コードを変更して起動	launch_helpers.download_code_artifact() -> 編集 -> create_and_launch_modified_job()
起動キューをリスト/作成	launch_helpers.list_queues() / create_queue()

---

バンドルされたファイル

ヘルパーライブラリ

import sys
sys.path.insert(0, "skills/wandb-primary/scripts")

# Weave helpers (traces, evals, GenAI)
from weave_helpers import (
    unwrap,                  # Weave タイプを再帰的にプレーン Python に変換
    get_token_usage,         # コールのサマリーからトークンカウントを抽出
    eval_results_to_dicts,   # predict_and_score 呼び出し -> 結果ディクトのリスト
    pivot_solve_rate,        # エージェント全体のタスクレベルピボットテーブルを構築
    results_summary,         # コンパクト評価サマリーを印刷
    eval_health,             # Evaluation.evaluate 呼び出しからステータス/カウントを抽出
    eval_efficiency,         # 評価呼び出し全体で成功あたりのトークンを計算
)

# W&B helpers (トレーニングラン、メトリクス) — 大規模プロジェクト最適化
from wandb_helpers import (
    get_api,             # 安全なタイムアウト付き API を作成 (デフォルト 60 秒)
    probe_project,       # クエリ前にプロジェクト規模、メトリクス、設定を発見
    fetch_runs,          # 高速: 選択的メトリクスで直接 GraphQL (17 倍高速)
    runs_to_dataframe,   # レガシー: ラン オブジェクトを反復 (遅い、代わりに fetch_runs を使用)
    diagnose_run,        # クイック診断サマリー (カスタマイズ可能なメトリクスキー)
    compare_configs,     # 2 つのラン間のサイドバイサイド設定差分
    scan_history,        # スマート履歴スキャン (大規模ランに対して自動的に beta_scan_history を選択)
)

# Launch helpers (ジョブ送信、ラン再現、キュー管理)
from launch_helpers import (
    parse_run_url,                       # W&B URL から (entity, project, run_id) を抽出
    list_queues,                         # すべての起動キューをリスト (ランク付け、推奨)
    get_job_artifact,                    # ランがジョブアーティファクトを持っているか確認
    inspect_job_artifact,                # ジョブアーティファクトのメタデータをダウンロード+検査
    download_code_artifact,              # ジョブアーティファクトからソースコードをダウンロード
    create_and_launch_modified_job,      # 変更されたコードをアップロード + 1 呼び出しで起動
    relaunch_run,                        # 設定オーバーライドで再実行 (コード変更なし)
    launch_job_artifact,                 # アーティファクトパスから直接起動
    submit_code_artifact_job,            # ジョブアーティファクトを作成し、1 呼び出しでエンキュー
    check_launched_run,                  # 起動されたランのステータス/メトリクスを確認
    create_queue,                        # K8s 起動キューを作成
    inspect_queue,                       # キュー詳細を印刷
    make_resource_args,                  # launch_add() の resource_args を構築
)

リファレンスドキュメント

必要に応じてこれらを読む — 完全な API サーフェスとレシピが含まれています:

• references/WANDB_CONCEPTS.md — W&B データモデル、用語、および曖昧性の解消 (entity/project/run 階層、config vs log vs summary、アーティファクト、レジストリ)。ユーザーが何を尋ねているかを理解するためにこれを読んでください。
• references/WANDB_SDK.md — トレーニングデータ用の W&B SDK (ラン、履歴、アーティファクト、スイープ、システムメトリクス)。API 呼び出しリファレンス。
• references/WEAVE_SDK.md — Weave SDK for GenAI トレース (client.get_calls(), CallsFilter, Query, 統計)。Weave クエリはここから始めてください。

---

重要なルール

プロジェクトごとにメトリクスキーを発見

コード例では、LOSS_KEY、VAL_LOSS_KEY、ACC_KEY、CONFIG_KEYS をプレースホルダーとして使用します。これらはプロジェクトによって異なります。各タスクの最初に probe_project() を使用するか、ユーザーのリクエストから発見してください。

# 間違い — ハードコードされたメトリクス名
rows = fetch_runs(api, path, metric_keys=["loss", "accuracy"])

# 正しい — probe_project またはユーザーのリクエストから発見
rows = fetch_runs(api, path, metric_keys=["train/loss", "train/acc"])

トレースとランを DATA として扱う

Weave トレースと W&B ラン履歴は非常に大きくなる可能性があります。生データを決してコンテキストにダンプしないでください。常に:

1. 最初に構造を検査 — 列名、dtypes、行数を見てください
2. pandas/numpy に読み込む — プログラムで統計を計算
3. サマリー、ダンプではなく — 生の行ではなく、計算された統計とテーブルを出力

常に最終的な答えを提供

分析の途中で作業を終わらせないでください。すべてのタスクは明確で構造化された応答で終わらなければなりません:

1. データをクエリ (最大 2 つのスクリプト)
2. 必要な番号を抽出
3. 提示: テーブル + 主な調査結果 + 各サブ質問への直接的な回答

「これで最終分析を構築する」と自分自身が言ったことに気づいたら、停止して持っているものを提示してください。

未知の Weave データに unwrap() を使用

Weave 出力に遭遇し、そのタイプが確実でない場合、最初にそれをアンラップします:

from weave_helpers import unwrap
import json

output = unwrap(call.output)
print(json.dumps(output, indent=2, default=str))

---

環境セットアップ

エンティティとプロジェクトは環境変数から来ます — ハードコードしないでください:

import os
entity  = os.environ["WANDB_ENTITY"]
project = os.environ["WANDB_PROJECT"]
path = f"{entity}/{project}"

---

キーパターン

非常に大規模なプロジェクトで正確にカウント (高速)

import wandb
api = wandb.Api(timeout=60)
path = f"{entity}/{project}"

total = len(api.runs(path, per_page=1, include_sweeps=False, lazy=True))
finished = len(api.runs(path, filters={"state": "finished"}, per_page=1, include_sweeps=False, lazy=True))

異なるタグ (O(1) — ランスキャンなし)

import wandb
from wandb_graphql.language import parser as gql_parser

api = wandb.Api(timeout=60)
doc = gql_parser.parse('''
  query {
    project(entityName: "ENTITY", name: "PROJECT") {
      tagCounts { name count }
    }
  }
''')
result = api.client.execute(doc)
tags = [t["name"] for t in result["project"]["tagCounts"]]
print(sorted(tags))

異なるグループ (O(1) — ランスキャンなし)

import wandb
from wandb_graphql.language import parser as gql_parser

api = wandb.Api(timeout=60)
doc = gql_parser.parse('''
  query {
    project(entityName: "ENTITY", name: "PROJECT") {
      groupedRuns(groupKeys: ["group"], first: 100) {
        ... on GroupedRunConnection {
          edges {
            node { group totalRuns }
          }
        }
      }
    }
  }
''')
result = api.client.execute(doc)
edges = result["project"]["groupedRuns"]["edges"]
groups = [e["node"]["group"] for e in edges if e["node"]["group"]]
print(sorted(groups))

W&B SDK — 高速ラン取得 (大規模プロジェクトで 17 倍高速)

import pandas as pd
from wandb_helpers import get_api, fetch_runs

api = get_api()
path = f"{entity}/{project}"

rows = fetch_runs(
    api, path,
    metric_keys=["LOSS_KEY", "ACC_KEY"],
    filters={"state": "finished"},
    limit=100,
)
df = pd.DataFrame(rows)
print(df.describe())

Weave — eval コール階層

Evaluation.evaluate (root)
  +-- Evaluation.predict_and_score (データセット行 x 試行ごとに 1 つ)
  |     +-- model.predict (実際のモデル呼び出し)
  |     +-- scorer_1.score
  |     +-- scorer_2.score
  +-- Evaluation.summarize

トークンの使用法

from weave_helpers import get_token_usage

usage = get_token_usage(call)
print(f"Tokens: {usage['total_tokens']} (in={usage['input_tokens']}, out={usage['output_tokens']})")

レポート作成 (W&B Reports)

from wandb.apis import reports as wr

runset = wr.Runset(entity=entity, project=project, name="All runs")
plots = wr.PanelGrid(
    runsets=[runset],
    panels=[
        wr.LinePlot(title="Loss", x="_step", y=["LOSS_KEY"]),
        wr.BarPlot(title="Accuracy", metrics=["ACC_KEY"], orientation="v"),
    ],
)

report = wr.Report(
    entity=entity, project=project,
    title="Project analysis",
    description="Summary of recent runs",
    width="fixed",
    blocks=[
        wr.H1(text="Project analysis"),
        wr.P(text="Auto-generated summary from W&B API."),
        plots,
    ],
)
report.save(draft=True)

---

うっかり

Weave API

うっかり	間違い	正しい
weave.init 引数	weave.init(project="x")	weave.init("x") (位置)
親フィルター	filter={'parent_id': 'x'}	filter={'parent_ids': ['x']} (複数形、リスト)
WeaveObject アクセス	rubric.get('passed')	getattr(rubric, 'passed', None)
ネストされた出力	out.get('succeeded')	out.get('output').get('succeeded') (output.output)
ObjectRef 比較	name_ref == "foo"	str(name_ref) == "foo"
CallsFilter インポート	from weave import CallsFilter	from weave.trace.weave_client import CallsFilter
Query インポート	from weave import Query	from weave.trace_server.interface.query import Query
Eval ステータスパス	summary["status"]	summary["weave"]["status"]
Eval 成功カウント	summary["success_count"]	summary["weave"]["status_counts"]["success"]
不確実な場合	型を推測	最初に unwrap() してから検査

W&B API

うっかり	間違い	正しい
API タイムアウト	wandb.Api() (19 秒デフォルト)	wandb.Api(timeout=60) または get_api()
サマリーアクセス	run.summary["loss"]	run.summary_metrics.get("LOSS_KEY")
すべてのランを読み込む	list(api.runs(...))	runs[:200] (常にスライス)
ランをカウント	len(list(api.runs(...)))	len(api.runs(..., per_page=1, include_sweeps=False, lazy=True))
異なるタグ	すべてのランを反復して run.tags を収集	GraphQL tagCounts クエリ
異なるグループ	すべてのランを反復して run.group を収集	GraphQL groupedRuns クエリ
遅延フェッチ後の run.config	run.config が {} を返す	config が必要な場合は lazy=False を使用
ページネーション	api.runs(path) (per_page=50 デフォルト)	api.runs(path, per_page=min(N, 1000))
履歴 — 大規模ランでキーなし	run.history(samples=10) -> 502	run.history(samples=10, keys=["LOSS_KEY"])
scan_history — キーなし	scan_history() -> タイムアウト	scan_history(keys=["LOSS_KEY"])
大規模履歴 (10K+ ステップ)	scan_history(keys=[...])	beta_scan_history(keys=[...]) (parquet)
クロスラン検索	クライアント側のすべてのランを反復	サーバー側フィルター: {"summary_metrics.X": {"$gt": Y}}

Launch

うっかり	間違い	正しい
キューのリスト	api.run_queues() または raw GQL	ヘルパーから list_queues(entity)
resource_args	キューのデフォルトに依存	make_resource_args() 経由で渡す
requirements.txt	venv から pip freeze	手動で記述 — ベースイメージに足りない deps のみ
ベースイメージアーキ	Mac で docker build	docker buildx build --platform linux/amd64
偽の起動	wandb.init() 設定で	relaunch_run() または launch_job_artifact()
未知の設定キー	relaunch_run(config={"conv_layers": 4})	コード変更 — ダウンロード、編集、create_and_launch_modified_job()

Weave ログのノイズ

import logging
logging.getLogger("weave").setLevel(logging.ERROR)