Pipelines de NLP para aplicaciones de producción

function preprocessText(raw: string): string {
  let text = raw;
 
  // Remove HTML tags
  text = text.replace(/<[^>]*>/g, ' ');
 
  // Decode HTML entities
  text = text
    .replace(/&amp;/g, '&')
    .replace(/&lt;/g, '<')
    .replace(/&gt;/g, '>')
    .replace(/&quot;/g, '"')
    .replace(/&#039;/g, "'");
 
  // Remove URLs
  text = text.replace(/https?:\/\/\S+/g, ' ');
 
  // Remove zero-width characters and other invisible Unicode
  text = text.replace(/[\u200B-\u200D\uFEFF\u00AD]/g, '');
 
  // Normalize Unicode (NFC form — composed characters)
  text = text.normalize('NFC');
 
  // Collapse multiple whitespace into single space
  text = text.replace(/\s+/g, ' ').trim();
 
  return text;
}

// ❌ Skipping preprocessing — model sees garbage
const input = "Check out   <b>this deal</b>!!!\u200B\n\n   http://spam.com";
// Model processes HTML tags, invisible chars, URLs as features
// Classification accuracy drops, embeddings are noisy
 
// ✅ Preprocessing first — model sees clean text
const cleaned = preprocessText(input);
// "Check out this deal!!!"
// Clean, normalized, ready for tokenization

# Word-level tokenization — simple but brittle
def word_tokenize(text: str) -> list[str]:
    return text.lower().split()
 
# Problem: "running", "runs", "ran" are three separate tokens
# Problem: "ChatGPT" is unknown if not in vocabulary
 
 
# Subword tokenization (BPE) — handles unknown words
from tokenizers import Tokenizer, models, trainers
 
def train_bpe_tokenizer(corpus: list[str], vocab_size: int = 30000):
    tokenizer = Tokenizer(models.BPE())
    trainer = trainers.BpeTrainer(
        vocab_size=vocab_size,
        special_tokens=["[PAD]", "[UNK]", "[CLS]", "[SEP]"]
    )
    tokenizer.train_from_iterator(corpus, trainer)
    return tokenizer
 
# "ChatGPT" → ["Chat", "G", "PT"] — handled as subwords
# "unfriendliness" → ["un", "friend", "li", "ness"]

# ❌ Using word tokenization for production — too many unknown words
tokens = text.split()
# "microservices" → ["microservices"]  (unknown word = [UNK])
# Loses all semantic meaning for rare words
 
# ✅ Using a pretrained tokenizer — robust subword handling
from transformers import AutoTokenizer
 
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
tokens = tokenizer.encode("microservices architecture patterns")
# Handles any word by breaking into known subwords
# Works across domains without retraining

from sentence_transformers import SentenceTransformer
import numpy as np
 
# Load a pretrained embedding model
model = SentenceTransformer("all-MiniLM-L6-v2")
 
# Generate embeddings
texts = [
    "How to reset my password",
    "I forgot my login credentials",
    "What are your business hours",
    "When is the store open",
]
 
embeddings = model.encode(texts)
# Shape: (4, 384) — four 384-dimensional vectors
 
# Compute similarity
from numpy.linalg import norm
 
def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    return float(np.dot(a, b) / (norm(a) * norm(b)))
 
# "reset password" vs "forgot credentials" → ~0.85 (very similar)
sim_1 = cosine_similarity(embeddings[0], embeddings[1])
 
# "reset password" vs "business hours" → ~0.15 (different topics)
sim_2 = cosine_similarity(embeddings[0], embeddings[2])

// TypeScript: Using embeddings for intent classification
 
interface Intent {
  name: string;
  examples: string[];
  embedding?: number[];  // Average embedding of examples
}
 
async function classifyIntent(
  text: string,
  intents: Intent[],
  embedder: EmbeddingModel
): Promise<{ intent: string; confidence: number }> {
  const queryEmbedding = await embedder.encode(text);
 
  let bestMatch = { intent: 'unknown', confidence: 0 };
 
  for (const intent of intents) {
    if (!intent.embedding) continue;
 
    const similarity = cosineSimilarity(queryEmbedding, intent.embedding);
 
    if (similarity > bestMatch.confidence) {
      bestMatch = { intent: intent.name, confidence: similarity };
    }
  }
 
  // Threshold: below 0.5 means no confident match
  if (bestMatch.confidence < 0.5) {
    return { intent: 'unknown', confidence: bestMatch.confidence };
  }
 
  return bestMatch;
}

import spacy
 
nlp = spacy.load("en_core_web_sm")
 
def extract_entities(text: str) -> list[dict]:
    doc = nlp(text)
    entities = []
 
    for ent in doc.ents:
        entities.append({
            "text": ent.text,
            "label": ent.label_,
            "start": ent.start_char,
            "end": ent.end_char,
        })
 
    return entities
 
# Input: "Apple announced a $3 billion deal with Samsung on March 15th"
# Output:
# [
#   {"text": "Apple", "label": "ORG", "start": 0, "end": 5},
#   {"text": "$3 billion", "label": "MONEY", "start": 18, "end": 28},
#   {"text": "Samsung", "label": "ORG", "start": 39, "end": 46},
#   {"text": "March 15th", "label": "DATE", "start": 50, "end": 60},
# ]

# Custom entity extraction for domain-specific data
# Example: extracting ticket IDs and severity levels from support messages
 
import re
from dataclasses import dataclass
 
@dataclass
class SupportEntities:
    ticket_ids: list[str]
    severity: str | None
    product: str | None
 
def extract_support_entities(text: str) -> SupportEntities:
    # Custom patterns for domain entities
    ticket_ids = re.findall(r'TICK-\d{4,8}', text)
 
    severity = None
    severity_patterns = {
        'critical': r'\b(critical|p0|sev-?0|outage)\b',
        'high': r'\b(high|p1|sev-?1|urgent)\b',
        'medium': r'\b(medium|p2|sev-?2|normal)\b',
        'low': r'\b(low|p3|sev-?3|minor)\b',
    }
    for level, pattern in severity_patterns.items():
        if re.search(pattern, text, re.IGNORECASE):
            severity = level
            break
 
    return SupportEntities(
        ticket_ids=ticket_ids,
        severity=severity,
        product=None,  # Would use NER model for product extraction
    )

interface PipelineResult {
  preprocessed: string;
  tokens: string[];
  embedding: number[];
  intent: { name: string; confidence: number };
  entities: Entity[];
  processingTimeMs: number;
}
 
async function processText(
  raw: string,
  config: PipelineConfig
): Promise<PipelineResult> {
  const start = performance.now();
 
  // Step 1: Preprocess
  const preprocessed = preprocessText(raw);
  if (preprocessed.length === 0) {
    throw new Error('Empty text after preprocessing');
  }
 
  // Step 2: Tokenize (for logging/debugging)
  const tokens = config.tokenizer.tokenize(preprocessed);
  if (tokens.length > config.maxTokens) {
    // Truncate gracefully — don't just slice mid-word
    tokens.length = config.maxTokens;
  }
 
  // Step 3: Generate embedding
  const embedding = await config.embedder.encode(preprocessed);
 
  // Step 4: Classify intent and extract entities in parallel
  const [intent, entities] = await Promise.all([
    classifyIntent(preprocessed, config.intents, config.embedder),
    extractEntities(preprocessed, config.nerModel),
  ]);
 
  return {
    preprocessed,
    tokens,
    embedding,
    intent,
    entities,
    processingTimeMs: performance.now() - start,
  };
}

// ❌ No pipeline structure — ad hoc processing
async function handleMessage(text: string) {
  const cleaned = text.replace(/<[^>]*>/g, ''); // Minimal cleaning
  const result = await model.classify(cleaned);   // Hope for the best
  return result;
}
 
// ✅ Structured pipeline with monitoring
async function handleMessage(text: string) {
  const result = await processText(text, pipelineConfig);
 
  // Log pipeline metrics
  metrics.histogram('nlp.pipeline.duration_ms', result.processingTimeMs);
  metrics.increment(`nlp.intent.${result.intent.name}`);
 
  if (result.intent.confidence < 0.5) {
    metrics.increment('nlp.intent.low_confidence');
  }
 
  return result;
}