两处：备战末尾触发，crew 方法里承接（crews/utils.py:356 / crew.py:1417）：

# crews/utils.py:356  在 prepare_kickoff 最后
if crew.planning:
    crew._handle_crew_planning()      # 开了 planning 才跑，且在建好 agent 之后

# crew.py:1417  crew 侧的承接方法
def _handle_crew_planning(self) -> None:
    self._logger.log("info", "Planning the crew execution")
    result = CrewPlanner(
        tasks=self.tasks, planning_agent_llm=self.planning_llm   # 用 planning_llm（D19）
    )._handle_crew_planning()          # ★真正干活的在 CrewPlanner（L03-L06）
    ...                                # 拿到 result 后回填（L06）

先看输出要长什么样（planning_handler.py:15）：

# planning_handler.py:15
class PlanPerTask(BaseModel):
    """Represents a plan for a specific task."""
    task_number: int = Field(description="The 1-indexed task number this plan corresponds to", ge=1)
    task: str = Field(description="The task for which the plan is created")
    plan: str = Field(description="The step by step plan on how the agents can execute "
                                  "their tasks using the available tools with mastery")

class PlannerTaskPydanticOutput(BaseModel):
    """Output format for task planning results."""
    list_of_plans_per_task: list[PlanPerTask] = Field(...,
        description="Step by step plan on how the agents can execute their tasks ...")

# planning_handler.py:37
class CrewPlanner:
    def __init__(self, tasks: list[Task], planning_agent_llm: str | BaseLLM | None = None):
        self.tasks = tasks
        self.planning_agent_llm = planning_agent_llm or "gpt-5.4-mini"   # ★兜底默认模型

规划的主流程（planning_handler.py:57）：

# planning_handler.py:57
def _handle_crew_planning(self) -> PlannerTaskPydanticOutput:
    planning_agent = self._create_planning_agent()      # ① 造规划 agent
    tasks_summary = self._create_tasks_summary()         # ② 把所有任务汇成一段摘要（L05）
    planner_task = self._create_planner_task(            # ③ 造一个"做规划"的任务
        planning_agent=planning_agent, tasks_summary=tasks_summary)
    result = planner_task.execute_sync()                 # ④ 同步执行这个规划任务（调 LLM）
    if isinstance(result.pydantic, PlannerTaskPydanticOutput):
        return result.pydantic                           # ⑤ 返回结构化计划
    raise ValueError("Failed to get the Planning output")

# planning_handler.py:80  规划 agent 的人设
def _create_planning_agent(self) -> Agent:
    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 ...",
        backstory="Planner agent for crew planning",
        llm=self.planning_agent_llm)

# planning_handler.py:97  规划任务本身
@staticmethod
def _create_planner_task(planning_agent: Agent, tasks_summary: str) -> Task:
    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 ...",
        expected_output="Step by step plan on how the agents can execute their tasks ...",
        agent=planning_agent,
        output_pydantic=PlannerTaskPydanticOutput)       # ★强制结构化输出

规划质量取决于喂进去什么（planning_handler.py:137）：

# planning_handler.py:137
def _create_tasks_summary(self) -> str:
    tasks_summary = []
    for idx, task in enumerate(self.tasks):
        knowledge_list = self._get_agent_knowledge(task)     # 取该任务 agent 的知识源
        agent_tools = (
            f"[{', '.join(str(tool) for tool in task.agent.tools)}]"
            if task.agent and task.agent.tools else '"agent has no tools"',
            ...)
        task_summary = 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": {"".join(agent_tools)}"""
        tasks_summary.append(task_summary)
    return " ".join(tasks_summary)

回到 crew 侧，拿到计划后的回填（crew.py:1417）：

# crew.py:1417
def _handle_crew_planning(self) -> None:
    result = CrewPlanner(tasks=self.tasks, planning_agent_llm=self.planning_llm)._handle_crew_planning()

    plan_map: dict[int, str] = {}
    for step_plan in result.list_of_plans_per_task:
        if step_plan.task_number in plan_map:
            self._logger.log("warning",
                f"Duplicate plan for Task Number {step_plan.task_number}, using the first plan")
        else:
            plan_map[step_plan.task_number] = step_plan.plan   # 编号 → 计划

    for idx, task in enumerate(self.tasks):
        task_number = idx + 1
        if task_number in plan_map:
            task.description += plan_map[task_number]          # ★计划拼到任务描述后面
        else:
            self._logger.log("warning", f"No plan found for Task Number {task_number}")

上面的 CrewPlanner 是"开工前一次性规划"。还有个更高级的 PlannerObserver（agents/planner_observer.py:39），做"执行中动态观察修正"：

# agents/planner_observer.py:39
class PlannerObserver:
    """Observes step execution results and decides on plan continuation.
    When ``observe_steps`` is enabled ..., after EVERY step execution this class:
    1. Analyzes what the step accomplished
    2. Identifies new information learned
    3. Decides if the remaining plan is still valid
    4. Suggests lightweight refinements or triggers full replanning
    ...
    LLM resolution (magical fallback):
    - If ``agent.planning_config.llm`` is explicitly set → use that
    - Otherwise → fall back to ``agent.llm`` (same LLM for everything)"""
    def __init__(self, agent, task=None, kickoff_input=""):
        self.agent = agent
        self.task = task
        self.llm = self._resolve_llm()