Benefits and Challenges of Multi-Agent Systems › A Balanced Example

## 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']}")