Conversational AI › Technical Architecture

messages = [
    {
        "role": "system",
        "content": "You are a helpful travel assistant. Help users plan trips and book travel.",
    },
    {"role": "user", "content": "I want to go somewhere warm this winter."},
    {
        "role": "assistant",
        "content": "Great! There are many wonderful warm destinations for winter travel. "
        "Are you interested in beach destinations, cultural cities, or adventure travel? "
        "And do you have a particular region in mind, like the Caribbean, Southeast Asia, or Europe?",
    },
    {
        "role": "user",
        "content": "Beach sounds perfect. I'm open to Southeast Asia.",
    },
]

def create_conversation(system_prompt):
    """Initialize a conversation with a system prompt."""
    return [{"role": "system", "content": system_prompt}]


def add_user_message(messages, user_input):
    """Append a user message to the conversation."""
    messages.append({"role": "user", "content": user_input})
    return messages


def add_assistant_message(messages, assistant_response):
    """Append an assistant message to the conversation."""
    messages.append({"role": "assistant", "content": assistant_response})
    return messages


def format_for_display(messages):
    """Format conversation for readable output."""
    output_lines = []
    for msg in messages:
        if msg["role"] == "system":
            continue
        prefix = "You: " if msg["role"] == "user" else "Assistant: "
        output_lines.append(f"{prefix}{msg['content']}")
    return "\n\n".join(output_lines)

def trim_conversation(messages, max_turns=10):
    """
    Keep system message and the most recent max_turns user+assistant pairs.

    Args:
        messages: Full conversation history
        max_turns: Maximum number of user/assistant turn pairs to keep

    Returns:
        Trimmed message list
    """
    system_messages = [m for m in messages if m["role"] == "system"]
    non_system = [m for m in messages if m["role"] != "system"]

    # Each "turn" is one user + one assistant message
    max_messages = max_turns * 2
    if len(non_system) > max_messages:
        non_system = non_system[-max_messages:]

    return system_messages + non_system


def estimate_token_count(messages, chars_per_token=4):
    """Rough token count estimate based on character count."""
    total_chars = sum(len(m["content"]) for m in messages)
    return total_chars // chars_per_token

class RestaurantBookingState:
    """Track the state of a restaurant booking conversation."""

    REQUIRED_SLOTS = ["date", "time", "party_size", "cuisine"]
    OPTIONAL_SLOTS = ["dietary_restrictions", "price_range", "occasion"]

    def __init__(self):
        self.slots = {
            slot: None for slot in self.REQUIRED_SLOTS + self.OPTIONAL_SLOTS
        }
        self.confirmed = False

    def fill_slot(self, slot_name, value):
        """Fill a slot with an extracted value."""
        if slot_name in self.slots:
            self.slots[slot_name] = value

    def missing_required(self):
        """Return list of required slots that haven't been filled."""
        return [s for s in self.REQUIRED_SLOTS if self.slots[s] is None]

    def is_complete(self):
        """Check if all required slots are filled."""
        return len(self.missing_required()) == 0

    def get_next_question(self):
        """Return a question to ask for the next missing required slot."""
        missing = self.missing_required()
        if not missing:
            return None

        questions = {
            "date": "What date would you like to book for?",
            "time": "What time would you prefer?",
            "party_size": "How many people will be dining?",
            "cuisine": "Do you have a cuisine preference?",
        }
        return questions.get(
            missing[0], f"Could you tell me your {missing[0]}?"
        )

    def summarize(self):
        """Return a summary of filled slots."""
        filled = {k: v for k, v in self.slots.items() if v is not None}
        return filled

def create_summary_from_history(messages, topic="the conversation so far"):
    """
    Create a compact summary of conversation history.

    In production, this would call an LLM to generate the summary.
    Here we simulate the output format.
    """
    user_msgs = [m for m in messages if m["role"] == "user"]
    topics_discussed = len(user_msgs)

    # In production: call LLM to summarize
    # For demonstration: extract key info from messages
    summary_parts = []
    for msg in messages:
        if msg["role"] == "user" and len(msg["content"]) > 20:
            summary_parts.append(msg["content"][:60])

    return (
        f"[CONVERSATION SUMMARY: The user discussed {topics_discussed} topics including: "
        + "; ".join(summary_parts[:3])
        + ("..." if len(summary_parts) > 3 else "")
        + "]"
    )


def compress_conversation(messages, keep_recent=4, system_prompt=None):
    """
    Compress old conversation turns into a summary message.

    Args:
        messages: Full conversation history
        keep_recent: Number of recent turns to keep verbatim
        system_prompt: Optional system prompt to prepend

    Returns:
        Compressed message list with summary
    """
    non_system = [m for m in messages if m["role"] != "system"]

    if len(non_system) <= keep_recent * 2:
        return messages  # Not long enough to compress

    to_summarize = non_system[: -keep_recent * 2]
    to_keep = non_system[-keep_recent * 2 :]

    summary_text = create_summary_from_history(to_summarize)
    summary_message = {"role": "system", "content": summary_text}

    system_messages = [m for m in messages if m["role"] == "system"]
    return system_messages + [summary_message] + to_keep