diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index eaeab6eade..0d20502063 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -1709,6 +1709,12 @@ 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 @@ -1725,6 +1731,12 @@ async def _finalize_stream() -> None: exception=result_stream._stream_error, # type: ignore timestamp=time_ns(), ) + _capture_operation_error( + attributes=attributes, + exception=result_stream._stream_error, # type: ignore[arg-type] + operation_duration_histogram=getattr(self, "duration_histogram", None), + duration=duration_state.get("duration"), + ) return response: ChatResponse[Any] = await result_stream.get_final_response() duration = duration_state.get("duration") @@ -1816,6 +1828,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) @@ -1901,6 +1919,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} @@ -3162,6 +3186,25 @@ def _get_response_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: dict[str, Any] = {k: v for k, v in attributes.items() if k in GEN_AI_METRIC_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], diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py index 1e767e2d1f..15dd5d2e8c 100644 --- a/python/packages/core/tests/core/test_observability.py +++ b/python/packages/core/tests/core/test_observability.py @@ -5355,6 +5355,98 @@ 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_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" + + 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