class_docstr: OPT BB3 Agent.
class_docstr_tok: ["OPT","BB3","Agent","."]
docstr: Override batch act for OPT BB3.
docstr_tok: ["Override","batch","act","for","OPT","BB3","."]
code: def batch_act(
      self, observations: List[Dict[Union[str, Module], Message]]
  ) -> List[Message]:
      """
      Override batch act for OPT BB3.
      """
      if self.vanilla:
          return self.batch_act_simple(
              observations, Module.VANILLA_DIALOGUE
          )
  
      if all(
          is_opener(
              o["raw"]["text"],
              o[Module.OPENING_DIALOGUE].get("memories"),
          )
          for o in observations
      ):
          # We're opening!
          return self.batch_act_simple(
              observations, Module.OPENING_DIALOGUE
          )
  
      batch_reply_final = [
          Message({"id": "BlenderBot3", "failures": []})
          for _ in range(len(observations))
      ]
      # Step 1: determine whether we're searching or accessing memory
      all_memory: List[Dict[str, int]] = [
          o["raw"]["memories"] for o in observations
      ]
      available_memory = self.get_available_memories(observations)
  
      batch_reply_sdm, search_indices = self.batch_act_decision(
          observations,
          Module.SEARCH_DECISION,
          batch_reply_final,
      )
      batch_reply_mdm, memory_indices = self.batch_act_decision(
          observations,
          Module.MEMORY_DECISION,
          batch_reply_final,
      )
  
      memory_indices = [
          i for i in memory_indices if available_memory[i]
      ]
      contextual_indices = [
          i
          for i in range(len(observations))
          if (
              self.contextual_knowledge_decision is Decision.ALWAYS
              or (
                  self.contextual_knowledge_decision is Decision.COMPUTE
                  and i not in search_indices + memory_indices
              )
          )
          and self.contextual_knowledge_decision is not Decision.NEVER
      ]
      all_inds = (
          set(contextual_indices)
          .union(set(search_indices))
          .union(set(memory_indices))
      )
      if not all(i in all_inds for i in range(len(observations))):
          # something went terribly, terribly wrong. Vanilla batch act.
          return self.batch_act_simple(
              observations, Module.VANILLA_DIALOGUE
          )
      # Step 2: Generate search queries, if necessary. Also generate memories for partner
      batch_reply_mgm_partner = self.batch_act_mgm(
          observations=observations,
          agent=self.batch_agents[Module.MEMORY_GENERATOR],
      )
      batch_reply_sgm = self.batch_act_sgm(
          observations,
          search_indices,
          self.batch_agents[Module.SEARCH_QUERY],
      )
      search_indices = [
          i
          for i in search_indices
          if i
          not in self.detect_and_handle_failures(
              Module.SEARCH_QUERY, batch_reply_sgm, batch_reply_final
          )
      ]
  
      search_results, top_docs = self.get_search_results(
          batch_reply_sgm, search_indices
      )
      invalid_search_results = [
          i for i in search_indices if not top_docs[i]
      ]
      contextual_indices += invalid_search_results
      search_indices = [
          i for i in search_indices if i not in invalid_search_results
      ]
  
      # step 3: memory/knowledge/contextual access.
      # for knowledge obs, we need to fill in context with external or persona knowledge
      batch_reply_knowledge = self.batch_act_knowledge(
          observations,
          search_indices,
          memory_indices,
          contextual_indices,
          {m: self.batch_agents[m] for m in Module if m.is_knowledge()},
          available_memory,
          search_results,
          top_docs,
          [
              [
                  Document(docid="", text=memory, title="")
                  for memory in memories
              ]
              for memories in available_memory
          ],
      )
  
      # step 4: memory/knowledge/contextual dialogue
      batch_reply_dialogue = self.batch_act_dialogue(
          observations, batch_reply_knowledge
      )
      batch_reply_mgm_self = self.batch_act_mgm(
          self_messages=[
              self.agents[Module.MEMORY_GENERATOR].observe(r)
              for r in batch_reply_dialogue
          ],
          agent=self.batch_agents[Module.MEMORY_GENERATOR],
      )
  
      batch_reply_final = self.collate_batch_acts(
          batch_reply_sdm,
          batch_reply_mdm,
          batch_reply_sgm,
          batch_reply_mgm_self,
          batch_reply_mgm_partner,
          batch_reply_knowledge,
          batch_reply_dialogue,
          all_memory,
      )
      for i, reply in enumerate(batch_reply_final):
          reply.force_set("id", "BlenderBot3")
          if i in invalid_search_results:
              reply.force_set("search_result_failure", True)
          if self.opt["debug_bb3"]:
              DisplayUtils.display_act(reply)
  
      batch_reply_final = self.detect_and_handle_final_failures(
          batch_reply_final
      )
  
      return batch_reply_final
  
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 23 for calling the BlenderBot3Agent obj's underlying member method code with example usage: obj.batch_act(observations) and returns: batch_reply_final, self, self, self