r"""Generate subtasks based on the input task prompt. Args: task_prompt (Union[str, TextPrompt]): The prompt for the task to be divided into subtasks. Returns: TextPrompt: A prompt for the subtasks generated by the agent.
(
self,
task_prompt: Union[str, TextPrompt],
)
| 166 | ) |
| 167 | |
| 168 | def run( |
| 169 | self, |
| 170 | task_prompt: Union[str, TextPrompt], |
| 171 | ) -> TextPrompt: |
| 172 | r"""Generate subtasks based on the input task prompt. |
| 173 | |
| 174 | Args: |
| 175 | task_prompt (Union[str, TextPrompt]): The prompt for the task to |
| 176 | be divided into subtasks. |
| 177 | |
| 178 | Returns: |
| 179 | TextPrompt: A prompt for the subtasks generated by the agent. |
| 180 | """ |
| 181 | # TODO: Maybe include roles information. |
| 182 | self.reset() |
| 183 | task_planner_prompt = self.task_planner_prompt.format(task=task_prompt) |
| 184 | |
| 185 | task_msg = BaseMessage.make_user_message( |
| 186 | role_name="Task Planner", content=task_planner_prompt |
| 187 | ) |
| 188 | |
| 189 | task_response = self.step(task_msg) |
| 190 | |
| 191 | if task_response.terminated: |
| 192 | raise RuntimeError("Task planning failed.") |
| 193 | if len(task_response.msgs) == 0: |
| 194 | raise RuntimeError("Got no task planning message.") |
| 195 | |
| 196 | sub_tasks_msg = task_response.msgs[0] |
| 197 | return TextPrompt(sub_tasks_msg.content) |
| 198 | |
| 199 | |
| 200 | @track_agent(name="TaskCreationAgent") |
no test coverage detected