InteractiveBenefits and Challenges of Multi-Agent Systems: When Complexity is Worth It

## Using GPT-5 for a generalist customer service agent
from openai import OpenAI
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

def generalist_agent(customer_query):
    """
    A single agent that tries to handle all customer service tasks.
    """
    system_prompt = """You are a customer service agent. Handle:
    - Technical support questions
    - Billing inquiries
    - Product recommendations
    - Returns and refunds
    - Account management
    
    Be helpful and professional."""
    
    response = client.chat.completions.create(
        model="gpt-5",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": customer_query}
        ]
    )
    
    return response.choices[0].message.content

## Example queries
print("Query 1:", generalist_agent("My payment failed but I was still charged"))
print("\nQuery 2:", generalist_agent("Which laptop is best for video editing?"))
print("\nQuery 3:", generalist_agent("How do I reset my password?"))

## Using Claude Sonnet 4.5 for specialized customer service agents
from anthropic import Anthropic
client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

class SpecializedCustomerService:
    """
    Customer service system with specialized agents.
    """
    def __init__(self):
        self.model = "claude-sonnet-4-5"
    
    def router_agent(self, query):
        """
        Routes queries to the appropriate specialist.
        """
        system_prompt = """You are a routing specialist. Categorize customer queries:
        - technical: password resets, login issues, bugs
        - billing: payments, charges, refunds, invoices
        - products: recommendations, specifications, comparisons
        - returns: return requests, warranty claims, exchanges
        
        Return only the category name, nothing else."""
        
        response = client.messages.create(
            model=self.model,
            max_tokens=50,
            system=system_prompt,
            messages=[{"role": "user", "content": query}]
        )
        
        return response.content[0].text.strip().lower()
    
    def technical_agent(self, query):
        """
        Specialist in technical support.
        """
        system_prompt = """You are a technical support specialist.
        You have deep knowledge of:
        - Authentication systems and password resets
        - Common technical issues and troubleshooting
        - System requirements and compatibility
        
        Provide clear, step-by-step technical guidance."""
        
        response = client.messages.create(
            model=self.model,
            max_tokens=512,
            system=system_prompt,
            messages=[{"role": "user", "content": query}]
        )
        
        return response.content[0].text
    
    def billing_agent(self, query):
        """
        Specialist in billing and payments.
        """
        system_prompt = """You are a billing specialist.
        You have deep knowledge of:
        - Payment processing and failed transactions
        - Refund policies and procedures
        - Invoice questions and billing disputes
        
        Be empathetic and clear about financial matters."""
        
        response = client.messages.create(
            model=self.model,
            max_tokens=512,
            system=system_prompt,
            messages=[{"role": "user", "content": query}]
        )
        
        return response.content[0].text
    
    def product_agent(self, query):
        """
        Specialist in product recommendations.
        """
        system_prompt = """You are a product specialist.
        You have deep knowledge of:
        - Product specifications and features
        - Use case matching and recommendations
        - Competitive comparisons
        
        Help customers find the right product for their needs."""
        
        response = client.messages.create(
            model=self.model,
            max_tokens=512,
            system=system_prompt,
            messages=[{"role": "user", "content": query}]
        )
        
        return response.content[0].text
    
    def handle_query(self, query):
        """
        Route and handle a customer query.
        """
        # Determine the right specialist
        category = self.router_agent(query)
        print(f"Routing to: {category} specialist")
        
        # Delegate to the specialist
        if "technical" in category:
            return self.technical_agent(query)
        elif "billing" in category:
            return self.billing_agent(query)
        elif "product" in category:
            return self.product_agent(query)
        else:
            return "I'll connect you with the right specialist."

## Example usage
service = SpecializedCustomerService()

print("=== Customer Service with Specialized Agents ===\n")

queries = [
    "My payment failed but I was still charged",
    "Which laptop is best for video editing?",
    "How do I reset my password?"
]

