跳转至

ztxexp.types

ztxexp.types

ztxexp 对外公开的运行时数据结构。

MetricEvent dataclass

单条指标事件。

属性：

名称 类型 描述
step int

指标对应的 step（epoch/global step）。
timestamp str

事件时间（ISO8601）。
metrics dict[str, float]

指标字典。
split str

数据划分（train/valid/test）。
phase str

阶段标识（fit/eval/infer）。
源代码位于： ztxexp/types.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@dataclass(slots=True)
class MetricEvent:
    """单条指标事件。

    Attributes:
        step: 指标对应的 step（epoch/global step）。
        timestamp: 事件时间（ISO8601）。
        metrics: 指标字典。
        split: 数据划分（train/valid/test）。
        phase: 阶段标识（fit/eval/infer）。
    """

    step: int
    timestamp: str
    metrics: dict[str, float]
    split: str = "train"
    phase: str = "fit"

    def to_dict(self) -> dict[str, Any]:
        """转换为字典。"""
        return asdict(self)

metrics instance-attribute 

metrics: dict[str, float]

phase class-attribute instance-attribute 

phase: str = 'fit'

split class-attribute instance-attribute 

split: str = 'train'

step instance-attribute 

step: int

timestamp instance-attribute 

timestamp: str

__init__ 

__init__(step: int, timestamp: str, metrics: dict[str, float], split: str = 'train', phase: str = 'fit') -> None

to_dict 

to_dict() -> dict[str, Any]

转换为字典。

源代码位于： ztxexp/types.py 
77
78
79
def to_dict(self) -> dict[str, Any]:
    """转换为字典。"""
    return asdict(self)

RunContext dataclass

单次实验运行上下文。

该对象由 ExpRunner 在每个 run 开始时构造，并传入用户实验函数。 exp_fn 的推荐契约如下：

  1. 函数签名：exp_fn(ctx: RunContext) -> dict | None；
  2. 返回 dict 时框架会写入 metrics.json；
  3. 返回 None 时不写 metrics.json，但 run 仍可成功；
  4. 业务产物统一写入 ctx.run_dir / "artifacts"；
  5. 过程指标用 ctx.log_metric(...) 写入 metrics.jsonl。

属性：

名称 类型 描述
run_id str

当前运行唯一 ID（同时也是 run 目录名）。
run_dir Path

当前运行目录绝对路径。
config dict[str, Any]

当前运行最终配置字典。
logger Logger

当前运行专属日志对象（输出到 run.log）。
meta RunMetadata

当前 run 元数据对象。

示例：

>>> def exp_fn(ctx: RunContext):
...     lr = ctx.config["lr"]
...     ctx.logger.info("lr=%s", lr)
...     return {"score": 1.0 - lr}
源代码位于： ztxexp/types.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
@dataclass(slots=True)
class RunContext:
    """单次实验运行上下文。

    该对象由 ``ExpRunner`` 在每个 run 开始时构造，并传入用户实验函数。
    ``exp_fn`` 的推荐契约如下：

    1. 函数签名：``exp_fn(ctx: RunContext) -> dict | None``；
    2. 返回 ``dict`` 时框架会写入 ``metrics.json``；
    3. 返回 ``None`` 时不写 ``metrics.json``，但 run 仍可成功；
    4. 业务产物统一写入 ``ctx.run_dir / "artifacts"``；
    5. 过程指标用 ``ctx.log_metric(...)`` 写入 ``metrics.jsonl``。

    Attributes:
        run_id: 当前运行唯一 ID（同时也是 run 目录名）。
        run_dir: 当前运行目录绝对路径。
        config: 当前运行最终配置字典。
        logger: 当前运行专属日志对象（输出到 run.log）。
        meta: 当前 run 元数据对象。

    Examples:
        >>> def exp_fn(ctx: RunContext):
        ...     lr = ctx.config["lr"]
        ...     ctx.logger.info("lr=%s", lr)
        ...     return {"score": 1.0 - lr}
    """

    run_id: str
    run_dir: Path
    config: dict[str, Any]
    logger: logging.Logger
    meta: RunMetadata = field(default_factory=RunMetadata)
    _metrics_jsonl_path: Path | None = field(default=None, repr=False)
    _trackers: list["Tracker"] = field(default_factory=list, repr=False)

    def log_metric(
        self,
        step: int,
        metrics: dict[str, float],
        split: str = "train",
        phase: str = "fit",
    ) -> None:
        """记录 step 级指标并通知 tracker。

        该方法用于写入过程曲线数据，不替代 ``exp_fn`` 的最终 ``return dict``。
        典型分工是：
        1. ``ctx.log_metric`` 负责每步/每轮中间指标；
        2. ``return dict`` 负责最终汇总指标。

        Args:
            step: 当前 step（例如 epoch 或 global step）。
            metrics: 指标字典，值应可转为 JSON（建议 ``float``）。
            split: 数据划分，如 ``train/valid/test``。
            phase: 阶段标识，如 ``fit/eval/infer``。

        Returns:
            None

        Examples:
            >>> ctx.log_metric(step=1, metrics={"loss": 0.91}, split="train", phase="fit")
        """
        event = MetricEvent(
            step=step,
            timestamp=datetime.now(timezone.utc).isoformat(),
            metrics=metrics,
            split=split,
            phase=phase,
        )
        payload = event.to_dict()

        if self._metrics_jsonl_path is not None:
            utils.append_jsonl(self._metrics_jsonl_path, payload)

        for tracker in self._trackers:
            tracker.on_metric(self, event)

