使用智能体运行机器学习实验

Trackio 从底层设计之初就旨在支持由 LLM 智能体驱动的自主 ML 实验。本指南将介绍如何组织训练代码、配置告警以及搭建反馈循环，使智能体能够启动实验、监控实验、应对问题并迭代超参数——全程无需人工干预。

为什么选择 Trackio 用于智能体？

大多数实验跟踪器都假设有人在查看仪表盘。Trackio 通过多个通道为智能体提供了一等公民级别的编程访问权限：

支持 JSON 输出的 CLI — 每个 trackio 命令都支持 --json，因此可以轻松地用任何语言进行解析
Python API — trackio.Api() 提供对运行、指标和警报的直接访问
HTTP API / MCP — Trackio 仪表盘暴露了 Gradio API 端点，并且可以作为 MCP 服务器运行，以便 LLM 直接调用工具
使用 Webhook 的警报 — 智能体可以像打印语句一样插入 trackio.alert() 调用，并在之后轮询它们
since 过滤 — 高效轮询新告警，无需重新读取完整历史记录

Agent 反馈循环

典型的自主实验循环如下所示：

┌─────────────────────────────────────┐
│  1. Agent writes training script    │
│     with trackio.init(), log(),     │
│     and alert() calls               │
├─────────────────────────────────────┤
│  2. Agent launches the script       │
├─────────────────────────────────────┤
│  3. Agent polls for alerts and      │
│     metrics while training runs     │
├─────────────────────────────────────┤
│  4. Agent reads results, decides    │
│     next hyperparameters            │
├─────────────────────────────────────┤
│  5. Agent writes a new script       │
│     and repeats from step 1         │
└─────────────────────────────────────┘

本指南的其余部分将通过具体代码逐步讲解每个步骤。

步骤 1：为智能体结构化训练代码

关键洞察在于，智能体可以像插入 print() 语句进行调试一样，插入 trackio.alert() 调用。与 print 语句相比，警报的结构化程度更高：它们具有标题、可选的描述、严重级别，并且会被自动持久化和查询。

以下是一个由智能体编写的训练脚本，在每个决策点都附有提示：

import trackio
import torch

trackio.init(
    project="cifar10-hparam-search",
    name="run-lr0.001-bs64",
    config={
        "learning_rate": 0.001,
        "batch_size": 64,
        "optimizer": "adamw",
        "epochs": 50,
        "weight_decay": 1e-4,
    },
)

model = build_model()
optimizer = torch.optim.AdamW(
    model.parameters(), lr=0.001, weight_decay=1e-4
)

best_val_loss = float("inf")
patience_counter = 0
patience_limit = 5

for epoch in range(50):
    train_loss = train_one_epoch(model, train_loader, optimizer)
    val_loss, val_acc = evaluate(model, val_loader)

    trackio.log({
        "train_loss": train_loss,
        "val_loss": val_loss,
        "val_accuracy": val_acc,
        "learning_rate": optimizer.param_groups[0]["lr"],
    })

    # Alert on loss divergence
    if train_loss > 10.0:
        trackio.alert(
            title="Training diverged",
            text=f"train_loss={train_loss:.4f} at epoch {epoch}. "
                 "Learning rate is likely too high.",
            level=trackio.AlertLevel.ERROR,
        )
        break

    # Alert when validation stops improving
    if val_loss < best_val_loss:
        best_val_loss = val_loss
        patience_counter = 0
        torch.save(model.state_dict(), "best_model.pt")
    else:
        patience_counter += 1
        if patience_counter >= patience_limit:
            trackio.alert(
                title="Early stopping triggered",
                text=f"No improvement for {patience_limit} epochs. "
                     f"Best val_loss={best_val_loss:.4f} at epoch "
                     f"{epoch - patience_limit}.",
                level=trackio.AlertLevel.WARN,
            )
            break

    # Alert on overfitting
    if epoch > 10 and (train_loss < 0.1 and val_loss > 1.5):
        trackio.alert(
            title="Overfitting detected",
            text=f"train_loss={train_loss:.4f} vs val_loss={val_loss:.4f}. "
                 "Consider adding regularization or reducing model capacity.",
            level=trackio.AlertLevel.WARN,
        )

    # Alert on good results
    if val_acc > 0.93:
        trackio.alert(
            title="High accuracy reached",
            text=f"val_accuracy={val_acc:.4f} at epoch {epoch}.",
            level=trackio.AlertLevel.INFO,
        )

trackio.alert(
    title="Training complete",
    text=f"Final val_loss={val_loss:.4f}, val_accuracy={val_acc:.4f}. "
         f"Best val_loss={best_val_loss:.4f}.",
    level=trackio.AlertLevel.INFO,
)
trackio.finish()

