AI Architect / AI Platform Engineer

The gateway is the single entry point for all AI requests. It handles authentication, rate limiting, request validation, and model routing.

AWS Implementation

The orchestration layer manages multi-step agent workflows with state management, tool selection, and human-in-the-loop capabilities.

Key Decisions

The knowledge layer handles document ingestion, chunking, embedding, vector storage, and retrieval with re-ranking.

Production Pipeline

A cross-cutting concern that spans all layers. Implements guardrails, audit logging, encryption, and monitoring.

Three-Layer Safety

Break complex queries into sub-queries. Example: "Tax implications of 2008 crisis on small businesses" becomes: ["2008 financial crisis causes", "tax policy changes 2008", "small business impact"]. Retrieve for each, combine results.

"Machine learning models" expands to: ["ML models", "neural networks", "deep learning", "AI algorithms"]. Improves recall when document terminology varies.

Generate a hypothetical relevant document for the query, then use its embedding for retrieval instead of the query embedding. Bridges the vocabulary gap between queries and documents.

Production Variant Configuration

import sagemaker
from sagemaker.model import Model

model = Model(
    image_uri="123456789.dkr.ecr.us-east-1.amazonaws.com/my-model:latest",
    model_data="s3://bucket/model.tar.gz",
    role="arn:aws:iam::123456789:role/SageMakerRole"
)

# Deploy with two variants for A/B testing
predictor = model.deploy(
    initial_instance_count=2,
    instance_type="ml.g5.xlarge",
    endpoint_name="my-model-endpoint",
    data_capture_config=DataCaptureConfig(
        enable_capture=True,
        sampling_percentage=100,         # Capture all for monitoring
        destination_s3_uri="s3://bucket/capture/"
    )
)

# Async endpoint config — scales to 0 when idle
async_config = AsyncInferenceConfig(
    output_path="s3://bucket/async-output/",
    max_concurrent_invocations_per_instance=4,
    notification_config=AsyncInferenceNotificationConfig(
        success_topic="arn:aws:sns:us-east-1:123:success-topic",
        error_topic="arn:aws:sns:us-east-1:123:error-topic"
    )
)

model.deploy(
    async_inference_config=async_config,
    instance_type="ml.g5.2xlarge",
    initial_instance_count=1
)

MME is ideal for multi-tenant AI platforms where each tenant has a fine-tuned model. Instead of one endpoint per tenant ($$$), you serve all tenants from a shared fleet and SageMaker handles model loading/unloading based on traffic patterns.

# Invoke a specific model on a multi-model endpoint
response = sm_runtime.invoke_endpoint(
    EndpointName="multi-model-endpoint",
    TargetModel="tenant-a/model.tar.gz",  # S3 key for this tenant's model
    Body=json.dumps(payload),
    ContentType="application/json"
)

SageMaker Inference Recommender benchmarks your model across different instance types and configurations to find the optimal cost/performance trade-off.

# Run inference recommender to find optimal instance
response = sm_client.create_inference_recommendations_job(
    JobName="my-model-benchmark",
    JobType="Default",  # or "Advanced" for custom traffic patterns
    RoleArn=role_arn,
    InputConfig={
        "ModelPackageVersionArn": model_package_arn,
        "JobDurationInSeconds": 3600
    }
)
# Returns: ranked list of instance types with latency, throughput, cost
# Example: ml.g5.xlarge — P95: 120ms, 40 req/s, $1.41/hr
#          ml.g5.2xlarge — P95: 85ms, 65 req/s, $2.36/hr

from sagemaker.estimator import Estimator

estimator = Estimator(
    image_uri="763104351884.dkr.ecr.us-east-1.amazonaws.com/pytorch-training:2.1-gpu",
    role=role,
    instance_count=4,                   # 4 instances for distributed training
    instance_type="ml.p4d.24xlarge",     # 8x A100 GPUs each = 32 GPUs total
    use_spot_instances=True,             # 70% cost savings
    max_wait=86400,                      # Max 24hr including spot interruptions
    max_run=72000,                       # Max 20hr actual training
    checkpoint_s3_uri="s3://bucket/checkpoints/",
    output_path="s3://bucket/output/",
    hyperparameters={
        "epochs": "10",
        "learning_rate": "5e-5",
        "batch_size": "32",
        "model_name": "bert-base-uncased"
    },
    distribution={
        "torch_distributed": {"enabled": True}  # PyTorch DDP
    },
    metric_definitions=[
        {"Name": "train:loss", "Regex": "train_loss: ([0-9\\.]+)"},
        {"Name": "eval:accuracy", "Regex": "eval_acc: ([0-9\\.]+)"}
    ]
)

