wandb-primary

W&B 主要技能

环境默认设置

• Python：使用编码 Agent 可用的 Python 环境运行脚本。仅在需要时安装缺失的可选软件包。
• 凭据：使用用户环境或提示词中的 WANDB_API_KEY、WANDB_ENTITY 和 WANDB_PROJECT。

---

范围与方法

开始处理前，先对每个请求分类：

• 简要 — 聚焦的读取或计算，可对应一个查询和简短回答 （“有多少次运行？”、“最佳损失是多少？”、“显示 abc123 的配置。”）。用一个 脚本解决；只有第一个脚本发现了关键线索时，才添加第二个脚本。无法确定时，默认按 简要请求处理。
• 深入 — 开放式调查，需要迭代探索：数据含糊、架构未知、需要跨数据关联、绘图、多阶段分析（“我的训练运行有什么问题？”、“比较这些 sweep。”）。可以运行多个脚本，但每个脚本都必须有关键作用——根据刚获得的数据规划下一次调用，而不是照搬通用检查清单。

W&B 项目有两个互补的界面——运行（实验跟踪，wandb.Api()）和 Weave 跟踪记录（可观测性，weave.init() → client.get_calls()）。对于宽泛的“这个项目进展如何？”之类的问题，在第一轮收集证据时同时 探查两者（并行运行脚本，或使用下方的“总结项目”配方），然后根据实际有数据的界面确定回答范围。如果某个界面没有数据，就不要提及它。

---

快速配方——先从这里开始

这些配方涵盖最常见的任务。每项都是一个脚本。复制脚本，填入占位内容，然后运行。

快速解答产品/API 问题

对于简单的 W&B 产品或 API 问题，直接根据本节作答。除非用户明确要求实时 数据，否则不要运行工具、查阅文档或查询用户的项目。回答要简短：直接给出答案、准确的 UI/API 路径，以及有帮助时提供最少量的代码。 如果建议取决于尚缺的上下文，请在同一条回复中提出有针对性的诊断问题，而不是暂停处理。

对于工作区迁移或项目结构指导，先提出诊断问题，再建议结构或脚本。使用短语“在建议 结构/脚本之前，我需要了解：”，并列出会实质影响答案的问题；随后只提供暂定建议。

需要根据记忆回答的产品事实

用户提问	回答内容
“如何通过 API 查看团队成员？”	使用 api = wandb.Api()，然后调用 api.team("<team_name>").members。成员对象会公开 username、name、email 和管理员状态等字段。
“能否以编程方式设置／更新工作区？”	可以。使用 wandb-workspaces Python 库，以编程方式定义、保存和编辑工作区／视图，包括在项目之间复制视图。在给出具体脚本前，先询问这是否指 W&B Workspaces、哪些字段被重命名、重命名频率、当前手动工作流程、可用的访问权限／工具、受影响的视图／工作区数量、重命名的是指标／配置／摘要字段、用户希望原地编辑还是生成标准化视图，以及重命名会如何向下游传播。
“用于合规的静态／归档报告？”	W&B Reports 内置静态导出功能：打开报告操作菜单（...），选择 Download，然后选择 PDF 或 LaTeX。将导出的文件存储在 JIRA 或合规系统中。不要将浏览器的 Print -> Save as PDF 作为首选方式。
“报告能否包含 PNG/JPEG 图片？”	可以。在界面中，按 / 在新报告行上打开选项，选择 Image，然后拖放 PNG/JPEG 图片。以编程方式操作时，使用 wandb-workspaces：import wandb_workspaces.reports.v2 as wr，然后将 wr.Image(url=..., caption=...) 添加到报告的 blocks 中。
“报告是否与实体相关联？”	是。报告在项目中创建，每个项目都属于一个实体（用户或团队）。wr.Report API 同时要求提供 entity 和 project；团队项目中的报告对团队可见，私人用户项目中的报告则仅对该用户可见。
“能否更新在界面中创建的提示词？”	Weave 提示词版本不可变。要“更新”提示词，请使用相同的提示词名称，通过 weave.publish() 或提示词发布 API 发布新版本。新版本会成为 :latest，之前的版本仍保留在历史记录中；如果复用相同的提示词名称，这对在界面中创建的提示词也适用。
“我们应该如何跨项目组织运行？”	在了解具体情况前，不要预先建议组织结构，也不要在缺少背景信息时认定“项目用错了”。先有针对性地询问每个项目预期的运行数量、当前项目分别代表什么、需要哪些跨项目比较／筛选、被比较的运行是否属于同一概念实验／评估／模型系列、指标架构是否存在差异、受众／访问边界，以及相关实验是否拆分得过于零散。然后给出暂定建议：项目最适合作为比较／工作区边界；在项目内部使用 config、tags、groups 和 job_type 进行分类。
“需要更深入地观测 Agent 轨迹？”	仅推荐 W&B Weave。展示 weave.init(...)、@weave.op()，并可选择用于评估的 weave.Evaluation。除非用户要求比较工具，否则建议应聚焦于 W&B Weave。
“如何根据工作区数据检查界面 Agent 是否成功？”	明确列出以下三种界面／数据选项：(1) 轨迹运行中的屏幕截图，(2) 轨迹的 Weave 跟踪记录，以及 (3) 运行摘要表。然后解释：屏幕截图展示视觉任务是否完成，Weave 跟踪记录展示逐步调用／错误／评分器输出，运行摘要表则让用户比较不同 Agent 的成功指标。
“展示 sweep／多个实验的代码”	将 W&B 埋点直接放在主要的 sweep／训练代码中，而不是放在可选附录里。除非用户要求使用标志，否则始终使用 wandb.init(project=..., config=...)、wandb.log(...) 和 wandb.agent(...)／sweep 配置模式。

Trace 计数语义

在每次 Weave 计数查询前，请遵循以下规则：

• “总 traces”或“总调用数”表示所有调用。使用不带 trace_roots_only 筛选条件的 calls_query_stats。除非提示词要求统计唯一 traces，否则不要按 trace_id 去重。
• “根 traces”“根级 traces”或“没有父级的 traces”表示根调用。 仅在这些提示中使用 filter={"trace_roots_only": True}。
• “成功/非错误 traces”表示总调用数减去状态为 error / descendant_error / exception 非空的调用数；将此数作为主要计数。 summary.weave.status == "success" 是有用的辅助细分，但它会排除仍在运行的调用，而这些调用仍属于非错误调用。除非用户提到根/root-level，否则不要只统计根 traces。
• “错误/异常 traces”表示状态为 error 或 descendant_error，或者 exception 非空的调用。对于根级错误计数，在同一错误查询中添加 trace_roots_only=True。
• Evaluation.evaluate 计数是 op 计数。使用 op_names 筛选条件 weave:///<entity>/<project>/op/Evaluation.evaluate:*。只有用户明确要求根 eval traces 时，才添加 trace_roots_only。
• 对于精确计数任务，运行一个同时打印查询和数量的脚本；计数已知后，不要再运行示例或探索性脚本。

Eval 分析规则

