Adaptive Planning: Responding to Change

Overview

Adaptive Planning means adjusting plans when: - Assumptions prove wrong - New information emerges - Circumstances change - Progress stalls

Rigid plans fail. Adaptive plans survive.

---

Why Adaptation Matters

Rigid Plan Failure

Initial plan:
  Step 1: Search for data (assume 1 hour)
  Step 2: Analyze data (assume 1 hour)
  Step 3: Write report (assume 2 hours)
  Total: 4 hours

Reality:
  Step 1: Search takes 3 hours (database is slow)
  Step 2: Data is incomplete (need to find more)
  Step 3: Analysis takes 4 hours (data is complex)

Rigid plan: Follow original 4-hour plan
Result: Incomplete, wrong report

Adaptive plan: Adjust as issues emerge
Result: Complete, correct report (6 hours instead of 4)

---

Adaptation Triggers

What Triggers Re-Planning?

class AdaptationMonitor:
    def should_replan(self, plan: dict, execution: dict) -> bool:
        """Check if plan needs adjustment"""

        # Check 1: Time deviation
        if execution['elapsed_time'] > plan['estimated_time'] * 1.5:
            return True  # Taking too long

        # Check 2: Progress stalling
        if execution['progress'] < expected_progress_at_this_time:
            return True  # Falling behind

        # Check 3: Unexpected obstacles
        if execution['errors'] > acceptable_error_count:
            return True  # Too many problems

        # Check 4: New information
        if new_context_changes_plan():
            return True  # Context changed

        # Check 5: Success metrics not tracking
        if not tracking_toward_goal():
            return True  # Won't reach goal

        return False  # Plan still viable

# Usage
monitor = AdaptationMonitor()
if monitor.should_replan(plan, execution):
    plan = replan(plan, execution)

---

Replanin Strategies

Strategy 1: Local Repair

Fix just the failing part, keep rest:

class LocalRepair:
    def adapt(self, plan: List[dict], failure_index: int, error: Exception):
        """Fix just the failing step"""

        # Keep everything before failure
        repaired_plan = plan[:failure_index]

        # Fix the failing step
        original_step = plan[failure_index]
        fixed_step = self.fix_step(original_step, error)

        repaired_plan.append(fixed_step)

        # Keep everything after (may need adjustment)
        repaired_plan.extend(plan[failure_index + 1:])

        return repaired_plan

    def fix_step(self, step: dict, error: Exception) -> dict:
        """Find alternative approach for step"""

        # Try different tool/method
        alternatives = self.find_alternatives(step)

        for alt in alternatives:
            if self.likely_to_work(alt, error):
                return alt

        # If no alternative, escalate
        return None

# Example
plan = [
    "Search for data (using API)",
    "Parse results",
    "Generate report"
]

# API fails at step 1
repair = LocalRepair()
repaired = repair.adapt(plan, 0, APIError("Rate limited"))
# Result: ["Search for data (using web scraping)", ...]

Strategy 2: Full Replan

Start over with new understanding:

class FullReplan:
    def adapt(self, original_plan: List[dict], execution_log: dict):
        """Replan based on lessons learned"""

        # Analyze what went wrong
        failures = execution_log['failures']
        learnings = self.extract_learnings(failures)

        # Update assumptions
        new_assumptions = self.update_assumptions(learnings)

        # Create new plan
        new_plan = self.create_plan_with_assumptions(new_assumptions)

        return new_plan

    def extract_learnings(self, failures):
        """What did we learn from failures?"""

        learnings = {
            'time_estimates': self.analyze_timing(failures),
            'risks': self.identify_new_risks(failures),
            'dependencies': self.identify_dependencies(failures),
            'blockers': self.identify_blockers(failures)
        }

        return learnings

# Example
old_plan_failed = {
    'failures': [
        {'step': 'Database query', 'issue': 'Query timeout', 'time': 45_min},
        {'step': 'Data validation', 'issue': '20% invalid records', 'time': 30_min}
    ]
}

replan = FullReplan()
new_plan = replan.adapt(old_plan, old_plan_failed)
# New plan accounts for:
# - Longer query times
# - Need to validate/clean data

