Python Engineering for AI：从 0 亲手构建第一个真实项目

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def load_lines(path: "Path | str") -> list[str]:
    """按 UTF-8 读取文件的每一行（不含行尾换行符）。"""
    return Path(path).read_text(encoding="utf-8").splitlines()

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

LINE_RE = re.compile(
    r"^(?P<ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})\s+"
    r"(?P<level>DEBUG|INFO|WARNING|ERROR)\s+"
    r"(?P<msg>.*)$"
)
KV_RE = re.compile(r"(?P<key>[A-Za-z_][A-Za-z0-9_/]*)=(?P<value>[^\s]+)")


def parse_line(line: str) -> Optional[LogEvent]:
    line = line.rstrip("\n")
    if not line.strip():
        return None
    m = LINE_RE.match(line)
    if not m:
        return None
    fields = {kv.group("key"): kv.group("value") for kv in KV_RE.finditer(m.group("msg"))}
    return LogEvent(
        timestamp=datetime.strptime(m.group("ts"), TS_FORMAT),
        level=m.group("level"),
        message=m.group("msg"),
        fields=fields,
    )

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

@property
def loss(self) -> Optional[float]:
    raw = self.fields.get("loss")
    if raw is None:
        return None
    try:
        return float(raw)
    except ValueError:
        return None

@property
def step(self) -> Optional[int]:
    raw = self.fields.get("step")
    return int(raw) if raw is not None and raw.isdigit() else None

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

@property
def final_loss(self) -> Optional[float]:
    return self.losses[-1] if self.losses else None

@property
def best_loss(self) -> Optional[float]:
    return min(self.losses) if self.losses else None

@property
def avg_tokens_per_sec(self) -> Optional[float]:
    if not self.throughput_samples:
        return None
    return sum(self.throughput_samples) / len(self.throughput_samples)

@property
def has_nan(self) -> bool:
    return len(self.nan_events) > 0

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

@dataclass
class TrainRunStats:
    ...
    loss_steps: list[int] = field(default_factory=list)   # 新增：只含带 loss 的 step

@property
def best_step(self) -> Optional[int]:
    if not self.losses or not self.loss_steps:
        return None
    best_idx = self.losses.index(min(self.losses))
    return self.loss_steps[best_idx]

# analyze() 内：
if event.loss is not None:
    stats.losses.append(event.loss)
    if event.step is not None:
        stats.loss_steps.append(event.step)     # 对齐修复：两个列表同步追加

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def format_report(stats) -> str:
    lines = [
        "== 训练日志分析 ==",
        f"事件数            : {stats.parsed_events}（原始行 {stats.total_lines}）",
        f"step 范围         : {stats.steps[0] if stats.steps else '-'} → {stats.steps[-1] if stats.steps else '-'}",
        f"final loss        : {stats.final_loss:.4f}" if stats.final_loss is not None else "final loss        : -",
        f"best  loss        : {stats.best_loss:.4f} (step {stats.best_step})" if stats.best_loss is not None else "best  loss        : -",
        f"平均吞吐          : {stats.avg_tokens_per_sec:.0f} tokens/s" if stats.avg_tokens_per_sec else "平均吞吐          : -",
        f"warning 数        : {len(stats.warnings)}",
        f"error   数        : {len(stats.errors)}",
        f"NaN 事件          : {'是（训练可能已发散！）' if stats.has_nan else '否'}",
    ]
    return "\n".join(lines)

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(levelname)s %(name)s: %(message)s",
    )
    if not args.logfile.exists():
        logger.error("日志文件不存在：%s", args.logfile)
        return 2
    events = parse_log_file(args.logfile)
    if not events:
        logger.error("没有解析到任何合法日志行：%s", args.logfile)
        return 1
    stats = analyze(events, total_lines=len(load_lines(args.logfile)))
    ...
    return 0

参考实现用于对照，不是抄写目标；打开不会自动标记本步完成。

完整实现：projects/log-analyzer/（README + 14 个测试函数的独立实现）。 把你的 starter 实现和它对照，重点看三个地方：

1. parser.py 的正则与 None 语义；
2. stats.py 的 loss_steps 对齐写法；
3. cli.py 的退出码与 logging 分支。 如果你的实现通过了 41 个测试又和它在风格上不同——这是好事，说明你有自己的工程判断。 但请确认你能解释它每一个设计选择的理由。