for query in queries:
    print(f"\nCustomer: {query}")
    response = service.handle_query(query)
    print(f"Agent: {response[:100]}...")  # Truncate for readability

## Using Claude Sonnet 4.5 for parallel agent execution
import os
from anthropic import Anthropic
import concurrent.futures
import time
client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

def flight_agent(destination, dates):
    """
    Researches flight options.
    """
    start = time.time()
    
    system_prompt = """You are a flight research specialist.
    Find the best flight options considering price, duration, and convenience."""
    
    query = f"Find flights to {destination} for {dates}"
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        system=system_prompt,
        messages=[{"role": "user", "content": query}]
    )
    
    elapsed = time.time() - start
    return {
        "agent": "flights",
        "result": response.content[0].text,
        "time": elapsed
    }

def hotel_agent(destination, dates):
    """
    Researches hotel options.
    """
    start = time.time()
    
    system_prompt = """You are a hotel research specialist.
    Find the best hotel options considering location, amenities, and value."""
    
    query = f"Find hotels in {destination} for {dates}"
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        system=system_prompt,
        messages=[{"role": "user", "content": query}]
    )
    
    elapsed = time.time() - start
    return {
        "agent": "hotels",
        "result": response.content[0].text,
        "time": elapsed
    }

def activities_agent(destination, interests):
    """
    Researches activities and attractions.
    """
    start = time.time()
    
    system_prompt = """You are a local activities specialist.
    Recommend activities, restaurants, and attractions based on interests."""
    
    query = f"Recommend activities in {destination} for someone interested in {interests}"
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        system=system_prompt,
        messages=[{"role": "user", "content": query}]
    )
    
    elapsed = time.time() - start
    return {
        "agent": "activities",
        "result": response.content[0].text,
        "time": elapsed
    }

## Sequential execution (single agent approach)
def plan_trip_sequential(destination, dates, interests):
    """
    Plan a trip with one agent doing everything sequentially.
    """
    print("=== Sequential Planning ===")
    total_start = time.time()
    
    results = []
    results.append(flight_agent(destination, dates))
    results.append(hotel_agent(destination, dates))
    results.append(activities_agent(destination, interests))
    
    total_time = time.time() - total_start
    
    for r in results:
        print(f"{r['agent']}: {r['time']:.2f}s")
    print(f"Total time: {total_time:.2f}s\n")
    
    return results

## Parallel execution (multi-agent approach)
def plan_trip_parallel(destination, dates, interests):
    """
    Plan a trip with multiple agents working simultaneously.
    """
    print("=== Parallel Planning ===")
    total_start = time.time()
    
    # Execute all agents concurrently
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        flight_future = executor.submit(flight_agent, destination, dates)
        hotel_future = executor.submit(hotel_agent, destination, dates)
        activities_future = executor.submit(activities_agent, destination, interests)
        
        # Wait for all to complete
        results = [
            flight_future.result(),
            hotel_future.result(),
            activities_future.result()
        ]
    
    total_time = time.time() - total_start
    
    for r in results:
        print(f"{r['agent']}: {r['time']:.2f}s")
    print(f"Total time: {total_time:.2f}s\n")
    
    return results

## Compare both approaches
sequential_results = plan_trip_sequential("Tokyo", "March 15-22", "food and history")
parallel_results = plan_trip_parallel("Tokyo", "March 15-22", "food and history")

## Calculate speedup
seq_time = sum(r['time'] for r in sequential_results)
par_time = max(r['time'] for r in parallel_results)
speedup = seq_time / par_time

print(f"Speedup: {speedup:.2f}x faster with parallel agents")

## Using Claude Sonnet 4.5 for agent verification
import os
from anthropic import Anthropic
import json

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

def research_agent(topic):
    """
    Researches a topic and provides findings.
    """
    system_prompt = """You are a research agent. Research the topic and provide factual information.
    Return your findings as JSON with:
    - claims: list of factual claims you're making
    - confidence: your confidence level (0-1) for each claim
    - sources: where this information comes from"""
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": f"Research: {topic}"}]
    )
    
    return response.content[0].text

