_getPatchedChatCompletionsCreate()

in packages/instrumentation-openai/src/instrumentation.ts [145:224]

• 61 lines of code
• 5 McCabe index (conditional complexity)

  _getPatchedChatCompletionsCreate() {
    const self = this;
    // eslint-disable-next-line @typescript-eslint/no-explicit-any
    return (original: (...args: unknown[]) => any) => {
      // https://platform.openai.com/docs/api-reference/chat/create
      // eslint-disable-next-line @typescript-eslint/no-explicit-any
      return function patchedCreate(this: any, ...args: unknown[]) {
        if (!self.isEnabled) {
          return original.apply(this, args);
        }

        debug('OpenAI.Chat.Completions.create args: %O', args);
        /** type ChatCompletionCreateParamsStreaming */
        // eslint-disable-next-line @typescript-eslint/no-explicit-any
        const params = args[0] as any;
        const config = self.getConfig();
        const startNow = performance.now();

        let startInfo;
        try {
          startInfo = self._startChatCompletionsSpan(
            params,
            config,
            this?._client?.baseURL
          );
        } catch (err) {
          self._diag.error('unexpected error starting span:', err);
          return original.apply(this, args);
        }
        const { span, ctx, commonAttrs } = startInfo;

        // eslint-disable-next-line @typescript-eslint/no-explicit-any
        const apiPromise: Promise<any> = context.with(ctx, () =>
          original.apply(this, args)
        );

        // Streaming.
        if (params && params.stream) {
          // When streaming, `apiPromise` resolves to `import('openai/streaming').Stream`,
          // an async iterable (i.e. has a `Symbol.asyncIterator` method). We
          // want to wrap that iteration to gather telemetry. Instead of wrapping
          // `Symbol.asyncIterator`, which would be nice, we wrap the `iterator`
          // method because it is used internally by `Stream#tee()`.
          return apiPromise.then(stream => {
            self._wrap(stream, 'iterator', origIterator => {
              return () => {
                return self._onChatCompletionsStreamIterator(
                  origIterator(),
                  span,
                  startNow,
                  config,
                  commonAttrs,
                  ctx
                );
              };
            });
            return stream;
          });
        }

        // Non-streaming.
        apiPromise
          .then(result => {
            self._onChatCompletionsCreateResult(
              span,
              startNow,
              commonAttrs,
              result,
              config,
              ctx
            );
          })
          .catch(
            self._createAPIPromiseRejectionHandler(startNow, span, commonAttrs)
          );

        return apiPromise;
      };
    };
  }