class_docstr: Test pipeline model
class_docstr_tok: ["Test","pipeline","model"]
code: def test_multi_params(self):
      p = (
          towhee.pipe.input("nums", "arr")
          .flat_map("nums", "num", lambda x: x)
          .map(("num", "arr"), "ret", lambda x, y: x + y)
          .output("ret")
      )
  
      input_data = ([1, 2, 3], np.random.rand(3, 3))
  
      json_data = to_json(input_data)
      triton_input = np.array([[json_data]], dtype=np.object_)
      input_tensors = pb_utils.InferenceRequest(
          [pb_utils.Tensor("INPUT0", triton_input)], [], ""
      )
  
      with TemporaryDirectory(dir="./") as root:
          pf = os.path.join(root, "pipe.pickle")
          with open(pf, "wb") as f:
              pickle.dump(p.dag_repr, f)
  
          pipe = pipe_model.TritonPythonModel()
          pipe._load_pipeline(pf)  # pylint: disable=protected-access
          res = pipe.execute([input_tensors])
          ret = from_json(res[0].output_tensors()[0].as_numpy()[0])[0]
          self.assertEqual(len(ret), 3)
          for index, item in enumerate(ret, 1):
              self.assertTrue(
                  (item[0] == (input_data[1] + index)).all()
              )
  
          # batch_inputs
          batch_triton_input = np.array(
              [[json_data], [json_data], [json_data]], dtype=np.object_
          )
          batch_input_tensors = pb_utils.InferenceRequest(
              [pb_utils.Tensor("INPUT0", batch_triton_input)], [], ""
          )
          res = pipe.execute([batch_input_tensors])
          ret = from_json(res[0].output_tensors()[0].as_numpy()[0])
          self.assertEqual(len(ret), 3)
          for single in ret:
              for index, item in enumerate(single, 1):
                  self.assertTrue(
                      (item[0] == (input_data[1] + index)).all()
                  )
  
file_path: towhee/tests/unittests/serve/triton/bls/test_pipeline_model.py
imports: import unittest
  import os
  from tempfile import TemporaryDirectory
  import dill
  import numpy
  import towhee
  from towhee.utils.serializer import to_json, from_json
  import towhee.serve.triton.bls.pipeline_model
  from towhee.serve.triton.bls.python_backend_wrapper import pb_utils
label_desc: Use image node_id 1 for calling the TestPipelineTritonModel obj's underlying member method code with example usage: obj.test_multi_params() without return types