class_docstr: OPT BB3 Agent.
class_docstr_tok: ["OPT","BB3","Agent","."]
docstr: Batch act for dialogue!
  
  If we have failures of knowledge conditioning, fall back to vanilla dialogue.
  
  :param observations:
      raw observations per module
  :param batch_reply_knowledge:
      list of knowledge replies
  
  :return batch_act_dialogue:
      return the dialogue replies!
docstr_tok: ["Batch","act","for","dialogue","!","If","we","have","failures","of","knowledge","conditioning",",","fall","back","to","vanilla","dialogue",".",":","param","observations",":","raw","observations","per","module",":","param","batch_reply_knowledge",":","list","of","knowledge","replies",":","return","batch_act_dialogue",":","return","the","dialogue","replies","!"]
code: def batch_act_dialogue(
      self,
      observations: List[Dict[Union[str, Module], Message]],
      batch_reply_knowledge: List[Message],
  ) -> List[Message]:
      """
      Batch act for dialogue!
  
      If we have failures of knowledge conditioning, fall back to vanilla dialogue.
  
      :param observations:
          raw observations per module
      :param batch_reply_knowledge:
          list of knowledge replies
  
      :return batch_act_dialogue:
          return the dialogue replies!
      """
      # construct dialogue obs with knowledge conditioning
      dialogue_obs, dialogue_replies = [], []
      for i, reply in enumerate(batch_reply_knowledge):
          for k_module in Module.knowledge_modules():
              if not reply.get(
                  k_module.message_name()
              ) or APIUtils.METASEQ_FAIL_MESSAGE_TEXT in reply.get(
                  k_module.message_name(), ""
              ):
                  continue
              logging.debug(
                  f"{k_module.message_name()}: {reply[k_module.message_name()]}"
              )
              d_module = None
              if k_module is Module.CONTEXTUAL_KNOWLEDGE:
                  d_module = Module.CONTEXTUAL_DIALOGUE
              elif k_module is Module.MEMORY_KNOWLEDGE:
                  d_module = Module.MEMORY_DIALOGUE
              else:
                  assert k_module is Module.SEARCH_KNOWLEDGE
                  d_module = Module.SEARCH_DIALOGUE
              d_obs = Message(observations[i][d_module])
              prompt = d_obs["prompt"].replace(
                  d_module.opt_post_context_tok(),
                  f"{d_module.opt_dialogue_knowledge_prefix()}{reply[k_module.message_name()]}",
              )
              d_obs.force_set("prompt", prompt)
              dialogue_obs.append((i, d_module, d_obs))
  
      # check if we have failures; if so, resort to vanilla dialogue
      all_inds = set([o[0] for o in dialogue_obs])
      for i in range(len(observations)):
          if i not in all_inds:
              dialogue_obs.append(
                  (
                      i,
                      Module.VANILLA_DIALOGUE,
                      Message(observations[i][Module.VANILLA_DIALOGUE]),
                  )
              )
  
      if self.opt["debug_bb3"]:
          DisplayUtils.display_observations(
              {o[1]: o[2] for o in dialogue_obs}
          )
  
      # check knowledge conditioning; combine observations if so
      if (
          self.knowledge_conditioning == "separate"
          or len(dialogue_obs) == 1
      ):
          dialogue_replies = self.batch_agents[
              Module.SEARCH_DIALOGUE
          ].batch_act([o[-1] for o in dialogue_obs])
      elif self.knowledge_conditioning == "combined":
          logging.debug("combining all knowledge")
          combined_obs: List[Message] = []
          for obs_i in range(len(observations)):
              extra_knowledge = []
              dialogue_obs_i = [
                  d for d in dialogue_obs if d[0] == obs_i
              ]
              primary_obs = dialogue_obs_i[0]
              for _, _, obs in dialogue_obs_i[1:]:
                  extra_knowledge.append(obs["prompt"].split("\n")[-2])
              old_prompt = primary_obs[-1]["prompt"].split("\n")
              primary_obs[-1].force_set(
                  "prompt",
                  "\n".join(
                      old_prompt[:-1]
                      + extra_knowledge
                      + old_prompt[-1:]
                  ),
              )
              combined_obs.append(primary_obs)
          batch_agent = self.batch_agents[self.combined_dialogue_module]
          dialogue_replies = batch_agent.batch_act(
              [o[-1] for o in combined_obs]
          )
          dialogue_obs = combined_obs
      else:
          raise NotImplementedError("Both is not implemented")
  
      batch_reply_dialogue = [
          Message({}) for _ in range(len(observations))
      ]
  
      # collate dialogue replies
      for (i, d_module, _), reply in zip(
          dialogue_obs, dialogue_replies
      ):
          if APIUtils.is_request_failed_response(reply):
              batch_reply_dialogue[i] = reply
              continue
          if self.knowledge_conditioning == "combined":
              logging.debug(f"combined dialogue reply: {reply}")
              assert "text" not in batch_reply_dialogue[i]
              batch_reply_dialogue[i]["text"] = reply["text"]
              batch_reply_dialogue[i]["score"] = reply["logprobs"]
              batch_reply_dialogue[i][
                  "best_module"
              ] = "Combined Response"
          else:
              logging.debug(f"{d_module.message_name()}: {reply}")
          batch_reply_dialogue[i][d_module.message_name()] = reply[
              "text"
          ]
          batch_reply_dialogue[i][
              f"{d_module.message_name()}_score"
          ] = reply["logprobs"]
  
      # if separate, make sure we set text to highest scoring object
      if self.knowledge_conditioning == "separate":
          for reply in batch_reply_dialogue:
              if APIUtils.is_request_failed_response(reply):
                  continue
              scores = []
              for module in Module.dialogue_modules():
                  scores.append(
                      (
                          module,
                          reply.get(
                              f"{module.message_name()}_score",
                              float("inf"),
                          ),
                      )
                  )
              best_module = min(scores, key=lambda x: x[-1])[0]
              reply["text"] = reply[best_module.message_name()]
              reply["score"] = reply[
                  f"{best_module.message_name()}_score"
              ]
              reply["best_module"] = best_module
  
      return batch_reply_dialogue
  
imports: import copy
  import random
  from typing import List, Optional, Dict, Any, Union, Tuple
  from parlai.agents.rag.retrievers import Document
  from parlai.core.agents import Agent, create_agent
  from parlai.core.message import Message
  from parlai.core.params import ParlaiParser
  from parlai.core.opt import Opt
  import parlai.utils.logging
  from projects.bb3.agents.module import Module
  from projects.bb3.agents.r2c2_bb3_agent import BlenderBot3Agent
  from projects.bb3.agents.opt_api_agent import BB3OPTAgent
  from projects.bb3.agents.search_agent import SearchAgent
  from projects.bb3.agents.utils import Decision, APIUtils, is_opener, DisplayUtils, set_failed_reply
  import projects.bb3.prompts
label_desc: Use image node_id 19 for calling the BlenderBot3Agent obj's underlying member method code with example usage: obj.batch_act_dialogue(observations, batch_reply_knowledge) and returns: batch_reply_dialogue