Python Slot Filling and Extraction — Deep Dive

The Slot Filling Pipeline

Production slot filling involves five stages: tokenization, entity extraction, entity-to-slot mapping, validation, and state update. Each stage is a separate module with a defined interface, making components swappable and testable.

Entity Extraction Techniques

Rule-Based Extractors

For entities with predictable formats, rules outperform ML in both speed and reliability:

import re
from dateutil import parser as dateparser
from dataclasses import dataclass

@dataclass
class Entity:
    value: str
    entity_type: str
    start: int
    end: int
    confidence: float = 1.0

class RegexExtractor:
    PATTERNS = {
        "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
        "phone": r"\b\+?1?[-.\s]?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b",
        "currency": r"\$\d+(?:,\d{3})*(?:\.\d{2})?",
    }

    def extract(self, text: str) -> list[Entity]:
        entities = []
        for entity_type, pattern in self.PATTERNS.items():
            for match in re.finditer(pattern, text):
                entities.append(Entity(
                    value=match.group(),
                    entity_type=entity_type,
                    start=match.start(),
                    end=match.end(),
                ))
        return entities

class DateExtractor:
    def extract(self, text: str) -> list[Entity]:
        # Use dateutil for flexible date parsing
        try:
            dt = dateparser.parse(text, fuzzy=True, fuzzy_with_tokens=True)
            if dt and dt[0]:
                return [Entity(
                    value=dt[0].isoformat(),
                    entity_type="date",
                    start=0, end=len(text),
                    confidence=0.9,
                )]
        except (ValueError, OverflowError):
            pass
        return []

CRF-Based Extraction with sklearn-crfsuite

Conditional Random Fields model entity extraction as sequence labeling. Features typically include the token itself, its prefix/suffix, part-of-speech tag, and surrounding tokens:

import sklearn_crfsuite
from sklearn_crfsuite import metrics

def word_features(sentence: list[str], i: int) -> dict:
    word = sentence[i]
    features = {
        "word.lower()": word.lower(),
        "word[-3:]": word[-3:],
        "word[-2:]": word[-2:],
        "word.isupper()": word.isupper(),
        "word.istitle()": word.istitle(),
        "word.isdigit()": word.isdigit(),
    }
    if i > 0:
        prev = sentence[i - 1]
        features["prev.lower()"] = prev.lower()
        features["prev.istitle()"] = prev.istitle()
    else:
        features["BOS"] = True  # beginning of sentence
    if i < len(sentence) - 1:
        nxt = sentence[i + 1]
        features["next.lower()"] = nxt.lower()
        features["next.istitle()"] = nxt.istitle()
    else:
        features["EOS"] = True  # end of sentence
    return features

crf = sklearn_crfsuite.CRF(
    algorithm="lbfgs",
    c1=0.1,
    c2=0.1,
    max_iterations=100,
)
# crf.fit(X_train_features, y_train_labels)

CRFs are lightweight and fast at inference. They work well when you have 500+ labeled examples and entities follow contextual patterns.

Transformer-Based NER

For complex entities in varied contexts, fine-tune a Transformer on BIO-tagged data:

from transformers import AutoTokenizer, AutoModelForTokenClassification, pipeline

model_name = "dslim/bert-base-NER"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForTokenClassification.from_pretrained(model_name)

ner_pipeline = pipeline("ner", model=model, tokenizer=tokenizer, aggregation_strategy="simple")

results = ner_pipeline("Book a flight to Berlin on March 20")
# [{'entity_group': 'LOC', 'word': 'Berlin', 'score': 0.99, ...},
#  {'entity_group': 'DATE', 'word': 'March 20', 'score': 0.95, ...}]

For domain-specific entities (product names, internal codes), fine-tune on your own labeled data using the Hugging Face Trainer.

Entity-to-Slot Mapping

Direct Mapping

The simplest approach maps entity types directly to slot names:

ENTITY_SLOT_MAP = {
    "city": "destination",
    "date": "travel_date",
    "number": "passenger_count",
}

def map_entities_to_slots(entities: list[Entity]) -> dict:
    slots = {}
    for entity in entities:
        slot_name = ENTITY_SLOT_MAP.get(entity.entity_type)
        if slot_name:
            slots[slot_name] = entity.value
    return slots

Role-Based Mapping

When the same entity type fills different slots (origin city vs. destination city), role detection is needed. Approaches include:

• Positional heuristics: The first city after “from” is origin; the first city after “to” is destination.
• Learned roles: Train a classifier that predicts the role of each entity given its surrounding context.

def map_cities_by_context(text: str, city_entities: list[Entity]) -> dict:
    slots = {}
    for entity in city_entities:
        prefix = text[:entity.start].lower().strip()
        if prefix.endswith("from"):
            slots["origin"] = entity.value
        elif prefix.endswith("to"):
            slots["destination"] = entity.value
        elif "origin" not in slots:
            slots["destination"] = entity.value  # default
    return slots

Validation Layer

Type-Specific Validators

from datetime import datetime, date
from typing import Any