Strategy 3: Predictive Adaptation

Anticipate issues and preemptively adjust:

class PredictiveAdapter:
    def anticipate_issues(self, plan: List[dict]) -> List[dict]:
        """Predict problems and add mitigation steps"""

        adapted_plan = []

        for step in plan:
            adapted_plan.append(step)

            # Predict potential issues
            predicted_issues = self.predict_failures(step)

            # Add mitigation steps
            for issue in predicted_issues:
                mitigation = self.create_mitigation(step, issue)
                adapted_plan.append(mitigation)

        return adapted_plan

    def predict_failures(self, step: dict) -> List[dict]:
        """Predict what could go wrong"""

        # Based on similar past experiences
        similar_past = self.find_similar_past_steps(step)

        # What issues occurred then?
        issues = []
        for past_step in similar_past:
            if past_step['failed']:
                issues.append({
                    'type': past_step['failure_type'],
                    'probability': self.estimate_probability(past_step),
                    'severity': past_step['severity']
                })

        return issues

# Example
plan = [
    "Query large dataset",
    "Parse JSON response",
    "Validate data"
]

predictor = PredictiveAdapter()
robust_plan = predictor.anticipate_issues(plan)
# Result:
# [
#   "Query large dataset",
#   "  [ADD] Set timeout of 60 seconds",
#   "  [ADD] Have fallback to cached data",
#   "Parse JSON response",
#   "  [ADD] Validate JSON format first",
#   ...
# ]

---

Adaptive Replanning System

Complete Implementation

class AdaptiveAgent:
    def __init__(self):
        self.plan = []
        self.execution_log = []
        self.adaptation_count = 0

    def execute_with_adaptation(self, goal: dict, initial_plan: List[dict]):
        """Execute plan, adapting as needed"""

        self.plan = initial_plan
        max_adaptations = 3

        for step_num, step in enumerate(self.plan):
            try:
                # Try to execute
                result = self.execute_step(step)
                self.execution_log.append({'step': step, 'result': result, 'success': True})

            except Exception as e:
                self.execution_log.append({'step': step, 'result': e, 'success': False})

                # Decide how to adapt
                if self.adaptation_count < max_adaptations:
                    adapted_plan = self.decide_adaptation_strategy(step, e)

                    if adapted_plan:
                        # Replan from this point forward
                        self.plan = self.plan[:step_num] + adapted_plan
                        self.adaptation_count += 1

                        # Continue with adapted plan
                        continue

                # If can't adapt, escalate
                return self.escalate(step, e, goal)

        # Plan completed
        return {
            'success': True,
            'adaptations': self.adaptation_count,
            'execution_log': self.execution_log
        }

    def decide_adaptation_strategy(self, failed_step: dict, error: Exception) -> List[dict]:
        """Choose how to adapt"""

        error_severity = self.classify_error(error)

        if error_severity == 'RECOVERABLE':
            # Local repair: fix just this step
            return self.local_repair(failed_step, error)

        elif error_severity == 'REQUIRES_CONTEXT_CHANGE':
            # Full replan: start over with new understanding
            return self.full_replan(failed_step, error)

        else:
            # Can't adapt
            return None

    def local_repair(self, step: dict, error: Exception) -> List[dict]:
        """Try alternative approach for failing step"""

        alternatives = self.find_alternatives(step)

        # Try each alternative
        for alt in alternatives:
            if self.try_alternative(alt):
                return [alt]  # Just replace this step

        return None  # No viable alternative

    def full_replan(self, step: dict, error: Exception) -> List[dict]:
        """Replan everything from this point"""

        # Learn from failure
        learnings = self.extract_learnings(error)

        # Create new plan with learnings
        remaining_goal = self.extract_remaining_goal(step)
        new_plan = self.create_plan(remaining_goal, context=learnings)

        return new_plan

    def classify_error(self, error: Exception) -> str:
        """How severe is this error?"""

        if isinstance(error, RecoverableError):
            return 'RECOVERABLE'
        elif isinstance(error, ContextChangeError):
            return 'REQUIRES_CONTEXT_CHANGE'
        else:
            return 'FATAL'

