diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 5dc9fa8828c..9ecc90d06c8 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -2064,59 +2064,85 @@ def _record_duration() -> None: ) except Exception as exception: capture_exception(span=span, exception=exception, timestamp=time_ns()) + _capture_operation_error( + attributes=attributes, + exception=exception, + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=perf_counter() - start_time, + ) _close_span() raise async def _finalize_stream() -> None: from ._types import ChatResponse - try: - if result_stream._stream_error is not None: # pyright: ignore[reportPrivateUsage] - # Stream errored; skip get_final_response() to avoid firing - # result hooks such as after_run context providers on error - # paths. Capture the error on the span before returning. - capture_exception( - span=span, - exception=result_stream._stream_error, # type: ignore - timestamp=time_ns(), - ) - return - response: ChatResponse[Any] = await result_stream.get_final_response() - duration = duration_state.get("duration") - response_attributes = _get_response_attributes(attributes, response) - self._backfill_request_model(span, response_attributes) - _capture_response( + if result_stream._stream_error is not None: # pyright: ignore[reportPrivateUsage] + # Stream errored; skip get_final_response() to avoid firing + # result hooks such as after_run context providers on error + # paths. Capture the error on the span before returning. + capture_exception( span=span, - attributes=response_attributes, - token_usage_histogram=getattr(self, "token_usage_histogram", None), + exception=result_stream._stream_error, # type: ignore + timestamp=time_ns(), + ) + _capture_operation_error( + attributes=attributes, + exception=result_stream._stream_error, # type: ignore operation_duration_histogram=getattr(self, "duration_histogram", None), - duration=duration, + duration=duration_state.get("duration", perf_counter() - start_time), ) - _mark_inner_response_telemetry_captured(response) - if ( - OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED - and isinstance(response, ChatResponse) - and response.messages - and span.is_recording() - ): - finish_reason = _get_response_finish_reason(response) - # Activate the span: this cleanup hook runs after the final pull has - # exited its _activate_span context, so it wouldn't otherwise be current. - with _activate_span(span): - _capture_message_events_v1_36( - provider_name=provider_name, + _close_span() + return + + try: + try: + response: ChatResponse[Any] = await result_stream.get_final_response() + except Exception as exception: + capture_exception(span=span, exception=exception, timestamp=time_ns()) + _capture_operation_error( + attributes=attributes, + exception=exception, + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=duration_state.get("duration", perf_counter() - start_time), + ) + raise + + try: + duration = duration_state.get("duration") + response_attributes = _get_response_attributes(attributes, response) + self._backfill_request_model(span, response_attributes) + _capture_response( + span=span, + attributes=response_attributes, + token_usage_histogram=getattr(self, "token_usage_histogram", None), + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=duration, + ) + _mark_inner_response_telemetry_captured(response) + if ( + OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED + and isinstance(response, ChatResponse) + and response.messages + and span.is_recording() + ): + finish_reason = _get_response_finish_reason(response) + # Activate the span: this cleanup hook runs after the final pull has + # exited its _activate_span context, so it wouldn't otherwise be current. + with _activate_span(span): + _capture_message_events_v1_36( + provider_name=provider_name, + messages=response.messages, + finish_reason=finish_reason, + output=True, + ) + _capture_message_span_attributes_latest_experimental( + span=span, messages=response.messages, finish_reason=finish_reason, output=True, ) - _capture_message_span_attributes_latest_experimental( - span=span, - messages=response.messages, - finish_reason=finish_reason, - output=True, - ) - except Exception as exception: - capture_exception(span=span, exception=exception, timestamp=time_ns()) + except Exception as telemetry_exception: + logger.debug("Failed to capture telemetry for stream: %s", telemetry_exception) finally: _close_span() @@ -2171,6 +2197,12 @@ async def _get_response() -> ChatResponse: ) except Exception as exception: capture_exception(span=span, exception=exception, timestamp=time_ns()) + _capture_operation_error( + attributes=attributes, + exception=exception, + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=perf_counter() - start_time_stamp, + ) raise duration = perf_counter() - start_time_stamp response_attributes = _get_response_attributes(attributes, response) @@ -2256,6 +2288,12 @@ async def get_embeddings( ) except Exception as exception: capture_exception(span=span, exception=exception, timestamp=time_ns()) + _capture_operation_error( + attributes=attributes, + exception=exception, + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=perf_counter() - start_time_stamp, + ) raise duration = perf_counter() - start_time_stamp response_attributes: dict[str, Any] = {**attributes} @@ -3517,6 +3555,30 @@ def _get_response_attributes( ) +def _filter_metric_attributes(attributes: dict[str, Any]) -> dict[str, Any]: + """Filter attributes to only the GenAI metric attribute set.""" + return {key: attributes[key] for key in GEN_AI_METRIC_ATTRIBUTES if key in attributes} + + +def _capture_operation_error( + attributes: dict[str, Any], + exception: BaseException, + operation_duration_histogram: metrics.Histogram | None = None, + duration: float | None = None, +) -> None: + """Record the operation duration metric for a call that failed. + + The GenAI semantic conventions define ``gen_ai.client.operation.duration`` for failed + operations as well as successful ones, with ``error.type`` set to the class of the error. + Recording only successes leaves error latency out of the metric and gives no error rate. + """ + if operation_duration_histogram is None or duration is None: + return + attrs = _filter_metric_attributes(attributes) + attrs[OtelAttr.ERROR_TYPE] = type(exception).__name__ + operation_duration_histogram.record(duration, attributes=attrs) + + def _capture_response( span: trace.Span, attributes: dict[str, Any], @@ -3526,7 +3588,7 @@ def _capture_response( ) -> None: """Set the response for a given span.""" span.set_attributes(attributes) - attrs: dict[str, Any] = {k: v for k, v in attributes.items() if k in GEN_AI_METRIC_ATTRIBUTES} + attrs = _filter_metric_attributes(attributes) if token_usage_histogram and (input_tokens := attributes.get(OtelAttr.INPUT_TOKENS)) is not None: token_usage_histogram.record(input_tokens, attributes={**attrs, OtelAttr.T_TYPE: OtelAttr.T_TYPE_INPUT}) if token_usage_histogram and (output_tokens := attributes.get(OtelAttr.OUTPUT_TOKENS)) is not None: diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py index f496e44d07f..0dcafce37f8 100644 --- a/python/packages/core/tests/core/test_observability.py +++ b/python/packages/core/tests/core/test_observability.py @@ -5723,6 +5723,254 @@ def test_capture_response_with_error_type(span_exporter: InMemorySpanExporter): assert spans[0].attributes.get(OtelAttr.ERROR_TYPE) == "ValueError" # type: ignore[union-attr] # ty: ignore[unresolved-attribute] +def test_filter_metric_attributes() -> None: + """_filter_metric_attributes preserves only the allowed GenAI metric attributes.""" + from agent_framework.observability import OtelAttr, _filter_metric_attributes + + attributes = { + OtelAttr.OPERATION: "chat", + OtelAttr.PROVIDER_NAME: "test_provider", + OtelAttr.SYSTEM: "test_system", + OtelAttr.REQUEST_MODEL: "gpt-4o", + OtelAttr.RESPONSE_MODEL: "gpt-4o-mini", + OtelAttr.ADDRESS: "127.0.0.1", + OtelAttr.PORT: 8000, + "custom_attribute": "ignored", + OtelAttr.CONVERSATION_ID: "conv-123", + } + filtered = _filter_metric_attributes(attributes) + assert filtered == { + OtelAttr.OPERATION: "chat", + OtelAttr.PROVIDER_NAME: "test_provider", + OtelAttr.SYSTEM: "test_system", + OtelAttr.REQUEST_MODEL: "gpt-4o", + OtelAttr.RESPONSE_MODEL: "gpt-4o-mini", + OtelAttr.ADDRESS: "127.0.0.1", + OtelAttr.PORT: 8000, + } + + +def test_capture_operation_error_keeps_only_metric_attributes() -> None: + """The error record carries the metric attribute set plus error.type, nothing else.""" + from agent_framework.observability import _capture_operation_error + + histogram = Mock() + _capture_operation_error( + attributes={ + OtelAttr.OPERATION: OtelAttr.CHAT_COMPLETION_OPERATION, + OtelAttr.REQUEST_MODEL: "test-model", + OtelAttr.CONVERSATION_ID: "conv-1", + }, + exception=TimeoutError("slow"), + operation_duration_histogram=histogram, + duration=0.25, + ) + + histogram.record.assert_called_once_with( + 0.25, + attributes={ + OtelAttr.OPERATION: OtelAttr.CHAT_COMPLETION_OPERATION, + OtelAttr.REQUEST_MODEL: "test-model", + OtelAttr.ERROR_TYPE: "TimeoutError", + }, + ) + + +def test_capture_operation_error_without_histogram_or_duration() -> None: + """No histogram or no duration means nothing is recorded.""" + from agent_framework.observability import _capture_operation_error + + histogram = Mock() + _capture_operation_error(attributes={}, exception=ValueError("x"), operation_duration_histogram=histogram) + _capture_operation_error(attributes={}, exception=ValueError("x"), duration=1.0) + histogram.record.assert_not_called() + + +async def test_chat_client_records_duration_on_error( + mock_chat_client: Any, span_exporter: InMemorySpanExporter +) -> None: + """A failed non-streaming call records gen_ai.client.operation.duration with error.type.""" + + class FailingChatClient(mock_chat_client): # type: ignore[misc, valid-type] + async def _get_non_streaming_response(self, **kwargs: Any) -> ChatResponse: + raise ValueError("boom") + + client = FailingChatClient() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(ValueError, match="boom"): + await client.get_response(messages=[Message(role="user", contents=["hi"])], options={"model": "Test"}) + + histogram.record.assert_called_once() + duration, kwargs = histogram.record.call_args[0][0], histogram.record.call_args[1] + assert duration >= 0 + attributes = kwargs["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert attributes[OtelAttr.REQUEST_MODEL] == "Test" + assert attributes[OtelAttr.OPERATION] == OtelAttr.CHAT_COMPLETION_OPERATION + + +async def test_chat_client_records_duration_on_streaming_error( + mock_chat_client: Any, span_exporter: InMemorySpanExporter +) -> None: + """A stream that fails mid-iteration also records the duration metric with error.type.""" + + class FailingStreamChatClient(mock_chat_client): # type: ignore[misc, valid-type] + def _get_streaming_response(self, **kwargs: Any) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + async def _stream() -> AsyncIterable[ChatResponseUpdate]: + yield ChatResponseUpdate(contents=[Content.from_text("Hello")], role="assistant") + raise RuntimeError("stream broke") + + return ResponseStream(_stream(), finalizer=lambda updates: ChatResponse.from_updates(updates)) + + client = FailingStreamChatClient() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(RuntimeError, match="stream broke"): + async for _ in client.get_response( + messages=[Message(role="user", contents=["hi"])], stream=True, options={"model": "Test"} + ): + pass + + histogram.record.assert_called_once() + attributes = histogram.record.call_args[1]["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "RuntimeError" + assert attributes[OtelAttr.REQUEST_MODEL] == "Test" + + +async def test_chat_client_records_duration_when_stream_setup_fails( + mock_chat_client: Any, span_exporter: InMemorySpanExporter +) -> None: + """A stream that fails before it is built records the duration metric with error.type.""" + + class FailingSetupChatClient(mock_chat_client): # type: ignore[misc, valid-type] + def _get_streaming_response(self, **kwargs: Any) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + raise RuntimeError("setup broke") + + client = FailingSetupChatClient() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(RuntimeError, match="setup broke"): + async for _ in client.get_response( + messages=[Message(role="user", contents=["hi"])], stream=True, options={"model": "Test"} + ): + pass + + histogram.record.assert_called_once() + duration, kwargs = histogram.record.call_args[0][0], histogram.record.call_args[1] + assert duration >= 0 + attributes = kwargs["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "RuntimeError" + assert attributes[OtelAttr.REQUEST_MODEL] == "Test" + assert attributes[OtelAttr.OPERATION] == OtelAttr.CHAT_COMPLETION_OPERATION + + +async def test_chat_client_records_duration_when_stream_finalizer_fails( + mock_chat_client: Any, span_exporter: InMemorySpanExporter +) -> None: + """A stream whose finalizer raises records the duration metric with error.type.""" + + class FailingFinalizerChatClient(mock_chat_client): # type: ignore[misc, valid-type] + def _get_streaming_response(self, **kwargs: Any) -> ResponseStream[ChatResponseUpdate, ChatResponse]: + async def _stream() -> AsyncIterable[ChatResponseUpdate]: + yield ChatResponseUpdate(contents=[Content.from_text("Hello")], role="assistant") + + def _broken_finalizer(updates: Any) -> ChatResponse: + raise ValueError("finalizer failed") + + return ResponseStream(_stream(), finalizer=_broken_finalizer) + + client = FailingFinalizerChatClient() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(ValueError, match="finalizer failed"): + stream = client.get_response( + messages=[Message(role="user", contents=["hi"])], stream=True, options={"model": "Test"} + ) + async for _ in stream: + pass + + histogram.record.assert_called_once() + duration, kwargs = histogram.record.call_args[0][0], histogram.record.call_args[1] + assert duration >= 0 + attributes = kwargs["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert attributes[OtelAttr.REQUEST_MODEL] == "Test" + assert attributes[OtelAttr.OPERATION] == OtelAttr.CHAT_COMPLETION_OPERATION + + +async def test_chat_client_records_duration_when_stream_result_hook_fails( + mock_chat_client: Any, span_exporter: InMemorySpanExporter +) -> None: + """A stream whose result hook raises records the duration metric with error.type.""" + + def _broken_result_hook(response: ChatResponse) -> ChatResponse: + raise RuntimeError("result hook failed") + + client = mock_chat_client() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(RuntimeError, match="result hook failed"): + stream = client.get_response( + messages=[Message(role="user", contents=["hi"])], stream=True, options={"model": "Test"} + ) + stream.with_result_hook(_broken_result_hook) + async for _ in stream: + pass + + histogram.record.assert_called_once() + duration, kwargs = histogram.record.call_args[0][0], histogram.record.call_args[1] + assert duration >= 0 + attributes = kwargs["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "RuntimeError" + assert attributes[OtelAttr.REQUEST_MODEL] == "Test" + assert attributes[OtelAttr.OPERATION] == OtelAttr.CHAT_COMPLETION_OPERATION + + +async def test_embedding_client_records_duration_on_error(span_exporter: InMemorySpanExporter) -> None: + """A failed embedding call records gen_ai.client.operation.duration with error.type.""" + from agent_framework import BaseEmbeddingClient, GeneratedEmbeddings + from agent_framework.observability import EmbeddingTelemetryLayer + + class RawFailingEmbeddingClient(BaseEmbeddingClient[str, list[float], Any]): # type: ignore[type-arg] + async def get_embeddings( + self, values: Sequence[str], *, options: Any = None + ) -> GeneratedEmbeddings[list[float], Any]: + raise ValueError("embed boom") + + class FailingEmbeddingClient(EmbeddingTelemetryLayer, RawFailingEmbeddingClient): # type: ignore[misc] + OTEL_PROVIDER_NAME = "test" + + def service_url(self) -> str: + return "https://test.example.com" + + client = FailingEmbeddingClient() + histogram = Mock() + client.duration_histogram = histogram + span_exporter.clear() + + with pytest.raises(ValueError, match="embed boom"): + await client.get_embeddings(["hi"], options={"model": "test-embed"}) + + histogram.record.assert_called_once() + duration, kwargs = histogram.record.call_args[0][0], histogram.record.call_args[1] + assert duration >= 0 + attributes = kwargs["attributes"] + assert attributes[OtelAttr.ERROR_TYPE] == "ValueError" + assert attributes[OtelAttr.REQUEST_MODEL] == "test-embed" + assert attributes[OtelAttr.OPERATION] == OtelAttr.EMBEDDING_OPERATION + + def test_backfill_request_model_when_unknown(span_exporter: InMemorySpanExporter): """_backfill_request_model updates the span name and REQUEST_MODEL attribute when unknown.""" from agent_framework.observability import OtelAttr, get_tracer