Created cricbot using genai

.gitignore

@@ -125,3 +125,5 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+exploration/

constants/constants.py

@@ -0,0 +1,12 @@
+class Constants:
+
+    INTENT_IDENTIFIER_GPT_MODEL: str = "gpt-3.5-turbo"
+    RESPONSE_GENERATOR_GPT_MODEL: str = "gpt-4o"
+    BASE_FILE_PATH: str = "prompts"
+    INTENT_IDENTIFIER_SYS_MSG_FILE_NAME: str = "intent_identifier_system_message.txt"
+    LIVE_SCORE_RESPONSE_PROMPT: str = "live_score_response_prompt.txt"
+    FALLBACK_RESPONSE_PROMPT: str = "fallback_response_prompt.txt"
+
+    REASON_NOT_PRESENT: str = "Not able to understand the given input."
+    TEAMS_NOT_PRESENT_REASON: str = "Couldn't identify teams from your response."
+    MATCH_NOT_PRESENT_REASON: str = "Couldn't find a match between {team1} and {team2}"

main.py

@@ -0,0 +1,15 @@
+import os
+from dotenv import find_dotenv, load_dotenv
+from services.cricbot_service import CricbotService
+
+# loading the API Keys from .env
+load_dotenv(find_dotenv(), override=True)
+
+if __name__ == "__main__":
+    openai_api_key = os.environ.get('OPENAI_API_KEY')
+    cricbot_service = CricbotService(openai_api_key)
+    while True:
+        user_input = input("User: ")
+        if user_input == "exit":
+            break
+        cricbot_service.bot_response(user_input)

models/match_details.py

@@ -0,0 +1,15 @@
+class MatchDetails:
+    def __init__(self):
+        self.id = None
+        self.status = ''
+        self.team_1 = TeamScoreDetails()
+        self.team_2 = TeamScoreDetails()
+
+class TeamScoreDetails:
+    def __init__(self):
+        self.name = ''
+        self.abr = ''
+        self.run = ''
+        self.wicket = ''
+        self.over = ''
+

prompts/fallback_response_prompt.txt

@@ -0,0 +1,21 @@
+Role:
+You are an expert in writing cricket related articles.
+
+Context:
+We are building a chatbot about Cricket where you need to generate a fallback response.
+
+Instructions:
+- Generate response based on the reason explaing why response could not be generated.
+- Fetching player and team rankings is under development. Will be available shortly.
+- Crickbot supports only live score right now. All other cricket related messages should be handled appropriately
+- Only generate response related to cricket.
+- Dont hallucinate.
+- Dont generate biased response.
+- Dont include any political sentiment.
+- Be polite and professional. 
+- Consider edge cases like unclear intents and provide reasonable interpretations.
+- Ensure all outputs are contextually accurate and specific to Cricket.
+
+User: {user_input}
+Reason: {reason}
+

prompts/intent_identifier_system_message.txt

