Starts the agent process for a given task. Args: task (dict): The task to be performed. required_infos (dict | None): The required information for the task. Returns: str: The result of the agent process.
(self, task: dict, required_infos: dict | None)
| 230 | return plan |
| 231 | |
| 232 | async def start_agent_process(self, task: dict, required_infos: dict | None) -> str: |
| 233 | """ |
| 234 | Starts the agent process for a given task. |
| 235 | Args: |
| 236 | task (dict): The task to be performed. |
| 237 | required_infos (dict | None): The required information for the task. |
| 238 | Returns: |
| 239 | str: The result of the agent process. |
| 240 | """ |
| 241 | self.status_message = f"Starting task {task['task']}..." |
| 242 | agent_prompt = self.make_prompt(task['task'], required_infos) |
| 243 | pretty_print(f"Agent {task['agent']} started working...", color="status") |
| 244 | self.logger.info(f"Agent {task['agent']} started working on {task['task']}.") |
| 245 | answer, reasoning = await self.agents[task['agent'].lower()].process(agent_prompt, None) |
| 246 | self.last_answer = answer |
| 247 | self.last_reasoning = reasoning |
| 248 | self.blocks_result = self.agents[task['agent'].lower()].blocks_result |
| 249 | agent_answer = self.agents[task['agent'].lower()].raw_answer_blocks(answer) |
| 250 | success = self.agents[task['agent'].lower()].get_success |
| 251 | self.agents[task['agent'].lower()].show_answer() |
| 252 | pretty_print(f"Agent {task['agent']} completed task.", color="status") |
| 253 | self.logger.info(f"Agent {task['agent']} finished working on {task['task']}. Success: {success}") |
| 254 | agent_answer += "\nAgent succeeded with task." if success else "\nAgent failed with task (Error detected)." |
| 255 | return agent_answer, success |
| 256 | |
| 257 | def get_work_result_agent(self, task_needs, agents_work_result): |
| 258 | res = {k: agents_work_result[k] for k in task_needs if k in agents_work_result} |
no test coverage detected