From 570ad12f2dea62eca5afb00c14837250d27ca963 Mon Sep 17 00:00:00 2001 From: CJstate <1507965754@qq.com> Date: Thu, 1 Oct 2026 16:53:49 +0800 Subject: [PATCH 1/2] fix(groq): keep usage from choice-less streaming chunks Groq sends the token usage payload on a trailing chunk that can carry no choices at all. _process_streaming_chunk returned before it read x_groq.usage, so that chunk contributed nothing and the streaming token histograms stayed empty even though the usage was sitting right there. Read usage before the empty-choices guard and hand it back instead of None. The metric-recording work this branch originally carried has since landed upstream in #4439, so the branch is rebased onto upstream/main and now holds only the guard fix and its regression test. Signed-off-by: CJstate <142857225+CJstate@users.noreply.github.com> --- .../opentelemetry/instrumentation/groq/__init__.py | 14 ++++++++------ .../tests/traces/test_init.py | 13 +++++++++++++ 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py index 2924b5de56..8adc3161db 100644 --- a/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py +++ b/packages/opentelemetry-instrumentation-groq/opentelemetry/instrumentation/groq/__init__.py @@ -129,8 +129,15 @@ def _create_metrics(meter: Meter): def _process_streaming_chunk(chunk): """Extract content, tool_calls_delta, finish_reasons and usage from a streaming chunk.""" + # Groq attaches x_groq.usage to a final chunk that may carry no choices at + # all, so read it before the empty-choices guard: returning early would drop + # the only usage-bearing chunk and leave the token histograms empty. + usage = None + if hasattr(chunk, "x_groq") and chunk.x_groq and chunk.x_groq.usage: + usage = chunk.x_groq.usage + if not chunk.choices: - return None, [], [], None + return None, [], [], usage content = "" tool_calls_delta = [] @@ -144,11 +151,6 @@ def _process_streaming_chunk(chunk): if choice.finish_reason: finish_reasons.append(choice.finish_reason) - # Extract usage from x_groq if present in the final chunk - usage = None - if hasattr(chunk, "x_groq") and chunk.x_groq and chunk.x_groq.usage: - usage = chunk.x_groq.usage - return content, tool_calls_delta, finish_reasons, usage diff --git a/packages/opentelemetry-instrumentation-groq/tests/traces/test_init.py b/packages/opentelemetry-instrumentation-groq/tests/traces/test_init.py index ab940304ff..2e9cd665d6 100644 --- a/packages/opentelemetry-instrumentation-groq/tests/traces/test_init.py +++ b/packages/opentelemetry-instrumentation-groq/tests/traces/test_init.py @@ -81,9 +81,22 @@ def _make_async_wrapper( class TestProcessStreamingChunk: def test_empty_choices_returns_none_quad(self): chunk = MagicMock() + chunk.x_groq = None chunk.choices = [] assert _process_streaming_chunk(chunk) == (None, [], [], None) + def test_empty_choices_preserves_usage(self): + # Groq sends the usage payload on a trailing chunk that has no choices; + # the guard must not discard it. + chunk = MagicMock() + chunk.choices = [] + chunk.x_groq.usage.prompt_tokens = 9 + chunk.x_groq.usage.completion_tokens = 4 + content, tool_calls_delta, finish_reasons, usage = _process_streaming_chunk(chunk) + assert (content, tool_calls_delta, finish_reasons) == (None, [], []) + assert usage.prompt_tokens == 9 + assert usage.completion_tokens == 4 + def test_multiple_choices_accumulates_content(self): chunk = MagicMock() chunk.x_groq = None From 3c5702ff6bb98530386121a5ff3b1f402d64f68d Mon Sep 17 00:00:00 2001 From: CJstate <142857225+CJstate@users.noreply.github.com> Date: Fri, 2 Oct 2026 19:17:30 +0800 Subject: [PATCH 2/2] test(groq): cover usage on a choice-less final chunk CodeRabbit pointed out that the sync and async metric tests feed usage on a chunk that carries a choice, so they never exercise the branch this PR fixes: the guard in `_process_streaming_chunk` that returns before reading `x_groq.usage`. Add one metric-level test per processor that puts the usage payload on a trailing chunk with no choices at all and asserts the token histograms reach the reader. Both tests fail on the pre-fix guard ("assert token_metric is not None") and pass with it, so the fix is now covered at the level a user observes, not only at the helper's return value. Signed-off-by: CJstate <142857225+CJstate@users.noreply.github.com> --- .../metrics/test_groq_streaming_metrics.py | 77 +++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py index dceea85d93..1dcaa56f59 100644 --- a/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py +++ b/packages/opentelemetry-instrumentation-groq/tests/metrics/test_groq_streaming_metrics.py @@ -33,6 +33,13 @@ def _chunk(content="", finish_reason=None, usage=None, model=None): ) +def _empty_choice_usage_chunk(usage): + """A trailing chunk with no choices that still carries the usage payload, as Groq sends it.""" + chunk = _chunk(usage=usage) + chunk.choices = [] + return chunk + + def _usage(prompt=10, completion=5): return SimpleNamespace( prompt_tokens=prompt, @@ -143,6 +150,42 @@ def test_records_token_and_duration_metrics(self, reader, tracer_provider, meter assert duration_metric is not None assert duration_metric.data.data_points[0].sum > 0 + def test_records_token_metrics_from_a_choice_less_final_chunk(self, reader, tracer_provider, meter_provider): + """Usage that arrives on a trailing chunk with no choices must still be recorded. + + Groq attaches the usage payload to a final chunk that carries no choices, so the + empty-choices guard in `_process_streaming_chunk` must not drop it: this is the + difference between an empty token histogram and a populated one. + """ + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + token_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_TOKEN_USAGE) + + stream = _FakeStream( + [ + _chunk(content="hello"), + _chunk(content=" world", finish_reason="stop"), + _empty_choice_usage_chunk(_usage(prompt=10, completion=5)), + ] + ) + + for _ in _create_stream_processor( + stream, + span, + None, + token_histogram, + None, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ): + pass + + token_metric = _find_metric(reader.get_metrics_data(), Meters.LLM_TOKEN_USAGE) + assert token_metric is not None + input_dp = _find_data_point(token_metric, {GenAIAttributes.GEN_AI_TOKEN_TYPE: "input"}) + output_dp = _find_data_point(token_metric, {GenAIAttributes.GEN_AI_TOKEN_TYPE: "output"}) + assert input_dp is not None and input_dp.sum == 10 + assert output_dp is not None and output_dp.sum == 5 + def test_token_points_carry_non_streaming_attribute_set(self, reader, tracer_provider, meter_provider): """Token points must match the non-streaming attribute keys (semconv requires operation.name).""" span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") @@ -334,6 +377,40 @@ async def test_async_records_token_and_duration_metrics(self, reader, tracer_pro assert duration_metric is not None assert duration_metric.data.data_points[0].sum > 0 + @pytest.mark.asyncio + async def test_async_records_token_metrics_from_a_choice_less_final_chunk( + self, reader, tracer_provider, meter_provider + ): + span = get_tracer("test", tracer_provider=tracer_provider).start_span("chat llama-3.3-70b-versatile") + token_histogram = meter_provider.get_meter("test").create_histogram(name=Meters.LLM_TOKEN_USAGE) + + stream = _FakeAsyncStream( + [ + _chunk(content="hello"), + _chunk(content=" world", finish_reason="stop"), + _empty_choice_usage_chunk(_usage(prompt=7, completion=3)), + ] + ) + + processor = _create_async_stream_processor( + stream, + span, + None, + token_histogram, + None, + start_time=0, + llm_model="llama-3.3-70b-versatile", + ) + async for _ in processor: + pass + + token_metric = _find_metric(reader.get_metrics_data(), Meters.LLM_TOKEN_USAGE) + assert token_metric is not None + input_dp = _find_data_point(token_metric, {GenAIAttributes.GEN_AI_TOKEN_TYPE: "input"}) + output_dp = _find_data_point(token_metric, {GenAIAttributes.GEN_AI_TOKEN_TYPE: "output"}) + assert input_dp is not None and input_dp.sum == 7 + assert output_dp is not None and output_dp.sum == 3 + # --------------------------------------------------------------------------- # _create_stream_processor failure paths (review: record duration on error)