From 7f869c87445a4915e62d4325b9817979b9213abc Mon Sep 17 00:00:00 2001 From: Hao Zhe Date: Thu, 15 Oct 2026 03:12:52 +0800 Subject: [PATCH] feat(telemetry): report LLM token usage coverage --- openviking/models/vlm/backends/litellm_vlm.py | 8 +- openviking/models/vlm/backends/openai_vlm.py | 14 ++- .../models/vlm/backends/volcengine_media.py | 1 + openviking/models/vlm/base.py | 17 +++ openviking/telemetry/operation.py | 33 +++++ tests/models/vlm/test_token_usage_coverage.py | 115 ++++++++++++++++++ 6 files changed, 179 insertions(+), 9 deletions(-) create mode 100644 tests/models/vlm/test_token_usage_coverage.py diff --git a/openviking/models/vlm/backends/litellm_vlm.py b/openviking/models/vlm/backends/litellm_vlm.py index 699ad343ff..634e66fece 100644 --- a/openviking/models/vlm/backends/litellm_vlm.py +++ b/openviking/models/vlm/backends/litellm_vlm.py @@ -560,9 +560,11 @@ def _update_token_usage_from_response( duration_seconds: float = 0.0, ) -> None: """Update token usage from response.""" - if hasattr(response, "usage") and response.usage: - prompt_tokens = response.usage.prompt_tokens - completion_tokens = response.usage.completion_tokens + usage = getattr(response, "usage", None) + self.record_token_usage_coverage(reported=usage is not None) + if usage: + prompt_tokens = usage.prompt_tokens + completion_tokens = usage.completion_tokens self.update_token_usage( model_name=self.model or "unknown", provider=self.provider, diff --git a/openviking/models/vlm/backends/openai_vlm.py b/openviking/models/vlm/backends/openai_vlm.py index 7f6023eadf..e3adf80299 100644 --- a/openviking/models/vlm/backends/openai_vlm.py +++ b/openviking/models/vlm/backends/openai_vlm.py @@ -205,12 +205,14 @@ def _update_token_usage_from_response( response, duration_seconds: float = 0.0, ): - if hasattr(response, "usage") and response.usage: - tracer.info(f"response.usage={response.usage}") - prompt_tokens = response.usage.prompt_tokens - completion_tokens = response.usage.completion_tokens - prompt_tokens_details = getattr(response.usage, "prompt_tokens_details", None) - completion_tokens_details = getattr(response.usage, "completion_tokens_details", None) + usage = getattr(response, "usage", None) + self.record_token_usage_coverage(reported=usage is not None) + if usage: + tracer.info(f"response.usage={usage}") + prompt_tokens = usage.prompt_tokens + completion_tokens = usage.completion_tokens + prompt_tokens_details = getattr(usage, "prompt_tokens_details", None) + completion_tokens_details = getattr(usage, "completion_tokens_details", None) prompt_cached_tokens = getattr(prompt_tokens_details, "cached_tokens", 0) or 0 completion_reasoning_tokens = ( getattr(completion_tokens_details, "reasoning_tokens", 0) or 0 diff --git a/openviking/models/vlm/backends/volcengine_media.py b/openviking/models/vlm/backends/volcengine_media.py index 3debdea48a..81e8a03d12 100644 --- a/openviking/models/vlm/backends/volcengine_media.py +++ b/openviking/models/vlm/backends/volcengine_media.py @@ -363,6 +363,7 @@ def _record_usage( ) -> None: try: usage = getattr(response, "usage", None) + vlm.record_token_usage_coverage(reported=usage is not None) input_details = getattr(usage, "input_tokens_details", None) output_details = getattr(usage, "output_tokens_details", None) vlm.update_token_usage( diff --git a/openviking/models/vlm/base.py b/openviking/models/vlm/base.py index 930b41ceea..a8bfe9e831 100644 --- a/openviking/models/vlm/base.py +++ b/openviking/models/vlm/base.py @@ -213,6 +213,23 @@ def is_available(self) -> bool: return self.api_key is not None or self.api_base is not None # Token usage tracking methods + def record_token_usage_coverage(self, *, reported: bool) -> None: + """Record token-usage coverage for one successful provider response.""" + try: + from openviking.telemetry import get_current_telemetry + + get_current_telemetry().record_llm_token_usage_report(reported=reported) + except Exception as e: + # Telemetry must never break model inference. + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + "vlm token-usage coverage emit failed provider=%s model_name=%s err=%s: %s", + self.provider, + self.model, + type(e).__name__, + e, + ) + def update_token_usage( self, model_name: str, diff --git a/openviking/telemetry/operation.py b/openviking/telemetry/operation.py index 6ff3eb626d..24944a3ec4 100644 --- a/openviking/telemetry/operation.py +++ b/openviking/telemetry/operation.py @@ -64,6 +64,8 @@ class TelemetrySummaryBuilder: _PRUNED = object() _PRESERVED_ZERO_METRIC_PATHS = { + ("tokens", "llm", "coverage", "missing_report_count"), + ("tokens", "llm", "coverage", "reported_response_count"), ("vector", "cuvs", "memory", "estimated_peak_bytes_max"), ("vector", "cuvs", "memory", "free_bytes_min"), ("vector", "cuvs", "memory", "usable_bytes_min"), @@ -229,6 +231,14 @@ def build( llm_completion_reasoning_tokens = cls._i(counters.get("tokens.llm.completion_reasoning"), 0) embedding_total_tokens = cls._i(counters.get("tokens.embedding.total"), 0) rerank_total_tokens = cls._i(counters.get("tokens.rerank.total"), 0) + llm_successful_response_count = max( + cls._i(counters.get("tokens.llm.coverage.successful_responses"), 0), + 0, + ) + llm_reported_response_count = min( + max(cls._i(counters.get("tokens.llm.coverage.reported_responses"), 0), 0), + llm_successful_response_count, + ) stage_token_summary = cls._build_stage_token_summary(counters) vector_candidates_scored = cls._i(counters.get("vector.scored"), 0) vectors_scanned = gauges.get("vector.scanned") @@ -257,6 +267,20 @@ def build( } if stage_token_summary: summary["tokens"]["stages"] = stage_token_summary + if llm_successful_response_count > 0: + llm_missing_report_count = llm_successful_response_count - llm_reported_response_count + if llm_reported_response_count == llm_successful_response_count: + coverage_status = "complete" + elif llm_reported_response_count == 0: + coverage_status = "missing" + else: + coverage_status = "partial" + summary["tokens"]["llm"]["coverage"] = { + "status": coverage_status, + "successful_response_count": llm_successful_response_count, + "reported_response_count": llm_reported_response_count, + "missing_report_count": llm_missing_report_count, + } if cls._has_metric_prefix("queue", counters, gauges): summary["queue"] = { @@ -716,6 +740,15 @@ def add_token_usage( completion_reasoning_tokens=completion_reasoning_tokens, ) + def record_llm_token_usage_report(self, *, reported: bool) -> None: + """Record whether one successful LLM response included token usage.""" + if not self.enabled: + return + with self._lock: + self._counters["tokens.llm.coverage.successful_responses"] += 1 + if reported: + self._counters["tokens.llm.coverage.reported_responses"] += 1 + def record_token_usage( self, source: str, diff --git a/tests/models/vlm/test_token_usage_coverage.py b/tests/models/vlm/test_token_usage_coverage.py new file mode 100644 index 0000000000..ff1e63772f --- /dev/null +++ b/tests/models/vlm/test_token_usage_coverage.py @@ -0,0 +1,115 @@ +# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. +# SPDX-License-Identifier: AGPL-3.0 +"""Tests for VLM token-usage coverage telemetry.""" + +from types import SimpleNamespace + +import pytest + +from openviking.models.vlm.backends.litellm_vlm import LiteLLMVLMProvider +from openviking.models.vlm.backends.openai_vlm import OpenAIVLM +from openviking.models.vlm.backends.volcengine_media import _record_usage +from openviking.models.vlm.backends.volcengine_vlm import VolcEngineVLM +from openviking.telemetry.backends.memory import MemoryOperationTelemetry +from openviking.telemetry.context import bind_telemetry + + +def _response(prompt_tokens: int | None, completion_tokens: int = 0): + if prompt_tokens is None: + return SimpleNamespace(usage=None) + return SimpleNamespace( + usage=SimpleNamespace( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + prompt_tokens_details=None, + completion_tokens_details=None, + ) + ) + + +@pytest.mark.parametrize( + ("provider_type", "config"), + [ + (OpenAIVLM, {"provider": "openai", "model": "test-model"}), + (LiteLLMVLMProvider, {"provider": "litellm", "model": "test-model"}), + ], + ids=["openai", "litellm"], +) +@pytest.mark.parametrize( + ("responses", "expected_status", "expected_counts"), + [ + ([(None, 0)], "missing", (1, 0, 1)), + ([(0, 0)], "complete", (1, 1, 0)), + ([(10, 4), (None, 0)], "partial", (2, 1, 1)), + ], + ids=["missing", "explicit-zero", "partial"], +) +def test_chat_response_token_usage_coverage( + provider_type, + config, + responses, + expected_status, + expected_counts, +): + telemetry = MemoryOperationTelemetry(operation="resources.add_resource", enabled=True) + provider = provider_type(config) + + with bind_telemetry(telemetry): + for prompt_tokens, completion_tokens in responses: + provider._update_token_usage_from_response(_response(prompt_tokens, completion_tokens)) + + summary = telemetry.finish().summary + coverage = summary["tokens"]["llm"]["coverage"] + successful, reported, missing = expected_counts + assert coverage == { + "status": expected_status, + "successful_response_count": successful, + "reported_response_count": reported, + "missing_report_count": missing, + } + if responses == [(10, 4), (None, 0)]: + assert summary["tokens"]["llm"]["total"] == 14 + + +@pytest.mark.parametrize( + ("usage", "expected_status"), + [(None, "missing"), ((0, 0), "complete")], + ids=["missing", "explicit-zero"], +) +def test_volcengine_media_response_token_usage_coverage(usage, expected_status): + telemetry = MemoryOperationTelemetry(operation="resources.add_resource", enabled=True) + provider = VolcEngineVLM({"provider": "volcengine", "model": "test-model"}) + if usage is None: + response = SimpleNamespace(usage=None) + else: + input_tokens, output_tokens = usage + response = SimpleNamespace( + usage=SimpleNamespace( + input_tokens=input_tokens, + output_tokens=output_tokens, + input_tokens_details=None, + output_tokens_details=None, + ) + ) + + with bind_telemetry(telemetry): + _record_usage(provider, response, duration_seconds=0.1) + + assert summary_coverage(telemetry) == { + "status": expected_status, + "successful_response_count": 1, + "reported_response_count": int(usage is not None), + "missing_report_count": int(usage is None), + } + + +def test_operation_without_successful_llm_response_omits_coverage(): + telemetry = MemoryOperationTelemetry(operation="resources.add_resource", enabled=True) + + summary = telemetry.finish().summary + + assert "tokens" not in summary + + +def summary_coverage(telemetry: MemoryOperationTelemetry): + return telemetry.finish().summary["tokens"]["llm"]["coverage"]