Adaptive Planning

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

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
if any_error_occurs:
 replan() # Every error triggers new plan
# Result

# RIGHT
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_plan(plan, error)
# Might adapt in direction away from goal

# RIGHT
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
replan_step_5()
# Step 6 now invalid because it depends on old step 5

# RIGHT
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
while executing:
 if should_replan():
 replan() # Early detection

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

2. Prioritize Solutions

# Good
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
adapt(plan) # Might thrash

3. Learn from Adaptation

# Good
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
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