# Usage
agent = AdaptiveAgent()

initial_plan = [
    "Search for research papers",
    "Analyze papers",
    "Write summary"
]

result = agent.execute_with_adaptation(goal, initial_plan)

print(f"Completed: {result['success']}")
print(f"Adaptations made: {result['adaptations']}")

---

Adaptive Warnings ⚠️

Warning 1: Thrashing (Too Much Adaptation)

# ❌ WRONG: Replan too frequently
if any_error_occurs:
    replan()  # Every error triggers new plan
# Result: Constantly replanning, never progressing

# ✅ RIGHT: Replan strategically
error_threshold = 3  # Tolerate some errors
if error_count > error_threshold:
    replan()  # Only replan when pattern emerges

Key Lesson: Some errors are normal. Only adapt when pattern detected.

Warning 2: Losing Sight of Goal

# ❌ WRONG: Adapt without checking goal
adapt_plan(plan, error)
# Might adapt in direction away from goal

# ✅ RIGHT: Verify adaptation moves toward goal
new_plan = adapt_plan(plan, error)
if not still_tracking_to_goal(new_plan):
    # Don't use adaptation
    use_different_adaptation()

Key Lesson: Adaptation must still serve the goal.

Warning 3: Cascading Changes

# ❌ WRONG: Change one thing in isolation
replan_step_5()
# Step 6 now invalid because it depends on old step 5

# ✅ RIGHT: Validate downstream impact
replan_step_5()
validate_remaining_steps()  # Check if still valid
fix_downstream_if_needed()

Key Lesson: Plans are interconnected. Validate ripple effects.

---

Best Practices

1. Detect Early

# ✅ Good: Monitor during execution
while executing:
    if should_replan():
        replan()  # Early detection

# ❌ Bad: Wait until end
while executing:
    pass
if failed():
    replan()  # Too late, wasted effort

2. Prioritize Solutions

# ✅ Good: Try solutions in order
def adapt(plan):
    if can_local_repair():
        return local_repair()  # Minimal change
    elif can_adjust_approach():
        return adjust_approach()  # Medium change
    else:
        return full_replan()  # Large change

# ❌ Bad: Random approach
adapt(plan)  # Might thrash

3. Learn from Adaptation

# ✅ Good: Track adaptations for future
if adapted_plan:
    record_adaptation({
        'original_step': step,
        'issue': error,
        'solution': adaptation,
        'success': did_work
    })

    # Use learnings next time
    if similar_problem_occurs:
        apply_past_solution()

# ❌ Bad: Don't track
if adapted_plan:
    use_it()  # No memory of what worked

---

Real-World Example

class CustomerServiceAgent:
    """Service agent that adapts handling strategy"""

    def handle_complaint(self, complaint: dict):
        """Handle customer complaint adaptively"""

        # Initial approach: Try standard resolution
        plan = [
            "Understand issue",
            "Offer refund",
            "Close ticket"
        ]

        # Execute with adaptation
        for step in plan:
            result = execute(step)

            if result.failed:
                # Adapt based on failure
                if result.error == "Customer_Unsatisfied":
                    # Standard refund didn't work
                    # Try different approach
                    plan = [
                        "Escalate to manager",
                        "Offer replacement product + discount",
                        "Get satisfaction confirmation"
                    ]

                elif result.error == "Manager_Unavailable":
                    # Escalation blocked
                    # Try different path
                    plan = [
                        "Offer immediate store credit",
                        "Schedule callback with manager",
                        "Send follow-up survey"
                    ]

                # Continue with adapted plan

---

Key Takeaways

1. Plans change; goals don't - Adapt approach, stay focused
2. Detect issues early - Monitor progress constantly
3. Try local fixes first - Minimal changes preferred
4. Validate adaptations - Check they still serve the goal
5. Learn from failures - Apply past solutions to similar issues
6. Avoid thrashing - Only adapt when needed
7. Document adaptations - Future agent can learn

---

Next Steps

• Go back to Goal Oriented Reasoning for more context
• Or move to Part 6: Tool Use

---

Last Updated: August 9, 2026