From a29a82f96f97298315fb067a86f384e61141eb79 Mon Sep 17 00:00:00 2001 From: Venkat Ramachandran <268347452+venkat-uk@users.noreply.github.com> Date: Sat, 22 Aug 2026 13:50:28 +0100 Subject: [PATCH] Python: record gen_ai.client.operation.duration on failed calls 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. ChatTelemetryLayer and EmbeddingTelemetryLayer only recorded the histogram on the success path, so error latency never reached the metric and the metric carried no error rate. _capture_response already had the branch that copies error.type into the metric attributes, but no caller ever put error.type into the attribute dict, so that branch was unreachable in production. Add _capture_operation_error and call it from the three chat and embedding failure paths: the streaming setup failure, the mid-stream failure, and the non-streaming failure. The metric attribute set is filtered through GEN_AI_METRIC_ATTRIBUTES, the same filter the success path uses, so no high-cardinality attribute reaches the metric. --- .../core/agent_framework/observability.py | 43 +++++++++ .../core/tests/core/test_observability.py | 92 +++++++++++++++++++ 2 files changed, 135 insertions(+) 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