def verification_agent(research_findings):
    """
    Verifies research findings for accuracy and completeness.
    """
    system_prompt = """You are a fact-checking agent. Review research findings and:
    - Check if claims are well-supported
    - Identify any potential errors or inconsistencies
    - Suggest additional information needed
    - Rate overall reliability
    
    Return JSON with:
    - verified_claims: claims that seem accurate
    - questionable_claims: claims that need more verification
    - missing_information: important gaps in the research
    - overall_confidence: your confidence in the research (0-1)"""
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": f"Verify this research:\n\n{research_findings}"}]
    )
    
    return response.content[0].text

def synthesis_agent(research, verification):
    """
    Synthesizes verified information into a final answer.
    """
    system_prompt = """You are a synthesis agent. Combine research and verification to create a final answer.
    Include only well-verified information. Acknowledge uncertainties.
    Be clear about confidence levels."""
    
    context = f"Research:\n{research}\n\nVerification:\n{verification}"
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": f"Synthesize:\n\n{context}"}]
    )
    
    return response.content[0].text

## Example: Research with verification
def research_with_verification(topic):
    """
    Multi-agent research with built-in verification.
    """
    print(f"=== Researching: {topic} ===\n")
    
    # Step 1: Initial research
    print("Research Agent working...")
    research = research_agent(topic)
    print(f"Research completed:\n{research[:200]}...\n")
    
    # Step 2: Verification
    print("Verification Agent checking...")
    verification = verification_agent(research)
    print(f"Verification completed:\n{verification[:200]}...\n")
    
    # Step 3: Final synthesis
    print("Synthesis Agent combining results...")
    final_answer = synthesis_agent(research, verification)
    print(f"Final Answer:\n{final_answer}")
    
    return final_answer

## Run the verified research
result = research_with_verification(
    "What are the health benefits of intermittent fasting?"
)

## Using Claude Sonnet 4.5 to demonstrate coordination challenges
import os
from anthropic import Anthropic
import time

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

class ReportWritingTeam:
    """
    Three agents collaborating on a report (with potential coordination issues).
    """
    def __init__(self):
        self.model = "claude-sonnet-4-5"
        self.shared_state = {
            "outline": None,
            "sections": {},
            "final_report": None
        }
    
    def outlining_agent(self, topic):
        """
        Creates a report outline.
        """
        system_prompt = """You are an outlining specialist.
        Create a clear outline for a report on the given topic.
        Return a simple numbered list of sections."""
        
        response = client.messages.create(
            model=self.model,
            max_tokens=512,
            system=system_prompt,
            messages=[{"role": "user", "content": f"Create outline for: {topic}"}]
        )
        
        self.shared_state["outline"] = response.content[0].text
        return self.shared_state["outline"]
    
    def writing_agent(self, section_number):
        """
        Writes a specific section of the report.
        """
        # Problem: What if the outline isn't ready yet?
        outline = self.shared_state.get("outline")
        if not outline:
            return "ERROR: No outline available yet!"
        
        system_prompt = f"""You are a writing specialist.
        Write section {section_number} based on this outline:\n\n{outline}"""
        
        response = client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=system_prompt,
            messages=[{"role": "user", "content": f"Write section {section_number}"}]
        )
        
        section_text = response.content[0].text
        self.shared_state["sections"][section_number] = section_text
        return section_text
    
    def editing_agent(self):
        """
        Edits and finalizes the report.
        """
        # Problem: What if sections aren't ready yet?
        sections = self.shared_state.get("sections")
        if not sections:
            return "ERROR: No sections to edit yet!"
        
        combined = "\n\n".join([
            f"Section {num}:\n{text}" 
            for num, text in sorted(sections.items())
        ])
        
        system_prompt = """You are an editing specialist.
        Review and polish this report for clarity and flow."""
        
        response = client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=system_prompt,
            messages=[{"role": "user", "content": combined}]
        )
        
        self.shared_state["final_report"] = response.content[0].text
        return self.shared_state["final_report"]

