fix: better token counting and fixes cache

This commit is contained in:
Alex committed 2026-04-28 01:47:53 +01:00
1 parent a6a5db631b
commit 552bfe016a
9 files changed
+498 -44

No files matched your search

@@ -12,6 +12,12 @@ logger = logging.getLogger(__name__)
class TokenCounter:
"""Centralized token counting for conversations and messages."""
# Per-image token estimate. Provider tokenizers vary widely
# (Gemini ~258, GPT-4o 85-1500, Claude ~1500) and the actual cost
# depends on resolution/detail we can't see here. Errs slightly high
# so the threshold check stays conservative.
_IMAGE_PART_TOKEN_ESTIMATE = 1500
@staticmethod
def count_message_tokens(messages: List[Dict]) -> int:
"""
@@ -29,12 +35,36 @@ class TokenCounter:
if isinstance(content, str):
total_tokens += num_tokens_from_string(content)
elif isinstance(content, list):
# Handle structured content (tool calls, etc.)
# Handle structured content (tool calls, image parts, etc.)
for item in content:
if isinstance(item, dict):
total_tokens += num_tokens_from_string(str(item))
total_tokens += TokenCounter._count_content_part(item)
return total_tokens
@staticmethod
def _count_content_part(item: Dict) -> int:
# Image/file attachments are billed by the provider per image,
# not proportional to the inline bytes/base64 string.
# ``str(item)`` on a 1MB image inflates the count by ~10000x,
# which trips spurious compression and overflows downstream
# input limits.
item_type = item.get("type")
if "files" in item:
files = item.get("files")
count = len(files) if isinstance(files, list) and files else 1
return TokenCounter._IMAGE_PART_TOKEN_ESTIMATE * count
if "image_url" in item or item_type in {
"image",
"image_url",
"input_image",
"file",
}:
return TokenCounter._IMAGE_PART_TOKEN_ESTIMATE
return num_tokens_from_string(str(item))
@staticmethod
def count_query_tokens(
queries: List[Dict[str, Any]], include_tool_calls: bool = True
+10 -1
View File
@@ -1,3 +1,4 @@
import hashlib
import json
import logging
import time
@@ -10,6 +11,14 @@ from application.utils import get_hash
logger = logging.getLogger(__name__)
def _cache_default(value):
# Image attachments arrive inline as bytes (see GoogleLLM.prepare_messages_with_attachments);
# hash so the cache key stays bounded in size and stable across identical content.
if isinstance(value, (bytes, bytearray, memoryview)):
return f"<bytes:sha256:{hashlib.sha256(bytes(value)).hexdigest()}>"
return repr(value)
_redis_instance = None
_redis_creation_failed = False
_instance_lock = Lock()
@@ -36,7 +45,7 @@ def get_redis_instance():
def gen_cache_key(messages, model="docgpt", tools=None):
if not all(isinstance(msg, dict) for msg in messages):
raise ValueError("All messages must be dictionaries.")
messages_str = json.dumps(messages)
messages_str = json.dumps(messages, default=_cache_default)
tools_str = json.dumps(str(tools)) if tools else ""
combined = f"{model}_{messages_str}_{tools_str}"
cache_key = get_hash(combined)
+62 -16
View File
@@ -166,7 +166,7 @@ class BaseLLM(ABC):
if is_stream:
return self._stream_with_fallback(
decorated_method, method_name, *args, **kwargs
decorated_method, method_name, decorators, *args, **kwargs
)
try:
@@ -187,14 +187,27 @@ class BaseLLM(ABC):
f"{fallback.model_id}. Error: {str(e)}"
)
fallback_method = getattr(
fallback, method_name.replace("_raw_", "")
)
# Apply decorators to fallback's raw method directly — calling
# fallback.gen() would re-enter the orchestrator and recurse via
# fallback.fallback_llm.
fallback_method = getattr(fallback, method_name)
for decorator in decorators:
fallback_method = decorator(fallback_method)
fallback_kwargs = {**kwargs, "model": fallback.model_id}
return fallback_method(*args, **fallback_kwargs)
try:
return fallback_method(fallback, *args, **fallback_kwargs)
except Exception as e2:
if self._is_non_retriable_client_error(e2):
logger.error(
f"Fallback LLM failed with non-retriable client "
f"error; giving up: {str(e2)}"
)
else:
logger.error(f"Fallback LLM also failed; giving up: {str(e2)}")
raise
def _stream_with_fallback(
self, decorated_method, method_name, *args, **kwargs
self, decorated_method, method_name, decorators, *args, **kwargs
):
"""
Wrapper generator that catches mid-stream errors and falls back.
@@ -223,11 +236,37 @@ class BaseLLM(ABC):
f"Primary LLM failed mid-stream. Falling back to "
f"{fallback.model_id}. Error: {str(e)}"
)
fallback_method = getattr(
fallback, method_name.replace("_raw_", "")
# Apply decorators to fallback's raw stream method directly —
# calling fallback.gen_stream() would re-enter the orchestrator
# and recurse via fallback.fallback_llm. Emit the stream-start
# event manually so dashboards still see the fallback's
# provider/model when the response actually comes from it.
fallback._emit_stream_start_log(
fallback.model_id,
kwargs.get("messages"),
kwargs.get("tools"),
bool(
kwargs.get("_usage_attachments")
or kwargs.get("attachments")
),
)
fallback_method = getattr(fallback, method_name)
for decorator in decorators:
fallback_method = decorator(fallback_method)
fallback_kwargs = {**kwargs, "model": fallback.model_id}
yield from fallback_method(*args, **fallback_kwargs)
try:
yield from fallback_method(fallback, *args, **fallback_kwargs)
except Exception as e2:
if self._is_non_retriable_client_error(e2):
logger.error(
f"Fallback LLM failed mid-stream with non-retriable "
f"client error; giving up: {str(e2)}"
)
else:
logger.error(
f"Fallback LLM also failed mid-stream; giving up: {str(e2)}"
)
raise
def gen(self, model, messages, stream=False, tools=None, *args, **kwargs):
decorators = [gen_token_usage, gen_cache]
@@ -242,22 +281,29 @@ class BaseLLM(ABC):
**kwargs,
)
def gen_stream(self, model, messages, stream=True, tools=None, *args, **kwargs):
# Attachments arrive as ``_usage_attachments`` from ``Agent._llm_gen``;
# the ``stream_token_usage`` decorator pops that key, but the log
# fires before the decorator runs so it's still in ``kwargs`` here.
def _emit_stream_start_log(self, model, messages, tools, has_attachments):
# Stamped with ``self.provider_name`` so dashboards can group calls
# by vendor; the fallback path emits its own copy on the fallback
# instance so the actual responding provider is recorded.
logging.info(
"llm_stream_start",
extra={
"model": model,
"provider": self.provider_name,
"message_count": len(messages) if messages is not None else 0,
"has_attachments": bool(
kwargs.get("_usage_attachments") or kwargs.get("attachments")
),
"has_attachments": bool(has_attachments),
"has_tools": bool(tools),
},
)
def gen_stream(self, model, messages, stream=True, tools=None, *args, **kwargs):
# Attachments arrive as ``_usage_attachments`` from ``Agent._llm_gen``;
# the ``stream_token_usage`` decorator pops that key, but the log
# fires before the decorator runs so it's still in ``kwargs`` here.
has_attachments = bool(
kwargs.get("_usage_attachments") or kwargs.get("attachments")
)
self._emit_stream_start_log(model, messages, tools, has_attachments)
decorators = [stream_cache, stream_token_usage]
return self._execute_with_fallback(
"_raw_gen_stream",
+20 -1
View File
@@ -280,7 +280,26 @@ class LLMHandler(ABC):
# Keep serialized function calls/responses so the compressor sees actions
parts_text.append(str(item))
elif "files" in item:
parts_text.append(str(item))
# Image attachments arrive with raw bytes / base64
# inline (see GoogleLLM.prepare_messages_with_attachments).
# ``str(item)`` would dump the whole byte/base64
# blob into the compression prompt and bust the
# compression LLM's input limit.
files = item.get("files") or []
descriptors = []
if isinstance(files, list):
for f in files:
if isinstance(f, dict):
descriptors.append(
f.get("mime_type") or "file"
)
elif isinstance(f, str):
descriptors.append(f)
if not descriptors:
descriptors = ["file"]
parts_text.append(
f"[attachment: {', '.join(descriptors)}]"
)
return "\n".join(parts_text)
return ""
+32 -1
View File
@@ -360,7 +360,38 @@ class TestExtractTextFromContent:
handler = ConcreteHandler()
content = [{"files": ["/tmp/a.txt"]}]
result = handler._extract_text_from_content(content)
assert "files" in result
assert result == "[attachment: /tmp/a.txt]"
def test_list_with_inline_image_bytes(self):
# Google attaches images as inline bytes; stringifying them into
# the compression prompt would bust the compression LLM's input
# limit. The placeholder must describe the attachment without
# embedding the bytes.
handler = ConcreteHandler()
content = [
{
"files": [
{"file_bytes": b"\x89PNG" + b"\x00" * 1000, "mime_type": "image/png"}
]
}
]
result = handler._extract_text_from_content(content)
assert result == "[attachment: image/png]"
assert "PNG" not in result
assert "\\x" not in result
def test_list_with_multiple_files(self):
handler = ConcreteHandler()
content = [
{
"files": [
{"file_bytes": b"a", "mime_type": "image/png"},
{"file_uri": "https://x", "mime_type": "image/jpeg"},
]
}
]
result = handler._extract_text_from_content(content)
assert result == "[attachment: image/png, image/jpeg]"
def test_list_with_none_text(self):
handler = ConcreteHandler()
+6 -11
View File
@@ -49,6 +49,9 @@ class FailingLLM(BaseLLM):
class FallbackLLM(BaseLLM):
# _execute_with_fallback applies decorators to the fallback's raw method
# directly and never calls .gen() / .gen_stream() on it, so
# tracking lives on the raw methods.
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.gen_called = False
@@ -62,14 +65,6 @@ class FallbackLLM(BaseLLM):
self.gen_stream_called = True
yield "fallback_chunk"
def gen(self, *args, **kwargs):
self.gen_called = True
return "fallback_gen_result"
def gen_stream(self, *args, **kwargs):
self.gen_stream_called = True
yield "fallback_stream_chunk"
# ---------------------------------------------------------------------------
# gen / gen_stream decorator application
@@ -230,7 +225,7 @@ class TestExecuteWithFallbackNonStreaming:
llm._fallback_llm = fallback
result = llm.gen(model="m", messages=[])
assert result == "fallback_gen_result"
assert result == "fallback_result"
assert fallback.gen_called
@@ -257,7 +252,7 @@ class TestStreamWithFallback:
llm._fallback_llm = fallback
result = list(llm.gen_stream(model="m", messages=[]))
assert "fallback_stream_chunk" in result
assert "fallback_chunk" in result
assert fallback.gen_stream_called
@@ -344,7 +339,7 @@ class TestNonRetriableClientError:
llm._fallback_llm = fallback
result = llm.gen(model="m", messages=[])
assert result == "fallback_gen_result"
assert result == "fallback_result"
assert fallback.gen_called
+142 -12
View File
@@ -31,12 +31,20 @@ class FakeLLM(BaseLLM):
self.gen_stream_called = False
self.last_model_received = None # tracks the model kwarg passed to gen/gen_stream
# Track at the raw-method level. _execute_with_fallback applies
# decorators to the fallback's raw method directly and
# never calls .gen() / .gen_stream() on it, so a public-method
# override would not register fallback hops.
def _raw_gen(self, baseself, model, messages, stream, tools=None, **kwargs):
self.gen_called = True
self.last_model_received = model
if self.fail_at is not None:
raise RuntimeError("primary model unavailable")
return self.responses[0]
def _raw_gen_stream(self, baseself, model, messages, stream, tools=None, **kwargs):
self.gen_stream_called = True
self.last_model_received = model
yielded = 0
for chunk in self.stream_chunks:
if self.fail_at is not None and yielded >= self.fail_at:
@@ -44,18 +52,6 @@ class FakeLLM(BaseLLM):
yield chunk
yielded += 1
# Wrap gen/gen_stream so we can track whether the fallback instance was used
# and which model kwarg it received
def gen(self, *args, **kwargs):
self.gen_called = True
self.last_model_received = kwargs.get("model")
return super().gen(*args, **kwargs)
def gen_stream(self, *args, **kwargs):
self.gen_stream_called = True
self.last_model_received = kwargs.get("model")
return super().gen_stream(*args, **kwargs)
# Helpers
@@ -294,6 +290,140 @@ class TestStreamingFallback:
with pytest.raises(RuntimeError, match="mid-stream failure"):
list(primary.gen_stream(**CALL_ARGS))
def test_fallback_emits_stream_start_with_fallback_provider(
self, patch_model_utils, caplog
):
# The fallback raw-stream path bypasses ``gen_stream``, so it must
# emit its own ``llm_stream_start`` event tagged with the fallback
# vendor — otherwise dashboards record only the failed primary
# even when the response came from the backup.
import logging as _logging
class FallbackProvider(FakeLLM):
provider_name = "fallback-vendor"
backup = FallbackProvider(
stream_chunks=["b1"], model_id="backup-model-id"
)
patch_model_utils(
get_provider=lambda m, **_kwargs: "openai",
get_api_key=lambda p: "k",
create_llm=lambda type, **kw: backup,
)
class PrimaryProvider(FakeLLM):
provider_name = "primary-vendor"
primary = PrimaryProvider(
stream_chunks=["x"],
fail_at=0,
backup_models=["backup-model-id"],
)
with caplog.at_level(_logging.INFO, logger="root"):
list(
primary.gen_stream(
model="primary-model",
messages=[{"role": "user", "content": "hi"}],
)
)
starts = [r for r in caplog.records if r.message == "llm_stream_start"]
assert len(starts) == 2
assert starts[0].provider == "primary-vendor"
assert starts[0].model == "primary-model"
assert starts[1].provider == "fallback-vendor"
assert starts[1].model == "backup-model-id"
# Tests — fallback never re-enters the orchestrator (Option B regression)
@pytest.mark.integration
class TestFallbackNoRecursion:
"""When the primary fails, _execute_with_fallback applies decorators to
the fallback's raw method directly. The fallback's own ``fallback_llm``
property must never be accessed — otherwise a fallback failure would
re-enter the orchestrator and walk the global FALLBACK_LLM_* chain
unboundedly."""
def test_backup_fallback_llm_property_never_accessed_on_gen_failure(
self, monkeypatch, patch_model_utils
):
backup = FakeLLM(fail_at=0) # backup also fails
accessed_on = []
original_property = BaseLLM.fallback_llm
def tracked_fallback_llm(self_llm):
accessed_on.append(self_llm)
return original_property.fget(self_llm)
monkeypatch.setattr(
BaseLLM, "fallback_llm", property(tracked_fallback_llm)
)
patch_model_utils(
get_provider=lambda m, **_kwargs: "openai",
get_api_key=lambda p: "k",
create_llm=lambda type, **kw: backup,
)
primary = FakeLLM(fail_at=0, backup_models=["backup-model"])
with pytest.raises(RuntimeError, match="primary model unavailable"):
primary.gen(**CALL_ARGS)
assert primary in accessed_on # primary lazy-loaded its fallback
assert backup not in accessed_on # backup's chain was never walked
def test_backup_fallback_llm_property_never_accessed_on_stream_failure(
self, monkeypatch, patch_model_utils
):
backup = FakeLLM(stream_chunks=["x"], fail_at=0)
accessed_on = []
original_property = BaseLLM.fallback_llm
def tracked_fallback_llm(self_llm):
accessed_on.append(self_llm)
return original_property.fget(self_llm)
monkeypatch.setattr(
BaseLLM, "fallback_llm", property(tracked_fallback_llm)
)
patch_model_utils(
get_provider=lambda m, **_kwargs: "openai",
get_api_key=lambda p: "k",
create_llm=lambda type, **kw: backup,
)
primary = FakeLLM(
stream_chunks=["y"], fail_at=0, backup_models=["backup-model"]
)
with pytest.raises(RuntimeError, match="mid-stream failure"):
list(primary.gen_stream(**CALL_ARGS))
assert primary in accessed_on
assert backup not in accessed_on
def test_fallback_failure_propagates_without_chain(self, patch_model_utils):
"""When both primary and fallback fail, the fallback's exception
propagates cleanly — no third hop, no extra retries."""
backup = FakeLLM(fail_at=0)
patch_model_utils(
get_provider=lambda m, **_kwargs: "openai",
get_api_key=lambda p: "k",
create_llm=lambda type, **kw: backup,
)
primary = FakeLLM(fail_at=0, backup_models=["backup-model"])
with pytest.raises(RuntimeError, match="primary model unavailable"):
primary.gen(**CALL_ARGS)
assert backup.gen_called # confirms fallback raw method WAS invoked
# Tests — backup model priority over global fallback
+91
View File
@@ -438,3 +438,94 @@ def test_stream_cache_key_generation_failure_yields(mock_make_redis):
messages = ["not_a_dict"]
result = list(mock_function(None, "model", messages, stream=True, tools=None))
assert result == ["fallback_chunk"]
# =====================================================================
# gen_cache_key with inline bytes (Google attachments)
# =====================================================================
@pytest.mark.unit
def test_gen_cache_key_handles_inline_bytes():
"""Image attachments arrive in messages as raw bytes (see
GoogleLLM.prepare_messages_with_attachments). gen_cache_key must not
crash on json.dumps of bytes."""
msgs = [
{
"role": "user",
"content": [{"file_bytes": b"\x00\x01\x02", "mime_type": "image/png"}],
}
]
key = gen_cache_key(msgs, model="x")
assert isinstance(key, str)
assert len(key) == 32
@pytest.mark.unit
def test_gen_cache_key_stable_for_same_bytes():
"""Two requests with identical image bytes must produce the same key
— otherwise we'd never get cache hits on image-bearing prompts."""
a = [
{
"role": "user",
"content": [{"file_bytes": b"abc", "mime_type": "image/png"}],
}
]
b = [
{
"role": "user",
"content": [{"file_bytes": b"abc", "mime_type": "image/png"}],
}
]
assert gen_cache_key(a, "m") == gen_cache_key(b, "m")
@pytest.mark.unit
def test_gen_cache_key_differs_for_different_bytes():
"""Different image bytes must produce different keys — otherwise two
different images would collide in cache."""
a = [
{
"role": "user",
"content": [{"file_bytes": b"abc", "mime_type": "image/png"}],
}
]
b = [
{
"role": "user",
"content": [{"file_bytes": b"xyz", "mime_type": "image/png"}],
}
]
assert gen_cache_key(a, "m") != gen_cache_key(b, "m")
@pytest.mark.unit
def test_gen_cache_key_handles_bytearray_and_memoryview():
"""The default helper covers all bytes-like types so refactors that
swap bytes for bytearray/memoryview don't silently re-introduce the
TypeError."""
msgs_ba = [
{
"role": "user",
"content": [
{"file_bytes": bytearray(b"abc"), "mime_type": "image/png"}
],
}
]
msgs_mv = [
{
"role": "user",
"content": [
{"file_bytes": memoryview(b"abc"), "mime_type": "image/png"}
],
}
]
msgs_b = [
{
"role": "user",
"content": [{"file_bytes": b"abc", "mime_type": "image/png"}],
}
]
# All three should hash the same content to the same key.
assert gen_cache_key(msgs_ba, "m") == gen_cache_key(msgs_b, "m")
assert gen_cache_key(msgs_mv, "m") == gen_cache_key(msgs_b, "m")
+103
View File
@@ -427,6 +427,109 @@ class TestCompressionService:
assert token_count_with_tools > token_count_without_tools
def test_count_message_tokens_skips_inline_image_bytes(self):
# Google attaches images as raw bytes inline. Stringifying the
# part would tokenize the byte repr (~2M tokens for a 1MB image),
# trigger spurious compression, and overflow downstream input
# limits. The estimate must stay bounded.
big_bytes = b"\x00" * 1_000_000
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "describe this"},
{
"files": [
{"file_bytes": big_bytes, "mime_type": "image/png"}
]
},
],
}
]
tokens = TokenCounter.count_message_tokens(messages)
assert tokens < 5000
def test_count_message_tokens_per_image_estimate_scales(self):
msgs_one = [
{
"role": "user",
"content": [
{
"files": [
{"file_bytes": b"x", "mime_type": "image/png"}
]
}
],
}
]
msgs_two = [
{
"role": "user",
"content": [
{
"files": [
{"file_bytes": b"x", "mime_type": "image/png"},
{"file_uri": "https://x", "mime_type": "image/jpeg"},
]
}
],
}
]
assert TokenCounter.count_message_tokens(
msgs_two
) > TokenCounter.count_message_tokens(msgs_one)
def test_count_message_tokens_skips_openai_image_url(self):
# OpenAI puts images inline as base64 data URLs. ``str(item)`` on
# the dict would tokenize the entire base64 payload.
big_b64 = "A" * 1_000_000
messages = [
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{big_b64}"
},
},
],
}
]
tokens = TokenCounter.count_message_tokens(messages)
assert tokens < 5000
def test_count_message_tokens_skips_anthropic_image(self):
# Anthropic puts images inline as ``source.data`` base64.
big_b64 = "A" * 1_000_000
messages = [
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": big_b64,
},
},
],
}
]
tokens = TokenCounter.count_message_tokens(messages)
assert tokens < 5000
def test_count_message_tokens_still_counts_text_parts(self):
# The image bypass must not regress regular text-part counting.
messages = [
{
"role": "user",
"content": [{"type": "text", "text": "hello world"}],
}
]
assert TokenCounter.count_message_tokens(messages) > 0
def test_format_conversation_for_compression(
self, prompt_builder, sample_conversation
):