Merge pull request #59 from tukcomCD2024/fix/#56-backend-rabbitmq-con…

…nection

Fix/#56 FastAPI와 RabbitMQ 스트림 연결 끊기는 현상 예외처리

backend/ai_response_processor/news/contentqueue/rabbitmq_consumer.py

@@ -1,8 +1,9 @@
 import os
+import time
 from dotenv import load_dotenv
 import json
 import logging
-from pika import BlockingConnection, ConnectionParameters, PlainCredentials
+from pika import BlockingConnection, ConnectionParameters, PlainCredentials, exceptions
 
 from news.crud.news_summarizer import summarize_news
 
@@ -37,4 +38,8 @@ def callback(ch, method, properties, body):
 
 
 channel.basic_consume(queue=os.getenv('SUMMARY_QUEUE'), on_message_callback=callback, auto_ack=True)
-channel.start_consuming()
+try:
+    channel.start_consuming()
+except (exceptions.AMQPConnectionError, exceptions.StreamLostError) as e:
+    logger.error("Connection lost or failed, retrying... Error: {}".format(e))
+    time.sleep(10)

backend/ai_response_processor/news/contentqueue/rabbitmq_producer.py

@@ -1,7 +1,8 @@
 import os
+import time
 from dotenv import load_dotenv
 import json
-from pika import BlockingConnection, ConnectionParameters, PlainCredentials, BasicProperties
+from pika import BlockingConnection, ConnectionParameters, PlainCredentials, BasicProperties, exceptions
 
 from news.schema.message_item import MessageItem
 
@@ -26,16 +27,22 @@ def get_connection_params():
                                 heartbeat=600,
                                 blocked_connection_timeout=300)
 
+
 def send_message(message: MessageItem):
-    connection = BlockingConnection(get_connection_params())
-    channel = connection.channel()
-    channel.queue_declare(queue=CONFIG['queue_name'], durable=True)
-
-    props = BasicProperties(content_type=CONTENT_TYPE, delivery_mode=1)
-    serialized_message = json.dumps(message.__dict__)
-
-    channel.basic_publish(exchange=CONFIG['exchange_name'],
-                          routing_key=CONFIG['routing_key'],
-                          body=serialized_message,
-                          properties=props)
-    connection.close()
+    try:
+        connection = BlockingConnection(get_connection_params())
+        channel = connection.channel()
+        channel.queue_declare(queue=CONFIG['queue_name'], durable=True)
+
+        props = BasicProperties(content_type=CONTENT_TYPE, delivery_mode=1)
+        serialized_message = json.dumps(message.__dict__)
+
+        channel.basic_publish(exchange=CONFIG['exchange_name'],
+                              routing_key=CONFIG['routing_key'],
+                              body=serialized_message,
+                              properties=props)
+        connection.close()
+    except (exceptions.AMQPConnectionError, exceptions.StreamLostError) as e:
+        print("Connection failed, retrying in 5 seconds... Error: {}".format(e))
+        time.sleep(5)
+        send_message(message)

backend/ai_response_processor/news/crud/news_summarizer.py

@@ -17,7 +17,7 @@ def summarize_news(news_id: int, content: str):
 
     template = ChatPromptTemplate.from_messages(
         [
-            ("system", "You're a news summarizer. Also, the answer must be no more than 500 characters in Korean."),
+            ("system", "You're a news summarizer. Also, the answer must be summarized in Korean within 30% of the user's request."),
             ("user", "{raw_news_content}"),
         ]
     )