@@ -0,0 +1,91 @@
+Role:
+You are an expert in classifying intent and identifying entities from a plain text.
+
+Context:
+We are building a chatbot about Cricket where you need to find intent and entities in the message.
+
+Tasks:
+You need to perform following tasks:
+- Identify the intent and entities in the given text. Possible intents and their corresponding entities are:
+    1. 'live_score': This means that user is trying to find the live score of a cricket match between 2 teams. Entities to find are:
+        * 'team1' - Cricket team 1 [Mandatory]
+        * 'team2' - Cricket team 2 [Mandatory]
+    2. 'player_ranking': This means that user is trying to find the international ranking of a player. Also, identify the format and skill mentioned in the text. Entities to find are:
+        * 'player' - Player name [Mandatory]
+        * 'format' - Format (test, odi, t20) [Optional]
+        * 'skill' - Skill (batting, bowling, allrounder) [Optional]
+    3. 'team_ranking': This means that user is trying to find the international ranking of a team. Identify the format as well, if present. Entities to find are:
+        * 'team' - Team name [Mandatory]
+        * 'format' - Format (test, odi, t20) [Optional]
+    4. 'fallback': If text doesn't fit in any of the above intents, then return this intent. Entities to find is:
+        * 'reason' - output the reason because of which you are not able to indetify the intent in the given text.
+- Return the response in json format. It should be in following structure
+    {
+        "intent": "<value>",
+        "entities": {
+            "team1": "<value>",
+            "team2": "<value>"
+            ...
+        }
+    }
+- Consider edge cases like multiple entities in a message or unclear intents and provide reasonable interpretations.
+- Ensure all outputs are contextually accurate and specific to Cricket.
+
+Example1:
+Input: Get me live scores of cricket match between india and australia.
+Output: {
+    "intent": "live_score",
+    "entities": {
+        "team1": "india",
+        "team2": "australia"
+    }
+}
+
+Example2:
+Input: mumbai indians vs gujarat titans
+Output: {
+    "intent": "live_score",
+    "entities": {
+        "team1": "mumbai indians",
+        "team2": "gujarat titans"
+    }
+}
+
+Example3: 
+Input: Fetch me ranking of Virat in odi
+Output: {
+    "intent": "player_ranking",
+    "entities": {
+        "player": "Virat",
+        "format": "odi"
+    }
+}
+
+Example4: 
+Input: Fetch me batting ranking of david warner in t20i
+Output: {
+    "intent": "player_ranking",
+    "entities": {
+        "player": "david warner",
+        "format": "t20",
+        "skill": "batting"
+    }
+}
+
+Example5:
+Input: Find ranking of india
+Output: {
+    "intent": "team_ranking",
+    "entities": {
+        "team": "india"
+    }
+}
+
+Example6:
+Input: Show me live score of football match
+Output: {
+    "intent": "fallback"
+    "entities": {
+        "reason": "Cannot show live score of a football match."
+    }
+}

prompts/live_score_response_prompt.txt

@@ -0,0 +1,30 @@
+Role:
+You are an expert in writing cricket related articles.
+
+Context:
+We are building a chatbot about Cricket where you need to generate a response for the live score of the match between 2 teams. Also, include the current status of match, if present.
+
+Instructions:
+- Summarize the live score in textual format.
+- Live score should be mentioned in the following format as well.
+<t1_name> vs <t2_name>
+<t1_abr>: <t1_run>/<t1_wkt> (<t1_ovr>)
+<t2_abr>: <t2_run>/<t2_wkt> (<t2_ovr>)
+<status>
+- Dont mention None if scores are not available for a team.
+- Batting team should be first in live score summary.
+- Only generate response related to cricket.
+- Dont hallucinate.
+- Dont generate biased response.
+- Dont include any political sentiment.
+- Be polite and professional. 
+- Consider edge cases like unclear intents and provide reasonable interpretations.
+- Ensure all outputs are contextually accurate and specific to Cricket.
+
+User: {user_input}
+Live Status of match:
+{t1_name} vs {t2_name}
+{t1_abr}: {t1_run}/{t1_wkt} ({t1_ovr})
+{t2_abr}: {t2_run}/{t2_wkt} ({t2_ovr})
+{status}
+

services/cricbot_service.py

@@ -0,0 +1,49 @@
+from constants.constants import Constants
+from services.intent_identifier_service import IntentIdentifierService
+from services.live_score_service import LiveScoreService
+from services.response_generator_service import ResponseGeneratorService
+
+
+class CricbotService:
+    def __init__(self, openai_api_key: str):
+        self.__intent_identifier_service = IntentIdentifierService(openai_api_key)
+        self.__live_score_service = LiveScoreService()
+        self.__response_generator_service = ResponseGeneratorService(openai_api_key)
+
+    def bot_response(self, user_input: str):
+        intent_details = self.__intent_identifier_service.invoke(user_input)
+        response = ""
+        match intent_details.get('intent'):
+            case 'live_score':
+                response = self.__handle_live_score_intent(user_input, intent_details)
+            case _:
+                response = self.__handle_fallback_intent(user_input, intent_details)
+        print("Cricbot:", response)
+
+    def __handle_live_score_intent(self, user_input: str, intent_details) -> str:
+        entities = intent_details.get('entities')
+        if 'team1' not in entities or 'team2' not in entities:
+            return self.__get_fallback_response(user_input, Constants.TEAMS_NOT_PRESENT_REASON)
+
+        live_score = self.__live_score_service.fetch_live_score(entities['team1'], entities['team2'])
+
+        if live_score is None:
+            return self.__get_fallback_response(
+                user_input, 
+                Constants.MATCH_NOT_PRESENT_REASON.format(
+                    team1=entities['team1'], 
+                    team2=entities['team2']
+                )
+            )
+
+        return self.__response_generator_service.get_live_score_response(user_input, live_score)
+
+    def __handle_fallback_intent(self, user_input: str, intent_details) -> str:
+        entities = intent_details.get('entities')
+        if 'reason' not in entities or not entities['reason']:
+            return self.__get_fallback_response(user_input, Constants.REASON_NOT_PRESENT)
+        return self.__get_fallback_response(user_input, entities['reason'])
+
+    def __get_fallback_response(self, user_input: str, reason: str) -> str:
+        return self.__response_generator_service.get_fallback_response(user_input, reason)
+

