diff --git a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py index 9aba00bed1..704a401113 100644 --- a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py +++ b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py @@ -35,6 +35,58 @@ logger = logging.getLogger(__name__) +FR_STREAMING_TIME_TO_FIRST_TOKEN_MS = "fortifyroot.llm.streaming.time_to_first_token_ms" +FR_STREAMING_TIME_TO_GENERATE_MS = "fortifyroot.llm.streaming.time_to_generate_ms" +GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS = getattr( + SpanAttributes, + "GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS", + SpanAttributes.LLM_USAGE_CACHE_READ_INPUT_TOKENS, +) +GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS = getattr( + SpanAttributes, + "GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS", + SpanAttributes.LLM_USAGE_CACHE_CREATION_INPUT_TOKENS, +) + + +def _streaming_latency_ms(start_time, end_time): + return int(round((end_time - start_time) * 1000)) + + +def _is_streaming_token_item(item): + if getattr(item, "type", None) != "content_block_delta": + return False + + delta = getattr(item, "delta", None) + delta_type = getattr(delta, "type", None) + if delta_type == "text_delta": + return bool(getattr(delta, "text", None)) + if delta_type == "thinking_delta": + return bool(getattr(delta, "thinking", None)) + if delta_type == "input_json_delta": + return bool(getattr(delta, "partial_json", None)) + return False + + +def _set_streaming_first_token_latency(span, start_time, first_token_time): + if not span or not span.is_recording(): + return + set_span_attribute( + span, + FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, + _streaming_latency_ms(start_time, first_token_time), + ) + + +def _set_streaming_time_to_generate(span, first_token_time, end_time): + if not first_token_time or not span or not span.is_recording(): + return + set_span_attribute( + span, + FR_STREAMING_TIME_TO_GENERATE_MS, + _streaming_latency_ms(first_token_time, end_time), + ) + @dont_throw def _process_response_item(item, complete_response): @@ -102,10 +154,10 @@ def _set_token_usage( set_span_attribute(span, SpanAttributes.LLM_USAGE_TOTAL_TOKENS, total_tokens) set_span_attribute( - span, SpanAttributes.GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, cache_read_tokens + span, GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, cache_read_tokens ) set_span_attribute( - span, SpanAttributes.GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, cache_creation_tokens + span, GEN_AI_USAGE_CACHE_CREATION_INPUT_TOKENS, cache_creation_tokens ) set_span_attribute( @@ -182,6 +234,7 @@ def __init__( self._complete_response = {"events": [], "model": "", "usage": {}, "id": ""} self._instrumentation_completed = False + self._first_token_time = None self._pending_item = None # FR: pending-item buffer for streaming safety self._streaming_safety = AnthropicStreamingSafety(span, getattr(span, "name", "anthropic.chat")) # FR: streaming safety init @@ -252,6 +305,13 @@ def __next__(self): # FR: rewritten for pending-item safety buffer raise e item = self._streaming_safety.process_item(item) # FR: streaming safety processing + if self._first_token_time is None and _is_streaming_token_item(item): + self._first_token_time = time.time() + _set_streaming_first_token_latency( + self._span, + self._start_time, + self._first_token_time, + ) if self._pending_item is None: # FR: pending-item buffer logic self._pending_item = item # FR: pending-item buffer logic continue # FR: pending-item buffer logic @@ -275,8 +335,14 @@ def _handle_completion(self): GenAIAttributes.GEN_AI_RESPONSE_ID, self._complete_response.get("id"), ) + end_time = time.time() + _set_streaming_time_to_generate( + self._span, + self._first_token_time, + end_time, + ) if self._duration_histogram: - duration = time.time() - self._start_time + duration = end_time - self._start_time self._duration_histogram.record( duration, attributes=metric_attributes, @@ -369,6 +435,7 @@ def __init__( self._complete_response = {"events": [], "model": "", "usage": {}, "id": ""} self._instrumentation_completed = False + self._first_token_time = None self._pending_item = None # FR: pending-item buffer for streaming safety self._streaming_safety = AnthropicStreamingSafety(span, getattr(span, "name", "anthropic.chat")) # FR: streaming safety init @@ -442,6 +509,13 @@ async def __anext__(self): # FR: rewritten for pending-item safety buffer raise item = self._streaming_safety.process_item(item) # FR: streaming safety processing + if self._first_token_time is None and _is_streaming_token_item(item): + self._first_token_time = time.time() + _set_streaming_first_token_latency( + self._span, + self._start_time, + self._first_token_time, + ) if self._pending_item is None: # FR: pending-item buffer logic self._pending_item = item # FR: pending-item buffer logic continue # FR: pending-item buffer logic @@ -463,9 +537,15 @@ def _complete_instrumentation(self): request_model=self._kwargs.get("model"), ) set_span_attribute(self._span, GenAIAttributes.GEN_AI_RESPONSE_ID, self._complete_response.get("id")) + end_time = time.time() + _set_streaming_time_to_generate( + self._span, + self._first_token_time, + end_time, + ) if self._duration_histogram: - duration = time.time() - self._start_time + duration = end_time - self._start_time self._duration_histogram.record( duration, attributes=metric_attributes, @@ -482,6 +562,7 @@ def _complete_instrumentation(self): if usage := self._complete_response.get("usage"): completion_tokens = usage.get("output_tokens", 0) else: + completion_tokens = 0 completion_content = "" if self._complete_response.get("events"): model_name = self._complete_response.get("model") or None diff --git a/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py b/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py index 278362f817..cfa2125c0a 100644 --- a/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py +++ b/packages/opentelemetry-instrumentation-anthropic/tests/test_safety_unit.py @@ -3,12 +3,15 @@ import pytest from opentelemetry.instrumentation.anthropic import safety +from opentelemetry.instrumentation.anthropic.config import Config from opentelemetry.instrumentation.anthropic.streaming_safety import ( AnthropicStreamingSafety, ) from opentelemetry.instrumentation.anthropic.streaming import ( AnthropicAsyncStream, AnthropicStream, + FR_STREAMING_TIME_TO_FIRST_TOKEN_MS, + FR_STREAMING_TIME_TO_GENERATE_MS, ) from opentelemetry.instrumentation.fortifyroot import SafetyDecision, SafetyResult from opentelemetry.instrumentation.fortifyroot import ( @@ -410,6 +413,89 @@ def test_anthropic_sync_stream_span_not_ended_before_last_item_consumed(): assert len(exporter.get_finished_spans()) == 1 +def test_anthropic_sync_stream_sets_streaming_latency_span_attrs(monkeypatch): + clear_safety_handlers() + exporter, tracer = _test_tracer() + + timestamps = iter([101.25, 103.0]) + monkeypatch.setattr( + "opentelemetry.instrumentation.anthropic.streaming.time.time", + lambda: next(timestamps), + ) + + span = tracer.start_span("anthropic.chat") + stream = AnthropicStream( + span, + _Iterator( + [ + SimpleNamespace( + type="content_block_start", + index=0, + content_block=SimpleNamespace(type="text"), + ), + SimpleNamespace( + type="content_block_delta", + index=0, + delta=SimpleNamespace(type="text_delta", text="hello"), + ), + ] + ), + SimpleNamespace(count_tokens=lambda text: len(text)), + 100.0, + kwargs={}, + ) + + first = next(stream) + second = next(stream) + with pytest.raises(StopIteration): + next(stream) + + finished = exporter.get_finished_spans() + assert first.type == "content_block_start" + assert second.type == "content_block_delta" + assert len(finished) == 1 + assert finished[0].attributes[FR_STREAMING_TIME_TO_FIRST_TOKEN_MS] == 1250 + assert finished[0].attributes[FR_STREAMING_TIME_TO_GENERATE_MS] == 1750 + + +def test_anthropic_sync_stream_without_token_omits_streaming_latency_attrs(monkeypatch): + clear_safety_handlers() + exporter, tracer = _test_tracer() + + timestamps = iter([301.0]) + monkeypatch.setattr( + "opentelemetry.instrumentation.anthropic.streaming.time.time", + lambda: next(timestamps), + ) + + span = tracer.start_span("anthropic.chat") + stream = AnthropicStream( + span, + _Iterator( + [ + SimpleNamespace( + type="content_block_start", + index=0, + content_block=SimpleNamespace(type="text"), + ) + ] + ), + SimpleNamespace(count_tokens=lambda text: len(text)), + 300.0, + kwargs={}, + ) + + first = next(stream) + with pytest.raises(StopIteration): + next(stream) + + finished = exporter.get_finished_spans() + assert first.type == "content_block_start" + assert len(finished) == 1 + assert FR_STREAMING_TIME_TO_FIRST_TOKEN_MS not in finished[0].attributes + assert FR_STREAMING_TIME_TO_GENERATE_MS not in finished[0].attributes + + @pytest.mark.asyncio async def test_anthropic_async_stream_span_not_ended_before_last_item_consumed(): """D-13: Async variant - span must not be ended until the consumer has processed the last pending item.""" @@ -444,3 +530,50 @@ async def test_anthropic_async_stream_span_not_ended_before_last_item_consumed() assert stream._instrumentation_completed assert len(exporter.get_finished_spans()) == 1 + + +@pytest.mark.asyncio +async def test_anthropic_async_stream_sets_streaming_latency_span_attrs(monkeypatch): + clear_safety_handlers() + exporter, tracer = _test_tracer() + monkeypatch.setattr(Config, "enrich_token_usage", True) + + timestamps = iter([201.1, 202.4]) + monkeypatch.setattr( + "opentelemetry.instrumentation.anthropic.streaming.time.time", + lambda: next(timestamps), + ) + + span = tracer.start_span("anthropic.chat") + stream = AnthropicAsyncStream( + span, + _AsyncIterator( + [ + SimpleNamespace( + type="content_block_start", + index=0, + content_block=SimpleNamespace(type="text"), + ), + SimpleNamespace( + type="content_block_delta", + index=0, + delta=SimpleNamespace(type="text_delta", text="hello"), + ), + ] + ), + SimpleNamespace(count_tokens=lambda text: len(text)), + 200.0, + kwargs={}, + ) + + first = await stream.__anext__() + second = await stream.__anext__() + with pytest.raises(StopAsyncIteration): + await stream.__anext__() + + finished = exporter.get_finished_spans() + assert first.type == "content_block_start" + assert second.type == "content_block_delta" + assert len(finished) == 1 + assert finished[0].attributes[FR_STREAMING_TIME_TO_FIRST_TOKEN_MS] == 1100 + assert finished[0].attributes[FR_STREAMING_TIME_TO_GENERATE_MS] == 1300