Skip to content

MessageHistory

This module contains the MessageHistory class, which is used to store the conversation history and the system prompt sent to OpenAI's chat API.

MessageHistory

Abstraction for the conversation history and the system prompt sent to OpenAI's chat API.

MessageHistory is used to store the a system prompt and the conversation history for a chat LLM. The system prompt is stored as the first message in the history. The conversation history is stored as a list of messages, where each message is a dictionary with "role" and "content" keys. In addition MessageHistory provides a set of methods for managing the messages and system prompt. MessageHistory is used by the ChatLLM generate and generate_async methods.

Parameters:

Name Type Description Default
max_messages int

The maximum number of messages to store in the history.

 0

Attributes:

Name Type Description
max_messages int

The maximum number of messages to store in the history.
messages list[dict[str, str]]

The conversation history.
Source code in llmflows/llms/message_history.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
class MessageHistory:
    """
    Abstraction for the conversation history and the system prompt sent to OpenAI's
    chat API.

    MessageHistory is used to store the a system prompt and the conversation history
    for a chat LLM. The system prompt is stored as the first message in the history.
    The conversation history is stored as a list of messages, where each message is a
    dictionary with "role" and "content" keys. In addition MessageHistory provides a
    set of methods for managing the messages and system prompt. MessageHistory is used
    by the ChatLLM generate and generate_async methods.

    Args:
        max_messages (int): The maximum number of messages to store in the history.

    Attributes:
        max_messages (int): The maximum number of messages to store in the history.
        messages (list[dict[str, str]]): The conversation history.
    """

    def __init__(
        self,
        max_messages: int = 0,
    ):
        self.max_messages = max_messages
        self.messages = []

    @property
    def system_prompt(self) -> str:
        """
        Returns the system prompt.

        Returns:
            str: The system prompt content.
        """
        if self.messages and self.messages[0]["role"] == "system":
            return self.messages[0]["content"]
        return ""

    @system_prompt.setter
    def system_prompt(self, new_prompt: str) -> None:
        """
        Sets the system prompt.

        Args:
            new_prompt (str): The new system prompt.
        """
        if not self.messages or self.messages[0]["role"] != "system":
            self.messages.insert(0, {"role": "system", "content": new_prompt})
        else:
            self.update_system_prompt(new_prompt)

    def update_system_prompt(self, new_prompt: str):
        """
        Updates the system prompt sent to the chat API.

        Args:
            new_prompt (str): The new system prompt.
        """
        self.messages[0] = {"role": "system", "content": new_prompt}

    def get_conversation_string(self):
        """
        Returns the conversation history as a string.
        """
        return "\n".join(
            [f"{message['role']}: {message['content']}" for message in self.messages]
        )

    @property
    def messages(self):
        """
        Returns the conversation history.
        """
        return self._messages

    @messages.setter
    def messages(self, value):
        """
        Sets the conversation history.

        Each message in the list should be a dictionary containing
        "role" and "content" keys.

        Args:
            value (list): A list of message dictionaries.

        Raises:
            ValueError: If the provided value is not a list or if any
            dictionary in the list is not a valid message.
        """
        if not isinstance(value, list):
            raise ValueError("messages must be a list of dicts")
        for item in value:
            self.validate_message(item)
        self._messages = value

    def add_user_message(self, message: str) -> None:
        """Adds a new user message to the conversation history."""
        self.add_message(message_str=message, role="user")

    def add_ai_message(self, message: str) -> None:
        """Adds a new assistant message to the conversation history."""
        self.add_message(message_str=message, role="assistant")

    def add_message(self, message_str: str, role: str) -> None:
        """
        Appends a new message to the message history.

        Args:
            message_str (str): Content of the message.
            role (str, optional): Role in the conversation. Can be "user" or
                "assistant". Defaults to "user".

        Returns:
            Updated conversation history.
        """
        role = self.validate_role(role)

        if self.max_messages and (len(self.messages) >= self.max_messages):
            self.remove_message(idx=1)

        self.messages.append({"role": role, "content": message_str})

    @staticmethod
    def validate_role(role: str) -> str:
        """
        Validates the role of a message.

        Args:
            role (str): The role of the message (either "user", "assistant",
                or "system").

        Returns:
            str: The validated role.

        Raises:
            ValueError: If the role is "system", as this should be updated using
                'update_system_prompt' method.
            ValueError: If the role is not "user" or "assistant".
        """

        if role not in ["user", "system", "assistant"]:
            raise ValueError(
                "The role should be either 'system', 'user' or 'assistant'"
            )

        return role

    def validate_message(self, message: dict[str, str]) -> dict[str, str]:
        """
        Validates a message for required fields.

        Args:
            message (dict[str, str]): The message to validate.

        Returns:
            The validated message.

        Raises:
            ValueError: If the provided message does not contain the necessary fields
                ("role" and "content").
        """
        if {"role", "content"}.issubset(message):
            self.validate_role(message["role"])
            return message
        raise ValueError("The provided message is not a valid message.")

    def replace_message(self, new_message: str, new_role: str, idx: int):
        """
        Replaces a message in the list of messages sent to the chat API.

        Args:
            new_message (str): The new message to replace the old message with.
            idx (int): The index of the message to replace.
        """
        message = {"role": new_role, "content": new_message}
        self.validate_message(message)
        self.messages[idx] = message

    def remove_message(self, idx=-1):
        """
        Removes a message from the list of messages sent to the chat API.

        Args:
            idx (int): The index of the message to remove.
        """
        self.messages.pop(idx)