class SlotValidator:
    @staticmethod
    def validate_date(value: str) -> tuple[bool, str | None]:
        try:
            dt = dateparser.parse(value)
            if dt and dt.date() >= date.today():
                return True, dt.date().isoformat()
            return False, "Date must be in the future"
        except Exception:
            return False, "Could not parse date"

    @staticmethod
    def validate_passenger_count(value: Any) -> tuple[bool, int | None]:
        try:
            n = int(value)
            if 1 <= n <= 9:
                return True, n
            return False, "Passenger count must be between 1 and 9"
        except (ValueError, TypeError):
            return False, "Not a valid number"

    @staticmethod
    def validate_destination(value: str, known_cities: set[str]) -> tuple[bool, str | None]:
        normalized = value.strip().title()
        if normalized in known_cities:
            return True, normalized
        # Fuzzy match
        from difflib import get_close_matches
        matches = get_close_matches(normalized, known_cities, n=1, cutoff=0.8)
        if matches:
            return True, matches[0]
        return False, f"Unknown destination: {value}"

Validation Chain

Chain validators per slot and collect errors:

from dataclasses import dataclass, field

@dataclass
class ValidationResult:
    valid: bool
    cleaned_value: Any = None
    error: str | None = None

@dataclass
class Frame:
    slots: dict[str, Any] = field(default_factory=dict)
    errors: dict[str, str] = field(default_factory=dict)

    VALIDATORS = {
        "travel_date": SlotValidator.validate_date,
        "passenger_count": SlotValidator.validate_passenger_count,
    }

    def fill_and_validate(self, raw_slots: dict[str, Any]) -> None:
        for slot_name, raw_value in raw_slots.items():
            validator = self.VALIDATORS.get(slot_name)
            if validator:
                valid, result = validator(raw_value)
                if valid:
                    self.slots[slot_name] = result
                    self.errors.pop(slot_name, None)
                else:
                    self.errors[slot_name] = result
            else:
                self.slots[slot_name] = raw_value

Multi-Turn State Machine

Frame Lifecycle

A frame progresses through states:

1. INIT — Frame created, all slots empty.
2. FILLING — Bot prompts for missing slots, user provides values.
3. VALIDATING — All slots filled, running final cross-field validation.
4. CONFIRMED — User confirmed the summary.
5. EXECUTED — Backend action completed.

from enum import Enum, auto

class FrameState(Enum):
    INIT = auto()
    FILLING = auto()
    VALIDATING = auto()
    CONFIRMED = auto()
    EXECUTED = auto()

class FrameManager:
    def __init__(self, required_slots: list[str]):
        self.required = required_slots
        self.frame = Frame()
        self.state = FrameState.INIT

    def process_turn(self, entities: list[Entity]) -> str:
        raw_slots = map_entities_to_slots(entities)
        self.frame.fill_and_validate(raw_slots)

        if self.frame.errors:
            slot, error = next(iter(self.frame.errors.items()))
            return f"There's an issue with {slot}: {error}. Could you try again?"

        missing = [s for s in self.required if s not in self.frame.slots]
        if missing:
            self.state = FrameState.FILLING
            return self._prompt_for(missing[0])

        self.state = FrameState.VALIDATING
        return self._generate_summary()

    def _prompt_for(self, slot_name: str) -> str:
        prompts = {
            "destination": "Where would you like to travel?",
            "travel_date": "What date works for you?",
            "passenger_count": "How many passengers?",
        }
        return prompts.get(slot_name, f"Please provide {slot_name}")

    def _generate_summary(self) -> str:
        details = ", ".join(f"{k}: {v}" for k, v in self.frame.slots.items())
        return f"Here's what I have: {details}. Shall I proceed?"

Handling Corrections

Detect correction patterns and update the appropriate slot:

CORRECTION_PATTERNS = [
    r"(?:actually|no|wait|change)\s+(?:the\s+)?(\w+)\s+to\s+(.+)",
    r"(?:make it|switch to|change to)\s+(.+)",
]

def detect_correction(text: str, current_slots: dict) -> tuple[str, str] | None:
    for pattern in CORRECTION_PATTERNS:
        match = re.search(pattern, text, re.IGNORECASE)
        if match:
            groups = match.groups()
            if len(groups) == 2:
                slot_hint, new_value = groups
                # Fuzzy match slot_hint to known slot names
                for slot_name in current_slots:
                    if slot_hint.lower() in slot_name.lower():
                        return slot_name, new_value.strip()
    return None

Composite Extractors

Production systems run multiple extractors and merge results with priority:

class CompositeExtractor:
    def __init__(self, extractors: list, priority: list[str]):
        self.extractors = extractors
        self.priority = priority  # extractor names in priority order

    def extract(self, text: str) -> list[Entity]:
        all_entities: dict[str, list[Entity]] = {}
        for extractor in self.extractors:
            name = extractor.__class__.__name__
            all_entities[name] = extractor.extract(text)

        # Merge: higher priority extractors win on overlapping spans
        merged = []
        occupied_spans = set()
        for name in self.priority:
            for entity in all_entities.get(name, []):
                span = (entity.start, entity.end)
                if not any(s <= entity.start < e or s < entity.end <= e
                          for s, e in occupied_spans):
                    merged.append(entity)
                    occupied_spans.add(span)
        return merged

Performance Benchmarks

Extractor	Entities/sec	F1 (domain)	Setup Effort
Regex	100,000+	95-100% (structured)	Low
CRF	10,000+	85-92%	Medium
spaCy NER	15,000+	80-90%	Low
Fine-tuned BERT	500-2,000	90-96%	High
Duckling (dates)	5,000+	97%+ (temporal)	Low

The production pattern: use regex/Duckling for structured entities (dates, emails, phone numbers), CRF or spaCy for domain entities with moderate training data, and fine-tuned Transformers only for complex entities where accuracy justifies the latency cost.

The one thing to remember: Production slot filling chains multiple extraction strategies — rules for structured data, ML for context-dependent entities — into a validation pipeline that tracks state across turns and handles corrections gracefully.