DEV Community: Sanjay Patoliya

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

Sanjay Patoliya — Wed, 24 Jun 2026 14:03:35 +0000

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:

Upload a PDF or image — it goes directly to S3 via presigned URL
Watch real-time processing status as Textract, Claude, and the embedding pipeline run — via WebSocket
Read the AI-generated classification and summary
Ask questions in natural language — get answers with cited page numbers
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

Layer	Technology
Frontend	React 18 + TypeScript + Vite + TailwindCSS + React Query
Backend	FastAPI (Python 3.12) + SQLAlchemy async + Alembic
AI	Anthropic Claude (`claude-sonnet-4-6`)
Embeddings	Voyage AI (`voyage-3`)
OCR	Amazon Textract
Vector Search	pgvector (Aurora PostgreSQL Serverless v2)
Auth	AWS Cognito + JWT
Real-time	API Gateway WebSocket
Hosting	ECS Fargate + ALB + CloudFront
IaC	AWS CDK (Python) — 8 stacks
Local Dev	Docker Compose (PostgreSQL + Redis + LocalStack)

AWS Services Used

Service	Purpose
ECS Fargate	Serverless container hosting for FastAPI
Application Load Balancer	Routes traffic to ECS tasks
CloudFront	CDN + HTTPS termination
S3	Document storage with presigned URLs
Amazon Textract	OCR, table extraction, form parsing
AWS Lambda	Event-driven async processing pipeline
Amazon SQS + SNS	Decoupled processing with dead letter queues
API Gateway WebSocket	Real-time processing status notifications
Aurora Serverless v2 (pgvector)	Document metadata and vector embeddings
ElastiCache Redis	Rate limiting and response caching
AWS Cognito	JWT authentication and user management
SSM Parameter Store	Secure API key storage
ECR	Docker image registry
AWS X-Ray	Distributed tracing
CloudWatch	Metrics, dashboards, cost alarms
AWS CDK (Python)	Full infrastructure as code

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

Method	Endpoint	Description
GET	`/health`	Health check
POST	`/api/v1/upload/presigned-url`	Get presigned S3 POST URL
POST	`/api/v1/upload/complete/{doc_id}`	Confirm upload, trigger processing
GET	`/api/v1/documents`	List documents (paginated)
GET	`/api/v1/documents/{doc_id}`	Get document metadata
DELETE	`/api/v1/documents/{doc_id}`	Delete document
POST	`/api/v1/documents/{doc_id}/chat`	RAG chat — streams SSE tokens
POST	`/api/v1/search`	Global search across all documents
WebSocket	`/api/v1/ws`	Real-time processing status

Infrastructure with AWS CDK — 8 Stacks

Each stack has a single concern and is deployed independently.

Stack	What it creates
`DocuAI-Network`	VPC, 2 AZs, NAT Gateway, S3 + SQS VPC endpoints
`DocuAI-Storage`	Raw + processed S3 buckets, SQS queue + DLQ
`DocuAI-Auth`	Cognito User Pool + App Client
`DocuAI-Database`	Aurora Serverless v2 (PostgreSQL + pgvector)
`DocuAI-Processing`	Lambda Orchestrator (S3 trigger) + Lambda Processor (SNS)
`DocuAI-WebSocket`	API Gateway WebSocket + Lambda connect/disconnect/send
`DocuAI-Compute`	ECS Fargate service + ALB + ECR
`DocuAI-Monitoring`	CloudWatch dashboards + alarms

# 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
GitHub: github.com/sanjaypatoliya

Building an AI Research Agent with LangGraph, Claude, and AWS

Sanjay Patoliya — Thu, 11 Jun 2026 15:00:12 +0000

What I Built

An AI research agent that answers any question by searching the web, Wikipedia, and academic papers — then streams its reasoning back to the browser in real time.

User flow:

Ask any research question in natural language
Watch the agent decide which tools to call — live, as it happens
Read a structured report: Summary, Key Findings, Academic Research, Sources, Conclusion
Ask follow-up questions — the agent remembers the full conversation

The key difference from a simple chatbot: this agent is autonomous. It decides when to search, what to search for, and which source to use. Claude does the reasoning; LangGraph controls the loop.

Architecture

Browser
  │
  ▼
CloudFront (HTTPS)
  ├── /* ──────────────► S3 (React static files)
  └── /api/v1/* ───────► ALB
                           │
                           ▼
                      ECS Fargate (FastAPI)
                           │
                           ▼
                      LangGraph ReAct Agent
                           │
              ┌────────────┼────────────┐
              ▼            ▼            ▼
        Tavily API    Wikipedia API  arXiv API
      (web search)   (background    (academic
                       facts)        papers)
              └────────────┼────────────┘
                           ▼
                     Anthropic Claude
                   (reasoning + answer)
                           │
                           ▼
                   SSE Stream → Browser
               (tokens + tool trace events)
                           │
                           ▼
                       DynamoDB
                  (conversation history)

Same CloudFront pattern as my previous post on the AI Resume Analyzer — /api/v1/* proxies to the ALB so the frontend has a single HTTPS endpoint with no CORS issues.

Tech Stack

Layer	Technology
Frontend	React 18 + TypeScript + Vite + TailwindCSS
Backend	FastAPI (Python 3.12)
AI Agent	LangGraph ReAct + Anthropic Claude (claude-sonnet-4-6)
Web Search	Tavily API
Knowledge	Wikipedia API
Academic Research	arXiv API
Persistence	Amazon DynamoDB
Streaming	Server-Sent Events (SSE)
Hosting	ECS Fargate + ALB + CloudFront
IaC	AWS CDK (Python)

AWS Services Used

Service	Purpose
ECS Fargate	Serverless container hosting for FastAPI
Application Load Balancer	Routes traffic to ECS tasks
CloudFront	CDN + HTTPS termination + SSE proxy
S3	React static file hosting
DynamoDB	Conversation history with GSI
SSM Parameter Store	Secure API key storage (Anthropic + Tavily)
ECR	Docker image registry
VPC + NAT Gateway	Private network with outbound internet for API calls
AWS CDK (Python)	Infrastructure as Code

The Core: LangGraph ReAct Agent

The most interesting part of this project is the agent itself. Instead of a fixed pipeline (step 1 → step 2 → step 3), a ReAct agent reasons its way to an answer:

Reason — think about what information is needed
Act — call a tool (web search, Wikipedia, arXiv)
Observe — read the result
Repeat — until it has enough to answer

LangGraph models this as a StateGraph — a directed graph where each node is either the Claude model or a tool executor, and edges define when to call tools vs. when to return the final answer.

# agent/graph.py
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

def build_agent(tools: list) -> CompiledGraph:
    model = ChatAnthropic(model=settings.model_name).bind_tools(tools)

    def call_model(state: AgentState) -> dict:
        messages = [SystemMessage(content=SYSTEM_PROMPT)] + state["messages"]
        response = model.invoke(messages)
        return {"messages": [response]}

    def should_continue(state: AgentState) -> str:
        last = state["messages"][-1]
        return "tools" if last.tool_calls else END

    graph = StateGraph(AgentState)
    graph.add_node("agent", call_model)
    graph.add_node("tools", ToolNode(tools))
    graph.set_entry_point("agent")
    graph.add_conditional_edges("agent", should_continue)
    graph.add_edge("tools", "agent")
    return graph.compile()

The should_continue function is the key decision point: if Claude's response includes tool calls, route to the tools node; otherwise, the answer is ready and the graph ends.

AgentState and Conversation Memory

AgentState holds the full message history using LangGraph's add_messages reducer — new messages are appended to the list rather than replacing it:

# agent/state.py
from langgraph.graph import add_messages
from typing import Annotated
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

This single field is all that's needed for multi-turn memory. Each conversation starts from DynamoDB history, so follow-up questions have full context of everything said before.

The Three Research Tools

The agent has three tools. Claude autonomously decides which ones to call based on the question.

Tavily — Web Search

Tavily is purpose-built for LLM agents. It returns clean, structured results without the noise of a raw search engine.

# agent/tools.py
from langchain_community.tools.tavily_search import TavilySearchResults

def get_tools() -> list:
    web_search = TavilySearchResults(
        max_results=5,
        description="Search the web for current information, news, and general knowledge."
    )
    ...

Wikipedia — Background Facts

For well-established topics, Wikipedia gives reliable background without burning Tavily quota.

@tool
def wikipedia_search(query: str) -> str:
    """Search Wikipedia for background information on a topic."""
    wiki = wikipediaapi.Wikipedia(language="en", user_agent="ai-research-agent/1.0")
    page = wiki.page(query)
    if not page.exists():
        return f"No Wikipedia article found for: {query}"
    # Truncate to 3000 chars to stay within Claude's context budget
    return page.summary[:3000]

arXiv — Academic Papers

For scientific or technical questions, the agent searches arXiv for peer-reviewed research. This differentiates the output from a standard web search.

@tool
def arxiv_search(query: str) -> str:
    """Search arXiv for academic papers and research on a topic."""
    import arxiv
    results = arxiv.Search(query=query, max_results=3,
                           sort_by=arxiv.SortCriterion.Relevance)
    papers = []
    for r in arxiv.Client().results(results):
        papers.append(f"Title: {r.title}\nAuthors: {', '.join(a.name for a in r.authors)}\n"
                      f"Summary: {r.summary[:500]}\nURL: {r.entry_id}")
    return "\n\n---\n\n".join(papers) if papers else "No papers found."

SSE Streaming — Tokens + Live Agent Trace

This is the part that makes the UI feel alive. Instead of waiting 20 seconds for a complete answer, the browser receives two types of events simultaneously:

Trace events — every tool call the agent makes, shown in a live panel
Token events — the answer streaming word by word

The SSE event protocol:

data: {"type": "conversation_id", "conversation_id": "abc-123"}

data: {"type": "trace", "step": "tool_start", "tool": "web_search", "input": "LangGraph tutorial"}

data: {"type": "trace", "step": "tool_end", "tool": "web_search", "output": "LangGraph is..."}

data: {"type": "token", "content": "LangGraph is a "}

data: {"type": "token", "content": "framework for building..."}

data: {"type": "done"}

The streaming endpoint uses LangGraph's astream_events — an async generator that yields every event in the agent's execution graph:

# routers/chat.py
async def stream_agent_response(message: str, conversation_id: str):
    history = await dynamodb_service.get_messages(conversation_id)
    state = {"messages": history + [HumanMessage(content=message)]}

    async for event in agent.astream_events(state, version="v2"):
        kind = event["event"]

        if kind == "on_tool_start":
            yield f'data: {{"type":"trace","step":"tool_start","tool":"{event["name"]}","input":"{event["data"].get("input","")[:200]}"}}\n\n'

        elif kind == "on_tool_end":
            yield f'data: {{"type":"trace","step":"tool_end","tool":"{event["name"]}","output":"{str(event["data"].get("output",""))[:300]}"}}\n\n'

        elif kind == "on_chat_model_stream":
            content = event["data"]["chunk"].content
            if content:
                yield f'data: {json.dumps({"type": "token", "content": content})}\n\n'

    yield 'data: {"type": "done"}\n\n'

The FastAPI endpoint sets media_type="text/event-stream" and returns a StreamingResponse — no WebSockets needed.

Multi-turn Conversations

Every conversation is stored in DynamoDB. When the user sends a follow-up question, the full history is loaded and passed back into the agent — so it has full context of everything discussed.

DynamoDB Schema

The table uses a single-table design with a GSI for listing conversations by recency:

Key	Value	Purpose
`pk`	`CONV#{conversation_id}`	Partition key
`sk`	`META` or `MSG#{timestamp}`	Sort key
`entity_type`	`conversation` or `message`	GSI partition key
`updated_at`	ISO timestamp	GSI sort key

The GSI (entity-type-index) makes it efficient to list all conversations sorted by most recently updated — without scanning the full table.

# services/dynamodb_service.py
async def save_messages(conversation_id: str, messages: list[BaseMessage]) -> None:
    with table.batch_writer() as batch:
        for i, msg in enumerate(messages):
            batch.put_item(Item={
                "pk": f"CONV#{conversation_id}",
                "sk": f"MSG#{i:06d}",
                "entity_type": "message",
                "role": msg.type,
                "content": msg.content,
                "updated_at": datetime.utcnow().isoformat(),
            })

API Endpoints

Method	Endpoint	Description
GET	`/health`	Health check
POST	`/api/v1/chat/stream`	Stream agent response + tool trace via SSE
GET	`/api/v1/conversations`	List all conversations
GET	`/api/v1/conversations/{id}`	Get conversation with full message history
DELETE	`/api/v1/conversations/{id}`	Delete a conversation

Infrastructure with AWS CDK

Two stacks — simpler than the resume analyzer because the frontend is a pure static site with no server-side rendering.

BackendStack

VPC with 2 AZs, 1 NAT Gateway, ECS Fargate service, and ALB. Both API keys are injected from SSM Parameter Store at container startup.

# stacks/backend_stack.py (simplified)
fargate_service = ecs_patterns.ApplicationLoadBalancedFargateService(
    self, "BackendService",
    cluster=cluster,
    cpu=512,
    memory_limit_mib=1024,
    task_image_options=ecs_patterns.ApplicationLoadBalancedTaskImageOptions(
        image=ecs.ContainerImage.from_ecr_repository(ecr_repo),
        environment={"DYNAMODB_TABLE": table.table_name},
        secrets={
            "ANTHROPIC_API_KEY": ecs.Secret.from_ssm_parameter(anthropic_key),
            "TAVILY_API_KEY": ecs.Secret.from_ssm_parameter(tavily_key),
        },
    ),
)
# Long agent reasoning chains need more than the default 60s
fargate_service.load_balancer.set_attribute("idle_timeout.timeout_seconds", "120")

FrontendStack

CloudFront with two important timeout settings for SSE:

# stacks/frontend_stack.py (simplified)
api_behaviour = cloudfront.BehaviorOptions(
    origin=alb_origin,
    viewer_protocol_policy=cloudfront.ViewerProtocolPolicy.REDIRECT_TO_HTTPS,
    cache_policy=cloudfront.CachePolicy.CACHING_DISABLED,
    origin_request_policy=cloudfront.OriginRequestPolicy.ALL_VIEWER,
    # SSE streams can run for 60-120s — increase read timeout
    # Default is 30s; streaming connections silently drop without this
)

The CloudFront read timeout (default 30s) must be extended for SSE. Unlike REST endpoints, SSE connections stay open while the agent reasons and writes — they can easily exceed 30 seconds on complex questions.

Testing Strategy

Backend — pytest with mocked astream_events and DynamoDB. No real API calls needed.

cd backend
pytest                          # all tests
pytest tests/agent/             # ReAct agent + tools
pytest tests/routers/           # API endpoints + SSE format
pytest --cov=app                # with coverage report

Test file	What it covers
`test_state.py`	`add_messages` appends correctly, preserves history
`test_tools.py`	Wikipedia found/not found/truncation, arXiv search, `get_tools` returns 3 tools
`test_chat.py`	SSE event format, trace events, token events, done/error, input validation
`test_conversations.py`	List/get/delete endpoints, 404 handling

Frontend — Vitest + Testing Library.

cd frontend
npm test
npm run test:coverage

Lessons Learned

1. ReAct agents decide — you don't have to hardcode logic
A fixed pipeline would need explicit code for "if scientific question → use arXiv". The ReAct loop lets Claude make that call based on context. The agent often combines all three tools on a single question.

2. CloudFront read timeout must be increased for SSE
The default CloudFront read timeout is 30 seconds. SSE connections for complex research questions run longer. Without setting this to 120s, streams silently drop mid-response — no error, just a frozen browser. This was the hardest bug to diagnose.

3. ALB idle timeout must also be extended
Both CloudFront (read timeout) and ALB (idle timeout) need to be set to 120s. Fixing only one still drops long connections. The ALB's default 60s kicks in before CloudFront's if not changed.

4. NAT Gateway is required for private ECS tasks
ECS tasks in a private subnet need a NAT Gateway to reach Anthropic, Tavily, Wikipedia, and arXiv. Without it, all external API calls silently time out — no DNS error, just a hanging request.

5. add_messages is all you need for conversation memory
LangGraph's add_messages reducer appends new messages to the state list automatically. Passing the full DynamoDB history into the initial state gives the agent complete context of past turns with zero extra logic.

6. Build Docker images for linux/amd64 on Apple Silicon
ECS Fargate runs on x86 by default. Building on an M-series Mac without specifying the platform creates an arm64 image that won't start on Fargate. Always add --platform linux/amd64 to the Docker build, or set it in the Dockerfile.

7. Two SSM parameters — both injected at container startup
This project needs two API keys: Anthropic and Tavily. Both are stored as SSM SecureString and injected as environment variables at container startup. Keys never appear in CDK output, CloudFormation templates, or Docker images.

GitHub

The full source code is available on GitHub:

👉 github.com/sanjaypatoliya/ai-research-agent

About the Author

I'm Sanjay Patoliya — AWS Certified engineer with 7 AWS certifications building production-ready AI systems on AWS.

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

Originally published at sanjaypatoliya.com

Building an AI Resume Scorer with AWS Textract and Claude

Sanjay Patoliya — Thu, 11 Jun 2026 14:43:44 +0000

What I Built

A full-stack AI application that scores how well a resume matches a job description — giving candidates a score from 0–100 across 5 categories with specific improvement suggestions.

User flow:

Upload a resume PDF
Paste a job description
Get an AI-powered match score with detailed feedback
View history of past analyses

The entire backend pipeline runs on AWS: Textract extracts the resume text, Claude AI does the intelligent scoring, and results are saved to DynamoDB.

Architecture

Browser
  │
  ▼
CloudFront (HTTPS)
  ├── /* ──────────────► S3 (React static files)
  └── /api/v1/* ───────► ALB
                           │
                           ▼
                      ECS Fargate (FastAPI)
                           │
                      S3 — fetch uploaded PDF
                           │
                      Textract — extract text (OCR)
                           │
                      Claude AI — score + suggestions
                           │
                      DynamoDB — save result

The key architectural decision was routing /api/v1/* through CloudFront to the ALB. This means the frontend uses a single HTTPS endpoint for both static files and API calls — no mixed content issues, no CORS problems.

Tech Stack

Layer	Technology
Frontend	React 18 + TypeScript + Vite + TailwindCSS
Backend	FastAPI (Python 3.12)
AI	Anthropic Claude (claude-sonnet-4-6)
OCR	AWS Textract
Storage	Amazon S3 + DynamoDB
Hosting	ECS Fargate + ALB + CloudFront
IaC	AWS CDK (Python)

AWS Services Used

Service	Purpose
ECS Fargate	Serverless container hosting for FastAPI
Application Load Balancer	Routes traffic to ECS tasks
CloudFront	CDN + HTTPS termination
S3	Resume PDF storage + React static hosting
DynamoDB	Analysis results and history
Textract	Extract text from resume PDFs
SSM Parameter Store	Secure Anthropic API key storage
ECR	Docker image registry
VPC + NAT Gateway	Private network with internet access for Claude API
AWS CDK (Python)	Infrastructure as Code

The Analysis Pipeline

When a user submits a resume and job description, this is what happens:

Step 1 — PDF Upload via Presigned URL

The frontend requests a presigned S3 URL from the backend, then uploads the PDF directly to S3 from the browser. This avoids routing the binary file through the FastAPI server.

# POST /api/v1/upload
def get_presigned_url(file_name: str, content_type: str) -> dict:
    s3_key = f"uploads/{uuid4()}/{file_name}"
    upload_url = s3_client.generate_presigned_url(
        "put_object",
        Params={"Bucket": S3_BUCKET, "Key": s3_key, "ContentType": content_type},
        ExpiresIn=300,
    )
    return {"upload_url": upload_url, "s3_key": s3_key}

Step 2 — Text Extraction with AWS Textract

Once the PDF is in S3, Textract extracts all the text. Textract handles complex PDF layouts — multi-column resumes, tables, headers — much better than simple PDF parsers.

# textract_service.py
def extract_text_from_s3(s3_key: str) -> str:
    response = textract_client.detect_document_text(
        Document={"S3Object": {"Bucket": S3_BUCKET, "Name": s3_key}}
    )
    blocks = [b["Text"] for b in response["Blocks"] if b["BlockType"] == "LINE"]
    return "\n".join(blocks)

Note: detect_document_text is synchronous and works well for standard resumes. If you need to support 10+ page documents, switch to start_document_analysis (asynchronous) — it handles larger files without timing out.

Step 3 — AI Scoring with Claude

The extracted resume text and job description are sent to Claude with a structured prompt that asks for scores across 5 categories and improvement suggestions.

The 5 scoring categories:

Skills Match — how well technical skills align
Experience Level — years and seniority match
Education — degree and field relevance
Keywords — ATS-critical keyword presence
ATS Formatting — resume structure and formatting

Claude returns a structured JSON response. To ensure the LLM's non-deterministic output doesn't break the frontend, I used Pydantic in the FastAPI layer to validate the JSON structure before saving to DynamoDB — if Claude returns an unexpected format, the request fails fast with a clear error rather than silently corrupting data.

{
  "overall_score": 78,
  "categories": [
    { "name": "Skills Match", "score": 85, "rationale": "Strong Python and AWS skills" },
    { "name": "Experience Level", "score": 70, "rationale": "5 years matches requirement" },
    { "name": "Education", "score": 90, "rationale": "Relevant CS degree" },
    { "name": "Keywords", "score": 75, "rationale": "Most keywords present" },
    { "name": "ATS Formatting", "score": 65, "rationale": "Some formatting improvements needed" }
  ],
  "suggestions": [
    "Add quantifiable achievements to your experience section",
    "Include AWS certification names explicitly"
  ]
}

Step 4 — Save to DynamoDB

Results are saved with a UUID as the partition key. DynamoDB PAY_PER_REQUEST billing means no capacity planning and costs nothing when idle.

# dynamodb_service.py
def save_result(result: AnalysisResult) -> None:
    table.put_item(Item={
        "id": result.id,
        "overall_score": result.overall_score,
        "categories": result.categories,
        "suggestions": result.suggestions,
        "created_at": result.created_at.isoformat(),
        "file_name": result.file_name,
    })

API Endpoints

Method	Endpoint	Description
GET	`/health`	Health check
POST	`/api/v1/upload`	Get presigned S3 URL
POST	`/api/v1/analyze`	Analyze resume vs job description
GET	`/api/v1/history`	List past analyses
GET	`/api/v1/history/{id}`	Get single result
DELETE	`/api/v1/history/{id}`	Delete a result

Infrastructure with AWS CDK

All AWS resources are defined as code using CDK Python, split into 4 stacks:

StorageStack

S3 bucket with server-side encryption and 30-day lifecycle expiry for uploaded PDFs.

DatabaseStack

DynamoDB table with PAY_PER_REQUEST billing — no capacity planning needed.

BackendStack

The most complex stack. VPC with 2 AZs, 1 NAT Gateway (ECS tasks need outbound internet to reach the Anthropic API), ECS Fargate service, and Application Load Balancer.

Key decision: ECS Fargate instead of Lambda for the backend. The Claude AI response can take 10–30 seconds for detailed analysis. Lambda has a 15-minute limit but cold starts and the complexity of streaming responses made Fargate a cleaner choice for this workload. Fargate's stability during long-running LLM inferences over Lambda's potential timeout and cold-start issues was a deliberate choice for reliability.

# backend_stack.py (simplified)
fargate_service = ecs_patterns.ApplicationLoadBalancedFargateService(
    self, "BackendService",
    cluster=cluster,
    cpu=256,
    memory_limit_mib=512,
    task_image_options=ecs_patterns.ApplicationLoadBalancedTaskImageOptions(
        image=ecs.ContainerImage.from_ecr_repository(ecr_repo),
        environment={"APP_ENV": "production"},
        secrets={"ANTHROPIC_API_KEY": ecs.Secret.from_ssm_parameter(api_key_param)},
    ),
)
# ALB timeout extended for Claude response time
fargate_service.load_balancer.set_attribute("idle_timeout.timeout_seconds", "120")

FrontendStack

CloudFront with OAC (Origin Access Control) for S3, SPA routing (404 → index.html), and a /api/v1/* behaviour that proxies to the ALB. Cache invalidation runs automatically on each deploy. This pattern — CloudFront + ALB + S3 — is my go-to for production SPAs. It eliminates the CORS dance entirely and centralises SSL termination at the edge.

Deploy Pipeline

A single make deploy command runs everything:

make deploy
# 1. Runs backend tests (pytest)
# 2. Runs frontend tests (vitest)
# 3. Builds React app (npm run build)
# 4. CDK deploys all 4 stacks

Deployment stops if any test fails — no broken code reaches AWS.

Testing Strategy

Backend — pytest with moto to mock all AWS services. No real AWS credentials needed for tests.

cd backend
pytest                    # all tests with coverage
pytest tests/services/    # service layer only
pytest tests/routers/     # API endpoints only

Frontend — Vitest + Testing Library.

cd frontend
npm test                  # all tests
npm run test:coverage     # with coverage report

Lessons Learned

1. Use presigned URLs for file uploads
Routing PDFs through FastAPI adds unnecessary load and latency. Direct S3 upload from the browser via presigned URL is faster and cheaper.

2. ECS Fargate over Lambda for long AI calls
Lambda works for fast operations but AI scoring takes 10–30 seconds. Fargate containers are always warm and handle this naturally.

3. NAT Gateway is required for private ECS tasks
ECS tasks in a private subnet need a NAT Gateway to reach the Anthropic API. I initially tried without it and the AI calls silently timed out.

4. Extend ALB idle timeout for AI workloads
The default ALB idle timeout is 60 seconds. Claude can take longer for detailed analysis. Setting it to 120s prevents connection resets mid-response.

5. CloudFront routing solves CORS completely
Proxying /api/v1/* through CloudFront to the ALB means the frontend and API share the same domain. Zero CORS configuration needed.

6. SSM Parameter Store for secrets
Never put API keys in environment variables or Docker images. SSM SecureString is injected at container startup — keys never appear in logs or CDK output.

7. Auth is intentionally deferred to v2
The current version has no user authentication. Auth with AWS Cognito is planned for v2: user login, private history per user, and Cognito-authorizer on the API Gateway. Deferring auth kept v1 scope small and shippable.

GitHub

The full source code is available on GitHub:

👉 github.com/sanjaypatoliya/ai-resume-analyzer

About the Author

I'm Sanjay Patoliya — AWS Certified engineer with 7 AWS certifications building production-ready AI systems on AWS.

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

Originally published at sanjaypatoliya.com