File size: 13,707 Bytes
b52b47e 6b17c8c b52b47e 30754bf b52b47e 30754bf f56bea9 30754bf e511f35 30754bf e511f35 30754bf e511f35 30754bf e511f35 30754bf e511f35 30754bf e511f35 30754bf e511f35 30754bf e511f35 30754bf f56bea9 30754bf b52b47e 30754bf b52b47e 30754bf b52b47e e511f35 901ba01 e511f35 901ba01 e511f35 901ba01 e511f35 901ba01 e511f35 901ba01 e511f35 901ba01 e511f35 901ba01 e511f35 901ba01 e511f35 901ba01 e511f35 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 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 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 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 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 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 |
import sys
from typing import Dict, Any
from aiflows.utils import logging
logging.set_verbosity_debug()
log = logging.get_logger(__name__)
from aiflows.interfaces import KeyInterface
from flow_modules.aiflows.ControllerExecutorFlowModule import ControllerExecutorFlow
class AutoGPTFlow(ControllerExecutorFlow):
""" This class implements a (very basic) AutoGPT flow. It is a flow that consists of multiple sub-flows that are executed circularly. It Contains the following subflows:
- A Controller Flow: A Flow that controls which subflow of the Executor Flow to execute next.
- A Memory Flow: A Flow used to save and retrieve messages or memories which might be useful for the Controller Flow.
- A HumanFeedback Flow: A flow use to get feedback from the user/human.
- A Executor Flow: A Flow that executes commands generated by the Controller Flow. Typically it's a branching flow (see BranchingFlow) and the commands are which branch to execute next.
An illustration of the flow is as follows:
| -------> Memory Flow -------> Controller Flow ------->|
^ |
| |
| v
| <----- HumanFeedback Flow <------- Executor Flow <----|
*Configuration Parameters*:
- `name` (str): The name of the flow. Default is "AutoGPTFlow".
- `description` (str): A description of the flow. Default is "An example implementation of AutoGPT with Flows."
- `max_rounds` (int): The maximum number of rounds the circular flow can run for. Default is 30.
- `early_exit_key` (str): The key that is used to terminate the flow early. Default is "EARLY_EXIT".
- `subflows_config` (Dict[str,Any]): A dictionary of subflows configurations. Default:
- `Controller` (Dict[str,Any]): The configuration of the Controller Flow. By default the controller flow is a ControllerAtomicFlow (see ControllerExecutorFlowModule). It's default values are
defined in ControllerAtomicFlow.yaml of the ControllerExecutorFlowModule. Except for the following parameters who are overwritten by the AutoGPTFlow in AutoGPTFlow.yaml:
- `finish` (Dict[str,Any]): The configuration of the finish command (used to terminate the flow early when the controller has accomplished its goal).
- `description` (str): The description of the command. Default is "The finish command is used to terminate the flow early when the controller has accomplished its goal."
- `input_args` (List[str]): The list of expected keys to run the finish command. Default is ["answer"].
- `human_message_prompt_template`(Dict[str,Any]): The prompt template used to generate the message that is shown to the user/human when the finish command is executed. Default is:
- `template` (str): The template of the humand message prompt (see AutoGPTFlow.yaml for default template)
- `input_variables` (List[str]): The list of variables to be included in the template. Default is ["observation", "human_feedback", "memory"].
- `ìnput_interface_initialized` (List[str]): The input interface that Controller Flow expects except for the first time in the flow. Default is ["observation", "human_feedback", "memory"].
- `Executor` (Dict[str,Any]): The configuration of the Executor Flow. By default the executor flow is a Branching Flow (see BranchingFlow). It's default values are the default values of the BranchingFlow. Fields to define:
- `subflows_config` (Dict[str,Any]): A Dictionary of subflows configurations.The keys are the names of the subflows and the values are the configurations of the subflows. Each subflow is a branch of the branching flow.
- `HumanFeedback` (Dict[str,Any]): The configuration of the HumanFeedback Flow. By default the human feedback flow is a HumanStandardInputFlow (see HumanStandardInputFlowModule ).
It's default values are specified in the REAMDE.md of HumanStandardInputFlowModule. Except for the following parameters who are overwritten by the AutoGPTFlow in AutoGPTFlow.yaml:
- `request_multi_line_input_flag` (bool): Flag to request multi-line input. Default is False.
- `query_message_prompt_template` (Dict[str,Any]): The prompt template presented to the user/human to request input. Default is:
- `template` (str): The template of the query message prompt (see AutoGPTFlow.yaml for default template)
- `input_variables` (List[str]): The list of variables to be included in the template. Default is ["goal","command","command_args",observation"]
- input_interface_initialized (List[str]): The input interface that HumanFeeback Flow expects except for the first time in the flow. Default is ["goal","command","command_args",observation"]
- `Memory` (Dict[str,Any]): The configuration of the Memory Flow. By default the memory flow is a ChromaDBFlow (see VectorStoreFlowModule). It's default values are defined in ChromaDBFlow.yaml of the VectorStoreFlowModule. Except for the following parameters who are overwritten by the AutoGPTFlow in AutoGPTFlow.yaml:
- `n_results`: The number of results to retrieve from the memory. Default is 2.
- `topology` (List[Dict[str,Any]]): The topology of the flow which is "circular". By default, the topology is the one shown in the illustration above (the topology is also described in AutoGPTFlow.yaml).
*Input Interface*:
- `goal` (str): The goal of the flow.
*Output Interface*:
- `answer` (str): The answer of the flow.
- `status` (str): The status of the flow. It can be "finished" or "unfinished".
:param flow_config: The configuration of the flow. Contains the parameters described above and the parameters required by the parent class (CircularFlow).
:type flow_config: Dict[str,Any]
:param subflows: A list of subflows constituating the circular flow. Required when instantiating the subflow programmatically (it replaces subflows_config from flow_config).
:type subflows: List[Flow]
"""
def __init__(self, **kwargs):
super().__init__( **kwargs)
self.rename_human_output_interface = KeyInterface(
keys_to_rename={"human_input": "human_feedback"}
)
self.input_interface_controller = KeyInterface(
keys_to_select = ["goal","observation","human_feedback", "memory"],
)
self.input_interface_human_feedback = KeyInterface(
keys_to_select = ["goal","command","command_args","observation"],
)
self.memory_read_ouput_interface = KeyInterface(
additional_transformations = [self.prepare_memory_read_output],
keys_to_select = ["memory"],
)
self.human_feedback_ouput_interface = KeyInterface(
keys_to_rename={"human_input": "human_feedback"},
keys_to_select = ["human_feedback"],
)
self.next_flow_to_call = {
None: "MemoryRead",
"MemoryRead": "Controller",
"Controller": "Executor",
"Executor": "HumanFeedback",
"HumanFeedback": "MemoryWrite",
"MemoryWrite": "MemoryRead",
}
def set_up_flow_state(self):
super().set_up_flow_state()
self.flow_state["early_exit_flag"] = False
def prepare_memory_read_output(self,data_dict: Dict[str, Any],**kwargs):
retrieved_memories = data_dict["retrieved"][0][1:]
return {"memory": "\n".join(retrieved_memories)}
def _get_memory_key(self):
""" This method returns the memory key that is used to retrieve memories from the ChromaDB model.
:param flow_state: The state of the flow
:type flow_state: Dict[str, Any]
:return: The current context
:rtype: str
"""
goal = self.flow_state.get("goal")
last_command = self.flow_state.get("command")
last_command_args = self.flow_state.get("command_args")
last_observation = self.flow_state.get("observation")
last_human_feedback = self.flow_state.get("human_feedback")
if last_command is None:
return ""
assert goal is not None, goal
assert last_command_args is not None, last_command_args
assert last_observation is not None, last_observation
current_context = \
f"""
== Goal ==
{goal}
== Command ==
{last_command}
== Args
{last_command_args}
== Result
{last_observation}
== Human Feedback ==
{last_human_feedback}
"""
return current_context
def prepare_memory_read_input(self) -> Dict[str, Any]:
query = self._get_memory_key()
return {
"operation": "read",
"content": query
}
def prepare_memory_write_input(self) -> Dict[str, Any]:
query = self._get_memory_key()
return {
"operation": "write",
"content": str(query)
}
def call_memory_read(self):
memory_read_input = self.prepare_memory_read_input()
message = self.package_input_message(
data = memory_read_input,
dst_flow = "Memory",
)
self.subflows["Memory"].get_reply(
message,
self.get_instance_id(),
)
def call_memory_write(self):
memory_write_input = self.prepare_memory_write_input()
message = self.package_input_message(
data = memory_write_input,
dst_flow = "Memory",
)
self.subflows["Memory"].get_reply(
message,
self.get_instance_id(),
)
def call_human_feedback(self):
message = self.package_input_message(
data = self.input_interface_human_feedback(self.flow_state),
dst_flow = "HumanFeedback",
)
self.subflows["HumanFeedback"].get_reply(
message,
self.get_instance_id(),
)
def register_data_to_state(self, input_message):
#Making this explicit so it's easier to understand
#I'm also showing different ways of writing to the state
# either programmatically or using the _state_update_dict and
# input and ouput interface methods
last_called = self.flow_state["last_called"]
if last_called is None:
self.flow_state["input_message"] = input_message
self.flow_state["goal"] = input_message.data["goal"]
elif last_called == "Executor":
self.flow_state["observation"] = input_message.data
elif last_called == "Controller":
self._state_update_dict(
{
"command": input_message.data["command"],
"command_args": input_message.data["command_args"]
}
)
#detect and early exit
if self.flow_state["command"] == "finish":
self._state_update_dict(
{
"EARLY_EXIT": True,
"answer": self.flow_state["command_args"]["answer"],
"status": "finished"
}
)
self.flow_state["early_exit_flag"] = True
elif last_called == "MemoryRead":
self._state_update_dict(
self.memory_read_ouput_interface(input_message).data
)
elif last_called == "HumanFeedback":
self._state_update_dict(
self.human_feedback_ouput_interface(input_message).data
)
#detect early exit
if self.flow_state["human_feedback"].strip().lower() == "q":
self._state_update_dict(
{
"EARLY_EXIT": True,
"answer": "The user has chosen to exit before a final answer was generated.",
"status": "unfinished",
}
)
self.flow_state["early_exit_flag"] = True
def run(self,input_message):
self.register_data_to_state(input_message)
flow_to_call = self.get_next_flow_to_call()
if self.flow_state.get("early_exit_flag",False):
self.generate_reply()
elif flow_to_call == "MemoryRead":
self.call_memory_read()
elif flow_to_call == "Controller":
self.call_controller()
elif flow_to_call == "Executor":
self.call_executor()
elif flow_to_call == "HumanFeedback":
self.call_human_feedback()
elif flow_to_call == "MemoryWrite":
self.call_memory_write()
self.flow_state["current_round"] += 1
else:
self._on_reach_max_round()
self.generate_reply()
self.flow_state["last_called"] = self.get_next_flow_to_call()
|