Module 5: AWS Services for AI (S3, Lambda, SageMaker Basics)

4. S3 + Lambda Integration

Next →

Overview

In this capsule you'll connect S3 and Lambda in a complete event-driven flow: S3 trigger → Lambda → processes → S3. Until now you worked with S3 (capsule 02) and Lambda (capsule 03) separately. Here you combine them. Lambda reads a prompt template from S3, takes a new document that fired the trigger, builds a contextualized prompt, invokes the LLM, and writes the response back to S3. It's the fundamental pattern of an AI service on AWS.

Context: In Module 4 you built a pipeline in LocalStack that integrated S3 and Lambda. Now you go deeper into the real integration patterns: S3 Event Notifications, the format of the event Lambda receives, how to avoid infinite loops (Lambda writes to S3 → S3 triggers Lambda again), and how to design flows that scale. By the end, you'll have an event-driven pipeline that processes documents automatically.

S3 Event Notifications → Lambda

How the trigger works

When an object is created (or modified, deleted) in S3, you can configure S3 to send an event to Lambda. Lambda runs automatically with the object's information:

                    ┌──────────────────────┐
  Upload doc ──────→│ S3 Bucket            │
  PUT documents/    │ ai-assets-xxx/       │
                    │ documents/new-doc.json│
                    └──────────┬───────────┘
                               │ S3 Event Notification
                               │ (s3:ObjectCreated:*)
                               ↓
                    ┌──────────────────────┐
                    │ Lambda Function      │
                    │ 1. Read S3 event     │
                    │ 2. Read doc from S3  │
                    │ 3. Read prompt (S3)  │
                    │ 4. Invoke LLM        │
                    │ 5. Write to S3       │
                    └──────────┬───────────┘
                               │ put_object
                               ↓
                    ┌──────────────────────┐
                    │ S3 Bucket            │
                    │ responses/2026/03/08/│
                    │ resp-abc123.json     │
                    └──────────────────────┘

Configure an S3 Event Notification with boto3

import boto3
import json
import os

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
lambda_client = boto3.client("lambda", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))

BUCKET = "ai-assets-dev"
LAMBDA_ARN = os.environ.get(
    "PROCESSOR_LAMBDA_ARN",
    "arn:aws:lambda:us-east-1:000000000000:function:ai-doc-processor"
)


def configure_s3_trigger(bucket: str, lambda_arn: str, prefix: str) -> dict:
    """Configures S3 to trigger Lambda when a file is uploaded."""
    response = s3.put_bucket_notification_configuration(
        Bucket=bucket,
        NotificationConfiguration={
            "LambdaFunctionConfigurations": [
                {
                    "Id": "process-new-documents",
                    "LambdaFunctionArn": lambda_arn,
                    "Events": ["s3:ObjectCreated:*"],
                    "Filter": {
                        "Key": {
                            "FilterRules": [
                                {"Name": "prefix", "Value": prefix},
                                {"Name": "suffix", "Value": ".json"},
                            ]
                        }
                    },
                }
            ]
        },
    )
    print(f"Trigger configured: {bucket}/{prefix}*.json → {lambda_arn}")
    return response


configure_s3_trigger(BUCKET, LAMBDA_ARN, "documents/inbox/")

The S3 event Lambda receives

When S3 triggers Lambda, the event has this structure:

# Event Lambda receives when a file is uploaded to S3
event = {
    "Records": [
        {
            "eventVersion": "2.1",
            "eventSource": "aws:s3",
            "awsRegion": "us-east-1",
            "eventTime": "2026-03-08T12:00:00.000Z",
            "eventName": "ObjectCreated:Put",
            "s3": {
                "bucket": {
                    "name": "ai-assets-dev",
                    "arn": "arn:aws:s3:::ai-assets-dev",
                },
                "object": {
                    "key": "documents/inbox/new-report.json",
                    "size": 4567,
                    "eTag": "abc123def456",
                },
            },
        }
    ]
}

Parse the S3 event in Lambda

from urllib.parse import unquote_plus


