Building an AI Document Processor with RAG, pgvector, and AWS

What I Built

A full-stack AI platform where you upload any document — PDF, invoice, contract, image — and the system automatically extracts text, classifies the document type, generates a summary, and lets you chat with it in natural language.

User flow:

1. Upload a PDF or image — it goes directly to S3 via presigned URL
2. Watch real-time processing status as Textract, Claude, and the embedding pipeline run — via WebSocket
3. Read the AI-generated classification and summary
4. Ask questions in natural language — get answers with cited page numbers
5. Search across your entire document library in one query

The key difference from a simple chatbot: the answers are grounded in your documents. Claude can only use the retrieved chunks, and every answer includes source citations with page numbers.

Architecture

Browser
  │
  ▼
CloudFront (HTTPS)
  ├── /* ──────────────► S3 (React static files)
  └── /api/v1/* ───────► ALB
                           │
                           ▼
                      ECS Fargate (FastAPI)
                       Cognito JWT auth
                           │
              ┌────────────┼────────────┐
              ▼            ▼            ▼
         S3 Upload      RAG Chat    WebSocket
        (presigned)     pgvector      status
              │            │            │
              ▼            ▼            ▼
          S3 trigger   Anthropic   API Gateway
              │          Claude     WebSocket
              ▼
       Lambda Orchestrator
              │
       Amazon Textract
         (OCR + tables)
              │
       Lambda Processor
              │
      ┌───────┴────────┐
      ▼                ▼
   Claude           pgvector
(classify +        (chunks +
 summarize)       embeddings)

This project is the most complex of my three — eight CDK stacks covering networking, storage, auth, database, async processing, WebSocket, compute, and monitoring. The async processing pipeline (S3 → Lambda → Textract → Lambda → Claude → pgvector) is the architectural core, separate from the FastAPI backend that handles the user-facing API.

Tech Stack

AWS Services Used

The Processing Pipeline

When a user uploads a document, it triggers an async pipeline that runs entirely outside the FastAPI request cycle.

Step 1 — Direct S3 Upload via Presigned URL

The frontend requests a presigned POST URL from the backend, then uploads the file directly to S3 — the binary never touches the FastAPI server.

# api/v1/upload.py
async def get_presigned_url(request: PresignedUrlRequest, user=Depends(get_current_user)):
    s3_key = f"{user.id}/{doc_id}/{request.file_name}"
    response = s3_client.generate_presigned_post(
        Bucket=settings.s3_raw_bucket,
        Key=s3_key,
        ExpiresIn=300,
    )
    return {"doc_id": doc_id, "upload_url": response["url"], "upload_fields": response["fields"]}

After the upload completes, the frontend calls /api/v1/upload/complete/{doc_id} — this triggers the async pipeline.

Step 2 — Lambda Orchestrator: Textract

The S3 ObjectCreated event triggers the Orchestrator Lambda, which starts an async Textract job and updates the document status to processing.

Textract runs asynchronously — for a 20-page PDF it can take 30–60 seconds. When Textract finishes, it publishes a completion event to an SNS topic, which triggers the Processor Lambda.

Step 3 — Lambda Processor: Claude + Voyage AI + pgvector

This is the most complex Lambda. It runs 6 steps in sequence:

# lambdas/processor/handler.py
def handler(event, context):
    for sns_record in event["Records"]:
        message = json.loads(sns_record["Sns"]["Message"])
        doc_id = message["JobTag"]

        # Step 1: fetch Textract result (paginated API)
        result = _fetch_textract_result(message["JobId"])

        # Step 2: save raw Textract output to S3 processed bucket
        _save_textract_result_to_s3(doc_id, result)

        # Step 3: classify document type + generate summary with Claude
        claude_result = _classify_with_claude(result["raw_text"], doc_id)

        # Step 4: update DB → status=COMPLETED, doc_type, summary, page_count
        _update_document(doc_id, {
            "status": "completed",
            "page_count": result["page_count"],
            "doc_type": claude_result["doc_type"],
            "summary": claude_result["summary"],
        })

        # Step 5: chunk text → embed with Voyage AI → store in pgvector
        _store_chunks(doc_id, result["pages"])

        # Step 6: push status update to browser via WebSocket
        _notify_websocket(doc_id, "completed", ...)

Claude classification prompt returns structured JSON — document type (invoice, contract, report, etc.) and a 2–3 sentence summary. Pydantic validates the JSON before saving.

Chunking strategy: 800 characters per chunk, 100-character overlap between chunks. Each chunk stores its page number — this is what enables page-level citations in the final answer.

Voyage AI embeddings: Each chunk is embedded using voyage-3 in a single batched API call — one call for all chunks in the document, not one call per chunk.

Step 4 — RAG Chat: Embed → Retrieve → Stream

When the user asks a question, the RAG pipeline runs inside FastAPI:

# services/rag_service.py
async def embed_question(question: str) -> list[float]:
    client = voyageai.Client(api_key=settings.voyage_api_key)
    result = client.embed([question], model="voyage-3", input_type="query")
    return result.embeddings[0]

async def find_relevant_chunks(db, document_id, query_embedding) -> list[dict]:
    await db.execute(text("SET LOCAL ivfflat.probes = 10"))
    result = await db.execute(text("""
        SELECT chunk_index, page_number, content,
               1 - (embedding_vec <=> CAST(:embedding AS vector)) AS similarity
        FROM document_chunks
        WHERE document_id = :document_id
          AND embedding_vec IS NOT NULL
        ORDER BY embedding_vec <=> CAST(:embedding AS vector)
        LIMIT 5
    """), {"document_id": document_id, "embedding": str(query_embedding)})
    return [dict(row) for row in result]

The top-5 most similar chunks are assembled into a context block and sent to Claude. Claude is constrained to only use the provided excerpts — if the answer isn't there, it says so. The response streams back to the browser via SSE, word by word.

WebSocket — Real-time Processing Status

Processing a document takes 30–90 seconds. Instead of polling, the browser opens a WebSocket connection via API Gateway on page load. When the Processor Lambda finishes, it invokes the WebSocket Lambda which pushes a status update directly to the browser.

Browser ──WebSocket──► API Gateway ──► Lambda WS handler
                                           │
                          DynamoDB ◄───────┘
                       (connection store)
                                           ▲
                          Lambda Processor ─┘
                        (invokes WS handler
                         on job completion)

The WebSocket connection store (DynamoDB) maps connection_id → user_id. When the Processor Lambda calls _notify_websocket(doc_id, status), the WS Lambda looks up the user's active connections and pushes the update to all of them. If the user has multiple browser tabs open, all get updated simultaneously.

Cognito Auth — Per-User Document Isolation

Every API request requires a valid Cognito JWT. The FastAPI dependency get_current_user verifies the token against Cognito's JWKS endpoint and extracts the sub claim as the user ID.

# core/security.py
async def verify_cognito_token(token: str) -> dict:
    jwks_url = f"https://cognito-idp.{settings.cognito_region}.amazonaws.com/{settings.cognito_user_pool_id}/.well-known/jwks.json"
    jwks_client = PyJWKClient(jwks_url, cache_keys=True)
    signing_key = jwks_client.get_signing_key_from_jwt(token)
    return jwt.decode(token, signing_key.key, algorithms=["RS256"],
                      audience=settings.cognito_client_id)

All database queries filter by owner_id = current_user.id — users can only see their own documents. This is enforced at the repository layer, not the route layer.

API Endpoints

Infrastructure with AWS CDK — 8 Stacks

Each stack has a single concern and is deployed independently.

# infrastructure/stacks/processing_stack.py (simplified)
orchestrator = lambda_.Function(
    self, "Orchestrator",
    runtime=lambda_.Runtime.PYTHON_3_12,
    handler="handler.handler",
    environment={
        "SQS_QUEUE_URL": props.sqs_queue.queue_url,
        "USE_TEXTRACT_STUB": "false",
    },
)
# S3 trigger → Orchestrator Lambda
props.raw_bucket.add_event_notification(
    s3.EventType.OBJECT_CREATED,
    s3n.LambdaDestination(orchestrator),
)

processor = lambda_.Function(
    self, "Processor",
    runtime=lambda_.Runtime.PYTHON_3_12,
    handler="handler.handler",
    timeout=Duration.minutes(5),
    environment={
        "DATABASE_URL": props.db_url_ssm.string_value,
        "WEBSOCKET_LAMBDA_NAME": props.ws_lambda.function_name,
    },
)
# Textract SNS completion → Processor Lambda
processor.add_event_source(
    lambda_event_sources.SnsEventSource(props.textract_sns_topic)
)

The Database stack uses removal_policy=RETAIN — Aurora is never deleted on cdk destroy. This prevents accidental data loss during infrastructure teardown.

Local Development with Docker Compose

All AWS services are emulated locally using LocalStack, so you can develop without AWS credentials or costs:

# Start Postgres + Redis + LocalStack
docker-compose up -d

# Run migrations
alembic upgrade head

# Start API
uvicorn app.main:app --reload --port 8000

# Start local Lambda worker (simulates S3 trigger + Textract + Processor)
make worker

The USE_TEXTRACT_STUB=true environment variable skips real Textract and uses a stub document — so the full pipeline runs locally without any AWS calls.

Testing Strategy

cd backend
pytest tests/ -v --cov=app          # all tests with coverage
pytest tests/api/test_upload.py     # upload endpoints
pytest tests/api/test_chat.py       # RAG chat + SSE streaming
pytest tests/services/              # service layer (S3, Textract, Claude, RAG)

All AWS services are mocked with moto. Claude and Voyage AI are mocked at the service level — no real API calls in tests.

Lessons Learned

1. Async processing is the right pattern for heavy AI workloads Running Textract + Claude + embedding in a Lambda decoupled from the API request means the HTTP call returns immediately. The user sees progress via WebSocket instead of waiting on a 60-second HTTP request that might time out.

2. WebSocket via API Gateway + Lambda is simpler than it looks The connection store (DynamoDB mapping connection IDs to user IDs) is the key insight. Once that's in place, pushing updates from any Lambda becomes a one-line API call to the WebSocket management endpoint.

3. Batch embeddings — one API call per document, not per chunk Generating embeddings in a single batched call (all chunks at once) vs. one call per chunk is 10x faster and significantly cheaper. Voyage AI's client.embed([text1, text2, ...]) handles batches natively.

4. pgvector on Aurora Serverless v2 is a production-ready RAG store Aurora Serverless v2 scales to zero when idle and handles the pgvector cosine similarity queries with ivfflat indexing. For a document Q&A workload with moderate traffic, it's cheaper and simpler than a dedicated vector database.

5. Eight CDK stacks kept the infrastructure manageable One concern per stack means cdk diff DocuAI-Processing-dev shows only Lambda changes, not the entire 200-resource deployment. Stacks also have clear dependency chains — Network → Storage/Auth → Database → Processing/WebSocket → Compute → Monitoring.

6. Cognito JWKS caching is essential Without caching, every API request would fetch the JWKS endpoint — adding 200ms+ latency and hitting Cognito rate limits under load. PyJWKClient(jwks_url, cache_keys=True) fixes both.

7. USE_TEXTRACT_STUB made local development possible Real Textract requires S3 objects in a real AWS bucket and takes 10–30 seconds per document. A stub that returns a fixed text string lets the full pipeline run in under a second locally — including Claude classification, chunking, embedding, and pgvector storage.

GitHub

Key code samples (RAG service, Lambda processor, CDK stacks) are available on GitHub:

👉 github.com/sanjaypatoliya/ai-document-processor-sample

About the Author

I'm Sanjay Patoliya — AWS Certified Solutions Architect & DevOps Engineer building production-ready AI systems on AWS. If you're looking to build something similar or need a remote AWS + AI engineer, feel free to reach out.

• Portfolio: sanjaypatoliya.com
• LinkedIn: linkedin.com/in/sanjaykumar-patoliya-b234a287
• Email: sbpatoliya@gmail.com