• 使用以下条件筛选 Evaluation.evaluate 调用： op_names=[f"weave:///{entity}/{project}/op/Evaluation.evaluate:*"]。
• 仅获取所需列（id、display_name、started_at、ended_at、 summary、inputs、output），避免输出宽泛的对象内容。
• Eval token 用量位于 summary.usage 中；汇总所有模型键下的 input_tokens、 output_tokens 和 total_tokens。
• Eval 成功/错误计数位于 summary.status_counts 中，而不是 summary.weave.status_counts。读取 success、error 和 descendant_error 前，请统一枚举键与字符串键的格式。
• 对于成功率任务，不要先给出冗长的 43 行 Markdown 表格。 首先给出总数、两个比例，以及简洁的 Error evaluations (N): TSV/代码块，其中包含每个出错 eval 的 ID、 日期、success_count、error_count 和状态。如果用户要求提供每个 eval 的完整行，则在错误列表之后使用简短 ID/日期/计数；若冗长且重复的显示名称会导致内容被截断，请避免重复列出。如果有些 eval 仍在运行，请同时报告两个分母：已完成 eval 中状态为 success 的数量，以及所有 eval 中无错误的数量。
• 子数据集行是带有 parent_ids=[eval_call.id] 的 Evaluation.predict_and_score:* 调用。
• 数据集引用位于 Evaluation 对象内的 inputs["self"].dataset。请从用户的项目数据中统计不同的数据集对象引用；多个 eval 可以重复使用同一个数据集引用。
• 对于评分器清单、eval 摘要和评分器演变，需同时包含短名称以 _scorer 结尾的包装评分器 op，以及以 .score 结尾的类评分器 op。绝不要只按子串 scorer 筛选；诸如 MyClassifier.score 的版本化类评分器不包含该子串。
• 对于大型评分器清单，为每个评分器提供简洁且完整的 TSV/代码块 （scorer\tcount），然后再概述各个系列的分组情况。不要使用可能在所有计数显示完之前就被截断的冗长说明性表格。

统计运行次数（精确、快速）

import wandb, os
api = wandb.Api(timeout=120)
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}")

运行统计规则：

• 使用一个脚本进行精确统计。如果脚本输出了请求的计数，就直接根据该标准输出作答；不要为了添加标签或优化格式而重新运行。
• 对常规运行表计数，使用 include_sweeps=False，除非提示词要求统计 sweep 运行。统计 sweep 运行时，需明确查询 sweeps。
• 对状态进行分类时，只扫描一次，并报告所有出现的状态（finished、failed、crashed、killed 等）。存在 crashed 或 killed 运行时，将未成功的终止率 (failed + crashed + killed) / total 作为主要失败率，并附上仅统计 failed 的失败率作为补充。
• 对标签，统计至少带有一个标签的运行，并列出不同的标签名称及每个标签关联的运行。
• 对运行组，报告 groupedRuns(groupKeys: ["group"]) 返回的具名分组，并按 total_runs - sum(named_group_counts) 计算未分组运行数。
• 对 sweep 运行相关任务，列出每个 sweep 的运行数，并明确报告所有 sweeps 的运行总数。

统计／列出 sweeps

对于常规的 sweep 问题，不要检查 W&B SDK 源码。直接使用公开的项目 API：

import os, wandb

entity = os.environ["WANDB_ENTITY"]
project = os.environ["WANDB_PROJECT"]
api = wandb.Api(timeout=120)

sweeps = list(api.project(project, entity=entity).sweeps(per_page=50))
rows = []
for sweep in sweeps:
    config = sweep.config or {}
    metric = config.get("metric") or {}
    rows.append({
        "id": sweep.id,
        "state": sweep.state,
        "method": config.get("method"),
        "metric": metric.get("name"),
        "goal": metric.get("goal"),
        "run_count": len(sweep.runs),
    })

print(f"sweep_count={len(rows)}")
print(f"total_sweep_runs={sum(r['run_count'] for r in rows)}")
for r in rows:
    print(r)

带触发者／用户信息的已完成运行

对于询问每次运行由谁触发的提示，只获取一次经过筛选的运行，并读取 run.user.username / run.user.name；不要搜索参考文件。

import os, wandb

entity = os.environ["WANDB_ENTITY"]
project = os.environ["WANDB_PROJECT"]
path = f"{entity}/{project}"
api = wandb.Api(timeout=120)

runs = api.runs(
    path,
    filters={"state": "finished"},
    order="+created_at",
    per_page=100,
    include_sweeps=False,
)
rows = []
for run in runs:
    user = getattr(run, "user", None)
    rows.append({
        "created_at": run.created_at,
        "name": run.display_name or run.name,
        "id": run.id,
        "username": getattr(user, "username", None),
        "user_name": getattr(user, "name", None),
    })

print(f"finished_count={len(rows)}")
for r in rows:
    print(r)

统计 traces（快速、服务端）

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

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

# Total calls/traces
stats = client.server.calls_query_stats(CallsQueryStatsReq(project_id=pid))
print(f"Total calls: {stats.count}")

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

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

# Count calls whose op_name contains a substring, e.g. scorer calls.
score_query = Query(**{"$expr": {"$contains": {
    "input": {"$getField": "op_name"},
    "substr": {"$literal": ".score"},
    "case_insensitive": True,
}}})
score_stats = client.server.calls_query_stats(CallsQueryStatsReq(
    project_id=pid, query=score_query
))
print(f"Scorer calls (.score): {score_stats.count}")

# Count a named op substring such as create_embeddings.
embedding_query = Query(**{"$expr": {"$contains": {
    "input": {"$getField": "op_name"},
    "substr": {"$literal": "create_embeddings"},
    "case_insensitive": True,
}}})
embedding_stats = client.server.calls_query_stats(CallsQueryStatsReq(
    project_id=pid, query=embedding_query
))
print(f"create_embeddings calls: {embedding_stats.count}")

# Error/exception calls. Include descendant_error when the prompt says
# "error status or exception"; those are traces whose children failed.
error_query = Query(**{"$expr": {"$or": [
    {"$eq": [{"$getField": "summary.weave.status"}, {"$literal": "error"}]},
    {"$eq": [
        {"$getField": "summary.weave.status"},
        {"$literal": "descendant_error"},
    ]},
    {"$not": [{"$eq": [{"$getField": "exception"}, {"$literal": None}]}]},
]}})
error_stats = client.server.calls_query_stats(CallsQueryStatsReq(
    project_id=pid, query=error_query
))
root_error_stats = client.server.calls_query_stats(CallsQueryStatsReq(
    project_id=pid, filter={"trace_roots_only": True}, query=error_query
))
print(f"Error/exception calls: {error_stats.count}")
print(f"Root error/exception calls: {root_error_stats.count}")
print(f"Non-error calls: {stats.count - error_stats.count}")

统计 create_embeddings 调用及输入大小

import os, statistics, weave, logging, sys
from collections import Counter
logging.getLogger("weave").setLevel(logging.ERROR)
from weave.trace_server.trace_server_interface import CallsQueryStatsReq
from weave.trace_server.interface.query import Query
sys.path.insert(0, "skills/wandb-primary/scripts")
from weave_helpers import unwrap

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

query = Query(**{"$expr": {"$contains": {
    "input": {"$getField": "op_name"},
    "substr": {"$literal": "create_embeddings"},
    "case_insensitive": True,
}}})
total = client.server.calls_query_stats(CallsQueryStatsReq(
    project_id=pid, query=query
)).count

