Proactive Notifications Apps (#1240)

## Overview
Issue: #1235 #1132

## Features
- Developer could create new apps with the proactive notification
capacity

## Usages
- Developer create a new app with capacity `proactive_notification`, and
`external_integration` to trigger the `transcript_processed`.
  ```
  {
    "id": "mentor-01",
    ...
    "capabilities": [
      "external_integration",
      "proactive_notification",
    ],
    "external_integration": {
      "triggers_on": "transcript_processed",
"webhook_url":
"https://based-hardware-development--plugins-api.modal.run/mentor",
      ...
    },
    "proactive_notification": {
        "scopes": ["user_name", "user_facts"]
    },
    ...
  }
  ```

- Webhook respond with the format:
  ```
  {
    "mentor": {
"prompt": "the prompt template, with `{{user_name}}` and
`{{user_facts}}`, Omi will use this prompt to ask LLM then send a
notification to user",
        "params": ["user_name", "user_facts"]
    }
  }
  ```

- (Optional) you could detect the codeword before responding with the
`mentor` instruction, either by using the LLM or just a simple regex:
  ```
    ai_names = ['Omi', 'Omie', 'Homi', 'Homie']
codewords = [f'hey {ai_name} what do you think' for ai_name in ai_names]
ai_name in ai_names]
    transcript = TranscriptSegment.segments_as_string(segments)
    text_lower = normalize(transcript)
pattern = r'\b(?:' + '|'.join(map(re.escape, [normalize(cw) for cw in
codewords])) + r')\b'
    if bool(re.search(pattern, text_lower)):
        # respond 
  ```


## Technical details

<img width="1074" alt="Screenshot 2024-11-06 at 10 43 46"
src="https://github.com/user-attachments/assets/09a18457-6116-454b-98a8-d36dd9a7b1d2">

## Examples

![382703065-af9b26f9-ddc5-4909-974c-c83ae4dc658e](https://github.com/user-attachments/assets/131cedfe-7588-4770-aa5e-9cea9a0e2d67)


## TODO
- [x] A dead simple app which could trigger codeword and send the
notification
- [x] Put the LLM to it
- [x] `user_facts` scope
- [x] Put capacity `proactive_notification`
- [x] Refine the document

## Future idea
- `memory_context` scope
- rate limits per plugin per user per 5s - 1

## 🚀 Deploy Steps
- [ ] Merge https://github.com/BasedHardware/omi/pull/1240/files
- [ ] Create plugin.
  ```
  curl -X 'POST' \

'https://based-hardware-development--backend-thinh-v2-api.modal.run/v3/plugins'
\
  -H 'accept: application/json' \
  -H 'authorization: <KEY>' \
  -H 'Content-Type: multipart/form-data' \
-F
'plugin_data={"name":"Mentor.01","author":"@thinh","description":"Mentor.01
- An AI-powered mentor, designed to elevate your meetings and help you
achieve your goals. With its insightful guidance and real-time support,
you'\''ll gain the confidence and skills to excel in every
interaction.","image":"/plugins/logos/mentor_01.jpg","capabilities":["external_integration","proactive_notification"],"external_integration":{"triggers_on":"transcript_processed","webhook_url":"https://based-hardware-development--plugins-api.modal.run/mentor","setup_completed_url":"https://based-hardware-development--plugins-api.modal.run/setup/mentor","setup_instructions_file_path":"https://raw.githubusercontent.com/BasedHardware/Omi/main/plugins/instructions/mentor_01/README.md"},"proactive_notification":{"scopes":["user_name","user_facts"]},"deleted":false,"private":false}'
\
  -F 'file=@mentor_01.jpg;type=image/jpeg'
  ```
- [ ] Deploy Pusher service
- [ ] Deploy Plugin service

backend/models/notification_message.py

@@ -1,11 +1,10 @@
 
-from typing import  Optional
+from typing import Optional
 import uuid
 from datetime import datetime
 from pydantic import BaseModel, Field
 
 
-
 class NotificationMessage(BaseModel):
     id: str = Field(default_factory=lambda: str(uuid.uuid4()))
     created_at: str = Field(default_factory=lambda: datetime.now().isoformat())
@@ -14,15 +13,15 @@ class NotificationMessage(BaseModel):
     from_integration: str
     type: str
     notification_type: str
-
+    text: Optional[str] = ""
 
     @staticmethod
     def get_message_as_dict(
             message: 'NotificationMessage',
     ) -> dict:
-        
+
         message_dict = message.dict()
-        
+
         # Remove 'plugin_id' if it is None
         if message.plugin_id is None:
             del message_dict['plugin_id']

backend/models/plugin.py

@@ -34,6 +34,8 @@ class ExternalIntegration(BaseModel):
     auth_steps: Optional[List[AuthStep]] = []
     # setup_instructions_markdown: str = ''
 
+class ProactiveNotification(BaseModel):
+    scopes: Set[str]
 
 class Plugin(BaseModel):
     id: str
@@ -53,6 +55,7 @@ class Plugin(BaseModel):
     deleted: bool = False
     trigger_workflow_memories: bool = True  # default true
     installs: int = 0
+    proactive_notification: Optional[ProactiveNotification] = None
 
     def get_rating_avg(self) -> Optional[str]:
         return f'{self.rating_avg:.1f}' if self.rating_avg is not None else None
@@ -75,6 +78,11 @@ def triggers_on_memory_creation(self) -> bool:
     def triggers_realtime(self) -> bool:
         return self.works_externally() and self.external_integration.triggers_on == 'transcript_processed'
 
+    def fitler_proactive_notification_scopes(self, params: [str]) -> []:
+        if not self.proactive_notification:
+            return []
+        return [param for param in params if param in self.proactive_notification.scopes]
+
     def get_image_url(self) -> str:
         return f'https://raw.githubusercontent.com/BasedHardware/Omi/main{self.image}'
 

backend/routers/chat.py

@@ -11,7 +11,6 @@
 from models.memory import Memory
 from utils.llm import initial_chat_message
 from utils.other import endpoints as auth
-from utils.plugins import get_plugin_by_id
 from utils.retrieval.graph import execute_graph_chat
 
 router = APIRouter()
@@ -107,4 +106,4 @@ def get_messages(uid: str = Depends(auth.get_current_user_uid)):
     messages = chat_db.get_messages(uid, limit=100, include_memories=True)  # for now retrieving first 100 messages
     if not messages:
         return [initial_message_util(uid)]
-    return messages
+    return messages

backend/routers/plugins.py

@@ -9,7 +9,7 @@
 from fastapi.params import File, Form, Header
 
 from database.plugins import get_plugin_usage_history, add_public_plugin, add_private_plugin, \
-     change_plugin_approval_status, \
+    change_plugin_approval_status, \
     get_plugin_by_id_db, change_plugin_visibility_db, get_unapproved_public_plugins_db, public_plugin_id_exists_db, \
     private_plugin_id_exists_db
 from database.redis_db import set_plugin_review, enable_plugin, disable_plugin, increase_plugin_installs_count, \
@@ -137,7 +137,7 @@ def add_plugin(plugin_data: str = Form(...), file: UploadFile = File(...), uid=D
     data['approved'] = False
     data['id'] = data['name'].replace(' ', '-').lower()
     data['uid'] = uid
-    if data['private']:
+    if 'private' in data and data['private']:
         data['id'] = data['id'] + '-private'
         if private_plugin_id_exists_db(data['id'], uid):
             data['id'] = data['id'] + '-' + ''.join([str(random.randint(0, 9)) for _ in range(5)])

backend/routers/transcribe_v2.py

@@ -17,7 +17,7 @@
 from models.message_event import MemoryEvent, MessageEvent
 from utils.memories.location import get_google_maps_location
 from utils.memories.process_memory import process_memory
-from utils.plugins import trigger_external_integrations, trigger_realtime_integrations
+from utils.plugins import trigger_external_integrations
 from utils.stt.streaming import *
 from utils.webhooks import send_audio_bytes_developer_webhook, realtime_transcript_webhook, \
     get_audio_bytes_webhook_seconds

backend/utils/llm.py

@@ -814,3 +814,22 @@ def provide_advice_message(uid: str, segments: List[TranscriptSegment], context:
     ```
     """.replace('    ', '').strip()
     return llm_mini.with_structured_output(OutputMessage).invoke(prompt).message
+
+# **************************************************
+# ************* MENTOR PLUGIN **************
+# **************************************************
+
+def get_metoring_message(uid: str, plugin_prompt: str, params: [str]) -> str:
+    user_name, facts_str = get_prompt_facts(uid)
+
+    prompt = plugin_prompt
+    for param in params:
+        if param == "user_name":
+            prompt = prompt.replace("{{user_name}}", user_name)
+            continue
+        if param == "user_facts":
+            prompt = prompt.replace("{{user_facts}}", facts_str)
+            continue
+    prompt = prompt.replace('    ', '').strip()
+
+    return llm_mini.invoke(prompt).content

backend/utils/plugins.py

@@ -13,6 +13,7 @@
 from models.plugin import Plugin, UsageHistoryType
 from utils.notifications import send_notification
 from utils.other.endpoints import timeit
+from utils.llm import get_metoring_message
 
 
 # ***********************************
@@ -176,7 +177,7 @@ async def trigger_realtime_integrations(uid: str, segments: list[dict]):
 
 
 def _trigger_realtime_integrations(uid: str, token: str, segments: List[dict]) -> dict:
-    plugins: List[Plugin] = get_plugins_data(uid, include_reviews=False)
+    plugins: List[Plugin] = get_plugins_data_from_db(uid, include_reviews=False)
     filtered_plugins = [
         plugin for plugin in plugins if
         plugin.triggers_realtime() and plugin.enabled and not plugin.deleted
@@ -206,11 +207,29 @@ def _single(plugin: Plugin):
             response_data = response.json()
             if not response_data:
                 return
+
+            # message
             message = response_data.get('message', '')
-            print('Plugin', plugin.id, 'response:', message)
+            mentor = response_data.get('mentor', None)
+            print('Plugin', plugin.id, 'response:', message, mentor)
             if message and len(message) > 5:
                 send_plugin_notification(token, plugin.name, plugin.id, message)
                 results[plugin.id] = message
+
+            # proactive_notification
+            if plugin.has_capability("proactive_notification"):
+                mentor = response_data.get('mentor', None)
+                if mentor:
+                    prompt = mentor.get('prompt', '')
+                    if len(prompt) > 0 and len(prompt) <= 5000:
+                        params = mentor.get('params', [])
+                        message = get_metoring_message(uid, prompt, plugin.fitler_proactive_notification_scopes(params))
+                        if message and len(message) > 5:
+                            send_plugin_notification(token, plugin.name, plugin.id, message)
+                            results[plugin.id] = message
+                    elif len(prompt) > 5000:
+                        print(f"Plugin {plugin.id} prompt too long, length: {len(prompt)}/5000")
+
         except Exception as e:
             print(f"Plugin integration error: {e}")
             return
@@ -225,6 +244,7 @@ def _single(plugin: Plugin):
         if not message:
             continue
         messages.append(add_plugin_message(message, key, uid))
+
     return messages
 
 

backend/utils/pusher.py

@@ -17,7 +17,6 @@
 from models.message_event import MemoryEvent, MessageEvent
 from utils.memories.location import get_google_maps_location
 from utils.memories.process_memory import process_memory
-from utils.plugins import trigger_external_integrations, trigger_realtime_integrations
 from utils.stt.streaming import *
 from utils.webhooks import send_audio_bytes_developer_webhook, realtime_transcript_webhook, \
     get_audio_bytes_webhook_seconds

plugins/example/basic/mentor.py

@@ -0,0 +1,77 @@
+import re
+import time
+
+from fastapi import APIRouter
+
+from models import *
+from db import get_upsert_segment_to_transcript_plugin
+
+router = APIRouter()
+
+scan_segment_session = {}
+
+# *******************************************************
+# ************ Basic Mentor Plugin ************
+# *******************************************************
+
+@router.post('/mentor', tags=['mentor', 'basic', 'realtime'], response_model=MentorEndpointResponse)
+def mentoring(data: RealtimePluginRequest):
+    def normalize(text):
+        return re.sub(r' +', ' ',re.sub(r'[,?.!]', ' ', text)).lower().strip()
+
+    # Add segments by session_id
+    session_id = data.session_id
+    segments = get_upsert_segment_to_transcript_plugin('mentor-01', session_id, data.segments)
+    scan_segment = scan_segment_session[session_id] if session_id in scan_segment_session and len(segments) > len(data.segments) else 0
+
+    # Detect codewords
+    ai_names = ['Omi', 'Omie', 'Homi', 'Homie']
+    codewords = [f'hey {ai_name} what do you think' for ai_name in ai_names]
+    scan_segments = TranscriptSegment.combine_segments([], segments[scan_segment:])
+    if len(scan_segments) == 0:
+        return {}
+    text_lower = normalize(scan_segments[-1].text)
+    pattern = r'\b(?:' + '|'.join(map(re.escape, [normalize(cw) for cw in codewords])) + r')\b'
+    if not bool(re.search(pattern, text_lower)):
+        return {}
+
+    # Generate mentoring prompt
+    scan_segment_session[session_id] = len(segments)
+    transcript = TranscriptSegment.segments_as_string(segments)
+
+    user_name = "{{user_name}}"
+    user_facts = "{{user_facts}}"
+
+    prompt = f"""
+    You are an experienced mentor, that helps people achieve their goals during the meeting.
+    You are advising {user_name} right now.
+
+    {user_facts}
+
+    The following is a {user_name}'s conversation, with the transcripts, that {user_name} had during the meeting.
+    {user_name} wants to get the call-to-action advice to move faster during the meetting based on the conversation.
+
+    First, identify the topics or problems that {user_name} is discussing or trying to resolve during the meeting, and then provide advice specific to those topics or problems. If you cannot find the topic or problem of the meeting, respond with an empty message.
+
+    The advice must focus on the specific object mentioned in the conversation. The object could be a product, a person, or an event.
+
+    The response must follow this format:
+    Noticed you are trying to <meeting topics or problems>.
+    If I were you, I'd <actions>.
+
+    Remember {user_name} is busy so this has to be very efficient and concise.
+    Respond in at most 100 words.
+
+    Output your response in plain text, without markdown.
+    ```
+    ${transcript}
+    ```
+    """.replace('    ', '').strip()
+
+    return {'session_id': data.session_id,
+            'mentor': {'prompt': prompt,
+                       'params': ['user_name', 'user_facts']}}
+
+@ router.get('/setup/mentor', tags=['mentor'])
+def is_setup_completed(uid: str):
+    return {'is_setup_completed': True}

plugins/example/db.py

@@ -148,3 +148,29 @@ def get_ahda_url(uid: str) -> str:
 def get_ahda_os(uid: str) -> str:
     val = r.get(f'ahda_os:{uid}')
     return val.decode('utf-8') if val else None
+
+
+# *******************************************************
+# ************ MENTOR PLUGIN UTILS ***********
+# *******************************************************
+
+def get_upsert_segment_to_transcript_plugin(plugin_id: str, session_id: str, new_segments: list[TranscriptSegment]) -> List[dict]:
+    key = f'plugin:{plugin_id}:session:{session_id}:transcript_segments'
+    segments = r.get(key)
+    if not segments:
+        segments = []
+    else:
+        segments = eval(segments)
+
+    segments.extend([segment.dict() for segment in new_segments])
+
+    # keep 1000
+    if len(segments) > 1000:
+        segments = segments[-1000:]
+
+    r.set(key, str(segments))
+
+    # expire 5m
+    r.expire(key, 60 * 5)
+
+    return [TranscriptSegment(**segment) for segment in segments]

plugins/example/main.py

@@ -8,8 +8,7 @@
 from basic import memory_created as basic_memory_created_router
 from oauth import memory_created as oauth_memory_created_router
 from zapier import memory_created as zapier_memory_created_router
-from ahda import client as ahda_realtime_transcription_router
-
+# from ahda import client as ahda_realtime_transcription_router
 # from advanced import openglass as advanced_openglass_router
 
 # ************* @DEPRECATED **************
@@ -19,6 +18,7 @@
 # 3. Didn't find killer use cases.
 # from advanced import realtime as advanced_realtime_router
 # from basic import realtime as basic_realtime_router
+from basic import mentor as basic_realtime_mentor_router
 # ****************************************
 
 app = FastAPI()
@@ -33,12 +33,12 @@
 
 @modal_app.function(
     image=(
-            Image.debian_slim()
-            # .apt_install('libgl1-mesa-glx', 'libglib2.0-0')
-            .pip_install_from_requirements('requirements.txt')
+        Image.debian_slim()
+        # .apt_install('libgl1-mesa-glx', 'libglib2.0-0')
+        .pip_install_from_requirements('requirements.txt')
     ),
     keep_warm=1,  # need 7 for 1rps
-    memory=(128, 256),
+    memory=(128, 512),
     cpu=1,
     allow_concurrent_inputs=10,
 )
@@ -50,8 +50,9 @@ def api():
 app.include_router(basic_memory_created_router.router)
 app.include_router(oauth_memory_created_router.router)
 app.include_router(zapier_memory_created_router.router)
-#app.include_router(ahda_realtime_transcription_router.router)
+# app.include_router(ahda_realtime_transcription_router.router)
 
+app.include_router(basic_realtime_mentor_router.router)
 # app.include_router(basic_realtime_router.router)
 # app.include_router(advanced_realtime_router.router)
 # app.include_router(advanced_openglass_router.router)