Scheduler to clear out inactive conversations

core/slack.py

@@ -5,12 +5,17 @@
     AgentTokenBufferMemory,
 )
 
+import datetime
+
 from abc import ABC, abstractmethod
 from statemachine import State
 from statemachine import StateMachine
 
 # --
 
+# the time in seconds, after which a conversation will be retried if inactive
+CONVERSATION_EXPIY_TIME=60
+
 class StatusStrategy(ABC):
     @abstractmethod
     def print(self, message) -> str:
@@ -68,6 +73,7 @@ class Conversation(StateMachine):
 
     def __init__(self, slack_client, channel, thread_ts):
 
+        self.last_activity = datetime.datetime.now()
         self.client = slack_client
         self.feedback = SlackStatus(slack_client=slack_client, channel=channel, thread_ts=thread_ts)
 
@@ -91,6 +97,9 @@ def __init__(self, slack_client, channel, thread_ts):
 
         super().__init__()
 
+    def is_expired(self):
+        return self.last_activity < datetime.datetime.now()-datetime.timedelta(seconds=CONVERSATION_EXPIY_TIME)
+
     def on_enter_greeting(self):            
             # mimic the first LLM response to get things started
             self.response_handle = {
@@ -103,6 +112,7 @@ def on_enter_running(self):
 
         print("running ..")
 
+        self.last_activity = datetime.datetime.now()
         self.feedback.set_tagline("Thinking ...")    
 
         # request chat completion
@@ -126,8 +136,7 @@ def on_enter_answered(self):
         slack_response = self.client.chat_postMessage(
             channel=self.channel, 
             thread_ts=self.thread_ts,
-            text=f"{response_content}")
-
+            text=f"{response_content}")       
 
         self.feedback.set_visible(False)
 

requirements.txt

@@ -28,4 +28,5 @@ tiktoken
 slack-bolt
 slack
 aiohttp
-httpx
+httpx
+apscheduler

slack_bot.py

@@ -2,6 +2,10 @@
 import signal
 import sys
 import time
+import threading
+
+import datetime as dt
+from apscheduler.schedulers.background import BackgroundScheduler
 
 from slack_bolt.adapter.socket_mode import SocketModeHandler
 from slack_sdk import WebClient
@@ -16,29 +20,47 @@
 app = App(token=os.environ['SLACK_BOT_TOKEN']) 
 client = WebClient(os.environ['SLACK_BOT_TOKEN'])
 socket = SocketModeHandler(app, os.environ['SLACK_APP_TOKEN'])
+scheduler = BackgroundScheduler()
 
 active_conversations = []
+conversation_lock = threading.Lock()
 
-# makre sure conversation are retried when bot stops
+# make sure conversation are retried when bot stops
 def graceful_shutdown(signum, frame):    
     print("Shutdown bot ...")
 
     # retire all active conversations
     [ref["conversation"].retire() for ref in active_conversations]    
     time.sleep(3)
 
+    # stop the scheduler
+    scheduler.shutdown(wait=False)
+
+    # stop the listener
     socket.disconnect()
     socket.close()
+
     sys.exit(0)
 
 def find_conversation(thread_ts):
-
-    for ref in active_conversations:
-        if(ref["id"]==thread_ts):
-            return ref["conversation"]
-        else:
-            return None
-
+    with conversation_lock:
+        for ref in active_conversations:
+            if(ref["id"]==thread_ts):
+                return ref["conversation"]
+            else:
+                return None
+
+def retire_inactive_conversation():
+    with conversation_lock:        
+        for ref in active_conversations:
+            conversation = ref["conversation"]
+            if(conversation.is_expired()):
+                if(conversation.current_state!='answered'):            
+                    conversation.retire()
+                    active_conversations.remove(ref)
+                else:
+                    print("Conversation is still active, keep for next cycle: ", str(conversation))    
+
 # This gets activated when the bot is tagged in a channel    
 # it will start a new thread that will hold the conversation
 @app.event("app_mention")
@@ -56,7 +78,7 @@ def handle_message_events(body, logger):
 
         response_channel = body["event"]["channel"]
         response_thread = body["event"]["event_ts"]
-
+        
         # register new conversation        
         conversation = Conversation(
             slack_client=client, 
@@ -100,9 +122,18 @@ def handle_message_events(event, say):
         # outside thread we ingore messages
         pass        
 
+
 if __name__ == "__main__":
 
     signal.signal(signal.SIGINT, graceful_shutdown)
     signal.signal(signal.SIGTERM, graceful_shutdown)
 
+    # scheduler reaper
+    scheduler.add_job(retire_inactive_conversation, 'interval', seconds=5, id='retirement_job')
+    scheduler.start()
+
+    # start listening for messages
     socket.start()
+
+
+