💬 Natural Language Processing

from transformers import BertForSequenceClassification, Trainer, TrainingArguments from datasets import load_dataset import torch # Load dataset dataset = load_dataset("imdb") # Tokenization function def tokenize_function(examples): return tokenizer(examples["text"], padding="max_length", truncation=True) # Tokenize dataset tokenized_datasets = dataset.map(tokenize_function, batched=True) # Load model model = BertForSequenceClassification.from_pretrained( "bert-base-uncased", num_labels=2 ) # Training arguments training_args = TrainingArguments( output_dir="./results", learning_rate=2e-5, per_device_train_batch_size=16, per_device_eval_batch_size=16, num_train_epochs=3, weight_decay=0.01, ) # Create trainer trainer = Trainer( model=model, args=training_args, train_dataset=tokenized_datasets["train"], eval_dataset=tokenized_datasets["test"], ) # Fine-tune trainer.train()

import pandas as pd from sklearn.model_selection import train_test_split from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.naive_bayes import MultinomialNB from sklearn.metrics import accuracy_score, classification_report # Load and prepare data def load_spam_data(): # Example data structure data = { 'text': [ "Win a free iPhone now!", "Meeting scheduled for tomorrow", "Claim your prize money", "Project deadline reminder", ], 'label': ['spam', 'ham', 'spam', 'ham'] } return pd.DataFrame(data) # Preprocessing pipeline def preprocess_pipeline(df): # Clean text df['text'] = df['text'].str.lower() df['text'] = df['text'].str.replace(r'[^\w\s]', '', regex=True) # Convert labels to binary df['label'] = df['label'].map({'spam': 1, 'ham': 0}) return df # Build and train model df = load_spam_data() df = preprocess_pipeline(df) X_train, X_test, y_train, y_test = train_test_split( df['text'], df['label'], test_size=0.2, random_state=42 ) # Vectorize text vectorizer = TfidfVectorizer(max_features=1000) X_train_tfidf = vectorizer.fit_transform(X_train) X_test_tfidf = vectorizer.transform(X_test) # Train classifier clf = MultinomialNB() clf.fit(X_train_tfidf, y_train) # Evaluate predictions = clf.predict(X_test_tfidf) print(f"Accuracy: {accuracy_score(y_test, predictions):.2f}") print(classification_report(y_test, predictions))

from textblob import TextBlob import matplotlib.pyplot as plt from datetime import datetime import numpy as np class SentimentAnalyzer: def __init__(self): self.sentiments = [] self.timestamps = [] def analyze_text(self, text): # Perform sentiment analysis blob = TextBlob(text) polarity = blob.sentiment.polarity subjectivity = blob.sentiment.subjectivity # Classify sentiment if polarity > 0.1: sentiment = 'positive' elif polarity < -0.1: sentiment = 'negative' else: sentiment = 'neutral' # Store results self.sentiments.append({ 'text': text, 'polarity': polarity, 'subjectivity': subjectivity, 'sentiment': sentiment, 'timestamp': datetime.now() }) return sentiment, polarity, subjectivity def visualize_trends(self): if not self.sentiments: print("No data to visualize") return # Extract data polarities = [s['polarity'] for s in self.sentiments] timestamps = [s['timestamp'] for s in self.sentiments] # Create visualization plt.figure(figsize=(12, 6)) # Sentiment over time plt.subplot(1, 2, 1) plt.plot(timestamps, polarities, 'b-', alpha=0.6) plt.axhline(y=0, color='r', linestyle='--', alpha=0.3) plt.xlabel('Time') plt.ylabel('Sentiment Polarity') plt.title('Sentiment Trend Over Time') plt.xticks(rotation=45) # Sentiment distribution plt.subplot(1, 2, 2) sentiments = [s['sentiment'] for s in self.sentiments] unique, counts = np.unique(sentiments, return_counts=True) plt.bar(unique, counts, color=['red', 'gray', 'green']) plt.xlabel('Sentiment') plt.ylabel('Count') plt.title('Sentiment Distribution') plt.tight_layout() plt.show() # Usage example analyzer = SentimentAnalyzer() # Analyze sample texts texts = [ "I absolutely love this product! It's amazing!", "This is terrible, worst experience ever.", "It's okay, nothing special.", "Fantastic service, highly recommended!", "Not impressed with the quality." ] for text in texts: sentiment, polarity, subjectivity = analyzer.analyze_text(text) print(f"Text: {text[:50]}...") print(f"Sentiment: {sentiment}, Polarity: {polarity:.2f}") print("-" * 50) # Visualize results analyzer.visualize_trends()