estimator.fit({
    "training": "s3://bucket/train/",
    "validation": "s3://bucket/val/"
})

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.step_collections import RegisterModel

# Step 1: Data Processing
processing_step = ProcessingStep(
    name="PreprocessData",
    processor=sklearn_processor,
    code="preprocess.py",
    inputs=[ProcessingInput(source=input_data, destination="/opt/ml/processing/input")],
    outputs=[ProcessingOutput(output_name="train"), ProcessingOutput(output_name="test")]
)

# Step 2: Training
training_step = TrainingStep(
    name="TrainModel",
    estimator=estimator,
    inputs={"train": TrainingInput(s3_data=processing_step.properties.ProcessingOutputConfig
        .Outputs["train"].S3Output.S3Uri)}
)

# Step 3: Evaluation
eval_step = ProcessingStep(
    name="EvaluateModel",
    processor=sklearn_processor,
    code="evaluate.py",
    inputs=[ProcessingInput(source=training_step.properties.ModelArtifacts.S3ModelArtifacts,
        destination="/opt/ml/processing/model")],
    property_files=[PropertyFile(name="evaluation", output_name="metrics",
        path="evaluation.json")]
)

# Step 4: Conditional deployment
cond = ConditionGreaterThanOrEqualTo(
    left=JsonGet(step_name="EvaluateModel", property_file="evaluation",
        json_path="accuracy"),
    right=0.85
)

register_step = RegisterModel(
    name="RegisterModel",
    estimator=estimator,
    model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
    content_types=["application/json"],
    approval_status="PendingManualApproval"  # Requires human sign-off
)

condition_step = ConditionStep(
    name="CheckAccuracy",
    conditions=[cond],
    if_steps=[register_step],     # Deploy if accurate enough
    else_steps=[FailStep(name="ModelNotGoodEnough")]
)

# Assemble pipeline
pipeline = Pipeline(
    name="AI-Hub-Training-Pipeline",
    steps=[processing_step, training_step, eval_step, condition_step],
    sagemaker_session=session
)

pipeline.upsert(role_arn=role)
pipeline.start()

from sagemaker.feature_store.feature_group import FeatureGroup

# Define feature group
feature_group = FeatureGroup(name="customer-features", sagemaker_session=session)
feature_group.load_feature_definitions(data_frame=df)

# Create with both online + offline stores
feature_group.create(
    s3_uri="s3://bucket/feature-store/",
    record_identifier_name="customer_id",
    event_time_feature_name="event_time",
    role_arn=role,
    enable_online_store=True    # DynamoDB for real-time
)

# Ingest features
feature_group.ingest(data_frame=df, max_workers=4, wait=True)

# Real-time lookup (online store)
record = featurestore_runtime.get_record(
    FeatureGroupName="customer-features",
    RecordIdentifierValueAsString="CUST-12345"
)

# Historical query for training (offline store — Athena)
query = feature_group.athena_query()
query.run(
    query_string="""
        SELECT * FROM "customer-features"
        WHERE event_time BETWEEN '2025-01-01' AND '2025-12-31'
    """,
    output_location="s3://bucket/athena-results/"
)

from sagemaker.model_monitor import DefaultModelMonitor, DataCaptureConfig

# Step 1: Enable Data Capture on endpoint
data_capture_config = DataCaptureConfig(
    enable_capture=True,
    sampling_percentage=100,
    destination_s3_uri="s3://bucket/data-capture/",
    capture_options=["Input", "Output"]  # Capture both request + response
)

# Step 2: Create baseline from training data
monitor = DefaultModelMonitor(role=role, instance_type="ml.m5.xlarge")
monitor.suggest_baseline(
    baseline_dataset="s3://bucket/baseline-data/train.csv",
    dataset_format=DatasetFormat.csv(header=True)
)