## Example: What happens with poor coordination?
def write_report_poor_coordination(topic):
    """
    Demonstrates coordination problems when agents aren't synchronized.
    """
    team = ReportWritingTeam()
    
    print("=== Poor Coordination Example ===\n")
    
    # Problem: Starting all agents at once without coordination
    import concurrent.futures
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        # All agents start simultaneously
        outline_future = executor.submit(team.outlining_agent, topic)
        writing_future = executor.submit(team.writing_agent, 1)
        editing_future = executor.submit(team.editing_agent)
        
        # Results
        outline = outline_future.result()
        section = writing_future.result()
        final = editing_future.result()
    
    print(f"Outline: {outline[:100]}...")
    print(f"Section: {section[:100]}...")
    print(f"Final: {final[:100]}...")
    print("\nNotice the errors: writing and editing agents failed because they started before outline was ready!")

## Example: Better coordination
def write_report_good_coordination(topic):
    """
    Demonstrates proper coordination with sequencing.
    """
    team = ReportWritingTeam()
    
    print("\n=== Good Coordination Example ===\n")
    
    # Step 1: Outline first
    print("Step 1: Creating outline...")
    outline = team.outlining_agent(topic)
    print(f"Outline ready: {outline[:100]}...\n")
    
    # Step 2: Write sections (could be parallel if multiple sections)
    print("Step 2: Writing sections...")
    section = team.writing_agent(1)
    print(f"Section complete: {section[:100]}...\n")
    
    # Step 3: Edit the complete report
    print("Step 3: Editing final report...")
    final = team.editing_agent()
    print(f"Final report: {final[:100]}...\n")
    
    print("Success: Proper sequencing avoided coordination errors!")

## Demonstrate both approaches
write_report_poor_coordination("The Future of Renewable Energy")
write_report_good_coordination("The Future of Renewable Energy")

## Using Claude Sonnet 4.5 to demonstrate communication misunderstandings
import os
from anthropic import Anthropic
import json

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

def data_agent():
    """
    Agent that provides data (but in an ambiguous format).
    """
    system_prompt = """You are a data collection agent.
    Provide the requested data in a clear format."""
    
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=256,
        system=system_prompt,
        messages=[{"role": "user", "content": "Provide the quarterly revenue figures"}]
    )
    
    return response.content[0].text

def analysis_agent(data):
    """
    Agent that analyzes data (expecting specific format).
    """
    system_prompt = """You are a data analysis agent.
    You receive data as JSON with fields: q1, q2, q3, q4 (all numbers).
    Calculate the total and average."""
    
    # Problem: What if data isn't in the expected format?
    try:
        data_dict = json.loads(data)
        total = sum(data_dict.values())
        average = total / len(data_dict)
        return f"Total: ${total:,.0f}, Average: ${average:,.0f}"
    except:
        return "ERROR: Could not parse data. Expected JSON format with quarterly numbers."

## Demonstrate the communication issue
print("=== Communication Misunderstanding ===\n")

data = data_agent()
print(f"Data Agent provided:\n{data}\n")

result = analysis_agent(data)
print(f"Analysis Agent result:\n{result}\n")

print("Problem: If Data Agent didn't return strict JSON, Analysis Agent fails!")

## Using Claude Sonnet 4.5 for a well-designed multi-agent system
import os
from anthropic import Anthropic
import json
from datetime import datetime

client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

