Agent Base

The base Agent class provides the core interface and functionality that all rLLM agents inherit from.

rllm.agents.agent.BaseAgent

Bases: ABC

Source code in rllm/agents/agent.py

class BaseAgent(ABC):
    @property
    def chat_completions(self) -> list[dict[str, str]]:
        """Converts agent's internal state into a list of OAI chat completions."""
        return []

    @property
    def trajectory(self) -> Trajectory:
        """Converts agent's internal state into a Trajectory object."""
        return Trajectory()

    def update_from_env(self, observation: Any, reward: float, done: bool, info: dict, **kwargs):
        """
        Updates the agent's internal state after an environment step.

        Args:
            observation (Any): The observation after stepping through environment.
            reward (float): The reward received after taking the action.
            done (bool): Whether the episode has ended due to termination.
            info (dict): Additional metadata from the environment.
        """
        raise NotImplementedError("Subclasses must implement this method if using AgentExecutionEngine")

    def update_from_model(self, response: str, **kwargs) -> Action:
        """
        Updates the agent's internal state after the model generates a response.

        Args:
            response (str): The response from the model.

        Returns:
            None
        """
        raise NotImplementedError("Subclasses must implement this method if using AgentExecutionEngine")

    @abstractmethod
    def reset(self):
        """
        Resets the agent's internal state, typically called at the beginning of a new episode.

        This function should clear any stored history or state information necessary
        for a fresh interaction.

        Returns:
            None
        """
        return

    def get_current_state(self) -> Step | None:
        """
        Returns the agent's current state as a dictionary.

        This method provides access to the agent's internal state at the current step,
        which can be useful for debugging, logging, or state management.

        Returns:
            Step: The agent's current state.
        """
        if not self.trajectory.steps:
            return None
        return self.trajectory.steps[-1]

chat_completions property

chat_completions: list[dict[str, str]]

Converts agent's internal state into a list of OAI chat completions.

trajectory property

trajectory: Trajectory

Converts agent's internal state into a Trajectory object.

update_from_env

update_from_env(observation: Any, reward: float, done: bool, info: dict, **kwargs)

Updates the agent's internal state after an environment step.

Parameters:

Name	Type	Description	Default
observation	Any	The observation after stepping through environment.	required
reward	float	The reward received after taking the action.	required
done	bool	Whether the episode has ended due to termination.	required
info	dict	Additional metadata from the environment.	required

Source code in rllm/agents/agent.py

def update_from_env(self, observation: Any, reward: float, done: bool, info: dict, **kwargs):
    """
    Updates the agent's internal state after an environment step.

    Args:
        observation (Any): The observation after stepping through environment.
        reward (float): The reward received after taking the action.
        done (bool): Whether the episode has ended due to termination.
        info (dict): Additional metadata from the environment.
    """
    raise NotImplementedError("Subclasses must implement this method if using AgentExecutionEngine")

update_from_model

update_from_model(response: str, **kwargs) -> Action

Updates the agent's internal state after the model generates a response.

Parameters:

Name	Type	Description	Default
response	str	The response from the model.	required

Returns:

Type	Description
Action	None

Source code in rllm/agents/agent.py

def update_from_model(self, response: str, **kwargs) -> Action:
    """
    Updates the agent's internal state after the model generates a response.

    Args:
        response (str): The response from the model.

    Returns:
        None
    """
    raise NotImplementedError("Subclasses must implement this method if using AgentExecutionEngine")

reset abstractmethod

reset()

Resets the agent's internal state, typically called at the beginning of a new episode.

This function should clear any stored history or state information necessary for a fresh interaction.

Returns:

Type	Description
	None

Source code in rllm/agents/agent.py

@abstractmethod
def reset(self):
    """
    Resets the agent's internal state, typically called at the beginning of a new episode.

    This function should clear any stored history or state information necessary
    for a fresh interaction.

    Returns:
        None
    """
    return

get_current_state

get_current_state() -> Step | None

Returns the agent's current state as a dictionary.

This method provides access to the agent's internal state at the current step, which can be useful for debugging, logging, or state management.

Returns:

Name	Type	Description
Step	Step | None	The agent's current state.

Source code in rllm/agents/agent.py

def get_current_state(self) -> Step | None:
    """
    Returns the agent's current state as a dictionary.

    This method provides access to the agent's internal state at the current step,
    which can be useful for debugging, logging, or state management.

    Returns:
        Step: The agent's current state.
    """
    if not self.trajectory.steps:
        return None
    return self.trajectory.steps[-1]

---

Action

rllm.agents.agent.Action dataclass

Source code in rllm/agents/agent.py

@dataclass
class Action:
    action: Any = None