Merge branch 'main' into ayushag/worker-to-mdc-parser-info

Author: ayushag-nv

examples/multimodal/README.md

@@ -59,14 +59,12 @@ flowchart LR
   pd_worker --> encode_worker
 ```
 
+***Note*** Only the LLaVA 1.5 7B model is supported. Qwen2.5-VL and Phi3V support will be added in the future.
+
 ```bash
 cd $DYNAMO_HOME/examples/multimodal
 # Serve a LLaVA 1.5 7B model:
 bash launch/agg.sh --model llava-hf/llava-1.5-7b-hf
-# Serve a Qwen2.5-VL model:
-# bash launch/agg.sh --model Qwen/Qwen2.5-VL-7B-Instruct
-# Serve a Phi3V model:
-# bash launch/agg.sh --model microsoft/Phi-3.5-vision-instruct
 ```
 
 ### Client
@@ -100,8 +98,6 @@ curl http://localhost:8080/v1/chat/completions \
     }'
 ```
 
-If serving the example Qwen model, replace `"llava-hf/llava-1.5-7b-hf"` in the `"model"` field with `"Qwen/Qwen2.5-VL-7B-Instruct"`. If serving the example Phi3V model, replace `"llava-hf/llava-1.5-7b-hf"` in the `"model"` field with `"microsoft/Phi-3.5-vision-instruct"`.
-
 You should see a response similar to this:
 ```json
 {"id": "c37b946e-9e58-4d54-88c8-2dbd92c47b0c", "object": "chat.completion", "created": 1747725277, "model": "llava-hf/llava-1.5-7b-hf", "choices": [{"index": 0, "message": {"role": "assistant", "content": " In the image, there is a city bus parked on a street, with a street sign nearby on the right side. The bus appears to be stopped out of service. The setting is in a foggy city, giving it a slightly moody atmosphere."}, "finish_reason": "stop"}]}

lib/llm/src/migration.rs

@@ -100,10 +100,13 @@ impl RetryManager {
             if let Some(response) = response_stream.next().await {
                 if let Some(err) = response.err() {
                     const STREAM_ERR_MSG: &str = "Stream ended before generation completed";
-                    if format!("{:?}", err) == STREAM_ERR_MSG {
+                    if err
+                        .chain()
+                        .any(|e| e.to_string().starts_with(STREAM_ERR_MSG))
+                    {
                         tracing::warn!("Stream disconnected... recreating stream...");
                         if let Err(err) = self.new_stream().await {
-                            tracing::warn!("Cannot recreate stream: {:?}", err);
+                            tracing::warn!("Cannot recreate stream: {:#}", err);
                         } else {
                             continue;
                         }
@@ -462,6 +465,7 @@ mod tests {
     /// Expected behavior: All 10 responses should be received successfully.
     #[tokio::test]
     async fn test_retry_manager_no_migration() {
+        dynamo_runtime::logging::init();
         let request = create_mock_request(10);
         let mock_engine = Arc::new(MockEngine::new(MockBehavior::Success, 10, 100));
         let next_generate: ServerStreamingEngine<PreprocessedRequest, Annotated<LLMEngineOutput>> =
@@ -493,6 +497,7 @@ mod tests {
     /// Expected behavior: All 10 responses should be received successfully after retry.
     #[tokio::test]
     async fn test_retry_manager_new_request_migration() {
+        dynamo_runtime::logging::init();
         let request = create_mock_request(10);
         let mock_engine = Arc::new(MockEngine::new(MockBehavior::FailThenSuccess, 10, 100));
         let next_generate: ServerStreamingEngine<PreprocessedRequest, Annotated<LLMEngineOutput>> =
@@ -524,6 +529,8 @@ mod tests {
     /// Expected behavior: 5 responses from first stream + 5 responses from retry stream = 10 total.
     #[tokio::test]
     async fn test_retry_manager_ongoing_request_migration() {
+        dynamo_runtime::logging::init();
+
         let request = create_mock_request(10);
         let mock_engine = Arc::new(MockEngine::new(
             MockBehavior::MidStreamFail { fail_after: 5 },
@@ -560,6 +567,7 @@ mod tests {
     /// Expected behavior: Should receive an error after all retries are exhausted, with the original error.
     #[tokio::test]
     async fn test_retry_manager_new_request_migration_indefinite_failure() {
+        dynamo_runtime::logging::init();
         let request = create_mock_request(0);
         let mock_engine = Arc::new(MockEngine::new(MockBehavior::AlwaysFail, 0, 100));
         let next_generate: ServerStreamingEngine<PreprocessedRequest, Annotated<LLMEngineOutput>> =
@@ -580,6 +588,7 @@ mod tests {
     /// Expected behavior: Should receive some responses from first stream, then error after retries exhausted.
     #[tokio::test]
     async fn test_retry_manager_ongoing_request_migration_indefinite_failure() {
+        dynamo_runtime::logging::init();
         let request = create_mock_request(10);
         let mock_engine = Arc::new(MockEngine::new(
             MockBehavior::MidStreamFailAlways { fail_after: 3 },
@@ -627,6 +636,7 @@ mod tests {
     /// Expected behavior: Should receive some responses from first stream, then error after retries exhausted.
     #[tokio::test]
     async fn test_retry_manager_ongoing_request_migration_indefinite_failure_stream_error() {
+        dynamo_runtime::logging::init();
         let request = create_mock_request(10);
         let mock_engine = Arc::new(MockEngine::new(
             MockBehavior::MidStreamFailAlwaysStreamError { fail_after: 3 },

lib/llm/src/protocols/common/llm_backend.rs

@@ -157,9 +157,9 @@ impl MaybeError for LLMEngineOutput {
         LLMEngineOutput::error(format!("{:?}", err))
     }
 
-    fn err(&self) -> Option<Box<dyn std::error::Error + Send + Sync>> {
+    fn err(&self) -> Option<anyhow::Error> {
         if let Some(FinishReason::Error(err_msg)) = &self.finish_reason {
-            Some(anyhow::Error::msg(err_msg.clone()).into())
+            Some(anyhow::Error::msg(err_msg.clone()))
         } else {
             None
         }

lib/runtime/src/protocols/annotated.rs

@@ -143,14 +143,14 @@ where
         Annotated::from_error(format!("{:?}", err))
     }
 
-    fn err(&self) -> Option<Box<dyn std::error::Error + Send + Sync>> {
+    fn err(&self) -> Option<anyhow::Error> {
         if self.is_error() {
             if let Some(comment) = &self.comment {
                 if !comment.is_empty() {
-                    return Some(anyhow::Error::msg(comment.join("; ")).into());
+                    return Some(anyhow::Error::msg(comment.join("; ")));
                 }
             }
-            Some(anyhow::Error::msg("unknown error").into())
+            Some(anyhow::Error::msg("unknown error"))
         } else {
             None
         }

lib/runtime/src/protocols/maybe_error.rs

@@ -20,7 +20,7 @@ pub trait MaybeError {
     fn from_err(err: Box<dyn Error + Send + Sync>) -> Self;
 
     /// Construct into an error instance.
-    fn err(&self) -> Option<Box<dyn Error + Send + Sync>>;
+    fn err(&self) -> Option<anyhow::Error>;
 
     /// Check if the current instance represents a success.
     fn is_ok(&self) -> bool {
@@ -46,8 +46,8 @@ mod tests {
                 message: err.to_string(),
             }
         }
-        fn err(&self) -> Option<Box<dyn Error + Send + Sync>> {
-            Some(anyhow::Error::msg(self.message.clone()).into())
+        fn err(&self) -> Option<anyhow::Error> {
+            Some(anyhow::Error::msg(self.message.clone()))
         }
     }