Skip to content

academy.behavior

Behavior

Agent behavior base class.

All Agent instances execute a behavior which is defined by a subclass of the Behavior. Each behavior is composed of three parts: 1. The on_startup() and on_shutdown() methods define callbacks that are invoked once at the start and end of an agent's execution, respectively. The methods should be used to initialize and cleanup stateful resources. Resource initialization should not be performed in __init__. 2. Action methods annotated with @action are methods that other agents can invoke on this agent. An agent may also call it's own action methods as normal methods. 3. Control loop methods annotated with @loop are executed in separate threads when the agent is executed.

Warning

This class cannot be instantiated directly and must be subclassed.

behavior_actions 

behavior_actions() -> dict[str, Action[Any, Any]]

Get methods of this behavior type that are decorated as actions.

Returns:

Source code in academy/behavior.py 
77
78
79
80
81
82
83
84
85
86
87
88
def behavior_actions(self) -> dict[str, Action[Any, Any]]:
    """Get methods of this behavior type that are decorated as actions.

    Returns:
        Dictionary mapping method names to action methods.
    """
    actions: dict[str, Action[Any, Any]] = {}
    for name in dir(self):
        attr = getattr(self, name)
        if _is_agent_method_type(attr, 'action'):
            actions[name] = attr
    return actions

behavior_loops 

behavior_loops() -> dict[str, ControlLoop]

Get methods of this behavior type that are decorated as loops.

Returns:

Source code in academy/behavior.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def behavior_loops(self) -> dict[str, ControlLoop]:
    """Get methods of this behavior type that are decorated as loops.

    Returns:
        Dictionary mapping method names to loop methods.
    """
    loops: dict[str, ControlLoop] = {}
    for name in dir(self):
        attr = getattr(self, name)
        if _is_agent_method_type(attr, 'loop'):
            loops[name] = attr
    return loops

behavior_handles 

behavior_handles() -> dict[
    str,
    Handle[Any] | HandleDict[Any, Any] | HandleList[Any],
]

Get instance attributes that are agent handles.

Returns:

Source code in academy/behavior.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def behavior_handles(
    self,
) -> dict[
    str,
    Handle[Any] | HandleDict[Any, Any] | HandleList[Any],
]:
    """Get instance attributes that are agent handles.

    Returns:
        Dictionary mapping attribute names to agent handles or \
        data structures of handles.
    """
    from academy.handle import Handle

    # This import is deferred to prevent a cyclic import with
    # academy.handle.
    handles: dict[
        str,
        Handle[Any] | HandleDict[Any, Any] | HandleList[Any],
    ] = {}
    for name in dir(self):
        attr = getattr(self, name)
        if isinstance(attr, (Handle, HandleDict, HandleList)):
            handles[name] = attr
    return handles

behavior_handles_bind 

behavior_handles_bind(
    bind: Callable[[Handle[BehaviorT]], Handle[BehaviorT]],
) -> None

Bind all instance attributes that are agent handles.

Parameters:

  • bind (Callable[[Handle[BehaviorT]], Handle[BehaviorT]]) –

    A callback that takes a handle and returns the same handle or a bound version of the handle.

Source code in academy/behavior.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def behavior_handles_bind(
    self,
    bind: Callable[[Handle[BehaviorT]], Handle[BehaviorT]],
) -> None:
    """Bind all instance attributes that are agent handles.

    Args:
        bind: A callback that takes a handle and returns the same handle
            or a bound version of the handle.
    """
    for attr, handles in self.behavior_handles().items():
        if isinstance(handles, Handle):
            setattr(self, attr, bind(handles))
        elif isinstance(handles, HandleDict):
            setattr(
                self,
                attr,
                HandleDict({k: bind(h) for k, h in handles.items()}),
            )
        elif isinstance(handles, HandleList):
            setattr(self, attr, HandleList(bind(h) for h in handles))
        else:
            raise AssertionError('Unreachable.')

