Merge pull request #712 from MervinPraison/claude/issue-491-20250628_070020

feat: implement agent-to-agent handoffs

Author: MervinPraison

examples/python/handoff/handoff_advanced.py

@@ -0,0 +1,141 @@
+"""
+Advanced handoff example with callbacks, input types, and filters.
+
+This example demonstrates advanced handoff features including:
+- Custom handoff callbacks
+- Structured input data
+- Input filters
+- Custom tool names and descriptions
+"""
+
+from praisonaiagents import Agent, handoff, handoff_filters, prompt_with_handoff_instructions
+from pydantic import BaseModel
+from typing import Optional
+import logging
+
+logging.basicConfig(level=logging.INFO)
+
+
+# Define structured input for escalation
+class EscalationData(BaseModel):
+    reason: str
+    priority: Optional[str] = "normal"
+    customer_sentiment: Optional[str] = None
+
+
+# Callback function for logging handoffs
+def log_handoff(source_agent: Agent):
+    """Log when a handoff occurs"""
+    logging.info(f"Handoff initiated from {source_agent.name}")
+
+
+# Callback function with input data
+def log_escalation(source_agent: Agent, input_data: EscalationData):
+    """Log escalation with structured data"""
+    logging.info(f"ESCALATION from {source_agent.name}: {input_data.reason} (Priority: {input_data.priority})")
+
+
+# Create specialized agents
+faq_agent = Agent(
+    name="FAQ Agent",
+    role="FAQ Specialist",
+    goal="Answer frequently asked questions using knowledge base",
+    backstory="I have access to comprehensive FAQ documentation and can quickly provide accurate answers."
+)
+
+escalation_agent = Agent(
+    name="Escalation Agent",
+    role="Senior Support Manager",
+    goal="Handle escalated issues that require special attention",
+    backstory="I handle complex cases that need senior-level intervention and decision-making."
+)
+
+# Create support agent with custom handoffs
+support_agent = Agent(
+    name="Support Agent",
+    role="Customer Support Representative",
+    goal="Provide first-line support and escalate when necessary",
+    backstory="I help customers with their issues and know when to involve specialists.",
+    instructions=prompt_with_handoff_instructions(
+        """Help customers with their requests. You should:
+        1. Try to resolve issues yourself first
+        2. Transfer to FAQ Agent for common questions you can't answer
+        3. Escalate to senior management for complex or sensitive issues
+        
+        When escalating, always provide a clear reason and assess the priority.""",
+        None  # Agent will be set later
+    ),
+    handoffs=[
+        # Simple handoff with callback
+        handoff(
+            faq_agent,
+            on_handoff=log_handoff,
+            input_filter=handoff_filters.remove_all_tools  # Remove tool calls from history
+        ),
+        # Advanced handoff with structured input
+        handoff(
+            escalation_agent,
+            tool_name_override="escalate_to_manager",
+            tool_description_override="Escalate complex or sensitive issues to senior management",
+            on_handoff=log_escalation,
+            input_type=EscalationData
+        )
+    ]
+)
+
+# Update the instructions with the agent reference
+support_agent.instructions = prompt_with_handoff_instructions(
+    support_agent.instructions,
+    support_agent
+)
+
+
+# Example with custom input filter
+def custom_filter(data):
+    """Keep only last 3 messages and remove system messages"""
+    data = handoff_filters.keep_last_n_messages(3)(data)
+    data = handoff_filters.remove_system_messages(data)
+    return data
+
+
+# Agent with custom filtered handoff
+filtered_agent = Agent(
+    name="Filtered Support",
+    role="Support with Privacy",
+    goal="Handle support while maintaining privacy",
+    backstory="I ensure customer privacy by filtering sensitive conversation history.",
+    handoffs=[
+        handoff(
+            faq_agent,
+            tool_description_override="Transfer to FAQ (filtered history)",
+            input_filter=custom_filter
+        )
+    ]
+)
+
+# Example usage
+if __name__ == "__main__":
+    print("=== Advanced Handoff Examples ===\n")
+    
+    # Test escalation with structured data
+    print("Test 1: Escalation with structured input")
+    response = support_agent.chat(
+        "I've been waiting for my refund for 3 weeks and I'm very frustrated! "
+        "I need this resolved immediately or I'll take legal action!"
+    )
+    print(f"Response: {response}\n")
+    
+    # Test FAQ handoff
+    print("\nTest 2: FAQ handoff with filtering")
+    response = support_agent.chat(
+        "How do I reset my password? I can't find the option in settings."
+    )
+    print(f"Response: {response}\n")
+    
+    # Test filtered handoff
+    print("\nTest 3: Filtered handoff")
+    # Add some conversation history
+    filtered_agent.chat("My account number is 12345")
+    filtered_agent.chat("I have a sensitive issue")
+    response = filtered_agent.chat("I need help with FAQ about privacy policy")
+    print(f"Response: {response}\n")

examples/python/handoff/handoff_basic.py

@@ -0,0 +1,61 @@
+"""
+Basic handoff example demonstrating agent-to-agent delegation.
+
+This example shows how agents can hand off tasks to specialized agents.
+"""
+
+from praisonaiagents import Agent
+
+# Create specialized agents
+billing_agent = Agent(
+    name="Billing Agent",
+    role="Billing Specialist",
+    goal="Handle all billing-related inquiries and tasks",
+    backstory="I am an expert in billing systems, payment processing, and invoice management."
+)
+
+refund_agent = Agent(
+    name="Refund Agent",
+    role="Refund Specialist",
+    goal="Process refund requests and handle refund-related issues",
+    backstory="I specialize in processing refunds, evaluating refund eligibility, and ensuring customer satisfaction."
+)
+
+technical_support_agent = Agent(
+    name="Technical Support",
+    role="Technical Support Specialist",
+    goal="Resolve technical issues and provide technical assistance",
+    backstory="I am skilled in troubleshooting technical problems and providing solutions."
+)
+
+# Create a triage agent with handoffs to specialized agents
+triage_agent = Agent(
+    name="Triage Agent",
+    role="Customer Service Triage",
+    goal="Understand customer needs and route them to the appropriate specialist",
+    backstory="I analyze customer requests and direct them to the most suitable specialist for efficient resolution.",
+    instructions="""Analyze the customer's request and determine which specialist can best help:
+    - For billing questions, payment issues, or invoices, transfer to the Billing Agent
+    - For refund requests or refund status inquiries, transfer to the Refund Agent  
+    - For technical problems or product issues, transfer to Technical Support
+    
+    Always explain why you're transferring the customer before doing so.""",
+    handoffs=[billing_agent, refund_agent, technical_support_agent]
+)
+
+# Example usage
+if __name__ == "__main__":
+    print("=== Customer Service Triage System ===\n")
+    
+    # Test different types of requests
+    test_requests = [
+        "I need a refund for my last purchase",
+        "Why was I charged twice on my credit card?",
+        "The app keeps crashing when I try to login"
+    ]
+    
+    for request in test_requests:
+        print(f"\nCustomer: {request}")
+        response = triage_agent.chat(request)
+        print(f"Response: {response}")
+        print("-" * 50)

examples/python/handoff/handoff_customer_service.py

@@ -0,0 +1,162 @@
+"""
+Complete customer service workflow example with handoffs.
+
+This example demonstrates a realistic customer service scenario with:
+- Order status checking
+- Refund processing
+- Technical support
+- FAQ handling
+- Escalation management
+"""
+
+from praisonaiagents import Agent, handoff, RECOMMENDED_PROMPT_PREFIX
+import random
+
+
+# Mock functions for demonstration
+def check_order_status(order_id: str) -> str:
+    """Check the status of an order"""
+    statuses = ["shipped", "processing", "delivered", "pending"]
+    return f"Order {order_id} is {random.choice(statuses)}"
+
+
+def process_refund(order_id: str, reason: str) -> str:
+    """Process a refund request"""
+    return f"Refund initiated for order {order_id}. Reason: {reason}. Expected in 3-5 business days."
+
+
+def get_faq_answer(question: str) -> str:
+    """Get answer from FAQ database"""
+    faqs = {
+        "shipping": "Standard shipping takes 5-7 business days. Express shipping takes 2-3 business days.",
+        "returns": "You can return items within 30 days of purchase in original condition.",
+        "warranty": "All products come with a 1-year manufacturer warranty.",
+        "payment": "We accept credit cards, debit cards, PayPal, and Apple Pay."
+    }
+    for key, answer in faqs.items():
+        if key in question.lower():
+            return answer
+    return "I'll need to check our documentation for that specific question."
+
+
+# Create specialized agents with tools
+order_agent = Agent(
+    name="Order Specialist",
+    role="Order Management Specialist",
+    goal="Handle all order-related inquiries including tracking, modifications, and status updates",
+    backstory="I specialize in order management and have access to the order tracking system.",
+    tools=[check_order_status],
+    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
+
+I can help with:
+- Checking order status
+- Tracking shipments
+- Order modifications
+- Delivery issues
+
+For refunds, I'll transfer you to our Refund Specialist.
+For technical issues, I'll connect you with Technical Support."""
+)
+
+refund_agent = Agent(
+    name="Refund Specialist",
+    role="Refund and Returns Specialist",
+    goal="Process refunds and handle return requests efficiently",
+    backstory="I'm authorized to process refunds and handle all return-related matters.",
+    tools=[process_refund],
+    instructions="""I process refunds and returns. I need the order ID and reason for the refund.
+    I ensure all refunds are processed according to our policy."""
+)
+
+faq_agent = Agent(
+    name="FAQ Assistant",
+    role="Knowledge Base Specialist",
+    goal="Provide quick answers to frequently asked questions",
+    backstory="I have access to our comprehensive FAQ database.",
+    tools=[get_faq_answer],
+    instructions="I provide answers to common questions about shipping, returns, warranty, and payments."
+)
+
+technical_agent = Agent(
+    name="Technical Support",
+    role="Technical Support Engineer",
+    goal="Resolve technical issues with products and services",
+    backstory="I'm trained in troubleshooting all our products and technical services.",
+    instructions="""I help with:
+    - Product setup and configuration
+    - Troubleshooting technical issues
+    - Software problems
+    - Hardware diagnostics"""
+)
+
+escalation_agent = Agent(
+    name="Senior Manager",
+    role="Customer Experience Manager",
+    goal="Handle escalated issues and ensure customer satisfaction",
+    backstory="I'm a senior manager with authority to make exceptions and handle complex cases.",
+    instructions="""I handle:
+    - Escalated complaints
+    - Special requests requiring manager approval
+    - Complex issues that need executive decisions
+    - Customer retention situations"""
+)
+
+# Create main customer service agent with handoffs
+customer_service_agent = Agent(
+    name="Customer Service",
+    role="Customer Service Representative",
+    goal="Provide excellent customer service by understanding needs and routing to the right specialist",
+    backstory="I'm your first point of contact and I'll make sure you get the help you need.",
+    instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
+
+Welcome! I'm here to help you today. I can assist with various requests or connect you with the right specialist:
+
+- For order tracking and status → Order Specialist
+- For refunds and returns → Refund Specialist  
+- For common questions → FAQ Assistant
+- For technical problems → Technical Support
+- For complaints or special requests → Senior Manager
+
+How can I help you today?""",
+    handoffs=[
+        order_agent,
+        refund_agent,
+        faq_agent,
+        technical_agent,
+        handoff(
+            escalation_agent,
+            tool_description_override="Escalate to senior management for complex issues or complaints"
+        )
+    ]
+)
+
+# Example interaction
+if __name__ == "__main__":
+    print("=== Customer Service System ===")
+    print("Welcome to our customer service! Type 'quit' to exit.\n")
+    
+    # Simulated customer interactions
+    test_scenarios = [
+        "I need to check the status of my order #12345",
+        "I want a refund for order #67890, the product was damaged",
+        "What's your return policy?",
+        "My device won't turn on after the update",
+        "I'm very unhappy with the service and want to speak to a manager!",
+        "How long does shipping usually take?"
+    ]
+    
+    print("Running automated test scenarios:\n")
+    for scenario in test_scenarios:
+        print(f"Customer: {scenario}")
+        response = customer_service_agent.chat(scenario)
+        print(f"Agent: {response}")
+        print("-" * 80)
+        
+    # Interactive mode
+    print("\nNow entering interactive mode. You can ask questions directly:")
+    while True:
+        user_input = input("\nYou: ")
+        if user_input.lower() == 'quit':
+            break
+        response = customer_service_agent.chat(user_input)
+        print(f"Agent: {response}")