archived/intent_recognition/intent_recognition.py

"""
Module for generating intent
"""
import json
import random

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import make_pipeline

DATA_PATH = "data/formatted_data.json"

app = FastAPI()

def load_data(data_path: str):
    """route for loading the raw data"""
    with open(
        file=data_path,
        mode="r",
        encoding="utf-8",
    ) as data:
        data = json.load(data)
    return data

training_data = load_data(data_path=DATA_PATH)

class Query(BaseModel):
    query: str

class IntentModel:
    def __init__(self, data):
        self.training_data = data
        self.vectorizer = CountVectorizer()
        self.classifier = MultinomialNB()
        self.train()

    def train(self):
        f_patterns = []
        f_intents = []

        for entry in self.training_data:
            patterns = entry["patterns"]
            responses = entry["responses"]
            intent = entry["intent"]

            for pattern in patterns:
                f_patterns.append(pattern)
                f_intents.append(intent)

            for response in responses:
                f_patterns.append(response)
                f_intents.append(intent)

        # print(len(f_intents))

        # patterns = [
        #     pattern for entry in self.training_data
        #     for pattern in entry["patterns"]
        # ]
        # intents = [
        #     entry["intent"] for entry in self.training_data
        #     for _ in entry["patterns"]
        # ]
        vector = self.vectorizer.fit_transform(f_patterns)
        self.classifier.fit(vector, f_intents)

    def predict_intent(self, query):
        vectorized_query = self.vectorizer.transform([query])
        intent = self.classifier.predict(vectorized_query)
        return intent[0]

intent_model = IntentModel(training_data)

@app.post("/predict_intent")
def predict_intent(query: Query):
    try:
        intent = intent_model.predict_intent(query.query)
        responses = [
            response["responses"] for response in training_data
            if response["intent"] == intent
        ]
        response = random.choice(responses[0])

        if not response:
            raise HTTPException(status_code=500, detail="Intent not found")

        return {
            "intent": intent#,
            # "response": response
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))