class_docstr: OPT BB3 Agent.
class_docstr_tok: ["OPT","BB3","Agent","."]
docstr: Add prompts to knowledge observations before calling batch act.
  
  :param observations:
      raw observations
  :param search_indices:
      indices in which we use search knowledge
  :param memory_indices:
      indices in which we use memories
  :param contextual_indices:
      indices in which we use an entity from the context
  :param batch_agents:
      mapping from module to agent with which to call batch act
  :param available_memory:
      memories per batch example
  :param search_results:
      search results per batch example
  :param top_docs:
      top docs corresponding to the search results
  :param top_memories:
      retrieved memories.
  
  :return batch_reply_knowledge:
      return batch reply from knowledge modules.
docstr_tok: ["Add","prompts","to","knowledge","observations","before","calling","batch","act",".",":","param","observations",":","raw","observations",":","param","search_indices",":","indices","in","which","we","use","search","knowledge",":","param","memory_indices",":","indices","in","which","we","use","memories",":","param","contextual_indices",":","indices","in","which","we","use","an","entity","from","the","context",":","param","batch_agents",":","mapping","from","module","to","agent","with","which","to","call","batch","act",":","param","available_memory",":","memories","per","batch","example",":","param","search_results",":","search","results","per","batch","example",":","param","top_docs",":","top","docs","corresponding","to","the","search","results",":","param","top_memories",":","retrieved","memories",".",":","return","batch_reply_knowledge",":","return","batch","reply","from","knowledge","modules","."]
code: def batch_act_knowledge(
      self,
      observations: List[Dict[Union[str, Module], Message]],
      search_indices: List[int],
      memory_indices: List[int],
      contextual_indices: List[int],
      batch_agents: Dict[Module, Agent],
      available_memory: List[List[str]],
      search_results: List[List[str]],
      top_docs: List[List[Document]],
      top_memories: List[List[str]],
  ) -> List[Message]:
      """
      Add prompts to knowledge observations before calling batch act.
  
      :param observations:
          raw observations
      :param search_indices:
          indices in which we use search knowledge
      :param memory_indices:
          indices in which we use memories
      :param contextual_indices:
          indices in which we use an entity from the context
      :param batch_agents:
          mapping from module to agent with which to call batch act
      :param available_memory:
          memories per batch example
      :param search_results:
          search results per batch example
      :param top_docs:
          top docs corresponding to the search results
      :param top_memories:
          retrieved memories.
  
      :return batch_reply_knowledge:
          return batch reply from knowledge modules.
      """
      for i, all_obs in enumerate(observations):
          for module in Module:
              obs = all_obs[module]
              if (
                  module is Module.MEMORY_KNOWLEDGE
                  and i in memory_indices
              ):
                  memories = "\n".join(available_memory[i])
                  new_prompt = self._check_and_limit_len(
                      obs["prompt"].replace(
                          module.opt_pre_context_tok(), memories
                      )
                  )
                  obs.force_set("prompt", new_prompt)
              elif (
                  module is Module.SEARCH_KNOWLEDGE
                  and i in search_indices
              ):
                  results = (
                      f"{PROMPT.EXTERNAL_KNOWLEDGE_PREFIX}: "
                      + f"\n{PROMPT.EXTERNAL_KNOWLEDGE_PREFIX}: ".join(
                          [
                              s.replace("\n", "")
                              for s in search_results[i]
                          ]
                      )
                  )
                  new_prompt = self._check_and_limit_len(
                      obs["prompt"].replace(
                          module.opt_pre_context_tok(), results
                      )
                  )
                  obs.force_set("prompt", new_prompt)
  
      batch_reply_knowledge = super().batch_act_knowledge(
          observations,
          search_indices,
          memory_indices,
          contextual_indices,
          batch_agents,
          top_docs=top_docs,
          top_memories=top_memories,
      )
      # re-assign memories
      for i, reply in enumerate(batch_reply_knowledge):
          if i in memory_indices:
              text = reply[
                  Module.MEMORY_KNOWLEDGE.message_name()
              ].strip()
              do_split = True
              if any(
                  text.startswith(p)
                  for p in [
                      self.partner_memory_prefix,
                      self.self_memory_prefix,
                  ]
              ):
                  # no need to split the memory to find it
                  do_split = False
              true_memory = ""
              for memory in available_memory[i]:
                  raw_mem = (
                      self.memory_utils.split_prefix_memory(memory)[-1]
                      if do_split
                      else memory
                  )
                  if raw_mem == text:
                      true_memory = memory
                      break
              if true_memory:
                  reply.force_set(
                      Module.MEMORY_KNOWLEDGE.message_name(),
                      true_memory,
                  )
              else:
                  logging.debug(
                      f"Memory {text} is invalid, given {available_memory[i]}"
                  )
                  reply.pop(
                      Module.MEMORY_KNOWLEDGE.message_name(), None
                  )
                  reply.pop(
                      f"{Module.MEMORY_KNOWLEDGE.message_name()}_docs",
                      None,
                  )
  
      return batch_reply_knowledge
  
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 16 for calling the BlenderBot3Agent obj's underlying member method code with example usage: obj.batch_act_knowledge(observations, search_indices, memory_indices, contextual_indices, batch_agents, available_memory, search_results, top_docs, top_memories) and returns: batch_reply_knowledge