def parse_s3_event(event: dict) -> list[dict]:
    """Extracts bucket and key information from an S3 event."""
    records = []
    for record in event.get("Records", []):
        s3_info = record.get("s3", {})
        bucket = s3_info.get("bucket", {}).get("name", "")
        key = unquote_plus(s3_info.get("object", {}).get("key", ""))
        size = s3_info.get("object", {}).get("size", 0)

        records.append({
            "bucket": bucket,
            "key": key,
            "size": size,
            "event_name": record.get("eventName", ""),
            "event_time": record.get("eventTime", ""),
            "region": record.get("awsRegion", ""),
        })

    return records

Handler: S3 Trigger → Process with LLM → S3

Complete flow

# s3_trigger_handler.py — Lambda that processes S3 documents with an LLM
import json
import logging
import os
import time
from datetime import datetime
from urllib.parse import unquote_plus

import boto3
from openai import OpenAI

logger = logging.getLogger()
logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", ""))

BUCKET = os.environ.get("AI_BUCKET", "ai-assets-dev")
PROMPT_KEY = os.environ.get("PROMPT_KEY", "prompts/summarizer/v1/system.txt")
RESPONSE_PREFIX = os.environ.get("RESPONSE_PREFIX", "responses/")
MODEL = os.environ.get("MODEL_NAME", "gpt-4o-mini")


def get_prompt_template() -> str:
    """Reads the prompt template from S3."""
    response = s3.get_object(Bucket=BUCKET, Key=PROMPT_KEY)
    return response["Body"].read().decode("utf-8")


def get_document(bucket: str, key: str) -> dict:
    """Reads a JSON document from S3."""
    response = s3.get_object(Bucket=bucket, Key=key)
    content = response["Body"].read().decode("utf-8")
    return json.loads(content)


def save_response(request_id: str, response_data: dict) -> str:
    """Saves the inference response to S3."""
    now = datetime.utcnow()
    key = f"{RESPONSE_PREFIX}{now.strftime('%Y/%m/%d')}/{request_id}.json"

    s3.put_object(
        Bucket=BUCKET,
        Key=key,
        Body=json.dumps(response_data, ensure_ascii=False).encode("utf-8"),
        ContentType="application/json",
    )
    return key


def process_document(document: dict, prompt_template: str) -> dict:
    """Processes a document using the LLM with the prompt template."""
    doc_content = document.get("content", "")
    doc_title = document.get("title", "Untitled")

    full_prompt = f"{prompt_template}\n\nDocument: {doc_title}\n\n{doc_content}"

    response = openai_client.chat.completions.create(
        model=MODEL,
        messages=[{"role": "user", "content": full_prompt}],
        max_tokens=1000,
    )

    return {
        "answer": response.choices[0].message.content,
        "model": response.model,
        "tokens_used": response.usage.total_tokens,
        "input_tokens": response.usage.prompt_tokens,
        "output_tokens": response.usage.completion_tokens,
    }


def handler(event, context):
    """Lambda triggered by S3 that processes documents with an LLM."""
    start_time = time.time()
    results = []

    prompt_template = get_prompt_template()

    for record in event.get("Records", []):
        s3_info = record.get("s3", {})
        source_bucket = s3_info.get("bucket", {}).get("name", "")
        source_key = unquote_plus(s3_info.get("object", {}).get("key", ""))

        request_id = f"req-{context.aws_request_id[:8]}"

        logger.info(json.dumps({
            "event": "processing_document",
            "request_id": request_id,
            "source": f"s3://{source_bucket}/{source_key}",
        }))

        try:
            document = get_document(source_bucket, source_key)
            llm_result = process_document(document, prompt_template)

            response_data = {
                "request_id": request_id,
                "source_key": source_key,
                "source_bucket": source_bucket,
                "document_title": document.get("title", ""),
                "processed_at": datetime.utcnow().isoformat(),
                **llm_result,
            }

            response_key = save_response(request_id, response_data)

            logger.info(json.dumps({
                "event": "document_processed",
                "request_id": request_id,
                "response_key": response_key,
                "tokens_used": llm_result["tokens_used"],
            }))

            results.append({
                "request_id": request_id,
                "source_key": source_key,
                "response_key": response_key,
                "status": "success",
            })

        except Exception as e:
            logger.error(json.dumps({
                "event": "processing_error",
                "request_id": request_id,
                "source_key": source_key,
                "error": str(e),
            }))
            results.append({
                "request_id": request_id,
                "source_key": source_key,
                "status": "error",
                "error": str(e),
            })

    duration_ms = int((time.time() - start_time) * 1000)

    return {
        "processed": len(results),
        "results": results,
        "duration_ms": duration_ms,
    }