config instance-attribute 

config: dict[str, Any]

logger instance-attribute 

logger: Logger

meta class-attribute instance-attribute 

meta: RunMetadata = field(default_factory=RunMetadata)

run_dir instance-attribute 

run_dir: Path

run_id instance-attribute 

run_id: str

__init__ 

__init__(run_id: str, run_dir: Path, config: dict[str, Any], logger: Logger, meta: RunMetadata = RunMetadata(), _metrics_jsonl_path: Path | None = None, _trackers: list['Tracker'] = list()) -> None

log_metric 

log_metric(step: int, metrics: dict[str, float], split: str = 'train', phase: str = 'fit') -> None

记录 step 级指标并通知 tracker。

该方法用于写入过程曲线数据，不替代 exp_fn 的最终 return dict。 典型分工是： 1. ctx.log_metric 负责每步/每轮中间指标； 2. return dict 负责最终汇总指标。

参数：

名称 类型 描述 默认
step int

当前 step（例如 epoch 或 global step）。

 必需
metrics dict[str, float]

指标字典，值应可转为 JSON（建议 float）。

 必需
split str

数据划分，如 train/valid/test。

 'train'
phase str

阶段标识，如 fit/eval/infer。

 'fit'

返回：

类型 描述
None

None

示例：

>>> ctx.log_metric(step=1, metrics={"loss": 0.91}, split="train", phase="fit")
源代码位于： ztxexp/types.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def log_metric(
    self,
    step: int,
    metrics: dict[str, float],
    split: str = "train",
    phase: str = "fit",
) -> None:
    """记录 step 级指标并通知 tracker。

    该方法用于写入过程曲线数据，不替代 ``exp_fn`` 的最终 ``return dict``。
    典型分工是：
    1. ``ctx.log_metric`` 负责每步/每轮中间指标；
    2. ``return dict`` 负责最终汇总指标。

    Args:
        step: 当前 step（例如 epoch 或 global step）。
        metrics: 指标字典，值应可转为 JSON（建议 ``float``）。
        split: 数据划分，如 ``train/valid/test``。
        phase: 阶段标识，如 ``fit/eval/infer``。

    Returns:
        None

    Examples:
        >>> ctx.log_metric(step=1, metrics={"loss": 0.91}, split="train", phase="fit")
    """
    event = MetricEvent(
        step=step,
        timestamp=datetime.now(timezone.utc).isoformat(),
        metrics=metrics,
        split=split,
        phase=phase,
    )
    payload = event.to_dict()

    if self._metrics_jsonl_path is not None:
        utils.append_jsonl(self._metrics_jsonl_path, payload)

    for tracker in self._trackers:
        tracker.on_metric(self, event)

RunMetadata dataclass

运行元数据。

用于描述一次 run 的治理与复现上下文。字段均为可选，框架会在运行时 自动填充可采集部分（如 python 版本、平台、命令行等）。

属性：

名称 类型 描述
experiment_name str | None

实验名称。
group str | None

实验分组。
tags dict[str, str] | list[str] | None

标签（可为字典或字符串列表）。
parent_run_id str | None

父 run ID（用于 lineage）。
attempt int | None

当前尝试次数（重试时递增）。
git_commit str | None

当前代码 commit。
python_version str | None

Python 版本。
platform str | None

运行平台描述。
hostname str | None

主机名。
started_cmd str | None

启动命令。
dataset_version str | None

数据版本标识。
seed int | None

随机种子。
extras dict[str, Any] | None

