WIP : bypassed the message incompatibility error, but now failing on …

…a validation error on the rss source schema

examples/llm/vdb_upload/helper.py

@@ -17,6 +17,7 @@
 
 from morpheus.config import Config
 from morpheus.messages import ControlMessage
+from morpheus.messages import MessageMeta
 from morpheus.pipeline.pipeline import Pipeline
 from morpheus.stages.general.linear_modules_source import LinearModuleSourceStage
 
@@ -73,7 +74,7 @@ def setup_rss_source(pipe: Pipeline, config: Config, source_name: str, rss_confi
         module_config={"rss_config": rss_config},
     )
     rss_pipe = pipe.add_stage(
-        LinearModuleSourceStage(config, module_definition, output_type=ControlMessage, output_port_name="output"))
+        LinearModuleSourceStage(config, module_definition, output_type=MessageMeta, output_port_name="output"))
 
     return rss_pipe
 

examples/llm/vdb_upload/pipeline.py

@@ -24,6 +24,7 @@
 from morpheus.stages.general.trigger_stage import TriggerStage
 from morpheus.stages.inference.triton_inference_stage import TritonInferenceStage
 from morpheus.stages.output.write_to_vector_db_stage import WriteToVectorDBStage
+from morpheus.stages.preprocess.deserialize_stage import DeserializeStage
 from morpheus.stages.preprocess.preprocess_nlp_stage import PreprocessNLPStage
 
 logger = logging.getLogger(__name__)
@@ -64,6 +65,8 @@ def pipeline(pipeline_config: Config,
 
     vdb_sources = process_vdb_sources(pipe, pipeline_config, source_config)
 
+    deserialize_stage = pipe.add_stage(DeserializeStage(pipeline_config))
+
     trigger = None
     if (isolate_embeddings):
         trigger = pipe.add_stage(TriggerStage(pipeline_config))
@@ -85,13 +88,13 @@ def pipeline(pipeline_config: Config,
 
     # Connect the pipeline
     for source_output in vdb_sources:
-        if (isolate_embeddings):
-            pipe.add_edge(source_output, trigger)
-        else:
-            pipe.add_edge(source_output, nlp_stage)
+        pipe.add_edge(source_output, deserialize_stage)
 
     if (isolate_embeddings):
+        pipe.add_edge(deserialize_stage, trigger)
         pipe.add_edge(trigger, nlp_stage)
+    else:
+        pipe.add_edge(deserialize_stage, nlp_stage)
 
     pipe.add_edge(nlp_stage, monitor_1)
     pipe.add_edge(monitor_1, embedding_stage)