Avoiding Infinite Loops

The problem

If Lambda writes to the same bucket that triggers it, you can create a loop:

S3 upload (documents/inbox/) → Lambda runs
    → Lambda writes to responses/ → does S3 trigger Lambda again?

If the trigger is configured for s3:ObjectCreated:* without a prefix filter, Lambda writes to S3, S3 triggers Lambda, Lambda writes to S3... infinite loop. Each iteration invokes Lambda and costs money.

Solution 1: Different prefixes (recommended)

# Trigger ONLY on documents/inbox/ (input prefix)
# Lambda writes to responses/ (output prefix)
# The trigger does NOT fire on responses/

configure_s3_trigger(
    bucket=BUCKET,
    lambda_arn=LAMBDA_ARN,
    prefix="documents/inbox/",  # Only this prefix triggers Lambda
)

Solution 2: Separate bucket for output

INPUT_BUCKET = "ai-input-dev"
OUTPUT_BUCKET = "ai-output-dev"

# Trigger on INPUT_BUCKET
# Lambda writes to OUTPUT_BUCKET
# No possibility of a loop

Solution 3: Check in the handler

def handler(event, context):
    for record in event.get("Records", []):
        key = record["s3"]["object"]["key"]

        if key.startswith("responses/") or key.startswith("processed/"):
            logger.info(f"Skipping output file: {key}")
            continue

        process_document(record)

Complete Flow: Prompt Template + RAG Document

Pipeline with a dynamic prompt template

# pipeline.py — Complete flow S3 → Lambda → LLM → S3
import boto3
import json
import os
from openai import OpenAI

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", ""))

BUCKET = os.environ.get("AI_BUCKET", "ai-assets-dev")


def run_ai_pipeline(
    prompt_name: str,
    prompt_version: str,
    document_collection: str,
    document_id: str,
    max_tokens: int = 1000,
) -> dict:
    """Runs the complete pipeline: reads prompt + document, invokes LLM, saves response."""

    # 1. Read prompt template from S3
    prompt_key = f"prompts/{prompt_name}/{prompt_version}/system.txt"
    prompt_resp = s3.get_object(Bucket=BUCKET, Key=prompt_key)
    prompt_template = prompt_resp["Body"].read().decode("utf-8")

    # 2. Read document from S3
    doc_key = f"documents/{document_collection}/{document_id}.json"
    doc_resp = s3.get_object(Bucket=BUCKET, Key=doc_key)
    document = json.loads(doc_resp["Body"].read().decode("utf-8"))

    # 3. Build the contextualized prompt
    doc_content = document.get("content", "")
    doc_title = document.get("title", "")
    full_prompt = f"{prompt_template}\n\nTitle: {doc_title}\n\nContent:\n{doc_content}"

    # 4. Invoke the LLM
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": full_prompt}],
        max_tokens=max_tokens,
    )

    result = {
        "answer": response.choices[0].message.content,
        "model": response.model,
        "tokens_used": response.usage.total_tokens,
        "prompt_name": prompt_name,
        "prompt_version": prompt_version,
        "document_id": document_id,
        "collection": document_collection,
    }

    # 5. Save the response to S3
    from datetime import datetime
    now = datetime.utcnow()
    response_key = f"responses/{now.strftime('%Y/%m/%d')}/{document_id}-{prompt_name}.json"

    s3.put_object(
        Bucket=BUCKET,
        Key=response_key,
        Body=json.dumps(result, ensure_ascii=False).encode("utf-8"),
        ContentType="application/json",
    )

    result["response_key"] = response_key
    return result


