AWS AI Services
Building AI Agents with Amazon Bedrock: From Concept to Production
Complete guide to building autonomous AI agents using Amazon Bedrock Agents for Australian businesses, covering agent design, action groups, knowledge bases, and orchestration patterns.
CloudPoint Team
AI agents represent the next evolution in generative AI applications. Unlike simple chatbots that respond to queries, agents can plan multi-step tasks, use tools, access knowledge bases, and take actions on your behalf. Amazon Bedrock Agents provides a fully managed service for building autonomous AI agents without managing infrastructure. For Australian businesses, this enables sophisticated automation while maintaining security and compliance.
What are AI Agents?
AI agents are autonomous systems that can:
- Understand complex, multi-step requests
- Plan sequences of actions to achieve goals
- Use external tools and APIs
- Access knowledge bases for information
- Make decisions based on context
- Execute tasks with minimal human intervention
Traditional Chatbot vs AI Agent
Traditional Chatbot:
User: "Book a meeting with the sales team next Tuesday"
Bot: "I can help with that. Please visit calendar.company.com to book a meeting."AI Agent:
User: "Book a meeting with the sales team next Tuesday"
Agent:
1. Checks calendar availability for sales team
2. Finds available slot on Tuesday at 2pm
3. Creates meeting invite
4. Sends calendar invitations
5. Confirms: "Meeting booked for Tuesday, Nov 28 at 2:00 PM AEDT with Sales Team"Agent Components
Foundation Model: The AI brain that understands requests and plans actions.
Action Groups: Capabilities the agent can perform (APIs, Lambda functions).
Knowledge Bases: Information repositories the agent can query.
Orchestration: The logic that coordinates actions to achieve goals.
Guardrails: Safety controls to prevent harmful outputs.
Agent Architecture
Basic Agent Structure
import boto3
import json
class BedrockAgent:
def __init__(self, agent_id: str, agent_alias_id: str):
self.bedrock_agent_runtime = boto3.client(
'bedrock-agent-runtime',
region_name='ap-southeast-2'
)
self.agent_id = agent_id
self.agent_alias_id = agent_alias_id
def invoke(self, prompt: str, session_id: str = None) -> dict:
"""Invoke Bedrock agent with a prompt."""
response = self.bedrock_agent_runtime.invoke_agent(
agentId=self.agent_id,
agentAliasId=self.agent_alias_id,
sessionId=session_id or self._generate_session_id(),
inputText=prompt
)
# Stream response
completion = ""
for event in response['completion']:
if 'chunk' in event:
chunk = event['chunk']
completion += chunk['bytes'].decode('utf-8')
return {
'completion': completion,
'session_id': response['sessionId']
}
def _generate_session_id(self) -> str:
"""Generate unique session ID."""
from uuid import uuid4
return str(uuid4())Agent Workflow
User Input
↓
Foundation Model (Understands intent)
↓
Planning (Breaks down into steps)
↓
Action Selection (Choose appropriate tool)
↓
Action Execution (Call API/Lambda)
↓
Response Processing (Interpret results)
↓
Next Action or Final ResponseCreating Your First Agent
1. Define Agent Purpose
Example: Australian Business Assistant:
agent_instruction = """You are an Australian business assistant that helps with:
1. Checking company information via ASIC lookup
2. Validating ABN (Australian Business Numbers)
3. Calculating GST and business expenses
4. Providing business registration guidance
5. Australian business compliance reminders
Always use Australian terminology, date formats (DD/MM/YYYY), and currency (AUD).
Be professional and helpful. For legal or financial advice, recommend consulting appropriate professionals."""2. Set Up Action Groups
Action groups define what the agent can do:
import boto3
class AgentActionGroups:
def __init__(self):
self.bedrock_agent = boto3.client('bedrock-agent', region_name='ap-southeast-2')
def create_action_group(
self,
agent_id: str,
action_group_name: str,
lambda_arn: str,
api_schema: dict
) -> str:
"""Create action group for agent."""
response = self.bedrock_agent.create_agent_action_group(
agentId=agent_id,
agentVersion='DRAFT',
actionGroupName=action_group_name,
actionGroupExecutor={
'lambda': lambda_arn
},
apiSchema={
'payload': json.dumps(api_schema)
},
description=f'Action group for {action_group_name}'
)
return response['agentActionGroup']['actionGroupId']
# Define API schema for ABN validation
abn_validation_schema = {
"openapi": "3.0.0",
"info": {
"title": "ABN Validation API",
"version": "1.0.0",
"description": "Validate Australian Business Numbers"
},
"paths": {
"/validate-abn": {
"post": {
"summary": "Validate ABN",
"description": "Validates an Australian Business Number",
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"abn": {
"type": "string",
"description": "11-digit Australian Business Number"
}
},
"required": ["abn"]
}
}
}
},
"responses": {
"200": {
"description": "Validation result",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"valid": {"type": "boolean"},
"abn": {"type": "string"},
"entity_name": {"type": "string"},
"status": {"type": "string"}
}
}
}
}
}
}
}
},
"/calculate-gst": {
"post": {
"summary": "Calculate GST",
"description": "Calculate GST for Australian prices",
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"amount": {
"type": "number",
"description": "Amount in AUD"
},
"gst_inclusive": {
"type": "boolean",
"description": "Whether amount includes GST"
}
},
"required": ["amount"]
}
}
}
},
"responses": {
"200": {
"description": "GST calculation",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"base_amount": {"type": "number"},
"gst_amount": {"type": "number"},
"total_amount": {"type": "number"}
}
}
}
}
}
}
}
}
}
}3. Create Lambda Functions
Implement actions as Lambda functions:
# lambda_function.py
import json
def validate_abn(abn: str) -> dict:
"""Validate Australian Business Number using weighted checksum."""
# Remove spaces and hyphens
clean_abn = abn.replace(' ', '').replace('-', '')
# Validate format
if len(clean_abn) != 11 or not clean_abn.isdigit():
return {
'valid': False,
'abn': abn,
'error': 'ABN must be 11 digits'
}
# Apply weighting
weights = [10, 1, 3, 5, 7, 9, 11, 13, 15, 17, 19]
digits = [int(d) for d in clean_abn]
# Subtract 1 from first digit
digits[0] -= 1
# Calculate weighted sum
total = sum(d * w for d, w in zip(digits, weights))
# Check if divisible by 89
is_valid = (total % 89) == 0
# In production, query ABR (Australian Business Register) API
# for entity details
entity_name = "Example Pty Ltd" if is_valid else None
status = "Active" if is_valid else "Invalid"
return {
'valid': is_valid,
'abn': clean_abn,
'entity_name': entity_name,
'status': status
}
def calculate_gst(amount: float, gst_inclusive: bool = False) -> dict:
"""Calculate GST for Australian prices."""
gst_rate = 0.10 # 10% GST
if gst_inclusive:
# Amount includes GST, extract it
base_amount = amount / (1 + gst_rate)
gst_amount = amount - base_amount
total_amount = amount
else:
# Amount excludes GST, add it
base_amount = amount
gst_amount = amount * gst_rate
total_amount = amount + gst_amount
return {
'base_amount': round(base_amount, 2),
'gst_amount': round(gst_amount, 2),
'total_amount': round(total_amount, 2),
'gst_rate': gst_rate
}
def lambda_handler(event, context):
"""AWS Lambda handler for Bedrock Agent actions."""
# Extract action and parameters
action = event['actionGroup']
api_path = event['apiPath']
parameters = event.get('parameters', [])
# Convert parameters to dict
params = {p['name']: p['value'] for p in parameters}
# Route to appropriate function
if api_path == '/validate-abn':
result = validate_abn(params['abn'])
elif api_path == '/calculate-gst':
result = calculate_gst(
float(params['amount']),
params.get('gst_inclusive', 'false').lower() == 'true'
)
else:
return {
'statusCode': 400,
'body': json.dumps({'error': 'Unknown action'})
}
# Return response in Bedrock Agent format
return {
'messageVersion': '1.0',
'response': {
'actionGroup': action,
'apiPath': api_path,
'httpMethod': 'POST',
'httpStatusCode': 200,
'responseBody': {
'application/json': {
'body': json.dumps(result)
}
}
}
}4. Create Knowledge Base
Add knowledge sources for the agent:
class KnowledgeBaseManager:
def __init__(self):
self.bedrock_agent = boto3.client('bedrock-agent', region_name='ap-southeast-2')
def create_knowledge_base(
self,
name: str,
description: str,
embeddings_model: str = 'amazon.titan-embed-text-v1'
) -> str:
"""Create knowledge base for agent."""
response = self.bedrock_agent.create_knowledge_base(
name=name,
description=description,
roleArn=self._get_kb_role_arn(),
knowledgeBaseConfiguration={
'type': 'VECTOR',
'vectorKnowledgeBaseConfiguration': {
'embeddingModelArn': f'arn:aws:bedrock:ap-southeast-2::foundation-model/{embeddings_model}'
}
},
storageConfiguration={
'type': 'OPENSEARCH_SERVERLESS',
'opensearchServerlessConfiguration': {
'collectionArn': self._get_opensearch_collection_arn(),
'vectorIndexName': f'{name}-index',
'fieldMapping': {
'vectorField': 'embedding',
'textField': 'text',
'metadataField': 'metadata'
}
}
}
)
return response['knowledgeBase']['knowledgeBaseId']
def add_data_source(
self,
knowledge_base_id: str,
s3_bucket: str,
s3_prefix: str
) -> str:
"""Add S3 data source to knowledge base."""
response = self.bedrock_agent.create_data_source(
knowledgeBaseId=knowledge_base_id,
name='australian-business-docs',
dataSourceConfiguration={
'type': 'S3',
's3Configuration': {
'bucketArn': f'arn:aws:s3:::{s3_bucket}',
'inclusionPrefixes': [s3_prefix]
}
}
)
# Start ingestion job
self.bedrock_agent.start_ingestion_job(
knowledgeBaseId=knowledge_base_id,
dataSourceId=response['dataSource']['dataSourceId']
)
return response['dataSource']['dataSourceId']
# Usage
kb_manager = KnowledgeBaseManager()
# Create knowledge base
kb_id = kb_manager.create_knowledge_base(
name='australian-business-knowledge',
description='Knowledge base for Australian business regulations and compliance'
)
# Add documents
kb_manager.add_data_source(
kb_id,
'my-business-docs-bucket',
'compliance-documents/'
)5. Assemble Complete Agent
class AustralianBusinessAgent:
def __init__(self):
self.bedrock_agent = boto3.client('bedrock-agent', region_name='ap-southeast-2')
def create_agent(
self,
agent_name: str,
foundation_model: str = 'anthropic.claude-v2'
) -> str:
"""Create complete Bedrock agent."""
# Create agent
agent_response = self.bedrock_agent.create_agent(
agentName=agent_name,
agentResourceRoleArn=self._get_agent_role_arn(),
foundationModel=foundation_model,
instruction=self._get_agent_instructions(),
idleSessionTTLInSeconds=600,
description='Australian business assistant for compliance and calculations'
)
agent_id = agent_response['agent']['agentId']
# Create action groups
self._create_action_groups(agent_id)
# Associate knowledge base
self._associate_knowledge_base(agent_id)
# Prepare agent
self.bedrock_agent.prepare_agent(agentId=agent_id)
# Create alias
alias_response = self.bedrock_agent.create_agent_alias(
agentId=agent_id,
agentAliasName='production'
)
return agent_id, alias_response['agentAlias']['agentAliasId']
def _get_agent_instructions(self) -> str:
return """You are an Australian business assistant specializing in:
1. ABN validation and ASIC lookups
2. GST calculations and tax guidance
3. Business registration requirements
4. Compliance reminders and deadlines
5. Australian business regulations
Guidelines:
- Use Australian English (organise, not organize)
- Format dates as DD/MM/YYYY
- Currency in AUD
- Reference Australian regulations and standards
- For complex legal/financial matters, recommend professional consultation
- Be helpful, professional, and accurate
When asked about business information:
1. First check if ABN validation is needed
2. Calculate GST if amounts are mentioned
3. Query knowledge base for compliance information
4. Provide clear, actionable answers"""
def _create_action_groups(self, agent_id: str):
"""Create all action groups for agent."""
action_groups = AgentActionGroups()
# ABN and GST action group
action_groups.create_action_group(
agent_id=agent_id,
action_group_name='business-tools',
lambda_arn=self._get_lambda_arn(),
api_schema=abn_validation_schema
)
def _associate_knowledge_base(self, agent_id: str):
"""Associate knowledge base with agent."""
self.bedrock_agent.associate_agent_knowledge_base(
agentId=agent_id,
agentVersion='DRAFT',
knowledgeBaseId=self._get_knowledge_base_id(),
description='Australian business compliance knowledge',
knowledgeBaseState='ENABLED'
)
# Create agent
agent_builder = AustralianBusinessAgent()
agent_id, alias_id = agent_builder.create_agent('australian-business-assistant')
print(f"Agent created: {agent_id}")
print(f"Alias: {alias_id}")Using the Agent
Basic Interaction
# Initialize agent
agent = BedrockAgent(agent_id, alias_id)
# Simple query
response = agent.invoke("Is ABN 53 004 085 616 valid?")
print(response['completion'])
# Multi-step task
response = agent.invoke("""
I need to invoice a client for $5,000 worth of consulting services.
1. Calculate the GST
2. Tell me the total amount to invoice
3. Explain what records I need to keep for tax purposes
""")
print(response['completion'])Session Management
class AgentSession:
def __init__(self, agent: BedrockAgent):
self.agent = agent
self.session_id = None
self.history = []
def chat(self, message: str) -> str:
"""Continue conversation in same session."""
response = self.agent.invoke(message, self.session_id)
# Store session ID for continuity
self.session_id = response['session_id']
# Track history
self.history.append({
'user': message,
'agent': response['completion']
})
return response['completion']
def get_history(self) -> list:
"""Get conversation history."""
return self.history
# Usage
session = AgentSession(agent)
# Multi-turn conversation
response1 = session.chat("What do I need to register a business in NSW?")
response2 = session.chat("How much does it cost?")
response3 = session.chat("What about Victoria instead?")
# Agent maintains context across messages
for interaction in session.get_history():
print(f"User: {interaction['user']}")
print(f"Agent: {interaction['agent']}\n")Advanced Agent Patterns
1. Multi-Agent Orchestration
class MultiAgentOrchestrator:
def __init__(self):
self.agents = {
'compliance': BedrockAgent(compliance_agent_id, compliance_alias),
'finance': BedrockAgent(finance_agent_id, finance_alias),
'hr': BedrockAgent(hr_agent_id, hr_alias)
}
def route_query(self, query: str) -> dict:
"""Route query to appropriate specialist agent."""
# Use LLM to classify query
classification = self._classify_query(query)
# Route to specialist agent
agent_type = classification['category']
agent = self.agents.get(agent_type)
if agent:
response = agent.invoke(query)
else:
response = self._handle_general_query(query)
return {
'agent_type': agent_type,
'response': response['completion']
}
def _classify_query(self, query: str) -> dict:
"""Classify query to determine appropriate agent."""
bedrock_runtime = boto3.client('bedrock-runtime', region_name='ap-southeast-2')
prompt = f"""Classify this business query into one category:
- compliance: regulations, legal requirements, permits
- finance: accounting, tax, invoicing, expenses
- hr: employees, workplace, Fair Work, leave
Query: {query}
Respond with only the category name."""
response = bedrock_runtime.invoke_model(
modelId='anthropic.claude-v2',
body=json.dumps({
"prompt": f"\n\nHuman: {prompt}\n\nAssistant:",
"max_tokens_to_sample": 20
})
)
result = json.loads(response['body'].read())
category = result['completion'].strip().lower()
return {'category': category}
# Usage
orchestrator = MultiAgentOrchestrator()
response = orchestrator.route_query(
"How many weeks of parental leave are employees entitled to?"
)
print(f"Routed to: {response['agent_type']}")
print(f"Response: {response['response']}")2. Agent with Human-in-the-Loop
class HumanInLoopAgent:
def __init__(self, agent: BedrockAgent):
self.agent = agent
self.pending_approvals = {}
def invoke_with_approval(self, query: str, approval_required_actions: list) -> dict:
"""Invoke agent with human approval for certain actions."""
response = self.agent.invoke(query)
# Check if response requires approval
if self._requires_approval(response, approval_required_actions):
approval_id = self._request_approval(query, response)
return {
'status': 'pending_approval',
'approval_id': approval_id,
'preview': response['completion']
}
return {
'status': 'completed',
'response': response['completion']
}
def _requires_approval(self, response: dict, approval_actions: list) -> bool:
"""Check if response involves actions requiring approval."""
# Parse response for action indicators
response_text = response['completion'].lower()
for action in approval_actions:
if action.lower() in response_text:
return True
return False
def _request_approval(self, query: str, response: dict) -> str:
"""Request human approval for action."""
import uuid
approval_id = str(uuid.uuid4())
self.pending_approvals[approval_id] = {
'query': query,
'response': response,
'timestamp': datetime.utcnow().isoformat(),
'status': 'pending'
}
# Send notification (SNS, email, etc.)
self._notify_approver(approval_id)
return approval_id
def approve_action(self, approval_id: str) -> dict:
"""Approve pending action."""
if approval_id not in self.pending_approvals:
return {'error': 'Invalid approval ID'}
approval = self.pending_approvals[approval_id]
approval['status'] = 'approved'
return {
'status': 'approved',
'response': approval['response']['completion']
}
# Usage
hitl_agent = HumanInLoopAgent(agent)
# Actions requiring approval
approval_actions = ['send invoice', 'create payment', 'submit form']
response = hitl_agent.invoke_with_approval(
"Send invoice to client for $10,000",
approval_actions
)
if response['status'] == 'pending_approval':
print(f"Action requires approval: {response['approval_id']}")
print(f"Preview: {response['preview']}")
# Later, after human review
final = hitl_agent.approve_action(response['approval_id'])
print(f"Final response: {final['response']}")3. Agent with Memory
class MemoryEnabledAgent:
def __init__(self, agent: BedrockAgent):
self.agent = agent
self.dynamodb = boto3.resource('dynamodb', region_name='ap-southeast-2')
self.memory_table = self.dynamodb.Table('agent-memory')
def invoke_with_memory(self, user_id: str, query: str) -> dict:
"""Invoke agent with persistent memory."""
# Retrieve user context
context = self._get_user_context(user_id)
# Enhance query with context
enhanced_query = self._enhance_with_context(query, context)
# Invoke agent
response = self.agent.invoke(enhanced_query)
# Update memory
self._update_memory(user_id, query, response)
return response
def _get_user_context(self, user_id: str) -> dict:
"""Retrieve user context from DynamoDB."""
try:
response = self.memory_table.get_item(Key={'user_id': user_id})
if 'Item' in response:
return response['Item']['context']
except:
pass
return {}
def _enhance_with_context(self, query: str, context: dict) -> str:
"""Add relevant context to query."""
if not context:
return query
context_str = f"""User context:
- Business name: {context.get('business_name', 'Not set')}
- ABN: {context.get('abn', 'Not set')}
- Industry: {context.get('industry', 'Not set')}
- Previous queries: {', '.join(context.get('recent_queries', [])[:3])}
Current query: {query}"""
return context_str
def _update_memory(self, user_id: str, query: str, response: dict):
"""Update user memory in DynamoDB."""
# Get existing context
context = self._get_user_context(user_id)
# Update recent queries
recent = context.get('recent_queries', [])
recent.insert(0, query)
recent = recent[:10] # Keep last 10
context['recent_queries'] = recent
# Save to DynamoDB
self.memory_table.put_item(
Item={
'user_id': user_id,
'context': context,
'last_interaction': datetime.utcnow().isoformat()
}
)
# Usage
memory_agent = MemoryEnabledAgent(agent)
# Agent remembers user context
response1 = memory_agent.invoke_with_memory(
'user123',
'My business ABN is 53 004 085 616'
)
# Later query uses remembered ABN
response2 = memory_agent.invoke_with_memory(
'user123',
'Calculate GST for a $1000 invoice' # Agent knows the ABN
)Monitoring and Observability
CloudWatch Integration
import boto3
from datetime import datetime
class AgentMonitoring:
def __init__(self):
self.cloudwatch = boto3.client('cloudwatch', region_name='ap-southeast-2')
self.logs = boto3.client('logs', region_name='ap-southeast-2')
def log_agent_interaction(
self,
agent_id: str,
user_query: str,
response: str,
latency_ms: float,
success: bool
):
"""Log agent interaction to CloudWatch."""
# Custom metrics
self.cloudwatch.put_metric_data(
Namespace='BedrockAgents',
MetricData=[
{
'MetricName': 'Invocations',
'Value': 1,
'Unit': 'Count',
'Timestamp': datetime.utcnow(),
'Dimensions': [
{'Name': 'AgentId', 'Value': agent_id},
{'Name': 'Success', 'Value': str(success)}
]
},
{
'MetricName': 'Latency',
'Value': latency_ms,
'Unit': 'Milliseconds',
'Timestamp': datetime.utcnow(),
'Dimensions': [
{'Name': 'AgentId', 'Value': agent_id}
]
}
]
)
# Detailed logs
log_entry = {
'timestamp': datetime.utcnow().isoformat(),
'agent_id': agent_id,
'user_query': user_query,
'response_preview': response[:200],
'latency_ms': latency_ms,
'success': success
}
self.logs.put_log_events(
logGroupName='/aws/bedrock/agents',
logStreamName=agent_id,
logEvents=[{
'timestamp': int(datetime.utcnow().timestamp() * 1000),
'message': json.dumps(log_entry)
}]
)
def create_dashboard(self, agent_id: str):
"""Create CloudWatch dashboard for agent monitoring."""
dashboard_body = {
"widgets": [
{
"type": "metric",
"properties": {
"metrics": [
["BedrockAgents", "Invocations", {"stat": "Sum"}]
],
"period": 300,
"stat": "Sum",
"region": "ap-southeast-2",
"title": "Agent Invocations"
}
},
{
"type": "metric",
"properties": {
"metrics": [
["BedrockAgents", "Latency", {"stat": "Average"}]
],
"period": 300,
"stat": "Average",
"region": "ap-southeast-2",
"title": "Average Latency"
}
}
]
}
self.cloudwatch.put_dashboard(
DashboardName=f'bedrock-agent-{agent_id}',
DashboardBody=json.dumps(dashboard_body)
)Cost Optimisation
Agent Cost Structure
def calculate_agent_cost(
num_invocations: int,
avg_input_tokens: int,
avg_output_tokens: int,
foundation_model: str = 'claude-v2'
) -> dict:
"""Calculate Bedrock Agent costs (prices in AUD)."""
# Model pricing per 1K tokens (example, converted to AUD)
model_pricing = {
'claude-v2': {
'input': 0.01102,
'output': 0.03306
},
'titan-text': {
'input': 0.0004,
'output': 0.0006
}
}
pricing = model_pricing[foundation_model]
# Calculate model costs
input_cost = (avg_input_tokens / 1000) * pricing['input'] * num_invocations
output_cost = (avg_output_tokens / 1000) * pricing['output'] * num_invocations
# Agent orchestration costs (example)
orchestration_cost = num_invocations * 0.0015 # Per invocation
# Knowledge base query costs (if used)
kb_cost = num_invocations * 0.0003 # Per query
total_cost = input_cost + output_cost + orchestration_cost + kb_cost
return {
'model_cost': round(input_cost + output_cost, 2),
'orchestration_cost': round(orchestration_cost, 2),
'knowledge_base_cost': round(kb_cost, 2),
'total_monthly_cost_aud': round(total_cost, 2)
}
# Example: 10,000 invocations/month
cost = calculate_agent_cost(
num_invocations=10_000,
avg_input_tokens=800,
avg_output_tokens=400,
foundation_model='claude-v2'
)
print(f"Monthly cost: ${cost['total_monthly_cost_aud']} AUD")Optimisation Strategies
1. Caching Common Queries:
from functools import lru_cache
import hashlib
@lru_cache(maxsize=1000)
def get_cached_response(query_hash: str):
# Cache frequently asked questions
pass2. Prompt Optimisation:
# Reduce token usage with concise prompts
# Keep instructions clear but brief3. Right-size Knowledge Base:
# Only include relevant documents
# Remove outdated content
# Optimize document chunkingAustralian Compliance
Privacy and Data Handling
class CompliantAgent:
def __init__(self, agent: BedrockAgent):
self.agent = agent
def invoke_with_pii_filtering(self, query: str) -> dict:
"""Invoke agent with PII filtering."""
# Detect and mask PII before sending
masked_query, pii_detected = self._mask_pii(query)
# Invoke agent
response = self.agent.invoke(masked_query)
# Log PII usage
if pii_detected:
self._log_pii_usage(pii_detected)
return response
def _mask_pii(self, text: str) -> tuple:
"""Detect and mask PII in text."""
import re
pii_patterns = {
'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
'phone': r'\b(?:\+?61|0)[2-478](?:[ -]?[0-9]){8}\b',
'abn': r'\b\d{2}\s?\d{3}\s?\d{3}\s?\d{3}\b'
}
pii_found = []
masked_text = text
for pii_type, pattern in pii_patterns.items():
matches = re.findall(pattern, text)
if matches:
pii_found.append({'type': pii_type, 'count': len(matches)})
masked_text = re.sub(pattern, f'[{pii_type.upper()}]', masked_text)
return masked_text, pii_foundReal-World Examples
Example 1: Customer Support Agent
# Agent that handles Australian e-commerce support
support_agent_instructions = """You are a customer support agent for an Australian e-commerce store.
Handle:
- Order tracking and status
- Returns and refunds (Australian Consumer Law)
- Product information and recommendations
- Delivery timeframes (Australia Post)
- Payment issues
Always:
- Check order status in database
- Validate Australian addresses
- Calculate GST correctly
- Reference Australian Consumer Guarantees
- Provide Australia Post tracking links
"""Example 2: Compliance Assistant
# Agent that helps with Australian business compliance
compliance_agent_instructions = """You are an Australian business compliance assistant.
Help with:
- BAS lodgement reminders
- Fair Work compliance
- Regulatory requirements for regulated entities
- Privacy Act obligations
- Workplace health and safety
Check knowledge base for:
- Current regulations
- Lodgement deadlines
- Compliance checklists
- Required documentation
"""Example 3: Property Management Agent
# Agent for Australian property management
property_agent_instructions = """You are a property management assistant for Australian real estate.
Manage:
- Tenant inquiries
- Lease agreements (state-specific)
- Maintenance requests
- Rental payments and bond
- Inspection scheduling
Apply:
- State residential tenancy laws
- Bond Authority requirements
- Fair Trading regulations
- Rental tribunal decisions
"""Conclusion
Amazon Bedrock Agents enables Australian businesses to build sophisticated AI automation that can understand context, use tools, and take actions autonomously. By combining foundation models, action groups, and knowledge bases, you can create agents that handle complex workflows while maintaining security and compliance.
Key takeaways:
- Agents orchestrate multi-step tasks autonomously
- Action groups define what agents can do
- Knowledge bases provide domain expertise
- Guardrails ensure safe operation
- Monitoring and human-in-the-loop provide control
CloudPoint helps Australian businesses design, implement, and deploy Bedrock Agents for customer support, business automation, and compliance assistance. We ensure your agents meet Australian regulatory requirements while delivering measurable business value.
Contact us for an AI Agent implementation consultation and automate your business workflows with confidence.
Ready to Build AI Agents?
CloudPoint helps Australian businesses implement Amazon Bedrock Agents that automate complex workflows and integrate with your systems. Get in touch to explore AI opportunities.