其它扩展元数据。
源代码位于： ztxexp/types.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@dataclass(slots=True)
class RunMetadata:
    """运行元数据。

    用于描述一次 run 的治理与复现上下文。字段均为可选，框架会在运行时
    自动填充可采集部分（如 python 版本、平台、命令行等）。

    Attributes:
        experiment_name: 实验名称。
        group: 实验分组。
        tags: 标签（可为字典或字符串列表）。
        parent_run_id: 父 run ID（用于 lineage）。
        attempt: 当前尝试次数（重试时递增）。
        git_commit: 当前代码 commit。
        python_version: Python 版本。
        platform: 运行平台描述。
        hostname: 主机名。
        started_cmd: 启动命令。
        dataset_version: 数据版本标识。
        seed: 随机种子。
        extras: 其它扩展元数据。
    """

    experiment_name: str | None = None
    group: str | None = None
    tags: dict[str, str] | list[str] | None = None
    parent_run_id: str | None = None
    attempt: int | None = None
    git_commit: str | None = None
    python_version: str | None = None
    platform: str | None = None
    hostname: str | None = None
    started_cmd: str | None = None
    dataset_version: str | None = None
    seed: int | None = None
    extras: dict[str, Any] | None = None

    def to_dict(self) -> dict[str, Any]:
        """转换为字典。"""
        return asdict(self)

attempt class-attribute instance-attribute 

attempt: int | None = None

dataset_version class-attribute instance-attribute 

dataset_version: str | None = None

experiment_name class-attribute instance-attribute 

experiment_name: str | None = None

extras class-attribute instance-attribute 

extras: dict[str, Any] | None = None

git_commit class-attribute instance-attribute 

git_commit: str | None = None

group class-attribute instance-attribute 

group: str | None = None

hostname class-attribute instance-attribute 

hostname: str | None = None

parent_run_id class-attribute instance-attribute 

parent_run_id: str | None = None

platform class-attribute instance-attribute 

platform: str | None = None

python_version class-attribute instance-attribute 

python_version: str | None = None

seed class-attribute instance-attribute 

seed: int | None = None

started_cmd class-attribute instance-attribute 

started_cmd: str | None = None

tags class-attribute instance-attribute 

tags: dict[str, str] | list[str] | None = None

__init__ 

__init__(experiment_name: str | None = None, group: str | None = None, tags: dict[str, str] | list[str] | None = None, parent_run_id: str | None = None, attempt: int | None = None, git_commit: str | None = None, python_version: str | None = None, platform: str | None = None, hostname: str | None = None, started_cmd: str | None = None, dataset_version: str | None = None, seed: int | None = None, extras: dict[str, Any] | None = None) -> None

to_dict 

to_dict() -> dict[str, Any]

转换为字典。

源代码位于： ztxexp/types.py 
54
55
56
def to_dict(self) -> dict[str, Any]:
    """转换为字典。"""
    return asdict(self)

RunSummary dataclass

一次批量执行的汇总结果。

属性：

名称 类型 描述
total int

本次执行计划中的配置总数。
succeeded int

成功运行数量。
failed int

失败运行数量。
skipped int

跳过运行数量。
duration_sec float

本次批量执行总耗时（秒）。
failed_run_ids list[str]

失败 run 的 ID 列表。

示例：

>>> summary = RunSummary(4, 3, 1, 0, 2.35, ["20260301_xxx"])
>>> summary.failed
1
源代码位于： ztxexp/types.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@dataclass(slots=True)
class RunSummary:
    """一次批量执行的汇总结果。

    Attributes:
        total: 本次执行计划中的配置总数。
        succeeded: 成功运行数量。
        failed: 失败运行数量。
        skipped: 跳过运行数量。
        duration_sec: 本次批量执行总耗时（秒）。
        failed_run_ids: 失败 run 的 ID 列表。

    Examples:
        >>> summary = RunSummary(4, 3, 1, 0, 2.35, ["20260301_xxx"])
        >>> summary.failed
        1
    """

    total: int
    succeeded: int
    failed: int
    skipped: int
    duration_sec: float
    failed_run_ids: list[str]

duration_sec instance-attribute 

duration_sec: float

failed instance-attribute 

failed: int

failed_run_ids instance-attribute 

failed_run_ids: list[str]

skipped instance-attribute 

skipped: int

succeeded instance-attribute 

succeeded: int

total instance-attribute 

total: int

__init__ 

__init__(total: int, succeeded: int, failed: int, skipped: int, duration_sec: float, failed_run_ids: list[str]) -> None