# Usage
output = run_ai_pipeline(
    prompt_name="summarizer",
    prompt_version="v1",
    document_collection="product-docs",
    document_id="doc-001",
)
print(f"Response saved to: s3://{BUCKET}/{output['response_key']}")

Batch Processing with S3 + Lambda

Process multiple documents

def batch_process_collection(
    prompt_name: str,
    prompt_version: str,
    collection: str,
) -> dict:
    """Processes all documents in a collection with the given prompt."""
    paginator = s3.get_paginator("list_objects_v2")
    prefix = f"documents/{collection}/"

    doc_keys = []
    for page in paginator.paginate(Bucket=BUCKET, Prefix=prefix):
        for obj in page.get("Contents", []):
            if obj["Key"].endswith(".json"):
                doc_keys.append(obj["Key"])

    print(f"Found {len(doc_keys)} documents in '{collection}'")

    results = {"success": 0, "failed": 0, "errors": []}

    for key in doc_keys:
        doc_id = key.split("/")[-1].replace(".json", "")
        try:
            output = run_ai_pipeline(
                prompt_name=prompt_name,
                prompt_version=prompt_version,
                document_collection=collection,
                document_id=doc_id,
            )
            results["success"] += 1
            print(f"  ✓ {doc_id}: {output['tokens_used']} tokens")
        except Exception as e:
            results["failed"] += 1
            results["errors"].append({"doc_id": doc_id, "error": str(e)})
            print(f"  ✗ {doc_id}: {str(e)}")

    return results

SAM Template with an S3 Trigger

template.yaml for S3 → Lambda

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: S3 + Lambda AI Pipeline — Module 5

Parameters:
  OpenAiApiKey:
    Type: String
    NoEcho: true
  BucketName:
    Type: String
    Default: ai-assets-dev

Resources:
  AIBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Ref BucketName

  DocProcessorFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: ai-doc-processor
      Handler: s3_trigger_handler.handler
      Runtime: python3.11
      Architectures: [arm64]
      MemorySize: 512
      Timeout: 120
      Environment:
        Variables:
          OPENAI_API_KEY: !Ref OpenAiApiKey
          AI_BUCKET: !Ref BucketName
          PROMPT_KEY: prompts/summarizer/v1/system.txt
          RESPONSE_PREFIX: responses/
          MODEL_NAME: gpt-4o-mini
      Policies:
        - S3ReadPolicy:
            BucketName: !Ref BucketName
        - S3CrudPolicy:
            BucketName: !Ref BucketName
      Events:
        NewDocument:
          Type: S3
          Properties:
            Bucket: !Ref AIBucket
            Events: s3:ObjectCreated:*
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: documents/inbox/
                  - Name: suffix
                    Value: .json

Outputs:
  BucketName:
    Value: !Ref AIBucket
  ProcessorArn:
    Value: !GetAtt DocProcessorFunction.Arn

Testing the Complete Flow

Manual test with boto3

import boto3
import json
import os
import time

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = "ai-assets-dev"


def setup_test_data():
    """Prepares test data in S3."""
    s3.put_object(
        Bucket=BUCKET,
        Key="prompts/summarizer/v1/system.txt",
        Body="Summarize the following document in 3 key points.".encode("utf-8"),
    )

    test_doc = {
        "id": "test-doc-001",
        "title": "Introduction to Serverless",
        "content": (
            "Serverless computing lets you run code without managing servers. "
            "AWS Lambda is the most popular service for serverless. "
            "Functions run in response to events and scale automatically. "
            "You pay only for execution time, not for idle servers."
        ),
    }

    s3.put_object(
        Bucket=BUCKET,
        Key="documents/inbox/test-doc-001.json",
        Body=json.dumps(test_doc, ensure_ascii=False).encode("utf-8"),
    )
    print("Test data uploaded")


