andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-01-09 08:08:32 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

feat: refactor and improve code for planning feature

Browse Source
This commit is contained in:
Eduardo Chiarotti
2024-07-11 16:45:24 -03:00
parent 2c2193cda3
commit 852c88f94a
3 changed files with 71 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
src/crewai/crew.py
View File

@@ -30,6 +30,7 @@ from crewai.tools.agent_tools import AgentTools
from crewai.utilities import I18N, FileHandler, Logger, RPMController
from crewai.utilities.constants import TRAINED_AGENTS_DATA_FILE, TRAINING_DATA_FILE
from crewai.utilities.evaluators.task_evaluator import TaskEvaluator
from crewai.utilities.planning_handler import CrewPlanner
from crewai.utilities.training_handler import CrewTrainingHandler
try:
@@ -133,7 +134,7 @@ class Crew(BaseModel):
default=False,
description="output_log_file",
)
plan: Optional[bool] = Field(
planning: Optional[bool] = Field(
default=False,
description="Plan the crew execution and add the plan to the crew.",
)
@@ -344,7 +345,7 @@ class Crew(BaseModel):
agent.create_agent_executor()
if self.plan:
if self.planning:
self._logger.log("info", "Planning the crew execution")
self._handle_crew_planning()
@@ -430,39 +431,8 @@ class Crew(BaseModel):
return results
def _handle_crew_planning(self):
"""Handles the crew planning."""
planning_agent = Agent(
role="Task Execution Planner",
goal="Your goal is to create an extremely detailed, step-by-step plan based on the tasks and tools "
"available to each agent so that they can perform the tasks in an exemplary manner",
backstory="Planner agent for crew planning",
)
tasks_summary = []
for idx, task in enumerate(self.tasks):
tasks_summary.append(
f"""
Task Number {idx + 1} - {task.description}
"task_description": {task.description}
"task_expected_output": {task.expected_output}
"agent": {task.agent.role if task.agent else "None"}
"agent_goal": {task.agent.goal if task.agent else "None"}
"task_tools": {task.tools}
"agent_tools": {task.agent.tools if task.agent else "None"}
"""
)
class PlannerTaskPydanticOutput(BaseModel):
list_of_plans_per_task: List[str]
planner_task = Task(
description=f"Based on this tasks summary: {' '.join(tasks_summary)} \n Create the most descriptive plan based on the tasks descriptions, tools availables and agents goals for them to execute with perfection their goals",
expected_output="Step by step plan on how the agents can execute their tasks using the available tools with mastery",
agent=planning_agent,
output_pydantic=PlannerTaskPydanticOutput,
)
result = planner_task.execute()
"""Handles the Crew planning."""
result = CrewPlanner(self.tasks)._handle_crew_planning()
for task, step_plan in zip(self.tasks, result.list_of_plans_per_task):
task.description += step_plan

4
src/crewai/task.py
View File

@@ -178,7 +178,7 @@ class Task(BaseModel):
agent: BaseAgent | None = None,
context: Optional[str] = None,
tools: Optional[List[Any]] = None,
) -> str | None:
) -> Any:
"""Execute the task.
Returns:
@@ -319,7 +319,7 @@ class Task(BaseModel):
)
return converter
def _export_output(self, result: str) -> Any:
def _export_output(self, result: str) -> Any: # TODO: Refactor and fix type hints
exported_result = result
instructions = "I'm gonna convert this raw text into valid JSON."

64
src/crewai/utilities/planning_handler.py Normal file
View File

@@ -0,0 +1,64 @@
from typing import List
from pydantic import BaseModel
from crewai.agent import Agent
from crewai.task import Task
class PlannerTaskPydanticOutput(BaseModel):
list_of_plans_per_task: List[str]
class CrewPlanner:
def __init__(self, tasks: List[Task]):
self.tasks = tasks
def _handle_crew_planning(self):
"""Handles the Crew planning by creating detailed step-by-step plans for each task."""
planning_agent = self._create_planning_agent()
tasks_summary = self._create_tasks_summary()
planner_task = self._create_planner_task(planning_agent, tasks_summary)
return planner_task.execute()
def _create_planning_agent(self) -> Agent:
"""Creates the planning agent for the crew planning."""
return Agent(
role="Task Execution Planner",
goal=(
"Your goal is to create an extremely detailed, step-by-step plan based on the tasks and tools "
"available to each agent so that they can perform the tasks in an exemplary manner"
),
backstory="Planner agent for crew planning",
)
def _create_planner_task(self, planning_agent: Agent, tasks_summary: str) -> Task:
"""Creates the planner task using the given agent and tasks summary."""
return Task(
description=(
f"Based on these tasks summary: {tasks_summary} \n Create the most descriptive plan based on the tasks "
"descriptions, tools available, and agents' goals for them to execute their goals with perfection."
),
expected_output="Step by step plan on how the agents can execute their tasks using the available tools with mastery",
agent=planning_agent,
output_pydantic=PlannerTaskPydanticOutput,
)
def _create_tasks_summary(self) -> str:
"""Creates a summary of all tasks."""
tasks_summary = []
for idx, task in enumerate(self.tasks):
tasks_summary.append(
f"""
Task Number {idx + 1} - {task.description}
"task_description": {task.description}
"task_expected_output": {task.expected_output}
"agent": {task.agent.role if task.agent else "None"}
"agent_goal": {task.agent.goal if task.agent else "None"}
"task_tools": {task.tools}
"agent_tools": {task.agent.tools if task.agent else "None"}
"""
)
return " ".join(tasks_summary)