docstr: Handles completion for an arbitrary amount of rows.
  
  There are a couple checks that should be done when calling OpenAI's API:
    - account max batch size, to maximize batch size first
    - account rate limit, to maximize parallel calls second
  
  Additionally, single completion calls are done with exponential backoff to guarantee all prompts are processed,
  because even with previous checks the tokens-per-minute limit may apply.
docstr_tok: ["Handles","completion","for","an","arbitrary","amount","of","rows",".","There","are","a","couple","checks","that","should","be","done","when","calling","OpenAI","'s","API",":","-","account","max","batch","size",",","to","maximize","batch","size","first","-","account","rate","limit",",","to","maximize","parallel","calls","second","Additionally",",","single","completion","calls","are","done","with","exponential","backoff","to","guarantee","all","prompts","are","processed",",","because","even","with","previous","checks","the","tokens-per-minute","limit","may","apply","."]
code: def _completion(
      self,
      model_name,
      prompts,
      api_key,
      api_args,
      args,
      df,
      parallel=True,
  ):
      """
      Handles completion for an arbitrary amount of rows.
  
      There are a couple checks that should be done when calling OpenAI's API:
        - account max batch size, to maximize batch size first
        - account rate limit, to maximize parallel calls second
  
      Additionally, single completion calls are done with exponential backoff to guarantee all prompts are processed,
      because even with previous checks the tokens-per-minute limit may apply.
      """
  
      @retry_with_exponential_backoff()
      def _submit_completion(model_name, prompts, api_args, args, df):
          kwargs = {
              "model": model_name,
          }
          if model_name in IMAGE_MODELS:
              return _submit_image_completion(kwargs, prompts, api_args)
          elif model_name == "embedding":
              return _submit_embedding_completion(
                  kwargs, prompts, api_args
              )
          elif model_name in self.chat_completion_models:
              return _submit_chat_completion(
                  kwargs,
                  prompts,
                  api_args,
                  df,
                  mode=args.get("mode", "conversational"),
              )
          else:
              return _submit_normal_completion(
                  kwargs, prompts, api_args
              )
  
      def _log_api_call(params, response):
          after_openai_query(params, response)
  
          params2 = params.copy()
          params2.pop("api_key", None)
          params2.pop("user", None)
          logger.debug(f">>>openai call: {params2}:\n{response}")
  
      def _submit_normal_completion(kwargs, prompts, api_args):
          def _tidy(comp):
              tidy_comps = []
              for c in comp.choices:
                  if hasattr(c, "text"):
                      tidy_comps.append(c.text.strip("\n").strip(""))
              return tidy_comps
  
          kwargs["prompt"] = prompts
          kwargs = {**kwargs, **api_args}
  
          before_openai_query(kwargs)
          resp = _tidy(client.completions.create(**kwargs))
          _log_api_call(kwargs, resp)
          return resp
  
      def _submit_embedding_completion(kwargs, prompts, api_args):
          def _tidy(comp):
              tidy_comps = []
              for c in comp.data:
                  if hasattr(c, "embedding"):
                      tidy_comps.append([c.embedding])
              return tidy_comps
  
          kwargs["input"] = prompts
          kwargs = {**kwargs, **api_args}
  
          before_openai_query(kwargs)
          resp = _tidy(client.embeddings.create(**kwargs))
          _log_api_call(kwargs, resp)
          return resp
  
      def _submit_chat_completion(
          kwargs, prompts, api_args, df, mode="conversational"
      ):
          def _tidy(comp):
              tidy_comps = []
              for c in comp.choices:
                  if hasattr(c, "message"):
                      tidy_comps.append(
                          c.message.content.strip("\n").strip("")
                      )
              return tidy_comps
  
          completions = []
          if mode != "conversational":
              initial_prompt = {
                  "role": "system",
                  "content": "You are a helpful assistant. Your task is to continue the chat.",
              }  # noqa
          else:
              # get prompt from model
              initial_prompt = {
                  "role": "system",
                  "content": args["prompt"],
              }  # noqa
  
          kwargs["messages"] = [initial_prompt]
          last_completion_content = None
  
          for pidx in range(len(prompts)):
              if mode != "conversational":
                  kwargs["messages"].append(
                      {"role": "user", "content": prompts[pidx]}
                  )
              else:
                  question = prompts[pidx]
                  if question:
                      kwargs["messages"].append(
                          {"role": "user", "content": question}
                      )
                  answer = df.iloc[pidx][args.get("assistant_column")]
                  if answer:
                      kwargs["messages"].append(
                          {"role": "assistant", "content": answer}
                      )
  
              if mode == "conversational-full" or (
                  mode == "conversational" and pidx == len(prompts) - 1
              ):
                  kwargs["messages"] = truncate_msgs_for_token_limit(
                      kwargs["messages"],
                      kwargs["model"],
                      api_args["max_tokens"],
                  )
                  pkwargs = {**kwargs, **api_args}
  
                  before_openai_query(kwargs)
                  resp = _tidy(
                      client.chat.completions.create(**pkwargs)
                  )
                  _log_api_call(pkwargs, resp)
  
                  completions.extend(resp)
              elif mode == "default":
                  kwargs["messages"] = [initial_prompt] + [
                      kwargs["messages"][-1]
                  ]
                  pkwargs = {**kwargs, **api_args}
  
                  before_openai_query(kwargs)
                  resp = _tidy(
                      client.chat.completions.create(**pkwargs)
                  )
                  _log_api_call(pkwargs, resp)
  
                  completions.extend(resp)
              else:
                  # in "normal" conversational mode, we request completions only for the last row
                  last_completion_content = None
                  if args.get("answer_column") in df.columns:
                      # insert completion if provided, which saves redundant API calls
                      completions.extend(
                          [df.iloc[pidx][args.get("answer_column")]]
                      )
                  else:
                      completions.extend([""])
  
              if args.get("answer_column") in df.columns:
                  kwargs["messages"].append(
                      {
                          "role": "assistant",
                          "content": df.iloc[pidx][
                              args.get("answer_column")
                          ],
                      }
                  )
              elif last_completion_content:
                  # interleave assistant responses with user input
                  kwargs["messages"].append(
                      {
                          "role": "assistant",
                          "content": last_completion_content[0],
                      }
                  )
  
          return completions
  
      def _submit_image_completion(kwargs, prompts, api_args):
          def _tidy(comp):
              return [
                  c.url if hasattr(c, "url") else c.b64_json
                  for c in comp
              ]
  
          completions = [
              client.images.generate(
                  **{"prompt": p, **kwargs, **api_args}
              ).data[0]
              for p in prompts
          ]
          return _tidy(completions)
  
      client = self._get_client(
          api_key=api_key,
          base_url=args.get("api_base"),
          org=args.pop("api_organization")
          if "api_organization" in args
          else None,
      )
      try:
          # check if simple completion works
          completion = _submit_completion(
              model_name, prompts, api_args, args, df
          )
          return completion
      except Exception as e:
          # else, we get the max batch size
          if (
              "you can currently request up to at most a total of"
              in str(e)
          ):
              pattern = "a total of"
              max_batch_size = int(
                  e[e.find(pattern) + len(pattern) :].split(").")[0]
              )
          else:
              max_batch_size = (
                  self.max_batch_size
              )  # guards against changes in the API message
  
      if not parallel:
          completion = None
          for i in range(math.ceil(len(prompts) / max_batch_size)):
              partial = _submit_completion(
                  model_name,
                  prompts[
                      i * max_batch_size : (i + 1) * max_batch_size
                  ],
                  api_args,
                  args,
                  df,
              )
              if not completion:
                  completion = partial
              else:
                  completion["choices"].extend(partial["choices"])
                  for field in (
                      "prompt_tokens",
                      "completion_tokens",
                      "total_tokens",
                  ):
                      completion["usage"][field] += partial["usage"][
                          field
                      ]
      else:
          promises = []
          with concurrent.futures.ThreadPoolExecutor() as executor:
              for i in range(math.ceil(len(prompts) / max_batch_size)):
                  logger.debug(
                      f"{i * max_batch_size}:{(i+1) * max_batch_size}/{len(prompts)}"
                  )
                  future = executor.submit(
                      _submit_completion,
                      model_name,
                      prompts[
                          i * max_batch_size : (i + 1) * max_batch_size
                      ],
                      api_args,
                      args,
                      df,
                  )
                  promises.append({"choices": future})
          completion = None
          for p in promises:
              if not completion:
                  completion = p["choices"].result()
              else:
                  completion.extend(p["choices"].result())
  
      return completion
  