messages property writable

Returns the conversation history.

system_prompt: str property writable

Returns the system prompt.

Returns:

Name Type Description
str str

The system prompt content.

add_ai_message(message)

Adds a new assistant message to the conversation history.

Source code in llmflows/llms/message_history.py 
109
110
111
def add_ai_message(self, message: str) -> None:
    """Adds a new assistant message to the conversation history."""
    self.add_message(message_str=message, role="assistant")

add_message(message_str, role)

Appends a new message to the message history.

Parameters:

Name Type Description Default
message_str str

Content of the message.

 required
role str

Role in the conversation. Can be "user" or "assistant". Defaults to "user".

 required

Returns:

Type Description
None

Updated conversation history.
Source code in llmflows/llms/message_history.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def add_message(self, message_str: str, role: str) -> None:
    """
    Appends a new message to the message history.

    Args:
        message_str (str): Content of the message.
        role (str, optional): Role in the conversation. Can be "user" or
            "assistant". Defaults to "user".

    Returns:
        Updated conversation history.
    """
    role = self.validate_role(role)

    if self.max_messages and (len(self.messages) >= self.max_messages):
        self.remove_message(idx=1)

    self.messages.append({"role": role, "content": message_str})

add_user_message(message)

Adds a new user message to the conversation history.

Source code in llmflows/llms/message_history.py 
105
106
107
def add_user_message(self, message: str) -> None:
    """Adds a new user message to the conversation history."""
    self.add_message(message_str=message, role="user")

get_conversation_string()

Returns the conversation history as a string.

Source code in llmflows/llms/message_history.py 
69
70
71
72
73
74
75
def get_conversation_string(self):
    """
    Returns the conversation history as a string.
    """
    return "\n".join(
        [f"{message['role']}: {message['content']}" for message in self.messages]
    )

remove_message(idx=-1)

Removes a message from the list of messages sent to the chat API.

Parameters:

Name Type Description Default
idx int

The index of the message to remove.

 -1
Source code in llmflows/llms/message_history.py 
188
189
190
191
192
193
194
195
def remove_message(self, idx=-1):
    """
    Removes a message from the list of messages sent to the chat API.

    Args:
        idx (int): The index of the message to remove.
    """
    self.messages.pop(idx)

replace_message(new_message, new_role, idx)

Replaces a message in the list of messages sent to the chat API.

Parameters:

Name Type Description Default
new_message str

The new message to replace the old message with.

 required
idx int

The index of the message to replace.

 required
Source code in llmflows/llms/message_history.py 
176
177
178
179
180
181
182
183
184
185
186
def replace_message(self, new_message: str, new_role: str, idx: int):
    """
    Replaces a message in the list of messages sent to the chat API.

    Args:
        new_message (str): The new message to replace the old message with.
        idx (int): The index of the message to replace.
    """
    message = {"role": new_role, "content": new_message}
    self.validate_message(message)
    self.messages[idx] = message

update_system_prompt(new_prompt)

Updates the system prompt sent to the chat API.

Parameters:

Name Type Description Default
new_prompt str

The new system prompt.

 required
Source code in llmflows/llms/message_history.py 
60
61
62
63
64
65
66
67
def update_system_prompt(self, new_prompt: str):
    """
    Updates the system prompt sent to the chat API.

    Args:
        new_prompt (str): The new system prompt.
    """
    self.messages[0] = {"role": "system", "content": new_prompt}

validate_message(message)

Validates a message for required fields.

Parameters:

Name Type Description Default
message dict[str, str]

The message to validate.

 required

Returns:

Type Description
dict[str, str]

The validated message.

Raises:

Type Description
ValueError

If the provided message does not contain the necessary fields ("role" and "content").
Source code in llmflows/llms/message_history.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def validate_message(self, message: dict[str, str]) -> dict[str, str]:
    """
    Validates a message for required fields.

    Args:
        message (dict[str, str]): The message to validate.

    Returns:
        The validated message.

    Raises:
        ValueError: If the provided message does not contain the necessary fields
            ("role" and "content").
    """
    if {"role", "content"}.issubset(message):
        self.validate_role(message["role"])
        return message
    raise ValueError("The provided message is not a valid message.")

validate_role(role) staticmethod

Validates the role of a message.

Parameters:

Name Type Description Default
role str

The role of the message (either "user", "assistant", or "system").

 required

Returns:

Name Type Description
str str

The validated role.

Raises:

Type Description
ValueError

If the role is "system", as this should be updated using 'update_system_prompt' method.
ValueError

If the role is not "user" or "assistant".
Source code in llmflows/llms/message_history.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@staticmethod
def validate_role(role: str) -> str:
    """
    Validates the role of a message.

    Args:
        role (str): The role of the message (either "user", "assistant",
            or "system").

    Returns:
        str: The validated role.

    Raises:
        ValueError: If the role is "system", as this should be updated using
            'update_system_prompt' method.
        ValueError: If the role is not "user" or "assistant".
    """

    if role not in ["user", "system", "assistant"]:
        raise ValueError(
            "The role should be either 'system', 'user' or 'assistant'"
        )

    return role