trading_agent.agent.conversation

A desk's conversation with its model: the system prompt and the messages so far.

Messages are appended, and only ever changed in two ways: compression replaces the whole history at once with a summary and the current turn, which is a fresh conversation as far as the model can tell, and older pictures are taken out so that only the newest few are sent with every model call.

The conversation also remembers how many tokens the model said it held at the last call, so the compressor can tell how full the context window is.

Typical usage example:

conversation = Conversation('You are a trading desk.') conversation.append(ConversationMessage.user('Hello'))

Conversation

Conversation(system_prompt)

An append-only list of messages with a fixed system prompt.

Attributes:

Name	Type	Description
system_prompt		The instructions sent before every message.
measured_tokens		The tokens the model reported for the conversation at its last call, or 0 before any call.

Creates an empty conversation.

Parameters:

Name	Type	Description	Default
system_prompt	str	The instructions sent before every message.	required

Source code in src/trading_agent/agent/conversation.py

def __init__(self, system_prompt: str):
    """Creates an empty conversation.

    Args:
        system_prompt (str): The instructions sent before every message.
    """
    self.system_prompt = system_prompt
    self.measured_tokens = 0
    self._measured_message_count = 0
    self._messages = []

append

append(message)

Adds a message to the end.

Parameters:

Name	Type	Description	Default
message	ConversationMessage	The message to add.	required

Source code in src/trading_agent/agent/conversation.py

def append(self, message: messages.ConversationMessage) -> None:
    """Adds a message to the end.

    Args:
        message (messages.ConversationMessage): The message to add.
    """
    self._messages.append(message)

copy

copy()

A separate conversation with the same prompt and messages, which can be extended without changing this one.

Returns:

Type	Description
Conversation	The copy.

Source code in src/trading_agent/agent/conversation.py

def copy(self) -> Conversation:
    """A separate conversation with the same prompt and messages, which can be extended without changing this one.

    Returns:
        Conversation: The copy.
    """
    duplicate = Conversation(self.system_prompt)
    duplicate.measured_tokens = self.measured_tokens
    duplicate._measured_message_count = self._measured_message_count
    duplicate._messages = list(self._messages)
    return duplicate

messages property

messages

The messages so far, oldest first, as a read-only tuple.

record_usage

record_usage(tokens)

Remembers how many tokens the model reported at the call that just finished.

Parameters:

Name	Type	Description	Default
tokens	int	The prompt tokens plus the reply's tokens.	required

Source code in src/trading_agent/agent/conversation.py

def record_usage(self, tokens: int) -> None:
    """Remembers how many tokens the model reported at the call that just finished.

    Args:
        tokens (int): The prompt tokens plus the reply's tokens.
    """
    self.measured_tokens = tokens
    self._measured_message_count = len(self._messages)

estimated_tokens

estimated_tokens()

How many tokens the next call will send, as closely as can be told now.

It is the last measured count plus an estimate for the messages added since, such as tool results: three characters per token of text, and 1500 tokens for each picture.

Returns:

Type	Description
int	The estimate.

Source code in src/trading_agent/agent/conversation.py

def estimated_tokens(self) -> int:
    """How many tokens the next call will send, as closely as can be told now.

    It is the last measured count plus an estimate for the messages added since, such as tool results: three characters per token of text, and 1500 tokens for each picture.

    Returns:
        int: The estimate.
    """
    added_characters = 0
    added_images = 0
    for message in self._messages[self._measured_message_count :]:
        added_characters += len(message.text)
        for call in message.tool_calls:
            added_characters += len(call.name) + len(str(call.arguments))
        added_images += len(message.images)
        if message.image is not None:
            added_images += 1
    return (
        self.measured_tokens
        + added_characters // _CHARACTERS_PER_TOKEN
        + added_images * _TOKENS_PER_IMAGE
    )

keep_newest_pictures

keep_newest_pictures(count)

Takes every picture but the newest few out of the conversation, leaving a note in each one's place.

Every message is sent again with each model call, so a picture left in the history costs about 1500 tokens on every later call.

Parameters:

Name	Type	Description	Default
count	int	How many of the newest pictures to keep.	required

Returns:

Type	Description
int	How many pictures were taken out.

Source code in src/trading_agent/agent/conversation.py

def keep_newest_pictures(self, count: int) -> int:
    """Takes every picture but the newest few out of the conversation, leaving a note in each one's place.

    Every message is sent again with each model call, so a picture left in the history costs about 1500 tokens on every later call.

    Args:
        count (int): How many of the newest pictures to keep.

    Returns:
        int: How many pictures were taken out.
    """
    seen = 0
    removed = 0
    for index in range(len(self._messages) - 1, -1, -1):
        message = self._messages[index]
        pictures = len(message.images)
        if message.image is not None:
            pictures += 1
        if pictures == 0:
            continue
        seen += pictures
        if seen - pictures >= count:
            self._messages[index] = self._without_pictures(message)
            removed += pictures
    return removed

restart

restart(new_messages)

Replaces the whole history at once, for compression.

Parameters:

Name	Type	Description	Default
new_messages	Sequence[ConversationMessage]	The messages the conversation starts again with.	required

Source code in src/trading_agent/agent/conversation.py

def restart(
    self,
    new_messages: Sequence[messages.ConversationMessage],
) -> None:
    """Replaces the whole history at once, for compression.

    Args:
        new_messages (Sequence[messages.ConversationMessage]): The messages the conversation starts again with.
    """
    self._messages = list(new_messages)
    self.measured_tokens = 0
    self._measured_message_count = 0