services/intent_identifier_service.py

@@ -0,0 +1,30 @@
+import json
+from typing import Any, List
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import SystemMessage, BaseMessage, HumanMessage
+from constants.constants import Constants
+from utils.common_util import read_prompt_from_file
+
+class IntentIdentifierService:
+    def __init__(self, openai_api_key: str):
+        self.__llm_chain = ChatOpenAI(
+            model=Constants.INTENT_IDENTIFIER_GPT_MODEL, 
+            api_key=openai_api_key
+        )
+
+    def invoke(self, user_text) -> Any:
+        messages = self.__get_llm_messages(user_text)
+        output = self.__llm_chain.invoke(messages)
+        return json.loads(output.content)
+
+    def __get_llm_messages(self, user_text) -> List[BaseMessage]:
+        return [
+            self.__get_system_message(),
+            self.__get_human_message(user_text)
+        ]
+
+    def __get_system_message(self) -> SystemMessage:
+        return SystemMessage(content=read_prompt_from_file(Constants.INTENT_IDENTIFIER_SYS_MSG_FILE_NAME))
+
+    def __get_human_message(self, user_text) -> HumanMessage:
+        return HumanMessage(content=user_text)

services/live_score_service.py

@@ -0,0 +1,63 @@
+from datetime import datetime
+from typing import Any, List
+import requests
+
+from models.match_details import MatchDetails
+from utils.common_util import clean_team_name, clean_team_names
+
+class LiveScoreService:
+    def fetch_live_score(self, team1, team2) -> MatchDetails:
+        live_matches = self.__fetch_all_live_matches()
+        return self.__find_match(live_matches, team1, team2)
+
+    def __fetch_all_live_matches(self) -> List[MatchDetails]:
+        cur_date = datetime.today().strftime("%Y%m%d")
+        url = f"https://prod-public-api.livescore.com/v1/api/app/date/cricket/{cur_date}/5.30?locale=en&MD=1"
+        response = requests.get(url)
+        if response.ok:
+            return self.__process_matches_data(response.json())
+        else:
+            return []
+
+    def __process_matches_data(self, response: Any) -> List[MatchDetails]:
+        matches = []
+        for stage in response['Stages']:
+            for event in stage['Events']:
+                match_details = MatchDetails()
+                match_details.id = event.get('Eid')
+                match_details.status = event.get('ECo') or ''
+
+                team1 = event.get('T1')
+                if team1 and len(team1) > 0:
+                    match_details.team_1.name = team1[0].get('Nm')
+                    match_details.team_1.abr = team1[0].get('Abr')
+                match_details.team_1.run = event.get('Tr1C1') or ''
+                match_details.team_1.wicket = event.get('Tr1CW1') or ''
+                match_details.team_1.over = event.get('Tr1CO1') or ''
+
+                team2 = event.get('T2')
+                if team2 and len(team2) > 0:
+                    match_details.team_2.name = team2[0].get('Nm')
+                    match_details.team_2.abr = team2[0].get('Abr')
+                match_details.team_2.run = event.get('Tr2C1') or ''
+                match_details.team_2.wicket = event.get('Tr2CW1') or ''
+                match_details.team_2.over = event.get('Tr2CO1') or ''
+                matches.append(match_details)
+        return matches
+
+    def __find_match(self, matches: List[MatchDetails], team1: str, team2: str) -> MatchDetails:
+        if team1.lower() == team2.lower():
+            return None
+        for match in matches:
+            teams = clean_team_names([
+                match.team_1.name, 
+                match.team_2.name, 
+                match.team_1.abr, 
+                match.team_2.abr
+            ])
+            if clean_team_name(team1) in teams and clean_team_name(team2) in teams:
+                return match
+        return None
+
+
+