File size: 2,495 Bytes
0245be8 c095e79 0245be8 c095e79 0245be8 c095e79 0245be8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 |
from rasa_sdk.executor import CollectingDispatcher
from typing import Any, Text, Dict, List
from rasa_sdk import Action, Tracker
from dotenv import load_dotenv
from logging import getLogger
from enum import IntEnum
import os
logger = getLogger(__name__)
env = os.getenv("ENV", "local")
env_file = f".env-{env}"
load_dotenv(dotenv_path=f"../../.env-{env}")
MODEL_NAME = os.getenv("MODEL_NAME")
CHANNEL_TYPE = IntEnum(
"CHANNEL_TYPE", ["SMS", "TELEGRAM", "WHATSAPP", "EMAIL", "WEBSITE"]
)
logger = getLogger(__name__)
# -------------------------------------------------
# Custom Rasa action to trigger our RasaGPT LLM API
# -------------------------------------------------
class ActionGPTFallback(Action):
def name(self) -> str:
return "action_gpt_fallback"
def get_channel(self, channel: str) -> CHANNEL_TYPE:
if channel == "telegram":
return CHANNEL_TYPE.TELEGRAM
elif channel == "whatsapp":
return CHANNEL_TYPE.WHATSAPP
elif channel == "sms":
return CHANNEL_TYPE.SMS
elif channel == "email":
return CHANNEL_TYPE.EMAIL
else:
return CHANNEL_TYPE.WEBSITE
def run(
self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> List[Dict[Text, Any]]:
# ------------
# Get metadata
# ------------
data = tracker.latest_message
metadata = data['metadata'] if data and 'metadata' in data else None
response = metadata['response'] if metadata and 'response' in metadata else None
tags = metadata['tags'] if metadata and 'tags' in metadata else None
is_escalate = (
metadata['is_escalate'] if metadata and 'is_escalate' in metadata else None
)
# -----------------
# Escalate to human
# -----------------
if is_escalate is True:
response = f'{response} \n\n ⚠️💁 [ESCALATE TO HUMAN]'
# -----------------------
# Labels generated by LLM
# -----------------------
if tags is not None:
response = f'{response} \n\n 🏷️ {",".join(tags)}'
logger.debug(
f"""[🤖 ActionGPTFallback]
data: {data}
metadata: {metadata}
response: {response}
tags: {tags}
is_escalate: {is_escalate}
"""
)
dispatcher.utter_message(text=response)
return []
|