Python API
agents
Redis Agent

RedisAgent Class Documentation

Overview

The RedisAgent class is an implementation of a base agent that interacts with a server using Redis as a message broker. This class extends from BaseAgent, and leverages a FastAPI server to handle messaging tasks.

Dependencies

  • asyncio: Provides support for asynchronous operations.
  • logging: Used for logging messages.
  • os: Used to fetch environment variables.
  • uuid: Used for generating unique UUIDs.
  • aiohttp: Asynchronous HTTP client/server framework.
  • pydantic: Data validation and settings management using Python-type annotations.
  • requests: HTTP library for sending HTTP requests.

Class Definition

class RedisAgent(BaseAgent[Observation, AgentAction]):
    """An agent use redis as a message broker."""

__init__ Method

Initializes the Redis agent.

Parameters

  • agent_name (Optional[str]): Name of the agent.
  • uuid_str (Optional[str]): Unique identifier for the agent.
  • session_id (Optional[str]): Unique identifier for the session, defaults to a newly generated UUID.
  • agent_profile (Optional[AgentProfile]): Profile details of the agent.
def __init__(self, agent_name: str | None = None, uuid_str: str | None = None, session_id: str | None = None, agent_profile: AgentProfile | None = None) -> None:

Details

  • Sets the session ID using the provided session_id or generates a new UUID.
  • Sets the sender ID using a generated UUID.
  • Sets a pre-defined model_name to "redis".
  • Checks for the existence of the FASTAPI_URL environment variable and sets _URL accordingly.
  • Attempts to connect to a server using the generated session and sender IDs.

act Method

To be implemented.

Parameters

  • obs (Observation): The observation data.

Returns

  • AgentAction: An action based on the observation.
def act(self, obs: Observation) -> AgentAction:
    raise NotImplementedError

aact Method

Asynchronous method to act upon receiving an observation.

Parameters

  • obs (Observation): The observation data.

Returns

  • AgentAction: An action based on the observation.
async def aact(self, obs: Observation) -> AgentAction:

Details

  • Receives and processes the observation.
  • Posts observation data to the server.
  • Unlocks the server for the client, waits for client's message, and processes the received message to determine the next action.
  • Validates and parses the received message as an AgentAction.

reset Method

Resets the agent's state.

Parameters

  • reset_reason (str): Reason for resetting the agent.
def reset(self, reset_reason: str = "") -> None:

Details

  • Uses the super().reset() method to reset the base class.
  • Attempts to send a reset reason to the server if provided.

Summary

The RedisAgent class provides an interface for interacting with a server using Redis as a message broker. It contains methods for initializing the agent, handling observations asynchronously, and resetting the agent's state.

Llm AgentLogs