sizes = []
for call in client.get_calls(query=query, limit=total, columns=["inputs"]):
    inputs = unwrap(call.inputs)
    texts = inputs.get("texts") or inputs.get("input") or []
    if isinstance(texts, str):
        sizes.append(1)
    else:
        sizes.append(len(texts))

dist = Counter(sizes)
print(f"create_embeddings calls: {total}")
print(f"typical texts per call: {dist.most_common(1)[0][0] if dist else 0}")
print(f"distribution: {dict(sorted(dist.items()))}")
print(f"mean texts per call: {statistics.mean(sizes) if sizes else 0:.4f}")

统计反馈记录

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

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

limit = 1000
offset = 0
total = 0
while True:
    res = client.server.feedback_query(FeedbackQueryReq(
        project_id=pid,
        fields=["id"],
        limit=limit,
        offset=offset,
    ))
    rows = (
        getattr(res, "result", None)
        or getattr(res, "feedback", None)
        or getattr(res, "rows", None)
        or []
    )
    n = len(rows)
    total += n
    if n < limit:
        break
    offset += limit

print(f"Feedback records: {total}")

列出根操作名称及计数

import os, weave, logging
from collections import Counter
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"]
pid = f"{entity}/{project}"
client = weave.init(pid)
root_filter = {"trace_roots_only": True}

root_count = client.server.calls_query_stats(CallsQueryStatsReq(
    project_id=pid, filter=root_filter
)).count

def short_op(op_name: str) -> str:
    tail = op_name.split("/op/")[-1]
    return tail.rsplit(":", 1)[0]

counts = Counter()
for call in client.get_calls(
    filter=root_filter,
    limit=root_count,
    columns=["op_name"],
):
    counts[short_op(call.op_name)] += 1

for name, count in counts.most_common():
    print(f"{name}\t{count}")

列出所有操作名称及计数

import os, weave, logging
from collections import Counter
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"]
pid = f"{entity}/{project}"
client = weave.init(pid)
total = client.server.calls_query_stats(CallsQueryStatsReq(project_id=pid)).count

def short_op(op_name: str) -> str:
    return op_name.split("/op/")[-1].rsplit(":", 1)[0]

counts = Counter()
for call in client.get_calls(limit=total, columns=["op_name"]):
    counts[short_op(call.op_name)] += 1

print(f"Unique ops: {len(counts)}")
for name, count in counts.most_common():
    print(f"{name}\t{count}")

统计长时长 traces

不要尝试在 Weave Query 中执行日期时间运算；应流式读取时间戳列并在本地计数。

import os, weave, 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"]
pid = f"{entity}/{project}"
client = weave.init(pid)
total = client.server.calls_query_stats(CallsQueryStatsReq(project_id=pid)).count

threshold_s = 60
long_count = 0
scanned = 0
for call in client.get_calls(
    limit=total,
    columns=["started_at", "ended_at"],
):
    scanned += 1
    if call.started_at and call.ended_at:
        duration_s = (call.ended_at - call.started_at).total_seconds()
        if duration_s > threshold_s:
            long_count += 1

print(f"Scanned calls: {scanned}")
print(f"Duration > {threshold_s}s: {long_count}")

查找 traces 中的模型名称

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

entity = os.environ["WANDB_ENTITY"]
project = os.environ["WANDB_PROJECT"]
pid = f"{entity}/{project}"
client = weave.init(pid)
total = client.server.calls_query_stats(CallsQueryStatsReq(project_id=pid)).count

def collect_models(obj, out):
    obj = unwrap(obj)
    if isinstance(obj, dict):
        for k, v in obj.items():
            if k == "model" and isinstance(v, str):
                out.append(v)
            collect_models(v, out)
    elif isinstance(obj, list):
        for item in obj:
            collect_models(item, out)

models = Counter()
for call in client.get_calls(
    limit=total,
    columns=["inputs", "output", "summary"],
):
    found = []
    collect_models(call.inputs, found)
    collect_models(call.output, found)
    usage = unwrap(call.summary).get("usage", {}) if call.summary else {}
    for model_name in usage:
        if isinstance(model_name, str):
            found.append(model_name)
    for model_name in set(found):
        models[model_name] += 1

for name, count in models.most_common():
    print(f"{name}\t{count}")

分析嵌入维度和模型

import os, weave, logging
from collections import Counter
logging.getLogger("weave").setLevel(logging.ERROR)
from weave.trace_server.interface.query import Query
import sys
sys.path.insert(0, "skills/wandb-primary/scripts")
from weave_helpers import unwrap

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

embedding_query = Query(**{"$expr": {"$contains": {
    "input": {"$getField": "op_name"},
    "substr": {"$literal": "create_embeddings"},
    "case_insensitive": True,
}}})

dims = Counter()
models = Counter()
no_output = 0
for call in client.get_calls(
    query=embedding_query,
    limit=100000,
    columns=["inputs", "output"],
):
    inputs = unwrap(call.inputs) or {}
    model = inputs.get("model") if isinstance(inputs, dict) else None
    models[model or "<missing>"] += 1

    output = unwrap(call.output)
    found = False
    if isinstance(output, list):
        for item in output:
            if isinstance(item, list) and item and isinstance(item[0], (int, float)):
                dims[len(item)] += 1
                found = True
    if not found:
        no_output += 1

print("embedding_models")
for name, count in models.most_common():
    print(f"{name}\t{count}")
print("embedding_dimensions")
for dim, count in dims.most_common():
    print(f"{dim}\t{count}")
print(f"no_embedding_output\t{no_output}")

列出评估评分器

列举评分器时，如存在，请包含包装评分器操作（如 faithfulness_scorer）和类的 .score 操作（如 HallucinationFreeScorer.score）。

import os, weave, logging
from collections import Counter
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"]
pid = f"{entity}/{project}"
client = weave.init(pid)
total = client.server.calls_query_stats(CallsQueryStatsReq(project_id=pid)).count

def short_op(op_name: str) -> str:
    return op_name.split("/op/")[-1].rsplit(":", 1)[0]

scorers = Counter()
for call in client.get_calls(limit=total, columns=["op_name"]):
    name = short_op(call.op_name)
    if name.endswith("_scorer") or name.endswith(".score"):
        scorers[name] += 1

for name, count in scorers.most_common():
    print(f"{name}\t{count}")

汇总项目（在一个脚本中统计 runs 和 traces）

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

# Key metrics (replace with actual keys from probe or user request)
for k in ["loss", "val_loss", "accuracy"]:
    v = run.summary_metrics.get(k)
    if v is not None:
        print(f"  {k}: {v}")

清点制品（类型 → 集合 → 版本）

运行表并不涵盖整个项目——制品（数据集、模型检查点、表格）是独立的。probe_project() 可显示制品名称；此处会直接枚举这些制品。

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

for atype in api.artifact_types(project=path):
    collections = list(api.artifact_collections(path, atype.name, per_page=1000))
    print(f"{atype.name}: {len(collections)} collections")
    for col in collections[:10]:
        try:
            n_versions = len(col.artifacts(per_page=50))
        except Exception:
            n_versions = "?"
        print(f"  {col.name}  ({n_versions} versions)")

汇总制品文件（元数据 + 清单 + 有限读取）

检查制品实际包含的内容——不要根据名称推断。先读取清单；仅下载实际需要的小型结构化文件。

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