这里的警报不仅面向人类——它们是结构化的信号，智能体稍后会读取这些信号来决定下一步操作。

面向智能体的告警指南

对于应停止运行的条件（发散、NaN 损失、内存溢出），请使用 ERROR。
对于表明应调整超参数的情况（过拟合、早停、收敛缓慢），请使用 WARN。
使用 INFO 记录里程碑（训练完成、达到高准确率、保存检查点）。
在 text 中包含数值，以便智能体能够解析并对其进行推理。
在 text 中包含可操作的建议（例如，“学习率可能过高”），以帮助智能体决定要更改什么。

步骤 2：启动脚本

Agent 可以在子进程或后台进程中启动该脚本。在大多数 Agent 框架中，只需运行一条 shell 命令即可：

python train.py

如果该智能体使用的是通过 MCP 连接的 LLM，它还可以通过 shell 工具启动脚本。

步骤 3：训练期间的监控

在训练运行期间，代理会轮询以获取警报和指标。--since 标志使这一过程更高效——仅返回自上次检查以来产生的新警报。

通过 CLI 轮询

# First poll — get all alerts
trackio get alerts --project "cifar10-hparam-search" --run "run-lr0.001-bs64" --json

# Subsequent polls — only new alerts
trackio get alerts --project "cifar10-hparam-search" --since "2025-06-15T14:30:00" --json

通过 Python API 进行轮询

import trackio
from datetime import datetime, timezone

api = trackio.Api()
last_check = None

while training_is_running():
    alerts = api.alerts(
        "cifar10-hparam-search",
        run="run-lr0.001-bs64",
        since=last_check,
    )
    if alerts:
        last_check = alerts[0]["timestamp"]  # alerts are sorted newest-first
        for a in alerts:
            print(f"[{a['level'].upper()}] {a['title']}: {a['text']}")

            if a["level"] == "error":
                # Stop the run and re-plan
                handle_error(a)

    time.sleep(30)

通过 MCP 进行轮询

如果 Trackio 仪表盘作为 MCP 服务器运行，智能体可以直接通过 MCP 协议调用 get_alerts 工具。有关设置说明，请参阅 Trackio 作为 API 和 MCP 服务器。

步骤 4：读取结果并决定下一步

运行结束后，智能体会读取完整的结果：

# Get run summary (metrics, config, step count)
trackio get run --project "cifar10-hparam-search" --run "run-lr0.001-bs64" --json

# Get the loss curve
trackio get metric --project "cifar10-hparam-search" --run "run-lr0.001-bs64" --metric "val_loss" --json

# Get all alerts (the most useful signal for decision-making)
trackio get alerts --project "cifar10-hparam-search" --run "run-lr0.001-bs64" --json

这些警报提供了训练期间发生情况的简明摘要。智能体无需解析数千个指标值，只需阅读几条警报即可了解：

运行是否发散？→ 降低学习率
是否过拟合？→ 添加正则化或降低模型容量
提前停止了？→ 超参数大致合适，但可能需要进行调优
它达到了高准确率吗？→ 此配置运行良好

步骤 5：迭代调整超参数

该智能体使用上一次运行产生的警报来选择新的超参数。以下是智能体可能遵循的决策逻辑的具体示例：

import json
import subprocess
import trackio

api = trackio.Api()

# Read alerts from the last run
alerts = api.alerts("cifar10-hparam-search", run="run-lr0.001-bs64")

# Read the config from the last run
runs = api.runs("cifar10-hparam-search")
last_run = [r for r in runs if r.name == "run-lr0.001-bs64"][0]
config = last_run.config

# Decision logic based on alerts
new_config = dict(config)
for alert in alerts:
    if "diverged" in alert["title"].lower():
        new_config["learning_rate"] = config["learning_rate"] * 0.1
    elif "overfitting" in alert["title"].lower():
        new_config["weight_decay"] = config.get("weight_decay", 1e-4) * 10
    elif "early stopping" in alert["title"].lower():
        new_config["learning_rate"] = config["learning_rate"] * 0.5
    elif "high accuracy" in alert["title"].lower():
        # This config is good — try fine-tuning
        new_config["learning_rate"] = config["learning_rate"] * 0.5
        new_config["epochs"] = config["epochs"] + 20

随后，该智能体使用更新后的配置生成新的训练脚本并启动它，从而重复此循环。

比较运行

经过多次迭代后，智能体可以比较项目中的所有运行：

# List all runs
trackio list runs --project "cifar10-hparam-search" --json

# Get a summary of each run
trackio get run --project "cifar10-hparam-search" --run "run-lr0.001-bs64" --json
trackio get run --project "cifar10-hparam-search" --run "run-lr0.0001-bs64" --json

