Agents#

Hierarchical agent classes for multi-level control systems.

Agent Base Class#

class heron.agents.base.Agent(agent_id: str, level: int = 1)#

Base class for all HERON agents.

Parameters:

agent_id – Unique identifier for the agent
level – Hierarchy level (1 = leaf, higher = closer to root)

agent_id: str#: Unique identifier for the agent.

level: int#: Position in hierarchy (1 = field agent, 2 = coordinator, etc.)

abstractmethod observe() → Observation#

Generate observation from current state.

Returns:: Observation object with local and optional global data

abstractmethod act(observation: Observation, **kwargs) → Action#

Process observation and return action.

Parameters:: observation – Current observation
Returns:: Action to execute

abstractmethod reset(seed: int = None)#

Reset agent to initial state.

Parameters:: seed – Random seed for reproducibility

Field Agent#

class heron.agents.FieldAgent(agent_id: str, **kwargs)#

Leaf-level agent for local sensing and actuation.

Inherits from Agent with level=1.

Usage:

from heron.agents import FieldAgent

class MySensor(FieldAgent):
    def __init__(self, agent_id: str):
        super().__init__(agent_id)
        self.state = FieldAgentState()

    def observe(self) -> Observation:
        return Observation(local={"state": self.state.vector()})

    def act(self, observation: Observation, **kwargs) -> Action:
        # Local control logic
        return self.action

Coordinator Agent#

class heron.agents.CoordinatorAgent(agent_id: str, subordinates: dict = None, **kwargs)#

Mid-level agent that manages subordinate agents.

Parameters:: subordinates – Dict mapping IDs to subordinate agents

subordinates: dict[str, Agent]#: Dictionary of subordinate agents.

aggregate_observations() → dict#: Collect observations from all subordinates.

Usage:

from heron.agents import CoordinatorAgent, FieldAgent

field1 = MySensor("sensor_1")
field2 = MySensor("sensor_2")

coordinator = CoordinatorAgent(
    agent_id="zone_controller",
    subordinates={"sensor_1": field1, "sensor_2": field2}
)

System Agent#

class heron.agents.SystemAgent(agent_id: str, subordinates: dict = None, **kwargs)#

Top-level agent for global coordination.

Inherits from CoordinatorAgent with highest hierarchy level.

Usage:

from heron.agents import SystemAgent

system = SystemAgent(
    agent_id="system_operator",
    subordinates={"zone_1": coordinator1, "zone_2": coordinator2}
)

Proxy Agent#

class heron.agents.Proxy(agent_id: str, actual_agent: Agent, broker: MessageBroker, upstream_id: str = None)#

Wrapper for message-based distributed execution.

Parameters:

actual_agent – The agent to wrap
broker – Message broker for communication
upstream_id – ID of upstream coordinator (if any)

async step_distributed()#: Execute distributed step with message passing.

Usage:

from heron.agents import Proxy
from heron.messaging.memory import InMemoryBroker

broker = InMemoryBroker()
proxy = Proxy(
    agent_id="sensor_1_proxy",
    actual_agent=sensor,
    broker=broker,
    upstream_id="zone_controller"
)

await proxy.step_distributed()