My Practical Insights of Using spaCy Library

import spacy from spacy import displacy import pandas as pd # Load the English model (install with: python -m spacy download en_core_web_sm) nlp = spacy.load("en_core_web_sm") # Examine the pipeline components print("Pipeline components:", nlp.pipe_names) print("Pipeline:", [name for name, component in nlp.pipeline]) # Process a sample text text = "Apple Inc. is planning to open a new store in New York City next month." doc = nlp(text) # Explore different linguistic features print(f"\nOriginal text: {text}") print(f"Number of tokens: {len(doc)}") print(f"Number of sentences: {len(list(doc.sents))}") # Token-level analysis for token in doc: print(f"{token.text:12} | {token.pos_:8} | {token.lemma_:12} | {token.is_stop}")

# Advanced linguistic feature extraction text = """ The researchers at Stanford University published groundbreaking findings about machine learning algorithms. Dr. Sarah Johnson, who led the study, explained that the new approach could revolutionize natural language processing. """ doc = nlp(text) print("=== NAMED ENTITY RECOGNITION ===") for ent in doc.ents: print(f"{ent.text:20} | {ent.label_:12} | {spacy.explain(ent.label_)}") print("\n=== DEPENDENCY PARSING ===") for token in doc: if token.dep_ != "punct": # Skip punctuation print(f"{token.text:15} | {token.dep_:10} | {token.head.text}") print("\n=== SENTENCE SEGMENTATION ===") for i, sent in enumerate(doc.sents, 1): print(f"Sentence {i}: {sent.text.strip()}") print("\n=== NOUN CHUNKS ===") for chunk in doc.noun_chunks: print(f"{chunk.text:25} | Root: {chunk.root.text} | Dep: {chunk.root.dep_}")

from spacy.matcher import Matcher from spacy.util import filter_spans # Initialize matcher for pattern-based entity recognition matcher = Matcher(nlp.vocab) # Define patterns for custom entities email_pattern = [{"TEXT": {"REGEX": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}"}}] phone_pattern = [ {"TEXT": {"REGEX": r"\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}"}}, {"TEXT": {"REGEX": r"\+\d{1,3}[-.\s]?\d{3}[-.\s]?\d{3}[-.\s]?\d{4}"}} ] product_code_pattern = [{"TEXT": {"REGEX": r"[A-Z]{2,3}-\d{3,4}"}}] # Add patterns to matcher matcher.add("EMAIL", [email_pattern]) matcher.add("PHONE", [phone_pattern]) matcher.add("PRODUCT_CODE", [product_code_pattern]) text = """ Contact Sarah at sarah.johnson@company.com or call (555) 123-4567. Product codes: ABC-1234, XYZ-5678. International number: +1-555-987-6543. """ doc = nlp(text) # Find pattern matches matches = matcher(doc) custom_entities = [] for match_id, start, end in matches: label = nlp.vocab.strings[match_id] span = doc[start:end] custom_entities.append((span.start_char, span.end_char, label)) print(f"Custom Entity: {span.text:25} | Label: {label}") # Combine with existing entities all_entities = [(ent.start_char, ent.end_char, ent.label_) for ent in doc.ents] all_entities.extend(custom_entities) print(f"\nFound {len(doc.ents)} standard entities and {len(custom_entities)} custom entities")