behavior_mro classmethod 

behavior_mro() -> tuple[str, ...]

Get the method resolution order of the behavior.

Example 
>>> from academy.behavior import Behavior
>>>
>>> class A(Behavior): ...
>>> class B(Behavior): ...
>>> class C(A): ...
>>> class D(A, B): ...
>>>
>>> A.behavior_mro()
('__main__.A',)
>>> B.behavior_mro()
('__main__.B',)
>>> C.behavior_mro()
('__main__.C', '__main__.A')
>>> D.behavior_mro()
('__main__.D', '__main__.A', '__main__.B')

Returns:

  • tuple[str, ...] –

    Tuple of fully-qualified paths of types in the MRO of this behavior type, not including the base Behavior or object.

Source code in academy/behavior.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
@classmethod
def behavior_mro(cls) -> tuple[str, ...]:
    """Get the method resolution order of the behavior.

    Example:
        ```python
        >>> from academy.behavior import Behavior
        >>>
        >>> class A(Behavior): ...
        >>> class B(Behavior): ...
        >>> class C(A): ...
        >>> class D(A, B): ...
        >>>
        >>> A.behavior_mro()
        ('__main__.A',)
        >>> B.behavior_mro()
        ('__main__.B',)
        >>> C.behavior_mro()
        ('__main__.C', '__main__.A')
        >>> D.behavior_mro()
        ('__main__.D', '__main__.A', '__main__.B')
        ```

    Returns:
        Tuple of fully-qualified paths of types in the MRO of this \
        behavior type, not including the base \
        [`Behavior`][academy.behavior.Behavior] or [`object`][object].
    """
    mro = cls.mro()
    base_index = mro.index(Behavior)
    mro = mro[:base_index]
    return tuple(f'{t.__module__}.{t.__qualname__}' for t in mro)

on_setup 

on_setup() -> None

Setup up resources needed for the agents execution.

This is called before any control loop threads are started.

Source code in academy/behavior.py 
186
187
188
189
190
191
def on_setup(self) -> None:
    """Setup up resources needed for the agents execution.

    This is called before any control loop threads are started.
    """
    pass

on_shutdown 

on_shutdown() -> None

Shutdown resources after the agents execution.

This is called after control loop threads have exited.

Source code in academy/behavior.py 
193
194
195
196
197
198
def on_shutdown(self) -> None:
    """Shutdown resources after the agents execution.

    This is called after control loop threads have exited.
    """
    pass

Action

Bases: Generic[P, R_co], Protocol

Action method protocol.

__call__ 

__call__(*arg: args, **kwargs: kwargs) -> R_co

Expected signature of methods decorated as an action.

In general, action methods can implement any signature.

Source code in academy/behavior.py 
206
207
208
209
210
211
def __call__(self, *arg: P.args, **kwargs: P.kwargs) -> R_co:
    """Expected signature of methods decorated as an action.

    In general, action methods can implement any signature.
    """
    ...

ControlLoop

Bases: Protocol

Control loop method protocol.

__call__ 

__call__(shutdown: Event) -> None

Expected signature of methods decorated as a control loop.

Parameters:

  • shutdown (Event) –

    Event indicating that the agent has been instructed to shutdown and all control loops should exit.

Returns:

  • None –

    Control loops should not return anything.

Source code in academy/behavior.py 
219
220
221
222
223
224
225
226
227
228
229
def __call__(self, shutdown: threading.Event) -> None:
    """Expected signature of methods decorated as a control loop.

    Args:
        shutdown: Event indicating that the agent has been instructed to
            shutdown and all control loops should exit.

    Returns:
        Control loops should not return anything.
    """
    ...

action 

action(method: Callable[P, R]) -> Callable[P, R]

Decorator that annotates a method of a behavior as an action.

Marking a method of a behavior as an action makes the method available to other agents. I.e., peers within a multi-agent system can only invoke methods marked as actions on each other. This enables behaviors to define "private" methods.

Example 
from academy.behavior import Behavior, action

class Example(Behavior):
    @action
    def perform(self):
        ...
Source code in academy/behavior.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def action(method: Callable[P, R]) -> Callable[P, R]:
    """Decorator that annotates a method of a behavior as an action.

    Marking a method of a behavior as an action makes the method available
    to other agents. I.e., peers within a multi-agent system can only invoke
    methods marked as actions on each other. This enables behaviors to
    define "private" methods.

    Example:
        ```python
        from academy.behavior import Behavior, action

        class Example(Behavior):
            @action
            def perform(self):
                ...
        ```
    """
    method._agent_method_type = 'action'  # type: ignore[attr-defined]
    return method

loop 

loop(
    method: Callable[Concatenate[BehaviorT, P], R],
) -> Callable[Concatenate[BehaviorT, P], R]

Decorator that annotates a method of a behavior as a control loop.

Control loop methods of a behavior are run as threads when an agent starts. A control loop can run for a well-defined period of time or indefinitely, provided the control loop exits when the shutdown event, passed as a parameter to all control loop methods, is set.

Example 
import threading
from academy.behavior import Behavior, loop

class Example(Behavior):
    @loop
    def listen(self, shutdown: threading.Event) -> None:
        while not shutdown.is_set():
            ...

Raises:

Source code in academy/behavior.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
def loop(
    method: Callable[Concatenate[BehaviorT, P], R],
) -> Callable[Concatenate[BehaviorT, P], R]:
    """Decorator that annotates a method of a behavior as a control loop.

    Control loop methods of a behavior are run as threads when an agent
    starts. A control loop can run for a well-defined period of time or
    indefinitely, provided the control loop exits when the `shutdown`
    event, passed as a parameter to all control loop methods, is set.

    Example:
        ```python
        import threading
        from academy.behavior import Behavior, loop

        class Example(Behavior):
            @loop
            def listen(self, shutdown: threading.Event) -> None:
                while not shutdown.is_set():
                    ...
        ```

    Raises:
        TypeError: if the method signature does not conform to the
            [`ControlLoop`][academy.behavior.ControlLoop] protocol.
    """
    method._agent_method_type = 'loop'  # type: ignore[attr-defined]

    if sys.version_info >= (3, 10):  # pragma: >=3.10 cover
        found_sig = inspect.signature(method, eval_str=True)
        expected_sig = inspect.signature(ControlLoop.__call__, eval_str=True)
    else:  # pragma: <3.10 cover
        found_sig = inspect.signature(method)
        expected_sig = inspect.signature(ControlLoop.__call__)

    if found_sig != expected_sig:
        raise TypeError(
            f'Signature of loop method "{method.__name__}" is {found_sig} '
            f'but should be {expected_sig}. If the signatures look the same '
            'except that types are stringified, try importing '
            '"from __future__ import annotations" at the top of the module '
            'where the behavior is defined.',
        )

    @functools.wraps(method)
    def _wrapped(self: BehaviorT, *args: P.args, **kwargs: P.kwargs) -> R:
        logger.debug('Started %r loop for %s', method.__name__, self)
        result = method(self, *args, **kwargs)
        logger.debug('Exited %r loop for %s', method.__name__, self)
        return result

    return _wrapped

event 

event(
    name: str,
) -> Callable[
    [Callable[[BehaviorT], None]],
    Callable[[BehaviorT, Event], None],
]

Decorator that annotates a method of a behavior as an event loop.

An event loop is a special type of control loop that runs when a threading.Event is set. The event is cleared after the loop runs.

Example 
import threading
from academy.behavior import Behavior, timer

class Example(Behavior):
    def __init__(self) -> None:
        self.alert = threading.Event()

    @event('alert')
    def handle(self) -> None:
        # Runs every time alter is set
        ...

Parameters:

Raises:

Source code in academy/behavior.py 
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
def event(
    name: str,
) -> Callable[
    [Callable[[BehaviorT], None]],
    Callable[[BehaviorT, threading.Event], None],
]:
    """Decorator that annotates a method of a behavior as an event loop.

    An event loop is a special type of control loop that runs when a
    [`threading.Event`][threading.Event] is set. The event is cleared
    after the loop runs.

    Example:
        ```python
        import threading
        from academy.behavior import Behavior, timer

        class Example(Behavior):
            def __init__(self) -> None:
                self.alert = threading.Event()

            @event('alert')
            def handle(self) -> None:
                # Runs every time alter is set
                ...
        ```

    Args:
        name: Attribute name of the [`threading.Event`][threading.Event]
            to wait on.

    Raises:
        AttributeError: Raised at runtime if no attribute named `name`
            exists on the behavior.
        TypeError: Raised at runtime if the attribute named `name` is not
            a [`threading.Event`][threading.Event].
    """

    def decorator(
        method: Callable[[BehaviorT], None],
    ) -> Callable[[BehaviorT, threading.Event], None]:
        method._agent_method_type = 'loop'  # type: ignore[attr-defined]

        @functools.wraps(method)
        def _wrapped(self: BehaviorT, shutdown: threading.Event) -> None:
            event = getattr(self, name)
            if not isinstance(event, threading.Event):
                raise TypeError(
                    f'Attribute {name} of {type(self).__class__} has type '
                    f'{type(event).__class__}. Expected threading.Event.',
                )

            logger.debug(
                'Started %r event loop for %s (event: %r)',
                method.__name__,
                self,
                name,
            )
            combined = or_event(shutdown, event)
            while True:
                combined.wait()
                if shutdown.is_set():
                    break
                elif event.is_set():
                    try:
                        method(self)
                    finally:
                        event.clear()
                else:
                    raise AssertionError('Unreachable.')
            logger.debug('Exited %r event loop for %s', method.__name__, self)

        return _wrapped

    return decorator

timer 

timer(
    interval: float | timedelta,
) -> Callable[
    [Callable[[BehaviorT], None]],
    Callable[[BehaviorT, Event], None],
]

Decorator that annotates a method of a behavior as a timer loop.

A timer loop is a special type of control loop that runs at a set interval. The method will always be called once before the first sleep.

Example 
from academy.behavior import Behavior, timer

class Example(Behavior):
    @timer(interval=1)
    def listen(self) -> None:
        # Runs every 1 second
        ...

Parameters:

Source code in academy/behavior.py 
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
def timer(
    interval: float | timedelta,
) -> Callable[
    [Callable[[BehaviorT], None]],
    Callable[[BehaviorT, threading.Event], None],
]:
    """Decorator that annotates a method of a behavior as a timer loop.

    A timer loop is a special type of control loop that runs at a set
    interval. The method will always be called once before the first
    sleep.

    Example:
        ```python
        from academy.behavior import Behavior, timer

        class Example(Behavior):
            @timer(interval=1)
            def listen(self) -> None:
                # Runs every 1 second
                ...
        ```

    Args:
        interval: Seconds or a [`timedelta`][datetime.timedelta] to wait
            between invoking the method.
    """
    interval = (
        interval.total_seconds()
        if isinstance(interval, timedelta)
        else interval
    )

    def decorator(
        method: Callable[[BehaviorT], None],
    ) -> Callable[[BehaviorT, threading.Event], None]:
        method._agent_method_type = 'loop'  # type: ignore[attr-defined]

        @functools.wraps(method)
        def _wrapped(self: BehaviorT, shutdown: threading.Event) -> None:
            logger.debug(
                'Started %r timer loop for %s (interval: %fs)',
                method.__name__,
                self,
                interval,
            )
            while not shutdown.wait(interval):
                method(self)
            logger.debug('Exited %r timer loop for %s', method.__name__, self)

        return _wrapped

    return decorator