Task#
- class src.openCHA.tasks.task.BaseTask(*, name: str = <property object>, chat_name: str, description: str, dependencies: ~typing.List[str] = <property object>, inputs: ~typing.List[str] = <property object>, outputs: ~typing.List[str] = [], datapipe: ~openCHA.datapipes.datapipe.DataPipe = None, output_type: bool = False, return_direct: bool = False)[source]#
Description:
This class is the base implementation for the Tasks. For every new task that you want to create, you should inherit from this class and override the attributes and methods based on your task’s need. This class defines a base class named BaseTask. This class serves as a foundation for defining common properties and behaviors among various tasks in the system.
- name#
The name of the task. It should be unique underscore_case to be defined in TaskType. sample_task_name
- Type:
str
- chat_name#
This is the name that later will be used if needed to mention the tasks inside the chat with the user. It should be Camel Case. SampleTaskChatName
- Type:
str
- description#
The description of the what specifically the task is doing. Try to define it as specific as possible to help the Task Planner decide better.
- Type:
str
- dependencies#
You can put the name of the TaskTypes that this task is dependent on. For example, in stress detection scenario, the stress analysis is dependent on the fetch hrv data task. [TaskType.SERPAPI, TASKTYPE.EXTRACT_TEXT]
- Type:
List[str]
- inputs#
This is the list of descriptions for the inputs that should be provided by the planner. For example if your task has two inputs: [“the first input description”, “the second input description”]
- Type:
List[str]
- outputs#
This is the list of the description of the outputs that the task returns. This helps the planner to understand the returned results better and use it as needed. For example, if the task returns a list of sleep hours for different sleep states, the description helps planner learn which number is related to what state.
- Type:
List[str]
- output_type#
This indicates if the task result should be stored in the DataPipe or be returned directly to the planner. This process will be done in the parse_input and post_execute methods. If needed you can overwrite them.
- Type:
bool
- return_direct#
This indicates if this task should completely interrupt the planning process or not. This is needed in cases like when you want to ask a question from user and no further planning is needed until the user gives the proper answer (look at ask_user task)
- Type:
bool
- abstract _execute(inputs: List[Any]) str[source]#
Abstract method representing the execution of the task. You should implement this method based on your need. This method is called by the execute method that provides the parsed inputs to this method.
- Parameters:
inputs (List[Any]) – Input data for the task.
- Returns:
Result of the task execution.
- Return type:
str
- Raises:
NotImplementedError – Subclasses must implement the execute method.
- _parse_input(input_args: List[str]) List[str][source]#
Parses the input string into a list of strings. If the input is in format datapipe:key, the parser will retrieve the data from datapipe before sending it over to the _execute method.
- Parameters:
List (input_args) – List of Input string provided by planner. It should be parsed and return a list of str variables.
- Returns:
List of parsed strings. These strings can be converted into desired types inside _execute method.
- Return type:
List[str]
- _post_execute(result: str = '') str[source]#
This method is called inside execute method after calling _execute. The result of _execute will be passed to this method in case the output_type attribute is True, the result will be stored inside the datapipe and the datapipe key is returned to the plenner instead of the raw result. This is good practice for times that you have intermediate data (like sleep data over a month) and it needs to be passed over to other tasks and the raw result is not immidiately needed. This will save a huge amount of tokens and makes sure that the planner will not pass wrong raw data to the tasks.
It is important to note that to make the DataPipe’s stored data standard and unified, we store the data in the json string format that currently contains ‘data’ and ‘description’ keys. The ‘data’ will be the returned data after execution and the ‘description’ is created using the outputs attribute of the task. Whenever the raw data is returned to the planner, these outputs descriptions will help the planner understand and learn how to interpret the ‘data’ to generate the final answer or continue planning.
- Parameters:
result (str) – string containig the task result.
- Returns:
List of parsed strings.
- Return type:
List[str]
- _validate_inputs(inputs: List[str]) bool[source]#
This method is called inside execute method after calling _parse_input. The result of _parse_input will be passed to this method to check the validity of the provided inputs by the Task Planner. Currently it only checks the length of the parsed input and the length of the inputs attribute of the tasks class. You can inherit this function to further add more input checkings for your own tasks.
- Parameters:
inputs (List) – List of strings containig the parsed inputs.
- Returns:
True if the inputs are valid, False otherwise
- Return type:
bool
- execute(input_args: List[str]) str[source]#
This method is called by the Orchestrator which provides the planner provided inputs. This method first calls _parse_input to parse the inputs and retrieve needed data from the DataPipe Then _execute is called and the parsed inputs are given to this method. Finally the final result of execution is passed to _post_execute and ith will either be stored inside DataPipe or directly returned to the planner to continue planning.
- Parameters:
input_args (str) – Input string provided by planner.
- Returns:
The final result of the task execution.
- Return type:
str