optional streaming support for slack & whatsapp

add finish_reason to streaming

README.md

@@ -108,6 +108,12 @@ ngrok http 8080
 5. Copy the temporary access token there and set env var `WHATSAPP_ACCESS_TOKEN = XXXX`
 
 
+**(Optional) Use the test script to send yourself messages** 
+
+```bash
+python manage.py runscript test_wa_msg_send --script-args 104696745926402 +918764022384
+```
+Replace `+918764022384` with your number and `104696745926402` with the test number ID
 
 ## Dangerous postgres commands
 
@@ -178,3 +184,4 @@ rsync -P -a <username>@captain.us-1.gooey.ai:/home/<username>/fixture.json .
 createdb -T template0 $PGDATABASE
 pg_dump $SOURCE_DATABASE | psql -q $PGDATABASE
 ```
+

bots/admin.py

@@ -168,6 +168,7 @@ class BotIntegrationAdmin(admin.ModelAdmin):
             "Settings",
             {
                 "fields": [
+                    "streaming_enabled",
                     "show_feedback_buttons",
                     "analysis_run",
                     "view_analysis_results",

bots/models.py

@@ -571,6 +571,11 @@ class BotIntegration(models.Model):
         help_text="If provided, the message content will be analyzed for this bot using this saved run",
     )
 
+    streaming_enabled = models.BooleanField(
+        default=False,
+        help_text="If set, the bot will stream messages to the frontend (Slack only)",
+    )
+
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
 

bots/tasks.py

@@ -128,7 +128,7 @@ def send_broadcast_msg(
                     channel_is_personal=convo.slack_channel_is_personal,
                     username=bi.name,
                     token=bi.slack_access_token,
-                )
+                )[0]
             case _:
                 raise NotImplementedError(
                     f"Platform {bi.platform} doesn't support broadcasts yet"

daras_ai_v2/base.py

@@ -189,15 +189,15 @@ def setup_render(self):
 
     def refresh_state(self):
         _, run_id, uid = extract_query_params(gooey_get_query_params())
-        channel = f"gooey-outputs/{self.slug_versions[0]}/{uid}/{run_id}"
+        channel = self.realtime_channel_name(run_id, uid)
         output = realtime_pull([channel])[0]
         if output:
             st.session_state.update(output)
 
     def render(self):
         self.setup_render()
 
-        if self.get_run_state() == RecipeRunState.running:
+        if self.get_run_state(st.session_state) == RecipeRunState.running:
             self.refresh_state()
         else:
             realtime_clear_subs()
@@ -1307,12 +1307,13 @@ def _render_input_col(self):
             )
         return submitted
 
-    def get_run_state(self) -> RecipeRunState:
-        if st.session_state.get(StateKeys.run_status):
+    @classmethod
+    def get_run_state(cls, state: dict[str, typing.Any]) -> RecipeRunState:
+        if state.get(StateKeys.run_status):
             return RecipeRunState.running
-        elif st.session_state.get(StateKeys.error_msg):
+        elif state.get(StateKeys.error_msg):
             return RecipeRunState.failed
-        elif st.session_state.get(StateKeys.run_time):
+        elif state.get(StateKeys.run_time):
             return RecipeRunState.completed
         else:
             # when user is at a recipe root, and not running anything
@@ -1331,7 +1332,7 @@ def _render_output_col(self, submitted: bool):
 
         self._render_before_output()
 
-        run_state = self.get_run_state()
+        run_state = self.get_run_state(st.session_state)
         match run_state:
             case RecipeRunState.completed:
                 self._render_completed_output()
@@ -1458,13 +1459,16 @@ def call_runner_task(self, example_id, run_id, uid, is_api_call=False):
             run_id=run_id,
             uid=uid,
             state=st.session_state,
-            channel=f"gooey-outputs/{self.slug_versions[0]}/{uid}/{run_id}",
+            channel=self.realtime_channel_name(run_id, uid),
             query_params=self.clean_query_params(
                 example_id=example_id, run_id=run_id, uid=uid
             ),
             is_api_call=is_api_call,
         )
 
+    def realtime_channel_name(self, run_id, uid):
+        return f"gooey-outputs/{self.slug_versions[0]}/{uid}/{run_id}"
+
     def generate_credit_error_message(self, example_id, run_id, uid) -> str:
         account_url = furl(settings.APP_BASE_URL) / "account/"
         if self.request.user.is_anonymous:

daras_ai_v2/bot_integration_widgets.py

@@ -24,6 +24,11 @@ def general_integration_settings(bi: BotIntegration):
         ] = BotIntegration._meta.get_field("show_feedback_buttons").default
         st.session_state[f"_bi_analysis_url_{bi.id}"] = None
 
+    bi.streaming_enabled = st.checkbox(
+        "**📡 Streaming Enabled**",
+        value=bi.streaming_enabled,
+        key=f"_bi_streaming_enabled_{bi.id}",
+    )
     bi.show_feedback_buttons = st.checkbox(
         "**👍🏾 👎🏽 Show Feedback Buttons**",
         value=bi.show_feedback_buttons,