# Advanced text preprocessing utilities import re import unicodedata def advanced_text_cleaner(text, remove_entities=None, min_token_length=2): """ Advanced text cleaning and normalization pipeline """ # Process with spaCy doc = nlp(text) cleaned_tokens = [] for token in doc: # Skip unwanted tokens if (token.is_stop or token.is_punct or token.is_space or token.like_num or len(token.text) < min_token_length): continue # Skip specific entity types if requested if remove_entities and token.ent_type_ in remove_entities: continue # Use lemmatized form cleaned_token = token.lemma_.lower().strip() # Additional cleaning if cleaned_token and cleaned_token.isalpha(): cleaned_tokens.append(cleaned_token) return cleaned_tokens # Test the preprocessing pipeline sample_texts = [ "The CEO of Apple Inc., Tim Cook, announced $50 billion in revenue for Q3 2023!", "Dr. Sarah Johnson (PhD) published 15 papers on NLP algorithms @ Stanford University.", "Visit https://example.com or email info@company.com for more details." ] print("=== TEXT PREPROCESSING RESULTS ===") for i, text in enumerate(sample_texts, 1): print(f"\nOriginal {i}: {text}") # Basic preprocessing basic_tokens = advanced_text_cleaner(text) print(f"Basic: {' '.join(basic_tokens)}") # Remove person and organization entities no_entities = advanced_text_cleaner(text, remove_entities=['PERSON', 'ORG']) print(f"No Entities: {' '.join(no_entities)}") # Advanced preprocessing utilities def extract_text_statistics(text): """Extract comprehensive text statistics""" doc = nlp(text) stats = { 'total_tokens': len(doc), 'unique_tokens': len(set([token.text.lower() for token in doc])), 'sentences': len(list(doc.sents)), 'entities': len(doc.ents), 'noun_chunks': len(list(doc.noun_chunks)), 'stop_words': sum(1 for token in doc if token.is_stop), 'pos_distribution': {} } # POS distribution for token in doc: pos = token.pos_ stats['pos_distribution'][pos] = stats['pos_distribution'].get(pos, 0) + 1 return stats # Demonstrate text statistics long_text = """ Natural language processing (NLP) is a subfield of linguistics, computer science, and artificial intelligence concerned with the interactions between computers and human language. It involves developing algorithms and models that can understand, interpret, and generate human language in a valuable way. """ stats = extract_text_statistics(long_text) print(f"\n=== TEXT STATISTICS ===") for key, value in stats.items(): if key != 'pos_distribution': print(f"{key}: {value}") print(f"\nTop POS tags:") sorted_pos = sorted(stats['pos_distribution'].items(), key=lambda x: x[1], reverse=True) for pos, count in sorted_pos[:5]: print(f" {pos}: {count}")

from spacy.language import Language from spacy.tokens import Doc, Span # Custom pipeline component for sentiment analysis @Language.component("sentiment_analyzer") def sentiment_component(doc): """Simple rule-based sentiment analysis component""" positive_words = {"good", "great", "excellent", "amazing", "wonderful", "fantastic"} negative_words = {"bad", "terrible", "awful", "horrible", "disappointing", "poor"} positive_count = sum(1 for token in doc if token.lemma_.lower() in positive_words) negative_count = sum(1 for token in doc if token.lemma_.lower() in negative_words) # Calculate sentiment score total_sentiment_words = positive_count + negative_count if total_sentiment_words > 0: sentiment_score = (positive_count - negative_count) / total_sentiment_words else: sentiment_score = 0.0 # Add sentiment to doc extensions doc._.sentiment_score = sentiment_score doc._.sentiment_label = "positive" if sentiment_score > 0.1 else "negative" if sentiment_score < -0.1 else "neutral" return doc # Custom component for text complexity analysis @Language.component("complexity_analyzer") def complexity_component(doc): """Analyze text complexity metrics""" total_tokens = len(doc) total_sentences = len(list(doc.sents)) # Calculate average sentence length avg_sentence_length = total_tokens / total_sentences if total_sentences > 0 else 0 # Count complex words (words with more than 2 syllables - simplified) complex_words = sum(1 for token in doc if len(token.text) > 7 and token.is_alpha) # Calculate readability scores doc._.avg_sentence_length = avg_sentence_length doc._.complex_word_ratio = complex_words / total_tokens if total_tokens > 0 else 0 doc._.readability_score = max(0, 100 - (avg_sentence_length * 1.5) - (doc._.complex_word_ratio * 100)) return doc # Register document extensions if not Doc.has_extension("sentiment_score"): Doc.set_extension("sentiment_score", default=0.0) if not Doc.has_extension("sentiment_label"): Doc.set_extension("sentiment_label", default="neutral") if not Doc.has_extension("avg_sentence_length"): Doc.set_extension("avg_sentence_length", default=0.0) if not Doc.has_extension("complex_word_ratio"): Doc.set_extension("complex_word_ratio", default=0.0) if not Doc.has_extension("readability_score"): Doc.set_extension("readability_score", default=0.0) # Create a new pipeline with custom components nlp_custom = spacy.load("en_core_web_sm") nlp_custom.add_pipe("sentiment_analyzer", last=True) nlp_custom.add_pipe("complexity_analyzer", last=True) # Test the custom pipeline test_texts = [ "This is an amazing product! The quality is excellent and the design is wonderful.", "The service was terrible. I had a horrible experience and would not recommend it.", "The comprehensive analysis demonstrates significant improvements in computational efficiency through advanced algorithmic optimizations.", "I like cats." ] print("=== CUSTOM PIPELINE ANALYSIS ===") for i, text in enumerate(test_texts, 1): doc = nlp_custom(text) print(f"\nText {i}: {text}") print(f"Sentiment: {doc._.sentiment_label} (score: {doc._.sentiment_score:.2f})") print(f"Avg sentence length: {doc._.avg_sentence_length:.1f}") print(f"Complex word ratio: {doc._.complex_word_ratio:.2f}") print(f"Readability score: {doc._.readability_score:.1f}")

