trading_agent.agent.agent_loop

The agent loop: ask the model, run the tools it asks for, give it the results, repeat.

This is the loop at the heart of Pi and every other agent. One run of the loop is one turn: it ends when the model replies without asking for a tool, or when the round limit is reached.

With a context compressor, a conversation that has filled 70% of the model's window is compressed before the next turn: the desk first gets a silent memory-flush turn to save what matters, then the older messages are replaced by a summary. Between the tool rounds of one long turn, a conversation past 85% is compressed at once, without the flush.

Before every model call, all but the four newest pictures are taken out of the conversation, because each one would otherwise be sent again with every later call.

Typical usage example:

loop = AgentLoop('main', backend, executor, event_bus, maximum_rounds=20) result = await loop.run_turn(conversation, trigger)

AgentLoop

AgentLoop(desk_name, backend, executor, event_bus, maximum_rounds, compressor=None, steering_inbox=None, image_preparer=None)

Runs turns for one desk against one backend.

Attributes:

Name	Type	Description
backend		The model the loop asks.

Creates the loop.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk the loop runs for, used to tag its events.	required
backend	ModelBackend	The model to ask.	required
executor	ToolCallExecutor	What runs the model's tool calls.	required
event_bus	EventBus	Where every step is published.	required
maximum_rounds	int	The most model calls one turn may make.	required
compressor	ContextCompressor | None	Compresses the conversation when it gets too full, or None to let it grow.	None
steering_inbox	SteeringInbox | None	The desk's mid-turn messages, added to the conversation after each tool round, or None for a loop that cannot be steered, such as a subagent's.	None
image_preparer	ImagePreparer | None	Sizes the pictures sent with a message for the model; None for a new one.	None

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

def __init__(
    self,
    desk_name: str,
    backend: backend_module.ModelBackend,
    executor: tool_call_executor_module.ToolCallExecutor,
    event_bus: event_bus_module.EventBus,
    maximum_rounds: int,
    compressor: context_compressor_module.ContextCompressor | None = None,
    steering_inbox: steering_inbox_module.SteeringInbox | None = None,
    image_preparer: image_preparer_module.ImagePreparer | None = None,
):
    """Creates the loop.

    Args:
        desk_name (str): The desk the loop runs for, used to tag its events.
        backend (backend_module.ModelBackend): The model to ask.
        executor (tool_call_executor_module.ToolCallExecutor): What runs the model's tool calls.
        event_bus (event_bus_module.EventBus): Where every step is published.
        maximum_rounds (int): The most model calls one turn may make.
        compressor (context_compressor_module.ContextCompressor | None): Compresses the conversation when it gets too full, or None to let it grow.
        steering_inbox (steering_inbox_module.SteeringInbox | None): The desk's mid-turn messages, added to the conversation after each tool round, or None for a loop that cannot be steered, such as a subagent's.
        image_preparer (image_preparer_module.ImagePreparer | None): Sizes the pictures sent with a message for the model; None for a new one.
    """
    self._desk_name = desk_name
    self.backend = backend
    self._executor = executor
    self._event_bus = event_bus
    self._maximum_rounds = maximum_rounds
    self._compressor = compressor
    self._steering_inbox = steering_inbox
    self._image_preparer = (
        image_preparer or image_preparer_module.ImagePreparer()
    )

tool_definitions

tool_definitions()

The tools the model is offered on every call.

Returns:

Type	Description
list[dict[str, Any]]	One function-calling definition per tool.

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

def tool_definitions(self) -> list[dict[str, Any]]:
    """The tools the model is offered on every call.

    Returns:
        list[dict[str, Any]]: One function-calling definition per tool.
    """
    return self._executor.tool_registry.definitions()

run_turn async

run_turn(conversation, trigger)

Runs one turn for a trigger, appending everything to the conversation.

Parameters:

Name	Type	Description	Default
conversation	Conversation	The desk's conversation, which grows by this turn's messages.	required
trigger	Trigger	The reason for the turn.	required

Returns:

Type	Description
TurnResult	The model's final text.

Raises:

Type	Description
BackendUnavailableError	The model could not be reached.

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

async def run_turn(
    self,
    conversation: conversation_module.Conversation,
    trigger: trigger_module.Trigger,
) -> turn_runner.TurnResult:
    """Runs one turn for a trigger, appending everything to the conversation.

    Args:
        conversation (conversation_module.Conversation): The desk's conversation, which grows by this turn's messages.
        trigger (trigger_module.Trigger): The reason for the turn.

    Returns:
        turn_runner.TurnResult: The model's final text.

    Raises:
        backend_module.BackendUnavailableError: The model could not be reached.
    """
    if self._compressor is not None and self._compressor.is_full(
        conversation
    ):
        await self._flush_and_compress(conversation, trigger, 'threshold')
    return await self._run_rounds(conversation, trigger)

compact async

compact(conversation, trigger)

Flushes and compresses the conversation now, for the /compact command.

Parameters:

Name	Type	Description	Default
conversation	Conversation	The desk's conversation.	required
trigger	Trigger	The command's trigger, used to tag events.	required

Returns:

Type	Description
bool	True when the conversation was compressed, False when there is no compressor or nothing to compress.

Raises:

Type	Description
BackendUnavailableError	The model could not be reached.

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

async def compact(
    self,
    conversation: conversation_module.Conversation,
    trigger: trigger_module.Trigger,
) -> bool:
    """Flushes and compresses the conversation now, for the /compact command.

    Args:
        conversation (conversation_module.Conversation): The desk's conversation.
        trigger (trigger_module.Trigger): The command's trigger, used to tag events.

    Returns:
        bool: True when the conversation was compressed, False when there is no compressor or nothing to compress.

    Raises:
        backend_module.BackendUnavailableError: The model could not be reached.
    """
    if self._compressor is None or not conversation.messages:
        return False
    return await self._flush_and_compress(conversation, trigger, 'command')