I Added Human-in-the-Loop to an Async Queue System — Here's How LangGraph Checkpointing Made It Possible

Most HITL tutorials show you how to pause a graph and wait for input in the same process. That's the easy version. The hard version is when your graph runs inside a queue worker, the gateway that needs to notify the reviewer is a separate service, and you need the paused state to survive a container restart.

This is what I ran into building AuraFlow AI.

---

What AuraFlow Does

AuraFlow is a distributed data cleaning system. The flow:

POST /jobs (NestJS/Fastify/Bun)
    → BullMQ job pushed to Redis
        → Python LangGraph worker (concurrent, 3 threads)
            → Sanitize: strip injections, normalize unicode
            → Parse: LLM cleans raw malformed data into JSON
            → Validate: LLM checks output, returns confidence score
            → Pre-review check: deterministic business rules
            → Human review (if triggered)
        → HTTP callback with retry + idempotency
    → Result persisted to PostgreSQL
GET /jobs/:id/progress → SSE realtime stream

The system handles things like this:

Input:  "BUDI SANTOSO | gaji: Rp 8.500.000 | tgl masuk: 15 Januari 2024"
Output: {"name": "Budi Santoso", "salary": 8500000, "date": "2024-01-15"}

Everything runs asynchronously. The gateway accepts a job and returns a 202 Accepted immediately. The worker processes it in the background and notifies the gateway via HTTP callback when done.

The problem: where does a human reviewer fit into this?

---

Why HITL in an Async System Is Hard

In a synchronous system, HITL is simple. You call interrupt(), the graph pauses, you get a value back from the caller, you resume. Everything is in one process, one thread, one call stack.

In an async queue system, none of that applies:

• The graph runs inside a BullMQ worker process
• The reviewer accesses a separate HTTP API (the gateway)
• The worker and gateway communicate only via Redis pub/sub and HTTP callbacks
• The worker might restart between the pause and the resume
• The paused state needs to survive container crashes

The state has to be persisted somewhere. This is where LangGraph checkpointing becomes essential.

---

The Checkpointing Setup

LangGraph can persist graph state after every node via a checkpointer. When a graph is interrupted, the state at that exact point is serialized and stored. When you resume later — even from a different process — LangGraph reads the checkpoint and picks up from where it left off.

For AuraFlow, the checkpoint backend is Redis (which was already in the stack for BullMQ):

from langgraph.checkpoint.redis.aio import AsyncRedisSaver

async def setup(self):
    self._checkpointer = AsyncRedisSaver(
        redis_url=REDIS_URL,
        ttl={"default_ttl": 1440},  # 24 hours
    )
    await self._checkpointer.asetup()
    self._graph = build_graph(checkpointer=self._checkpointer)

The key thing: thread_id = job_id. Every job gets its own checkpoint namespace. If BullMQ retries a job (same job ID), LangGraph finds the existing checkpoint and resumes from where processing stopped — not from the beginning.

---

The Trigger: Deterministic Business Rules, Not LLM Confidence

My first attempt at HITL used the LLM's confidence score as the trigger. If confidence < 0.9, pause for review. This was a mistake.

LLMs are inconsistent about confidence scores. "Ali Ba" returned confidence 1.0. "A B" returned confidence 1.0. The LLM had no way to know that "Ba" might be initials rather than a surname — it just saw two capitalized words.

The fix was to move the trigger to a deterministic pre_review_check node that runs after validation:

def _check_hitl_rules(cleaned_data: str) -> list[str]:
    reasons = []
    data = json.loads(cleaned_data)

    # Each word in name must be >= 3 characters
    short_words = [w for w in data["name"].split() if len(w) < 3]
    if short_words:
        reasons.append(f"name contains short word(s): {short_words} — may be initials")

    # Salary must be within realistic IDR range
    if data["salary"] < 500_000:
        reasons.append(f"salary {data['salary']} is unusually low")
    if data["salary"] > 500_000_000:
        reasons.append(f"salary {data['salary']} is unusually high")

    # Date must be within reasonable range
    year = data["date"][:4]
    if year < "2000" or year > "2030":
        reasons.append(f"date year {year} is outside expected range")

    return reasons

Empty list = auto complete. Non-empty = pause for review.

This is more reliable than LLM confidence because the rules are auditable, testable, and deterministic. A colleague can read them and understand exactly when human review is triggered. You can't say that about an LLM's internal confidence calibration.

---

How the Pause Works

When pre_review_check returns HITL reasons, the graph routes to human_review_node:

def human_review_node(state: AgentState) -> AgentState:
    _publish_progress(job_id, "pending_review", {
        "cleanedData": state["cleaned_data"],
        "confidence": state["confidence"],
        "hitlReasons": state["hitl_reasons"],
    })

    # This pauses the graph and persists state to Redis
    review_input = interrupt({
        "jobId": job_id,
        "cleanedData": state["cleaned_data"],
        "hitlReasons": state["hitl_reasons"],
        "message": "Data requires human review due to business rule violations.",
    })

    # Execution resumes here when Command(resume=...) is called
    return {
        **state,
        "review_decision": review_input.get("decision", "reject"),
        "review_edited_data": review_input.get("editedData", ""),
        "review_note": review_input.get("note", ""),
    }

interrupt() raises an internal LangGraph exception that unwinds the call stack back to graph.invoke(). The caller receives the current state with result["__interrupt__"] populated.

The worker detects this and sends a pending_review callback to the gateway:

result = self._graph.invoke(initial_state, invoke_config)

