Conversational AI

Dialogue System Components

Modern conversational AI systems combine multiple components: understanding user intent, managing conversation state, generating appropriate responses, and maintaining context across turns.

Building a Chatbot Framework

from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
import json

class Intent(Enum):
    GREETING = "greeting"
    FAREWELL = "farewell"
    QUESTION = "question"
    COMPLAINT = "complaint"
    UNKNOWN = "unknown"

@dataclass
class Message:
    role: str
    content: str
    timestamp: float
    metadata: Dict = None

@dataclass
class DialogState:
    user_id: str
    current_intent: Intent
    context: Dict
    history: List[Message]
    slot_values: Dict

class DialogManager:
    def __init__(self):
        self.sessions: Dict[str, DialogState] = {}
        self.max_history = 20
    
    def get_or_create_session(self, user_id: str) -> DialogState:
        if user_id not in self.sessions:
            self.sessions[user_id] = DialogState(
                user_id=user_id,
                current_intent=Intent.UNKNOWN,
                context={},
                history=[],
                slot_values={}
            )
        return self.sessions[user_id]
    
    def update_state(self, user_id: str, message: Message, intent: Intent):
        state = self.get_or_create_session(user_id)
        state.current_intent = intent
        state.history.append(message)
        
        if len(state.history) > self.max_history:
            state.history = state.history[-self.max_history:]
    
    def get_context_window(self, user_id: str, max_turns: int = 10) -> List[Dict]:
        state = self.get_or_create_session(user_id)
        recent = state.history[-max_turns:]
        return [{"role": msg.role, "content": msg.content} for msg in recent]

manager = DialogManager()

Intent Classification

import openai
from typing import Tuple

class IntentClassifier:
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
        self.intent_descriptions = {
            Intent.GREETING: "User is saying hello or greeting",
            Intent.FAREWELL: "User is saying goodbye",
            Intent.QUESTION: "User is asking a question",
            Intent.COMPLAINT: "User is expressing dissatisfaction",
        }
    
    def classify(self, text: str) -> Tuple[Intent, float]:
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": f"""Classify user message into one of these intents:
{json.dumps(self.intent_descriptions, indent=2)}

Return JSON with "intent" and "confidence" fields."""},
                {"role": "user", "content": text}
            ],
            temperature=0,
            response_format={"type": "json_object"}
        )
        
        result = json.loads(response.choices[0].message.content)
        
        intent_map = {i.value: i for i in Intent}
        intent = intent_map.get(result["intent"], Intent.UNKNOWN)
        confidence = result.get("confidence", 0.5)
        
        return intent, confidence

classifier = IntentClassifier(api_key="your-api-key")
intent, confidence = classifier.classify("Hello, how are you?")
print(f"Intent: {intent.value}, Confidence: {confidence:.2f}")

Response Generation

class ResponseGenerator:
    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
    
    def generate_response(
        self,
        user_message: str,
        context: List[Dict],
        persona: str = "helpful assistant"
    ) -> str:
        messages = [
            {"role": "system", "content": f"""You are a {persona}.
Use the conversation context to provide relevant responses.
Be concise and helpful."""}
        ]
        
        messages.extend(context[-10:])
        messages.append({"role": "user", "content": user_message})
        
        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            temperature=0.7,
            max_tokens=150
        )
        
        return response.choices[0].message.content
    
    def generate_with_functions(
        self,
        user_message: str,
        available_functions: List[Dict]
    ) -> Dict:
        response = self.client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "Use available functions when appropriate."},
                {"role": "user", "content": user_message}
            ],
            functions=available_functions,
            function_call="auto"
        )
        
        message = response.choices[0].message
        
        if message.function_call:
            return {
                "type": "function",
                "name": message.function_call.name,
                "arguments": json.loads(message.function_call.arguments)
            }
        
        return {"type": "text", "content": message.content}

generator = ResponseGenerator(api_key="your-api-key")
response = generator.generate_response(
    "What's the weather in New York?",
    context=[],
    persona="weather assistant"
)

Context Management

from collections import deque
import hashlib

class ContextManager:
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
        self.context_store: Dict[str, deque] = {}
    
    def add_message(self, session_id: str, role: str, content: str):
        if session_id not in self.context_store:
            self.context_store[session_id] = deque(maxlen=100)
        
        self.context_store[session_id].append({
            "role": role,
            "content": content,
            "token_count": self.estimate_tokens(content)
        })
        
        self._trim_context(session_id)
    
    def get_context(self, session_id: str, max_turns: int = 10) -> List[Dict]:
        if session_id not in self.context_store:
            return []
        
        messages = list(self.context_store[session_id])
        return [{"role": m["role"], "content": m["content"]} for m in messages[-max_turns:]]
    
    def _trim_context(self, session_id: str):
        messages = self.context_store[session_id]
        total_tokens = sum(m["token_count"] for m in messages)
        
        while total_tokens > self.max_tokens and len(messages) > 2:
            removed = messages.popleft()
            total_tokens -= removed["token_count"]
    
    def estimate_tokens(self, text: str) -> int:
        return len(text.split()) * 1.3

context_mgr = ContextManager(max_tokens=3000)
context_mgr.add_message("session_1", "user", "Hello!")
context_mgr.add_message("session_1", "assistant", "Hi there! How can I help?")
context = context_mgr.get_context("session_1")

Complete Chatbot

class Chatbot:
    def __init__(self, api_key: str):
        self.classifier = IntentClassifier(api_key)
        self.generator = ResponseGenerator(api_key)
        self.dialog_manager = DialogManager()
        self.context_manager = ContextManager()
    
    def chat(self, user_id: str, message: str) -> str:
        import time
        
        msg = Message(role="user", content=message, timestamp=time.time())
        intent, confidence = self.classifier.classify(message)
        
        self.dialog_manager.update_state(user_id, msg, intent)
        self.context_manager.add_message(user_id, "user", message)
        
        context = self.context_manager.get_context(user_id)
        
        if intent == Intent.FAREWELL:
            response = "Goodbye! Have a great day!"
        elif intent == Intent.GREETING:
            response = "Hello! How can I assist you today?"
        else:
            response = self.generator.generate_response(message, context)
        
        self.context_manager.add_message(user_id, "assistant", response)
        
        return response

bot = Chatbot(api_key="your-api-key")
response = bot.chat("user_123", "What is machine learning?")
print(response)

Best Practices

• Maintain conversation history with proper truncation
• Implement fallback responses for unrecognized intents
• Use slot filling for structured data collection
• Add personality and tone consistency
• Handle multi-turn conversations gracefully
• Implement error recovery mechanisms
• Monitor conversation quality metrics