code: def _save_outputs(self, opt, world, logger, episode_metrics):
      if (
          is_distributed()
      ):  # flatten everything intelligently if need be
          world_report = aggregate_unnamed_reports(
              all_gather_list(world.report())
          )
          episode_metrics_unflattened = all_gather_list(episode_metrics)
          flattened = []
          for rank_elem in episode_metrics_unflattened:
              for elem in rank_elem:
                  flattened.append(elem)
          episode_metrics = flattened
      else:
          world_report = world.report()
      logging.report("Final report:\n" + nice_report(world_report))
  
      report = dict_report(world_report)
  
      def get_episode_report(goal, episode_metric):
          metrics_dict = dict_report(episode_metric.report())
          metrics_dict["goal"] = goal
          return metrics_dict
  
      report["tod_metrics"] = [
          get_episode_report(g, e) for g, e in episode_metrics
      ]
  
      if (
          "report_filename" in opt
          and opt["report_filename"] is not None
      ):
          if len(world_report) == 0:
              logging.warning("Report is empty; not saving report")
  
          report_fname = f"{opt['report_filename']}.json"
          # Save report
          if not is_distributed() or is_primary_worker():
              with PathManager.open(report_fname, "w") as f:
                  logging.info(f"Saving model report to {report_fname}")
                  json.dump({"opt": opt, "report": report}, f, indent=4)
                  f.write("\n")  # for jq
  
      if "world_logs" in opt and opt["world_logs"] is not None:
          if (
              is_distributed()
          ):  # Save separately, then aggregate together
              rank = get_rank()
              log_outfile_part = f"{opt['world_logs']}_{opt['save_format']}_{rank}.jsonl"
              logger.write(
                  log_outfile_part,
                  world,
                  file_format=opt["save_format"],
              )
              sync_object(None)
              if is_primary_worker():
                  log_outfile = (
                      f"{opt['world_logs']}_{opt['save_format']}.jsonl"
                  )
                  log_outfile_metadata = f"{opt['world_logs']}_{opt['save_format']}.metadata"
                  with open(log_outfile, "w+") as outfile:
                      for rank in range(num_workers()):
                          log_outfile_part = f"{opt['world_logs']}_{opt['save_format']}_{rank}.jsonl"
                          with open(log_outfile_part) as infile:
                              for line in infile:
                                  json_blob = json.loads(line.strip())
                                  if (
                                      len(json_blob["dialog"]) < 2
                                  ):  # skip when we don't have generation
                                      continue
                                  json_blob[
                                      "metadata_path"
                                  ] = log_outfile_metadata
                                  outfile.write(json.dumps(json_blob))
                                  outfile.write("\n")
                          log_output_part_metadata = f"{opt['world_logs']}_{opt['save_format']}_{rank}.metadata"
                          if rank == 0:
                              copyfile(
                                  log_output_part_metadata,
                                  log_outfile_metadata,
                              ),
                          os.remove(log_outfile_part)
                          os.remove(log_output_part_metadata)
          else:
              log_outfile = (
                  f"{opt['world_logs']}_{opt['save_format']}.jsonl"
              )
              logger.write(
                  log_outfile, world, file_format=opt["save_format"]
              )
  
      return report
  
imports: import json
  from copy import deepcopy
  from shutil import copyfile
  import os
  import parlai.utils.logging
  import parlai.core.tod.tod_world
  import parlai.core.tod.tod_agents
  from parlai.core.agents import create_agent
  from parlai.core.metrics import dict_report, aggregate_unnamed_reports
  from parlai.core.opt import Opt
  from parlai.core.params import ParlaiParser
  from parlai.core.script import ParlaiScript, register_script
  from parlai.utils.distributed import is_primary_worker, all_gather_list, is_distributed, get_rank, sync_object, num_workers
  from parlai.utils.io import PathManager
  from parlai.utils.misc import TimeLogger, nice_report
  from parlai.utils.world_logging import WorldLogger
label_desc: Use image node_id 8 for calling the TodWorldScript obj's underlying member method code with example usage: obj._save_outputs(opt, world, logger, episode_metrics) and returns: report, metrics_dict