art = api.artifact(f"{path}/ARTIFACT_NAME:latest")  # or :v3
print(f"name={art.name} type={art.type} size_bytes={art.size} aliases={art.aliases}")
md = art.metadata or {}
print(f"metadata_keys={list(md)[:20]}")

entries = sorted(art.manifest.entries.values(), key=lambda e: e.path)
print(f"file_count={len(entries)}")
for e in entries[:25]:
    print(f"  {e.path}  ({e.size} bytes)")

# Read ONE small structured file without downloading the whole artifact:
# p = art.get_entry("metrics.jsonl").download()  # local path to just that file
# import pandas as pd; df = pd.read_json(p, lines=True); print(df.describe())

制品规则：

• 对于“此制品中有什么？”之类的问题，读取清单和有限样本行；不要为回答结构性问题而下载数 GB 的制品。
• 使用 run.logged_artifacts() 查找某次运行的输出（例如检查点位置），并使用 run.used_artifacts() 查找其输入。

系统指标（GPU / CPU / 内存）——必须使用 stream='system'

GPU、CPU、内存、网络和磁盘指标位于独立的系统流中。 不带 stream='system' 调用 run.history() 只会返回训练指标——所有 system.gpu.*、system.cpu.*、system.memory.* 键都不会出现。在默认流中 找不到系统键，并不能证明它们不存在。

在得出 GPU 或系统指标不可用的结论之前，必须调用 run.history(stream='system')。

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

runs = api.runs(path, filters={"state": "finished"}, per_page=100)
rows = []
for run in runs:
    sys_df = run.history(stream="system", samples=500)
    if sys_df.empty or "system.gpu.0.gpu" not in sys_df.columns:
        rows.append({"run": run.name, "gpu_mean": None, "gpu_min": None, "gpu_max": None})
        continue
    gpu = sys_df["system.gpu.0.gpu"].dropna()
    rows.append({
        "run": run.name,
        "gpu_mean": round(gpu.mean(), 1),
        "gpu_min": round(gpu.min(), 1),
        "gpu_max": round(gpu.max(), 1),
        "low_util_pct": round(100 * (gpu < 30).sum() / len(gpu), 1) if len(gpu) else None,
    })

df = pd.DataFrame(rows)
print(df.to_string(index=False))

运行查找规则：

• 对于面向用户的运行名称，优先使用 run.display_name 或 run.name；如果有用，也请单独列出 run.id。不要只把运行 ID 作为名称报告。
• 对于“最佳”“最高”“最低”“最新”和“最长”任务，使用一个脚本 打印获胜运行的名称、ID、指标值、状态、组、job_type 和标签。在最终回答中使用这些上下文。
• 对于基线与超参数优化的问题，group is None 且 job_type/标签为空 通常表示未分组的基线；超参数优化试验通常有命名组，和/或 job_type="hyperopt"。如果获胜运行未分组，而排名第二的运行是分组的超参数优化试验，请明确说明。
• 对于最终指标问题，先检查汇总指标；只有在汇总指标缺失或任务明确要求历史记录时，才使用 scan_history(keys=[...])。
• 对于配置/模型变体问题，尝试使用 api.runs(..., lazy=False) 和 GraphQL 配置读取。如果配置为空，请说明这一点，并使用运行名称、标签、组、job_type 或文件作为信息来源；不要编造配置值。
• 对于 YOLOv5 权重清单，在最终计数表中将 yolov5s.pt 等原始文件名规范化为 yolov5s 等规范变体名称；必要时添加原始/来源列。

运行分析 / 项目摘要规则：

• 对于项目摘要，运行一个脚本，打印观测到的运行数量、配置键/值频率、指标键系列、产物类型和 sweep 状态。 最终回答中只引用脚本打印出的确切运行 ID、指标值或配置值；否则仅概述观测到的高层模式。
• 对于特定项目的摘要，不要依赖记忆中的项目事实。运行相关的 W&B/Weave 查询，打印简洁的证据，并且只根据观测到的数据撰写最终回答。
• 对于异常值分析，根据用户的运行计算所需的指标/历史统计数据；只有证据支持时，才将观测到的最大异常值作为标题重点。

OpenAI + Weave 跟踪设置：

• 对于 OpenAI 跟踪设置问题，明确提及 OpenAI 自动跟踪： 调用 weave.init(...) 后，Weave 会自动跟踪受支持的 OpenAI 客户端调用；用户也可以使用 weave.integrations.openai.OpenAI。说明系统会记录 提示词、响应、Token 用量、延迟和错误；使用应用函数上的 @weave.op() 添加应用级调用树。

W&B Sweep 设置：

• 对于 sweep 设置问题，始终在代码中展示具体的生命周期： 定义包含 method、metric 和 parameters 的 sweep 配置；通过 sweep_id = wandb.sweep(sweep_config, project=...) 创建；通过 wandb.agent(sweep_id, function=train, count=...) 运行 Agent；并在训练函数中使用 wandb.init(config=...) 和 wandb.log(...) 记录指标。
• 明确讨论网格、随机和贝叶斯搜索：网格搜索适用于极小的离散空间，随机搜索适用于广泛且成本低的探索和对数尺度的学习率，贝叶斯搜索适用于指标稳定后的高成本优化。提及平行坐标图、参数重要性、按排序排列的运行表，以及在选定赢家前重新运行排名靠前的配置/随机种子。

诊断训练历史（曲线、尖峰、NaN、稳定性）

对于“训练是否稳定？”／“哪些运行发散了？”／“有没有损失尖峰？”等问题，请扫描某个指标在各次运行中的历史记录，并在本地计算稳定性统计数据。始终传入 keys=[...]；对于步数达到 10K 以上的运行，请使用 beta_scan_history，而不是 history。

import wandb, os, numpy as np, pandas as pd
api = wandb.Api(timeout=120)
path = f"{os.environ['WANDB_ENTITY']}/{os.environ['WANDB_PROJECT']}"
metric = "train/loss"  # discover the real key first (probe_project / inspect a run)

runs = api.runs(path, filters={"state": "finished"}, per_page=100)[:40]
rows = []
for run in runs:
    df = run.history(samples=300, keys=[metric])  # never omit keys on large runs
    series = df[metric].dropna() if metric in getattr(df, "columns", []) else pd.Series(dtype=float)
    arr = series.to_numpy(dtype=float)
    finite = arr[np.isfinite(arr)]
    diffs = np.abs(np.diff(finite)) if finite.size > 2 else np.array([])
    spike_threshold = 5 * (np.median(diffs) or 1.0)
    rows.append({
        "run": run.display_name or run.name,
        "id": run.id,
        "points": int(arr.size),
        "nan_or_inf": int((~np.isfinite(arr)).sum()),
        "min": round(float(finite.min()), 5) if finite.size else None,
        "final": round(float(finite[-1]), 5) if finite.size else None,
        "spikes": int((diffs > spike_threshold).sum()),
    })

out = pd.DataFrame(rows).sort_values("min", na_position="last")
print(out.to_string(index=False))

min 是历史记录中的最佳值（而非终点值）；较大的 final - min 差值、非零的 nan_or_inf 或大量 spikes 都表示运行不稳定或已发散。如需排查 GPU 利用率不足，请使用上面的系统流配方。

比较两次运行

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

比较不同队列／变体（按配置项或运行维度分组）

