(
self,
api_key: str,
description: str, # description of the worker/character card (PROMPT)
get_state_fn: Callable,
action_space: List[Function],
# specific additional instruction for the worker (PROMPT)
instruction: Optional[str] = "",
)
| 9 | """ |
| 10 | |
| 11 | def __init__( |
| 12 | self, |
| 13 | api_key: str, |
| 14 | description: str, # description of the worker/character card (PROMPT) |
| 15 | get_state_fn: Callable, |
| 16 | action_space: List[Function], |
| 17 | # specific additional instruction for the worker (PROMPT) |
| 18 | instruction: Optional[str] = "", |
| 19 | ): |
| 20 | |
| 21 | self._base_url: str = "https://game.virtuals.io" |
| 22 | self._api_key: str = api_key |
| 23 | |
| 24 | # checks |
| 25 | if not self._api_key: |
| 26 | raise ValueError("API key not set") |
| 27 | |
| 28 | self.description: str = description |
| 29 | self.instruction: str = instruction |
| 30 | |
| 31 | # setup get state function and initial state |
| 32 | self.get_state_fn = lambda function_result, current_state: { |
| 33 | "instructions": self.instruction, # instructions are set up in the state |
| 34 | # places the rest of the output of the get_state_fn in the state |
| 35 | **get_state_fn(function_result, current_state), |
| 36 | } |
| 37 | dummy_function_result = FunctionResult( |
| 38 | action_id="", |
| 39 | action_status=FunctionResultStatus.DONE, |
| 40 | feedback_message="", |
| 41 | info={}, |
| 42 | ) |
| 43 | # get state |
| 44 | self.state = self.get_state_fn(dummy_function_result, None) |
| 45 | |
| 46 | # # setup action space (functions/tools available to the worker) |
| 47 | # check action space type - if not a dict |
| 48 | if not isinstance(action_space, dict): |
| 49 | self.action_space = { |
| 50 | f.get_function_def()["fn_name"]: f for f in action_space} |
| 51 | else: |
| 52 | self.action_space = action_space |
| 53 | |
| 54 | # initialize an agent instance for the worker |
| 55 | self._agent_id: str = create_agent( |
| 56 | self._base_url, self._api_key, "StandaloneWorker", self.description, "N/A" |
| 57 | ) |
| 58 | |
| 59 | # persistent variables that is maintained through the worker running |
| 60 | # task ID for everytime you provide/update the task (i.e. ask the agent to do something) |
| 61 | self._submission_id: Optional[str] = None |
| 62 | # current response from the Agent |
| 63 | self._function_result: Optional[FunctionResult] = None |
| 64 | |
| 65 | def set_task(self, task: str): |
| 66 | """ |
nothing calls this directly
no test coverage detected