Integrating Bedrock With Amazon S3 Part 2

# read_generate_write_s3.py
import boto3
import json
import logging

# -------- Configuration --------
REGION = "us-east-1"
BUCKET_NAME = "kodekloud-intro-bedrock"
INPUT_KEY = "input/customer_reviews.txt"
OUTPUT_KEY = "output/customer_reviews_summary.txt"

# Example model identifier (replace with the model available in your account)
MODEL_ID = "us.anthropic.claude-haiku-4-5-20251001-v1:0"

# -------- Logging --------
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# -------- AWS Clients --------
s3 = boto3.client("s3", region_name=REGION)
bedrock = boto3.client("bedrock-runtime", region_name=REGION)

def read_s3_text(bucket: str, key: str) -> str:
    resp = s3.get_object(Bucket=bucket, Key=key)
    return resp["Body"].read().decode("utf-8")

def write_s3_text(bucket: str, key: str, text: str) -> None:
    s3.put_object(Bucket=bucket, Key=key, Body=text.encode("utf-8"), ContentType="text/plain")

def build_prompt(input_text: str) -> str:
    return f"""
Instruction:
You are summarizing customer feedback for a product team.

Requirements:
- Be concise
- Highlight common themes
- Use bullet points
- Maximum of 5 bullets

Input:
{input_text}
"""

def invoke_bedrock(model_id: str, prompt: str) -> str:
    payload = {"input": prompt}
    response = bedrock.invoke_model(
        modelId=model_id,
        contentType="application/json",
        accept="application/json",
        body=json.dumps(payload).encode("utf-8"),
        inferenceConfig={
            "maxTokensToSample": 300,
            "temperature": 0.2
        }
    )
    return response["body"].read().decode("utf-8")

def parse_bedrock_response(resp_body: str) -> str:
    """
    Attempt to handle several common Bedrock JSON shapes:
    - `{"outputs": [{"content":[{"text":"..."}]}]}`
    - `{"output": {"message": {"content": [{"text":"..."}]}}}`
    - `{"generated_text": "..."}` or `{"text": "..."}`

    Falls back to returning the raw response body if parsing fails.
    """
    try:
        obj = json.loads(resp_body)
        # outputs -> content -> text
        if isinstance(obj, dict):
            if "outputs" in obj and isinstance(obj["outputs"], list):
                parts = []
                for out in obj["outputs"]:
                    if isinstance(out, dict) and "content" in out and isinstance(out["content"], list):
                        for c in out["content"]:
                            if isinstance(c, dict) and "text" in c:
                                parts.append(c["text"])
                if parts:
                    return "".join(parts)
            # output -> message -> content -> text
            if "output" in obj and isinstance(obj["output"], dict):
                output = obj["output"]
                if "message" in output and isinstance(output["message"], dict) and "content" in output["message"]:
                    parts = []
                    for c in output["message"]["content"]:
                        if isinstance(c, dict) and "text" in c:
                            parts.append(c["text"])
                    if parts:
                        return "".join(parts)
            # simpler fields
            if "generated_text" in obj:
                return obj["generated_text"]
            if "text" in obj:
                return obj["text"]
        # If no known pattern matched, return the raw response
        return resp_body
    except Exception:
        # On parse error, return raw response body
        return resp_body

def main():
    logger.info("Reading input file from S3...")
    input_text = read_s3_text(BUCKET_NAME, INPUT_KEY)
    logger.info("Input read (truncated): %s", input_text[:200])

    prompt = build_prompt(input_text)

    logger.info("Invoking Bedrock model...")
    raw_resp = invoke_bedrock(MODEL_ID, prompt)

    logger.info("Parsing model response...")
    summary = parse_bedrock_response(raw_resp)

    logger.info("Writing summary back to S3...")
    write_s3_text(BUCKET_NAME, OUTPUT_KEY, summary)

    logger.info("Summary written to s3://%s/%s", BUCKET_NAME, OUTPUT_KEY)

if __name__ == "__main__":
    main()

$ python3 read_generate_write_s3.py
INFO:__main__:Reading input file from S3...
INFO:__main__:Input read (truncated): 1. Absolutely love this product. Setup took less than five minutes and it works flawlessly.
...
INFO:__main__:Invoking Bedrock model...
INFO:__main__:Parsing model response...
INFO:__main__:Writing summary back to S3...
INFO:__main__:Summary written to s3://kodekloud-intro-bedrock/output/customer_reviews_summary.txt

- **Setup & Performance**: Quick setup and reliable functionality when working properly.
- **Quality Control & Shipping**: Multiple reports of defective units and damaged packaging upon delivery.
- **Customer Support**: Inconsistent experience — some customers received prompt replacements while others received canned responses.
- **Documentation & Battery Life**: Instructions unclear and occasional complaints about battery life.
- **Value**: Generally perceived as good value for the price.