imports: import os
  import math
  import json
  import shutil
  import tempfile
  import datetime
  import textwrap
  import subprocess
  import concurrent.futures
  from typing import Optional, Dict
  import openai
  from openai import OpenAI
  import numpy
  import pandas
  from mindsdb.utilities.hooks import before_openai_query, after_openai_query
  from mindsdb.utilities import log
  from mindsdb.integrations.libs.base import BaseMLEngine
  from mindsdb.integrations.handlers.openai_handler.helpers import retry_with_exponential_backoff, truncate_msgs_for_token_limit, get_available_models, PendingFT
  from mindsdb.integrations.handlers.openai_handler.constants import CHAT_MODELS, IMAGE_MODELS, FINETUNING_MODELS, OPENAI_API_BASE
  from mindsdb.integrations.utilities.handler_utils import get_api_key
  from mindsdb.integrations.libs.llm_utils import get_completed_prompts
label_desc: Use image node_id 5 for calling the OpenAIHandler obj's underlying member method code with example usage: obj._completion(model_name, prompts, api_key, api_args, args, df, parallel) and returns: completion, resp, resp, completions, _tidy, completion, _submit_image_completion, tidy_comps, tidy_comps, tidy_comps, unknown, _submit_embedding_completion, _submit_chat_completion, _submit_normal_completion