import logging from typing import List, Dict, Optional from dataclasses import dataclass import json # Configure logging for production logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class ProcessingResult: """Structured result for text processing""" text: str tokens: List[str] entities: List[Dict] sentiment: Optional[str] = None language: Optional[str] = None processing_time: Optional[float] = None class ProductionNLPProcessor: """Production-ready NLP processor with error handling and monitoring""" def __init__(self, model_name: str = "en_core_web_sm", enable_custom_components: bool = True): try: self.nlp = spacy.load(model_name) if enable_custom_components: # Add custom components if needed if "sentiment_analyzer" not in self.nlp.pipe_names: self.nlp.add_pipe("sentiment_analyzer", last=True) logger.info(f"NLP processor initialized with model: {model_name}") logger.info(f"Pipeline components: {self.nlp.pipe_names}") except OSError as e: logger.error(f"Failed to load model {model_name}: {e}") raise def process_text(self, text: str) -> ProcessingResult: """Process single text with error handling""" if not text or not text.strip(): return ProcessingResult(text="", tokens=[], entities=[], sentiment="neutral") try: import time start_time = time.time() # Process with spaCy doc = self.nlp(text.strip()) # Extract information tokens = [token.lemma_.lower() for token in doc if not token.is_stop and not token.is_punct] entities = [ { "text": ent.text, "label": ent.label_, "start": ent.start_char, "end": ent.end_char, "description": spacy.explain(ent.label_) } for ent in doc.ents ] # Get sentiment if available sentiment = getattr(doc._, "sentiment_label", "neutral") processing_time = time.time() - start_time return ProcessingResult( text=text, tokens=tokens, entities=entities, sentiment=sentiment, language=doc.lang_, processing_time=processing_time ) except Exception as e: logger.error(f"Error processing text: {e}") return ProcessingResult(text=text, tokens=[], entities=[], sentiment="error") def process_batch(self, texts: List[str], batch_size: int = 50) -> List[ProcessingResult]: """Process multiple texts efficiently""" results = [] try: # Use spaCy's pipe for efficient batch processing docs = list(self.nlp.pipe(texts, batch_size=batch_size)) for text, doc in zip(texts, docs): tokens = [token.lemma_.lower() for token in doc if not token.is_stop and not token.is_punct] entities = [ { "text": ent.text, "label": ent.label_, "start": ent.start_char, "end": ent.end_char } for ent in doc.ents ] sentiment = getattr(doc._, "sentiment_label", "neutral") results.append(ProcessingResult( text=text, tokens=tokens, entities=entities, sentiment=sentiment, language=doc.lang_ )) except Exception as e: logger.error(f"Error in batch processing: {e}") # Return error results for all texts results = [ProcessingResult(text=text, tokens=[], entities=[], sentiment="error") for text in texts] return results # Demonstrate production processor processor = ProductionNLPProcessor() # Single text processing sample_text = "Apple Inc. is planning to release an amazing new iPhone model next year." result = processor.process_text(sample_text) print("=== SINGLE TEXT PROCESSING ===") print(f"Original: {result.text}") print(f"Key tokens: {result.tokens[:10]}") # First 10 tokens print(f"Entities found: {len(result.entities)}") for ent in result.entities: print(f" - {ent['text']}: {ent['label']}") print(f"Sentiment: {result.sentiment}") print(f"Processing time: {result.processing_time:.3f}s") # Batch processing demonstration batch_texts = [ "Google announced new AI capabilities.", "Microsoft released updates to their cloud platform.", "Amazon's stock price increased significantly.", "Tesla revealed their latest electric vehicle innovations." ] batch_results = processor.process_batch(batch_texts) print(f"\n=== BATCH PROCESSING ===") print(f"Processed {len(batch_results)} texts") for i, result in enumerate(batch_results, 1): print(f"{i}. Entities: {len(result.entities)}, Sentiment: {result.sentiment}")