update worker.py

worker.py

@@ -52,9 +52,8 @@ class RabbitMQWorker:
                 self.publisher_channel.basic_publish(
                     exchange="",
                     routing_key="ml_server",
-                    body=json.dumps(body_dict),
+                    body=json.dumps(body_dict).encode('utf-8'),
                     properties=pika.BasicProperties(
-                        delivery_mode=2,
                         headers=headers
                     )
                 )
@@ -84,15 +83,15 @@ class RabbitMQWorker:
         logger.info("[Worker %s] Received message: %s, headers: %s", thread_id, body, headers)
 
         try:
-            if headers.get("request_type") == "process_files":
-                ch.basic_ack(delivery_tag=method.delivery_tag)
-                contexts = []
-
-                body_dict = json.loads(body)
+            ch.basic_ack(delivery_tag=method.delivery_tag)
+            contexts = []
 
-                data = body_dict.get("data", {})
-                input_files = data.get("input_files", [])
-                
+            body_dict = json.loads(body)
+            
+            pattern = body_dict.get("pattern")
+            if pattern == "process_files":
+                data = body_dict.get("data")
+                input_files = data.get("input_files")
                 logger.info("[Worker %s] Found %d file(s) to process.", thread_id, len(input_files))
 
                 for files in input_files:
@@ -106,12 +105,13 @@ class RabbitMQWorker:
                         err_str = f"Error processing file {files.get('key', '')}: {e}"
                         logger.error(err_str)
                         contexts.append({"key": files.get("key", ""), "body": err_str})
-
-
+    
+    
                 data["md_context"] = contexts
                 # topics = data.get("topics", [])
-
-                headers["request_type"] = "question_extraction_update_from_gpu_server"
+    
+                body_dict["pattern"] = "question_extraction_update_from_gpu_server"
+                body_dict["data"] = data
                 
                 # Publish results
                 if self.publish_message(body_dict, headers):
@@ -121,9 +121,10 @@ class RabbitMQWorker:
                     logger.error("[Worker %s] Failed to publish results.", thread_id)
                 
                 logger.info("[Worker %s] Contexts: %s", thread_id, contexts)
+                
             else:
                 ch.basic_ack(delivery_tag=method.delivery_tag)
-                logger.warning("[Worker %s] Unknown process type in headers: %s", thread_id, headers.get("request_type"))
+                logger.warning("[Worker %s] Unknown pattern type in headers: %s", thread_id, pattern)
                 
         except Exception as e:
             logger.error("Error in callback: %s", e)