To be overriden by the descendant class
(self, input:Dict[str,str], spatial_info:Dict[str,Dict], temporal_info:Dict[str,Dict],**kwargs)
| 116 | return {"system_prompt": system_prompt, "user_prompt": user_prompt} |
| 117 | |
| 118 | def _execute(self, input:Dict[str,str], spatial_info:Dict[str,Dict], temporal_info:Dict[str,Dict],**kwargs): |
| 119 | """ To be overriden by the descendant class """ |
| 120 | """ Use the processed input to get the result """ |
| 121 | |
| 122 | inputs = asyncio.run( |
| 123 | self._process_inputs( |
| 124 | input, |
| 125 | spatial_info, |
| 126 | temporal_info, |
| 127 | mode="default", |
| 128 | **kwargs, |
| 129 | ) |
| 130 | ) |
| 131 | system_prompt = inputs["system_prompt"] |
| 132 | user_prompt = inputs["user_prompt"] |
| 133 | message = [{'role':'system','content':system_prompt},{'role':'user','content':user_prompt}] |
| 134 | response = self.llm.gen(message) |
| 135 | return response |
| 136 | |
| 137 | async def _async_execute(self, input:Dict[str,str], spatial_info:Dict[str,Dict], temporal_info:Dict[str,Dict], mode: str = "default", **kwargs): |
| 138 | """Handle asynchronous execution across different KV cache modes.""" |
nothing calls this directly
no test coverage detected