JOJO
bb420b0d04
- 取消追踪 552 个文件（运行态数据、密钥历史、实验/归档目录、 agentskills/easyagent/demo/android-webview-app 等不公开内容） - .gitignore 追加公开发布排除项与 cache/ 忽略 - 重写 README.md：mac 为主/开发中声明、web/host 双模式、Docker 镜像、 数据存储、settings.json/custom_models.json 配置说明 - .env.example 删除失效项（AGENT_API_* 等），标注推荐 settings.json - 新增 config/custom_models.json.example（Kimi-K3 蓝本 + DeepSeek 示例） - 修复 utils/perf_log.py 硬编码本机路径（改走 config.paths.LOGS_DIR）
55 lines
2.0 KiB
Python
55 lines
2.0 KiB
Python
"""Simple performance log helper for debugging latency issues."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
try:
|
|
from config.paths import LOGS_DIR as _LOGS_DIR
|
|
PERF_LOG_PATH = Path(_LOGS_DIR) / "perf_debug.log"
|
|
except Exception:
|
|
PERF_LOG_PATH = Path.home() / ".astrion" / "astrion" / "logs" / "perf_debug.log"
|
|
|
|
|
|
def perf_log(label: str, elapsed_ms: Optional[float] = None, extra: Optional[Dict[str, Any]] = None) -> None:
|
|
"""Append a timestamped performance entry to the perf log file."""
|
|
try:
|
|
PERF_LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
now = datetime.now().isoformat()
|
|
parts = [now, label]
|
|
if elapsed_ms is not None:
|
|
parts.append(f"elapsed_ms={elapsed_ms:.2f}")
|
|
if extra:
|
|
parts.append(str(extra))
|
|
line = " | ".join(parts) + "\n"
|
|
with open(PERF_LOG_PATH, "a", encoding="utf-8") as fh:
|
|
fh.write(line)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class PerfTimer:
|
|
"""Context manager / helper to measure and log elapsed time."""
|
|
|
|
def __init__(self, label: str, extra: Optional[Dict[str, Any]] = None):
|
|
self.label = label
|
|
self.extra = extra or {}
|
|
self.start: Optional[float] = None
|
|
|
|
def __enter__(self) -> "PerfTimer":
|
|
self.start = time.perf_counter()
|
|
perf_log(f"START {self.label}", extra=self.extra)
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
|
|
elapsed = (time.perf_counter() - self.start) * 1000 if self.start else 0.0
|
|
perf_log(f"END {self.label}", elapsed_ms=elapsed, extra=self.extra)
|
|
|
|
def mark(self, sub_label: str, extra: Optional[Dict[str, Any]] = None) -> None:
|
|
elapsed = (time.perf_counter() - self.start) * 1000 if self.start else 0.0
|
|
merged = {**self.extra, **(extra or {})}
|
|
perf_log(f"MARK {self.label}::{sub_label}", elapsed_ms=elapsed, extra=merged)
|