对于“哪个变体／优化器／组最好？”等问题，请按某个维度（配置键、run.group 或 run.job_type）对运行分组，并比较各组中的某个指标。请报告完整的阶梯分布，而不只是最佳和最差值。

import wandb, os, numpy as np, pandas as pd
from collections import defaultdict
api = wandb.Api(timeout=120)
path = f"{os.environ['WANDB_ENTITY']}/{os.environ['WANDB_PROJECT']}"
metric = "accuracy"   # discover the real key first
axis = "optimizer"    # a config key; or use run.group / run.job_type

runs = api.runs(path, filters={"state": "finished"}, per_page=200)[:200]
buckets = defaultdict(list)
for run in runs:
    key = run.config.get(axis, "<missing>")  # or: run.group / run.job_type
    value = run.summary_metrics.get(metric)
    if value is not None:
        buckets[str(key)].append(float(value))

rows = [
    {axis: key, "n": len(vals), "mean": round(np.mean(vals), 4),
     "min": round(np.min(vals), 4), "max": round(np.max(vals), 4)}
    for key, vals in buckets.items()
]
out = pd.DataFrame(rows).sort_values("mean", ascending=False)
print(out.to_string(index=False))

如果返回的配置为空，说明这些运行是延迟获取的——请使用 api.runs(..., per_page=200) 重新获取，并访问每次运行的配置；或者改用 run.group／run.job_type／标签作为分组维度。不要臆造分组值。

汇总最近一次评估

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 报告

使用 wandb-workspaces 以编程方式定义报告。有关运行集筛选器、面板、加载和分享，请参阅 references/REPORTS.md。

import os

import wandb_workspaces.reports.v2 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("Project Analysis"),
        wr.P("Auto-generated summary from W&B API."),
        plots,
    ],
)
report.save(draft=True)
print(f"Report saved: {report.url}")

report.save() 正常返回并不能证明报告已成功保存——保存可能会静默失败。对于临时草稿以外的报告，请通过 report_helpers 保存，以便读取并验证结果，而不是假定保存成功：

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

result = save_report_verified(report)  # draft=True by default
print(result["answer"])                # answer=... verified=True/False url=...

启动

使用 skills/wandb-primary/scripts/launch_helpers.py。不要在本地训练来测试 GPU 工作，也不要用本地 wandb.init() 假装使用 Launch。

你创建的每个 Launch 入口点都必须调用 wandb.init(...)、至少记录一个指标，并结束运行。若 requirements.txt 中缺少 wandb，这些辅助工具会将其添加进去。

对于新的代码任务，如果用户未指定镜像，请使用 python:3.11-slim。辅助工具默认使用该镜像，并会在需要时将 wandb 添加到 requirements.txt。如果任务需要 CUDA、PyTorch 或其他特定框架的依赖，请在提交 Launch 任务前询问或选择合适的基础镜像。

从头开始设置 Launch 时，默认使用 Kubernetes。如果用户询问如何设置 Launch，且没有队列，请说明此 Agent 可以帮助设置 Kubernetes Launch 队列和 Agent。只有在用户提出要求或已有非 Kubernetes 环境时，才提及其他后端。

如果用户明确希望使用本地 GPU 或本地计算机，可以使用本地 Docker。在这种情况下，请提供 Local Docker Launch 队列，而不是 Kubernetes 队列，并告知用户需要 Docker、NVIDIA GPU 容器支持、W&B CLI 和 W&B 服务账号 API 密钥。告知他们将服务账号密钥存储在 WANDB_SERVICE_ACCOUNT_API_KEY 中；本地 Agent 仍会通过 WANDB_API_KEY 接收该密钥：

wandb launch-agent --queue QUEUE --entity ENTITY  # requires WANDB_API_KEY in the environment

如果缺少 Kubernetes 队列，除非用户指定了队列名称，否则请提出创建 wandb-launch-k8s。使用默认值 namespace="wandb-launch"、gpus=1、cpu=8、memory="80Gi"；只有在无法推断 W&B entity 时才询问。除非用户明确指定其他命名空间，否则使用 wandb-launch。队列默认值只能包含命名空间和资源请求；不要在其中放入 WANDB_API_KEY、服务账号 API 密钥或其他机密。

python skills/wandb-primary/scripts/launch_helpers.py create-queue ENTITY \
  --queue wandb-launch-k8s \
  --namespace wandb-launch \
  --gpus 1 --cpu 8 --memory 80Gi

创建队列后，请提出使用现有 W&B Launch Helm chart 引导配置 Kubernetes Launch Agent。不要自行运行 kubectl、helm 或工具检查；不要假定用户环境中安装了这些工具。提供一条可复制粘贴的命令，让用户在自己的集群环境中运行。

默认使用 Helm。告知用户在运行命令前，将 WANDB_SERVICE_ACCOUNT_API_KEY 设置为 W&B 服务账号 API 密钥，而不是个人用户密钥：

helm upgrade --install wandb-launch launch-agent \
  --repo https://charts.wandb.ai \
  --namespace wandb-launch \
  --create-namespace \
  --set agent.apiKey="$WANDB_SERVICE_ACCOUNT_API_KEY" \
  --set namespace=wandb-launch \
  --set "additionalTargetNamespaces={wandb-launch}" \
  --set-file launchConfig=<(cat <<YAML
entity: ${WANDB_ENTITY}
queues:
  - wandb-launch-k8s
max_jobs: 10
builder:
  type: noop
verbosity: 1
YAML
)

只有用户明确表示无法使用 Helm 时，才提供 kubectl 备用方案。

启动前检查队列：

python skills/wandb-primary/scripts/launch_helpers.py list-queues ENTITY

选择明确的 queue_name。辅助函数不会替你选择。如果用户没有指定队列，列出队列后就停下来，询问要使用哪个队列。提供简短的选项列表，包含队列名称、活跃 Agent 数量、近期项目状态和资源默认值。 agents=N 表示有一个 Launch Agent 正在轮询该队列；agents=0 表示已提交的项目会一直等待，直到 Agent 启动。items=state:count 显示队列项目近期的状态；CLAIMED 是正常的最终队列项目状态，并不能证明任务仍在运行。ns=... gpu=... cpu=... mem=... 是队列的默认资源。明确选择队列；除非传入 gpus=、cpu= 或 memory=，否则辅助函数会使用该队列的默认值。

默认启动行为：只等待 Launch 分配 W&B 运行 ID，然后告知用户运行 URL 和队列项目 ID。除非用户明确要求，或任务是需要在启动下一个运行前检查上一个运行的串行实验循环，否则不要等待完成。此类情况传入 wait_for="done"；其他情况保留默认值 wait_for="launched"。 Launch 辅助函数会返回一个状态字典，其中包含 queue_item_id、run_url、run_id、queue_state、run_state、launched、done 和 check_command。

使用配置覆盖项简单地重新启动：

python skills/wandb-primary/scripts/launch_helpers.py relaunch RUN_URL \
  --queue QUEUE \
  --config '{"epochs": 100}'

新建代码任务：

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

status = submit_code_artifact_job(
    code_files=["train.py"],
    entrypoint="python train.py",
    entity="ENTITY",
    project="PROJECT",
    queue_name="QUEUE",
    job_name="JOB_NAME",
)
print(status["run_url"], status["queue_item_id"])

