danoan.llm_assistant.runner.cli.commands.session.task_runner module

class danoan.llm_assistant.runner.cli.commands.session.task_runner.EventNotifier(event_name: str, register_fn: Callable[[str, Callable[[...], Any]], Any], unregister_fn: Callable[[str, Callable[[...], Any]], Any])

Bases: object

Helper class that notifies caller when an event is complete.

The EventNotifier is usually returned by an external class (let us say Manager) that accepts registering listeners to its events. A common use is to create a context and check the value of the notifier.

with Manager.get_notifier(event_name) as notifier:

while True:

if notifier.is_set():

break

do_something()

Parameters:

• event_name (str)

• register_fn (Callable[[str, Callable[[...], Any]], Any])

• unregister_fn (Callable[[str, Callable[[...], Any]], Any])

class danoan.llm_assistant.runner.cli.commands.session.task_runner.Task

Bases: ABC

Represent a task of TaskRunner.

class Event(*values)

Bases: Enum

Complete = 'complete'

Stop = 'stop'

get_event_notifier(event: Event) → EventNotifier

Create EventNotifier for an event.

The EventNotifier allows us to check if events such as Task.Event.Complete or Task.Event.Stop were triggered.

Parameters:

event (Event)

Return type:

EventNotifier

register_listener(event: Event, callback)

Register a callback function to be called whenever the event is triggered.

Parameters:

event (Event)

abstractmethod run(**kwargs)

The task logic.

start(**kwargs)

Start executing the task logic.

It triggers all listeners of Task.Event.Complete whenever is finished.

stop()

Trigger all listeners of event Task.Event.Stop.

The logic contained in run is not aborted. It is up to the user to take this decision.

# Run logic with self.get_event_notifier(Task.Event.Stop) as notifier:

if notifier.is_set():

take action

unregister_listener(event: Event, callback)

Unregister a previously registered listener.

Raises:

TaskListenerNotRegisteredError if the listener to unregister is not found. –

Parameters:

event (Event)

class danoan.llm_assistant.runner.cli.commands.session.task_runner.TaskInstruction(task_name: str, task_input: Dict[str, Any] | None)

Bases: object

Parameters:

• task_name (str)

• task_input (Dict[str, Any] | None)

task_input: Dict[str, Any] | None

task_name: str

exception danoan.llm_assistant.runner.cli.commands.session.task_runner.TaskListenerNotRegisteredError

Bases: Exception

exception danoan.llm_assistant.runner.cli.commands.session.task_runner.TaskNotRegisteredError

Bases: Exception

class danoan.llm_assistant.runner.cli.commands.session.task_runner.TaskRunner

Bases: object

Execute TaskInstruction added to its queue until emptness.

TaskRunner executes tasks in the same order they are added in its internal queue via a TaskInstruction. The TaskInstruction contains the task name and its input variables. To execute a task, the latter needs to be priorly registered.

The Task output is an optional TaskInstruction and it is added to the TaskRunner queue as soon it is completed. The added TaskInstruction is then executed in the next turn.

The queue object is adapted to concurrent thread execution.

QuitInstruction = TaskInstruction(task_name='quit', task_input=None)

add(td: TaskInstruction)

Add a new TaskInstruction to the queue.

Parameters:

td (TaskInstruction)

clear()

Clear task queue.

next()

Stop the current task.

register(task_name: str, first_task: bool = False, callback=None)

Decorator to register a function as a task.

Parameters:

• task_name (str)

• first_task (bool)

run()

Traverse the queue executing all tasks there present.

stop()

Stop all tasks.