stephane.david/agenticSeek
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat : planner adapt to task failure + temporary remove frontend task panel

Browse Source
This commit is contained in:
martin legrand
2025-04-23 15:51:47 +02:00
parent d01ae7217f
commit 564a09c96d
6 changed files with 143 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sources/agents/agent.py
+4
View File
@@ -78,6 +78,10 @@ class Agent():
def get_tools(self) -> dict:
return self.tools
@property
def get_success(self) -> bool:
return self.success
def get_blocks_result(self) -> list:
return self.blocks_result
sources/agents/planner_agent.py
+132 -21
View File
@@ -26,11 +26,18 @@ class PlannerAgent(Agent):
}
self.role = "planification"
self.type = "planner_agent"
def parse_agent_tasks(self, text):
tasks = []
def get_task_names(self, text: str) -> List[str]:
"""
Extracts task names from the given text.
This method processes a multi-line string, where each line may represent a task name.
containing '##' or starting with a digit. The valid task names are collected and returned.
Args:
text (str): A string containing potential task titles (eg: Task 1: I will...).
Returns:
List[str]: A list of extracted task names that meet the specified criteria.
"""
tasks_names = []
lines = text.strip().split('\n')
for line in lines:
if line is None:
@@ -41,9 +48,23 @@ class PlannerAgent(Agent):
if '##' in line or line[0].isdigit():
tasks_names.append(line)
continue
return tasks_names
def parse_agent_tasks(self, text: str) -> List[Tuple[str, str]]:
"""
Parses agent tasks from the given LLM text.
This method extracts task information from a JSON. It identifies task names and their details.
Args:
text (str): The input text containing task information in a JSON-like format.
Returns:
List[Tuple[str, str]]: A list of tuples containing task names and their details.
"""
tasks = []
tasks_names = self.get_task_names(text)
blocks, _ = self.tools["json"].load_exec_block(text)
if blocks == None:
return (None, None)
return []
for block in blocks:
line_json = json.loads(block)
if 'plan' in line_json:
@@ -58,10 +79,18 @@ class PlannerAgent(Agent):
tasks.append(agent)
if len(tasks_names) != len(tasks):
names = [task['task'] for task in tasks]
return zip(names, tasks)
return zip(tasks_names, tasks)
return list(map(list, zip(names, tasks)))
return list(map(list, zip(names, tasks)))
def make_prompt(self, task: dict, agent_infos_dict: dict):
def make_prompt(self, task: str, agent_infos_dict: dict) -> str:
"""
Generates a prompt for the agent based on the task and previous agents work information.
Args:
task (str): The task to be performed.
agent_infos_dict (dict): A dictionary containing information from other agents.
Returns:
str: The formatted prompt for the agent.
"""
infos = ""
if agent_infos_dict is None or len(agent_infos_dict) == 0:
infos = "No needed informations."
@@ -76,8 +105,14 @@ class PlannerAgent(Agent):
"""
return prompt
def show_plan(self, agents_tasks: dict, answer: str) -> None:
if agents_tasks == (None, None):
def show_plan(self, agents_tasks: List[dict], answer: str) -> None:
"""
Displays the plan made by the agent.
Args:
agents_tasks (dict): The tasks assigned to each agent.
answer (str): The answer from the LLM.
"""
if agents_tasks == []:
pretty_print(answer, color="warning")
pretty_print("Failed to make a plan. This can happen with (too) small LLM. Clarify your request and insist on it making a plan within ```json.", color="failure")
return
@@ -85,8 +120,15 @@ class PlannerAgent(Agent):
for task_name, task in agents_tasks:
pretty_print(f"{task['agent']} -> {task['task']}", color="info")
pretty_print("▔▗ E N D ▖▔", color="status")
async def make_plan(self, prompt: str) -> str:
"""
Asks the LLM to make a plan.
Args:
prompt (str): The prompt to be sent to the LLM.
Returns:
str: The plan made by the LLM.
"""
ok = False
answer = None
while not ok:
@@ -94,35 +136,98 @@ class PlannerAgent(Agent):
self.memory.push('user', prompt)
answer, _ = await self.llm_request()
agents_tasks = self.parse_agent_tasks(answer)
if agents_tasks == (None, None):
if "NO_UPDATE" in agents_tasks:
return []
if agents_tasks == []:
prompt = f"Failed to parse the tasks. Please make a plan within ```json.\n"
pretty_print("Failed to make plan. Retrying...", color="warning")
continue
self.show_plan(agents_tasks, answer)
ok = True
return answer
return self.parse_agent_tasks(answer)
async def update_plan(self, goal: str, agents_tasks: List[dict], agents_work_result: dict, id: str, success: bool) -> dict:
"""
Updates the plan with the results of the agents work.
Args:
goal (str): The goal to be achieved.
agents_tasks (list): The tasks assigned to each agent.
agents_work_result (dict): The results of the agents work.
Returns:
dict: The updated plan.
"""
#self.memory.clear()
last_agent_work = agents_work_result[id]
tool_success_str = "success" if success else "failure"
pretty_print(f"Agent {id} work {tool_success_str}.", color="success" if success else "failure")
next_task = agents_tasks[int(id)][0]
if success:
return agents_tasks # we only update the plan if last task failed, for now
update_prompt = f"""
Your goal was : {goal}
You previously made a plan, agents are currently working on it.
The last agent working on task: {id}, did the following work:
{last_agent_work}
But the agent {id} failed with the task.
The agent {id} about to work on task: {next_task}
Is the work done for task {id} leading to sucess or failure ? Did an agent fail with a task?
If agent work lead to success: answer "NO_UPDATE"
If agent work lead might to failure: update the plan.
plan should be within ```json like before.
You need to rewrite the whole plan, but only change the tasks after {id}.
Keep the plan as short as the original one if possible.
"""
print("PROMPT")
print(update_prompt)
print("END PROMPT")
pretty_print("Updating plan...", color="status")
plan = await self.make_plan(update_prompt)
if plan == []:
pretty_print("No plan update required.", color="info")
return agents_tasks
return plan
async def start_agent_process(self, task: dict, required_infos: dict | None) -> str:
"""
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.
"""
agent_prompt = self.make_prompt(task['task'], required_infos)
pretty_print(f"Agent {task['agent']} started working...", color="status")
agent_answer, _ = await self.agents[task['agent'].lower()].process(agent_prompt, None)
success = self.agents[task['agent'].lower()].get_success
self.agents[task['agent'].lower()].show_answer()
pretty_print(f"Agent {task['agent']} completed task.", color="status")
return agent_answer
return agent_answer, success
def get_work_result_agent(self, task_needs, agents_work_result):
return {k: agents_work_result[k] for k in task_needs if k in agents_work_result}
async def process(self, prompt: str, speech_module: Speech) -> Tuple[str, str]:
agents_tasks = (None, None)
async def process(self, goal: str, speech_module: Speech) -> Tuple[str, str]:
"""
Process the goal by dividing it into tasks and assigning them to agents.
Args:
goal (str): The goal to be achieved (user prompt).
speech_module (Speech): The speech module for text-to-speech.
Returns:
Tuple[str, str]: The result of the agent process and empty reasoning string.
"""
agents_tasks = []
agents_work_result = dict()
answer = await self.make_plan(prompt)
agents_tasks = self.parse_agent_tasks(answer)
agents_tasks = await self.make_plan(goal)
if agents_tasks == (None, None):
if agents_tasks == []:
return "Failed to parse the tasks.", ""
for task_name, task in agents_tasks:
i = 0
steps = len(agents_tasks)
while i < steps:
task_name, task = agents_tasks[i][0], agents_tasks[i][1]
self.status_message = "Starting agent process..."
pretty_print(f"I will {task_name}.", color="info")
pretty_print(f"Assigned agent {task['agent']} to {task_name}", color="info")
@@ -131,8 +236,14 @@ class PlannerAgent(Agent):
if agents_work_result is not None:
required_infos = self.get_work_result_agent(task['need'], agents_work_result)
try:
self.last_answer = await self.start_agent_process(task, required_infos)
self.last_answer, success = await self.start_agent_process(task, required_infos)
except Exception as e:
raise e
agents_work_result[task['id']] = self.last_answer
if i == steps - 1:
break
agents_tasks = await self.update_plan(goal, agents_tasks, agents_work_result, task['id'], success)
steps = len(agents_tasks)
i += 1
return self.last_answer, ""