或者在 Python 中：

api = trackio.Api()
runs = api.runs("cifar10-hparam-search")

for run in runs:
    alerts = run.alerts()
    errors = [a for a in alerts if a["level"] == "error"]
    warns = [a for a in alerts if a["level"] == "warn"]
    infos = [a for a in alerts if a["level"] == "info"]
    print(f"{run.name}: {len(errors)} errors, {len(warns)} warnings, {len(infos)} info")

通过 Webhooks 向 Agent 发送警报

如果你的智能体框架具有 HTTP 端点，或者能够接收来自 Slack/Discord 的消息，你可以使用 Webhook 将警报实时推送到智能体，而无需轮询：

trackio.init(
    project="cifar10-hparam-search",
    webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
)

这样，警报会发送到智能体监控的 Slack 频道，从而无需轮询即可实现闭环。有关 Slack 和 Discord 的详细设置说明，请参阅警报指南。

完整示例：自主超参数搜索

下面是一个完整的、自包含的示例，将上述所有内容串联起来。该脚本运行一个简单的超参数网格搜索，并基于自动告警做出决策：

import trackio
import random
import time

def run_experiment(config):
    """Simulate a training run. Replace with real training code."""
    run = trackio.init(
        project="auto-hparam-search",
        config=config,
    )

    lr = config["learning_rate"]
    best_val_loss = float("inf")
    patience = 0

    for epoch in range(config["epochs"]):
        # Simulated metrics — replace with real training
        noise = random.gauss(0, lr * 10)
        train_loss = max(0.01, 1.0 / (epoch + 1) + noise)
        val_loss = train_loss + random.uniform(0, 0.3)
        val_acc = max(0, min(1, 1.0 - val_loss + random.uniform(-0.05, 0.05)))

        trackio.log({
            "train_loss": train_loss,
            "val_loss": val_loss,
            "val_accuracy": val_acc,
        })

        if train_loss > 5.0:
            trackio.alert(
                title="Training diverged",
                text=f"train_loss={train_loss:.4f}, lr={lr}",
                level=trackio.AlertLevel.ERROR,
            )
            break

        if val_loss < best_val_loss:
            best_val_loss = val_loss
            patience = 0
        else:
            patience += 1
            if patience >= 5:
                trackio.alert(
                    title="Early stopping",
                    text=f"Best val_loss={best_val_loss:.4f}",
                    level=trackio.AlertLevel.WARN,
                )
                break

        time.sleep(0.05)

    trackio.alert(
        title="Run complete",
        text=f"Final val_loss={val_loss:.4f}, val_acc={val_acc:.4f}, "
             f"best_val_loss={best_val_loss:.4f}",
        level=trackio.AlertLevel.INFO,
    )
    trackio.finish()
    return run.name


def analyze_and_decide(api, project, run_name, config):
    """Read alerts from a run and suggest new hyperparameters."""
    alerts = api.alerts(project, run=run_name)

    new_config = dict(config)
    for alert in alerts:
        title = alert["title"].lower()
        if "diverged" in title:
            new_config["learning_rate"] *= 0.1
        elif "early stopping" in title:
            new_config["learning_rate"] *= 0.5
        elif "run complete" in title and "val_acc" in (alert.get("text") or ""):
            # Slight refinement
            new_config["learning_rate"] *= 0.8
            new_config["epochs"] += 10

    return new_config


api = trackio.Api()
project = "auto-hparam-search"

config = {
    "learning_rate": 0.01,
    "batch_size": 64,
    "epochs": 30,
}

for iteration in range(5):
    print(f"\n--- Iteration {iteration + 1} ---")
    print(f"Config: {config}")

    run_name = run_experiment(config)
    config = analyze_and_decide(api, project, run_name, config)

此示例是独立的，并使用模拟指标。将 run_experiment 函数体替换为您的实际训练代码，模式保持不变。

在 Transformers 和 TRL 中使用 Alerts

在 Hugging Face Trainer 中使用 report_to="trackio" 时，内置的 TrackioCallback 会自动处理 trackio.init()、trackio.log() 和 trackio.finish()。由于 trackio.init() 在训练开始前被调用，当前运行实例已设置完毕，因此同一进程中的任何其他回调都可以使用 trackio.alert()。

要添加警报，请编写一个小型的 TrainerCallback，并通过 callbacks= 将其传递给 Trainer：

Transformers

import trackio
from transformers import Trainer, TrainerCallback, TrainingArguments

