From e9f6eb53ffd5d9dee490b5dd1391692b6653ba9b Mon Sep 17 00:00:00 2001 From: Doron Kopit Date: Thu, 5 Nov 2026 15:09:38 +0200 Subject: [PATCH 1/3] fix(openai): fix(openai): record duration and token metrics for the Responses API Fixes #4585 --- .../instrumentation/openai/utils.py | 18 ++ .../instrumentation/openai/v1/__init__.py | 12 +- .../openai/v1/responses_wrappers.py | 104 +++++++++- .../tests/metrics/test_openai_metrics.py | 190 +++++++++++++++++- 4 files changed, 312 insertions(+), 12 deletions(-) diff --git a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/utils.py b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/utils.py index 338550693b..cf0f92d9f5 100644 --- a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/utils.py +++ b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/utils.py @@ -129,6 +129,24 @@ def wrapper(wrapped, instance, args, kwargs): return _with_chat_telemetry +def _with_responses_telemetry_wrapper(func): + def _with_responses_telemetry(tracer, token_histogram=None, duration_histogram=None): + def wrapper(wrapped, instance, args, kwargs): + return func( + tracer, + token_histogram, + duration_histogram, + wrapped, + instance, + args, + kwargs, + ) + + return wrapper + + return _with_responses_telemetry + + def _with_tracer_wrapper(func): def _with_tracer(tracer): def wrapper(wrapped, instance, args, kwargs): diff --git a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/__init__.py b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/__init__.py index e7e3deb56b..7c14fa7d92 100644 --- a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/__init__.py +++ b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/__init__.py @@ -305,17 +305,17 @@ def _instrument(self, **kwargs): self._try_wrap( "openai.resources.responses", "Responses.create", - responses_get_or_create_wrapper(tracer), + responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", "Responses.retrieve", - responses_get_or_create_wrapper(tracer), + responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", "Responses.parse", - responses_get_or_create_wrapper(tracer), + responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", @@ -325,17 +325,17 @@ def _instrument(self, **kwargs): self._try_wrap( "openai.resources.responses", "AsyncResponses.create", - async_responses_get_or_create_wrapper(tracer), + async_responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", "AsyncResponses.retrieve", - async_responses_get_or_create_wrapper(tracer), + async_responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", "AsyncResponses.parse", - async_responses_get_or_create_wrapper(tracer), + async_responses_get_or_create_wrapper(tracer, tokens_histogram, duration_histogram), ) self._try_wrap( "openai.resources.responses", diff --git a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py index 340402ea9b..12caf2bfc5 100644 --- a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py +++ b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py @@ -23,14 +23,16 @@ from opentelemetry.instrumentation.openai.shared import ( _build_tool_def_dict, _extract_model_name_from_provider_format, + _get_openai_base_url, _parse_arguments, _set_request_attributes, _set_span_attribute, _set_tool_definitions_json, + metric_shared_attributes, model_as_dict, ) - from opentelemetry.instrumentation.openai.utils import ( + _with_responses_telemetry_wrapper, _with_tracer_wrapper, dont_throw, json_dumps, @@ -534,9 +536,44 @@ def set_data_attributes(traced_response: TracedData, span: Span): _set_responses_json_messages(traced_response, span) +def _responses_metric_attributes(instance, response_model, is_streaming=False): + return metric_shared_attributes( + response_model=_extract_model_name_from_provider_format(response_model) or None, + operation="chat", + server_address=_get_openai_base_url(instance), + is_streaming=is_streaming, + ) + + +def _error_metric_attributes(instance, request_model, exception, is_streaming=False): + attributes = _responses_metric_attributes(instance, None, is_streaming) + del attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] + if request_model: + attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] = request_model + attributes[ERROR_TYPE] = exception.__class__.__name__ + return attributes + + @dont_throw -@_with_tracer_wrapper -def responses_get_or_create_wrapper(tracer: Tracer, wrapped, instance, args, kwargs): +def _record_duration(duration_histogram, start_time, attributes): + if duration_histogram: + duration_histogram.record((time.time_ns() - start_time) / 1e9, attributes=attributes) + + +@dont_throw +def _record_token_usage(token_histogram, usage, attributes): + if token_histogram and usage: + for token_type, count in (("input", usage.input_tokens), ("output", usage.output_tokens)): + token_histogram.record( + count, attributes={**attributes, GenAIAttributes.GEN_AI_TOKEN_TYPE: token_type} + ) + + +@dont_throw +@_with_responses_telemetry_wrapper +def responses_get_or_create_wrapper( + tracer: Tracer, token_histogram, duration_histogram, wrapped, instance, args, kwargs +): if context_api.get_value(_SUPPRESS_INSTRUMENTATION_KEY): return wrapped(*args, **kwargs) start_time = time.time_ns() @@ -563,8 +600,18 @@ def responses_get_or_create_wrapper(tracer: Tracer, wrapped, instance, args, kwa start_time=start_time, request_kwargs=non_sentinel_kwargs, tracer=tracer, + instance=instance, + token_histogram=token_histogram, + duration_histogram=duration_histogram, ) except Exception as e: + _record_duration( + duration_histogram, + start_time, + _error_metric_attributes( + instance, non_sentinel_kwargs.get("model"), e, bool(non_sentinel_kwargs.get("stream")) + ), + ) response_id = non_sentinel_kwargs.get("response_id") existing_data = {} if response_id and response_id in responses: @@ -708,14 +755,18 @@ def responses_get_or_create_wrapper(tracer: Tracer, wrapped, instance, args, kwa _set_request_attributes(span, prepare_kwargs_for_shared_attributes(non_sentinel_kwargs), instance) set_data_attributes(traced_data, span) span.end() + # One point per emitted span, so background polls and repeat retrieves don't add operations + metric_attributes = _responses_metric_attributes(instance, parsed_response.model) + _record_duration(duration_histogram, traced_data.start_time, metric_attributes) + _record_token_usage(token_histogram, traced_data.usage, metric_attributes) return response @dont_throw -@_with_tracer_wrapper +@_with_responses_telemetry_wrapper async def async_responses_get_or_create_wrapper( - tracer: Tracer, wrapped, instance, args, kwargs + tracer: Tracer, token_histogram, duration_histogram, wrapped, instance, args, kwargs ): if context_api.get_value(_SUPPRESS_INSTRUMENTATION_KEY): return await wrapped(*args, **kwargs) @@ -743,8 +794,18 @@ async def async_responses_get_or_create_wrapper( start_time=start_time, request_kwargs=non_sentinel_kwargs, tracer=tracer, + instance=instance, + token_histogram=token_histogram, + duration_histogram=duration_histogram, ) except Exception as e: + _record_duration( + duration_histogram, + start_time, + _error_metric_attributes( + instance, non_sentinel_kwargs.get("model"), e, bool(non_sentinel_kwargs.get("stream")) + ), + ) response_id = non_sentinel_kwargs.get("response_id") existing_data = {} if response_id and response_id in responses: @@ -886,6 +947,10 @@ async def async_responses_get_or_create_wrapper( _set_request_attributes(span, prepare_kwargs_for_shared_attributes(non_sentinel_kwargs), instance) set_data_attributes(traced_data, span) span.end() + # One point per emitted span, so background polls and repeat retrieves don't add operations + metric_attributes = _responses_metric_attributes(instance, parsed_response.model) + _record_duration(duration_histogram, traced_data.start_time, metric_attributes) + _record_token_usage(token_histogram, traced_data.usage, metric_attributes) return response @@ -960,6 +1025,9 @@ class ResponseStream(ObjectProxy): _request_kwargs = None _tracer = None _traced_data = None + _instance = None + _token_histogram = None + _duration_histogram = None def __init__( self, @@ -969,10 +1037,16 @@ def __init__( request_kwargs=None, tracer=None, traced_data=None, + instance=None, + token_histogram=None, + duration_histogram=None, ): super().__init__(response) self._span = span self._start_time = start_time + self._instance = instance + self._token_histogram = token_histogram + self._duration_histogram = duration_histogram # Filter sentinel values (defensive, in case called directly without prior filtering) self._request_kwargs = _sanitize_sentinel_values(request_kwargs or {}) self._tracer = tracer @@ -1109,6 +1183,20 @@ def _process_chunk(self, chunk): if hasattr(chunk, "response") and chunk.response: self._complete_response_data = chunk.response + def _metric_attributes(self): + return _responses_metric_attributes( + self._instance, + self._traced_data.response_model or self._request_kwargs.get("model"), + is_streaming=True, + ) + + def _record_error_duration(self, exception): + _record_duration( + self._duration_histogram, + self._start_time, + _error_metric_attributes(self._instance, self._request_kwargs.get("model"), exception, True), + ) + @dont_throw def _process_complete_response(self): """Process the complete response and emit span""" @@ -1149,12 +1237,16 @@ def _process_complete_response(self): self._span.set_status(StatusCode.OK) self._span.end() self._cleanup_completed = True + metric_attributes = self._metric_attributes() + _record_duration(self._duration_histogram, self._start_time, metric_attributes) + _record_token_usage(self._token_histogram, self._traced_data.usage, metric_attributes) except Exception as e: if self._span and self._span.is_recording(): self._span.set_attribute(ERROR_TYPE, e.__class__.__name__) self._span.set_status(StatusCode.ERROR, str(e)) self._span.end() + self._record_error_duration(e) self._cleanup_completed = True @dont_throw @@ -1170,6 +1262,7 @@ def _handle_exception(self, exception): self._span.set_status(StatusCode.ERROR, str(exception)) self._span.end() + self._record_error_duration(exception) self._cleanup_completed = True @dont_throw @@ -1185,6 +1278,7 @@ def _ensure_cleanup(self): self._span.set_status(StatusCode.OK) self._span.end() + _record_duration(self._duration_histogram, self._start_time, self._metric_attributes()) self._cleanup_completed = True except Exception: diff --git a/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py b/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py index efa7f7f018..0a063b27e6 100644 --- a/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py +++ b/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py @@ -1,11 +1,16 @@ +import inspect +import json +import time +import httpx import pytest -from openai import OpenAI +from openai import AsyncOpenAI, InternalServerError, OpenAI from opentelemetry.semconv._incubating.attributes import ( gen_ai_attributes as GenAIAttributes, ) from opentelemetry.semconv._incubating.metrics import ( gen_ai_metrics as GenAIMetrics, ) +from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE from opentelemetry.semconv_ai import Meters from pydantic import BaseModel @@ -309,3 +314,186 @@ def test_image_gen_metrics(instrument_legacy, reader, openai_client): assert len(data_point.attributes["server.address"]) > 0 assert found_duration_metric + + +_RESPONSES_BODY = { + "object": "response", + "created_at": 0, + "status": "completed", + "model": "gpt-4o-mini", + "output": [{ + "type": "message", "id": "msg_1", "status": "completed", "role": "assistant", + "content": [{"type": "output_text", "text": "hi", "annotations": []}], + }], + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": [], + "usage": { + "input_tokens": 7, "output_tokens": 3, "total_tokens": 10, + "input_tokens_details": {"cached_tokens": 0}, "output_tokens_details": {"reasoning_tokens": 0}, + }, +} + + +def _responses_ok(request): + # A unique id per call, as already-completed ids are not traced again + body = {**_RESPONSES_BODY, "id": f"resp_metrics_{time.time_ns()}"} + if json.loads(request.content).get("stream"): + event = {"type": "response.completed", "sequence_number": 0, "response": body} + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=f"event: response.completed\ndata: {json.dumps(event)}\n\n", + ) + return httpx.Response(200, json=body) + + +def _responses_server_error(request): + return httpx.Response(500, json={"error": {"message": "boom"}}) + + +class _BrokenStream(httpx.SyncByteStream, httpx.AsyncByteStream): + def __iter__(self): + yield from () + raise httpx.ReadError("stream broke") + + async def __aiter__(self): + for chunk in self: + yield chunk + + +def _responses_broken_stream(request): + return httpx.Response(200, headers={"content-type": "text/event-stream"}, stream=_BrokenStream()) + + +def _background_handler(response_id): + def handler(request): + status = "queued" if request.method == "POST" else "completed" + return httpx.Response(200, json={**_RESPONSES_BODY, "id": response_id, "status": status}) + + return handler + + +def _client(is_async, handler): + transport = httpx.MockTransport(handler) + if is_async: + return AsyncOpenAI(api_key="test", max_retries=0, http_client=httpx.AsyncClient(transport=transport)) + return OpenAI(api_key="test", max_retries=0, http_client=httpx.Client(transport=transport)) + + +async def _result(value): + return await value if inspect.isawaitable(value) else value + + +async def _call_responses(is_async, handler, method="create", stream=False): + client = _client(is_async, handler) + kwargs = {"stream": True} if stream else {} + response = await _result(getattr(client.responses, method)(model="gpt-4o-mini", input="hello", **kwargs)) + if stream and is_async: + async for _ in response: + pass + elif stream: + for _ in response: + pass + return client + + +def _assert_shared_attributes(attributes, client, stream): + assert attributes[GenAIAttributes.GEN_AI_OPERATION_NAME] == "chat" + assert attributes[GenAIAttributes.GEN_AI_PROVIDER_NAME] == "openai" + assert attributes["server.address"] == str(client.base_url) + assert attributes["stream"] is stream + + +def _collect_points(reader): + data = reader.get_metrics_data() + points = {} + for rm in data.resource_metrics if data else []: + for sm in rm.scope_metrics: + for metric in sm.metrics: + points.setdefault(metric.name, []).extend(metric.data.data_points) + return points + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method,stream", [("create", False), ("create", True), ("parse", False)]) +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_metrics(instrument_legacy, reader, is_async, method, stream): + _collect_points(reader) + + client = await _call_responses(is_async, _responses_ok, method, stream) + + points = _collect_points(reader) + durations = points.get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + assert durations[0].sum > 0 + assert durations[0].attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == "gpt-4o-mini" + _assert_shared_attributes(durations[0].attributes, client, stream) + tokens = points.get(Meters.LLM_TOKEN_USAGE, []) + assert {p.attributes[GenAIAttributes.GEN_AI_TOKEN_TYPE]: p.sum for p in tokens} == {"input": 7, "output": 3} + for point in tokens: + assert point.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == "gpt-4o-mini" + _assert_shared_attributes(point.attributes, client, stream) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "handler,stream,error", + [ + (_responses_server_error, False, InternalServerError), + (_responses_server_error, True, InternalServerError), + (_responses_broken_stream, True, httpx.ReadError), + ], +) +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_error_metrics(instrument_legacy, reader, is_async, handler, stream, error): + _collect_points(reader) + + with pytest.raises(error): + await _call_responses(is_async, handler, stream=stream) + + points = _collect_points(reader) + durations = points.get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + attributes = durations[0].attributes + assert attributes[ERROR_TYPE] == error.__name__ + assert attributes[GenAIAttributes.GEN_AI_REQUEST_MODEL] == "gpt-4o-mini" + _assert_shared_attributes(attributes, _client(is_async, handler), stream) + assert Meters.LLM_TOKEN_USAGE not in points + + +@pytest.mark.asyncio +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_stream_closed_early_records_duration(instrument_legacy, reader, is_async): + _collect_points(reader) + client = _client(is_async, _responses_ok) + + response = await _result(client.responses.create(model="gpt-4o-mini", input="hello", stream=True)) + await _result(response.aclose() if is_async else response.close()) + + durations = _collect_points(reader).get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + assert ERROR_TYPE not in durations[0].attributes + _assert_shared_attributes(durations[0].attributes, client, True) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_background_retrieve_metrics(instrument_legacy, reader, is_async): + _collect_points(reader) + response_id = f"resp_background_{time.time_ns()}" + client = _client(is_async, _background_handler(response_id)) + + await _result(client.responses.create(model="gpt-4o-mini", input="hello", background=True)) + assert Meters.LLM_OPERATION_DURATION not in _collect_points(reader) + + await _result(client.responses.retrieve(response_id)) + points = _collect_points(reader) + durations = points.get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + _assert_shared_attributes(durations[0].attributes, client, False) + assert sorted(p.sum for p in points.get(Meters.LLM_TOKEN_USAGE, [])) == [3, 7] + + # Retrieving an already-completed response emits no span, so no new operation either + await _result(client.responses.retrieve(response_id)) + assert _collect_points(reader) == {} From bc2f0d77f30d9873709d20b7d77998b5f0719681 Mon Sep 17 00:00:00 2001 From: Doron Kopit Date: Mon, 9 Nov 2026 13:46:47 +0200 Subject: [PATCH 2/3] fix(openai): record failed status and close-time tokens for responses Addresses review comments: https://github.com/traceloop/openllmetry/pull/4697#discussion_r4480878712 https://github.com/traceloop/openllmetry/pull/4697#discussion_r4480878720 Fixes #4585 --- .../openai/v1/responses_wrappers.py | 23 ++++- .../tests/metrics/test_openai_metrics.py | 89 +++++++++++++++++++ 2 files changed, 108 insertions(+), 4 deletions(-) diff --git a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py index 12caf2bfc5..9da1989084 100644 --- a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py +++ b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/v1/responses_wrappers.py @@ -536,13 +536,17 @@ def set_data_attributes(traced_response: TracedData, span: Span): _set_responses_json_messages(traced_response, span) -def _responses_metric_attributes(instance, response_model, is_streaming=False): - return metric_shared_attributes( +def _responses_metric_attributes(instance, response_model, is_streaming=False, response=None): + attributes = metric_shared_attributes( response_model=_extract_model_name_from_provider_format(response_model) or None, operation="chat", server_address=_get_openai_base_url(instance), is_streaming=is_streaming, ) + # A failed generation comes back as a terminal response rather than a raised exception + if getattr(response, "status", None) == "failed": + attributes[ERROR_TYPE] = getattr(getattr(response, "error", None), "code", None) or "failed" + return attributes def _error_metric_attributes(instance, request_model, exception, is_streaming=False): @@ -563,6 +567,8 @@ def _record_duration(duration_histogram, start_time, attributes): @dont_throw def _record_token_usage(token_histogram, usage, attributes): if token_histogram and usage: + # error.type describes the operation, not its token usage + attributes = {key: value for key, value in attributes.items() if key != ERROR_TYPE} for token_type, count in (("input", usage.input_tokens), ("output", usage.output_tokens)): token_histogram.record( count, attributes={**attributes, GenAIAttributes.GEN_AI_TOKEN_TYPE: token_type} @@ -756,7 +762,7 @@ def responses_get_or_create_wrapper( set_data_attributes(traced_data, span) span.end() # One point per emitted span, so background polls and repeat retrieves don't add operations - metric_attributes = _responses_metric_attributes(instance, parsed_response.model) + metric_attributes = _responses_metric_attributes(instance, parsed_response.model, response=parsed_response) _record_duration(duration_histogram, traced_data.start_time, metric_attributes) _record_token_usage(token_histogram, traced_data.usage, metric_attributes) @@ -948,7 +954,7 @@ async def async_responses_get_or_create_wrapper( set_data_attributes(traced_data, span) span.end() # One point per emitted span, so background polls and repeat retrieves don't add operations - metric_attributes = _responses_metric_attributes(instance, parsed_response.model) + metric_attributes = _responses_metric_attributes(instance, parsed_response.model, response=parsed_response) _record_duration(duration_histogram, traced_data.start_time, metric_attributes) _record_token_usage(token_histogram, traced_data.usage, metric_attributes) @@ -1121,12 +1127,20 @@ async def __aexit__(self, exc_type, exc_val, exc_tb): suppress = bool(await self.__wrapped__.__aexit__(exc_type, exc_val, exc_tb)) return suppress + def _has_terminal_response(self): + return getattr(self._complete_response_data, "status", None) in _TERMINAL_STATUSES + def close(self): + # A terminal response already seen still gets its full span and token usage + if self._has_terminal_response(): + self._process_complete_response() self._ensure_cleanup() if hasattr(self.__wrapped__, "close"): return self.__wrapped__.close() async def aclose(self): + if self._has_terminal_response(): + self._process_complete_response() self._ensure_cleanup() if hasattr(self.__wrapped__, "aclose"): return await self.__wrapped__.aclose() @@ -1188,6 +1202,7 @@ def _metric_attributes(self): self._instance, self._traced_data.response_model or self._request_kwargs.get("model"), is_streaming=True, + response=self._complete_response_data, ) def _record_error_duration(self, exception): diff --git a/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py b/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py index 0a063b27e6..032bdc7e45 100644 --- a/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py +++ b/packages/opentelemetry-instrumentation-openai/tests/metrics/test_openai_metrics.py @@ -348,6 +348,35 @@ def _responses_ok(request): return httpx.Response(200, json=body) +def _responses_failed(error): + def handler(request): + body = {**_RESPONSES_BODY, "id": f"resp_failed_{time.time_ns()}", "status": "failed", "error": error} + if json.loads(request.content).get("stream"): + event = {"type": "response.failed", "sequence_number": 0, "response": body} + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=f"event: response.failed\ndata: {json.dumps(event)}\n\n", + ) + return httpx.Response(200, json=body) + + return handler + + +def _responses_created_then_completed(request): + body = {**_RESPONSES_BODY, "id": f"resp_created_{time.time_ns()}"} + events = [ + # Usage is kept so that treating this snapshot as the final response would show in the metrics + ("response.created", {**body, "status": "in_progress", "output": []}), + ("response.completed", body), + ] + content = "".join( + f"event: {name}\ndata: {json.dumps({'type': name, 'sequence_number': i, 'response': response})}\n\n" + for i, (name, response) in enumerate(events) + ) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=content) + + def _responses_server_error(request): return httpx.Response(500, json={"error": {"message": "boom"}}) @@ -462,6 +491,66 @@ async def test_responses_error_metrics(instrument_legacy, reader, is_async, hand assert Meters.LLM_TOKEN_USAGE not in points +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error,error_type", + [({"code": "server_error", "message": "boom"}, "server_error"), (None, "failed")], +) +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_failed_status_metrics(instrument_legacy, reader, is_async, stream, error, error_type): + _collect_points(reader) + + client = await _call_responses(is_async, _responses_failed(error), stream=stream) + + points = _collect_points(reader) + durations = points.get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + assert durations[0].attributes[ERROR_TYPE] == error_type + _assert_shared_attributes(durations[0].attributes, client, stream) + # error.type is not a token usage attribute + tokens = points.get(Meters.LLM_TOKEN_USAGE, []) + assert {p.attributes[GenAIAttributes.GEN_AI_TOKEN_TYPE]: p.sum for p in tokens} == {"input": 7, "output": 3} + for point in tokens: + assert ERROR_TYPE not in point.attributes + _assert_shared_attributes(point.attributes, client, stream) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_stream_closed_after_created_records_duration_only(instrument_legacy, reader, is_async): + _collect_points(reader) + client = _client(is_async, _responses_created_then_completed) + + response = await _result(client.responses.create(model="gpt-4o-mini", input="hello", stream=True)) + await _result(response.__anext__() if is_async else next(response)) + await _result(response.aclose() if is_async else response.close()) + + points = _collect_points(reader) + durations = points.get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + assert ERROR_TYPE not in durations[0].attributes + assert Meters.LLM_TOKEN_USAGE not in points + + +@pytest.mark.asyncio +@pytest.mark.parametrize("is_async", [False, True]) +async def test_responses_stream_closed_after_completed_records_tokens(instrument_legacy, reader, is_async): + _collect_points(reader) + client = _client(is_async, _responses_ok) + + response = await _result(client.responses.create(model="gpt-4o-mini", input="hello", stream=True)) + await _result(response.__anext__() if is_async else next(response)) + await _result(response.aclose() if is_async else response.close()) + + points = _collect_points(reader) + durations = points.get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + assert durations[0].attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == "gpt-4o-mini" + tokens = points.get(Meters.LLM_TOKEN_USAGE, []) + assert {p.attributes[GenAIAttributes.GEN_AI_TOKEN_TYPE]: p.sum for p in tokens} == {"input": 7, "output": 3} + + @pytest.mark.asyncio @pytest.mark.parametrize("is_async", [False, True]) async def test_responses_stream_closed_early_records_duration(instrument_legacy, reader, is_async): From fb83bdb7818ddf5865afbdeabe0fcb4d86bb4753 Mon Sep 17 00:00:00 2001 From: Doron Kopit Date: Mon, 9 Nov 2026 15:57:43 +0200 Subject: [PATCH 3/3] fix(openai): add vcr tests for responses stream token metrics Addresses review comments: https://github.com/traceloop/openllmetry/pull/4697#discussion_r4512987632 Fixes #4585 --- ...closed_after_completed_records_tokens.yaml | 137 ++++++++++++++++++ .../metrics/test_responses_stream_metrics.py | 80 ++++++++++ 2 files changed, 217 insertions(+) create mode 100644 packages/opentelemetry-instrumentation-openai/tests/metrics/cassettes/test_responses_stream_metrics/test_responses_stream_closed_after_completed_records_tokens.yaml create mode 100644 packages/opentelemetry-instrumentation-openai/tests/metrics/test_responses_stream_metrics.py diff --git a/packages/opentelemetry-instrumentation-openai/tests/metrics/cassettes/test_responses_stream_metrics/test_responses_stream_closed_after_completed_records_tokens.yaml b/packages/opentelemetry-instrumentation-openai/tests/metrics/cassettes/test_responses_stream_metrics/test_responses_stream_closed_after_completed_records_tokens.yaml new file mode 100644 index 0000000000..5a1f629a0f --- /dev/null +++ b/packages/opentelemetry-instrumentation-openai/tests/metrics/cassettes/test_responses_stream_metrics/test_responses_stream_closed_after_completed_records_tokens.yaml @@ -0,0 +1,137 @@ +interactions: +- request: + body: '{"input":"Say hello in one word.","max_output_tokens":256,"model":"gpt-6-luna","stream":true}' + headers: + Accept: + - application/json + Accept-Encoding: + - gzip, deflate + Connection: + - keep-alive + Content-Length: + - '93' + Content-Type: + - application/json + Host: + - api.openai.com + User-Agent: + - OpenAI/Python 1.99.7 + X-Stainless-Arch: + - arm64 + X-Stainless-Async: + - 'false' + X-Stainless-Lang: + - python + X-Stainless-OS: + - Linux + X-Stainless-Package-Version: + - 1.99.7 + X-Stainless-Runtime: + - CPython + X-Stainless-Runtime-Version: + - 3.11.18 + x-stainless-read-timeout: + - '600' + x-stainless-retry-count: + - '0' + method: POST + uri: https://api.openai.com/v1/responses + response: + body: + string: 'event: response.created + + data: {"type":"response.created","response":{"id":"resp_035cf543467a72f3016af1d0b75b8c87d2b503d8591608e1b6","object":"response","created_at":1794232503,"status":"in_progress","access_programs":null,"background":false,"completed_at":null,"error":null,"frequency_penalty":0.0,"image":{"context_compression":"disabled"},"incomplete_details":null,"instructions":null,"max_output_tokens":256,"max_tool_calls":null,"model":"gpt-6-luna","moderation":null,"output":[],"parallel_tool_calls":true,"presence_penalty":0.0,"previous_response_id":null,"prompt_cache_key":null,"prompt_cache_retention":"24h","reasoning":{"context":"all_turns","effort":"medium","mode":"standard","summary":null},"safety_identifier":null,"service_tier":"auto","store":false,"temperature":1.0,"text":{"format":{"type":"text"},"verbosity":"medium"},"tool_choice":"auto","tool_usage":{"image_gen":{"input_tokens":0,"input_tokens_details":{"image_tokens":0,"text_tokens":0},"output_tokens":0,"output_tokens_details":{"image_tokens":0,"text_tokens":0},"total_tokens":0},"web_search":{"num_requests":0}},"tools":[],"top_logprobs":0,"top_p":0.98,"truncation":"disabled","usage":null,"user":null,"metadata":{}},"sequence_number":0} + + + event: response.in_progress + + data: {"type":"response.in_progress","response":{"id":"resp_035cf543467a72f3016af1d0b75b8c87d2b503d8591608e1b6","object":"response","created_at":1794232503,"status":"in_progress","access_programs":null,"background":false,"completed_at":null,"error":null,"frequency_penalty":0.0,"image":{"context_compression":"disabled"},"incomplete_details":null,"instructions":null,"max_output_tokens":256,"max_tool_calls":null,"model":"gpt-6-luna","moderation":null,"output":[],"parallel_tool_calls":true,"presence_penalty":0.0,"previous_response_id":null,"prompt_cache_key":null,"prompt_cache_retention":"24h","reasoning":{"context":"all_turns","effort":"medium","mode":"standard","summary":null},"safety_identifier":null,"service_tier":"auto","store":false,"temperature":1.0,"text":{"format":{"type":"text"},"verbosity":"medium"},"tool_choice":"auto","tool_usage":{"image_gen":{"input_tokens":0,"input_tokens_details":{"image_tokens":0,"text_tokens":0},"output_tokens":0,"output_tokens_details":{"image_tokens":0,"text_tokens":0},"total_tokens":0},"web_search":{"num_requests":0}},"tools":[],"top_logprobs":0,"top_p":0.98,"truncation":"disabled","usage":null,"user":null,"metadata":{}},"sequence_number":1} + + + event: response.output_item.added + + data: {"type":"response.output_item.added","item":{"id":"msg_035cf543467a72f3016af1d0b8a81487d2a9adfb39f6708397","type":"message","status":"in_progress","content":[],"phase":"final_answer","role":"assistant"},"output_index":0,"sequence_number":2} + + + event: response.content_part.added + + data: {"type":"response.content_part.added","content_index":0,"item_id":"msg_035cf543467a72f3016af1d0b8a81487d2a9adfb39f6708397","output_index":0,"part":{"type":"output_text","annotations":[],"logprobs":[],"text":""},"sequence_number":3} + + + event: response.output_text.delta + + data: {"type":"response.output_text.delta","content_index":0,"delta":"Hello","item_id":"msg_035cf543467a72f3016af1d0b8a81487d2a9adfb39f6708397","logprobs":[],"obfuscation":"xfC6AVgyRt0","output_index":0,"sequence_number":4} + + + event: response.output_text.done + + data: {"type":"response.output_text.done","content_index":0,"item_id":"msg_035cf543467a72f3016af1d0b8a81487d2a9adfb39f6708397","logprobs":[],"output_index":0,"sequence_number":5,"text":"Hello"} + + + event: response.content_part.done + + data: {"type":"response.content_part.done","content_index":0,"item_id":"msg_035cf543467a72f3016af1d0b8a81487d2a9adfb39f6708397","output_index":0,"part":{"type":"output_text","annotations":[],"logprobs":[],"text":"Hello"},"sequence_number":6} + + + event: response.output_item.done + + data: {"type":"response.output_item.done","item":{"id":"msg_035cf543467a72f3016af1d0b8a81487d2a9adfb39f6708397","type":"message","status":"completed","content":[{"type":"output_text","annotations":[],"logprobs":[],"text":"Hello"}],"phase":"final_answer","role":"assistant"},"output_index":0,"sequence_number":7} + + + event: response.completed + + data: {"type":"response.completed","response":{"id":"resp_035cf543467a72f3016af1d0b75b8c87d2b503d8591608e1b6","object":"response","created_at":1794232503,"status":"completed","access_programs":null,"background":false,"completed_at":1794232504,"error":null,"frequency_penalty":0.0,"image":{"context_compression":"disabled"},"incomplete_details":null,"instructions":null,"max_output_tokens":256,"max_tool_calls":null,"model":"gpt-6-luna","moderation":null,"output":[{"id":"msg_035cf543467a72f3016af1d0b8a81487d2a9adfb39f6708397","type":"message","status":"completed","content":[{"type":"output_text","annotations":[],"logprobs":[],"text":"Hello"}],"phase":"final_answer","role":"assistant"}],"parallel_tool_calls":true,"presence_penalty":0.0,"previous_response_id":null,"prompt_cache_key":null,"prompt_cache_retention":"24h","reasoning":{"context":"all_turns","effort":"medium","mode":"standard","summary":null},"safety_identifier":null,"service_tier":"default","store":false,"temperature":1.0,"text":{"format":{"type":"text"},"verbosity":"medium"},"tool_choice":"auto","tool_usage":{"image_gen":{"input_tokens":0,"input_tokens_details":{"image_tokens":0,"text_tokens":0},"output_tokens":0,"output_tokens_details":{"image_tokens":0,"text_tokens":0},"total_tokens":0},"web_search":{"num_requests":0}},"tools":[],"top_logprobs":0,"top_p":0.98,"truncation":"disabled","usage":{"input_tokens":12,"input_tokens_details":{"cache_write_tokens":0,"cached_tokens":0},"output_tokens":5,"output_tokens_details":{"reasoning_tokens":0},"total_tokens":17},"user":null,"metadata":{}},"sequence_number":8} + + + ' + headers: + access-control-expose-headers: + - X-Request-ID, Server-Timing-Total + - CF-Ray + - CF-Ray + alt-svc: + - h3=":443"; ma=86400 + cf-cache-status: + - DYNAMIC + cf-ray: + - redacted + content-type: + - text/event-stream; charset=utf-8 + date: + - Mon, 09 Nov 2026 13:55:04 GMT + openai-organization: + - redacted + openai-processing-ms: + - '985' + openai-project: + - redacted + openai-version: + - '2020-10-01' + server: + - cloudflare + set-cookie: + - redacted + strict-transport-security: + - max-age=31536000; includeSubDomains; preload + transfer-encoding: + - chunked + x-content-type-options: + - nosniff + x-ratelimit-limit-requests: + - '30000' + x-ratelimit-limit-tokens: + - '180000000' + x-ratelimit-remaining-requests: + - '29999' + x-ratelimit-remaining-tokens: + - '179999844' + x-ratelimit-reset-requests: + - 2ms + x-ratelimit-reset-tokens: + - 0s + x-request-id: + - redacted + status: + code: 200 + message: OK +version: 1 diff --git a/packages/opentelemetry-instrumentation-openai/tests/metrics/test_responses_stream_metrics.py b/packages/opentelemetry-instrumentation-openai/tests/metrics/test_responses_stream_metrics.py new file mode 100644 index 0000000000..17261663a3 --- /dev/null +++ b/packages/opentelemetry-instrumentation-openai/tests/metrics/test_responses_stream_metrics.py @@ -0,0 +1,80 @@ +from pathlib import Path + +import pytest +from opentelemetry.semconv._incubating.attributes import ( + gen_ai_attributes as GenAIAttributes, +) +from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE +from opentelemetry.semconv_ai import Meters + +# The async test replays the same recorded call, as the stream is identical +_CASSETTE = str( + Path(__file__).parent + / "cassettes" + / "test_responses_stream_metrics" + / "test_responses_stream_closed_after_completed_records_tokens.yaml" +) + + +def _collect_points(reader): + data = reader.get_metrics_data() + points = {} + for rm in data.resource_metrics if data else []: + for sm in rm.scope_metrics: + for metric in sm.metrics: + points.setdefault(metric.name, []).extend(metric.data.data_points) + return points + + +def _assert_token_points(points): + durations = points.get(Meters.LLM_OPERATION_DURATION, []) + assert [p.count for p in durations] == [1] + assert ERROR_TYPE not in durations[0].attributes + tokens = points.get(Meters.LLM_TOKEN_USAGE, []) + assert {p.attributes[GenAIAttributes.GEN_AI_TOKEN_TYPE]: p.sum for p in tokens} == {"input": 12, "output": 5} + for point in tokens: + assert point.attributes[GenAIAttributes.GEN_AI_OPERATION_NAME] == "chat" + assert point.attributes[GenAIAttributes.GEN_AI_PROVIDER_NAME] == "openai" + assert point.attributes["stream"] is True + assert point.attributes[GenAIAttributes.GEN_AI_RESPONSE_MODEL] == "gpt-6-luna" + + +@pytest.mark.vcr +def test_responses_stream_closed_after_completed_records_tokens( + instrument_legacy, reader, openai_client +): + _collect_points(reader) + + stream = openai_client.responses.create( + model="gpt-6-luna", + input="Say hello in one word.", + max_output_tokens=256, + stream=True, + ) + for event in stream: + if event.type == "response.completed": + break + stream.close() + + _assert_token_points(_collect_points(reader)) + + +@pytest.mark.vcr(_CASSETTE) +@pytest.mark.asyncio +async def test_responses_stream_closed_after_completed_records_tokens_async( + instrument_legacy, reader, async_openai_client +): + _collect_points(reader) + + stream = await async_openai_client.responses.create( + model="gpt-6-luna", + input="Say hello in one word.", + max_output_tokens=256, + stream=True, + ) + async for event in stream: + if event.type == "response.completed": + break + await stream.aclose() + + _assert_token_points(_collect_points(reader))