code: def semantic_search(
      matcher,
      query_data: Optional[Union[pd.DataFrame, dict, list]] = None,
      response_data: Optional[Union[pd.DataFrame, dict, list]] = None,
      query_embeddings: Optional[torch.Tensor] = None,
      response_embeddings: Optional[torch.Tensor] = None,
      query_chunk_size: int = 128,
      response_chunk_size: int = 500000,
      top_k: int = 10,
      id_mappings: Optional[
          Union[Dict[str, Dict], Dict[str, pd.Series]]
      ] = None,
      similarity_type: Optional[str] = "cosine",
  ):
      """
      Perform a cosine similarity search between query data and response data.
  
      Parameters
      ----------
      query_data
          The query data.
      response_data
          The response data.
      query_embeddings
          2-D query embeddings.
      response_embeddings
          2-D response embeddings.
      id_mappings
          Id-to-content mappings. The contents can be text, image, etc.
          This is used when the dataframe contains the query/response indexes instead of their contents.
      query_chunk_size
          Process queries by query_chunk_size each time.
      response_chunk_size
          Process response data by response_chunk_size each time.
      top_k
          Retrieve top k matching entries.
      similarity_type
          Use what function (cosine/dot_prod) to score the similarity (default: cosine).
  
      Returns
      -------
      Search results.
      """
      assert (
          query_data is None or query_embeddings is None
      ), "Both query_data and query_embeddings are detected, but you can only use one of them."
      assert (
          query_data is not None or query_embeddings is not None
      ), "Both query_data and query_embeddings are None."
      assert (
          response_data is None or response_embeddings is None
      ), "Both response_data and response_embeddings are detected, but you can only use one of them."
      assert (
          response_data is not None or response_embeddings is not None
      ), "Both response_data and response_embeddings are None."
  
      if query_embeddings is None:
          query_header = (
              matcher.query[0] if matcher.query is not None else QUERY
          )
          query_data = data_to_df(query_data, header=query_header)
      if response_embeddings is None:
          response_header = (
              matcher.response[0] if matcher.response else RESPONSE
          )
          response_data = data_to_df(
              response_data, header=response_header
          )
  
      if query_embeddings is None:
          num_queries = len(query_data)
      else:
          num_queries = len(query_embeddings)
  
      if response_embeddings is None:
          num_responses = len(response_data)
      else:
          num_responses = len(response_embeddings)
  
      queries_result_list = [[] for _ in range(num_queries)]
  
      for query_start_idx in range(0, num_queries, query_chunk_size):
          if query_embeddings is None:
              batch_query_embeddings = matcher.extract_embedding(
                  query_data[
                      query_start_idx : query_start_idx
                      + query_chunk_size
                  ],
                  id_mappings=id_mappings,
                  as_tensor=True,
              )
          else:
              batch_query_embeddings = query_embeddings[
                  query_start_idx : query_start_idx + query_chunk_size
              ]
          # Iterate over chunks of the corpus
          for response_start_idx in range(
              0, num_responses, response_chunk_size
          ):
              if response_embeddings is None:
                  batch_response_embeddings = matcher.extract_embedding(
                      response_data[
                          response_start_idx : response_start_idx
                          + response_chunk_size
                      ],
                      id_mappings=id_mappings,
                      as_tensor=True,
                  )
              else:
                  batch_response_embeddings = response_embeddings[
                      response_start_idx : response_start_idx
                      + response_chunk_size
                  ]
              # Compute cosine similarities
              scores = compute_semantic_similarity(
                  a=batch_query_embeddings,
                  b=batch_response_embeddings,
                  similarity_type=similarity_type,
              )
  
              # Get top-k scores
              scores_top_k_values, scores_top_k_idx = torch.topk(
                  scores,
                  k=min(top_k, len(scores[0])),
                  dim=1,
                  largest=True,
                  sorted=False,
              )
              scores_top_k_values = scores_top_k_values.cpu().tolist()
              scores_top_k_idx = scores_top_k_idx.cpu().tolist()
  
              for query_itr in range(len(scores)):
                  for sub_response_id, score in zip(
                      scores_top_k_idx[query_itr],
                      scores_top_k_values[query_itr],
                  ):
                      corpus_id = response_start_idx + sub_response_id
                      query_id = query_start_idx + query_itr
                      if len(queries_result_list[query_id]) < top_k:
                          heapq.heappush(
                              queries_result_list[query_id],
                              (score, corpus_id),
                          )  # heaqp tracks the quantity of the first element in the tuple
                      else:
                          heapq.heappushpop(
                              queries_result_list[query_id],
                              (score, corpus_id),
                          )
  
      # change the data format and sort
      for query_id in range(len(queries_result_list)):
          for doc_itr in range(len(queries_result_list[query_id])):
              score, corpus_id = queries_result_list[query_id][doc_itr]
              queries_result_list[query_id][doc_itr] = {
                  "response_id": corpus_id,
                  "score": score,
              }
          queries_result_list[query_id] = sorted(
              queries_result_list[query_id],
              key=lambda x: x["score"],
              reverse=True,
          )
  
      return queries_result_list
  
imports: import copy
  import functools
  import heapq
  import logging
  from typing import Dict, List, Optional, Union
  import pandas
  import torch
  from omegaconf import DictConfig
  from torch import nn
  from torch.nn import functional
  from ..constants import FUSION, QUERY, RESPONSE
  from .data import data_to_df
  from .model import create_model
label_desc: Use image node_id 7 for calling a global function with example usage: semantic_search(matcher, query_data, response_data, query_embeddings, response_embeddings, query_chunk_size, response_chunk_size, top_k, id_mappings, similarity_type) and returns: queries_result_list