Handling Partial Success: Degraded Mode in Multi-Agent Orchestration

In production multi-agent systems, total success is a luxury. Components fail, APIs timeout, and dependencies become unavailable. The real question isn’t whether failures will occur, but how your system behaves when they do.

The Problem with All-or-Nothing Thinking

Traditional software design often assumes binary outcomes: success or failure. In multi-agent orchestration, this assumption breaks down catastrophically. When you have five agents working on a complex task and one fails, should the entire workflow fail?

Consider a customer support system where:

• Agent 1 successfully retrieves customer data
• Agent 2 fetches order history
• Agent 3 fails to access the shipping API
• Agent 4 generates a response
• Agent 5 formats the output

If we treat this as all-or-nothing, we discard valuable work from four agents because one dependency failed. The customer gets nothing instead of a partial answer.

Designing for Partial Success

Define Your Success Algebra

Before writing code, establish a clear success taxonomy:

from enum import Enum
from dataclasses import dataclass
from typing import List, Set

class SuccessLevel(Enum):
    VERIFIED_OK = "verified_ok"          # All mandatory predicates true
    DEGRADED_OK = "degraded_ok"          # Mandatory subset true, gaps documented
    PARTIAL_APPLY = "partial_apply"      # Some side effects landed
    ABORTED_CLEAN = "aborted_clean"      # Stopped before irreversible actions
    FAILED = "failed"                     # No meaningful progress

@dataclass
class MissionSpec:
    mission_id: str
    mandatory: Set[str]    # All required for verified_ok
    optional: Set[str]     # May land in degraded_ok
    abort_safe_state: str  # What to do if mid-mission fail

The key insight: mandatory predicates must be explicitly defined. Don’t let agents demote failed steps to “optional” mid-execution.

Implement Circuit Breakers at the Right Granularity

Circuit breakers should be scoped to individual dependencies, not entire workflows:

class DependencyCircuitBreaker:
    def __init__(self, dependency_name: str, failure_threshold: int = 3):
        self.dependency_name = dependency_name
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.state = "closed"  # closed, open, half-open
        
    def record_failure(self):
        self.failure_count += 1
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
            
    def record_success(self):
        self.failure_count = 0
        self.state = "closed"
        
    def allow_request(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            return False
        # half-open: allow one request
        return True

When a circuit breaker trips, the system should enter degraded mode for that specific dependency, not fail the entire workflow.

Degraded Mode Patterns

1. Graceful Feature Omission

When a non-critical component fails, continue without its output:

async def process_with_degradation(task):
    results = {}
    
    # Critical: must succeed
    results['customer_data'] = await fetch_customer_data(task.customer_id)
    
    # Important: try, but continue if fails
    try:
        results['recommendations'] = await get_recommendations(task.customer_id)
    except ServiceUnavailable:
        results['recommendations'] = []
        results['degradation_flags'].append('recommendations_unavailable')
    
    # Optional: skip if slow or failing
    try:
        results['personalization'] = await get_personalization(task.customer_id)
    except (TimeoutError, ServiceUnavailable):
        results['personalization'] = None
        
    return results

2. Parallel Fallback Paths

Run multiple strategies simultaneously, use the fastest successful result:

import asyncio
from typing import Any

async def parallel_with_fallback(primary_func, fallback_func, timeout=5.0):
    """Run primary and fallback in parallel, return first success."""
    async def run_with_timeout(func, timeout):
        try:
            return await asyncio.wait_for(func(), timeout=timeout)
        except asyncio.TimeoutError:
            return None
    
    # Start both tasks
    primary_task = asyncio.create_task(run_with_timeout(primary_func, timeout))
    fallback_task = asyncio.create_task(run_with_timeout(fallback_func, timeout))
    
    # Wait for first success
    done, pending = await asyncio.wait(
        [primary_task, fallback_task],
        return_when=asyncio.FIRST_COMPLETED
    )
    
    # Cancel pending tasks
    for task in pending:
        task.cancel()
    
    # Return first successful result
    for task in done:
        if task.exception() is None:
            return task.result()
    
    raise RuntimeError("All paths failed")

3. Compensating Transactions

When partial failure leaves inconsistent state, implement rollback:

class WorkflowWithCompensation:
    def __init__(self):
        self.compensation_handlers = {}
        
    def register_compensation(self, step_name: str, handler):
        self.compensation_handlers[step_name] = handler
        
    async def execute_with_compensation(self, steps):
        completed_steps = []
        
        for step in steps:
            try:
                result = await step.execute()
                completed_steps.append((step, result))
            except Exception as e:
                # Compensation in reverse order
                for completed_step, _ in reversed(completed_steps):
                    if completed_step.name in self.compensation_handlers:
                        await self.compensation_handlers[completed_step.name]()
                
                raise WorkflowFailed(
                    failed_step=step,
                    error=e,
                    compensated_steps=[s.name for s, _ in completed_steps]
                )

Reporting Degraded Results

Transparent Communication

Never hide degradation from users or downstream systems:

@dataclass
class DegradedResult:
    data: Any
    success_level: SuccessLevel
    achieved_predicates: Set[str]
    missing_predicates: Set[str]
    degradation_reasons: List[str]
    
    def to_response(self) -> dict:
        return {
            "data": self.data,
            "status": self.success_level.value,
            "achievements": list(self.achieved_predicates),
            "gaps": list(self.missing_predicates),
            "warnings": self.degradation_reasons
        }

Metrics That Matter

Track these metrics to understand your system’s degradation patterns:

class DegradationMetrics:
    def __init__(self):
        self.degradation_rate = 0.0
        self.cascade_depth = 0
        self.recovery_success_rate = 0.0
        
    def record_degradation(self, workflow_id: str, degraded_components: List[str]):
        # Track how often degradation occurs
        self.degradation_rate = self._update_rate(self.degradation_rate)
        
        # Track how failures propagate
        self.cascade_depth = max(self.cascade_depth, len(degraded_components))
        
    def record_recovery(self, workflow_id: str, recovered: bool):
        # Track if degraded mode actually helps
        self.recovery_success_rate = self._update_recovery_rate(recovered)

Production Considerations

1. Define Clear Boundaries

Every agent should have explicit boundaries:

• What it can decide autonomously
• When it must escalate to a manager
• What constitutes success vs. degradation

2. Version Your Success Criteria

As your system evolves, so do your success definitions:

# mission_spec_v1.yaml
mission_id: "customer_support_response"
mandatory:
  - customer_identified
  - order_retrieved
  - response_generated
optional:
  - recommendations_provided
  - personalization_applied

# mission_spec_v2.yaml - after learning from production
mission_id: "customer_support_response"
mandatory:
  - customer_identified
  - order_retrieved
  - response_generated
  - shipping_status_checked  # Promoted to mandatory
optional:
  - recommendations_provided
  - personalization_applied

3. Test Degradation Paths

Your test suite should include explicit degradation scenarios:

async def test_degraded_mode_when_recommendation_service_fails():
    # Mock recommendation service failure
    with mock_service_failure('recommendation_service'):
        result = await workflow.execute(test_task)
        
    assert result.success_level == SuccessLevel.DEGRADED_OK
    assert 'customer_identified' in result.achieved_predicates
    assert 'recommendations_provided' in result.missing_predicates
    assert result.data['recommendations'] == []

Conclusion

Partial success isn’t a compromise—it’s a realistic approach to building resilient multi-agent systems. By defining clear success criteria, implementing granular circuit breakers, and transparently reporting degradation, you create systems that deliver maximum value even under imperfect conditions.

The key takeaways:

1. Define success explicitly - Don’t let agents decide what’s optional mid-execution
2. Isolate failures - Circuit breakers at dependency level, not workflow level
3. Communicate degradation - Users deserve to know what’s working and what’s not
4. Test degradation paths - Your happy path tests aren’t enough
5. Track degradation metrics - You can’t improve what you don’t measure

In the real world, “80% working” is better than “100% failed.” Design your systems accordingly.