NLP Engineering

import re
import unicodedata
from bs4 import BeautifulSoup

def preprocess(text: str, lowercase: bool = True) -> str:
    # 1. Decode HTML entities and strip tags
    text = BeautifulSoup(text, "html.parser").get_text(separator=" ")

    # 2. Normalize unicode (NFD -> NFC, remove combining chars if needed)
    text = unicodedata.normalize("NFC", text)

    # 3. Lowercase
    if lowercase:
        text = text.lower()

    # 4. Remove URLs, emails, special tokens
    text = re.sub(r"https?://\S+|www\.\S+", " ", text)
    text = re.sub(r"\S+@\S+\.\S+", " ", text)

    # 5. Collapse whitespace
    text = re.sub(r"\s+", " ", text).strip()

    return text

# Usage
clean = preprocess("<p>Visit https://example.com for more info.</p>")
# -> "visit for more info."

# Option A: sentence-transformers (local, free, fast on GPU)
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-small-en-v1.5")

documents = ["The quick brown fox", "Machine learning is fun", "NLP rocks"]

# encode() handles batching internally; show_progress_bar for large corpora
embeddings = model.encode(documents, normalize_embeddings=True, show_progress_bar=True)
# -> numpy array, shape (3, 384)

# Option B: OpenAI embeddings API
from openai import OpenAI

client = OpenAI()

def embed_batch(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    # Strip newlines - they degrade embedding quality per OpenAI docs
    texts = [t.replace("\n", " ") for t in texts]
    response = client.embeddings.create(input=texts, model=model)
    return [item.embedding for item in response.data]

import numpy as np
import faiss
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-small-en-v1.5")

# --- Indexing ---
docs = ["Python is a programming language.", "The Eiffel Tower is in Paris.", ...]
doc_embeddings = model.encode(docs, normalize_embeddings=True).astype("float32")

# Inner product on normalized vectors = cosine similarity
index = faiss.IndexFlatIP(doc_embeddings.shape[1])
index.add(doc_embeddings)

# --- Retrieval ---
def search(query: str, top_k: int = 5) -> list[tuple[str, float]]:
    q_emb = model.encode([query], normalize_embeddings=True).astype("float32")
    scores, indices = index.search(q_emb, top_k)
    return [(docs[i], float(scores[0][j])) for j, i in enumerate(indices[0])]

results = search("programming languages for data science")
# -> [("Python is a programming language.", 0.87), ...]

from datasets import Dataset
from transformers import (
    AutoTokenizer,
    AutoModelForSequenceClassification,
    TrainingArguments,
    Trainer,
)
import torch

MODEL_ID = "distilbert-base-uncased"
LABELS = ["negative", "neutral", "positive"]
id2label = {i: l for i, l in enumerate(LABELS)}
label2id = {l: i for i, l in enumerate(LABELS)}

tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
model = AutoModelForSequenceClassification.from_pretrained(
    MODEL_ID, num_labels=len(LABELS), id2label=id2label, label2id=label2id
)

def tokenize(batch):
    return tokenizer(batch["text"], truncation=True, padding="max_length", max_length=128)

# train_data: list of {"text": str, "label": int}
train_ds = Dataset.from_list(train_data).map(tokenize, batched=True)

args = TrainingArguments(
    output_dir="./sentiment-model",
    num_train_epochs=3,
    per_device_train_batch_size=32,
    evaluation_strategy="epoch",
    save_strategy="best",
    load_best_model_at_end=True,
)

trainer = Trainer(model=model, args=args, train_dataset=train_ds, eval_dataset=eval_ds)
trainer.train()

import spacy
from transformers import pipeline

# Option A: spaCy (fast, battle-tested for standard entities)
nlp = spacy.load("en_core_web_sm")

def extract_entities(text: str) -> list[dict]:
    doc = nlp(text)
    return [
        {"text": ent.text, "label": ent.label_, "start": ent.start_char, "end": ent.end_char}
        for ent in doc.ents
    ]

entities = extract_entities("Apple Inc. was founded by Steve Jobs in Cupertino.")
# -> [{"text": "Apple Inc.", "label": "ORG", ...}, {"text": "Steve Jobs", "label": "PERSON", ...}]

# Option B: HuggingFace token classification (custom entities, higher accuracy)
ner = pipeline(
    "token-classification",
    model="dslim/bert-base-NER",
    aggregation_strategy="simple",  # merges B-/I- tokens into spans
)
results = ner("OpenAI released GPT-4 in San Francisco.")

# --- Extractive: rank sentences by TF-IDF centrality ---
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def extractive_summary(text: str, n_sentences: int = 3) -> str:
    sentences = [s.strip() for s in text.split(".") if s.strip()]
    tfidf = TfidfVectorizer().fit_transform(sentences)
    sim_matrix = cosine_similarity(tfidf)
    scores = sim_matrix.sum(axis=1)
    top_indices = np.argsort(scores)[-n_sentences:][::-1]
    return ". ".join(sentences[i] for i in sorted(top_indices)) + "."

# --- Abstractive: seq2seq model ---
from transformers import pipeline

summarizer = pipeline("summarization", model="facebook/bart-large-cnn")

def abstractive_summary(text: str, max_length: int = 130) -> str:
    # BART has a 1024-token context window - chunk long documents first
    result = summarizer(text, max_length=max_length, min_length=30, do_sample=False)
    return result[0]["summary_text"]

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_document(
    text: str,
    chunk_size: int = 512,
    chunk_overlap: int = 64,
) -> list[dict]:
    """
    Recursive splitter tries paragraph -> sentence -> word boundaries in order.
    chunk_overlap ensures context continuity across chunk boundaries.
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    chunks = splitter.split_text(text)
    return [{"text": chunk, "chunk_index": i, "total_chunks": len(chunks)} for i, chunk in enumerate(chunks)]

# Semantic chunking (group sentences by embedding similarity instead of length)
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings

semantic_splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",  # split where similarity drops sharply
    breakpoint_threshold_amount=95,
)
semantic_chunks = semantic_splitter.create_documents([text])