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/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) 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