diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/654.fixed b/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/654.fixed new file mode 100644 index 000000000..3805f1637 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/.changelog/654.fixed @@ -0,0 +1 @@ +Record response telemetry on ``messages.stream()`` when the stream is consumed through ``text_stream``, ``get_final_message()``, ``get_final_text()`` or ``until_done()``, which previously produced a span with request attributes only. diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py index fbdd448a4..8f4253194 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py @@ -117,6 +117,9 @@ class _MessagesStreamMixin(Generic[ResponseFormatT]): def _stop(self) -> None: if self._self_message_telemetry_finalized: return + # text_stream and the get_final_* helpers bypass _process_chunk, so the + # snapshot can be the only record of the response. + self._adopt_sdk_snapshot() _set_response_attributes( self._self_invocation, self._self_message, @@ -137,6 +140,28 @@ def _on_stream_end(self) -> None: def _on_stream_error(self, error: BaseException) -> None: self._fail(error) + def _adopt_sdk_snapshot(self) -> bool: + """Adopt the SDK stream's accumulated message, when it keeps one. + + ``MessageStream`` updates ``current_message_snapshot`` as the response + arrives, whichever accessor the caller reads it through. A plain + ``Stream`` has no snapshot and leaves accumulation to us. + """ + stream = cast(_StreamWrapperWithStream, self).stream + try: + snapshot = cast( + "ParsedMessage[ResponseFormatT] | None", + getattr(stream, "current_message_snapshot", None), + ) + except AssertionError: + # The property asserts the snapshot is set, so an unconsumed stream + # raises rather than answering None. + return False + if snapshot is None: + return False + self._self_message = snapshot + return True + def _process_chunk( self, chunk: RawMessageStreamEvent @@ -144,13 +169,7 @@ def _process_chunk( ) -> None: """Accumulate a final message snapshot from a streaming chunk.""" global _accumulation_disabled - stream = cast(_StreamWrapperWithStream, self).stream - snapshot = cast( - "ParsedMessage[ResponseFormatT] | None", - getattr(stream, "current_message_snapshot", None), - ) - if snapshot is not None: - self._self_message = snapshot + if self._adopt_sdk_snapshot(): return if accumulate_event is None or _accumulation_disabled: return diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py index b33377053..b75ccb0bd 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_async_messages.py @@ -2093,3 +2093,180 @@ def __init__(self): assert first_stops == [True] assert second_stops == [True] assert http_response.close_calls == 1 + + +@pytest.mark.asyncio +@pytest.mark.vcr() +@pytest.mark.cassette("test_async_messages_stream") +async def test_async_messages_stream_text_stream_records_response( + span_exporter, async_anthropic_client, instrument_with_content +): + """``stream.text_stream`` records the response the same as event iteration.""" + model = "claude-haiku-4-5" + + text = "" + async with async_anthropic_client.messages.stream( + model=model, + max_tokens=100, + messages=[{"role": "user", "content": "Say hello in one word."}], + ) as stream: + async for chunk in stream.text_stream: + text += chunk + + assert text == "Hello." + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == model + assert ( + span.attributes[GenAIAttributes.GEN_AI_RESPONSE_ID] + == "msg_01N2EGWxw2zHUcjTjToMivN6" + ) + assert ( + span.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] + == "claude-haiku-4-5-20251001" + ) + assert span.attributes[GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS] == 13 + assert span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS] == 5 + assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS] == ( + "stop", + ) + output_messages = _load_span_messages( + span, GenAIAttributes.GEN_AI_OUTPUT_MESSAGES + ) + assert output_messages[0]["role"] == "assistant" + assert output_messages[0]["parts"] == [ + {"type": "text", "content": "Hello."} + ] + + +@pytest.mark.asyncio +@pytest.mark.vcr() +@pytest.mark.cassette("test_async_messages_stream") +async def test_async_messages_stream_get_final_message_records_response( + span_exporter, async_anthropic_client, instrument_no_content +): + """``get_final_message()`` drains the SDK's iterator and still records.""" + model = "claude-haiku-4-5" + + async with async_anthropic_client.messages.stream( + model=model, + max_tokens=100, + messages=[{"role": "user", "content": "Say hello in one word."}], + ) as stream: + message = await stream.get_final_message() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == model + assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_ID] == message.id + assert ( + span.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == message.model + ) + assert span.attributes[ + GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS + ] == expected_input_tokens(message.usage) + assert ( + span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS] + == message.usage.output_tokens + ) + assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS] == ( + normalize_stop_reason(message.stop_reason), + ) + + +@pytest.mark.asyncio +@pytest.mark.vcr() +@pytest.mark.cassette("test_async_messages_stream") +async def test_async_messages_stream_get_final_text_records_response( + span_exporter, async_anthropic_client, instrument_no_content +): + """``get_final_text()`` drains the SDK's iterator and still records.""" + model = "claude-haiku-4-5" + + async with async_anthropic_client.messages.stream( + model=model, + max_tokens=100, + messages=[{"role": "user", "content": "Say hello in one word."}], + ) as stream: + text = await stream.get_final_text() + + assert text == "Hello." + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == model + assert ( + span.attributes[GenAIAttributes.GEN_AI_RESPONSE_ID] + == "msg_01N2EGWxw2zHUcjTjToMivN6" + ) + assert ( + span.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] + == "claude-haiku-4-5-20251001" + ) + assert span.attributes[GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS] == 13 + assert span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS] == 5 + assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS] == ( + "stop", + ) + + +@pytest.mark.asyncio +@pytest.mark.vcr() +@pytest.mark.cassette("test_async_messages_stream") +async def test_async_messages_stream_until_done_records_response( + span_exporter, async_anthropic_client, instrument_no_content +): + """``until_done()`` drains the SDK's iterator and still records.""" + model = "claude-haiku-4-5" + + async with async_anthropic_client.messages.stream( + model=model, + max_tokens=100, + messages=[{"role": "user", "content": "Say hello in one word."}], + ) as stream: + await stream.until_done() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert ( + span.attributes[GenAIAttributes.GEN_AI_RESPONSE_ID] + == "msg_01N2EGWxw2zHUcjTjToMivN6" + ) + assert ( + span.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] + == "claude-haiku-4-5-20251001" + ) + assert span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS] == 5 + assert span.attributes[GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS] == ( + "stop", + ) + + +@pytest.mark.asyncio +@pytest.mark.vcr() +@pytest.mark.cassette("test_async_messages_stream") +async def test_async_messages_stream_text_stream_user_exception( + span_exporter, async_anthropic_client, instrument_no_content +): + """A caller error while reading ``text_stream`` propagates and is recorded.""" + model = "claude-haiku-4-5" + + with pytest.raises(ValueError, match="caller failed"): + async with async_anthropic_client.messages.stream( + model=model, + max_tokens=100, + messages=[{"role": "user", "content": "Say hello in one word."}], + ) as stream: + async for _ in stream.text_stream: + raise ValueError("caller failed") + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == model + assert span.attributes[ErrorAttributes.ERROR_TYPE] == "ValueError" diff --git a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py index 4ba22eafd..85f9cd4aa 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py +++ b/instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/test_sync_messages.py @@ -2310,3 +2310,158 @@ def test_sync_messages_raw_response_parse_after_exit( assert spans[0].attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == model assert raw_response.parse().model == model + + +@pytest.mark.vcr() +@pytest.mark.cassette("test_sync_messages_stream") +def test_sync_messages_stream_text_stream_records_response( + span_exporter, anthropic_client, instrument_with_content +): + """``stream.text_stream`` records the response the same as event iteration.""" + model = "claude-sonnet-4-20250514" + + with anthropic_client.messages.stream( + model=model, + max_tokens=100, + messages=[{"role": "user", "content": "Say hello in one word."}], + ) as stream: + text = "".join(stream.text_stream) + + assert text == "Hello!" + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert_span_attributes( + span, + request_model=model, + response_id="msg_01FpWuSsvRgJp3eYbdHBinNp", + response_model=model, + input_tokens=13, + output_tokens=5, + finish_reasons=["stop"], + ) + assert isinstance(span.attributes[GenAIAttributes.GEN_AI_RESPONSE_ID], str) + assert isinstance( + span.attributes[GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS], int + ) + assert isinstance( + span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS], int + ) + output_messages = _load_span_messages( + span, GenAIAttributes.GEN_AI_OUTPUT_MESSAGES + ) + assert output_messages[0]["role"] == "assistant" + assert output_messages[0]["parts"] == [ + {"type": "text", "content": "Hello!"} + ] + + +@pytest.mark.vcr() +@pytest.mark.cassette("test_sync_messages_stream") +def test_sync_messages_stream_get_final_message_records_response( + span_exporter, anthropic_client, instrument_no_content +): + """``get_final_message()`` drains the SDK's iterator and still records.""" + model = "claude-sonnet-4-20250514" + + with anthropic_client.messages.stream( + model=model, + max_tokens=100, + messages=[{"role": "user", "content": "Say hello in one word."}], + ) as stream: + message = stream.get_final_message() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + assert_span_attributes( + spans[0], + request_model=model, + response_id=message.id, + response_model=message.model, + input_tokens=expected_input_tokens(message.usage), + output_tokens=message.usage.output_tokens, + finish_reasons=[normalize_stop_reason(message.stop_reason)], + ) + + +@pytest.mark.vcr() +@pytest.mark.cassette("test_sync_messages_stream") +def test_sync_messages_stream_get_final_text_records_response( + span_exporter, anthropic_client, instrument_no_content +): + """``get_final_text()`` drains the SDK's iterator and still records.""" + model = "claude-sonnet-4-20250514" + + with anthropic_client.messages.stream( + model=model, + max_tokens=100, + messages=[{"role": "user", "content": "Say hello in one word."}], + ) as stream: + text = stream.get_final_text() + + assert text == "Hello!" + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + assert_span_attributes( + spans[0], + request_model=model, + response_id="msg_01FpWuSsvRgJp3eYbdHBinNp", + response_model=model, + input_tokens=13, + output_tokens=5, + finish_reasons=["stop"], + ) + + +@pytest.mark.vcr() +@pytest.mark.cassette("test_sync_messages_stream") +def test_sync_messages_stream_until_done_records_response( + span_exporter, anthropic_client, instrument_no_content +): + """``until_done()`` drains the SDK's iterator and still records.""" + model = "claude-sonnet-4-20250514" + + with anthropic_client.messages.stream( + model=model, + max_tokens=100, + messages=[{"role": "user", "content": "Say hello in one word."}], + ) as stream: + stream.until_done() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + assert_span_attributes( + spans[0], + request_model=model, + response_id="msg_01FpWuSsvRgJp3eYbdHBinNp", + response_model=model, + input_tokens=13, + output_tokens=5, + finish_reasons=["stop"], + ) + + +@pytest.mark.vcr() +@pytest.mark.cassette("test_sync_messages_stream") +def test_sync_messages_stream_text_stream_user_exception( + span_exporter, anthropic_client, instrument_no_content +): + """A caller error while reading ``text_stream`` propagates and is recorded.""" + model = "claude-sonnet-4-20250514" + + with pytest.raises(ValueError, match="caller failed"): + with anthropic_client.messages.stream( + model=model, + max_tokens=100, + messages=[{"role": "user", "content": "Say hello in one word."}], + ) as stream: + for _ in stream.text_stream: + raise ValueError("caller failed") + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == model + assert span.attributes[ErrorAttributes.ERROR_TYPE] == "ValueError"