OpenAI Responses and Compaction APIs

Complete testing, wrapping and coverage for the latest OpenAI
responses API and Compaction API features in the python client.

Author: iamemilio

instrumentation-genai/opentelemetry-instrumentation-openai-v2/pyproject.toml

@@ -25,9 +25,10 @@ classifiers = [
   "Programming Language :: Python :: 3.13",
 ]
 dependencies = [
+  "openai>=1.109.1",
   "opentelemetry-api ~= 1.37",
   "opentelemetry-instrumentation ~= 0.58b0",
-  "opentelemetry-semantic-conventions ~= 0.58b0"
+  "opentelemetry-semantic-conventions ~= 0.58b0",
 ]
 
 [project.optional-dependencies]

instrumentation-genai/opentelemetry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/__init__.py

@@ -40,6 +40,7 @@
 ---
 """
 
+import importlib
 from typing import Collection
 
 from wrapt import wrap_function_wrapper
@@ -60,6 +61,12 @@
     chat_completions_create,
     embeddings_create,
 )
+from .responses_patch import (
+    async_responses_compact,
+    async_responses_create,
+    responses_compact,
+    responses_create,
+)
 
 
 class OpenAIInstrumentor(BaseInstrumentor):
@@ -128,10 +135,56 @@ def _instrument(self, **kwargs):
             ),
         )
 
-    def _uninstrument(self, **kwargs):
-        import openai  # pylint: disable=import-outside-toplevel  # noqa: PLC0415
+        # Add instrumentation for the Responses API
+        wrap_function_wrapper(
+            module="openai.resources.responses",
+            name="Responses.create",
+            wrapper=responses_create(
+                tracer, logger, instruments, is_content_enabled()
+            ),
+        )
 
-        unwrap(openai.resources.chat.completions.Completions, "create")
-        unwrap(openai.resources.chat.completions.AsyncCompletions, "create")
-        unwrap(openai.resources.embeddings.Embeddings, "create")
-        unwrap(openai.resources.embeddings.AsyncEmbeddings, "create")
+        wrap_function_wrapper(
+            module="openai.resources.responses",
+            name="AsyncResponses.create",
+            wrapper=async_responses_create(
+                tracer, logger, instruments, is_content_enabled()
+            ),
+        )
+
+        # `Responses.compact` was added later in openai-python; guard so older
+        # supported versions don't fail instrumentation.
+        try:
+            wrap_function_wrapper(
+                module="openai.resources.responses",
+                name="Responses.compact",
+                wrapper=responses_compact(
+                    tracer, logger, instruments, is_content_enabled()
+                ),
+            )
+            wrap_function_wrapper(
+                module="openai.resources.responses",
+                name="AsyncResponses.compact",
+                wrapper=async_responses_compact(
+                    tracer, logger, instruments, is_content_enabled()
+                ),
+            )
+        except AttributeError:
+            pass
+
+    def _uninstrument(self, **kwargs):
+        chat_mod = importlib.import_module("openai.resources.chat.completions")
+        unwrap(chat_mod.Completions, "create")
+        unwrap(chat_mod.AsyncCompletions, "create")
+
+        embeddings_mod = importlib.import_module("openai.resources.embeddings")
+        unwrap(embeddings_mod.Embeddings, "create")
+        unwrap(embeddings_mod.AsyncEmbeddings, "create")
+
+        responses_mod = importlib.import_module("openai.resources.responses")
+        unwrap(responses_mod.Responses, "create")
+        unwrap(responses_mod.AsyncResponses, "create")
+        if hasattr(responses_mod.Responses, "compact"):
+            unwrap(responses_mod.Responses, "compact")
+        if hasattr(responses_mod.AsyncResponses, "compact"):
+            unwrap(responses_mod.AsyncResponses, "compact")

instrumentation-genai/opentelemetry-instrumentation-openai-v2/src/opentelemetry/instrumentation/openai_v2/patch.py

@@ -13,6 +13,8 @@
 # limitations under the License.
 
 
+import asyncio
+import inspect
 from timeit import default_timer
 from typing import Any, Optional
 
@@ -336,28 +338,39 @@ def _record_metrics(
     )
 
     if result and getattr(result, "usage", None):
-        # Always record input tokens
-        input_attributes = {
-            **common_attributes,
-            GenAIAttributes.GEN_AI_TOKEN_TYPE: GenAIAttributes.GenAiTokenTypeValues.INPUT.value,
-        }
-        instruments.token_usage_histogram.record(
-            result.usage.prompt_tokens,
-            attributes=input_attributes,
+        # Get input tokens - Responses API uses input_tokens, Chat Completions uses prompt_tokens
+        input_tokens = getattr(result.usage, "input_tokens", None) or getattr(
+            result.usage, "prompt_tokens", None
         )
 
+        if input_tokens is not None:
+            input_attributes = {
+                **common_attributes,
+                GenAIAttributes.GEN_AI_TOKEN_TYPE: GenAIAttributes.GenAiTokenTypeValues.INPUT.value,
+            }
+            instruments.token_usage_histogram.record(
+                input_tokens,
+                attributes=input_attributes,
+            )
+
         # For embeddings, don't record output tokens as all tokens are input tokens
         if (
             operation_name
             != GenAIAttributes.GenAiOperationNameValues.EMBEDDINGS.value
         ):
-            output_attributes = {
-                **common_attributes,
-                GenAIAttributes.GEN_AI_TOKEN_TYPE: GenAIAttributes.GenAiTokenTypeValues.COMPLETION.value,
-            }
-            instruments.token_usage_histogram.record(
-                result.usage.completion_tokens, attributes=output_attributes
-            )
+            # Get output tokens - Responses API uses output_tokens, Chat Completions uses completion_tokens
+            output_tokens = getattr(
+                result.usage, "output_tokens", None
+            ) or getattr(result.usage, "completion_tokens", None)
+
+            if output_tokens is not None:
+                output_attributes = {
+                    **common_attributes,
+                    GenAIAttributes.GEN_AI_TOKEN_TYPE: GenAIAttributes.GenAiTokenTypeValues.COMPLETION.value,
+                }
+                instruments.token_usage_histogram.record(
+                    output_tokens, attributes=output_attributes
+                )
 
 
 def _set_response_attributes(
@@ -403,6 +416,50 @@ def _set_response_attributes(
         )
 
 
+def _set_responses_response_attributes(
+    span, result, logger: Logger, capture_content: bool
+):
+    """Set span attributes from a Responses API result."""
+    set_span_attribute(
+        span,
+        GenAIAttributes.GEN_AI_RESPONSE_MODEL,
+        getattr(result, "model", None),
+    )
+
+    if getattr(result, "id", None):
+        set_span_attribute(span, GenAIAttributes.GEN_AI_RESPONSE_ID, result.id)
+
+    # Responses API uses "output" instead of "choices", and "status" instead of "finish_reason"
+    if getattr(result, "output", None):
+        finish_reasons = []
+        for output_item in result.output:
+            status = getattr(output_item, "status", None)
+            finish_reasons.append(status or "error")
+        set_span_attribute(
+            span,
+            GenAIAttributes.GEN_AI_RESPONSE_FINISH_REASONS,
+            finish_reasons,
+        )
+
+    # Get the usage - Responses API uses input_tokens/output_tokens
+    if getattr(result, "usage", None):
+        input_tokens = getattr(result.usage, "input_tokens", None)
+        if input_tokens is not None:
+            set_span_attribute(
+                span,
+                GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS,
+                input_tokens,
+            )
+
+        output_tokens = getattr(result.usage, "output_tokens", None)
+        if output_tokens is not None:
+            set_span_attribute(
+                span,
+                GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS,
+                output_tokens,
+            )
+
+
 def _set_embeddings_response_attributes(
     span: Span,
     result: Any,
@@ -539,7 +596,7 @@ def cleanup(self):
                 )
 
             for idx, choice in enumerate(self.choice_buffers):
-                message = {"role": "assistant"}
+                message: dict[str, Any] = {"role": "assistant"}
                 if self.capture_content and choice.text_content:
                     message["content"] = "".join(choice.text_content)
                 if choice.tool_calls_buffers:
@@ -605,8 +662,17 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
         return False  # Propagate the exception
 
     def close(self):
-        self.stream.close()
-        self.cleanup()
+        try:
+            close_result = self.stream.close()
+            if inspect.isawaitable(close_result):
+                try:
+                    loop = asyncio.get_running_loop()
+                except RuntimeError:
+                    asyncio.run(close_result)
+                else:
+                    loop.create_task(close_result)
+        finally:
+            self.cleanup()
 
     def __iter__(self):
         return self
@@ -629,7 +695,7 @@ def __next__(self):
 
     async def __anext__(self):
         try:
-            chunk = await self.stream.__anext__()
+            chunk = await self.stream.__anext__()  # type: ignore[attr-defined]
             self.process_chunk(chunk)
             return chunk
         except StopAsyncIteration: