perf(compression): gate CJK-aware token estimation behind an ASCII fast path

The salvaged estimator ran a per-character Python loop on every
estimate_tokens_rough() call — a ~28,000,000x slowdown vs (len+3)//4 on a
1MB ASCII tool output (measured ~3.0s per call). Gate it:

- str.isascii() O(1) fast path keeps pure-ASCII text bit-identical to the
  classic (len+3)//4 rule at ~1.3x baseline cost (0.23us vs 0.17us per
  1MB call).
- Non-ASCII text counts dense CJK chars via a compiled character-class
  regex in C (len(text) - len(re.sub(''))): ~352ms/1MB hangul vs ~2.1s
  for the per-char loop.
- Non-ASCII-but-non-CJK text (accents, Cyrillic, emoji) keeps the classic
  rule.

Also: parity tests against the per-char reference implementation, and
updated two stale expectations that encoded the old behavior (CJK now
counted ~1 token/char; short string content now ceil-divided instead of
floored to 0). The continuity test now detects merged-into-tail summaries
via _is_context_summary_content.
This commit is contained in:
Teknium 2026-07-22 04:53:15 -07:00
parent 3f33a1c5aa
commit ea0fd393db
6 changed files with 87 additions and 15 deletions

View file

@ -2678,6 +2678,20 @@ def _is_cjk_token_dense_char(ch: str) -> bool:
)
# Same codepoint ranges as _is_cjk_token_dense_char, as a compiled character
# class so dense-char counting runs in C (``len(text) - len(re.sub(...))``)
# instead of a per-char Python loop. MUST stay in sync with
# _is_cjk_token_dense_char.
_CJK_DENSE_RE = re.compile(
"[\u1100-\u11ff" # Hangul Jamo
"\u2e80-\u9fff" # CJK radicals/ideographs
"\ua960-\ua97f" # Hangul Jamo Extended-A
"\uac00-\ud7af" # Hangul Syllables
"\uf900-\ufaff" # CJK compatibility ideographs
"\uff00-\uffef]" # Fullwidth forms / halfwidth kana
)
def estimate_tokens_rough(text: str) -> int:
"""Rough token estimate for pre-flight checks.
@ -2687,17 +2701,25 @@ def estimate_tokens_rough(text: str) -> int:
CJK/Hangul/Kana text is much denser than English under common LLM
tokenizers, so count those codepoints as roughly one token each instead
of applying the English-centric ~4 chars/token rule.
Perf: this runs on every message in every preflight/compaction walk,
including MB-scale tool outputs, so the common all-ASCII case must stay
O(1). ``str.isascii()`` is a flag check on CPython's compact unicode
representation (no scan), and the CJK counting itself is a single
C-level ``re.findall`` rather than a per-character Python loop.
"""
if not text:
return 0
text = str(text)
dense = 0
sparse = 0
for ch in text:
if _is_cjk_token_dense_char(ch):
dense += 1
else:
sparse += 1
if text.isascii():
# O(1) fast path — ASCII text cannot contain token-dense CJK chars.
return (len(text) + 3) // 4
dense = len(text) - len(_CJK_DENSE_RE.sub("", text))
if not dense:
# Non-ASCII but no CJK (accents, Cyrillic, emoji, ...): keep the
# classic ~4 chars/token rule.
return (len(text) + 3) // 4
sparse = len(text) - dense
return dense + ((sparse + 3) // 4)

View file

@ -0,0 +1 @@
plainOldCode

View file

@ -1,7 +1,11 @@
from unittest.mock import patch
from agent.context_compressor import ContextCompressor, _estimate_msg_budget_tokens
from agent.model_metadata import estimate_messages_tokens_rough, estimate_tokens_rough
from agent.model_metadata import (
_is_cjk_token_dense_char,
estimate_messages_tokens_rough,
estimate_tokens_rough,
)
def test_cjk_text_is_not_estimated_as_four_chars_per_token():
@ -48,3 +52,43 @@ def test_cjk_tail_does_not_expand_to_english_char_budget():
compress_end = compressor._find_tail_cut_by_tokens(messages, compress_start)
assert len(messages) - compress_end < 31
def _reference_per_char_estimate(text: str) -> int:
"""The pre-perf-gate per-character reference implementation."""
dense = 0
sparse = 0
for ch in text:
if _is_cjk_token_dense_char(ch):
dense += 1
else:
sparse += 1
return dense + ((sparse + 3) // 4)
def test_perf_gated_estimator_matches_per_char_reference():
samples = [
"",
"ab",
"a" * 400,
"" * 400,
"압축 테스트 " + ("" * 1000),
"café résumé naïve", # non-ASCII, no CJK
"hello 안녕 world",
"アイウエオ テスト", # halfwidth kana (fullwidth-forms block)
"漢字とかな交じり文です。",
"русский текст", # Cyrillic — non-ASCII, non-CJK
]
for text in samples:
assert estimate_tokens_rough(text) == _reference_per_char_estimate(text), repr(text)
def test_ascii_fast_path_keeps_classic_four_chars_per_token():
# Pure ASCII must be bit-identical to the historical (len+3)//4 rule.
for text in ("x", "xyz", "a" * 1000, "tool output\n" * 500):
assert estimate_tokens_rough(text) == (len(text) + 3) // 4
def test_non_ascii_non_cjk_keeps_classic_rule():
text = "café résumé " * 40
assert estimate_tokens_rough(text) == (len(text) + 3) // 4

View file

@ -70,8 +70,10 @@ class TestToolCallEnvelopeEstimate:
def test_non_dict_tool_calls_do_not_crash(self):
msg = {"role": "assistant", "content": "hi", "tool_calls": ["weird", None]}
# Non-dict entries are ignored (as before) without raising.
assert _estimate_msg_budget_tokens(msg) == len("hi") // _CHARS_PER_TOKEN + 10
# Non-dict entries are ignored (as before) without raising. String
# content now goes through estimate_tokens_rough (ceiling division),
# so 2 chars estimate as 1 token instead of flooring to 0.
assert _estimate_msg_budget_tokens(msg) == (len("hi") + 3) // _CHARS_PER_TOKEN + 10
@pytest.fixture()

View file

@ -99,13 +99,16 @@ def test_handoff_in_protected_head_is_replaced_not_duplicated():
with patch("agent.context_compressor.call_llm", return_value=_response("UPDATED summary body")):
compressed = compressor.compress(_messages_with_handoff(old_summary))
# The summary may be emitted standalone or merged into the first tail
# message (alternation corner case), so detect it the same way the
# compressor does rather than via a startswith(SUMMARY_PREFIX) check.
summary_messages = [
msg
for msg in compressed
if isinstance(msg, dict)
and str(msg.get("content") or "").startswith(SUMMARY_PREFIX)
and ContextCompressor._is_context_summary_content(msg.get("content"))
]
assert len(summary_messages) == 1
assert "UPDATED summary body" in summary_messages[0]["content"]
assert old_summary not in summary_messages[0]["content"]
assert "UPDATED summary body" in str(summary_messages[0]["content"])
assert old_summary not in str(summary_messages[0]["content"])
assert old_summary not in "\n".join(str(msg.get("content") or "") for msg in compressed)

View file

@ -58,9 +58,9 @@ class TestEstimateTokensRough:
assert long > short
def test_unicode_multibyte(self):
"""Unicode chars are still 1 Python char each — 4 chars/token holds."""
"""CJK chars are token-dense: counted ~1 token each, not 4 chars/token."""
text = "你好世界" # 4 CJK characters
assert estimate_tokens_rough(text) == 1
assert estimate_tokens_rough(text) == 4
class TestEstimateMessagesTokensRough: