A simple, lightweight loop for your agent.
Project description
agentloop
A simple, lightweight loop for your agent. Start/stop, step-through, and more.
Quickstart
from agentloop import start, stop
def step_one(next_output, loop_data):
print("step_one")
return next_output
def step_two(next_output, loop_data):
print("step_two")
return next_output
# Run the loop
loop_data = start(steps=[step_one, step_two])
# Stop the loop
stop(loop_data)
Installation
pip install agentloop
Usage
This package provides a set of functions to perform a stepped or continuous loop of operations in a separate thread. This can be used for various purposes like running a continuous process that can be controlled from the outside, a debugging tool for a cycle of operations, etc.
Steps
Each step must take in input from the last step and return output for the next step. The first step will receive None as input, and this will need to be handled. You can either start with an initialization step that returns the initial input, or you can check for None in the first step and return the initial input if it is None.
Example steps:
def step_one(next_output, loop_data): # next output None first run, then received from step_two
print("step_one")
return next_output # next output sent to step_two
def step_two(next_output, loop_data): # next output received from step_one
print("step_two")
return next_output # next output sent to step_one
steps = [step_one, step_two]
Function start
start(steps, stepped=False)
Description
Starts the main loop in a separate thread. This loop will run the steps given, in a continuous or stepped manner.
Parameters
-
steps
: a list of functions that should be executed in the loop. Each function should accept a single argument and return a single value which will be passed as an argument to the next function. The first function will receiveNone
as an argument. -
stepped
(optional): a boolean value that determines whether the loop should run in stepped mode or not. Defaults toFalse
.
Returns
A dictionary containing
stop_event
: an instance ofthreading.Event
that's used to control stopping of the loop.step_event
: an instance ofthreading.Event
that's used to control stepping.started_event
: an instance ofthreading.Event
that's set when the loop starts running.thread
: an instance ofthreading.Thread
that's running the main loop.
Function stop
stop(loop_data)
Description
Handles stopping of the loop.
Parameters
loop_data
: a dictionary containing thestop_event
andthread
which is returned by thestart
function.
Returns
None
Function step
step(loop_data)
Description
Performs a single step in the loop.
Parameters
loop_data
: a dictionary containing thestep_event
which is returned by thestart
function.
Returns
None
Function loop
loop(steps, stepped=False, loop_data=None)
Description
Runs the step array in a loop until stopped.
Parameters
-
steps
: a list of functions that should be executed in the loop. Each function should accept a single argument and return a single value which will be passed as an argument to the next function. The first function will receiveNone
as an argument. -
stepped
(optional): a boolean value that determines whether the loop should run in stepped mode or not. Defaults toFalse
. -
loop_data
(optional): a dictionary containingstop_event
andstep_event
instances. If not provided, new events will be created.
Returns
None
Function use_keyboard
use_keyboard(loop_data, input_key=keyboard.Key.space)
Description
Creates a keyboard listener and attaches it to the provided loop data object. This listener listens for a specified key press, and when detected, steps the loop (sets the 'step_event').
Parameters
-
loop_data
: A loop data object, which is typically created by thestart
function. -
input_key
(optional): The keyboard key that the listener will react to. Defaults tokeyboard.Key.space
.
Returns
Returns the updated loop dictionary with the newly created keyboard listener added to it. Note: Pass the updated dictionary to the stop function to also stop the keyboard listener
Defaults to Spacebar
Example
loop_data = {
"stop_event": threading.Event(),
"step_event": threading.Event(),
"started_event": threading.Event(),
"thread": None,
}
updated_loop_dict = use_keyboard(loop_data, input_key)
Testing
pytest test.py
Publishing
bash publish.sh --version=<version> --username=<pypi_username> --password=<pypi_password>
Contributions Welcome
If you like this library and want to contribute in any way, please feel free to submit a PR and I will review it. Please note that the goal here is simplicity and accesibility, using common language and few dependencies.
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for agentloop-0.1.4-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | e39970a140d5797aead912d15e8d938dcfbb8197e210a47230dde4d0ccadc40b |
|
MD5 | 0c1f8b05f9b622748e3384e6e4497b03 |
|
BLAKE2b-256 | d90ff876e3531e8576a544498b3c7fc260093dffc018db9cb33c96cffce1e5a4 |