52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
"""PyFlowX 性能基准套件.
|
|
|
|
用法::
|
|
|
|
python -m benchmarks # 运行全部基准
|
|
python -m benchmarks graph # 仅图构建基准
|
|
python -m benchmarks execution # 仅执行基准
|
|
python -m benchmarks context # 仅上下文注入基准
|
|
python -m benchmarks storage # 仅状态后端基准
|
|
python -m benchmarks advanced # 仅高级基准
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
__all__ = ["print_results", "time_it"]
|
|
|
|
|
|
def time_it(fn: Callable[[], Any], iterations: int = 100, warmup: int = 5) -> tuple[float, float]:
|
|
"""计时工具：返回 (平均耗时 ms, 吞吐 ops/sec)."""
|
|
for _ in range(warmup):
|
|
fn()
|
|
times: list[float] = []
|
|
for _ in range(iterations):
|
|
t0 = time.perf_counter()
|
|
fn()
|
|
times.append(time.perf_counter() - t0)
|
|
avg = sum(times) / len(times)
|
|
return avg * 1000, 1.0 / avg if avg > 0 else float("inf")
|
|
|
|
|
|
def print_results(title: str, results: list[tuple[str, int, float, float]]) -> None:
|
|
"""打印格式化基准结果表."""
|
|
console = Console()
|
|
table = Table(title=title, show_header=True, header_style="bold")
|
|
table.add_column("场景", style="cyan", no_wrap=True)
|
|
table.add_column("迭代", justify="right")
|
|
table.add_column("平均耗时", justify="right", style="yellow")
|
|
table.add_column("吞吐", justify="right", style="green")
|
|
for name, iters, ms, ops in results:
|
|
ms_str = f"{ms:.3f} ms" if ms < 1 else f"{ms:.2f} ms"
|
|
ops_str = f"{ops:.0f} ops/s" if ops > 1000 else f"{ops:.1f} ops/s"
|
|
table.add_row(name, str(iters), ms_str, ops_str)
|
|
console.print(table)
|
|
console.print()
|