Introduction: AWS re:Invent 2023 delivered transformative announcements for enterprise AI adoption, with Amazon Bedrock reaching general availability and Amazon Q emerging as AWS’s answer to AI-powered enterprise assistance. These services represent AWS’s strategic vision for making generative AI accessible, secure, and enterprise-ready. After integrating Bedrock into production workloads, I’ve found its model-agnostic approach and native AWS integration dramatically simplify AI deployment while maintaining enterprise security standards. Organizations should evaluate Bedrock for generative AI workloads requiring model flexibility, data privacy, and seamless AWS ecosystem integration.
Amazon Bedrock: Enterprise Foundation Models
Amazon Bedrock provides serverless access to foundation models from leading AI companies including Anthropic, AI21 Labs, Cohere, Meta, and Amazon’s own Titan models. This model-agnostic approach enables organizations to select the best model for each use case without infrastructure management. Bedrock handles scaling, security, and compliance, allowing teams to focus on application development.
Knowledge Bases for Amazon Bedrock enable Retrieval-Augmented Generation (RAG) without managing vector databases or embedding pipelines. Connect enterprise data sources, and Bedrock automatically handles chunking, embedding, and retrieval. This managed RAG capability dramatically reduces time-to-production for knowledge-grounded AI applications.
Agents for Amazon Bedrock orchestrate multi-step tasks by combining foundation models with enterprise APIs. Define actions and knowledge sources, and agents autonomously plan and execute complex workflows. This capability enables sophisticated automation scenarios from customer service to data analysis without custom orchestration code.
Amazon Q: AI-Powered Enterprise Assistant
Amazon Q represents AWS’s vision for enterprise AI assistance, offering specialized capabilities for developers, business users, and IT administrators. Q for Developer integrates with IDEs to provide code generation, explanation, and transformation. Q for Business connects to enterprise data sources for knowledge-grounded responses. Q for IT assists with AWS resource management and troubleshooting.
The security model for Amazon Q addresses enterprise concerns about AI data handling. Q operates within AWS’s security perimeter, respecting IAM permissions and data boundaries. Conversations and data remain within the customer’s AWS environment, addressing compliance requirements for regulated industries.
Python Implementation: Building with Amazon Bedrock
Here’s a comprehensive implementation demonstrating Amazon Bedrock integration patterns:
"""Amazon Bedrock Production Integration Patterns"""
import json
import asyncio
import logging
from typing import Dict, Any, List, Optional, AsyncIterator
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import boto3
from botocore.config import Config
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# ==================== Configuration ====================
class BedrockModel(Enum):
"""Available Bedrock foundation models."""
CLAUDE_3_SONNET = "anthropic.claude-3-sonnet-20240229-v1:0"
CLAUDE_3_HAIKU = "anthropic.claude-3-haiku-20240307-v1:0"
CLAUDE_INSTANT = "anthropic.claude-instant-v1"
TITAN_TEXT_EXPRESS = "amazon.titan-text-express-v1"
TITAN_TEXT_LITE = "amazon.titan-text-lite-v1"
LLAMA2_70B = "meta.llama2-70b-chat-v1"
COHERE_COMMAND = "cohere.command-text-v14"
@dataclass
class BedrockConfig:
"""Configuration for Bedrock client."""
region: str = "us-east-1"
model_id: str = BedrockModel.CLAUDE_3_SONNET.value
max_tokens: int = 4096
temperature: float = 0.7
top_p: float = 0.9
retry_attempts: int = 3
timeout: int = 60
@dataclass
class Message:
"""Chat message structure."""
role: str # "user" or "assistant"
content: str
timestamp: datetime = field(default_factory=datetime.utcnow)
@dataclass
class ConversationContext:
"""Maintains conversation state."""
conversation_id: str
messages: List[Message] = field(default_factory=list)
system_prompt: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
def add_message(self, role: str, content: str) -> None:
"""Add a message to the conversation."""
self.messages.append(Message(role=role, content=content))
def get_messages_for_api(self) -> List[Dict[str, str]]:
"""Format messages for Bedrock API."""
return [
{"role": msg.role, "content": msg.content}
for msg in self.messages
]
# ==================== Bedrock Client ====================
class BedrockClient:
"""Production-ready Amazon Bedrock client."""
def __init__(self, config: BedrockConfig):
self.config = config
boto_config = Config(
region_name=config.region,
retries={'max_attempts': config.retry_attempts},
connect_timeout=config.timeout,
read_timeout=config.timeout
)
self.client = boto3.client(
'bedrock-runtime',
config=boto_config
)
self.bedrock_agent = boto3.client(
'bedrock-agent-runtime',
config=boto_config
)
def invoke_model(
self,
prompt: str,
system_prompt: Optional[str] = None,
**kwargs
) -> str:
"""Invoke a foundation model with a single prompt."""
messages = [{"role": "user", "content": prompt}]
body = self._build_request_body(messages, system_prompt, **kwargs)
response = self.client.invoke_model(
modelId=self.config.model_id,
body=json.dumps(body),
contentType="application/json",
accept="application/json"
)
response_body = json.loads(response['body'].read())
return self._extract_response_text(response_body)
def invoke_with_conversation(
self,
context: ConversationContext,
user_message: str,
**kwargs
) -> str:
"""Invoke model with conversation context."""
context.add_message("user", user_message)
body = self._build_request_body(
context.get_messages_for_api(),
context.system_prompt,
**kwargs
)
response = self.client.invoke_model(
modelId=self.config.model_id,
body=json.dumps(body),
contentType="application/json",
accept="application/json"
)
response_body = json.loads(response['body'].read())
assistant_message = self._extract_response_text(response_body)
context.add_message("assistant", assistant_message)
return assistant_message
def invoke_model_streaming(
self,
prompt: str,
system_prompt: Optional[str] = None,
**kwargs
) -> AsyncIterator[str]:
"""Stream responses from the model."""
messages = [{"role": "user", "content": prompt}]
body = self._build_request_body(messages, system_prompt, **kwargs)
response = self.client.invoke_model_with_response_stream(
modelId=self.config.model_id,
body=json.dumps(body),
contentType="application/json",
accept="application/json"
)
for event in response['body']:
chunk = json.loads(event['chunk']['bytes'])
if 'delta' in chunk and 'text' in chunk['delta']:
yield chunk['delta']['text']
def _build_request_body(
self,
messages: List[Dict[str, str]],
system_prompt: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""Build request body based on model type."""
model_id = self.config.model_id
if "anthropic" in model_id:
return self._build_anthropic_body(messages, system_prompt, **kwargs)
elif "amazon.titan" in model_id:
return self._build_titan_body(messages, system_prompt, **kwargs)
elif "meta.llama" in model_id:
return self._build_llama_body(messages, system_prompt, **kwargs)
elif "cohere" in model_id:
return self._build_cohere_body(messages, system_prompt, **kwargs)
else:
raise ValueError(f"Unsupported model: {model_id}")
def _build_anthropic_body(
self,
messages: List[Dict[str, str]],
system_prompt: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""Build request body for Anthropic Claude models."""
body = {
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": kwargs.get("max_tokens", self.config.max_tokens),
"temperature": kwargs.get("temperature", self.config.temperature),
"top_p": kwargs.get("top_p", self.config.top_p),
"messages": messages
}
if system_prompt:
body["system"] = system_prompt
return body
def _build_titan_body(
self,
messages: List[Dict[str, str]],
system_prompt: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""Build request body for Amazon Titan models."""
# Combine messages into single prompt for Titan
prompt_parts = []
if system_prompt:
prompt_parts.append(f"System: {system_prompt}")
for msg in messages:
role = "Human" if msg["role"] == "user" else "Assistant"
prompt_parts.append(f"{role}: {msg['content']}")
prompt_parts.append("Assistant:")
return {
"inputText": "\n\n".join(prompt_parts),
"textGenerationConfig": {
"maxTokenCount": kwargs.get("max_tokens", self.config.max_tokens),
"temperature": kwargs.get("temperature", self.config.temperature),
"topP": kwargs.get("top_p", self.config.top_p)
}
}
def _build_llama_body(
self,
messages: List[Dict[str, str]],
system_prompt: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""Build request body for Meta Llama models."""
prompt_parts = []
if system_prompt:
prompt_parts.append(f"[INST] <>\n{system_prompt}\n< >")
for i, msg in enumerate(messages):
if msg["role"] == "user":
if i == 0 and not system_prompt:
prompt_parts.append(f"[INST] {msg['content']} [/INST]")
else:
prompt_parts.append(f"{msg['content']} [/INST]")
else:
prompt_parts.append(f"{msg['content']} [INST] ")
return {
"prompt": "".join(prompt_parts),
"max_gen_len": kwargs.get("max_tokens", self.config.max_tokens),
"temperature": kwargs.get("temperature", self.config.temperature),
"top_p": kwargs.get("top_p", self.config.top_p)
}
def _build_cohere_body(
self,
messages: List[Dict[str, str]],
system_prompt: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""Build request body for Cohere models."""
# Extract last user message as prompt
prompt = messages[-1]["content"] if messages else ""
return {
"prompt": prompt,
"max_tokens": kwargs.get("max_tokens", self.config.max_tokens),
"temperature": kwargs.get("temperature", self.config.temperature),
"p": kwargs.get("top_p", self.config.top_p)
}
def _extract_response_text(self, response_body: Dict[str, Any]) -> str:
"""Extract text from model response based on model type."""
model_id = self.config.model_id
if "anthropic" in model_id:
return response_body.get("content", [{}])[0].get("text", "")
elif "amazon.titan" in model_id:
return response_body.get("results", [{}])[0].get("outputText", "")
elif "meta.llama" in model_id:
return response_body.get("generation", "")
elif "cohere" in model_id:
return response_body.get("generations", [{}])[0].get("text", "")
return ""
# ==================== Knowledge Base Integration ====================
class KnowledgeBaseClient:
"""Client for Bedrock Knowledge Bases (RAG)."""
def __init__(
self,
knowledge_base_id: str,
model_arn: str,
region: str = "us-east-1"
):
self.knowledge_base_id = knowledge_base_id
self.model_arn = model_arn
self.client = boto3.client(
'bedrock-agent-runtime',
region_name=region
)
def retrieve(
self,
query: str,
num_results: int = 5
) -> List[Dict[str, Any]]:
"""Retrieve relevant documents from knowledge base."""
response = self.client.retrieve(
knowledgeBaseId=self.knowledge_base_id,
retrievalQuery={"text": query},
retrievalConfiguration={
"vectorSearchConfiguration": {
"numberOfResults": num_results
}
}
)
results = []
for result in response.get("retrievalResults", []):
results.append({
"content": result.get("content", {}).get("text", ""),
"score": result.get("score", 0.0),
"location": result.get("location", {}),
"metadata": result.get("metadata", {})
})
return results
def retrieve_and_generate(
self,
query: str,
session_id: Optional[str] = None
) -> Dict[str, Any]:
"""Retrieve context and generate response."""
request = {
"input": {"text": query},
"retrieveAndGenerateConfiguration": {
"type": "KNOWLEDGE_BASE",
"knowledgeBaseConfiguration": {
"knowledgeBaseId": self.knowledge_base_id,
"modelArn": self.model_arn
}
}
}
if session_id:
request["sessionId"] = session_id
response = self.client.retrieve_and_generate(**request)
return {
"output": response.get("output", {}).get("text", ""),
"session_id": response.get("sessionId"),
"citations": response.get("citations", [])
}
# ==================== Bedrock Agents ====================
class BedrockAgentClient:
"""Client for Bedrock Agents."""
def __init__(
self,
agent_id: str,
agent_alias_id: str,
region: str = "us-east-1"
):
self.agent_id = agent_id
self.agent_alias_id = agent_alias_id
self.client = boto3.client(
'bedrock-agent-runtime',
region_name=region
)
def invoke_agent(
self,
prompt: str,
session_id: str,
enable_trace: bool = False
) -> Dict[str, Any]:
"""Invoke a Bedrock agent."""
response = self.client.invoke_agent(
agentId=self.agent_id,
agentAliasId=self.agent_alias_id,
sessionId=session_id,
inputText=prompt,
enableTrace=enable_trace
)
# Process streaming response
completion = ""
traces = []
for event in response.get("completion", []):
if "chunk" in event:
chunk_data = event["chunk"]
if "bytes" in chunk_data:
completion += chunk_data["bytes"].decode("utf-8")
if enable_trace and "trace" in event:
traces.append(event["trace"])
return {
"completion": completion,
"traces": traces,
"session_id": session_id
}
# ==================== High-Level Application ====================
class EnterpriseAIAssistant:
"""Enterprise AI assistant using Bedrock."""
def __init__(
self,
bedrock_config: BedrockConfig,
knowledge_base_id: Optional[str] = None,
model_arn: Optional[str] = None
):
self.bedrock = BedrockClient(bedrock_config)
self.knowledge_base = None
if knowledge_base_id and model_arn:
self.knowledge_base = KnowledgeBaseClient(
knowledge_base_id,
model_arn,
bedrock_config.region
)
self.conversations: Dict[str, ConversationContext] = {}
def get_or_create_conversation(
self,
conversation_id: str,
system_prompt: Optional[str] = None
) -> ConversationContext:
"""Get existing or create new conversation."""
if conversation_id not in self.conversations:
self.conversations[conversation_id] = ConversationContext(
conversation_id=conversation_id,
system_prompt=system_prompt or self._default_system_prompt()
)
return self.conversations[conversation_id]
def _default_system_prompt(self) -> str:
"""Default system prompt for the assistant."""
return """You are an enterprise AI assistant. Provide accurate,
helpful responses while maintaining professional tone.
If you're unsure about something, acknowledge the uncertainty.
Always cite sources when using retrieved information."""
def chat(
self,
conversation_id: str,
user_message: str,
use_knowledge_base: bool = False
) -> str:
"""Process a chat message."""
context = self.get_or_create_conversation(conversation_id)
if use_knowledge_base and self.knowledge_base:
# Retrieve relevant context
retrieved = self.knowledge_base.retrieve(user_message, num_results=3)
# Augment prompt with retrieved context
context_text = "\n\n".join([
f"Source: {r['content']}" for r in retrieved
])
augmented_message = f"""Based on the following context:
{context_text}
User question: {user_message}
Please provide a response based on the context above."""
return self.bedrock.invoke_with_conversation(
context,
augmented_message
)
return self.bedrock.invoke_with_conversation(context, user_message)
def summarize_document(self, document: str) -> str:
"""Summarize a document."""
prompt = f"""Please provide a concise summary of the following document:
{document}
Summary:"""
return self.bedrock.invoke_model(
prompt,
system_prompt="You are a document summarization expert. Provide clear, concise summaries."
)
def analyze_sentiment(self, text: str) -> Dict[str, Any]:
"""Analyze sentiment of text."""
prompt = f"""Analyze the sentiment of the following text and respond in JSON format:
Text: {text}
Respond with JSON containing:
- sentiment: "positive", "negative", or "neutral"
- confidence: 0.0 to 1.0
- key_phrases: list of important phrases
- summary: brief explanation"""
response = self.bedrock.invoke_model(
prompt,
system_prompt="You are a sentiment analysis expert. Always respond in valid JSON."
)
try:
return json.loads(response)
except json.JSONDecodeError:
return {"raw_response": response}
# ==================== Example Usage ====================
def main():
"""Demonstrate Bedrock integration."""
# Initialize configuration
config = BedrockConfig(
region="us-east-1",
model_id=BedrockModel.CLAUDE_3_SONNET.value,
max_tokens=2048,
temperature=0.7
)
# Create assistant
assistant = EnterpriseAIAssistant(config)
# Simple chat
response = assistant.chat(
"session_001",
"What are the key benefits of using Amazon Bedrock for enterprise AI?"
)
print(f"Response: {response}")
# Document summarization
document = """
Amazon Bedrock is a fully managed service that offers a choice of
high-performing foundation models from leading AI companies...
"""
summary = assistant.summarize_document(document)
print(f"Summary: {summary}")
# Sentiment analysis
sentiment = assistant.analyze_sentiment(
"The new Bedrock features are amazing! Really impressed with the performance."
)
print(f"Sentiment: {sentiment}")
if __name__ == "__main__":
main()Terraform Infrastructure for Bedrock
Here’s Terraform configuration for deploying Bedrock infrastructure:
# Bedrock Infrastructure with Terraform
# Provider configuration
terraform {
required_providers {
aws = {
source = "hashicorp/aws"
version = "~> 5.0"
}
}
}
provider "aws" {
region = var.aws_region
}
# Variables
variable "aws_region" {
default = "us-east-1"
}
variable "environment" {
default = "production"
}
# S3 bucket for Knowledge Base data
resource "aws_s3_bucket" "knowledge_base_data" {
bucket = "bedrock-kb-data-${var.environment}-${random_id.suffix.hex}"
}
resource "random_id" "suffix" {
byte_length = 4
}
resource "aws_s3_bucket_versioning" "knowledge_base_data" {
bucket = aws_s3_bucket.knowledge_base_data.id
versioning_configuration {
status = "Enabled"
}
}
# IAM role for Bedrock Knowledge Base
resource "aws_iam_role" "bedrock_kb_role" {
name = "bedrock-kb-role-${var.environment}"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = {
Service = "bedrock.amazonaws.com"
}
}
]
})
}
resource "aws_iam_role_policy" "bedrock_kb_policy" {
name = "bedrock-kb-policy"
role = aws_iam_role.bedrock_kb_role.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Effect = "Allow"
Action = [
"s3:GetObject",
"s3:ListBucket"
]
Resource = [
aws_s3_bucket.knowledge_base_data.arn,
"${aws_s3_bucket.knowledge_base_data.arn}/*"
]
},
{
Effect = "Allow"
Action = [
"bedrock:InvokeModel"
]
Resource = "*"
}
]
})
}
# Lambda function for Bedrock integration
resource "aws_lambda_function" "bedrock_handler" {
filename = "bedrock_handler.zip"
function_name = "bedrock-handler-${var.environment}"
role = aws_iam_role.lambda_bedrock_role.arn
handler = "handler.lambda_handler"
runtime = "python3.12"
timeout = 60
memory_size = 512
environment {
variables = {
BEDROCK_MODEL_ID = "anthropic.claude-3-sonnet-20240229-v1:0"
ENVIRONMENT = var.environment
}
}
}
resource "aws_iam_role" "lambda_bedrock_role" {
name = "lambda-bedrock-role-${var.environment}"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = {
Service = "lambda.amazonaws.com"
}
}
]
})
}
resource "aws_iam_role_policy_attachment" "lambda_basic" {
role = aws_iam_role.lambda_bedrock_role.name
policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
}
resource "aws_iam_role_policy" "lambda_bedrock_policy" {
name = "lambda-bedrock-policy"
role = aws_iam_role.lambda_bedrock_role.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Effect = "Allow"
Action = [
"bedrock:InvokeModel",
"bedrock:InvokeModelWithResponseStream"
]
Resource = "*"
}
]
})
}
# Outputs
output "knowledge_base_bucket" {
value = aws_s3_bucket.knowledge_base_data.bucket
}
output "lambda_function_arn" {
value = aws_lambda_function.bedrock_handler.arn
}
Key Takeaways and Implementation Strategy
AWS re:Invent 2023 positioned Bedrock as the enterprise foundation for generative AI. The model-agnostic approach provides flexibility, while managed RAG and agents reduce implementation complexity. Amazon Q extends these capabilities to specific enterprise personas, from developers to business analysts.
For implementation, start with direct model invocation for simple use cases, then progress to Knowledge Bases for document-grounded applications. Evaluate agents for complex multi-step workflows requiring external integrations. The native AWS integration ensures security and compliance requirements are met without additional infrastructure.
Discover more from Code, Cloud & Context
Subscribe to get the latest posts sent to your email.