openai responses instrumentation (#6583)

* openai responses instrumentation

* clean up code

* add streaming processing, fix span tag adding

* make changes to conform with updated tool call tagging behavior tests

* fix tags after merge master

* remove unecessary stream chunk processing,  refactor metadata tags

* address comments, remove unecessary empty content tag handling

* run linter

* move const allowedParams to top of file

* clean up

* add test for openai response span

* add streamed test, fix non-streamed response test

* add cassettes

* remove default to empty string for non-existent name case

* lint

* fix test

* lint

* Update packages/datadog-plugin-openai/src/stream-helpers.js

Co-authored-by: Ruben Bridgewater <[email protected]>

* Update packages/dd-trace/src/llmobs/plugins/openai.js

Co-authored-by: Ruben Bridgewater <[email protected]>

* clean up response chunk extracting

* fix tests

* lint remove expect statements

---------

Co-authored-by: Ruben Bridgewater <[email protected]>

Author: 2 people

packages/datadog-instrumentations/src/openai.js

@@ -22,6 +22,14 @@ const V4_PACKAGE_SHIMS = [
     methods: ['create'],
     streamedResponse: true
   },
+  {
+    file: 'resources/responses/responses',
+    targetClass: 'Responses',
+    baseResource: 'responses',
+    methods: ['create'],
+    streamedResponse: true,
+    versions: ['>=4.87.0']
+  },
   {
     file: 'resources/embeddings',
     targetClass: 'Embeddings',

packages/datadog-plugin-openai/src/stream-helpers.js

@@ -107,8 +107,33 @@ function constructChatCompletionResponseFromStreamedChunks (chunks, n) {
   })
 }
 
+/**
+ * Constructs the entire response from a stream of OpenAI responses chunks.
+ * The responses API uses event-based streaming with delta chunks.
+ * @param {Array<Record<string, any>>} chunks
+ * @returns {Record<string, any>}
+ */
+function constructResponseResponseFromStreamedChunks (chunks) {
+  // The responses API streams events with different types:
+  // - response.output_text.delta: incremental text deltas
+  // - response.output_text.done: complete text for a content part
+  // - response.output_item.done: complete output item with role
+  // - response.done/response.incomplete/response.completed: final response with output array and usage
+
+  // Find the last chunk with a complete response object (status: done, incomplete, or completed)
+  const responseStatusSet = new Set(['done', 'incomplete', 'completed'])
+
+  for (let i = chunks.length - 1; i >= 0; i--) {
+    const chunk = chunks[i]
+    if (chunk.response && responseStatusSet.has(chunk.response.status)) {
+      return chunk.response
+    }
+  }
+}
+
 module.exports = {
   convertBuffersToObjects,
   constructCompletionResponseFromStreamedChunks,
-  constructChatCompletionResponseFromStreamedChunks
+  constructChatCompletionResponseFromStreamedChunks,
+  constructResponseResponseFromStreamedChunks
 }

packages/datadog-plugin-openai/src/tracing.js

@@ -11,7 +11,8 @@ const { MEASURED } = require('../../../ext/tags')
 const {
   convertBuffersToObjects,
   constructCompletionResponseFromStreamedChunks,
-  constructChatCompletionResponseFromStreamedChunks
+  constructChatCompletionResponseFromStreamedChunks,
+  constructResponseResponseFromStreamedChunks
 } = require('./stream-helpers')
 
 const { DD_MAJOR } = require('../../../version')
@@ -59,6 +60,8 @@ class OpenAiTracingPlugin extends TracingPlugin {
         response = constructCompletionResponseFromStreamedChunks(chunks, n)
       } else if (methodName === 'createChatCompletion') {
         response = constructChatCompletionResponseFromStreamedChunks(chunks, n)
+      } else if (methodName === 'createResponse') {
+        response = constructResponseResponseFromStreamedChunks(chunks)
       }
 
       ctx.result = { data: response }
@@ -134,6 +137,10 @@ class OpenAiTracingPlugin extends TracingPlugin {
       case 'createEdit':
         createEditRequestExtraction(tags, payload, openaiStore)
         break
+
+      case 'createResponse':
+        createResponseRequestExtraction(tags, payload, openaiStore)
+        break
     }
 
     span.addTags(tags)
@@ -313,6 +320,10 @@ function normalizeMethodName (methodName) {
     case 'embeddings.create':
       return 'createEmbedding'
 
+    // responses
+    case 'responses.create':
+      return 'createResponse'
+
     // files
     case 'files.create':
       return 'createFile'
@@ -376,6 +387,16 @@ function createEditRequestExtraction (tags, payload, openaiStore) {
   openaiStore.instruction = instruction
 }
 
+function createResponseRequestExtraction (tags, payload, openaiStore) {
+  // Extract model information
+  if (payload.model) {
+    tags['openai.request.model'] = payload.model
+  }
+
+  // Store the full payload for response extraction
+  openaiStore.responseData = payload
+}
+
 function retrieveModelRequestExtraction (tags, payload) {
   tags['openai.request.id'] = payload.id
 }
@@ -410,6 +431,10 @@ function responseDataExtractionByMethod (methodName, tags, body, openaiStore) {
       commonCreateResponseExtraction(tags, body, openaiStore, methodName)
       break
 
+    case 'createResponse':
+      createResponseResponseExtraction(tags, body, openaiStore)
+      break
+
     case 'listFiles':
     case 'listFineTunes':
     case 'listFineTuneEvents':
@@ -513,6 +538,26 @@ function commonCreateResponseExtraction (tags, body, openaiStore, methodName) {
   openaiStore.choices = body.choices
 }
 
+function createResponseResponseExtraction (tags, body, openaiStore) {
+  // Extract response ID if available
+  if (body.id) {
+    tags['openai.response.id'] = body.id
+  }
+
+  // Extract status if available
+  if (body.status) {
+    tags['openai.response.status'] = body.status
+  }
+
+  // Extract model from response if available
+  if (body.model) {
+    tags['openai.response.model'] = body.model
+  }
+
+  // Store the full response for potential future use
+  openaiStore.response = body
+}
+
 // The server almost always responds with JSON
 function coerceResponseBody (body, methodName) {
   switch (methodName) {