# Step 3: Schedule monitoring (hourly)
monitor.create_monitoring_schedule(
    monitor_schedule_name="fraud-model-monitor",
    endpoint_input=EndpointInput(
        endpoint_name="fraud-detection-endpoint",
        destination="/opt/ml/processing/input"
    ),
    output_s3_uri="s3://bucket/monitoring-output/",
    schedule_cron_expression="cron(0 * ? * * *)"  # Every hour
)

# Violations trigger CloudWatch alarms → SNS → PagerDuty

SageMaker Clarify uses SHAP (SHapley Additive exPlanations) to explain individual predictions. This is critical for BFSI compliance (GDPR right to explanation).

from sagemaker import clarify

shap_config = clarify.SHAPConfig(
    baseline=[[0, 0, 0, 0]],  # Reference point for SHAP
    num_samples=500,           # Number of perturbations
    agg_method="mean_abs"      # Aggregation for feature importance
)

clarify_processor = clarify.SageMakerClarifyProcessor(
    role=role, instance_count=1, instance_type="ml.m5.xlarge"
)

clarify_processor.run_explainability(
    data_config=data_config,
    model_config=model_config,
    explainability_config=shap_config
)
# Output: per-feature importance scores for each prediction
# "This loan was denied primarily due to: credit_score (45%),
#  debt_to_income (30%), employment_length (15%)"

from sagemaker.jumpstart.model import JumpStartModel

# Deploy Llama 3.1 8B in one line
model = JumpStartModel(
    model_id="meta-textgeneration-llama-3-1-8b-instruct",
    instance_type="ml.g5.2xlarge",
    role=role
)

predictor = model.deploy(
    initial_instance_count=1,
    endpoint_name="llama-3-1-endpoint"
)

# Fine-tune with your data
model.fit({
    "training": "s3://bucket/fine-tune-data/"
})

# Hugging Face Estimator with PEFT/LoRA
from sagemaker.huggingface import HuggingFace

huggingface_estimator = HuggingFace(
    entry_point="train_lora.py",
    instance_type="ml.g5.12xlarge",  # 4x A10G GPUs
    instance_count=1,
    transformers_version="4.36",
    pytorch_version="2.1",
    py_version="py310",
    role=role,
    hyperparameters={
        "model_id": "meta-llama/Llama-3.1-8B-Instruct",
        "lora_r": "16",          # LoRA rank (lower = faster, less capacity)
        "lora_alpha": "32",     # LoRA scaling factor
        "lora_dropout": "0.05",
        "epochs": "3",
        "learning_rate": "2e-4",
        "per_device_train_batch_size": "4",
        "gradient_accumulation_steps": "4",  # Effective batch = 16
    }
)

huggingface_estimator.fit({"training": "s3://bucket/lora-data/"})

{
    "Effect": "Allow",
    "Action": "bedrock:InvokeAgent",
    "Resource": "arn:aws:bedrock:*:*:agent/agent-123",
    "Condition": {
        "StringEquals": {
            "bedrock:tenant": "tenant-a"
        }
    }
}

def handle_request(event):
    tenant_id = event['requestContext']['authorizer']['claims']['tenant']
    # Validate tenant_id matches request
    if event.get('body', {}).get('tenant_id') != tenant_id:
        raise UnauthorizedException("Tenant mismatch")
    # Inject tenant context into all downstream calls
    result = bedrock_client.invoke_agent(
        agentId='agent-123',
        sessionStateValues={'tenant_id': tenant_id}
    )

-- Row-level security policy
CREATE POLICY tenant_isolation ON conversations
    USING (tenant_id = current_setting('app.current_tenant'));

# K8s Deployment for multi-GPU training
kind: Deployment
spec:
  replicas: 4
  template:
    spec:
      containers:
      - name: trainer
        resources:
          limits:
            nvidia.com/gpu: 8
        env:
        - name: MASTER_ADDR
          value: master-service.default

Scale inference pods based on CPU/memory/custom metrics. Set minReplicas: 2 for availability, maxReplicas: 50 for burst, target 70% CPU utilization.

Use node affinity to schedule GPU workloads on nvidia-a100 nodes. Combine with Karpenter spot instances for 70% cost reduction. Handle spot interruptions with graceful migration.