diff --git a/app/classifier.py b/app/classifier.py index cf8581e..b954ab1 100644 --- a/app/classifier.py +++ b/app/classifier.py @@ -1,14 +1,18 @@ -# Simple rule-based urgency classifier. Replace with sklearn/transformer later. -KEYWORDS_HIGH = {"emergency", "urgent", "help now", "asap", "please help", "immediately", "can't", "cant", "cannot", "can't breathe"} -KEYWORDS_MED = {"issue", "problem", "not working", "error", "fail", "unable"} +from app.config import settings +from typing import Tuple +# Wrapper that delegates to rule-based or ML classifier depending on settings.CLASSIFIER_MODE -def classify_urgency(text: str): - t = (text or "").lower() - for k in KEYWORDS_HIGH: - if k in t: - return "high", 0.99 - for k in KEYWORDS_MED: - if k in t: - return "medium", 0.75 - return "low", 0.3 +def classify_urgency(text: str) -> Tuple[str, float]: + mode = (settings.CLASSIFIER_MODE or "rule").lower() + if mode == "ml": + try: + from app.classifier_ml_wrapper import classify_urgency_ml + return classify_urgency_ml(text) + except Exception: + # fallback to rule-based + from app.classifier_rule import classify_urgency_rule + return classify_urgency_rule(text) + else: + from app.classifier_rule import classify_urgency_rule + return classify_urgency_rule(text) diff --git a/app/classifier_ml_wrapper.py b/app/classifier_ml_wrapper.py new file mode 100644 index 0000000..fa9d0f7 --- /dev/null +++ b/app/classifier_ml_wrapper.py @@ -0,0 +1,28 @@ +import os +import joblib +from app.config import settings +from typing import Tuple + +# lazy load to avoid startup cost +_model = None + + +def classify_urgency_ml(text: str) -> Tuple[str, float]: + global _model + model_path = os.getenv("MODEL_PATH") or settings.MODEL_PATH + if _model is None: + try: + _model = joblib.load(model_path) + except Exception: + # If model not available, fall back to rule-based + from app.classifier_rule import classify_urgency_rule + return classify_urgency_rule(text) + try: + probs = _model.predict_proba([text])[0] + idx = probs.argmax() + label = _model.classes_[idx] + confidence = float(probs[idx]) + return label, confidence + except Exception: + from app.classifier_rule import classify_urgency_rule + return classify_urgency_rule(text) diff --git a/app/classifier_rule.py b/app/classifier_rule.py new file mode 100644 index 0000000..130bcba --- /dev/null +++ b/app/classifier_rule.py @@ -0,0 +1,14 @@ +# Simple rule-based urgency classifier kept as a fallback and default implementation. +KEYWORDS_HIGH = {"emergency", "urgent", "help now", "asap", "please help", "immediately", "can't", "cant", "cannot", "can't breathe", "911"} +KEYWORDS_MED = {"issue", "problem", "not working", "error", "fail", "unable", "bug", "crash"} + + +def classify_urgency_rule(text: str): + t = (text or "").lower() + for k in KEYWORDS_HIGH: + if k in t: + return "high", 0.99 + for k in KEYWORDS_MED: + if k in t: + return "medium", 0.75 + return "low", 0.3 diff --git a/app/config.py b/app/config.py index 3030389..39fee21 100644 --- a/app/config.py +++ b/app/config.py @@ -9,6 +9,19 @@ class Settings(BaseSettings): PORT: int = 8000 WEBHOOK_SECRET_TOKEN: str = "" + # ML & model settings + CLASSIFIER_MODE: str = "rule" # 'rule' or 'ml' + MODEL_PATH: str = "models/urgency_model.joblib" + + # Messaging provider + WHATSAPP_PROVIDER: str = "twilio" + + # Meta settings (optional) + META_TOKEN: str = "" + META_PHONE_NUMBER_ID: str = "" + META_APP_SECRET: str = "" + META_VERIFY_TOKEN: str = "" + class Config: env_file = ".env" diff --git a/app/ml_classifier.py b/app/ml_classifier.py new file mode 100644 index 0000000..cfca252 --- /dev/null +++ b/app/ml_classifier.py @@ -0,0 +1,23 @@ +import joblib +from typing import Tuple +from sklearn.pipeline import Pipeline + +_model = None + + +def load_model(path: str) -> Pipeline: + global _model + if _model is None: + _model = joblib.load(path) + return _model + + +def predict_text(model: Pipeline, text: str) -> Tuple[str, float]: + """Return (label, confidence)""" + if not text: + return "low", 0.0 + probs = model.predict_proba([text])[0] + idx = probs.argmax() + label = model.classes_[idx] + confidence = float(probs[idx]) + return label, confidence diff --git a/app/reply_generator.py b/app/reply_generator.py index 15a36d6..393992a 100644 --- a/app/reply_generator.py +++ b/app/reply_generator.py @@ -1,25 +1,48 @@ import os import openai +import asyncio +from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type from app.config import settings openai.api_key = settings.OPENAI_API_KEY +SYSTEM_PROMPT = ( + "You are a concise, empathetic assistant that helps users over WhatsApp. " + "When appropriate, ask clarifying questions. When a message appears urgent, suggest escalation. " + "Keep replies short (1-3 sentences) and actionable." +) + + +@retry(wait=wait_exponential(min=1, max=10), stop=stop_after_attempt(3), retry=retry_if_exception_type(Exception)) +def _call_openai_sync(messages, model="gpt-3.5-turbo"): + resp = openai.ChatCompletion.create(model=model, messages=messages, max_tokens=200, temperature=0.3) + return resp.choices[0].message.content.strip() + + async def generate_reply(incoming_text: str, history): - # If OPENAI_API_KEY is set, use ChatCompletion for contextual replies, otherwise fallback. - if settings.OPENAI_API_KEY: - # Build a simple conversation for the LLM from history - messages = [ - {"role": "system", "content": "You are a helpful assistant that replies concisely and empathetically."} - ] - for item in history: - # assume message dicts have 'text' and maybe 'sender' (not enforced here) - messages.append({"role": "user", "content": item.get("text", "")}) - messages.append({"role": "user", "content": incoming_text}) - try: - resp = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=messages, max_tokens=200) - return resp.choices[0].message.content.strip() - except Exception: - # fallback - return "Thanks — I received your message. Can you tell me more?" - else: + # If OPENAI_API_KEY not set, return fallback + if not settings.OPENAI_API_KEY: + return "Thanks — I received your message. Can you tell me more?" + + # If history is long, compress/summarize the earlier part (simple approach: keep last 8 messages) + recent = history[-8:] if history and len(history) > 8 else history + + messages = [ + {"role": "system", "content": SYSTEM_PROMPT} + ] + + # Add history as user messages + if recent: + for item in recent: + text = item.get("text", "") + messages.append({"role": "user", "content": text}) + + messages.append({"role": "user", "content": incoming_text}) + + # Call OpenAI synchronously inside thread to keep compatibility with sync SDK + loop = asyncio.get_event_loop() + try: + reply = await loop.run_in_executor(None, _call_openai_sync, messages) + return reply + except Exception: return "Thanks — I received your message. Can you tell me more?" diff --git a/app/train_classifier.py b/app/train_classifier.py new file mode 100644 index 0000000..ff0bcb4 --- /dev/null +++ b/app/train_classifier.py @@ -0,0 +1,39 @@ +import argparse +import pandas as pd +from sklearn.linear_model import LogisticRegression +from sklearn.feature_extraction.text import TfidfVectorizer +from sklearn.pipeline import Pipeline +from sklearn.model_selection import train_test_split +from sklearn.metrics import classification_report +import joblib + + +def train(input_csv: str, output_path: str, test_size: float = 0.2, random_state: int = 42): + df = pd.read_csv(input_csv) + if "text" not in df.columns or "label" not in df.columns: + raise SystemExit("Input CSV must contain 'text' and 'label' columns") + X = df["text"].astype(str) + y = df["label"].astype(str) + + X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size, random_state=random_state) + + pipeline = Pipeline([ + ("tfidf", TfidfVectorizer(ngram_range=(1,2), max_features=10000)), + ("clf", LogisticRegression(max_iter=1000, class_weight='balanced')) + ]) + + pipeline.fit(X_train, y_train) + + preds = pipeline.predict(X_test) + print(classification_report(y_test, preds)) + + joblib.dump(pipeline, output_path) + print(f"Saved model to {output_path}") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Train urgency classifier") + parser.add_argument("--input", required=True, help="Path to labeled CSV (columns: text,label)") + parser.add_argument("--output", required=True, help="Output path for model joblib") + args = parser.parse_args() + train(args.input, args.output) diff --git a/data/sample_labeled.csv b/data/sample_labeled.csv new file mode 100644 index 0000000..fae0d05 --- /dev/null +++ b/data/sample_labeled.csv @@ -0,0 +1,10 @@ +text,label +"I need help now, my server is down",high +"Please respond asap, it's urgent",high +"I can't login and need access",high +"I'm seeing an error when I try to login",medium +"The API returns 500 sometimes",medium +"Facing intermittent failures in the app",medium +"What are your pricing plans?",low +"How do I upgrade?",low +"Hello, I have a general question",low diff --git a/docs/ML_AND_OPENAI.md b/docs/ML_AND_OPENAI.md new file mode 100644 index 0000000..e98001c --- /dev/null +++ b/docs/ML_AND_OPENAI.md @@ -0,0 +1,22 @@ +# ML & OpenAI usage + +This document explains how to train the TF-IDF + LogisticRegression urgency classifier and how to enable OpenAI replies. + +Training the classifier +1. Install requirements: pip install -r requirements.txt +2. Train using the sample data or your own labeled CSV (columns: text,label): + python app/train_classifier.py --input data/sample_labeled.csv --output models/urgency_model.joblib +3. The output model will be saved to models/urgency_model.joblib. Commit it if you want it in the repo (optional). + +Using the ML classifier at runtime +- Set environment variable to use the ML classifier: + export CLASSIFIER_MODE=ml +- Ensure MODEL_PATH (or settings.MODEL_PATH) points to the trained model file. + +OpenAI reply generator +- Set OPENAI_API_KEY in your .env file or environment variables. +- The reply generator will call OpenAI ChatCompletion (gpt-3.5-turbo by default). If the key is missing or OpenAI fails, the system falls back to a short canned reply. + +Notes +- The app defaults to the rule-based classifier if CLASSIFIER_MODE is not set to 'ml' or if the model fails to load. +- Keep model files out of source control for larger models; store in an artifacts bucket or volume for production. diff --git a/requirements.txt b/requirements.txt index 29de474..84003b7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,7 @@ twilio openai pydantic pytest +scikit-learn +joblib +pandas +tenacity diff --git a/tests/test_ml_classifier.py b/tests/test_ml_classifier.py new file mode 100644 index 0000000..3905e0d --- /dev/null +++ b/tests/test_ml_classifier.py @@ -0,0 +1,18 @@ +import pytest +from app.train_classifier import train +from sklearn.pipeline import Pipeline +import joblib +import os + + +def test_train_and_predict(tmp_path): + # train on sample data and ensure model saves and can predict + input_csv = os.path.join(os.getcwd(), "data", "sample_labeled.csv") + output = tmp_path / "model.joblib" + train(input_csv, str(output)) + assert output.exists() + model = joblib.load(str(output)) + assert isinstance(model, Pipeline) + # quick prediction checks + pred = model.predict(["This is urgent, please help now"])[0] + assert pred in {"high", "medium", "low"}