code: def _video_copy_detection(
      decode_op,
      emb_op,
      milvus_op,
      kv_op,
      select_op,
      tn_op,
      norm_op,
      filter_op,
      allow_triton=False,
      device=-1,
  ):
      op_config = {}
      if allow_triton:
          if device >= 0:
              op_config = AutoConfig.TritonGPUConfig(
                  device_ids=[device], max_batch_size=128
              )
          else:
              op_config = AutoConfig.TritonCPUConfig()
      if norm_op:
          emb_pipe = (
              pipe.input("url")
              .flat_map("url", "frames", decode_op)
              .map("frames", "emb", emb_op, config=op_config)
              .map("emb", "emb", norm_op)
          )
      else:
          emb_pipe = (
              pipe.input("url")
              .flat_map("url", "frames", decode_op)
              .map("frames", "emb", emb_op, config=op_config)
          )
  
      search_pipe = (
          emb_pipe.flat_map("emb", "res", milvus_op)
          .window_all(
              "res",
              ("retrieved_urls", "score"),
              lambda x: ([i[2] for i in x], [i[1] for i in x]),
          )
          .window_all("emb", "video_emb", merge_ndarray)
          .flat_map(
              ("retrieved_urls", "score"), "candidates", select_op
          )
          .map("candidates", "retrieved_emb", kv_op)
      )
  
      if filter_op:
          detect_pipe = (
              search_pipe.map(
                  ("video_emb", "retrieved_emb"),
                  ("similar_segment", "segment_score"),
                  tn_op,
              )
              .filter(
                  ("candidates", "similar_segment", "segment_score"),
                  ("candidates", "similar_segment", "segment_score"),
                  ("similar_segment", "segment_score", "video_emb"),
                  filter_op,
              )
              .output(
                  "url",
                  "candidates",
                  "similar_segment",
                  "segment_score",
              )
          )
      else:
          detect_pipe = search_pipe.map(
              ("video_emb", "retrieved_emb"),
              ("similar_segment", "segment_score"),
              tn_op,
          ).output(
              "url", "candidates", "similar_segment", "segment_score"
          )
  
      return detect_pipe
  
file_path: towhee/towhee/pipelines/video_copy_detection.py
imports: from typing import Optional
  from pydantic import BaseModel
  from towhee import ops, pipe, AutoPipes, AutoConfig
label_desc: Use image node_id 2 for calling a global function with example usage: _video_copy_detection(decode_op, emb_op, milvus_op, kv_op, select_op, tn_op, norm_op, filter_op, allow_triton, device) and returns: detect_pipe