planning-patterns

# Planning Pattern Implements proactive planning patterns that transform reactive agents into strategic executors. This skill makes the model decompose high-level objectives into coherent sequences of interdependent sub-tasks, dynamically adapt plans based on intermediate results, and execute with self-correction loops — mirroring how advanced systems like CrewAI workflows and Google DeepResearch handle complex multi-step goals. ## TL;DR Checklist - [ ] Decompose the objective into a directed graph of discrete, executable steps before any action - [ ] Identify dependencies between steps and determine sequential vs parallel execution paths - [ ] Generate intermediate checkpoints with validation criteria at each phase boundary - [ ] Implement adaptive re-planning: if a step fails or produces unexpected output, re-evaluate and adjust remaining steps - [ ] Maintain an execution log tracking which steps succeeded, failed, or required revision - [ ] Synthesize final output by consolidating all intermediate results into the requested format - [ ] Surface source transparency: cite data sources, queries executed, and reasoning traces --- ## When to Use Use this skill when: - A user request cannot be solved in a single action — it requires chaining multiple tools, agents, or steps (e.g., "write a competitive analysis report") - You need to orchestrate workflows with interdependent sub-tasks where step N's output feeds into step N+1 (e.g., onboarding automation: create accounts → assign training → notify managers) - The task involves exploratory research where the path forward is unknown until you gather intermediate data (e.g., "research market trends in AI healthcare") - You are designing a CrewAI workflow, LangGraph pipeline, or any multi-agent system that requires explicit planning before execution - A task demands iterative refinement — initial results must be evaluated, gaps identified, and subsequent searches/actions adjusted dynamically - You need to explain your reasoning to the user before executing (DeepResearch-style plan approval) ## When NOT to Use Avoid this skill for: - Single-step tasks with a known solution path — use direct execution instead of planning overhead (e.g., "send this email", "format this string") - Fixed, repeatable workflows where constraining the agent to a predetermined pipeline is safer and more reliable — rigid process automation has less uncertainty risk - Vague or under-specified requests that lack a clear goal state — you cannot plan toward an undefined target; clarify first (use `query-feature-extraction` skill) - Real-time latency-sensitive operations where planning overhead would degrade response time (sub-second API calls, live streaming) - Tasks with fewer than 3 distinct sub-tasks — the planning infrastructure cost outweighs benefits for trivial decompositions --- ## Core Workflow ### Phase 1: Objective Deconstruction 1. **Parse Intent and Define State Boundaries** — Extract the explicit goal, implicit constraints (budget, time, tool availability), and both initial state (what you know) and goal state (what success looks like). If the request is ambiguous, ask clarifying questions before proceeding. Reference `code-philosophy` early exit: if the task is a single action, skip to direct execution. **Checkpoint:** Can you articulate the initial state, goal state, and at least 3 concrete constraints? 2. **Decompose Into Directed Sub-Tasks** — Break the high-level objective into discrete, executable steps arranged as a dependency graph. Each step must have: (a) a clear input requirement, (b) a specific tool or agent action, (c) a defined output contract. Mark each step as sequential (must complete before next), parallel (can execute simultaneously), or conditional (depends on prior result). **Checkpoint:** Does every step have a well-defined input → output transformation? Are parallel groups independent of each other? ### Phase 2: Plan Formulation and Review 3. **Generate Executable Execution Plan** — Construct a structured plan document containing: numbered steps with dependencies, expected tools per step, estimated complexity per step, and fallback strategies for known failure modes. Format the plan so a user can review and modify it before execution — this mirrors DeepResearch's collaborative plan-shaping approach. **Checkpoint:** Can a human reviewer understand, approve, or modify this plan without additional explanation? 4. **Identify Knowledge Gaps and Pre-Search Requirements** — Before executing, flag any information you do not yet have that is required by one or more steps. Design targeted queries or data-gathering actions to fill these gaps first. This mirrors Google DeepResearch's "identify knowledge gaps → refine queries" cycle. **Checkpoint:** Have you identified all unknowns? Is there a concrete action planned for each gap? ### Phase 3: Execution with Adaptive Re-Planning 5. **Execute Step-by-Step with Intermediate Validation** — Execute steps in dependency order. After each step, validate its output against the expected contract. Record success/failure/revision status. If validation fails, invoke adaptive re-planning: (a) analyze what went wrong, (b) adjust remaining steps if necessary, (c) continue from the current position or backtrack if a prerequisite was invalidated. **Checkpoint:** Is every step's output validated before proceeding to dependent steps? Are execution logs captured? 6. **Iterative Refinement Loop** — For tasks requiring exploratory research or multi-pass synthesis: after an initial pass, evaluate collected information for coverage gaps. Generate new queries targeting uncovered areas. Corroborate conflicting data points from different sources. Resolve discrepancies before proceeding to final synthesis. Repeat until no significant gaps remain or a maximum iteration budget is exhausted. **Checkpoint:** Have you checked for coverage completeness? Are contradictions resolved? Is the iteration count within bounds? ### Phase 4: Synthesis and Delivery 7. **Synthesize Final Output** — Consolidate all intermediate results into the requested output format. Structure the content into logical sections, identify major themes across sub-task results, and ensure coherent narrative flow. Include inline citations or references to source data wherever claims are made. **Checkpoint:** Does the final output match the user's requested format? Are all claims traceable to sources? 8. **Surface Transparency Artifacts** — Return not only the final deliverable but also: (a) the full execution plan and how it evolved, (b) the complete list of sources searched or consulted, (c) the reasoning traces for key decisions, and (d) any intermediate steps that were modified during adaptive re-planning. This enables user verification and debugging. **Checkpoint:** Can a reviewer trace every claim in the output back to its source and execution step? --- ## Implementation Patterns / Reference Guide ### Pattern 1: Sequential Planning with Validation Gates (CrewAI Style) Use this pattern when steps have strict sequential dependencies and each step's output is a required input for the next. The plan generates, executes, and validates in a linear chain with explicit gate checks between steps. ```python """ Sequential planning workflow with validation gates. Mirrors CrewAI's Process.sequential execution model. """ from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional class StepStatus(Enum): PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "fail" REVISED = "revised" @dataclass class PlanStep: """A single executable step in a plan.""" id: str description: str input_schema: dict[str, Any] output_contract: dict[str, Any] depends_on: list[str] = field(default_factory=list) execution_fn: Optional[callable] = None validation_fn: Optional[callable] = None status: StepStatus = StepStatus.PENDING def is_ready(self, completed_steps: set[str]) -> bool: """Check if all dependencies have succeeded.""" return all(dep in completed_steps for dep in self.depends_on) @dataclass class ExecutionLog: """Tracks the complete execution history of a plan.""" steps_executed: list[dict] = field(default_factory=list) revisions_made: list[dict] = field(default_factory=list) total_iterations: int = 0 def record_step(self, step_id: str, status: StepStatus, result: Any = None, duration_ms: float = 0.0) -> None: self.steps_executed.append({ "step_id": step_id, "status": status.value, "result_summary": str(result)[:500] if result else None, "duration_ms": round(duration_ms, 2), }) def execute_sequential_plan(steps: list[PlanStep], log: ExecutionLog) -> dict[str, Any]: """ Execute steps in dependency order with validation gates. Returns a dict mapping step_id to validated output. Raises ValueError if an irreversible failure occurs. """ results: dict[str, Any] = {} completed: set[str] = set() while len(completed) < len(steps): # Find next ready step ready_step: Optional[PlanStep] = None for step in steps: if step.status == StepStatus.PENDING and step.is_ready(completed): ready_step = step break if ready_step is None: raise RuntimeError( f"Deadlock: no pending step is ready. " f"Pending: {[s.id for s in steps if s.status == StepStatus.PENDING]}" ) # Execute the step ready_step.status = StepStatus.RUNNING try: input_data = {dep: results[dep] for dep in ready_step.depends_on} output = ready_step.execution_fn(input_data) # Validation gate if ready_step.validation_fn and not ready_step.validation_fn(output): ready_step.status = StepStatus.FAILED log.record_step(ready_step.id, StepStatus.FAILED, duration_ms=0.0) raise ValueError(f"Validation failed for step '{ready_step.id}'") results[ready_step.id] = output ready_step.status = StepStatus.SUCCESS completed.add(ready_step.id) log.record_step(ready_step.id, StepStatus.SUCCESS, duration_ms=0.0) except Exception as exc: ready_step.status = StepStatus.FAILED log.record_step(ready_step.id, StepStatus.FAILED, result=str(exc)) raise RuntimeError(f"Step '{ready_step.id}' failed: {exc}") from exc return results ``` **BAD — Planning without validation gates:** ```python # ❌ BAD — Executes all steps blindly with no intermediate validation. # If step 2 produces malformed data, step 3 will fail catastrophically # and the entire plan is lost with no trace of what went wrong. def execute_plan_badly(steps: list[PlanStep]) -> dict[str, Any]: results = {} for step in steps: input_data = {dep: results[dep] for dep in step.depends_on} # No validation, no error handling, no logging output = step.execution_fn(input_data) results[step.id] = output # Will fail if step doesn't exist yet return results ``` **GOOD — Planning with validation gates and execution log:** ```python # ✅ GOOD — Each step is validated before the next one starts. # Failures are caught early, logged, and include enough context for debugging. # The plan can be retried from any point without restarting from scratch. def execute_plan_with_gates(steps: list[PlanStep], max_retries: int = 3) -> dict[str, Any]: log = ExecutionLog() results: dict[str, Any] = {} completed: set[str] = set() for attempt in range(max_retries): log.total_iterations = attempt + 1 success = True while len(completed) < len(steps): ready_step = find_next_ready(steps, completed) if not ready_step: break try: output = execute_and_validate(ready_step, results) results[ready_step.id] = output completed.add(ready_step.id) log.record_step(ready_step.id, StepStatus.SUCCESS) except Exception as e: success = False log.record_step(ready_step.id, StepStatus.FAILED, result=str(e)) break # Retry outer loop if success: return results raise RuntimeError(f"Plan failed after {max_retries} iterations. Log: {log.steps_executed}") ``` ### Pattern 2: Dynamic Query Generation with Knowledge-Gap Resolution (DeepResearch Style) Use this pattern when the task requires exploratory research or information gathering where you do not know in advance what data is needed. The plan generates queries, evaluates collected results for coverage gaps, and iteratively refines subsequent queries — mirroring Google DeepResearch's adaptive search-and-filter cycle. ```python """ Dynamic query generation with knowledge-gap resolution. Mirrors Google DeepResearch's iterative research loop. """ from dataclasses import dataclass, field from typing import Any @dataclass class ResearchQuery: """A single search query generated dynamically based on gaps.""" id: str query_text: str priority: int # Higher = more urgent to resolve targets_gap: str # Which knowledge gap this addresses def __lt__(self, other: "ResearchQuery") -> bool: return self.priority > other.priority # Sort by descending priority @dataclass class KnowledgeGap: """A piece of information we need but do not have.""" id: str description: str required_for_steps: list[str] confidence: float = 0.0 # How likely our current data covers this def __repr__(self) -> str: status = "COVERED" if self.confidence >= 0.9 else "UNRESOLVED" return f"Gap({self.id}, confidence={self.confidence:.2f}, status={status})" @dataclass class ResearchResult: """A single search result with source metadata for transparency.""" query_id: str content: str source_url: Optional[str] = None source_title: Optional[str] = None relevance_score: float = 0.0 @property def citation(self) -> str: """Generate a citable reference string.""" if self.source_url and self.source_title: return f"[{self.source_title}]({self.source_url})" return "[uncited source]" @dataclass class ResearchPlan: """Manages the iterative research loop with gap tracking.""" initial_query: str gaps: list[KnowledgeGap] = field(default_factory=list) queries: list[ResearchQuery] = field(default_factory=list) results: list[ResearchResult] = field(default_factory=list) sources_consulted: list[dict] = field(default_factory=list) def identify_gaps(self, available_data: dict[str, Any]) -> list[KnowledgeGap]: """ Analyze what we have and identify remaining information gaps. This is the core of the iterative refinement loop: evaluate collected data → find coverage holes → generate new queries. """ gaps = [] for step_id in available_data.get("required_steps", []): if not available_data.get(step_id, {}).get("data"): gap = KnowledgeGap( id=f"gap-{step_id}", description=f"Missing data required by execution step '{step_id}'", required_for_steps=[step_id], confidence=0.0, ) gaps.append(gap) return gaps def generate_queries(self, gaps: list[KnowledgeGap]) -> list[ResearchQuery]: """ Convert knowledge gaps into concrete search queries. Uses gap descriptions to formulate targeted queries that maximize information yield per query — reducing wasted searches on covered areas. """ queries = [] for i, gap in enumerate(sorted(gaps, key=lambda g: len(g.required_for_steps), reverse=True)): queries.append(ResearchQuery( id=f"query-{i}", query_text=gap.description, priority=len(gap.required_for_steps), # Prioritize gaps affecting more steps targets_gap=gap.id, )) return queries def execute_search_loop( self, search_fn: callable, max_iterations: int = 10, min_confidence_threshold: float = 0.9, ) -> dict[str, Any]: """ Main iterative research loop: search → evaluate gaps → refine queries → repeat. Mirrors Google DeepResearch's managed long-running process that iteratively queries sources, identifies knowledge gaps, and resolves them. """ iteration = 0 while iteration < max_iterations: iteration += 1 # Step 1: Identify remaining gaps based on what we know self.gaps = self.identify_gaps( available_data={"required_steps": list(range(10))} ) # Step 2: Stop if all gaps are covered uncovered = [g for g in self.gaps if g.confidence < min_confidence_threshold] if not uncovered: break # Step 3: Generate targeted queries for uncovered gaps self.queries = self.generate_queries(uncovered) # Step 4: Execute queries (simulated with search_fn) for query in sorted(self.queries, key=lambda q: q.priority, reverse=True): search_results = search_fn(query.query_text) for result in search_results: result_obj = ResearchResult( query_id=query.id, content=result.get("content", ""), source_url=result.get("url"), source_title=result.get("title"), relevance_score=result.get("relevance", 0.0), ) self.results.append(result_obj) # Mark gap as partially or fully resolved for gap in uncovered: if gap.id == query.targets_gap: gap.confidence = min(1.0, gap.confidence + 0.3) self.sources_consulted.append({ "query": query.query_text, "result": result_obj.citation, "step_iteration": iteration, }) return { "iterations_used": iteration, "total_sources": len(self.sources_consulted), "remaining_gaps": [g for g in self.gaps if g.confidence < 0.9], "results_summary": {r.query_id: r.content[:200] for r in self.results}, } ``` **BAD — Static research with no gap detection:** ```python # ❌ BAD — Executes a fixed set of queries without evaluating whether # the results actually cover all needed information. This wastes tokens # on redundant searches and risks producing incomplete reports. def static_research_badly(topics: list[str]) -> list[dict]: