Skip to main content

​
Handoff Configuration

This page provides comprehensive documentation for configuring handoffs in PraisonAI, including handoff filters, delegation strategies, routing rules, and advanced orchestration patterns.

​
Handoff System Overview

The handoff system enables agents to delegate tasks to other agents based on expertise, availability, or specific conditions. This creates flexible multi-agent workflows with intelligent task routing.

​
Basic Handoff Configuration

from praisonaiagents import Agent, Handoff

# Basic handoff configuration
handoff = Handoff(
    name="research_handoff",
    target="research_agent",
    description="Hand off research tasks to specialized researcher",
    condition="task requires research"
)

agent = Agent(
    name="Coordinator",
    handoffs=[handoff],
    handoff_config={
        "strategy": "best_match",
        "timeout": 30,
        "fallback_behavior": "handle_self"
    }
)

​
Handoff Filters

​
Filter Types and Configuration

# Expertise-based filter
expertise_filter = {
    "type": "expertise",
    "required_skills": ["python", "data_analysis"],
    "minimum_skill_level": 0.8,
    "match_all": True  # Require all skills
}

# Availability filter
availability_filter = {
    "type": "availability",
    "max_queue_size": 5,
    "max_active_tasks": 3,
    "estimated_duration": 300,  # 5 minutes
    "check_schedule": True
}

# Performance filter
performance_filter = {
    "type": "performance",
    "min_success_rate": 0.9,
    "max_avg_duration": 120,
    "recent_window": 100,  # Last 100 tasks
    "exclude_outliers": True
}

# Resource filter
resource_filter = {
    "type": "resource",
    "required_memory": 2048,  # MB
    "required_cpu": 2,        # cores
    "required_gpu": False,
    "check_availability": True
}

# Custom filter
custom_filter = {
    "type": "custom",
    "function": "evaluate_agent_compatibility",
    "parameters": {
        "task_complexity": "high",
        "domain": "finance"
    }
}

​
Composite Filters

# Combine multiple filters with AND/OR logic
composite_filter = {
    "type": "composite",
    "operator": "AND",
    "filters": [
        {
            "type": "expertise",
            "required_skills": ["machine_learning"],
            "minimum_skill_level": 0.7
        },
        {
            "type": "composite",
            "operator": "OR",
            "filters": [
                {
                    "type": "availability",
                    "max_queue_size": 3
                },
                {
                    "type": "performance",
                    "min_success_rate": 0.95
                }
            ]
        }
    ]
}

​
Dynamic Filter Functions

def dynamic_agent_filter(agent, task, context):
    """Custom filter function for complex logic"""
    # Check agent capabilities
    if not agent.has_capability(task.required_capability):
        return False
    
    # Check workload
    if agent.current_load > 0.8:
        return False
    
    # Check domain expertise
    domain_score = agent.get_domain_score(task.domain)
    if domain_score < context.get("min_domain_score", 0.7):
        return False
    
    # Check historical performance
    if task.priority == "high":
        success_rate = agent.get_success_rate(task.type)
        if success_rate < 0.9:
            return False
    
    return True

# Use dynamic filter
handoff_config = {
    "filters": [
        {
            "type": "dynamic",
            "function": dynamic_agent_filter,
            "context": {
                "min_domain_score": 0.8
            }
        }
    ]
}

​
Advanced Delegation Settings

​
Delegation Strategies

# Best match strategy - Select most suitable agent
best_match_config = {
    "strategy": "best_match",
    "scoring": {
        "expertise_weight": 0.4,
        "availability_weight": 0.3,
        "performance_weight": 0.3
    },
    "min_score": 0.7,
    "tie_breaker": "least_loaded"
}

# Round robin strategy - Distribute evenly
round_robin_config = {
    "strategy": "round_robin",
    "skip_unavailable": True,
    "sticky_sessions": True,  # Same agent for related tasks
    "session_duration": 3600  # 1 hour
}

# Load balancing strategy
load_balancing_config = {
    "strategy": "load_balance",
    "algorithm": "weighted_least_connections",
    "weights": {
        "agent_1": 3,
        "agent_2": 2,
        "agent_3": 1
    },
    "health_check_interval": 30
}

# Priority-based strategy
priority_config = {
    "strategy": "priority",
    "agent_priorities": {
        "expert_agent": 1,
        "senior_agent": 2,
        "junior_agent": 3
    },
    "escalation_enabled": True,
    "escalation_timeout": 60
}

​
Conditional Handoffs

# Task-based conditions
task_condition_handoff = Handoff(
    name="task_router",
    condition={
        "type": "task_properties",
        "rules": [
            {
                "if": {"task_type": "analysis", "complexity": "high"},
                "then": {"target": "senior_analyst", "priority": "high"}
            },
            {
                "if": {"task_type": "analysis", "complexity": "low"},
                "then": {"target": "junior_analyst", "priority": "normal"}
            },
            {
                "if": {"task_type": "research"},
                "then": {"target": "research_team", "strategy": "round_robin"}
            }
        ],
        "default": {"target": "general_agent"}
    }
)

# Context-based conditions
context_condition_handoff = Handoff(
    name="context_router",
    condition={
        "type": "context_evaluation",
        "evaluator": "evaluate_context",
        "rules": {
            "high_value_customer": {
                "target": "senior_support",
                "sla": 300  # 5 minute response
            },
            "technical_issue": {
                "target": "tech_support",
                "include_context": True
            },
            "billing_query": {
                "target": "billing_team",
                "secure_handoff": True
            }
        }
    }
)

​
Handoff Chains and Workflows

# Sequential handoff chain
sequential_handoff = {
    "type": "sequential",
    "chain": [
        {
            "agent": "validator",
            "action": "validate_input",
            "on_success": "next",
            "on_failure": "abort"
        },
        {
            "agent": "processor",
            "action": "process_data",
            "on_success": "next",
            "on_failure": "retry"
        },
        {
            "agent": "reviewer",
            "action": "review_output",
            "on_success": "complete",
            "on_failure": "escalate"
        }
    ],
    "timeout": 600,
    "preserve_context": True
}

# Parallel handoff
parallel_handoff = {
    "type": "parallel",
    "targets": ["analyzer_1", "analyzer_2", "analyzer_3"],
    "aggregation": "consensus",  # or "merge", "first", "all"
    "min_responses": 2,
    "timeout": 120,
    "continue_on_partial": True
}

# Conditional branching
branching_handoff = {
    "type": "branching",
    "condition": "evaluate_priority",
    "branches": {
        "urgent": {
            "target": "emergency_team",
            "escalate_after": 60
        },
        "high": {
            "target": "senior_team",
            "queue_position": "front"
        },
        "normal": {
            "target": "standard_team",
            "queue_position": "back"
        }
    }
}

​
Handoff Routing Rules

​
Static Routing

static_routing = {
    "routes": {
        "customer_support": ["support_agent_1", "support_agent_2"],
        "technical": ["tech_agent_1", "tech_agent_2", "tech_agent_3"],
        "sales": ["sales_agent_1"],
        "general": ["general_agent_1", "general_agent_2"]
    },
    "selection_method": "least_busy",
    "fallback_route": "general"
}

​
Dynamic Routing

def dynamic_router(task, agents, context):
    """Dynamic routing based on real-time conditions"""
    # Get task requirements
    required_skills = task.get_required_skills()
    urgency = task.get_urgency()
    
    # Score each agent
    scores = {}
    for agent in agents:
        score = 0
        
        # Skill match
        skill_match = agent.match_skills(required_skills)
        score += skill_match * 0.4
        
        # Availability
        availability = agent.get_availability()
        score += availability * 0.3
        
        # Performance history
        performance = agent.get_performance_score(task.type)
        score += performance * 0.3
        
        # Urgency bonus
        if urgency == "high" and agent.priority_capable:
            score += 0.2
        
        scores[agent.name] = score
    
    # Select best agent
    best_agent = max(scores, key=scores.get)
    return best_agent if scores[best_agent] > 0.6 else None

dynamic_routing_config = {
    "router_function": dynamic_router,
    "refresh_interval": 30,  # Re-evaluate every 30 seconds
    "cache_decisions": True,
    "cache_ttl": 60
}

​
Rule-Based Routing

rule_based_routing = {
    "rules": [
        {
            "name": "vip_customer_rule",
            "condition": "customer.tier == 'VIP'",
            "route": "senior_support",
            "priority": 1
        },
        {
            "name": "technical_issue_rule",
            "condition": "issue.category == 'technical' and issue.severity > 7",
            "route": "tech_specialist",
            "priority": 2
        },
        {
            "name": "language_rule",
            "condition": "customer.language != 'english'",
            "route": "multilingual_support",
            "priority": 3
        }
    ],
    "evaluation_order": "priority",  # or "sequential", "parallel"
    "stop_on_match": True
}

