Enterprise AI Agent Platforms¶
⏱️ Estimated reading time: 19 minutes
Introduction¶
Enterprise AI agent platforms provide production-ready infrastructure for deploying, managing, and scaling AI agents in business environments. These platforms offer enterprise-grade features including security, compliance, monitoring, and integration with existing business systems.
Major Enterprise Platforms¶
1. AWS Bedrock Agents¶
Overview¶
Amazon Bedrock Agents enable building and deploying AI agents that can execute multi-step tasks using foundation models and enterprise data.
Architecture¶
import boto3
from typing import Dict, List
class BedrockAgent:
def __init__(self, agent_name: str):
self.bedrock = boto3.client('bedrock-agent')
self.bedrock_runtime = boto3.client('bedrock-agent-runtime')
self.agent_name = agent_name
def create_agent(self, config: Dict):
"""Create a new Bedrock agent."""
response = self.bedrock.create_agent(
agentName=self.agent_name,
foundationModel='anthropic.claude-v2',
instruction=config['instruction'],
agentResourceRoleArn=config['role_arn'],
actionGroups=[
{
'actionGroupName': 'enterprise-tools',
'actionGroupExecutor': {
'lambda': config['lambda_arn']
},
'apiSchema': {
's3': {
's3BucketName': config['schema_bucket'],
's3ObjectKey': 'api-schema.yaml'
}
}
}
],
knowledgeBases=[
{
'knowledgeBaseId': config['kb_id'],
'description': 'Enterprise knowledge base'
}
]
)
return response['agent']['agentId']
async def invoke_agent(self, prompt: str, session_id: str = None):
"""Invoke the agent with a prompt."""
response = self.bedrock_runtime.invoke_agent(
agentId=self.agent_id,
agentAliasId=self.agent_alias_id,
sessionId=session_id or str(uuid.uuid4()),
inputText=prompt
)
# Stream the response
for event in response['completion']:
if 'chunk' in event:
yield event['chunk']['bytes'].decode('utf-8')
Knowledge Base Integration¶
class BedrockKnowledgeBase:
def __init__(self):
self.bedrock = boto3.client('bedrock-agent')
def create_knowledge_base(self, name: str, data_source: str):
# Create knowledge base
kb_response = self.bedrock.create_knowledge_base(
name=name,
roleArn='arn:aws:iam::123456789012:role/BedrockKBRole',
knowledgeBaseConfiguration={
'type': 'VECTOR',
'vectorKnowledgeBaseConfiguration': {
'embeddingModelArn': 'arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v1'
}
},
storageConfiguration={
'type': 'OPENSEARCH_SERVERLESS',
'opensearchServerlessConfiguration': {
'collectionArn': 'arn:aws:aoss:us-east-1:123456789012:collection/kb-collection',
'vectorIndexName': 'bedrock-knowledge-base-index',
'fieldMapping': {
'vectorField': 'embedding',
'textField': 'text',
'metadataField': 'metadata'
}
}
}
)
# Add data source
self.bedrock.create_data_source(
knowledgeBaseId=kb_response['knowledgeBase']['knowledgeBaseId'],
name=f'{name}-datasource',
dataSourceConfiguration={
'type': 'S3',
's3Configuration': {
'bucketArn': f'arn:aws:s3:::{data_source}'
}
}
)
2. Google Vertex AI Agents¶
Overview¶
Vertex AI Agents provide a comprehensive platform for building, deploying, and managing AI agents with Google's foundation models.
Implementation¶
from google.cloud import aiplatform
from google.cloud.aiplatform import agents
class VertexAIAgent:
def __init__(self, project_id: str, location: str):
aiplatform.init(project=project_id, location=location)
self.project_id = project_id
self.location = location
def create_agent(self, agent_config: Dict):
"""Create a Vertex AI agent."""
agent = agents.Agent.create(
display_name=agent_config['name'],
description=agent_config['description'],
model_name='gemini-pro',
tools=[
agents.Tool(
function_declarations=[
agents.FunctionDeclaration(
name='search_database',
description='Search enterprise database',
parameters={
'type': 'object',
'properties': {
'query': {'type': 'string'},
'filters': {'type': 'object'}
}
}
)
]
)
],
system_instruction=agent_config['system_prompt']
)
return agent
def create_conversation(self, agent):
"""Create a conversation with the agent."""
conversation = agent.start_conversation()
return conversation
async def send_message(self, conversation, message: str):
"""Send message to agent and get response."""
response = await conversation.send_message_async(
message,
generation_config={
'temperature': 0.7,
'max_output_tokens': 2048,
}
)
return response.text
Integration with Google Cloud Services¶
class VertexAIEnterpriseAgent:
def __init__(self):
self.bigquery = bigquery.Client()
self.storage = storage.Client()
self.firestore = firestore.Client()
def setup_tools(self):
"""Configure enterprise tool integrations."""
tools = [
{
'name': 'query_bigquery',
'function': self.query_bigquery,
'description': 'Query BigQuery data warehouse'
},
{
'name': 'access_storage',
'function': self.access_cloud_storage,
'description': 'Access Cloud Storage files'
},
{
'name': 'update_firestore',
'function': self.update_firestore,
'description': 'Update Firestore database'
}
]
return tools
async def query_bigquery(self, sql: str):
"""Execute BigQuery SQL query."""
query_job = self.bigquery.query(sql)
results = query_job.result()
return [dict(row) for row in results]
3. Microsoft Azure AI Agent Service¶
Overview¶
Azure AI Agent Service provides enterprise-grade agent deployment with deep integration into the Microsoft ecosystem.
Implementation¶
from azure.ai.agents import AgentClient
from azure.identity import DefaultAzureCredential
class AzureAIAgent:
def __init__(self, endpoint: str, deployment_name: str):
self.credential = DefaultAzureCredential()
self.client = AgentClient(
endpoint=endpoint,
credential=self.credential
)
self.deployment_name = deployment_name
def create_agent(self, config: Dict):
"""Create an Azure AI agent."""
agent = self.client.agents.create(
deployment_name=self.deployment_name,
model='gpt-4',
name=config['name'],
instructions=config['instructions'],
tools=[
{
'type': 'code_interpreter',
'enabled': True
},
{
'type': 'retrieval',
'retrieval': {
'search_type': 'hybrid',
'top_k': 5
}
},
{
'type': 'function',
'function': {
'name': 'query_dynamics',
'description': 'Query Dynamics 365',
'parameters': {...}
}
}
],
file_ids=config.get('file_ids', [])
)
return agent
async def run_agent(self, agent_id: str, prompt: str):
"""Run the agent with a prompt."""
thread = self.client.threads.create()
message = self.client.messages.create(
thread_id=thread.id,
role='user',
content=prompt
)
run = self.client.runs.create(
thread_id=thread.id,
assistant_id=agent_id
)
# Wait for completion
while run.status in ['queued', 'in_progress']:
await asyncio.sleep(1)
run = self.client.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
# Get messages
messages = self.client.messages.list(thread_id=thread.id)
return messages.data[0].content
4. Salesforce Einstein Agents¶
Overview¶
Einstein Agents bring AI capabilities directly into Salesforce CRM, enabling intelligent automation of sales and service processes.
Configuration¶
from simple_salesforce import Salesforce
import requests
class EinsteinAgent:
def __init__(self, username: str, password: str, security_token: str):
self.sf = Salesforce(
username=username,
password=password,
security_token=security_token
)
self.einstein_url = 'https://api.einstein.ai/v2'
def create_service_agent(self, config: Dict):
"""Create a customer service agent."""
agent_config = {
'name': config['name'],
'type': 'SERVICE_AGENT',
'capabilities': [
'CASE_CLASSIFICATION',
'SENTIMENT_ANALYSIS',
'RESPONSE_GENERATION',
'KNOWLEDGE_SEARCH'
],
'knowledge_bases': [
{
'type': 'SALESFORCE_KNOWLEDGE',
'object': 'Knowledge__kav'
},
{
'type': 'CASE_HISTORY',
'lookback_days': 90
}
],
'automation_rules': [
{
'trigger': 'NEW_CASE',
'actions': ['CLASSIFY', 'ROUTE', 'SUGGEST_RESPONSE']
}
]
}
response = requests.post(
f'{self.einstein_url}/agents',
json=agent_config,
headers=self.get_auth_headers()
)
return response.json()
def process_case(self, case_id: str, agent_id: str):
"""Process a support case with the agent."""
case = self.sf.Case.get(case_id)
response = requests.post(
f'{self.einstein_url}/agents/{agent_id}/process',
json={
'input': {
'subject': case['Subject'],
'description': case['Description'],
'priority': case['Priority']
}
},
headers=self.get_auth_headers()
)
result = response.json()
# Update case with agent recommendations
self.sf.Case.update(case_id, {
'Category__c': result['classification'],
'Sentiment__c': result['sentiment'],
'Suggested_Response__c': result['suggested_response']
})
return result
5. ServiceNow AI Agents¶
Overview¶
ServiceNow provides AI agents for IT service management, automating incident resolution and service requests.
Implementation¶
import requests
from typing import Dict, List
class ServiceNowAgent:
def __init__(self, instance: str, username: str, password: str):
self.instance = instance
self.auth = (username, password)
self.base_url = f'https://{instance}.service-now.com/api'
def create_virtual_agent(self, config: Dict):
"""Create a virtual agent for IT support."""
agent_config = {
'name': config['name'],
'type': 'virtual_agent',
'topics': [
{
'name': 'Password Reset',
'intent': 'reset_password',
'flow': 'password_reset_flow'
},
{
'name': 'Software Installation',
'intent': 'install_software',
'flow': 'software_install_flow'
}
],
'nlp_provider': 'servicenow_nlu',
'integrations': ['active_directory', 'software_catalog']
}
response = requests.post(
f'{self.base_url}/now/va/agents',
json=agent_config,
auth=self.auth
)
return response.json()
def handle_incident(self, incident_number: str):
"""Automatically handle an incident."""
# Get incident details
incident = self.get_incident(incident_number)
# Analyze with AI
analysis = requests.post(
f'{self.base_url}/now/ai/analyze',
json={'text': incident['description']},
auth=self.auth
).json()
# Find similar resolved incidents
similar = self.find_similar_incidents(
incident['description'],
limit=5
)
# Generate resolution
resolution = self.generate_resolution(
incident=incident,
analysis=analysis,
similar_incidents=similar
)
# Update incident
self.update_incident(incident_number, {
'work_notes': resolution['explanation'],
'resolution_code': resolution['code'],
'close_notes': resolution['summary']
})
return resolution
6. Snowflake Cortex Agents¶
Overview¶
Snowflake Cortex provides AI agents that operate directly on data warehouse data, enabling intelligent data analysis and automation.
Implementation¶
import snowflake.connector
from snowflake.ml.model import ModelRegistry
class SnowflakeCortexAgent:
def __init__(self, account: str, user: str, password: str):
self.conn = snowflake.connector.connect(
account=account,
user=user,
password=password
)
self.cursor = self.conn.cursor()
def create_data_agent(self, config: Dict):
"""Create an agent for data analysis."""
# Create agent function
create_agent_sql = f"""
CREATE OR REPLACE FUNCTION {config['name']}_agent(prompt STRING)
RETURNS TABLE(response STRING, data VARIANT)
LANGUAGE PYTHON
RUNTIME_VERSION = '3.9'
PACKAGES = ('snowflake-ml-python', 'pandas')
HANDLER = 'DataAgent.process'
AS '''
import pandas as pd
from snowflake.ml.llm import complete
class DataAgent:
@staticmethod
def process(prompt):
# Analyze prompt to determine data needs
analysis = complete(
model='llama2-70b',
prompt=f"Analyze this request and determine SQL needed: {prompt}"
)
# Generate and execute SQL
sql = analysis['sql']
df = pd.read_sql(sql, connection)
# Generate insights
insights = complete(
model='llama2-70b',
prompt=f"Provide insights on this data: {df.to_json()}"
)
return [(insights, df.to_dict())]
''';
"""
self.cursor.execute(create_agent_sql)
# Create stored procedure for autonomous operation
create_proc_sql = f"""
CREATE OR REPLACE PROCEDURE {config['name']}_autonomous()
RETURNS STRING
LANGUAGE SQL
AS
BEGIN
-- Scheduled autonomous operations
CALL {config['name']}_agent('Analyze daily sales trends');
CALL {config['name']}_agent('Identify anomalies in user behavior');
CALL {config['name']}_agent('Generate executive summary');
RETURN 'Autonomous analysis complete';
END;
"""
self.cursor.execute(create_proc_sql)
7. Specialized Enterprise Platforms¶
Lindy.ai¶
class LindyAgent:
"""No-code platform for business automation."""
def create_workflow(self, config: Dict):
workflow = {
'name': config['name'],
'trigger': {
'type': 'email',
'conditions': {'subject_contains': 'invoice'}
},
'actions': [
{
'type': 'extract_data',
'source': 'email_attachment',
'format': 'invoice'
},
{
'type': 'update_system',
'target': 'accounting_software',
'mapping': config['field_mapping']
},
{
'type': 'send_notification',
'channel': 'slack',
'message': 'Invoice processed: {invoice_number}'
}
]
}
return workflow
IBM watsonx Assistant¶
class WatsonxAgent:
def __init__(self, api_key: str, url: str):
self.authenticator = IAMAuthenticator(api_key)
self.assistant = AssistantV2(
version='2024-01-01',
authenticator=self.authenticator
)
self.assistant.set_service_url(url)
def create_assistant(self, config: Dict):
response = self.assistant.create_assistant(
name=config['name'],
description=config['description'],
language='en',
assistant_id=config['assistant_id']
).get_result()
# Add skills
for skill in config['skills']:
self.add_skill(response['assistant_id'], skill)
return response
Enterprise Features Comparison¶
| Platform | Strengths | Integration | Pricing Model | Best For |
|---|---|---|---|---|
| AWS Bedrock | Scalability, Multi-model | AWS ecosystem | Pay-per-use | Cloud-native apps |
| Google Vertex | ML capabilities | Google Cloud | Pay-per-use | Data-heavy workflows |
| Azure AI | Microsoft integration | Office 365, Dynamics | Subscription | Enterprise Microsoft |
| Salesforce Einstein | CRM integration | Salesforce ecosystem | Per-user | Sales & Service |
| ServiceNow | ITSM focus | IT systems | Platform license | IT operations |
| Snowflake Cortex | Data warehouse native | Data platforms | Compute credits | Analytics |
Security & Compliance¶
Enterprise Security Features¶
class EnterpriseSecurityAgent:
def __init__(self):
self.encryption = AES256Encryption()
self.audit_logger = AuditLogger()
self.compliance_checker = ComplianceChecker()
def secure_execution(self, func, *args, **kwargs):
# Pre-execution security checks
self.compliance_checker.verify_operation(func.__name__)
# Encrypt sensitive data
encrypted_args = self.encryption.encrypt_sensitive(args)
# Log operation
self.audit_logger.log_operation(
operation=func.__name__,
user=self.get_current_user(),
timestamp=datetime.now(),
data_classification=self.classify_data(args)
)
try:
# Execute with monitoring
result = func(*encrypted_args, **kwargs)
# Post-execution validation
self.validate_output(result)
return result
except Exception as e:
self.audit_logger.log_security_event(
event_type='EXECUTION_FAILURE',
details=str(e)
)
raise
Compliance Requirements¶
class ComplianceAgent:
def __init__(self):
self.regulations = {
'GDPR': GDPRCompliance(),
'HIPAA': HIPAACompliance(),
'SOC2': SOC2Compliance(),
'PCI': PCICompliance()
}
def ensure_compliance(self, data, operation):
for regulation, checker in self.regulations.items():
if checker.applies_to(data):
checker.validate(data, operation)
checker.apply_controls(data)
Deployment Patterns¶
Multi-Region Deployment¶
class MultiRegionAgent:
def __init__(self):
self.regions = {
'us-east-1': USEastAgent(),
'eu-west-1': EUWestAgent(),
'ap-southeast-1': APSoutheastAgent()
}
def route_request(self, request):
# Determine optimal region
region = self.get_optimal_region(
user_location=request.user_location,
data_residency=request.data_requirements,
latency_requirements=request.sla
)
return self.regions[region].process(request)
Best Practices¶
1. Scalability Design¶
class ScalableEnterpriseAgent:
def __init__(self):
self.connection_pool = ConnectionPool(max_size=100)
self.cache = RedisCache()
self.rate_limiter = RateLimiter(requests_per_second=1000)
async def handle_request(self, request):
# Check cache
cached = await self.cache.get(request.cache_key)
if cached:
return cached
# Rate limiting
await self.rate_limiter.acquire()
# Process with connection pooling
async with self.connection_pool.acquire() as conn:
result = await self.process(request, conn)
# Cache result
await self.cache.set(request.cache_key, result, ttl=300)
return result
2. Monitoring & Observability¶
class MonitoredEnterpriseAgent:
def __init__(self):
self.metrics = MetricsCollector()
self.tracer = DistributedTracer()
self.health_checker = HealthChecker()
@self.tracer.trace()
@self.metrics.measure()
async def execute(self, task):
span = self.tracer.start_span('agent_execution')
try:
# Record metrics
self.metrics.increment('tasks.started')
start_time = time.time()
result = await self.process_task(task)
# Record success metrics
self.metrics.increment('tasks.completed')
self.metrics.record('task.duration', time.time() - start_time)
return result
except Exception as e:
self.metrics.increment('tasks.failed')
span.set_tag('error', True)
raise
finally:
span.finish()
Conclusion¶
Enterprise AI agent platforms provide the infrastructure, security, and scalability required for production deployments. The choice of platform depends on existing technology stack, compliance requirements, and specific use cases. Key considerations include integration capabilities, security features, scalability, and total cost of ownership.
Resources¶
Documentation¶
Enterprise Guides¶
Next Steps¶
- Review Security & Observability for production safety
- Explore Multi-Agent Systems for complex deployments
- Study Communication Protocols for agent interoperability