code: def process_api_requests(
      requests: list[dict[str, any]],
      request_url: str,
      api_token: _OAITokenHolder,
      # Reference: https://platform.openai.com/docs/guides/rate-limits/overview
      max_requests_per_minute: float = 3_500,
      max_tokens_per_minute: float = 90_000,
      token_encoding_name: str = "cl100k_base",
      max_attempts: int = 5,
      max_workers: int = 10,
      throw_original_error=False,
  ):
      """Processes API requests in parallel, throttling to stay under rate limits."""
      # constants
      seconds_to_pause_after_rate_limit_error = 15
  
      # initialize trackers
      retry_queue = queue.Queue()
      status_tracker = (
          StatusTracker()
      )  # single instance to track a collection of variables
      next_request = None  # variable to hold the next request to call
  
      # initialize available capacity counts
      available_request_capacity = max_requests_per_minute
      available_token_capacity = max_tokens_per_minute
      last_update_time = time.time()
      results: list[tuple[int, any]] = []
      requests_iter = enumerate(requests)
      last_index = len(requests) - 1
      requests_exhausted = False
      _logger.debug(
          f"Request pool executor will run {len(requests)} requests"
      )
      with ThreadPoolExecutor(max_workers=max_workers) as executor:
          while True:
              # get next request (if one is not already waiting for capacity)
              if next_request is None:
                  if not retry_queue.empty():
                      next_request = retry_queue.get_nowait()
                      _logger.debug(
                          f"Retrying request {next_request.index}: {next_request}"
                      )
                  elif req := next(requests_iter, None):
                      # get new request
                      index, request_json = req
                      current_time = time.time()
                      next_request = APIRequest(
                          request_url=request_url,
                          index=index,
                          request_json=request_json,
                          token_consumption=num_tokens_consumed_from_request(
                              request_json,
                              request_url,
                              token_encoding_name,
                          ),
                          attempts_left=max_attempts,
                          results=results,
                          start_time=current_time,
                          last_log_time=current_time,
                      )
                      status_tracker.start_task()
                      requests_exhausted = index == last_index
  
              # update available capacity
              current_time = time.time()
              seconds_since_update = current_time - last_update_time
              available_request_capacity = min(
                  available_request_capacity
                  + int(
                      max_requests_per_minute
                      * seconds_since_update
                      / 60.0
                  ),
                  max_requests_per_minute,
              )
              available_token_capacity = min(
                  available_token_capacity
                  + int(
                      max_tokens_per_minute
                      * seconds_since_update
                      / 60.0
                  ),
                  max_tokens_per_minute,
              )
              last_update_time = current_time
  
              if next_request:
                  _logger.debug(
                      f"Available request capacity: {available_request_capacity}"
                  )
                  _logger.debug(
                      f"Available token capacity: {available_token_capacity}"
                  )
                  next_request_tokens = next_request.token_consumption
                  if (
                      available_request_capacity >= 1
                      and available_token_capacity
                      >= next_request_tokens
                  ):
                      # update counters
                      available_request_capacity -= 1
                      available_token_capacity -= next_request_tokens
                      next_request.attempts_left -= 1
                      # call API
                      api_token.validate(_logger)
                      executor.submit(
                          next_request.call_api,
                          retry_queue=retry_queue,
                          status_tracker=status_tracker,
                      )
                      next_request = None  # reset next_request to empty
                  else:
                      next_request = None
                      status_tracker.complete_task(success=False)
                      status_tracker.error = mlflow.MlflowException(
                          "Request size exceeded max tokens."
                      )
  
              # if all tasks are finished, break
              if (
                  requests_exhausted
                  and status_tracker.num_tasks_in_progress == 0
              ):
                  break
  
              # if a rate limit error was hit recently, pause to cool down
              seconds_since_rate_limit_error = (
                  time.time()
                  - status_tracker.time_of_last_rate_limit_error
              )
              if (
                  seconds_since_rate_limit_error
                  < seconds_to_pause_after_rate_limit_error
              ):
                  remaining_seconds_to_pause = (
                      seconds_to_pause_after_rate_limit_error
                      - seconds_since_rate_limit_error
                  )
                  _logger.debug(
                      "Encountered rate limit error. Pausing to cool down for "
                      f"{remaining_seconds_to_pause} seconds..."
                  )
                  time.sleep(remaining_seconds_to_pause)
                  # ^e.g., if pause is 15 seconds and final limit was hit 5 seconds ago
  
              time.sleep(0.001)  # avoid busy waiting
  
      # after finishing, log final status
      if status_tracker.num_tasks_failed > 0:
          if throw_original_error and len(requests) == 1:
              raise status_tracker.error
          raise mlflow.MlflowException(
              f"{status_tracker.num_tasks_failed} tasks failed. See logs for details."
          )
      if status_tracker.num_rate_limit_errors > 0:
          _logger.debug(
              f"{status_tracker.num_rate_limit_errors} rate limit errors received. "
              "Consider running at a lower rate."
          )
  
      return [res for _, res in sorted(results)]
  
imports: from __future__ import annotations
  import logging
  import os
  import queue
  import threading
  import time
  from concurrent.futures import ThreadPoolExecutor
  from dataclasses import dataclass
  import requests
  import tiktoken
  import mlflow
  from mlflow.protos.databricks_pb2 import UNAUTHENTICATED
  from mlflow.utils.openai_utils import _OAITokenHolder
label_desc: Use image node_id 2 for calling a global function with example usage: process_api_requests(requests, request_url, api_token, max_requests_per_minute, max_tokens_per_minute, token_encoding_name, max_attempts, max_workers, throw_original_error) and returns: unknown