def verify_response():
    """Verifies that Lambda processed the document and saved the response."""
    time.sleep(5)

    response = s3.list_objects_v2(Bucket=BUCKET, Prefix="responses/")
    contents = response.get("Contents", [])

    if not contents:
        print("No responses found yet. Lambda may still be processing.")
        return

    for obj in contents:
        print(f"Response found: {obj['Key']}")
        resp = s3.get_object(Bucket=BUCKET, Key=obj["Key"])
        data = json.loads(resp["Body"].read().decode("utf-8"))
        print(f"  Document: {data.get('document_title')}")
        print(f"  Answer: {data.get('answer', '')[:200]}...")
        print(f"  Tokens: {data.get('tokens_used')}")


setup_test_data()
verify_response()

Troubleshooting

Problem 1: Lambda doesn't trigger when I upload a file to S3

Check the notification configuration and the permissions.

# Check the notification configuration
aws s3api get-bucket-notification-configuration --bucket ai-assets-dev

# Check that Lambda has a permission policy for S3
aws lambda get-policy --function-name ai-doc-processor

# If the permission is missing:
aws lambda add-permission \
    --function-name ai-doc-processor \
    --statement-id s3-trigger \
    --action lambda:InvokeFunction \
    --principal s3.amazonaws.com \
    --source-arn arn:aws:s3:::ai-assets-dev

Problem 2: Infinite loop — Lambda triggers itself

Lambda writes to a prefix that fires another trigger.

# Make sure your trigger filters by the INPUT prefix
# and that Lambda writes to a different OUTPUT prefix

# ✅ Trigger: documents/inbox/*.json
# ✅ Output:  responses/2026/03/08/*.json

# ❌ Trigger: documents/*.json (too broad)
# ❌ Output:  documents/processed/*.json (same base prefix)

Problem 3: "NoSuchKey" when reading the prompt template

The prompt template doesn't exist in S3 when Lambda runs.

try:
    prompt = get_prompt_template()
except s3.exceptions.NoSuchKey:
    logger.error(f"Prompt template not found: {PROMPT_KEY}")
    prompt = "Summarize the following document concisely."

Problem 4: S3 event with a URL-encoded key

S3 sends URL-encoded keys. Spaces become +.

from urllib.parse import unquote_plus

raw_key = record["s3"]["object"]["key"]
key = unquote_plus(raw_key)

Practical Exercises

Exercise 1: Multi-prompt pipeline

Modify the pipeline so it processes each document with multiple prompts (for example, "summarizer" and "classifier") and saves each result in a separate prefix.

See solution 
import boto3
import json
import os
from datetime import datetime
from openai import OpenAI

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", ""))
BUCKET = os.environ.get("AI_BUCKET", "ai-assets-dev")

PIPELINES = [
    {"prompt_name": "summarizer", "prompt_version": "v1", "output_prefix": "summaries/"},
    {"prompt_name": "classifier", "prompt_version": "v1", "output_prefix": "classifications/"},
]


def multi_prompt_handler(event, context):
    results = []

    for record in event.get("Records", []):
        source_key = record["s3"]["object"]["key"]
        source_bucket = record["s3"]["bucket"]["name"]

        doc_resp = s3.get_object(Bucket=source_bucket, Key=source_key)
        document = json.loads(doc_resp["Body"].read().decode("utf-8"))

        for pipeline in PIPELINES:
            prompt_key = f"prompts/{pipeline['prompt_name']}/{pipeline['prompt_version']}/system.txt"
            prompt_resp = s3.get_object(Bucket=BUCKET, Key=prompt_key)
            prompt_template = prompt_resp["Body"].read().decode("utf-8")

            full_prompt = f"{prompt_template}\n\n{document.get('content', '')}"

            response = openai_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": full_prompt}],
                max_tokens=500,
            )

            doc_id = document.get("id", source_key.split("/")[-1].replace(".json", ""))
            now = datetime.utcnow()
            output_key = (
                f"{pipeline['output_prefix']}{now.strftime('%Y/%m/%d')}/"
                f"{doc_id}-{pipeline['prompt_name']}.json"
            )

            result = {
                "document_id": doc_id,
                "pipeline": pipeline["prompt_name"],
                "answer": response.choices[0].message.content,
                "tokens_used": response.usage.total_tokens,
                "processed_at": now.isoformat(),
            }

            s3.put_object(
                Bucket=BUCKET,
                Key=output_key,
                Body=json.dumps(result, ensure_ascii=False).encode("utf-8"),
                ContentType="application/json",
            )

            results.append({"doc_id": doc_id, "pipeline": pipeline["prompt_name"], "output": output_key})

    return {"processed": len(results), "results": results}

