class_docstr: This Blueprint uses somewhat specialized arguments for turn annotations, manages
  their validation, and also has specialized data storage for the result format.
  
  It also has options for the onboarding data answers and the annotation bucket
  definitions.
class_docstr_tok: ["This","Blueprint","uses","somewhat","specialized","arguments","for","turn","annotations",",","manages","their","validation",",","and","also","has","specialized","data","storage","for","the","result","format",".","It","also","has","options","for","the","onboarding","data","answers","and","the","annotation","bucket","definitions","."]
code: def __init__(
      self,
      task_run: "TaskRun",
      args: "DictConfig",
      shared_state: "SharedTaskState",
  ):
      # Default conversation initialization
      super().__init__(task_run, args=args, shared_state=shared_state)
      random.seed(self.args.blueprint.random_seed)
      np.random.seed(self.args.blueprint.random_seed)
      torch.manual_seed(self.args.blueprint.random_seed)
  
      # Load task configuration data beyond the task description, as the super does
      # that
      left_pane_path = os.path.expanduser(
          args.blueprint.left_pane_text_path
      )
      with open(left_pane_path, "r") as left_pane_file:
          self.left_pane_text = left_pane_file.read()
      self.format_left_pane_text(args)
      self.annotations_config: Optional[str] = None
      if args.blueprint.get("annotations_config_path", "") != "":
          annotations_config_path = os.path.expanduser(
              args.blueprint.annotations_config_path
          )
          with open(
              annotations_config_path, "r"
          ) as annotations_config_file:
              self.annotations_config = annotations_config_file.read()
  
      # Initialize models
      shared_state.shared_models = self._get_shared_models(args)
  
      # Limits the number of models that can generate at once
      semaphore = Semaphore(args.blueprint.max_concurrent_responses)
  
      # Move shared state into the world opt, so that it can be used by the world
      shared_state.onboarding_world_opt.update(
          {"skip_onboarding": self.annotations_config is None}
      )
      # The onboarding checks how well workers annotate conversations, so it should be
      # skipped if we are not annotating
      shared_state.world_opt.update(
          {
              "block_qualification": args.blueprint.block_qualification,
              "annotations_config": self.annotations_config,
              "semaphore": semaphore,
              "shared_bot_agents": shared_state.shared_models,
              "num_turns": args.blueprint.num_turns,
              "max_resp_time": args.blueprint.max_resp_time,
              "is_sandbox": args.provider.requester_name
              == "MOCK_REQUESTER",
              "check_acceptability": args.blueprint.check_acceptability,
              "chat_data_folder": args.blueprint.chat_data_folder,
          }
      )
  
imports: import json
  import os
  import pickle
  import random
  from abc import ABC, abstractmethod
  from dataclasses import dataclass, field
  from threading import Semaphore, Condition
  from typing import Any, Dict, List, Optional, TYPE_CHECKING
  import numpy
  import torch
  import yaml
  from mephisto.operations.registry import register_mephisto_abstraction
  from mephisto.abstractions.blueprint import SharedTaskState
  from mephisto.abstractions.blueprints.parlai_chat.parlai_chat_blueprint import ParlAIChatBlueprint, SharedParlAITaskState, ParlAIChatBlueprintArgs
  from omegaconf import DictConfig, MISSING
  from parlai.crowdsourcing.tasks.model_chat.bot_agent import TurkLikeAgent
  from parlai.crowdsourcing.tasks.model_chat.utils import ImageStack, get_context_generator
  from parlai.tasks.blended_skill_talk.agents import ContextGenerator
label_desc: Use image node_id 2 to create a new BaseModelChatBlueprint object from inherited base classes: ParlAIChatBlueprint, ABC with example: obj = BaseModelChatBlueprint(task_run, args, shared_state)