class_docstr: BLIP2 Vicuna model.
  Supported model types:
      - vicuna7b
      - vicuna13b
  Usage:
      >>> from lavis.models import load_model
      >>> model = load_model("blip2_vicuna_xinstruct", "vicuna7b")
class_docstr_tok: ["BLIP2","Vicuna","model",".","Supported","model","types",":","-","vicuna7b","-","vicuna13b","Usage",":",">",">",">","from","lavis.models","import","load_model",">",">",">","model","=","load_model","(","``","blip2_vicuna_xinstruct","''",",","``","vicuna7b","''",")"]
code: def get_query_outputs(self, samples):
      if samples == None or samples == {}:
          return
  
      curr_modalities = [
          modality
          for modality in self.modalities
          if modality in samples
      ]
      if len(curr_modalities) == 0:
          print("Model modalities do not match sample modalities.")
          return
  
      # get batch size
      bs = None
      for modality in curr_modalities:
          data = samples[modality]
          bs = data.size(0)
          break
  
      if "prompt" in samples.keys():
          prompt = samples["prompt"]
      elif "text_input" in samples.keys():
          prompt = samples["text_input"]
      else:
          prompt = self.prompt
  
      if isinstance(prompt, str):
          prompt = [prompt] * bs
      else:
          assert (
              len(prompt) == bs
          ), "The number of prompts must be equal to the batch size."
  
      embeds = {}
      query_tokens = {}
      data_atts = {}
  
      for modality in curr_modalities:
          data = samples[modality]
          ln = getattr(self, f"{modality}_ln")
          encoder = getattr(self, f"{modality}_encoder")
          if (
              modality == "video"
              and self.video_enc_name
              in Blip2VicunaXInstruct.SEQUENCIAL_ENCODERS
          ):
              embeds[modality] = []
              data_atts[modality] = []
              for j in range(data.size(2)):
                  this_frame = data[:, :, j, :, :]
                  with self.maybe_autocast():
                      embeds[modality].append(ln(encoder(this_frame)))
                      if self.shared_qformer:
                          embeds[modality][-1] = getattr(
                              self, f"{modality}_encoder_projection"
                          )(embeds[modality][j])
                      data_atts[modality].append(
                          torch.ones(
                              embeds[modality][j].size()[:-1],
                              dtype=torch.long,
                          ).to(self.device)
                      )
              # B, Token Size, LM EMB
              query_tokens[modality] = getattr(
                  self, f"{modality}_query_tokens"
              ).expand(data.size(0), -1, -1)
  
          elif (
              modality == "audio"
              and self.audio_enc_name
              in Blip2VicunaXInstruct.SEQUENCIAL_ENCODERS
          ):
              embeds[modality] = []
              data_atts[modality] = []
              for j in range(data.size(1)):
                  this_frame = data[:, j, :, :]
                  with self.maybe_autocast():
                      embeds[modality].append(ln(encoder(this_frame)))
                      if self.shared_qformer:
                          embeds[modality][j] = getattr(
                              self, f"{modality}_encoder_projection"
                          )(embeds[modality][j])
                  data_atts[modality].append(
                      torch.ones(
                          embeds[modality][j].size()[:-1],
                          dtype=torch.long,
                      ).to(self.device)
                  )
              # B, Token Size, LM EMB
              if not self.projection_only and not getattr(
                  self, f"projection_only_{modality}"
              ):
                  query_tokens[modality] = getattr(
                      self, f"{modality}_query_tokens"
                  ).expand(data.size(0), -1, -1)
          else:
              with self.maybe_autocast():
                  embeds[modality] = ln(encoder(data))
              if len(embeds[modality].size()) == 2:
                  # B, C, D
                  embeds[modality] = embeds[modality].unsqueeze(1)
              # B, C
              if self.shared_qformer:
                  embeds[modality] = getattr(
                      self, f"{modality}_encoder_projection"
                  )(embeds[modality])
  
              data_atts[modality] = torch.ones(
                  embeds[modality].size()[:-1], dtype=torch.long
              ).to(self.device)
  
              # B, Token Size, LM EMB
              if not self.projection_only and not getattr(
                  self, f"projection_only_{modality}"
              ):
                  query_tokens[modality] = getattr(
                      self, f"{modality}_query_tokens"
                  ).expand(embeds[modality].shape[0], -1, -1)
  
      query_outputs = {}
      if self.qformer_text_input:
          text_Qformer = self.tokenizer(
              prompt,
              padding="longest",
              truncation=True,
              max_length=self.max_txt_len,
              return_tensors="pt",
          ).to(self.device)
  
          Qformer_atts = {}
          query_atts = {}
          num = {}
          for modality in curr_modalities:
              # B, Token Size
              if not self.projection_only and not getattr(
                  self, f"projection_only_{modality}"
              ):
                  query_atts[modality] = torch.ones(
                      query_tokens[modality].size()[:-1],
                      dtype=torch.long,
                  ).to(self.device)
                  # B, Token Size + Inp Size
                  Qformer_atts[modality] = torch.cat(
                      [
                          query_atts[modality],
                          text_Qformer.attention_mask,
                      ],
                      dim=1,
                  )
              if (
                  modality in Blip2VicunaXInstruct.SEQUENCIAL_MODALITIES
                  and getattr(self, f"{modality}_enc_name")
                  in Blip2VicunaXInstruct.SEQUENCIAL_ENCODERS
              ):
                  num[modality] = len(embeds[modality])
                  bs = embeds[modality][0].shape[0]
                  indices = [
                      j_ + r
                      for r, j in enumerate(
                          [[i * bs for i in range(num[modality])]] * bs
                      )
                      for j_ in j
                  ]
                  reordered_embeds = torch.cat(embeds[modality])[
                      indices
                  ]
                  reordered_atts = torch.cat(data_atts[modality])[
                      indices
                  ]
                  if self.projection_only or getattr(
                      self, f"projection_only_{modality}"
                  ):
                      if self.proj_dim != 1:
                          query_outputs[modality] = getattr(
                              self, f"{modality}_projection"
                          )(
                              reordered_embeds.mean(1, keepdim=True)
                          ).view(
                              bs * num[modality],
                              self.num_query_token,
                              -1,
                          )
                      else:
                          query_outputs[modality] = getattr(
                              self, f"{modality}_projection"
                          )(
                              reordered_embeds.view(
                                  reordered_embeds.shape[0], -1
                              )
                          )
                      continue
                  query_output = getattr(
                      self, f"{modality}_Qformer"
                  ).bert(
                      text_Qformer.input_ids.repeat(num[modality], 1),
                      attention_mask=Qformer_atts[modality].repeat(
                          num[modality], 1
                      ),
                      query_embeds=query_tokens[modality].repeat(
                          num[modality], 1, 1
                      ),
                      encoder_hidden_states=reordered_embeds,
                      encoder_attention_mask=reordered_atts,
                      return_dict=True,
                  )
                  query_outputs[modality] = query_output
              else:
                  if self.projection_only or getattr(
                      self, f"projection_only_{modality}"
                  ):
                      if self.proj_dim != 1:
                          query_outputs[modality] = getattr(
                              self, f"{modality}_projection"
                          )(
                              embeds[modality].mean(1, keepdim=True)
                          ).reshape(
                              bs, self.num_query_token, -1
                          )
                      else:
                          query_outputs[modality] = getattr(
                              self, f"{modality}_projection"
                          )(embeds[modality]).reshape(
                              bs, self.num_query_token, -1
                          )
                      continue
                  query_outputs[modality] = getattr(
                      self, f"{modality}_Qformer"
                  ).bert(
                      text_Qformer.input_ids,
                      attention_mask=Qformer_atts[modality],
                      query_embeds=query_tokens[modality],
                      encoder_hidden_states=embeds[modality].to(
                          torch.float32
                      ),
                      encoder_attention_mask=data_atts[modality],
                      return_dict=True,
                  )
      else:
          num = {}
          for modality in curr_modalities:
              if (
                  modality in Blip2VicunaXInstruct.SEQUENCIAL_MODALITIES
                  and getattr(self, f"{modality}_enc_name")
                  in Blip2VicunaXInstruct.SEQUENCIAL_ENCODERS
              ):
                  num[modality] = len(embeds[modality])
                  bs = embeds[modality][0].shape[0]
                  indices = [
                      j_ + r
                      for r, j in enumerate(
                          [[i * bs for i in range(num[modality])]] * bs
                      )
                      for j_ in j
                  ]
                  reordered_embeds = torch.cat(embeds[modality])[
                      indices
                  ]
                  reordered_atts = torch.cat(data_atts[modality])[
                      indices
                  ]
                  if self.projection_only or getattr(
                      self, f"projection_only_{modality}"
                  ):
                      if self.proj_dim != 1:
                          query_outputs[modality] = getattr(
                              self, f"{modality}_projection"
                          )(
                              reordered_embeds.mean(1, keepdim=True)
                          ).view(
                              bs * num, self.num_query_token, -1
                          )
                      else:
                          query_outputs[modality] = getattr(
                              self, f"{modality}_projection"
                          )(
                              reordered_embeds.view(
                                  reordered_embeds.shape[0], -1
                              )
                          )
                      continue
                  query_output = getattr(
                      self, f"{modality}_Qformer"
                  ).bert(
                      query_embeds=query_tokens[modality].repeat(
                          num[modality], 1, 1
                      ),
                      encoder_hidden_states=reordered_embeds,
                      encoder_attention_mask=reordered_atts,
                      return_dict=True,
                  )
                  query_outputs[modality] = query_output
              else:
                  if self.projection_only or getattr(
                      self, f"projection_only_{modality}"
                  ):
                      if self.proj_dim != 1:
                          query_outputs[modality] = getattr(
                              self, f"{modality}_projection"
                          )(
                              embeds[modality].mean(1, keepdim=True)
                          ).reshape(
                              bs, self.num_query_token, -1
                          )
                      else:
                          query_outputs[modality] = getattr(
                              self, f"{modality}_projection"
                          )(embeds[modality]).reshape(
                              bs, self.num_query_token, -1
                          )
                      continue
                  query_outputs[modality] = getattr(
                      self, f"{modality}_Qformer"
                  ).bert(
                      query_embeds=query_tokens[modality],
                      encoder_hidden_states=embeds[modality].to(
                          torch.float32
                      ),  # pc data is floa16.
                      encoder_attention_mask=data_atts[modality],
                      return_dict=True,
                  )
  
      for modality in curr_modalities:
          if (
              modality in Blip2VicunaXInstruct.SEQUENCIAL_MODALITIES
              and getattr(self, f"{modality}_enc_name")
              in Blip2VicunaXInstruct.SEQUENCIAL_ENCODERS
          ):
              if self.projection_only or getattr(
                  self, f"projection_only_{modality}"
              ):
                  if self.proj_dim != 1:
                      query_outputs[f"llm_proj_{modality}"] = getattr(
                          self, f"{modality}_llm_proj"
                      )(query_outputs[modality].unsqueeze(1)).reshape(
                          bs * num, self.num_query_token, -1
                      )
                  else:
                      query_outputs[f"llm_proj_{modality}"] = getattr(
                          self, f"{modality}_llm_proj"
                      )(query_outputs[modality]).reshape(
                          bs * num, self.num_query_token, -1
                      )
                  query_outputs[f"llm_proj_{modality}"] = (
                      query_outputs[f"llm_proj_{modality}"]
                      .reshape(
                          bs, num[modality], self.num_query_token, -1
                      )
                      .contiguous()
                      .view(
                          bs, num[modality] * self.num_query_token, -1
                      )
                  )
                  query_outputs[modality] = query_outputs[
                      modality
                  ].view(bs, num[modality] * self.num_query_token, -1)
              else:
                  query_outputs[f"llm_proj_{modality}"] = (
                      getattr(self, f"{modality}_llm_proj")(
                          query_outputs[modality]["last_hidden_state"][
                              :, : query_tokens[modality].size(1), :
                          ]
                      )
                      .contiguous()
                      .view(
                          bs, num[modality] * self.num_query_token, -1
                      )
                  )
                  query_outputs[modality] = (
                      query_outputs[modality]["last_hidden_state"][
                          :, : query_tokens[modality].size(1), :
                      ]
                      .contiguous()
                      .view(
                          bs, num[modality] * self.num_query_token, -1
                      )
                  )
  
          else:
              if self.projection_only or getattr(
                  self, f"projection_only_{modality}"
              ):
                  if self.proj_dim == 1:
                      query_outputs[f"llm_proj_{modality}"] = getattr(
                          self, f"{modality}_llm_proj"
                      )(query_outputs[modality].mean(-1)).reshape(
                          bs, self.num_query_token, -1
                      )
                  else:
                      query_outputs[f"llm_proj_{modality}"] = getattr(
                          self, f"{modality}_llm_proj"
                      )(
                          query_outputs[modality].reshape(
                              bs, self.num_query_token, -1
                          )
                      )
              else:
                  query_outputs[modality] = query_outputs[
                      modality
                  ].last_hidden_state[
                      :, : query_tokens[modality].size(1), :
                  ]
                  query_outputs[f"llm_proj_{modality}"] = getattr(
                      self, f"{modality}_llm_proj"
                  )(query_outputs[modality])
  
      for modality in curr_modalities:
          query_outputs[f"embeds_{modality}"] = embeds[modality]
      return query_outputs
  
imports: import logging
  import string
  from packaging import version
  import os
  from omegaconf import OmegaConf
  import torch
  from torch.cuda.amp import autocast
  import torch.nn
  from torch.nn.modules.module import _IncompatibleKeys
  from peft import get_peft_model, LoraConfig, TaskType
  import transformers
  import random
  from lavis.common.registry import registry
  from lavis.models.base_model import BaseModel
  from lavis.models.blip2_models.blip2 import Blip2Base, disabled_train, LayerNorm
  from lavis.models.ulip_models.ULIP_models import ULIP_PointBERT
  from lavis.tasks.multimodal_classification import MultimodalClassificationTask
  from lavis.common.utils import is_url
  from lavis.models.blip2_models.Qformer import BertConfig, BertLMHeadModel
  from lavis.common.dist_utils import download_cached_file
  from lavis.processors.blip_processors import BlipCaptionProcessor
label_desc: Use image node_id 8 for calling the Blip2VicunaXInstruct obj's underlying member method code with example usage: obj.get_query_outputs(samples) and returns: query_outputs