Exercise 2: S3 trigger with deduplication

Implement a mechanism that avoids processing the same document twice. Use a "processed/" prefix in S3 as a record of already-processed documents.

See solution 
import boto3
import json
import os
import hashlib
from datetime import datetime
from urllib.parse import unquote_plus

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = os.environ.get("AI_BUCKET", "ai-assets-dev")
PROCESSED_PREFIX = "processed/"


def is_already_processed(source_key: str) -> bool:
    """Checks whether a document has already been processed."""
    marker_key = f"{PROCESSED_PREFIX}{hashlib.md5(source_key.encode()).hexdigest()}"
    try:
        s3.head_object(Bucket=BUCKET, Key=marker_key)
        return True
    except s3.exceptions.ClientError as e:
        if e.response["Error"]["Code"] == "404":
            return False
        raise


def mark_as_processed(source_key: str, result_key: str):
    """Marks a document as processed."""
    marker_key = f"{PROCESSED_PREFIX}{hashlib.md5(source_key.encode()).hexdigest()}"
    s3.put_object(
        Bucket=BUCKET,
        Key=marker_key,
        Body=json.dumps({
            "source_key": source_key,
            "result_key": result_key,
            "processed_at": datetime.utcnow().isoformat(),
        }).encode("utf-8"),
    )


def handler(event, context):
    results = []

    for record in event.get("Records", []):
        source_key = unquote_plus(record["s3"]["object"]["key"])

        if is_already_processed(source_key):
            results.append({"key": source_key, "status": "skipped", "reason": "already processed"})
            continue

        try:
            result_key = process_and_save(source_key)
            mark_as_processed(source_key, result_key)
            results.append({"key": source_key, "status": "processed", "result": result_key})
        except Exception as e:
            results.append({"key": source_key, "status": "error", "error": str(e)})

    return {"results": results}


def process_and_save(source_key: str) -> str:
    """Processes the document and returns the result key."""
    doc_resp = s3.get_object(Bucket=BUCKET, Key=source_key)
    document = json.loads(doc_resp["Body"].read().decode("utf-8"))
    now = datetime.utcnow()
    doc_id = source_key.split("/")[-1].replace(".json", "")
    result_key = f"responses/{now.strftime('%Y/%m/%d')}/{doc_id}.json"

    s3.put_object(
        Bucket=BUCKET,
        Key=result_key,
        Body=json.dumps({"processed": True, "source": source_key}).encode("utf-8"),
    )
    return result_key

Exercise 3: Pipeline with A/B prompt versioning

Implement a flow that processes each document with two prompt versions (A/B test) and saves both results to compare.

See solution 
import boto3
import json
import os
import time
from datetime import datetime
from openai import OpenAI

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", ""))
BUCKET = os.environ.get("AI_BUCKET", "ai-assets-dev")