​
Handoff Security and Validation

security_config = {
    "authentication": {
        "enabled": True,
        "method": "token",  # or "certificate", "mutual_tls"
        "token_ttl": 3600
    },
    
    "authorization": {
        "enabled": True,
        "check_permissions": True,
        "role_based": True,
        "resource_based": True
    },
    
    "data_handling": {
        "encrypt_in_transit": True,
        "sanitize_sensitive": True,
        "audit_trail": True,
        "data_retention": 90  # days
    },
    
    "validation": {
        "validate_schema": True,
        "validate_constraints": True,
        "max_payload_size": 1048576  # 1MB
    }
}

​
Performance Optimization

​
Caching and Optimization

optimization_config = {
    "caching": {
        "cache_routing_decisions": True,
        "cache_ttl": 300,
        "cache_size": 1000,
        "invalidation_strategy": "ttl"  # or "lru", "event_based"
    },
    
    "batching": {
        "enable_batching": True,
        "batch_size": 10,
        "batch_timeout": 1.0,
        "group_by": ["target_agent", "priority"]
    },
    
    "connection_pooling": {
        "pool_size": 20,
        "max_overflow": 10,
        "pool_timeout": 30,
        "recycle": 3600
    },
    
    "monitoring": {
        "track_latency": True,
        "track_success_rate": True,
        "alert_thresholds": {
            "latency_p95": 1000,  # ms
            "error_rate": 0.05
        }
    }
}

​
Complete Handoff Configuration Example

from praisonaiagents import Agent, Handoff

# Create comprehensive handoff configuration
coordinator = Agent(
    name="MasterCoordinator",
    handoffs=[
        Handoff(
            name="research_handoff",
            target="research_team",
            description="Complex research tasks",
            filters=[
                {
                    "type": "expertise",
                    "required_skills": ["research", "analysis"],
                    "minimum_skill_level": 0.8
                }
            ]
        ),
        Handoff(
            name="urgent_handoff",
            target="emergency_team",
            description="High-priority urgent tasks",
            condition={"urgency": "critical"},
            priority=1
        )
    ],
    handoff_config={
        # Strategy configuration
        "default_strategy": "best_match",
        "fallback_strategy": "round_robin",
        
        # Filter configuration
        "global_filters": [
            {
                "type": "availability",
                "max_queue_size": 10
            }
        ],
        
        # Routing configuration
        "routing": {
            "type": "hybrid",
            "static_routes": {
                "research": ["researcher_1", "researcher_2"],
                "analysis": ["analyst_1", "analyst_2"]
            },
            "dynamic_router": "smart_router_function"
        },
        
        # Performance settings
        "timeout": 60,
        "max_retries": 2,
        "retry_delay": 5,
        
        # Security settings
        "secure_handoffs": True,
        "encrypt_data": True,
        
        # Monitoring
        "track_metrics": True,
        "log_handoffs": True
    }
)

​
Environment Variables

# Handoff strategy
export PRAISONAI_HANDOFF_STRATEGY="best_match"
export PRAISONAI_HANDOFF_TIMEOUT="60"

# Filter settings
export PRAISONAI_HANDOFF_MIN_SCORE="0.7"
export PRAISONAI_HANDOFF_CHECK_AVAILABILITY="true"

# Routing settings
export PRAISONAI_HANDOFF_ROUTING="dynamic"
export PRAISONAI_HANDOFF_CACHE_ROUTES="true"

# Performance settings
export PRAISONAI_HANDOFF_BATCH_SIZE="10"
export PRAISONAI_HANDOFF_MAX_RETRIES="3"

# Security settings
export PRAISONAI_HANDOFF_SECURE="true"
export PRAISONAI_HANDOFF_ENCRYPT="true"

​
Best Practices

  1. Use appropriate filters to ensure tasks are routed to capable agents
  2. Implement fallback mechanisms for handling failures
  3. Monitor handoff performance and adjust strategies accordingly
  4. Cache routing decisions for frequently occurring patterns
  5. Set reasonable timeouts to prevent indefinite waiting
  6. Implement circuit breakers for unreliable agents
  7. Use batching for high-volume scenarios
  8. Maintain audit trails for compliance and debugging

​
See Also