Google Agent Development Kit

from google.cloud import aiplatform
from google.cloud.aiplatform import agent
import json
from typing import Dict, Any, List

class CustomerSupportAgent:
    """Basic customer support agent using Google Agent Development Kit"""
    
    def __init__(self, project_id: str, location: str = "us-central1"):
        self.project_id = project_id
        self.location = location
        
        # Initialize the AI Platform client
        aiplatform.init(project=project_id, location=location)
        
        # Configure the agent
        self.agent_config = {
            "display_name": "Customer Support Agent",
            "description": "AI assistant for customer support inquiries",
            "default_language_code": "en-US",
            "supported_language_codes": ["en-US", "es-ES", "fr-FR"],
            "time_zone": "America/New_York"
        }
        
        # Initialize conversation state
        self.conversation_state = {}
    
    def create_agent(self) -> str:
        """Create and deploy the agent"""
        
        # Define agent capabilities
        agent_definition = {
            "instructions": """
            You are a helpful customer support agent. You can:
            1. Answer product questions
            2. Help with order status
            3. Process returns and refunds
            4. Escalate complex issues to human agents
            
            Always be polite, professional, and helpful.
            If you can't help with something, offer to connect them with a human agent.
            """,
            
            "tools": [
                {
                    "name": "order_lookup",
                    "description": "Look up order information by order ID",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "order_id": {
                                "type": "string",
                                "description": "The customer's order ID"
                            }
                        },
                        "required": ["order_id"]
                    }
                },
                {
                    "name": "process_refund",
                    "description": "Initiate a refund for an order",
                    "parameters": {
                        "type": "object", 
                        "properties": {
                            "order_id": {"type": "string"},
                            "reason": {"type": "string"},
                            "amount": {"type": "number"}
                        },
                        "required": ["order_id", "reason"]
                    }
                },
                {
                    "name": "escalate_to_human",
                    "description": "Escalate the conversation to a human agent",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "issue_summary": {"type": "string"},
                            "urgency": {"type": "string", "enum": ["low", "medium", "high"]}
                        },
                        "required": ["issue_summary"]
                    }
                }
            ]
        }
        
        # Create the agent
        created_agent = agent.Agent.create(
            display_name=self.agent_config["display_name"],
            description=self.agent_config["description"],
            instructions=agent_definition["instructions"],
            tools=agent_definition["tools"]
        )
        
        print(f"Agent created: {created_agent.resource_name}")
        return created_agent.resource_name
    
    def handle_tool_call(self, tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
        """Handle tool function calls"""
        
        if tool_name == "order_lookup":
            return self.lookup_order(parameters["order_id"])
        elif tool_name == "process_refund":
            return self.process_refund(
                parameters["order_id"],
                parameters["reason"],
                parameters.get("amount")
            )
        elif tool_name == "escalate_to_human":
            return self.escalate_to_human(
                parameters["issue_summary"],
                parameters.get("urgency", "medium")
            )
        else:
            return {"error": f"Unknown tool: {tool_name}"}
    
    def lookup_order(self, order_id: str) -> Dict[str, Any]:
        """Mock order lookup function"""
        # In a real implementation, this would query your order database
        mock_orders = {
            "ORD-12345": {
                "order_id": "ORD-12345",
                "status": "Shipped",
                "items": ["Laptop", "Mouse"],
                "total": 1299.99,
                "tracking": "1Z999AA1234567890"
            },
            "ORD-67890": {
                "order_id": "ORD-67890", 
                "status": "Processing",
                "items": ["Phone Case"],
                "total": 29.99,
                "tracking": None
            }
        }
        
        if order_id in mock_orders:
            return mock_orders[order_id]
        else:
            return {"error": "Order not found"}
    
    def process_refund(self, order_id: str, reason: str, amount: float = None) -> Dict[str, Any]:
        """Mock refund processing function"""
        # In a real implementation, this would integrate with your payment system
        order = self.lookup_order(order_id)
        
        if "error" in order:
            return {"error": "Cannot process refund - order not found"}
        
        refund_amount = amount or order.get("total", 0)
        
        # Mock refund processing
        return {
            "refund_id": f"REF-{order_id}-001",
            "amount": refund_amount,
            "status": "Processing",
            "estimated_completion": "3-5 business days"
        }
    
    def escalate_to_human(self, issue_summary: str, urgency: str = "medium") -> Dict[str, Any]:
        """Mock human escalation function"""
        # In a real implementation, this would create a support ticket
        ticket_id = f"TICKET-{int(time.time())}"
        
        return {
            "ticket_id": ticket_id,
            "status": "Escalated to human agent",
            "urgency": urgency,
            "estimated_response": "30 minutes" if urgency == "high" else "2 hours"
        }
    
    def chat(self, message: str, session_id: str = "default") -> str:
        """Process a chat message"""
        # In a real implementation, this would use the Google Agent API
        # For now, we'll simulate basic responses
        
        message_lower = message.lower()
        
        if "order" in message_lower and any(char.isdigit() for char in message):
            # Extract order ID (simplified)
            import re
            order_match = re.search(r'ord-d+', message_lower)
            if order_match:
                order_id = order_match.group().upper()
                order_info = self.lookup_order(order_id)
                if "error" not in order_info:
                    return f"I found your order {order_id}. Status: {order_info['status']}. Items: {', '.join(order_info['items'])}."
                else:
                    return "I couldn't find that order. Please double-check the order ID."
        
        elif "refund" in message_lower or "return" in message_lower:
            return "I can help you with a refund. Please provide your order ID so I can look it up."
        
        elif "help" in message_lower:
            return "I'm here to help! I can assist with order status, refunds, returns, and general questions. What do you need help with?"
        
        else:
            return "I understand you need assistance. Could you please provide more details about how I can help you today?"

# Usage example
def main():
    # Initialize the agent
    agent = CustomerSupportAgent(project_id="your-project-id")
    
    # Create the agent (do this once during setup)
    # agent_resource_name = agent.create_agent()
    
    # Example conversation
    messages = [
        "Hi, I need help with my order",
        "My order ID is ORD-12345",
        "I want to return this order",
        "The laptop arrived damaged"
    ]
    
    session_id = "customer_123"
    
    for message in messages:
        response = agent.chat(message, session_id)
        print(f"User: {message}")
        print(f"Agent: {response}\n")

if __name__ == "__main__":
    main()

import asyncio
from google.cloud import aiplatform
from google.cloud.aiplatform.gapic import endpoint_service_client
import requests
import json
from datetime import datetime
from typing import Dict, List, Any, Optional

class EnterpriseAgent:
    """Advanced enterprise agent with multiple tool integrations"""
    
    def __init__(self, project_id: str, model_endpoint: str):
        self.project_id = project_id
        self.model_endpoint = model_endpoint
        self.conversation_history = {}
        
        # Initialize external service clients
        self.setup_integrations()
    
    def setup_integrations(self):
        """Setup external service integrations"""
        self.integrations = {
            "crm": CRMIntegration(),
            "calendar": CalendarIntegration(), 
            "email": EmailIntegration(),
            "analytics": AnalyticsIntegration()
        }
    
    async def process_conversation(self, user_message: str, user_id: str, 
                                 context: Dict[str, Any] = None) -> Dict[str, Any]:
        """Process user message with context awareness"""
        
        # Get or create conversation context
        if user_id not in self.conversation_history:
            self.conversation_history[user_id] = {
                "messages": [],
                "context": context or {},
                "user_preferences": await self.get_user_preferences(user_id)
            }
        
        conversation = self.conversation_history[user_id]
        conversation["messages"].append({
            "role": "user",
            "content": user_message,
            "timestamp": datetime.now().isoformat()
        })
        
        # Analyze user intent and extract entities
        intent_analysis = await self.analyze_intent(user_message, conversation["context"])
        
        # Determine if tool calls are needed
        tool_calls = await self.plan_tool_calls(intent_analysis, conversation)
        
        # Execute tool calls if needed
        tool_results = {}
        if tool_calls:
            tool_results = await self.execute_tools(tool_calls, user_id)
        
        # Generate response with tool results
        response = await self.generate_response(
            user_message, 
            conversation,
            intent_analysis,
            tool_results
        )
        
        # Store agent response
        conversation["messages"].append({
            "role": "assistant", 
            "content": response["text"],
            "tool_calls": tool_calls,
            "tool_results": tool_results,
            "timestamp": datetime.now().isoformat()
        })
        
        return response
    
    async def analyze_intent(self, message: str, context: Dict) -> Dict[str, Any]:
        """Analyze user intent using NLU"""
        # This would typically use Google's NLU APIs
        intents_patterns = {
            "schedule_meeting": ["schedule", "meeting", "calendar", "book", "appointment"],
            "get_analytics": ["analytics", "data", "metrics", "performance", "statistics"],
            "send_email": ["email", "send", "message", "contact"],
            "customer_lookup": ["customer", "client", "account", "lookup", "find"],
            "task_management": ["task", "todo", "reminder", "deadline", "project"]
        }
        
        message_lower = message.lower()
        detected_intents = []
        
        for intent, patterns in intents_patterns.items():
            if any(pattern in message_lower for pattern in patterns):
                detected_intents.append(intent)
        
        # Extract entities (simplified)
        entities = self.extract_entities(message)
        
        return {
            "intents": detected_intents,
            "entities": entities,
            "confidence": 0.85  # Would come from actual NLU model
        }
    
    def extract_entities(self, message: str) -> Dict[str, List[str]]:
        """Extract entities from message"""
        import re
        entities = {
            "dates": [],
            "times": [],
            "emails": [],
            "names": [],
            "companies": []
        }
        
        # Extract emails
        email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
        entities["emails"] = re.findall(email_pattern, message)
        
        # Extract dates (simplified)
        date_patterns = [
            r'\b(?:today|tomorrow|yesterday)\b',
            r'\b\d{1,2}/\d{1,2}/\d{4}\b',
            r'\b(?:Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday)\b'
        ]
        for pattern in date_patterns:
            entities["dates"].extend(re.findall(pattern, message, re.IGNORECASE))
        
        return entities
    
    async def plan_tool_calls(self, intent_analysis: Dict, conversation: Dict) -> List[Dict]:
        """Plan which tools to call based on intent"""
        tool_calls = []
        
        for intent in intent_analysis["intents"]:
            if intent == "schedule_meeting":
                tool_calls.append({
                    "tool": "calendar",
                    "action": "create_meeting",
                    "parameters": {
                        "title": "Meeting",  # Would extract from message
                        "date": intent_analysis["entities"]["dates"][0] if intent_analysis["entities"]["dates"] else "today",
                        "duration": 60
                    }
                })
            
            elif intent == "get_analytics":
                tool_calls.append({
                    "tool": "analytics",
                    "action": "get_dashboard_data",
                    "parameters": {
                        "timeframe": "last_week",
                        "metrics": ["revenue", "users", "conversion_rate"]
                    }
                })
            
            elif intent == "customer_lookup":
                if intent_analysis["entities"]["emails"]:
                    tool_calls.append({
                        "tool": "crm",
                        "action": "lookup_customer",
                        "parameters": {
                            "email": intent_analysis["entities"]["emails"][0]
                        }
                    })
        
        return tool_calls
    
    async def execute_tools(self, tool_calls: List[Dict], user_id: str) -> Dict[str, Any]:
        """Execute tool calls asynchronously"""
        results = {}
        
        for tool_call in tool_calls:
            tool_name = tool_call["tool"]
            action = tool_call["action"]
            parameters = tool_call["parameters"]
            
            if tool_name in self.integrations:
                try:
                    result = await self.integrations[tool_name].call_action(action, parameters)
                    results[f"{tool_name}_{action}"] = result
                except Exception as e:
                    results[f"{tool_name}_{action}"] = {"error": str(e)}
        
        return results
    
    async def generate_response(self, user_message: str, conversation: Dict,
                              intent_analysis: Dict, tool_results: Dict) -> Dict[str, Any]:
        """Generate natural language response"""
        
        # Build context for response generation
        context = {
            "user_message": user_message,
            "conversation_history": conversation["messages"][-5:],  # Last 5 messages
            "intent_analysis": intent_analysis,
            "tool_results": tool_results,
            "user_preferences": conversation.get("user_preferences", {})
        }
        
        # Generate response based on tools results
        if tool_results:
            response_text = self.format_tool_response(tool_results, intent_analysis)
        else:
            response_text = self.generate_conversational_response(user_message, context)
        
        return {
            "text": response_text,
            "context": context,
            "suggestions": self.generate_suggestions(intent_analysis, tool_results)
        }
    
    def format_tool_response(self, tool_results: Dict, intent_analysis: Dict) -> str:
        """Format tool results into natural language"""
        responses = []
        
        for tool_result_key, result in tool_results.items():
            if "error" in result:
                responses.append(f"I encountered an issue: {result['error']}")
                continue
            
            if "calendar" in tool_result_key:
                if "meeting_id" in result:
                    responses.append(f"I've scheduled your meeting for {result.get('date', 'the requested time')}. Meeting ID: {result['meeting_id']}")
            
            elif "analytics" in tool_result_key:
                metrics = result.get("metrics", {})
                if metrics:
                    responses.append(f"Here are your analytics: Revenue: ${metrics.get('revenue', 'N/A')}, Users: {metrics.get('users', 'N/A')}, Conversion Rate: {metrics.get('conversion_rate', 'N/A')}%")
            
            elif "crm" in tool_result_key:
                customer = result.get("customer", {})
                if customer:
                    responses.append(f"Found customer: {customer.get('name', 'Unknown')} - Status: {customer.get('status', 'Unknown')}, Last Contact: {customer.get('last_contact', 'Never')}")
        
        return " ".join(responses) if responses else "I've completed your request."
    
    def generate_conversational_response(self, message: str, context: Dict) -> str:
        """Generate conversational response without tools"""
        # Simplified response generation
        greetings = ["hello", "hi", "hey", "good morning", "good afternoon"]
        if any(greeting in message.lower() for greeting in greetings):
            return "Hello! I'm your enterprise AI assistant. I can help you with scheduling, analytics, customer data, and more. What would you like to do today?"
        
        return "I understand you need assistance. Could you provide more specific details about what you'd like me to help you with?"
    
    def generate_suggestions(self, intent_analysis: Dict, tool_results: Dict) -> List[str]:
        """Generate helpful suggestions for the user"""
        suggestions = []
        
        if "schedule_meeting" in intent_analysis["intents"]:
            suggestions.append("Would you like to send calendar invites to attendees?")
            suggestions.append("Should I set a reminder for this meeting?")
        
        if "analytics" in intent_analysis["intents"]:
            suggestions.append("Would you like me to create a report with this data?")
            suggestions.append("Should I set up automated analytics alerts?")
        
        return suggestions
    
    async def get_user_preferences(self, user_id: str) -> Dict[str, Any]:
        """Get user preferences and settings"""
        # This would typically query a user preferences database
        return {
            "timezone": "America/New_York",
            "language": "en-US", 
            "notification_preferences": {
                "email": True,
                "sms": False,
                "push": True
            },
            "working_hours": {
                "start": "09:00",
                "end": "17:00"
            }
        }

# Mock integration classes
class CRMIntegration:
    async def call_action(self, action: str, parameters: Dict) -> Dict:
        if action == "lookup_customer":
            # Mock customer data
            return {
                "customer": {
                    "name": "John Doe",
                    "email": parameters.get("email", ""),
                    "status": "Active",
                    "last_contact": "2024-01-15",
                    "value": "$15,000"
                }
            }
        return {"error": "Unknown action"}

class CalendarIntegration:
    async def call_action(self, action: str, parameters: Dict) -> Dict:
        if action == "create_meeting":
            return {
                "meeting_id": "MTG-123456",
                "date": parameters.get("date", "today"),
                "title": parameters.get("title", "Meeting"),
                "calendar_link": "https://calendar.google.com/event?id=MTG-123456"
            }
        return {"error": "Unknown action"}

class EmailIntegration:
    async def call_action(self, action: str, parameters: Dict) -> Dict:
        if action == "send_email":
            return {
                "message_id": "EMAIL-789",
                "status": "sent",
                "recipients": parameters.get("recipients", [])
            }
        return {"error": "Unknown action"}

class AnalyticsIntegration:
    async def call_action(self, action: str, parameters: Dict) -> Dict:
        if action == "get_dashboard_data":
            return {
                "metrics": {
                    "revenue": 125000,
                    "users": 5420,
                    "conversion_rate": 3.2
                },
                "timeframe": parameters.get("timeframe", "last_week")
            }
        return {"error": "Unknown action"}

# Usage example
async def main():
    agent = EnterpriseAgent(
        project_id="your-project-id",
        model_endpoint="your-model-endpoint"
    )
    
    # Example conversation
    response = await agent.process_conversation(
        "Schedule a meeting with john@example.com for tomorrow at 2 PM",
        user_id="user_456"
    )
    
    print(f"Agent Response: {response['text']}")
    print(f"Suggestions: {response['suggestions']}")

if __name__ == "__main__":
    asyncio.run(main())