[BUG]: vdb_upload pipeline failing on incompatible message type

[dagardner-nv]

Version

24.03

Which installation method(s) does this occur on?

Source

Describe the bug.

RuntimeError: The inference stage cannot handle input of <class 'morpheus._lib.messages.ControlMessage'>. Accepted input types: (<class 'morpheus.messages.multi_inference_message.MultiInferenceMessage'>,)

Minimum reproducible example

python examples/llm/main.py vdb_upload pipeline --stop_after=1024

Relevant log output

Click here to see error details

====Pipeline Pre-build====

====Pre-Building Segment: main====

Error occurred during Pipeline.build(). Exiting.

Traceback (most recent call last):

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 458, in build_and_start

self.build()

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 294, in build

self._pre_build()

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 257, in _pre_build

stage._pre_build()

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/source_stage.py", line 73, in _pre_build

return super()._pre_build(do_propagate=do_propagate)

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/stage_base.py", line 372, in _pre_build

dep._pre_build(do_propagate=do_propagate)

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/stage_base.py", line 372, in _pre_build

dep._pre_build(do_propagate=do_propagate)

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/stage_base.py", line 372, in _pre_build

dep._pre_build(do_propagate=do_propagate)

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/stage_base.py", line 349, in _pre_build

self._pre_compute_schema(schema)

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/single_port_stage.py", line 65, in _pre_compute_schema

raise RuntimeError((f"The {self.name} stage cannot handle input of {input_type}. "

RuntimeError: The inference stage cannot handle input of <class 'morpheus._lib.messages.ControlMessage'>. Accepted input types: (<class 'morpheus.messages.multi_inference_message.MultiInferenceMessage'>,)

====Pipeline Complete====

Traceback (most recent call last):

File "/home/dagardner/work/morpheus-24.03/examples/llm/main.py", line 33, in 

run_cli()

File "/home/dagardner/work/morpheus-24.03/examples/llm/main.py", line 29, in run_cli

cli(obj={}, auto_envvar_prefix='MORPHEUS_LLM', show_default=True, prog_name="morpheus_llm")

File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 1157, in call

return self.main(*args, **kwargs)

File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 1078, in main

rv = self.invoke(ctx)

File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 1688, in invoke

return _process_result(sub_ctx.command.invoke(sub_ctx))

File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 1688, in invoke

return _process_result(sub_ctx.command.invoke(sub_ctx))

File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 1434, in invoke

return ctx.invoke(self.callback, **ctx.params)

File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/site-packages/click/core.py", line 783, in invoke

return __callback(*args, **kwargs)

File "/home/dagardner/work/morpheus-24.03/examples/llm/vdb_upload/run.py", line 175, in pipeline

return _pipeline(**final_config)

File "/home/dagardner/work/morpheus-24.03/examples/llm/vdb_upload/pipeline.py", line 104, in pipeline

pipe.run()

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 651, in run

asyncio.run(self.run_async())

File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/asyncio/runners.py", line 44, in run

return loop.run_until_complete(main)

File "/home/dagardner/work/conda/envs/morpheus-24.03/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete

return future.result()

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 632, in run_async

await self.join()

File "/home/dagardner/work/morpheus-24.03/morpheus/pipeline/pipeline.py", line 447, in join

assert self._post_start_future is not None, "Pipeline must be started before joining"

AssertionError: Pipeline must be started before joining

Full env printout

Click here to see environment details

[Paste the results of print_env.sh here, it will be hidden by default]

Other/Misc.

No response

Code of Conduct

• I agree to follow Morpheus' Code of Conduct
• I have searched the open bugs and have found no duplicates for this bug report