def ab_test_handler(event, context):
    """Processes each document with two prompt versions for A/B testing."""
    versions = ["v1", "v2"]
    prompt_name = os.environ.get("PROMPT_NAME", "summarizer")
    results = []

    for record in event.get("Records", []):
        source_key = record["s3"]["object"]["key"]
        doc_resp = s3.get_object(Bucket=BUCKET, Key=source_key)
        document = json.loads(doc_resp["Body"].read().decode("utf-8"))
        doc_id = document.get("id", "unknown")

        ab_results = {}

        for version in versions:
            prompt_key = f"prompts/{prompt_name}/{version}/system.txt"
            try:
                prompt_resp = s3.get_object(Bucket=BUCKET, Key=prompt_key)
                prompt_template = prompt_resp["Body"].read().decode("utf-8")
            except Exception:
                continue

            full_prompt = f"{prompt_template}\n\n{document.get('content', '')}"

            start = time.time()
            response = openai_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": full_prompt}],
                max_tokens=500,
            )
            duration = int((time.time() - start) * 1000)

            ab_results[version] = {
                "answer": response.choices[0].message.content,
                "tokens_used": response.usage.total_tokens,
                "duration_ms": duration,
            }

        now = datetime.utcnow()
        comparison_key = f"ab-tests/{prompt_name}/{now.strftime('%Y/%m/%d')}/{doc_id}.json"

        s3.put_object(
            Bucket=BUCKET,
            Key=comparison_key,
            Body=json.dumps({
                "document_id": doc_id,
                "prompt_name": prompt_name,
                "versions_tested": versions,
                "results": ab_results,
                "tested_at": now.isoformat(),
            }, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )

        results.append({"doc_id": doc_id, "comparison_key": comparison_key})

    return {"ab_tests": len(results), "results": results}

Exercise 4: Pipeline monitor with alerts

Create a function that reviews the responses bucket and generates a pipeline health report: documents processed in the last 24h, errors, average tokens, etc.

See solution 
import boto3
import json
import os
from datetime import datetime, timedelta

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = os.environ.get("AI_BUCKET", "ai-assets-dev")


def pipeline_health_check() -> dict:
    """Generates a pipeline health report for the last 24h."""
    now = datetime.utcnow()
    yesterday = now - timedelta(days=1)

    prefixes = [
        f"responses/{now.strftime('%Y/%m/%d')}/",
        f"responses/{yesterday.strftime('%Y/%m/%d')}/",
    ]

    all_responses = []
    for prefix in prefixes:
        paginator = s3.get_paginator("list_objects_v2")
        for page in paginator.paginate(Bucket=BUCKET, Prefix=prefix):
            for obj in page.get("Contents", []):
                if obj["LastModified"].replace(tzinfo=None) >= yesterday:
                    try:
                        resp = s3.get_object(Bucket=BUCKET, Key=obj["Key"])
                        data = json.loads(resp["Body"].read().decode("utf-8"))
                        all_responses.append(data)
                    except Exception:
                        pass

    if not all_responses:
        return {"status": "warning", "message": "No responses in last 24h", "count": 0}

    tokens_list = [r.get("tokens_used", 0) for r in all_responses]
    errors = [r for r in all_responses if r.get("status") == "error"]

    report = {
        "status": "healthy" if len(errors) == 0 else "degraded",
        "period": f"{yesterday.isoformat()} — {now.isoformat()}",
        "total_processed": len(all_responses),
        "errors": len(errors),
        "error_rate": f"{(len(errors) / len(all_responses)) * 100:.1f}%",
        "tokens": {
            "total": sum(tokens_list),
            "average": round(sum(tokens_list) / len(tokens_list)),
            "max": max(tokens_list),
        },
        "checked_at": now.isoformat(),
    }

    if len(errors) / max(len(all_responses), 1) > 0.1:
        report["alert"] = "Error rate > 10%. Check pipeline configuration."

    return report


health = pipeline_health_check()
print(json.dumps(health, indent=2))

Summary

  • S3 Event Notifications trigger Lambda automatically when a file is uploaded. Configure prefix and suffix filters for precision.
  • The S3 → Lambda → S3 flow is the core pattern of an AI service on AWS: Lambda reads assets (prompts, documents), invokes the LLM, and persists results.
  • Avoid infinite loops by using separate input and output prefixes, or different buckets for input and output.
  • Parse the S3 event correctly: decode URL-encoded keys with unquote_plus, handle multiple Records.
  • The SAM Template defines the S3 → Lambda trigger as code — reproducible and versionable.
  • Batch processing lets you process entire document collections with a single prompt or multiple prompts (A/B testing).

Additional Resources

  1. S3 Event Notifications — Configuring triggers
  2. Lambda S3 Event Tutorial — Official tutorial
  3. S3 Event Message Structure — Event format
  4. SAM S3 Event — SAM template for S3 triggers
  5. Lambda Permissions for S3 — Required permissions
  6. Avoiding Recursive Invocations — Preventing loops
  7. S3 Batch Operations — Native mass processing
  8. Lambda Dead Letter Queues — Handling async failures
← Previous3. Lambda for AI InferenceNext →5. SageMaker Basics