code: def _video_embedding(
      decode_op,
      emb_op,
      milvus_op,
      kv_op,
      norm_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()
  
      def _norm():
          return (
              pipe.input("url")
              .flat_map("url", "frames", decode_op)
              .map("frames", "emb", emb_op, config=op_config)
              .map("emb", "emb", norm_op)
              .map(("url", "emb"), "milvus_res", milvus_op)
              .window_all("emb", "video_emb", merge_ndarray)
              .map(("url", "video_emb"), ("insert_status"), kv_op)
              .output()
          )
  
      def _unnorm():
          return (
              pipe.input("url")
              .flat_map("url", "frames", decode_op)
              .map("frames", "emb", emb_op, config=op_config)
              .map(("url", "emb"), "milvus_res", milvus_op)
              .window_all("emb", "video_emb", merge_ndarray)
              .map(("url", "video_emb"), ("insert_status"), kv_op)
              .output()
          )
  
      if norm_op:
          return _norm()
      return _unnorm()
  
file_path: towhee/towhee/pipelines/video_embedding.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_embedding(decode_op, emb_op, milvus_op, kv_op, norm_op, allow_triton, device) and returns: _unnorm, pipe, pipe, _norm