import json import random from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity class IntentChatbot: def __init__(self): self.intents = { "greeting": { "patterns": ["hello", "hi", "hey", "good morning", "good evening"], "responses": ["Hello! How can I help you?", "Hi there!", "Hey! What can I do for you?"] }, "farewell": { "patterns": ["bye", "goodbye", "see you", "farewell"], "responses": ["Goodbye!", "See you later!", "Have a great day!"] }, "help": { "patterns": ["help", "assist", "support", "what can you do"], "responses": ["I can help you with various tasks. Just ask!", "I'm here to assist you!"] }, "weather": { "patterns": ["weather", "temperature", "forecast", "rain", "sunny"], "responses": ["I can't check live weather, but you can try weather.com!", "For weather updates, check your local weather service."] } } self.vectorizer = TfidfVectorizer() self.prepare_patterns() def prepare_patterns(self): # Flatten all patterns for vectorization all_patterns = [] self.pattern_to_intent = {} for intent, data in self.intents.items(): for pattern in data['patterns']: all_patterns.append(pattern) self.pattern_to_intent[pattern] = intent # Fit vectorizer self.pattern_vectors = self.vectorizer.fit_transform(all_patterns) def get_intent(self, user_input): # Vectorize user input user_vector = self.vectorizer.transform([user_input.lower()]) # Calculate similarities similarities = cosine_similarity(user_vector, self.pattern_vectors) # Get best match best_match_idx = similarities.argmax() best_match_score = similarities[0, best_match_idx] # Threshold for matching if best_match_score > 0.3: pattern = list(self.pattern_to_intent.keys())[best_match_idx] intent = self.pattern_to_intent[pattern] return intent, best_match_score return None, 0 def get_response(self, user_input): intent, confidence = self.get_intent(user_input) if intent: responses = self.intents[intent]['responses'] return random.choice(responses), intent, confidence else: return "I'm not sure I understand. Can you rephrase?", None, 0 def chat(self): print("Chatbot: Hello! I'm your assistant. Type 'quit' to exit.") while True: user_input = input("You: ") if user_input.lower() == 'quit': print("Chatbot: Goodbye!") break response, intent, confidence = self.get_response(user_input) print(f"Chatbot: {response}") print(f"[Debug - Intent: {intent}, Confidence: {confidence:.2f}]") # Initialize and run chatbot chatbot = IntentChatbot() # Uncomment to run interactive chat # chatbot.chat() # Test the chatbot test_inputs = [ "Hello there!", "What's the weather like?", "Can you help me?", "Goodbye!" ] for test in test_inputs: response, intent, confidence = chatbot.get_response(test) print(f"User: {test}") print(f"Bot: {response}") print(f"Intent: {intent} (Confidence: {confidence:.2f})") print("-" * 50)

# Production NLP Pipeline import joblib from typing import List, Dict, Any import logging from datetime import datetime class NLPPipeline: def __init__(self, model_path: str): self.logger = logging.getLogger(__name__) self.model = self.load_model(model_path) self.preprocessor = TextPreprocessor() self.postprocessor = ResultPostprocessor() def load_model(self, path: str): """Load pre-trained model""" try: model = joblib.load(path) self.logger.info(f"Model loaded from {path}") return model except Exception as e: self.logger.error(f"Failed to load model: {e}") raise def process_batch(self, texts: List[str]) -> List[Dict[str, Any]]: """Process batch of texts""" results = [] for text in texts: try: # Preprocess processed = self.preprocessor.process(text) # Predict prediction = self.model.predict([processed])[0] confidence = self.model.predict_proba([processed])[0].max() # Postprocess result = self.postprocessor.format_result( text=text, prediction=prediction, confidence=confidence, timestamp=datetime.now() ) results.append(result) except Exception as e: self.logger.error(f"Error processing text: {e}") results.append({ 'text': text, 'error': str(e), 'timestamp': datetime.now() }) return results def monitor_performance(self, predictions: List, actuals: List): """Monitor model performance in production""" from sklearn.metrics import accuracy_score, confusion_matrix accuracy = accuracy_score(actuals, predictions) cm = confusion_matrix(actuals, predictions) # Log metrics self.logger.info(f"Model Accuracy: {accuracy:.2f}") self.logger.info(f"Confusion Matrix:\n{cm}") # Alert if performance drops if accuracy < 0.8: self.logger.warning("Model performance below threshold!") self.send_alert("Model performance degradation detected") return { 'accuracy': accuracy, 'confusion_matrix': cm.tolist(), 'timestamp': datetime.now() } def send_alert(self, message: str): """Send alerts for critical issues""" # Implement your alerting mechanism here # (email, Slack, PagerDuty, etc.) pass class TextPreprocessor: def process(self, text: str) -> str: # Implement preprocessing logic text = text.lower() # Add more preprocessing steps return text class ResultPostprocessor: def format_result(self, **kwargs) -> Dict[str, Any]: return { 'text': kwargs.get('text'), 'prediction': kwargs.get('prediction'), 'confidence': float(kwargs.get('confidence', 0)), 'timestamp': kwargs.get('timestamp').isoformat() } # Usage example if __name__ == "__main__": # Initialize pipeline pipeline = NLPPipeline("model.pkl") # Process texts texts = ["Sample text for processing", "Another example"] results = pipeline.process_batch(texts) for result in results: print(result)