feat: change log to inside function

src/crewai/crew.py

@@ -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,6 +134,10 @@ class Crew(BaseModel):
         default=False,
         description="output_log_file",
     )
+    planning: Optional[bool] = Field(
+        default=False,
+        description="Plan the crew execution and add the plan to the crew.",
+    )
 
     @field_validator("id", mode="before")
     @classmethod
@@ -340,6 +345,9 @@ class Crew(BaseModel):
 
             agent.create_agent_executor()
 
+        if self.planning:
+            self._handle_crew_planning()
+
         metrics = []
 
         if self.process == Process.sequential:
@@ -421,6 +429,14 @@ class Crew(BaseModel):
 
         return results
 
+    def _handle_crew_planning(self):
+        """Handles the Crew planning."""
+        self._logger.log("info", "Planning the crew execution")
+        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
+
     def _run_sequential_process(self) -> Union[str, Dict[str, Any]]:
         """Executes tasks sequentially and returns the final output."""
         task_output = None

src/crewai/task.py

@@ -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."
 

src/crewai/utilities/planning_handler.py

@@ -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)