修改现有任务：

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

info = download_code_artifact("ENTITY/PROJECT/JOB_NAME:latest")
# edit files in info["code_dir"]; if using apply_patch, use headerless V4A
# diffs without standard unified-diff file headers like --- or +++.
status = create_and_launch_modified_job(
    code_dir=info["code_dir"],
    entrypoint=info["entrypoint"],
    entity=info["entity"],
    project=info["project"],
    queue_name="QUEUE",
    job_name="JOB_NAME",
    base_image=info["base_image"],
)
print(status["run_url"], status["queue_item_id"])

除非用户要求指定覆盖项，否则使用队列默认的 GPU、CPU 和内存。如果需要覆盖项，直接将 gpus=、cpu= 和／或 memory= 传给同一个辅助函数。

如果用户要求的更改不是训练脚本已读取的配置字段，请修改代码，而不是传入新的配置键。

先用 check 调试 Launch 任务。它会打印队列、Agent、运行、运行日志和任务版本的 UI 链接，以及与 UI 侧边栏使用相同字段获取的队列项目问题。在 Python 中，使用 check_launch(...)。

python skills/wandb-primary/scripts/launch_helpers.py check ENTITY PROJECT QUEUE QUEUE_ITEM_ID

常用 UI URL 格式：

https://wandb.ai/ENTITY/launch/QUEUE_ID
https://wandb.ai/ENTITY/launch/QUEUE_ID/agents
https://wandb.ai/ENTITY/launch/QUEUE_ID/config
https://wandb.ai/ENTITY/launch/agents/AGENT_ID
https://wandb.ai/ENTITY/launch/agents/AGENT_ID/logs
https://wandb.ai/RUN_ENTITY/RUN_PROJECT/runs/RUN_ID
https://wandb.ai/RUN_ENTITY/RUN_PROJECT/runs/RUN_ID/logs
https://wandb.ai/JOB_ENTITY/JOB_PROJECT/jobs/JOB_COLLECTION_ID/version_details/ALIAS

没有稳定的队列项目详情 URL。打开队列运行页面，然后点击该队列项目的 Details；其 Issues 部分由队列项目的 error / warnings 字段提供数据。如果问题中包含 filePaths，这些文件位于 ENTITY/model-registry 中的 Launch Agent 运行记录里。

当工作负载已到达集群时，仍需要 K8s 资源：

kubectl logs -n wandb deploy/launch-agent-wandb-launch --since=1h | rg 'QUEUE_ITEM_ID|RUN_ID|JOB_NAME'
kubectl get jobs,pods -n NAMESPACE --sort-by=.metadata.creationTimestamp
kubectl describe pod -n NAMESPACE POD_NAME
kubectl logs -n NAMESPACE POD_NAME --tail=200

队列项目状态不是运行状态。PENDING 表示正在等待 Agent，LEASED 表示 Agent 已取出该项目，CLAIMED 表示 Agent 已确认该项目并分配了 associatedRunId，FAILED 表示 Launch 已将该项目标记为失败。成功启动的 Launch 运行，其正常的最终队列项目状态是 CLAIMED；应检查关联的 W&B 运行状态、运行日志、K8s 任务／Pod，以及任务／源代码工件，来判断执行是否成功。

如果在 Python 中进行轮询，不要等待队列状态变为 finished；队列项目没有这种状态。check_launch() 会返回 queue_state、launched、run_id、run_url、run_state 和 done。对于普通启动请求，launched 为 true 时就停止。只有当用户要求任务完成，或正在运行串行自主实验循环时，才在 done 时停止。

当 Pod 找不到代码或使用了错误的入口点时，使用工件：

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

inspect_job_artifact("ENTITY/PROJECT/JOB_NAME:latest")
info = download_code_artifact("ENTITY/PROJECT/JOB_NAME:latest")
print(info["code_dir"], info["files"], info["entrypoint"], info["base_image"])

前端查询映射：队列页面在 ENTITY/model-registry 上使用 QueueByID 获取队列、Agent 列表、队列项目、问题和资源配置。Agent 页面使用 FetchAgentDetails；Agent 日志使用 FetchRunOutputLog 和 RunLogLines，数据来自 ENTITY/model-registry 中的 Agent 运行记录。问题文件预览使用 Agent 运行记录上的 SingleFile。队列行使用 RunsInformation 获取关联运行的状态。

---

工作区

工作区视图是项目运行页面上已保存的分区／面板布局，与报告不同。在检查或编辑工作区（添加／重命名分区；添加面板，包括自定义表达式面板；设置运行集筛选条件／分组方式／排序；固定列；为运行着色或隐藏运行；保存视图）之前，请先阅读 references/WORKSPACES.md。请使用 wandb_workspaces.workspaces SDK——它有一些容易踩的坑：保存会立即生效，没有草稿状态；而 Workspace.from_url 会拒绝用户的默认 ?nw=nwuser... 工作区（这种情况需要使用参考文档中说明的原始规范 GraphQL 路径）。添加面板前，请验证指标／配置键。虚构的键会导致图表为空。

---

重要：大型项目性能规则

这些规则可避免在包含 10K+ 个运行或每个运行有 1K+ 个指标的项目上出现 502 错误、超时和长达数分钟的卡顿。违反其中任何一条都会导致大型项目出现故障。

1. 始终使用 wandb.Api(timeout=120)——默认的 19s 超时时间会导致频繁失败
2. 调用 history() 或 scan_history() 时，绝不能不显式指定 keys=[...]——获取所有列时，包含 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. 使用 scan_history(keys=[...]) 精确读取历史记录
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()
使用有界、可重复的 CLI 配方检查或比较运行	wandb_run_ops.py——参见 references/RUN_OPS.md
读取某个运行的控制台日志／根据日志诊断崩溃	Run.logLines GraphQL——参见 references/RUN_LOGS.md
大规模清点、筛选或检查 Artifacts	artifact_helpers.py／wandb_run_ops.py——参见 references/ARTIFACT_OPS.md
设计 Artifact 血缘关系或 Registry 晋级流程	references/ARTIFACTS_AND_REGISTRY.md
检查或更改 Runs 表状态	references/RUNS_TABLE.md，以及带防护的工作区辅助工具
提取评估结果进行分析	weave_helpers.eval_results_to_dicts()
在大型项目中探索评估且避免 OOM（先计数，再限制负载）	weave_helpers.safe_project_eval_summary()／safe_eval_child_summary()
不获取跟踪记录，仅统计数量	Weave 服务 API 中的 calls_query_stats
需要底层 Weave 筛选功能（CallsFilter、Query）	原始 Weave SDK——参见 references/WEAVE_SDK.md
读取 Weave Agent 对话或 Agent 跟踪记录	weave_agent_ops.py——参见 references/WEAVE_SDK.md
创建报告	wandb-workspaces（wandb_workspaces.reports.v2）——参见 references/REPORTS.md
检查或编辑工作区视图（分区、面板、运行集筛选器、固定列、运行颜色）	wandb_workspaces.workspaces——参见 references/WORKSPACES.md
创建 Artifact Collection Overview 草稿	collection_overview_helpers.py——参见 references/COLLECTION_OVERVIEWS.md
列出或创建 Automations（提醒、Slack／webhook／推送通知）	automation_helpers.py——参见 references/AUTOMATION_OPS.md
创建、检查或分析 Sweeps	sweeps_helpers.py／wandb_run_ops.py——参见 references/SWEEPS.md
设置 Launch 队列、重新启动运行或调试队列项目	launch_helpers.py——参见 references/LAUNCH.md
检查或创建 Weave Signal 监控器	weave_monitor_ops.py——参见 references/WEAVE_SDK.md
Agent Lens 对话、Insights 和标签	weave_agent_ops.py／Agent Lens API——参见 references/AGENT_LENS.md
设置生产环境监控	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 队列	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,                  # Recursively convert Weave types -> plain Python
    get_token_usage,         # Extract token counts from a call's summary
    eval_results_to_dicts,   # predict_and_score calls -> list of result dicts
    pivot_solve_rate,        # Build task-level pivot table across agents
    results_summary,         # Print compact eval summary
    eval_health,             # Extract status/counts from Evaluation.evaluate calls
    eval_efficiency,         # Compute tokens-per-success across eval calls
    safe_eval_root_summary,  # Compact aggregate evidence from one Evaluation.evaluate call
    safe_eval_child_summary, # Count predict_and_score rows first; sample full payloads (capped)
    safe_project_eval_summary,  # Project eval landscape without predict_and_score payload scans
)

