- fail closed on MCP input and compile database policy - add shared scan catalogs, finding queries, and tiered gates
62 lines
2 KiB
Python
62 lines
2 KiB
Python
"""Model-agnostic token-efficiency accounting for benchmark artifacts.
|
|
|
|
The harness does not assume a tokenizer or provider. It reports a reproducible
|
|
upper-bound-like estimate of one token per four UTF-8 bytes so reports can be
|
|
compared across runs without adding a model-specific dependency.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
from collections.abc import Iterable, Mapping
|
|
from typing import Any
|
|
|
|
from .metrics import distribution_summary
|
|
|
|
TOKEN_ESTIMATOR = "ceil(utf8_bytes/4)"
|
|
|
|
|
|
def estimate_text_tokens(text: str) -> int:
|
|
"""Return a deterministic, model-agnostic estimate for UTF-8 text."""
|
|
|
|
return math.ceil(len(text.encode("utf-8")) / 4)
|
|
|
|
|
|
def _metric_summary(bytes_samples: list[int], findings: list[int]) -> dict[str, Any]:
|
|
tokens = [math.ceil(value / 4) for value in bytes_samples]
|
|
density = [
|
|
1000.0 * count / token
|
|
for count, token in zip(findings, tokens, strict=True)
|
|
if token > 0
|
|
]
|
|
return {
|
|
"bytes": distribution_summary(bytes_samples),
|
|
"estimated_tokens": distribution_summary(tokens),
|
|
"findings_per_1k_estimated_tokens": distribution_summary(density),
|
|
}
|
|
|
|
|
|
def _nonnegative_int(record: Mapping[str, Any], key: str) -> int:
|
|
value = record.get(key, 0)
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
return 0
|
|
try:
|
|
return max(0, int(value))
|
|
except (OverflowError, ValueError):
|
|
return 0
|
|
|
|
|
|
def artifact_token_efficiency(runs: Iterable[Mapping[str, Any]]) -> dict[str, Any]:
|
|
"""Summarize driver and SARIF artifact token costs for completed runs."""
|
|
|
|
records = list(runs)
|
|
findings = [_nonnegative_int(record, "finding_count") for record in records]
|
|
return {
|
|
"estimator": TOKEN_ESTIMATOR,
|
|
"driver_output": _metric_summary(
|
|
[_nonnegative_int(record, "output_bytes") for record in records], findings
|
|
),
|
|
"sarif_report": _metric_summary(
|
|
[_nonnegative_int(record, "report_bytes") for record in records], findings
|
|
),
|
|
}
|