(
self,
function_result: Optional[FunctionResult] = None
)
| 138 | ) |
| 139 | |
| 140 | def _get_action( |
| 141 | self, |
| 142 | function_result: Optional[FunctionResult] = None |
| 143 | ) -> ActionResponse: |
| 144 | |
| 145 | # dummy function result if None is provided - for get_state_fn to take the same input all the time |
| 146 | if function_result is None: |
| 147 | function_result = FunctionResult( |
| 148 | action_id="", |
| 149 | action_status=FunctionResultStatus.DONE, |
| 150 | feedback_message="", |
| 151 | info={}, |
| 152 | ) |
| 153 | |
| 154 | # set up payload |
| 155 | data = { |
| 156 | "location": self.current_worker_id, |
| 157 | "map_id": self._map_id, |
| 158 | "environment": self.worker_states[self.current_worker_id], |
| 159 | "functions": [ |
| 160 | f.get_function_def() |
| 161 | for f in self.workers[self.current_worker_id].action_space.values() |
| 162 | ], |
| 163 | "events": {}, |
| 164 | "agent_state": self.agent_state, |
| 165 | "current_action": ( |
| 166 | function_result.model_dump( |
| 167 | exclude={'info'}) if function_result else None |
| 168 | ), |
| 169 | "version": "v2", |
| 170 | } |
| 171 | |
| 172 | # make API call |
| 173 | response = post( |
| 174 | base_url=self._base_url, |
| 175 | api_key=self._api_key, |
| 176 | endpoint=f"/v2/agents/{self.agent_id}/actions", |
| 177 | data=data, |
| 178 | ) |
| 179 | |
| 180 | return ActionResponse.model_validate(response) |
| 181 | |
| 182 | def step(self): |
| 183 |
no test coverage detected