Classification with Large Language Models

import json
from openai import OpenAI

client = OpenAI()

def classify_with_llm(text, labels):
    """Classify text into one of the given labels using an LLM."""
    prompt = f"""Classify the following text into exactly one of these labels:
{json.dumps(labels)}

Text: "{text}"

Respond with JSON: {{"label": "...", "reason": "..."}}"

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,  # Deterministic for classification
    )

    return json.loads(response.choices[0].message.content)

# Example usage
result = classify_with_llm(
    "This product is incredible, best purchase all year!",
    ["positive", "negative", "neutral"]
)
print(result)
# {"label": "positive", "reason": "Strongly positive language..."}

# === Approach 1: Prompt-based classification ===
# Fast to set up, no training data needed
def prompt_classify(text, labels):
    prompt = f"Classify into one of {labels}: \"{text}\"\nLabel:"
    # response = llm.generate(prompt, temperature=0)
    # return response.strip()
    pass

# === Approach 2: Fine-tuned classifier ===
# Requires labeled data, but faster and cheaper at scale
from transformers import pipeline

# After fine-tuning on your labeled dataset:
classifier = pipeline(
    "text-classification",
    model="./my-fine-tuned-classifier",
)

# Inference: ~5ms, deterministic, well-calibrated
result = classifier("The delivery was two weeks late.")
print(result)
# [{"label": "complaint", "score": 0.94}]

# Cost comparison at 100K predictions/day:
# Prompting: ~$50-200/day (API costs)
# Fine-tuned: ~$5-10/day (self-hosted GPU)

# === Zero-shot: instructions only ===
zero_shot_prompt = """Classify the customer message as: billing, technical, or general.

Message: "My card was charged twice for the same order."
Label:"""

# === Few-shot: instructions + examples ===
few_shot_prompt = """Classify customer messages. Examples:

"I can't log into my account" -> technical
"Please update my billing address" -> billing
"What are your business hours?" -> general

Message: "My card was charged twice for the same order."
Label:"""

# Few-shot typically improves accuracy by 10-30%
# on ambiguous or organization-specific labels

# Pro tip: use diverse, representative examples
# that cover edge cases, not just easy examples
def select_examples(examples, n_per_class=2):
    """Select diverse examples covering each class."""
    selected = []
    for label in set(e["label"] for e in examples):
        class_examples = [e for e in examples if e["label"] == label]
        # Pick diverse examples (e.g., by embedding distance)
        selected.extend(class_examples[:n_per_class])
    return selected

# A well-structured taxonomy definition for an LLM classifier
taxonomy = {
    "billing": {
        "description": "Issues related to charges, invoices, refunds, or payment methods",
        "includes": ["double charges", "refund requests", "invoice errors"],
        "excludes": ["account access issues", "feature requests"],
        "examples": [
            "I was charged twice for my subscription",
            "Can I get a refund for last month?",
        ],
    },
    "technical": {
        "description": "Issues with product functionality, bugs, or access",
        "includes": ["login failures", "app crashes", "feature not working"],
        "excludes": ["billing questions", "general inquiries"],
        "examples": [
            "The app crashes when I try to upload a file",
            "I can't reset my password",
        ],
    },
}

# Validate taxonomy quality: check for overlapping keywords
def check_overlap(taxonomy):
    for l1, d1 in taxonomy.items():
        for l2, d2 in taxonomy.items():
            if l1 >= l2: continue
            overlap = set(d1["includes"]) & set(d2["includes"])
            if overlap:
                print(f"WARNING: {l1} and {l2} overlap on: {overlap}")

from sklearn.metrics import classification_report
import numpy as np

# Simulated imbalanced predictions
y_true = ["normal"]*980 + ["fraud"]*20
y_pred = ["normal"]*990 + ["fraud"]*10  # Misses 50% of fraud

# Overall accuracy looks great but is misleading
accuracy = sum(t == p for t, p in zip(y_true, y_pred)) / len(y_true)
print(f"Overall accuracy: {accuracy:.1%}")  # 98.0% - deceptive!

# Per-class metrics reveal the real problem
print(classification_report(y_true, y_pred))
# fraud: recall = 0.50 (missing half of all fraud cases!)

# === For prompted LLMs: emphasize minority class ===
imbalance_aware_prompt = """Classify transactions. Pay special attention to
fraud indicators - it is critical not to miss fraud cases.

Fraud signals: unusual amounts, new locations, rapid succession,
mismatched billing details.

IMPORTANT: When uncertain between fraud and normal, flag as fraud
for human review. False negatives are much costlier than false positives."""

import json

def multi_label_classify(text, labels):
    """Classify text with ALL applicable labels (multi-label)."""
    prompt = f"""Analyze the following text and select ALL labels that apply.
Only include a label if it is clearly relevant.

Available labels: {json.dumps(labels)}

Text: "{text}"

