feat(m1): attribute vocab + sampling (T1.2)

- ATTRIBUTE_VOCAB (16 AIOps channels w/ units)
- FREQ_BANDS (6 sampling resolutions)
- sample_attributes(n_channels, seed) -> unique-name, reproducible
- tests/test_attributes.py (8 tests, RED->GREEN)

Exit criteria: fields complete, count correct, seed-reproducible.
This commit is contained in:
张宗平
2026-06-29 23:22:40 +08:00
parent 748bf41c53
commit 6d10454484
3 changed files with 140 additions and 1 deletions
+77
View File
@@ -0,0 +1,77 @@
"""Variable attribute vocabulary + attribute sampling (T1.2).
Defines a curated vocabulary of AIOps-relevant channel *names* with their
*units* and a set of *sampling frequency bands*. ``sample_attributes`` draws a
reproducible, unique-name subset to describe the channels of a multivariate
series — metadata that the instruction generator and the multimodal splicer
will later turn into text tokens like ``[属性: CPU usage (%), 1s]``.
"""
from __future__ import annotations
import random
from typing import Any
# ─── vocabulary ────────────────────────────────────────────────────────────
# Each entry: name (human label), unit. Frequencies are drawn separately from
# FREQ_BANDS so the same channel name can appear at different resolutions.
ATTRIBUTE_VOCAB: list[dict[str, str]] = [
{"name": "CPU usage", "unit": "%"},
{"name": "Memory usage", "unit": "%"},
{"name": "Disk I/O", "unit": "MB/s"},
{"name": "Network inbound traffic", "unit": "Mbps"},
{"name": "Network outbound traffic", "unit": "Mbps"},
{"name": "Request rate", "unit": "req/s"},
{"name": "Request latency p50", "unit": "ms"},
{"name": "Request latency p99", "unit": "ms"},
{"name": "Error rate", "unit": "%"},
{"name": "Temperature", "unit": "°C"},
{"name": "Queue length", "unit": "tasks"},
{"name": "Active connections", "unit": "conn"},
{"name": "Cache hit ratio", "unit": "%"},
{"name": "GPU utilization", "unit": "%"},
{"name": "Battery level", "unit": "%"},
{"name": "Pod restart count", "unit": "count"},
]
# Sampling frequency bands (resolution at which the series is recorded).
FREQ_BANDS: list[str] = ["1s", "5s", "10s", "30s", "1min", "5min"]
def sample_attributes(
n_channels: int, *, seed: int | None = None
) -> list[dict[str, Any]]:
"""Sample ``n_channels`` distinct channel attribute descriptors.
Each descriptor is a dict with keys ``name``, ``unit`` (drawn from
:data:`ATTRIBUTE_VOCAB`) and ``freq`` (drawn from :data:`FREQ_BANDS`).
Channel names are guaranteed unique within the returned list.
Parameters
----------
n_channels:
Number of channels to sample. Must be ``<= len(ATTRIBUTE_VOCAB)``.
seed:
RNG seed for reproducibility.
Raises
------
ValueError
If ``n_channels`` exceeds the available distinct names.
"""
if not isinstance(n_channels, int) or n_channels < 0:
raise ValueError(f"n_channels must be a non-negative int, got {n_channels!r}")
if n_channels > len(ATTRIBUTE_VOCAB):
raise ValueError(
f"requested {n_channels} channels but vocab only has "
f"{len(ATTRIBUTE_VOCAB)} distinct names"
)
rng = random.Random(seed)
vocab = list(ATTRIBUTE_VOCAB)
chosen = rng.sample(vocab, n_channels)
freqs = rng.choices(FREQ_BANDS, k=n_channels)
return [
{"name": c["name"], "unit": c["unit"], "freq": f}
for c, f in zip(chosen, freqs)
]