Dialogue Systems

Dialogue systems enable multi-turn conversations with context tracking, intent recognition, slot filling, and coherent response generation.

Dialogue State Tracker

from dataclasses import dataclass, field
from typing import Dict, List, Optional
import json

@dataclass
class DialogueState:
    turn: int = 0
    intent: str = ""
    slots: Dict[str, str] = field(default_factory=dict)
    history: List[Dict] = field(default_factory=list)
    context: Dict = field(default_factory=dict)

class DialogueStateTracker:
    def __init__(self):
        self.sessions: Dict[str, DialogueState] = {}

    def get_state(self, session_id: str) -> DialogueState:
        if session_id not in self.sessions:
            self.sessions[session_id] = DialogueState()
        return self.sessions[session_id]

    def update_state(self, session_id: str, intent: str, entities: Dict):
        state = self.get_state(session_id)
        state.intent = intent
        state.slots.update(entities)
        state.turn += 1
        state.history.append({
            "turn": state.turn,
            "intent": intent,
            "entities": entities
        })
        return state

    def reset_state(self, session_id: str):
        self.sessions[session_id] = DialogueState()

    def get_missing_slots(self, session_id: str, required_slots: List[str]) -> List[str]:
        state = self.get_state(session_id)
        return [slot for slot in required_slots if slot not in state.slots]

    def is_complete(self, session_id: str, required_slots: List[str]) -> bool:
        return len(self.get_missing_slots(session_id, required_slots)) == 0

# Usage
tracker = DialogueStateTracker()
tracker.update_state("user123", "book_flight", {"destination": "New York"})
tracker.update_state("user123", "provide_date", {"date": "2024-03-15"})
missing = tracker.get_missing_slots("user123", ["destination", "date", "passengers"])
# ["passengers"]

Intent Classification

from transformers import pipeline

class IntentClassifier:
    def __init__(self, model_name: str = "microsoft/DialoGPT-medium"):
        self.classifier = pipeline("zero-shot-classification")
        self.intents = [
            "book_flight", "check_status", "cancel_booking",
            "get_info", "complaint", "greeting", "goodbye"
        ]

    def classify(self, text: str) -> dict:
        result = self.classifier(text, self.intents)
        return {
            "intent": result["labels"][0],
            "confidence": result["scores"][0],
            "all_scores": dict(zip(result["labels"], result["scores"]))
        }

    def classify_with_threshold(self, text: str, threshold: float = 0.5) -> dict:
        result = self.classify(text)
        if result["confidence"] < threshold:
            return {"intent": "unknown", "confidence": result["confidence"]}
        return result

# Usage
classifier = IntentClassifier()
result = classifier.classify("I want to book a flight to Paris")
# {"intent": "book_flight", "confidence": 0.92}

Slot Filling with LLM

class SlotFiller:
    def __init__(self, llm):
        self.llm = llm
        self.slot_schemas = {
            "book_flight": ["destination", "date", "passengers", "class"],
            "book_hotel": ["city", "check_in", "check_out", "guests"],
            "restaurant": ["cuisine", "location", "time", "party_size"]
        }

    def extract_slots(self, text: str, intent: str) -> dict:
        schema = self.slot_schemas.get(intent, [])
        prompt = f"""Extract these slots from the text: {schema}
        Text: {text}
        Return JSON: {{slot_name: value}} or null if not found:"""

        response = self.llm.invoke(prompt).content
        import json
        return json.loads(response)

    def fill_missing_slots(self, text: str, intent: str, existing_slots: dict) -> dict:
        new_slots = self.extract_slots(text, intent)
        all_slots = {**existing_slots, **new_slots}
        return {k: v for k, v in all_slots.items() if v is not None}

    def generate_slot_query(self, missing_slots: list) -> str:
        if not missing_slots:
            return ""
        slot_descriptions = {
            "destination": "Where would you like to go?",
            "date": "When would you like to travel?",
            "passengers": "How many passengers?",
            "class": "What class would you prefer?"
        }
        questions = [slot_descriptions.get(s, f"Please provide {s}") for s in missing_slots[:2]]
        return " ".join(questions)

# Usage
filler = SlotFiller(llm)
slots = filler.extract_slots("I want to fly to Tokyo next Monday", "book_flight")
# {"destination": "Tokyo", "date": "next Monday"}

Dialogue Policy

class DialoguePolicy:
    def __init__(self, llm):
        self.llm = llm
        self.action_space = [
            "ask_slot", "confirm", "inform", "execute",
            "clarify", "greet", "farewell"
        ]

    def decide_action(self, state: DialogueState, required_slots: list) -> str:
        if state.turn == 0:
            return "greet"

        if not state.intent:
            return "clarify"

        missing = [s for s in required_slots if s not in state.slots]
        if missing:
            return "ask_slot"

        if state.turn <= 3:
            return "confirm"

        return "execute"

    def generate_response(self, action: str, state: DialogueState) -> str:
        if action == "greet":
            return "Hello! How can I help you today?"
        elif action == "ask_slot":
            missing = [s for s in ["destination", "date"] if s not in state.slots]
            if missing:
                return f"I'd be happy to help! What {missing[0]} would you like?"
        elif action == "confirm":
            slots_str = ", ".join([f"{k}: {v}" for k, v in state.slots.items()])
            return f"Let me confirm: {slots_str}. Is this correct?"
        elif action == "execute":
            return "Great! I'm processing your request now."
        return "I'm not sure I understand. Could you rephrase?"

    def decide_action_llm(self, state: DialogueState) -> str:
        prompt = f"""Based on this dialogue state, choose the next action:
        Turn: {state.turn}
        Intent: {state.intent}
        Slots: {state.slots}
        Actions: {self.action_space}
        Best action:"""
        return self.llm.invoke(prompt).content.strip()

# Usage
policy = DialoguePolicy(llm)
action = policy.decide_action(state, ["destination", "date"])
response = policy.generate_response(action, state)

Full Dialogue System

class DialogueSystem:
    def __init__(self, intent_classifier, slot_filler, policy, state_tracker):
        self.classifier = intent_classifier
        self.slot_filler = slot_filler
        self.policy = policy
        self.tracker = state_tracker
        self.required_slots = {
            "book_flight": ["destination", "date"],
            "check_status": ["booking_id"]
        }

    def process_turn(self, session_id: str, user_input: str) -> str:
        intent_result = self.classifier.classify(user_input)
        intent = intent_result["intent"]

        state = self.tracker.get_state(session_id)
        existing_slots = state.slots.copy()

        new_slots = self.slot_filler.fill_missing_slots(
            user_input, intent, existing_slots
        )
        self.tracker.update_state(session_id, intent, new_slots)

        required = self.required_slots.get(intent, [])
        action = self.policy.decide_action(
            self.tracker.get_state(session_id), required
        )

        response = self.policy.generate_response(action, self.tracker.get_state(session_id))
        return response

# Usage
system = DialogueSystem(classifier, filler, policy, tracker)
response1 = system.process_turn("s1", "I want to book a flight")
response2 = system.process_turn("s1", "To New York next week")

Key Takeaways

• State tracking maintains conversation context across turns
• Intent classification identifies user goals from utterances
• Slot filling extracts specific entities needed for actions
• Policy determines the next dialogue action based on state
• Full systems combine NLU, state tracking, policy, and NLG