class AlertCallback(TrainerCallback):
    def on_log(self, args, state, control, logs=None, **kwargs):
        if "trackio" not in args.report_to or logs is None:
            return
        loss = logs.get("loss")
        if loss is not None and loss > 5.0:
            trackio.alert(
                title="Training loss spike",
                text=f"loss={loss:.4f} at step {state.global_step}. "
                     "Consider lowering the learning rate.",
                level=trackio.AlertLevel.ERROR,
            )

    def on_evaluate(self, args, state, control, metrics=None, **kwargs):
        if "trackio" not in args.report_to or metrics is None:
            return
        eval_loss = metrics.get("eval_loss")
        if eval_loss is not None and eval_loss > 2.0:
            trackio.alert(
                title="High eval loss",
                text=f"eval_loss={eval_loss:.4f} at step {state.global_step}.",
                level=trackio.AlertLevel.WARN,
            )

trainer = Trainer(
    model=model,
    args=TrainingArguments(
        report_to="trackio",
        project="my-project",
        output_dir="./output",
        num_train_epochs=3,
        eval_strategy="steps",
        eval_steps=100,
    ),
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
    callbacks=[AlertCallback()],
)
trainer.train()

生成此代码的智能体只需定义具有正确条件的 AlertCallback 类，然后将其添加到 callbacks 列表中即可。无论使用何种模型或数据集，该模式均相同。

TRL

同样的方法也适用于 TRL 训练器，例如 GRPOTrainer 或 SFTTrainer。以下是一个强化学习的示例，其中包含针对奖励崩溃和 KL 散度的告警：

import trackio
from transformers import TrainerCallback

class RLAlertCallback(TrainerCallback):
    def on_log(self, args, state, control, logs=None, **kwargs):
        if "trackio" not in args.report_to or logs is None:
            return

        reward = logs.get("train/reward")
        if reward is not None and reward < -1.0:
            trackio.alert(
                title="Reward collapse",
                text=f"reward={reward:.4f} at step {state.global_step}. "
                     "Consider lowering the learning rate or checking the "
                     "reward model.",
                level=trackio.AlertLevel.ERROR,
            )

        kl = logs.get("train/kl")
        if kl is not None and kl > 10.0:
            trackio.alert(
                title="KL divergence too high",
                text=f"kl={kl:.4f} at step {state.global_step}. "
                     "The policy is drifting too far from the reference model.",
                level=trackio.AlertLevel.WARN,
            )

        completion_length = logs.get("train/completion_length")
        if completion_length is not None and completion_length < 5.0:
            trackio.alert(
                title="Completions too short",
                text=f"Mean completion length={completion_length:.1f} tokens. "
                     "The model may be collapsing to short outputs.",
                level=trackio.AlertLevel.WARN,
            )

然后将其传递给任意 TRL 训练器：

from trl import GRPOTrainer, GRPOConfig

trainer = GRPOTrainer(
    model=model,
    config=GRPOConfig(
        report_to="trackio",
        project="rl-experiment",
        output_dir="./output",
    ),
    train_dataset=dataset,
    callbacks=[RLAlertCallback()],
)
trainer.train()

编写告警回调的 Agent 技巧

保持条件简单。 每个 if 块应只针对一个阈值检查一个指标。这样便于智能体在多次运行之间添加、删除或调整条件。
在 text 中包含指标值和步数。 这能让智能体在决定下一组超参数时，有具体的数值可供推理。
提供可操作的建议。诸如“考虑降低学习率”之类的文本有助于智能体决定下一步要更改什么。
使用 on_evaluate 来获取评估指标。 评估指标仅在 on_evaluate 中可用，而在 on_log 中不可用。诸如 loss 之类的训练指标会出现在 on_log 中。

最佳实践总结

将警报用作结构化信号，而不仅仅是通知。在 text 中包含数值和可执行的建议，以便智能体能够解析并据此采取行动。
一致地使用严重级别。 ERROR = 停止并更改方法。 WARN = 调整参数。 INFO = 达到里程碑。
使用 since 进行轮询以提高效率。 保存你处理的最后一条警报的时间戳，并在下一次轮询时将其作为 since 传入，以避免重新读取旧警报。
将所有运行保留在同一项目中。 这可以让智能体并排比较运行结果，并查看尝试过的完整历史记录。
在每次运行时记录配置。始终将 config= 传递给 trackio.init()，以便 agent 能够读取回哪些超参数产生了哪些结果。
在代理脚本中使用 CLI 时，请加上 --json 参数。 JSON 输出便于可靠地解析，即使人类可读的格式发生变化也不会导致解析失败。
如需紧密集成，请考虑使用 MCP。 如果您的智能体框架支持 MCP，运行 trackio show --mcp-server 可以让智能体直接调用 Trackio 工具，而无需生成子进程。

Update on GitHub