Respond with JSON: {{"labels": ["label1", "label2", ...], "reasons": {{"label1": "...", ...}}}}"""

    # response = llm.generate(prompt)
    # result = json.loads(response)
    # return result
    pass

# === Multi-label evaluation ===
from sklearn.metrics import hamming_loss, f1_score
import numpy as np

# Binary vectors: [billing, technical, account, general]
y_true = np.array([[1,1,0,0], [0,0,1,1], [1,0,0,0]])
y_pred = np.array([[1,1,0,0], [0,1,1,0], [1,0,0,0]])

print(f"Hamming Loss: {hamming_loss(y_true, y_pred):.3f}")
print(f"Micro F1:     {f1_score(y_true, y_pred, average='micro'):.3f}")
print(f"Macro F1:     {f1_score(y_true, y_pred, average='macro'):.3f}")

from sklearn.metrics import (
    classification_report, confusion_matrix, f1_score
)
import numpy as np

# Ground truth and model predictions
y_true = ["billing"]*30 + ["technical"]*50 + ["general"]*20
y_pred = ["billing"]*25 + ["technical"]*5 + \
         ["technical"]*45 + ["billing"]*5 + \
         ["general"]*18 + ["technical"]*2

# Full classification report with per-class metrics
print(classification_report(y_true, y_pred))

# Macro-F1 treats all classes equally (important for imbalanced data)
macro = f1_score(y_true, y_pred, average="macro")
print(f"Macro F1: {macro:.3f}")

# Confusion matrix reveals which classes are confused
cm = confusion_matrix(y_true, y_pred, labels=["billing","technical","general"])
print("Confusion Matrix:")
print(cm)
# Rows = true, Cols = predicted
# Off-diagonal values show misclassification patterns

from collections import Counter

def self_consistency_classify(text, labels, n_samples=5):
    """Run classification N times and use agreement as confidence."""
    predictions = []
    for _ in range(n_samples):
        # Each call uses temperature > 0 for variation
        # pred = llm.classify(text, labels, temperature=0.7)
        # predictions.append(pred)
        pass

    # Count label frequencies across samples
    counts = Counter(predictions)
    top_label, top_count = counts.most_common(1)[0]

    # Agreement rate = confidence estimate
    confidence = top_count / n_samples

    return {
        "label": top_label,
        "confidence": confidence,
        "distribution": dict(counts),
        "should_escalate": confidence < 0.6,  # Low agreement = uncertain
    }

# Example: 5 runs produce ["billing","billing","billing","technical","billing"]
# → label="billing", confidence=0.8, should_escalate=False

# Example: 5 runs produce ["billing","technical","billing","general","technical"]
# → label="billing", confidence=0.4, should_escalate=True

def classify_and_route(text, confidence_threshold=0.8):
    """Classify with automatic routing based on confidence."""
    # Step 1: Get classification with confidence
    result = classify_with_confidence(text)

    # Step 2: Route based on confidence
    if result["confidence"] >= confidence_threshold:
        # High confidence: auto-route
        return {
            "action": "auto_route",
            "label": result["label"],
            "confidence": result["confidence"],
        }
    else:
        # Low confidence: escalate to human reviewer
        return {
            "action": "human_review",
            "suggested_label": result["label"],
            "confidence": result["confidence"],
            "reason": result.get("reason", "Low confidence"),
            "priority": "high" if result["confidence"] < 0.5 else "normal",
        }

# Step 3: Log human decisions for model improvement
def log_review_decision(text, model_label, human_label):
    """Store human review outcomes for retraining."""
    record = {
        "text": text,
        "model_label": model_label,
        "human_label": human_label,
        "overturned": model_label != human_label,
    }
    # Append to training/evaluation dataset
    # These hard cases are the most valuable training data
    return record

from collections import Counter
from datetime import datetime

class ClassificationMonitor:
    """Monitor classification system health in production."""

    def __init__(self, expected_distribution, drift_threshold=0.1):
        self.expected = expected_distribution
        self.threshold = drift_threshold
        self.predictions = []

    def log_prediction(self, label, confidence, text_hash):
        """Log each prediction for monitoring."""
        self.predictions.append({
            "label": label,
            "confidence": confidence,
            "timestamp": datetime.utcnow(),
            "text_hash": text_hash,
        })

    def check_distribution_drift(self):
        """Alert if label distribution has shifted significantly."""
        counts = Counter(p["label"] for p in self.predictions)
        total = sum(counts.values())
        alerts = []
        for label, expected_frac in self.expected.items():
            actual_frac = counts.get(label, 0) / total
            drift = abs(actual_frac - expected_frac)
            if drift > self.threshold:
                alerts.append(f"DRIFT: {label} expected {expected_frac:.0%}, got {actual_frac:.0%}")
        return alerts

    def check_confidence_drop(self, min_avg=0.7):
        """Alert if average confidence drops below threshold."""
        if not self.predictions: return []
        avg_conf = sum(p["confidence"] for p in self.predictions) / len(self.predictions)
        if avg_conf < min_avg:
            return [f"LOW CONFIDENCE: avg={avg_conf:.2f} (threshold={min_avg})"]
        return []