# W&B helpers (training runs, metrics) — large-project optimized
from wandb_helpers import (
    get_api,             # Create API with safe timeout (default 120s)
    probe_project,       # Discover project scale, metrics, config, artifacts BEFORE querying
    fetch_runs,          # FAST: Direct GraphQL with selective metrics (17x faster)
    runs_to_dataframe,   # Legacy: iterate run objects (slower, use fetch_runs instead)
    diagnose_run,        # Quick diagnostic summary (configurable metric keys)
    compare_configs,     # Side-by-side config diff between two runs
    scan_history,        # Exact history scan with explicit metric keys
)

# Launch helpers (job submission, run reproduction, queue management)
from launch_helpers import (
    parse_run_url,                       # Extract (entity, project, run_id) from a W&B URL
    list_queues,                         # List launch queues with active agents and resource defaults
    get_job_artifact,                    # Check if a run has a job artifact
    inspect_job_artifact,                # Download + inspect a job artifact's metadata
    download_code_artifact,              # Download source code from a job artifact
    create_and_launch_modified_job,      # Upload modified code + launch in one call
    relaunch_run,                        # Re-run with config overrides (no code change)
    launch_job_artifact,                 # Launch directly from an artifact path
    submit_code_artifact_job,            # Create job artifact and enqueue in one call
    check_launch,                        # Check queue item, run URL, run state, issues
    create_queue,                        # Create a K8s launch queue
    inspect_queue,                       # Print queue details
)

# Report helpers (save/edit W&B Reports with read-back verification)
from report_helpers import (
    save_report_verified,   # save a report (draft by default) then re-read to confirm it landed
    edit_report_verified,   # load via from_url, mutate in place, save + verify
)

运维 CLI

这些脚本会输出有界证据，并明确说明样本／限制方面的注意事项：

python skills/wandb-primary/scripts/wandb_run_ops.py --help
python skills/wandb-primary/scripts/artifact_helpers.py --help
python skills/wandb-primary/scripts/workspace_ops.py --help
python skills/wandb-primary/scripts/collection_overview_helpers.py --help
python skills/wandb-primary/scripts/weave_agent_ops.py --help
python skills/wandb-primary/scripts/weave_monitor_ops.py --help
python skills/wandb-primary/scripts/automation_helpers.py --help
python skills/wandb-primary/scripts/launch_helpers.py --help

• wandb_run_ops.py 涵盖运行计数、排名、变体、配置／群组比较、历史记录／稳定性、Tables、Artifacts、检查点、项目快照和工作区清单。
• artifact_helpers.py 可列出和筛选 Artifact 版本，而不下载其内容。
• workspace_ops.py 通过乐观并发重试和回读验证，对原始规格执行带防护的工作区视图变更。
• collection_overview_helpers.py 获取有界集合证据，并根据 JSON 计划创建经过验证的 Overview 草稿。
• weave_agent_ops.py 读取独立的 Weave Agent 对话／跟踪记录平面。

参考文档

按需阅读这些文档——其中包含完整的 API 功能和操作指南：

• references/WANDB_CONCEPTS.md — W&B 数据模型、术语和概念辨析（entity/project/run 层级、config 与 log 和 summary 的区别、artifacts、registry）。阅读此文档，了解用户在询问什么。
• references/WANDB_SDK.md — 用于训练数据的 W&B SDK（runs、history、artifacts、sweeps、system metrics）。API 调用参考。
• references/RUN_OPS.md — 用于 run/project 分析的规范有界 CLI 操作指南，以及其输出和注意事项约定。
• references/RUNS_TABLE.md — Runs Table 的标识、筛选/分组/排序、可见性、固定项、基线、列、备注、移动、导出和删除。
• references/RUN_LOGS.md — 通过 logLines GraphQL connection 读取 run 控制台日志（分页或追踪最新日志）、多部分 output.log 布局与拼接，以及崩溃/恢复日志注意事项。
• references/ARTIFACT_OPS.md — 有界 Artifact 清单、版本筛选、存储核算和选择性文件检查。
• references/ARTIFACTS_AND_REGISTRY.md — 资产版本控制、lineage、Registry 组织、晋级和治理。
• references/COLLECTION_OVERVIEWS.md — 基于证据生成 Artifact Collection Overview 草稿。
• references/WEAVE_SDK.md — 用于 GenAI traces 的 Weave SDK（client.get_calls()、CallsFilter、Query、stats）。从这里开始查询 Weave。
• references/HYPOTHESIS_GENERATION.md — 四阶段协同假设生成方法。对于任何涉及实验分析、异常诊断、“出了什么问题？”或“接下来该尝试什么？”的任务，都请阅读此文档。
• references/REPORTS.md — W&B Report 的编写/编辑：runsets、结构化筛选器、面板、媒体、列、加载和共享链接。
• references/WORKSPACES.md — 以编程方式管理工作区视图：加载/创建、sections 和 panels、runset 筛选器/groupby/列/run 颜色、保存语义，以及默认用户工作区的原始规格路径。
• references/AUTOMATION_OPS.md — W&B Automations：范围、事件、筛选器、操作；通过 GraphQL 和 automation_helpers.py 列出/创建。
• references/SWEEPS.md — Sweep 配置、创建、agents、调度器和命名 sweep 聚合。
• references/LAUNCH.md — Launch 队列（本地 Docker 和 Kubernetes）、代码作业、重新启动和队列项诊断。
• references/AGENT_LENS.md — Agent Lens 数据平面、Forge SDK 检测、Insights/tags 语义和服务 API。

---

分析项目

从项目自身的领域中读取发现——涵盖 runs 和 Weave——并揭示其结构掩盖的信息：

