Ingest Pipeline

The ingest pipeline fetches AWS announcements from public RSS feeds, generates

embeddings embeddings
A vector representation of text — a list of floating-point numbers that captures semantic meaning. Generated by an embedding model. using Titan Embeddings V2 Titan Embeddings V2
Amazon Titan Embeddings V2 — the Bedrock foundation model used to generate 1024-dimensional text embeddings. , and stores them in S3 Vectors S3 Vectors
Amazon S3 Vectors — a purpose-built vector storage capability within S3 that enables similarity search over embeddings without a separate vector database. for similarity search. Each run processes the full corpus and upserts vectors — making it safe to re-run at any time.

Trigger sources

The ingest Lambda Lambda
AWS Lambda — serverless compute. This project uses two functions: ingest (RSS → embed → store) and query (search → answer). can be invoked three ways:

Trigger	Source value	When it fires
EventBridge EventBridge Amazon EventBridge — a serverless event bus. Used here for the daily scheduled corpus ingest cron trigger. schedule	eventbridge	Daily at 06:00 UTC (configurable via ingest_schedule_expression)
API Gateway endpoint	api	Admin sends POST /ingest (requires Cognito JWT in admin group)
Bootstrap (terraform apply)	terraform	Runs once during initial deployment via aws_lambda_invocation

Note

The bootstrap invocation uses a lifecycle.replace_triggered_by block tied to the Lambda source hash — it re-runs automatically whenever the function code changes.

Step-by-step flow

The ingest pipeline runs through these stages:

1. Fetch RSS feeds — download articles from AWS What’s New and AWS News Blog
2. Strip HTML — extract plain text, normalize unicode, deduplicate by URL
3. Truncate — limit each article body to 8,000 characters
4. Upload to S3 — store each article as a .txt file in the source bucket
5. Embed — generate a 1024-dimensional vector for each article via Bedrock
6. PutVectors — store vectors with metadata in S3 Vectors
7. Write marker — record .last-ingest with timestamp and count

RSS fetching

The fetch_articles() function downloads from two feeds:

rag/fetch.py

FEEDS = (
    ("aws-whats-new", "https://aws.amazon.com/about-aws/whats-new/recent/feed/", 100),
    ("aws-news-blog", "https://aws.amazon.com/blogs/aws/feed/", 20),
)

Each feed has a per-source limit matching what the public endpoints expose. Articles are deduplicated by URL across feeds and a minimum of 5 articles is required for a successful run.

The HTTP fetch includes retry logic:

rag/fetch.py

def _fetch_bytes(url: str, retries: int = 2) -> bytes:
    request = Request(url, headers={"User-Agent": USER_AGENT})
    last_error: Exception | None = None
    for attempt in range(retries):
        try:
            with urlopen(request, timeout=60) as response:
                return response.read()
        except (HTTPError, URLError, TimeoutError) as exc:
            last_error = exc
            if attempt + 1 < retries:
                time.sleep(2)
    raise RuntimeError(f"Failed to fetch {url}: {last_error}") from last_error

HTML stripping and text preparation

Raw RSS content is HTML. The pipeline strips tags, unescapes entities, normalizes unicode to ASCII, and truncates to MAX_BODY_CHARS (8,000):

rag/fetch.py

def _strip_html(value: str) -> str:
    if not value:
        return ""
    unescaped = html.unescape(value)
    parser = _HTMLTextExtractor()
    parser.feed(unescaped)
    parser.close()
    text = parser.get_text() or re.sub(r"<[^>]+>", " ", unescaped)
    return re.sub(r"\s+", " ", text).strip()

def _truncate_body(text: str) -> str:
    if len(text) <= MAX_BODY_CHARS:
        return text
    trimmed = text[:MAX_BODY_CHARS].rsplit(" ", 1)[0]
    return trimmed + "\n\n[Truncated for demo embedding size.]\n"

Tip