class BalancedMultiAgentSystem:
    """
    A multi-agent system with clear interfaces and error handling.
    """
    def __init__(self):
        self.model = "claude-sonnet-4-5"
        self.log = []
    
    def _log_event(self, agent, event, details=None):
        """
        Centralized logging for observability.
        """
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "agent": agent,
            "event": event,
            "details": details
        }
        self.log.append(entry)
        print(f"[{agent}] {event}")
    
    def coordinator(self, user_request):
        """
        Coordinates the workflow with clear error handling.
        """
        self._log_event("coordinator", "Request received", user_request)
        
        try:
            # Step 1: Understand intent
            intent = self.intent_agent(user_request)
            if intent.get("error"):
                return self._handle_error("intent", intent["error"])
            
            # Step 2: Gather information
            research = self.research_agent(intent["topic"])
            if research.get("error"):
                return self._handle_error("research", research["error"])
            
            # Step 3: Formulate response
            response = self.response_agent(research, intent)
            if response.get("error"):
                return self._handle_error("response", response["error"])
            
            self._log_event("coordinator", "Request completed successfully")
            return response
            
        except Exception as e:
            self._log_event("coordinator", "Unexpected error", str(e))
            return {"error": "System error occurred", "details": str(e)}
    
    def intent_agent(self, request):
        """
        Understands user intent with structured output.
        """
        self._log_event("intent_agent", "Processing intent")
        
        try:
            system_prompt = """Extract the intent from user requests.
            Return JSON with:
            - intent_type: "question", "task", or "command"
            - topic: the main topic
            - details: any specific requirements
            
            Only return JSON, nothing else."""
            
            response = client.messages.create(
                model=self.model,
                max_tokens=256,
                system=system_prompt,
                messages=[{"role": "user", "content": request}]
            )
            
            intent = json.loads(response.content[0].text)
            self._log_event("intent_agent", "Intent extracted", intent.get("intent_type"))
            return intent
            
        except Exception as e:
            self._log_event("intent_agent", "Failed", str(e))
            return {"error": str(e)}
    
    def research_agent(self, topic):
        """
        Researches the topic with error handling.
        """
        self._log_event("research_agent", "Researching", topic)
        
        try:
            system_prompt = """Research the given topic and provide key information.
            Return JSON with:
            - summary: brief overview
            - key_points: list of main points
            - confidence: 0-1 confidence score
            
            Only return JSON, nothing else."""
            
            response = client.messages.create(
                model=self.model,
                max_tokens=512,
                system=system_prompt,
                messages=[{"role": "user", "content": f"Research: {topic}"}]
            )
            
            research = json.loads(response.content[0].text)
            self._log_event("research_agent", "Research completed", 
                          f"confidence: {research.get('confidence')}")
            return research
            
        except Exception as e:
            self._log_event("research_agent", "Failed", str(e))
            return {"error": str(e)}
    
    def response_agent(self, research, intent):
        """
        Formulates the final response.
        """
        self._log_event("response_agent", "Formulating response")
        
        try:
            system_prompt = """Create a clear, helpful response based on research and intent.
            Be concise and directly address the user's needs."""
            
            context = f"Intent: {json.dumps(intent)}\n\nResearch: {json.dumps(research)}"
            
            response = client.messages.create(
                model=self.model,
                max_tokens=512,
                system=system_prompt,
                messages=[{"role": "user", "content": context}]
            )
            
            self._log_event("response_agent", "Response created")
            return {"response": response.content[0].text}
            
        except Exception as e:
            self._log_event("response_agent", "Failed", str(e))
            return {"error": str(e)}
    
    def _handle_error(self, agent, error):
        """
        Graceful error handling.
        """
        self._log_event("coordinator", f"Handling error from {agent}")
        return {
            "response": f"I encountered an issue while processing your request. Could you try rephrasing?",
            "internal_error": error
        }
    
    def get_log(self):
        """
        Return the execution log for debugging.
        """
        return self.log

## Example usage
system = BalancedMultiAgentSystem()

print("=== Balanced Multi-Agent System ===\n")

result = system.coordinator("What are the main benefits of renewable energy?")
print(f"\nFinal Response: {result.get('response')}")

print("\n=== Execution Log ===")
for entry in system.get_log():
    print(f"{entry['timestamp']} | {entry['agent']}: {entry['event']}")