Building an AI agent with LangGraph, FastAPI and SSE

Resiko is an AI resume optimization agent I built with React, FastAPI, LangGraph, Groq and PostgreSQL. You give it a resume and a job description. It scores the resume against the job, rewrites it, scores it again and keeps going until the score is good enough. The user watches it happen live.

This post walks through the pattern behind it. The code is simplified to show the idea, not copied from the production app.

Why a graph and not a single prompt?

A single "improve this resume" prompt gives you one attempt and no way to tell whether it got better. Splitting the work into steps gives you three things:

• Measurable progress. A scoring step after every rewrite tells you if the change helped.
• A stopping rule. Stop when the score passes a threshold or after N attempts, so cost stays predictable.
• Visibility. Every step is an event you can stream to the UI.

LangGraph models this as a state machine: nodes are functions, edges decide what runs next, and a shared state object flows through it.

1. Define the state

from typing import TypedDict

class ResumeState(TypedDict):
    resume: str
    job_description: str
    ats_score: int
    missing_skills: list[str]
    attempts: int

Keep the state small and explicit. Everything a node needs to read or write goes here, and nothing else.

2. Write the nodes

Each node takes the state and returns only the keys it changed. The LLM calls sit behind small helper functions, which keeps the nodes easy to test.

async def score_ats(state: ResumeState) -> dict:
    result = await llm_score(state["resume"], state["job_description"])
    return {"ats_score": result.score, "missing_skills": result.missing}

async def rewrite(state: ResumeState) -> dict:
    new_resume = await llm_rewrite(
        state["resume"], state["job_description"], state["missing_skills"]
    )
    return {"resume": new_resume, "attempts": state["attempts"] + 1}

Ask the model for structured output (a JSON schema or a Pydantic model) in the scoring step. A number you can compare is much more useful than a paragraph of feedback.

3. Wire the loop

from langgraph.graph import StateGraph, START, END

TARGET_SCORE = 85
MAX_ATTEMPTS = 3

def should_iterate(state: ResumeState) -> str:
    if state["ats_score"] >= TARGET_SCORE or state["attempts"] >= MAX_ATTEMPTS:
        return "done"
    return "again"

graph = StateGraph(ResumeState)
graph.add_node("score_ats", score_ats)
graph.add_node("rewrite", rewrite)
graph.add_edge(START, "score_ats")
graph.add_conditional_edges("score_ats", should_iterate, {"again": "rewrite", "done": END})
graph.add_edge("rewrite", "score_ats")

agent = graph.compile()

The conditional edge is what turns this into an agent. MAX_ATTEMPTS is not optional. Without it, a model that can't reach the target score will loop and burn tokens forever.

4. Stream the steps with FastAPI and SSE

An optimization run takes several LLM calls. Making the user stare at a spinner for that long feels broken. Server-Sent Events (SSE) let the backend push each step as it finishes, over a normal HTTP response.

import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()

class OptimizeRequest(BaseModel):
    resume: str
    job_description: str

@app.post("/optimize")
async def optimize(req: OptimizeRequest):
    initial = {**req.model_dump(), "ats_score": 0, "missing_skills": [], "attempts": 0}

    async def events():
        async for update in agent.astream(initial, stream_mode="updates"):
            for node, changes in update.items():
                yield f"event: {node}\ndata: {json.dumps(changes)}\n\n"
        yield "event: done\ndata: {}\n\n"

    return StreamingResponse(events(), media_type="text/event-stream")

stream_mode="updates" gives you one item per finished node with only the keys that node changed, which maps neatly onto one SSE event per step.

5. Read the stream in React

The browser's built-in EventSource only supports GET requests. Because the resume goes in a POST body, I read the stream with fetch instead:

const res = await fetch("/optimize", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ resume, job_description }),
});
const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
let buffer = "";
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  buffer += value;
  const events = buffer.split("\n\n");
  buffer = events.pop();               // keep any half-received event
  for (const raw of events) handleEvent(raw);  // update score, show diff, etc.
}

Lessons from production

• Pick a fast inference provider. In a loop, latency multiplies. Fast inference (I use Groq) is the difference between a live feel and a long wait.
• Cap everything. Max attempts, max tokens per call and a request timeout. Agents fail in creative ways.
• Persist the result, not the stream. Save the final state to PostgreSQL once the run ends. The stream is for the user's eyes.
• Disable proxy buffering. Some reverse proxies buffer responses and your "live" stream arrives all at once. Send X-Accel-Buffering: no or turn buffering off for that route.
• Make the output useful outside the app. Resiko renders the final resume to PDF with WeasyPrint. The agent is only valuable if the result leaves the screen.