Truncation splits on the last space before 8,000 characters to avoid cutting words mid-token. This keeps the embedding input clean.

Embedding with Bedrock

Each article body is embedded using Titan Embeddings V2 Titan Embeddings V2
Amazon Titan Embeddings V2 (amazon.titan-embed-text-v2:0) — converts text into 1024-dimensional vectors for similarity search. via the Bedrock runtime API. The function includes exponential backoff for throttling:

rag/ingest.py

def _bedrock_embed(client, model_id: str, text: str) -> list[float]:
    last_error = None
    for attempt in range(6):
        try:
            response = client.invoke_model(
                modelId=model_id,
                contentType="application/json",
                accept="application/json",
                body=json.dumps({"inputText": text}),
            )
            payload = json.loads(response["body"].read())
            return payload["embedding"]
        except ClientError as exc:
            code = exc.response.get("Error", {}).get("Code", "")
            if code not in {"ThrottlingException", "ServiceUnavailableException"}:
                raise
            last_error = exc
            time.sleep(min(2**attempt, 30))
    raise RuntimeError(f"Bedrock embed throttled after retries: {last_error}")

Key details:

• Up to 6 retry attempts with exponential backoff (1s, 2s, 4s, 8s, 16s, 30s max)
• Only retries ThrottlingException and ServiceUnavailableException
• All other errors raise immediately
• A 150ms sleep between articles avoids hitting Bedrock rate limits

Storing vectors with PutVectors

After embedding, all vectors are written to S3 Vectors in a single PutVectors PutVectors
S3 Vectors PutVectors API — stores one or more vectors with associated metadata in a vector index. call:

rag/ingest.py

vectors.append(
    {
        "key": f"article-{article.url_hash}-chunk-000",
        "data": {"float32": embedding},
        "metadata": {
            "source": article.filename,
            "title": article.title,
            "url": article.url,
            "published": article.published,
            "feed": article.feed,
            "chunk": "0",
        },
    }
)

_put_vectors(s3vectors, config, vectors)

The vector key uses the URL hash, so re-ingesting the same article upserts rather than duplicates.

Vector metadata

Each stored vector carries metadata used during query and scoring:

Field	Example	Used for
source	aws-whats-new-a1b2c3d4.txt	Loading full document from S3 at query time
title	Amazon S3 Vectors is now GA	Display in search results and LLM context
url	https://aws.amazon.com/…	Citation links in generated answers
published	2025-06-01T12:00:00Z	Recency scoring (newer articles rank higher)
feed	aws-whats-new	Source attribution
chunk	0	Always “0” — single-chunk strategy (one vector per article)

Last-ingest marker

After all vectors are stored, the pipeline writes a JSON marker to S3:

rag/ingest.py

marker = {
    "ingested_at": fetched_at.strftime("%Y-%m-%dT%H:%M:%SZ"),
    "article_count": len(items),
}
s3.put_object(
    Bucket=config.source_bucket,
    Key=config.last_ingest_key,
    Body=json.dumps(marker).encode("utf-8"),
    ContentType="application/json",
)

This marker is read by the status endpoint to report corpus freshness.

Error handling and retries

Failure point	Behavior	Max retries
RSS feed HTTP errors	Retry with 2s backoff, then raise	2
Too few articles fetched	Raise RuntimeError (need ≥ 5)	0
Bedrock throttling	Exponential backoff (1s–30s cap)	6
Bedrock non-throttle errors	Raise immediately (no retry)	0
S3/S3 Vectors write failures	Raise (Lambda timeout is 480s)	0

Caution

The Lambda timeout is set to 480 seconds (8 minutes). If Bedrock throttling is severe, the function may time out before processing all articles. Check CloudWatch logs at /aws/lambda/<project>-<env>-ingest for timeout indicators.

Related pages

• Embeddings strategy — why single-chunk and Titan V2
• Query pipeline — how stored vectors are searched
• Scoring — how recency metadata affects ranking