Get the next task.
(task_creation_chain: LLMChain, result: Dict, task_description: str, task_list: List[str], objective: str)
| 99 | return cls(prompt=prompt, llm=llm, verbose=verbose) |
| 100 | |
| 101 | def get_next_task(task_creation_chain: LLMChain, result: Dict, task_description: str, task_list: List[str], objective: str) -> List[Dict]: |
| 102 | """Get the next task.""" |
| 103 | incomplete_tasks = ", ".join(task_list) |
| 104 | response = task_creation_chain.run(result=result, task_description=task_description, incomplete_tasks=incomplete_tasks, objective=objective) |
| 105 | # change the split method to re matching |
| 106 | # new_tasks = response.split('\n') |
| 107 | task_pattern = re.compile(r'\d+\. (.+?)\n') |
| 108 | new_tasks = task_pattern.findall(response) |
| 109 | |
| 110 | return [{"task_name": task_name} for task_name in new_tasks if task_name.strip()] |
| 111 | |
| 112 | def prioritize_tasks(task_prioritization_chain: LLMChain, this_task_id: int, task_list: List[Dict], objective: str) -> List[Dict]: |
| 113 | """Prioritize tasks.""" |