if result.get("__interrupt__"):
    await self._send_review_callback(job_id, result)
    # Start background listener for resume signal
    asyncio.create_task(self._listen_for_resume(job_id))
    return {"status": "pending_review"}

The gateway updates the job record in PostgreSQL to status=pending_review. The reviewer can then query GET /jobs/pending-review to see what needs attention.

---

How the Resume Works

The resume path uses Redis pub/sub. When a reviewer submits their decision:

POST /jobs/:id/review
{
  "decision": "approve",
  "reviewedBy": "admin",
  "note": "Ba is a valid short surname"
}

The gateway publishes to job-resume:{jobId} in Redis. The worker has a background coroutine listening on that channel:

async def _listen_for_resume(self, job_id: str):
    pubsub = redis.pubsub()
    await pubsub.subscribe(f"job-resume:{job_id}")

    async for message in pubsub.listen():
        if message["type"] != "message":
            continue

        payload = json.loads(message["data"])

        # Resume graph from checkpoint with reviewer's decision
        result = self._graph.invoke(
            Command(resume={
                "decision": payload["decision"],
                "editedData": payload.get("editedData", ""),
                "note": payload.get("note", ""),
            }),
            {"configurable": {"thread_id": job_id}},
        )

        await self._send_callback(job_id, result, None)
        break

Command(resume=...) tells LangGraph to find the checkpoint for this thread_id, restore state, and continue from where interrupt() was called — injecting the resume value as the return value of interrupt().

Three outcomes are possible:

• approve → graph routes to END, job marked completed
• reject → graph routes back to parse, LLM tries again
• edit → reviewer provides corrected data, graph reparses it

---

The SSE Stream

Reviewers don't need to poll. The gateway streams progress via SSE:

GET /jobs/:id/progress

event: connected
data: {"jobId":"cmt71ahc6...","channel":"job-progress:cmt71ahc6..."}

event: progress
data: {"stage":"sanitizing","timestamp":"..."}

event: progress
data: {"stage":"validated","confidence":0.85,"issues":["name is unusually short"]}

event: progress
data: {"stage":"pending_review","hitlReasons":["name contains short word(s): ['Ba']"]}

event: done
data: {"stage":"pending_review"}

The SSE connection closes when the job reaches a terminal state. For pending_review, it closes when the pause is detected — the reviewer then knows it's their turn.

---

What the Full Flow Looks Like in Practice

Submitting "Ali Ba, 5000000, 2024-01-15":

worker: sanitize_node no_changes_needed len=27
worker: parse_node attempt=1
worker: parse_node output={"name": "Ali Ba", "salary": 5000000, "date": "2024-01-15"}
worker: validate_node is_valid=True confidence=0.85 reason="Valid but name is short"
worker: pre_review_check hitl_triggered reasons=["name contains short word(s): ['Ba']"]
worker: graph_decision result=pending_review
worker: job_interrupted waiting_for_review

gateway: callback_received status=pending_review
gateway: job status → pending_review

# Reviewer checks and approves
POST /jobs/cmt71ahc6.../review {"decision": "approve", "reviewedBy": "admin"}

worker: resume_received decision=approve
worker: graph → END
worker: callback_sent status=completed

gateway: job status → completed, reviewedBy=admin, reviewedAt=...

The whole flow from submit to completed (including human review time) is captured in the job record.

---

The Architecture Decision I'd Revisit

The background listener (_listen_for_resume) is a fire-and-forget coroutine started with asyncio.create_task. If the worker restarts between the pause and the resume, the listener is gone.

The checkpoint survives the restart — LangGraph can still resume the graph. But no one is listening on job-resume:{jobId} anymore.

The correct fix is to make the gateway's POST /jobs/:id/review endpoint directly trigger the resume, rather than publishing to Redis and waiting for the worker to pick it up. The gateway would need to connect to the LangGraph checkpoint store directly and call Command(resume=...) itself.

This would make the architecture cleaner:

• Worker: process jobs, handle interrupts, write checkpoints
• Gateway: accept reviews, resume paused graphs, write results

I haven't implemented this yet. The current approach works for the scale AuraFlow runs at, but it's a real limitation worth naming.

---

What I'd Tell Someone Starting This

Use deterministic rules for HITL triggers, not LLM confidence scores. LLMs are not reliable confidence calibrators. Write code that says "if name word length < 3, flag for review" — not "if LLM says confidence < 0.9, flag for review."

Checkpoint everything before you need it. The value of checkpointing becomes obvious only when something breaks mid-run. Set it up from the start.

Separate the pause signal from the resume trigger. The interrupt tells the system something needs human attention. The resume tells the graph to continue. These are two different communication channels and should be treated as such.

SSE closes at the right moment. When pending_review is the terminal stage for SSE, the reviewer knows exactly when to act. Don't keep the SSE stream open indefinitely.

---

Source code: github.com/awaluddin-dev/auraflow-ai

---

I'm Awaluddin — Backend Engineer & AI Integrator based in Jakarta, currently consulting at an enterprise automotive company. Building toward a fully remote role. Portfolio at awaluddin-dev.vercel.app · LinkedIn · dev.to/awaluddin

Comments

[Jo Do]

Using job_id as thread_id makes the retry story clean, but it also makes the resume endpoint a concurrency boundary. I would require an expected checkpoint version with every reviewer decision and atomically mark that decision consumed before resuming. That turns double-clicks, gateway retries, and two reviewers racing into explicit conflicts instead of two continuations from the same pause. The audit record should keep the rejected attempt too, because it is evidence of a real race.