purpose: Implementation guide for self-correcting agent systems with feedback interpretation and retry strategies
type: technical-implementation
course: Tactical Agentic Coding
lesson: 05-supplemental
difficulty: INTERMEDIATE-ADVANCED
dependencies: lesson-05-close-the-loops, closed-loop-architecture, validation-strategies
last-updated: 2025-09-30
Self-correcting agents combine execution capabilities with validation feedback interpretation to autonomously fix errors and improve results without human intervention. This document provides complete implementation patterns for building agents that understand validation failures, analyze root causes, generate fixes, and retry intelligently until success.
# SELF-CORRECTING AGENTS
## HOW AGENTS SELF-CORRECT
### The Self-Correction Cycle
```
AGENT SELF-CORRECTION FLOW
┌─────────────────────────────────────────────────┐
│ PHASE 1: EXECUTION │
│ Agent generates work product │
│ (code, changes, artifacts) │
└──────────────────┬──────────────────────────────┘
↓
┌─────────────────────────────────────────────────┐
│ PHASE 2: VALIDATION │
│ Automated checks run │
│ Results: PASS or FAIL + details │
└──────────────────┬──────────────────────────────┘
↓
┌─────────┐
│ Success?│
└────┬────┘
│
┌─────────┴─────────┐
│ │
YES NO
│ │
▼ ▼
┌────────┐ ┌─────────────────────────────┐
│ DONE │ │ PHASE 3: INTERPRETATION │
└────────┘ │ Agent reads error messages │
│ Understands what failed │
└──────────┬──────────────────┘
↓
┌─────────────────────────────┐
│ PHASE 4: ANALYSIS │
│ Determine root cause │
│ Identify fix strategy │
└──────────┬──────────────────┘
↓
┌─────────────────────────────┐
│ PHASE 5: CORRECTION │
│ Generate and apply fix │
│ Prepare for retry │
└──────────┬──────────────────┘
│
└──────┐
│
(Return to PHASE 1 with context)
```
### Key Capabilities
**WHAT MAKES AN AGENT SELF-CORRECTING:**
1. **Error Interpretation:** Can read and understand validation output
2. **Root Cause Analysis:** Determines why validation failed
3. **Fix Generation:** Creates appropriate corrections
4. **Context Accumulation:** Learns from previous attempts
5. **Strategy Adaptation:** Changes approach based on feedback
6. **Retry Intelligence:** Knows when to retry vs. escalate
---
## FEEDBACK INTERPRETATION
### Understanding Validation Output
**AGENT MUST PARSE:**
1. **Exit codes:** 0 = success, non-zero = failure
2. **Error messages:** Specific failure descriptions
3. **File/line information:** Where errors occurred
4. **Stack traces:** Execution context
5. **Test results:** Which tests failed
6. **Performance metrics:** Speed, memory usage
### Implementation: Error Parser
```python
class ValidationOutputParser:
"""Parse validation output for agent consumption."""
def parse_validation_result(self, command_result):
"""
Convert raw validation output into structured data.
Args:
command_result: Result from subprocess.run()
Returns:
ParsedValidation with structured error information
"""
return ParsedValidation(
passed=(command_result.returncode == 0),
exit_code=command_result.returncode,
stdout=command_result.stdout,
stderr=command_result.stderr,
errors=self.extract_errors(command_result),
warnings=self.extract_warnings(command_result),
context=self.build_error_context(command_result)
)
def extract_errors(self, result):
"""Extract structured error information."""
errors = []
# Parse stderr for error patterns
for line in result.stderr.split('\n'):
error = self.parse_error_line(line)
if error:
errors.append(error)
# Parse stdout for test failures
test_failures = self.parse_test_failures(result.stdout)
errors.extend(test_failures)
return errors
def parse_error_line(self, line):
"""Parse a single error line."""
# Pattern: file:line:col: error: message
pattern = r'(.+):(\d+):(\d+):\s*error:\s*(.+)'
match = re.match(pattern, line)
if match:
return ErrorInfo(
file=match.group(1),
line=int(match.group(2)),
column=int(match.group(3)),
message=match.group(4),
severity="error"
)
return None
def build_error_context(self, result):
"""Build context for agent to understand error."""
context = {
"command": result.args,
"working_dir": os.getcwd(),
"exit_code": result.returncode,
"error_count": len(self.extract_errors(result)),
}
# Add framework-specific context
if "npm test" in str(result.args):
context["framework"] = "jest"
context["test_failures"] = self.count_test_failures(result.stdout)
return context
```
### Language-Specific Error Patterns
```python
class JavaScriptErrorParser:
"""Parse JS/TS specific errors."""
ERROR_PATTERNS = {
"syntax": r"SyntaxError: (.+)",
"type": r"TypeError: (.+)",
"reference": r"ReferenceError: (.+)",
"eslint": r"(.+):(\d+):(\d+): (.+) \[(.+)\]",
"test_fail": r"FAIL (.+)",
}
def parse(self, output):
"""Parse JavaScript error output."""
errors = []
for line in output.split('\n'):
for error_type, pattern in self.ERROR_PATTERNS.items():
match = re.search(pattern, line)
if match:
errors.append({
"type": error_type,
"message": match.group(1) if len(match.groups()) >= 1 else line,
"raw": line
})
return errors
```
```python
class PythonErrorParser:
"""Parse Python specific errors."""
def parse_traceback(self, stderr):
"""Parse Python traceback."""
lines = stderr.split('\n')
traceback = []
error_message = None
in_traceback = False
for line in lines:
if line.startswith('Traceback'):
in_traceback = True
elif in_traceback:
if line.startswith(' File'):
# Extract file, line, function
match = re.search(r'File "(.+)", line (\d+), in (.+)', line)
if match:
traceback.append({
"file": match.group(1),
"line": int(match.group(2)),
"function": match.group(3)
})
elif line and not line.startswith(' '):
# Error type and message
error_message = line
in_traceback = False
return {
"traceback": traceback,
"error": error_message
}
def parse_pytest_output(self, stdout):
"""Parse pytest failure output."""
failures = []
# Pattern: FAILED test_file.py::test_name - AssertionError: message
pattern = r'FAILED (.+)::(.+) - (.+): (.+)'
for match in re.finditer(pattern, stdout):
failures.append({
"file": match.group(1),
"test": match.group(2),
"error_type": match.group(3),
"message": match.group(4)
})
return failures
```
```python
class GoErrorParser:
"""Parse Go specific errors."""
def parse_build_errors(self, stderr):
"""Parse Go build/vet errors."""
errors = []
# Pattern: file.go:line:col: message
pattern = r'(.+\.go):(\d+):(\d+): (.+)'
for match in re.finditer(pattern, stderr):
errors.append({
"file": match.group(1),
"line": int(match.group(2)),
"column": int(match.group(3)),
"message": match.group(4)
})
return errors
def parse_test_failures(self, stdout):
"""Parse Go test failures."""
failures = []
# Pattern: --- FAIL: TestName (0.00s)
in_failure = False
current_test = None
for line in stdout.split('\n'):
if line.startswith('--- FAIL:'):
match = re.search(r'--- FAIL: (\w+)', line)
if match:
current_test = match.group(1)
in_failure = True
elif in_failure and line.strip():
failures.append({
"test": current_test,
"output": line.strip()
})
elif line.startswith('---'):
in_failure = False
return failures
```
---
## ERROR ANALYSIS PATTERNS
### Root Cause Identification
```python
class ErrorAnalyzer:
"""Analyze errors to identify root causes and fixes."""
def analyze(self, parsed_validation):
"""
Analyze validation result to determine root cause.
Returns:
Analysis with root cause and suggested fix strategy
"""
errors = parsed_validation.errors
# Categorize errors
syntax_errors = [e for e in errors if self.is_syntax_error(e)]
type_errors = [e for e in errors if self.is_type_error(e)]
logic_errors = [e for e in errors if self.is_logic_error(e)]
import_errors = [e for e in errors if self.is_import_error(e)]
# Determine primary issue
if syntax_errors:
root_cause = "SYNTAX_ERROR"
fix_strategy = "fix_syntax"
priority_errors = syntax_errors
elif import_errors:
root_cause = "MISSING_IMPORT"
fix_strategy = "add_imports"
priority_errors = import_errors
elif type_errors:
root_cause = "TYPE_MISMATCH"
fix_strategy = "fix_types"
priority_errors = type_errors
elif logic_errors:
root_cause = "LOGIC_ERROR"
fix_strategy = "fix_logic"
priority_errors = logic_errors
else:
root_cause = "UNKNOWN"
fix_strategy = "general_fix"
priority_errors = errors
return ErrorAnalysis(
root_cause=root_cause,
fix_strategy=fix_strategy,
priority_errors=priority_errors,
all_errors=errors,
confidence=self.calculate_confidence(errors),
suggested_actions=self.suggest_actions(root_cause)
)
def is_syntax_error(self, error):
"""Check if error is syntax-related."""
syntax_keywords = [
"SyntaxError", "unexpected token", "expected",
"missing", "invalid syntax"
]
return any(kw in error.message.lower() for kw in syntax_keywords)
def is_type_error(self, error):
"""Check if error is type-related."""
type_keywords = [
"TypeError", "type mismatch", "cannot convert",
"expected type", "incompatible types"
]
return any(kw in error.message.lower() for kw in type_keywords)
def is_logic_error(self, error):
"""Check if error is logic-related (test failures)."""
logic_keywords = [
"AssertionError", "expected", "actual",
"test failed", "FAIL"
]
return any(kw in error.message.lower() for kw in logic_keywords)
def is_import_error(self, error):
"""Check if error is import-related."""
import_keywords = [
"ImportError", "ModuleNotFoundError", "cannot find module",
"unresolved import"
]
return any(kw in error.message.lower() for kw in import_keywords)
def suggest_actions(self, root_cause):
"""Suggest specific actions based on root cause."""
actions = {
"SYNTAX_ERROR": [
"Check for missing brackets, parentheses, or quotes",
"Verify proper indentation",
"Review syntax for the language/framework"
],
"MISSING_IMPORT": [
"Add missing import statements",
"Check if module is installed",
"Verify import path is correct"
],
"TYPE_MISMATCH": [
"Review function signatures",
"Check variable types",
"Add type conversions if needed"
],
"LOGIC_ERROR": [
"Review test expectations",
"Check function logic",
"Verify edge case handling"
]
}
return actions.get(root_cause, ["Review error messages carefully"])
```
### Pattern-Based Error Recognition
```python
class ErrorPatternMatcher:
"""Match errors against known patterns with solutions."""
def __init__(self):
self.patterns = self.load_patterns()
def load_patterns(self):
"""Load known error patterns and solutions."""
return [
ErrorPattern(
name="missing_semicolon",
regex=r"expected ';'",
solution="Add semicolon to end of statement",
auto_fixable=True
),
ErrorPattern(
name="undefined_variable",
regex=r"'(\w+)' is not defined",
solution="Declare variable {1} or check spelling",
auto_fixable=False
),
ErrorPattern(
name="import_not_found",
regex=r"Cannot find module '(.+)'",
solution="Install package: npm install {1}",
auto_fixable=True,
fix_command="npm install {1}"
),
ErrorPattern(
name="async_await_missing",
regex=r"await.*not.*async",
solution="Add 'async' keyword to function declaration",
auto_fixable=True
),
# Add more patterns...
]
def match(self, error_message):
"""Find matching pattern for error."""
for pattern in self.patterns:
match = re.search(pattern.regex, error_message)
if match:
return MatchedPattern(
pattern=pattern,
matches=match.groups(),
solution=pattern.solution.format(*match.groups()),
auto_fixable=pattern.auto_fixable
)
return None
def get_auto_fixes(self, errors):
"""Get all auto-fixable errors."""
auto_fixes = []
for error in errors:
matched = self.match(error.message)
if matched and matched.auto_fixable:
auto_fixes.append(matched)
return auto_fixes
```
---
## RETRY STRATEGIES
### Strategy 1: Simple Retry with Context
```python
class SimpleRetryStrategy:
"""Basic retry with error context accumulation."""
def __init__(self, max_attempts=5):
self.max_attempts = max_attempts
def execute_with_retry(self, agent, task):
"""Execute task with simple retry logic."""
attempt = 0
error_history = []
while attempt < self.max_attempts:
try:
# Execute with accumulated context
result = agent.execute(
task=task,
previous_errors=error_history
)
# Validate
validation = self.validate(result)
if validation.passed:
return result
# Add to error history
error_history.append({
"attempt": attempt,
"errors": validation.errors,
"timestamp": time.time()
})
attempt += 1
except Exception as e:
error_history.append({
"attempt": attempt,
"exception": str(e)
})
attempt += 1
raise RetryExhaustedError(
f"Failed after {self.max_attempts} attempts",
error_history=error_history
)
```
### Strategy 2: Progressive Strategy Adaptation
```python
class AdaptiveRetryStrategy:
"""Adapt strategy based on error patterns."""
def __init__(self):
self.strategies = ["fast", "careful", "minimal", "verbose"]
self.current_strategy_index = 0
def execute_with_adaptation(self, agent, task):
"""Execute with strategy adaptation."""
max_total_attempts = 10
attempt = 0
while attempt < max_total_attempts:
strategy = self.strategies[self.current_strategy_index]
result = agent.execute(
task=task,
strategy=strategy
)
validation = self.validate(result)
if validation.passed:
return result
# Analyze why it failed
analysis = self.analyze_failure(validation.errors)
# Adapt strategy
if analysis.should_change_strategy:
self.current_strategy_index = min(
self.current_strategy_index + 1,
len(self.strategies) - 1
)
attempt += 1
raise RetryExhaustedError("All strategies exhausted")
def analyze_failure(self, errors):
"""Analyze if strategy change is needed."""
# If same errors repeating, change strategy
if self.errors_repeating(errors):
return AnalysisResult(should_change_strategy=True)
# If errors increasing, change strategy
if len(errors) > len(self.previous_errors):
return AnalysisResult(should_change_strategy=True)
return AnalysisResult(should_change_strategy=False)
```
### Strategy 3: Incremental Fix Approach
```python
class IncrementalFixStrategy:
"""Fix one error at a time, validate incrementally."""
def execute_incrementally(self, agent, task):
"""Fix errors one at a time."""
max_iterations = 20 # Higher limit for incremental
iteration = 0
current_code = task.initial_code
while iteration < max_iterations:
# Validate current state
validation = self.validate(current_code)
if validation.passed:
return current_code
# Take FIRST error only
first_error = validation.errors[0]
# Fix only this error
fixed_code = agent.fix_single_error(
code=current_code,
error=first_error
)
# Update current code
current_code = fixed_code
iteration += 1
raise RetryExhaustedError("Too many incremental fixes needed")
```
### Strategy 4: Backtracking Retry
```python
class BacktrackingStrategy:
"""Try different approaches, backtrack on failure."""
def __init__(self):
self.history = [] # Stack of (code, validation) pairs
def execute_with_backtracking(self, agent, task):
"""Execute with backtracking capability."""
max_depth = 5
current_code = task.initial_code
def try_approach(code, depth):
if depth > max_depth:
return None
# Validate current state
validation = self.validate(code)
if validation.passed:
return code
# Save state
self.history.append((code, validation))
# Try multiple fix approaches
approaches = agent.generate_fix_approaches(
code=code,
errors=validation.errors
)
for approach in approaches:
fixed_code = agent.apply_approach(code, approach)
# Recursive try
result = try_approach(fixed_code, depth + 1)
if result is not None:
return result
# Backtrack - no approach worked
return None
result = try_approach(current_code, 0)
if result is None:
raise RetryExhaustedError("No working approach found")
return result
```
### Strategy 5: Ensemble Retry
```python
class EnsembleRetryStrategy:
"""Try multiple agents/models in parallel."""
def __init__(self, agents):
self.agents = agents # List of different agents/models
async def execute_ensemble(self, task):
"""Execute with multiple agents, take best result."""
# Execute all agents in parallel
results = await asyncio.gather(*[
agent.execute_with_validation(task)
for agent in self.agents
], return_exceptions=True)
# Filter successful results
successful = [
r for r in results
if not isinstance(r, Exception) and r.validation.passed
]
if not successful:
# All failed - collect error patterns
all_errors = [
r.validation.errors if not isinstance(r, Exception) else [str(r)]
for r in results
]
raise EnsembleFailureError(
"All agents failed",
error_patterns=all_errors
)
# Return best result (e.g., fewest warnings)
return min(successful, key=lambda r: len(r.validation.warnings))
```
---
## WHEN TO ESCALATE TO HUMAN
### Escalation Triggers
**ESCALATE WHEN:**
1. **Max Attempts Reached**
```python
if attempt >= max_attempts:
escalate_to_human("Max retry attempts exhausted")
```
2. **No Progress**
```python
if same_errors_for_n_attempts(3):
escalate_to_human("No progress after 3 attempts")
```
3. **Degradation**
```python
if error_count_increasing():
escalate_to_human("Situation getting worse, not better")
```
4. **Unknown Error Pattern**
```python
if not error_matcher.has_pattern(error):
escalate_to_human("Unknown error pattern encountered")
```
5. **Resource Exhaustion**
```python
if time_elapsed > timeout or cost > budget:
escalate_to_human("Resource limits exceeded")
```
6. **External Dependency Failure**
```python
if error.is_external_service_failure():
escalate_to_human("External service unavailable")
```
7. **Security Concern**
```python
if error.indicates_security_issue():
escalate_to_human("Potential security issue detected", priority="high")
```
### Escalation Implementation
```python
class EscalationManager:
"""Manage escalation to human oversight."""
def __init__(self, notification_service):
self.notifications = notification_service
self.escalation_history = []
def should_escalate(self, retry_context):
"""Determine if escalation is needed."""
checks = [
self.check_max_attempts(retry_context),
self.check_no_progress(retry_context),
self.check_degradation(retry_context),
self.check_unknown_errors(retry_context),
self.check_resources(retry_context),
]
return any(checks)
def escalate(self, reason, context):
"""Escalate to human with full context."""
escalation = Escalation(
reason=reason,
timestamp=time.time(),
task=context.task,
attempts=context.attempts,
error_history=context.error_history,
last_validation=context.last_validation,
suggested_actions=self.suggest_human_actions(reason)
)
# Notify human
self.notifications.send_escalation(escalation)
# Log escalation
self.escalation_history.append(escalation)
# Wait for human input or timeout
response = self.await_human_response(
escalation,
timeout=3600 # 1 hour
)
return response
def suggest_human_actions(self, reason):
"""Suggest actions for human."""
suggestions = {
"max_attempts": [
"Review error patterns",
"Simplify task requirements",
"Try different approach manually"
],
"no_progress": [
"Check if requirements are achievable",
"Review error messages for clues",
"Consider breaking task into smaller pieces"
],
"unknown_error": [
"Research error message",
"Check external dependencies",
"Review recent changes"
]
}
return suggestions.get(reason, ["Review situation manually"])
```
---
## LEARNING FROM CORRECTIONS
### Pattern Storage
```python
class CorrectionPatternStore:
"""Store successful corrections for future use."""
def __init__(self, storage_path="corrections.json"):
self.storage_path = storage_path
self.patterns = self.load_patterns()
def store_correction(self, error, fix, result):
"""Store a successful correction."""
pattern = CorrectionPattern(
error_signature=self.create_signature(error),
error_message=error.message,
fix_applied=fix,
validation_result=result,
timestamp=time.time(),
success_count=1
)
# Check if similar pattern exists
existing = self.find_similar(pattern)
if existing:
# Increment success count
existing.success_count += 1
existing.last_used = time.time()
else:
# Add new pattern
self.patterns.append(pattern)
self.save_patterns()
def find_correction(self, error):
"""Find stored correction for similar error."""
signature = self.create_signature(error)
# Find matching patterns
matches = [
p for p in self.patterns
if self.signatures_match(p.error_signature, signature)
]
if not matches:
return None
# Return most successful pattern
return max(matches, key=lambda p: p.success_count)
def create_signature(self, error):
"""Create unique signature for error type."""
return {
"type": error.type,
"file_ext": Path(error.file).suffix if error.file else None,
"error_keywords": self.extract_keywords(error.message)
}
```
### Continuous Improvement
```python
class SelfImprovingAgent:
"""Agent that learns from corrections over time."""
def __init__(self, pattern_store):
self.pattern_store = pattern_store
self.performance_metrics = PerformanceTracker()
def execute_with_learning(self, task):
"""Execute with learning from past corrections."""
start_time = time.time()
# Check if we've seen similar task before
similar_corrections = self.pattern_store.find_similar_task(task)
if similar_corrections:
# Apply learned patterns proactively
task = task.with_preemptive_fixes(similar_corrections)
# Execute normally
result = self.execute_with_retry(task)
# Track performance
duration = time.time() - start_time
self.performance_metrics.record(
task_type=task.type,
duration=duration,
attempts=result.attempts,
success=True
)
return result
def analyze_improvements(self):
"""Analyze how agent is improving over time."""
metrics = self.performance_metrics.get_trends()
return ImprovementAnalysis(
average_attempts_trend=metrics.attempts_trend, # Should decrease
success_rate_trend=metrics.success_rate_trend, # Should increase
time_to_success_trend=metrics.duration_trend, # Should decrease
patterns_learned=len(self.pattern_store.patterns)
)
```
---
## COMPLETE SELF-CORRECTING AGENT
### Full Implementation
```python
class SelfCorrectingAgent:
"""Complete self-correcting agent implementation."""
def __init__(
self,
executor,
validator,
analyzer,
pattern_matcher,
retry_strategy,
escalation_manager
):
self.executor = executor
self.validator = validator
self.analyzer = analyzer
self.pattern_matcher = pattern_matcher
self.retry_strategy = retry_strategy
self.escalation_manager = escalation_manager
def execute_task(self, task):
"""
Execute task with full self-correction loop.
Returns:
Validated result or raises escalation exception
"""
context = ExecutionContext(
task=task,
attempts=0,
error_history=[],
start_time=time.time()
)
while context.attempts < self.retry_strategy.max_attempts:
try:
# EXECUTE
result = self.executor.execute(
task=task,
context=context
)
# VALIDATE
validation = self.validator.validate(result)
if validation.passed:
# Success!
return self.create_success_result(result, context)
# INTERPRET
parsed_errors = self.validator.parse_errors(validation)
# ANALYZE
analysis = self.analyzer.analyze(parsed_errors)
# CHECK KNOWN PATTERNS
known_fix = self.pattern_matcher.find_correction(
parsed_errors[0] if parsed_errors else None
)
if known_fix:
# Apply known fix
task = task.apply_fix(known_fix)
else:
# Generate new fix
task = task.with_error_context(analysis)
# Update context
context.attempts += 1
context.error_history.append({
"validation": validation,
"analysis": analysis,
"applied_fix": known_fix
})
# Check if should escalate
if self.escalation_manager.should_escalate(context):
return self.escalation_manager.escalate(
reason="retry_exhausted",
context=context
)
except Exception as e:
# Unexpected error
context.exceptions.append(e)
if self.is_unrecoverable(e):
return self.escalation_manager.escalate(
reason="unrecoverable_error",
context=context
)
# Max attempts reached
return self.escalation_manager.escalate(
reason="max_attempts",
context=context
)
def create_success_result(self, result, context):
"""Package successful result with metadata."""
return SuccessResult(
result=result,
attempts=context.attempts,
duration=time.time() - context.start_time,
error_history=context.error_history,
patterns_applied=context.patterns_applied
)
```
---
## SUMMARY
**SELF-CORRECTING AGENT COMPONENTS:**
1. **Feedback Interpretation** - Parse validation output
2. **Error Analysis** - Identify root causes
3. **Fix Generation** - Create appropriate corrections
4. **Retry Strategies** - Intelligent retry logic
5. **Escalation Management** - Know when to involve humans
6. **Pattern Learning** - Improve over time
**RETRY STRATEGIES:**
- Simple retry with context
- Adaptive strategy selection
- Incremental fix approach
- Backtracking retry
- Ensemble (multiple agents)
**ESCALATION TRIGGERS:**
- Max attempts reached
- No progress
- Degradation
- Unknown errors
- Resource exhaustion
- Security concerns
**KEY PRINCIPLES:**
- Accumulate context between attempts
- Learn from successful corrections
- Fail fast on unrecoverable errors
- Escalate intelligently
- Track performance metrics
- Continuous improvement
**RESULT:**
Autonomous agents that handle the vast majority of error corrections without human intervention, only escalating truly ambiguous or impossible cases.