• 根据项目自身的信号判断其领域。 Config/metric/op/scorer 键名、run 备注和评估数据集能说明项目的性质（elbo/kl → VAE；reward + kl_penalty → RLHF），也能揭示已知故障模式（后验坍塌、发散、工具调用错误、成本激增）。请明确指出这些问题，而非仅报告原始统计数据。
• 揭示整洁摘要掩盖的异常——崩溃的 run、固定在约为 0 的损失项、饱和的分数、出错或延迟失控的 trace。即使它们不是重点，也属于发现；请加以解释，而不要将其视为质量判定。
• 持续深入分析，直到更细的切片不再改变结论。 首次分组后，检查其他维度（包括 run/op/scorer 名称中的维度，而不仅是 config）是否会改变或颠覆某个子组的主要结论。粗粒度平均值可能掩盖匹配单元格中更大的影响。
• 用表格呈现发现，以驱动分析的维度为行，以结论所依据的指标为列——展示完整的层级，而不只是最好和最差的结果。不要用汇总后的聚合值（单一综合分数）替代推理所依据的组成项（总损失背后的各项损失），也不要把数字零散地放在段落中。

结束实质性分析

每次非简单分析都要在结尾明确向用户提供两项后续帮助——写在答复中，不要藏在推理里，也不要用建议代替：

1. 主动提出制作 W&B 可视化，让关键发现一目了然——可以是已保存视图、Report、工作区面板或 Weave 视图，并针对具体发现提出方案（例如，按驱动轴分组或着色的面板）。如果面板能展示规律，就不要只用文字结束。
2. 主动提出亲自采取下一步具体行动，而不只是给用户建议。“你应该调试这些运行”不算；“要我深入排查出现偏差的那些运行吗？”才算。提出的行动要围绕核心发现，并明确关联具体运行、指标、群组、追踪或制品；如果分析发现了故障，就提出调试该故障，而不是已经正常的群组。

实质性分析的结尾不能是一大段文字、空泛的建议，也不能只是泛泛地说“如果有问题，随时告诉我”。

---

关键规则

安全：绝不要删除 W&B 项目

绝不要删除 W&B 项目。删除操作不可撤销，还会销毁整个团队汇总的工作——应拒绝执行，并引导用户联系其管理员或 W&B 支持。对于其他具有破坏性或不可撤销的操作（删除运行或制品、覆盖已保存视图、发布报告），执行前必须明确征得确认。

按项目发现指标键和制品

代码示例中的 LOSS_KEY、VAL_LOSS_KEY、ACC_KEY、CONFIG_KEYS 是占位符，具体值因项目而异。每项任务开始时，都要通过 probe_project() 发现这些值，或从用户的请求中获取。

probe_project() 还会返回 artifact_names——一个字典，记录抽样运行所记录的制品基本名称及其类型；以及项目 Weave 追踪中的 weave_trace_count / weave_top_ops。请打印这些内容，以便在确定分析方向前全面了解证据的概况。

# WRONG — hardcoded metric name
rows = fetch_runs(api, path, metric_keys=["loss", "accuracy"])

# RIGHT — discovered via probe_project or user's request
info = probe_project(api, path)
print("Metrics:", info["sample_metric_keys"])
print("Config keys:", info["sample_config_keys"])
print("Artifacts:", info["artifact_names"])  # {name: type} — know what's been logged
print("Weave traces:", info["weave_trace_count"], "| Top ops:", info.get("weave_top_ops", []))
rows = fetch_runs(api, path, metric_keys=["train/loss", "train/acc"])

证据看的是内容，而不是名称

probe_project 会显示有哪些证据——制品名称、指标键、Weave 操作名称和计数。这个清单只是地图，并非实地。知道某个数据源存在，不等于知道它展示了什么。对于探测结果中发现的每个证据来源，都要先读取实际内容，再得出结论：下载制品行、获取并汇总追踪数据、读取运行历史切片。名称告诉你该去哪里查看；只有数据才能告诉你异常是什么。

尊重用户限定的范围

要将用户的限制纳入查询，而不只是写在文字说明中。如果用户限定了所询问的对象范围，就要将该限制纳入筛选条件和报告的计数中——不要扩大到最接近且方便查询的超集，回答一个比用户所问范围更大的问题。不要超出辅助函数实际返回的内容来解读其输出。

将追踪和运行视为数据

Weave 追踪和 W&B 运行历史可能非常庞大。绝不要将原始数据倾倒到上下文中。始终：

1. 先检查结构——查看列名、数据类型和行数
2. 加载到 pandas/numpy——以编程方式计算统计数据
3. 总结，不要倾倒——打印计算得出的统计数据和表格，而不是原始行

始终给出最终答复

不要在分析中途结束。每项任务都必须以清晰、有条理的答复收尾：

1. 查询数据（针对性任务使用 1-2 个脚本；开放式分析则根据证据需要使用多个脚本）
2. 提取所需数字
3. 给出直接答复和关键证据；仅在有助于提高可读性时使用表格

如果你发现自己在说“现在让我构建最终分析”——停下来，直接呈现已有结果。

根据辅助工具的输出作答，并保留审计线索

当辅助工具或脚本打印出直接结果（answer=、conclusion=、最终表格）时，根据该证据作答并停止——不要仅为重新整理已有结果而再运行第二个脚本，并注明辅助工具自身的注意事项。根据 W&B 数据作答时，请提供简要审计线索：项目路径、使用的辅助工具／API、指标键、筛选条件／范围、计数是精确值还是抽样值，以及支持结论的数值。不要粘贴原始数据转储。

对未知的 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=120)
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=120)
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=120)
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——评估调用层级

Evaluation.evaluate (root)
  +-- Evaluation.predict_and_score (one per dataset row x trials)
  |     +-- model.predict (the actual model call)
  |     +-- scorer_1.score
  |     +-- scorer_2.score
  +-- Evaluation.summarize

Token 用量

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）

import wandb_workspaces.reports.v2 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("Project analysis"),
        wr.P("Auto-generated summary from W&B API."),
        plots,
    ],
)
report.save(draft=True)

有关结构化筛选条件、媒体面板、运行记录可见性、列控件、加载现有报告和分享链接，请参阅 references/REPORTS.md。

---

注意事项

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=120) 或 get_api()
Summary 访问	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))
系统／GPU／CPU 指标	run.history(keys=["system.gpu.0.gpu"]) → 空	run.history(stream='system', samples=500) GPU、CPU、内存、网络和磁盘指标位于单独的数据流中；默认数据流仅返回训练指标。默认数据流中缺少这些指标，并不证明数据不存在。
History——大型运行记录未指定键	run.history(samples=10) -> 502	run.history(samples=10, keys=["LOSS_KEY"])
scan_history——未指定键	scan_history() -> 超时	scan_history(keys=["LOSS_KEY"])
跨运行记录搜索	在客户端遍历所有运行记录	服务端筛选：{"summary_metrics.X": {"$gt": Y}}
按运行类型筛选	filters={"job_type": "train"} → Unknown column 'job_type'	filters={"jobType": "train"}（后端字段使用驼峰命名）

Launch

易错点	错误写法	正确写法
列出队列	api.run_queues() 或原始 GQL	使用 helpers 中的 list_queues(entity)
资源	构造原始 Launch 资源参数	选择一个队列；helpers 会使用其默认值。仅在需要明确覆盖时使用 gpus=、cpu= 或 memory=
requirements.txt	从 venv 运行 pip freeze	让 helpers 添加 wandb；仅在需要额外软件包时传入 requirements=[...]
基础镜像架构	在 Mac 上运行 docker build	docker buildx build --platform linux/amd64
模拟 Launch	使用配置调用 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)