To be overriden by the descendant class
(self, input:Dict[str,str], spatial_info:Dict[str,Any], temporal_info:Dict[str,Any], mode: str = "default", **kwargs)
| 140 | return response |
| 141 | |
| 142 | async def _async_execute(self, input:Dict[str,str], spatial_info:Dict[str,Any], temporal_info:Dict[str,Any], mode: str = "default", **kwargs): |
| 143 | """ To be overriden by the descendant class """ |
| 144 | """ Use the processed input to get the result """ |
| 145 | """ The input type of this node is Dict """ |
| 146 | if mode == "default": |
| 147 | request_uid = input.get("_request_uid") |
| 148 | inputs = await self._process_inputs( |
| 149 | input, |
| 150 | spatial_info, |
| 151 | temporal_info, |
| 152 | mode=mode, |
| 153 | **kwargs, |
| 154 | ) |
| 155 | system_prompt = inputs["system_prompt"] |
| 156 | user_prompt = inputs["user_prompt"] |
| 157 | message = [{'role':'system','content':system_prompt},{'role':'user','content':user_prompt}] |
| 158 | result = await self.llm.agen( |
| 159 | message, |
| 160 | request_uid=request_uid, |
| 161 | agent_id=self.id, |
| 162 | agent_name=self.agent_name, |
| 163 | agent_role=self.role, |
| 164 | ) |
| 165 | if self.role == "Programming Expert": |
| 166 | answer = execute_code_get_return(result.text.lstrip("```python\n").rstrip("\n```")) |
| 167 | result.text += f"\nthe answer is {answer}" |
| 168 | return result |
| 169 | |
| 170 | assert mode == "allow_kv_reuse", f"Unsupported async execution mode: {mode}" |
| 171 | request_uid = input.get("_request_uid") or kwargs.get("request_uid") |
| 172 | if request_uid is None: |
| 173 | raise ValueError("request_uid is required for request-scoped anchor updates.") |
| 174 | |
| 175 | mode_data = await self._process_inputs( |
| 176 | input, |
| 177 | spatial_info, |
| 178 | temporal_info, |
| 179 | mode=mode, |
| 180 | **kwargs, |
| 181 | ) |
| 182 | preferred_mode = mode_data["preferred_mode"] |
| 183 | result = await self.llm.generate_for_agent( |
| 184 | request_uid=request_uid, |
| 185 | message=input['task'], |
| 186 | preferred_mode=preferred_mode, |
| 187 | max_tokens=self.llm.DEFAULT_MAX_TOKENS, |
| 188 | output_dir=kwargs.get('output_dir'), |
| 189 | agent_id=self.id, |
| 190 | agent_name=self.agent_name, |
| 191 | agent_role=self.role, |
| 192 | ) |
| 193 | |
| 194 | if self.role == "Programming Expert": |
| 195 | answer = execute_code_get_return(result.text.split("```python\n")[-1].split("\n```")[0]) |
| 196 | result.text += f"\nthe answer is {answer}" |
| 197 | return input['task'], result |
nothing calls this directly
no test coverage detected