DEV Community: Awaluddin

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

Awaluddin — Wed, 16 Sep 2026 03:04:52 +0000

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

I Built a Multi-Provider LLM Router for My AI Worker - Here's What I Learned

Awaluddin — Mon, 24 Aug 2026 11:59:37 +0000

I Built a Multi-Provider LLM Router for My AI Worker — Here's What I Learned

When I started integrating LLMs into my side project AuraFlow AI, I made the same mistake most backend engineers make: I hardcoded a single provider.

One week in, Gemini free tier hit its rate limit at 11 PM while I was testing. Everything stopped. I had to manually swap the API key, restart the worker, and lose 20 minutes of debugging momentum.

That was the last time I let a single LLM provider be a single point of failure.

What Is AuraFlow AI?

AuraFlow AI is a distributed data cleaning system I built for my portfolio. The architecture is straightforward:

POST /jobs (NestJS/Fastify/Bun)
    → BullMQ job pushed to Redis
        → Python LangGraph worker picks up job
            → Parser Agent: clean raw/malformed data using LLM
            → Validator Agent: verify output, loop back if invalid
        → HTTP callback with retry + idempotency
    → Result persisted to PostgreSQL

The LLM is at the core of both agents. If the LLM provider goes down, the entire system stops. That's the problem I needed to solve.

The Naive Version

My first implementation looked like this:

from langchain_google_genai import ChatGoogleGenerativeAI

llm = ChatGoogleGenerativeAI(
    model="gemini-3.5-flash",
    google_api_key=os.getenv("GEMINI_API_KEY")
)

def parse_node(state):
    result = llm.invoke(prompt)
    return {"cleaned_data": result.content}

Simple. Works. Completely fragile.

The problems were obvious once I started thinking about production:

Rate limit on Gemini free tier? System down.
Want to switch to a cheaper provider? Change code, redeploy.
Want to test with a local model? Swap the entire implementation.
Gemini API has an outage? System down until I wake up and notice.

I needed a proper abstraction layer.

The Design

I wanted three things:

Multiple providers — Gemini, OpenAI, Claude, Groq, and any OpenAI-compatible custom endpoint
Automatic fallback — if provider A fails, try B, then C, without any code change
Zero-code configuration — swap providers and priority via environment variables

The result is LLMRouter: a registry-based provider abstraction with a configurable fallback chain.

class LLMRouter:
    def __init__(self):
        self._chain: list[tuple[str, BaseChatModel]] = []
        self._build_chain()

    def _build_chain(self):
        # Priority order from env — change without touching code
        raw = os.getenv("LLM_PROVIDER_ORDER", "gemini,groq,openai,claude,custom")
        order = [p.strip() for p in raw.split(",")]

        for provider_name in order:
            loader, default_model = PROVIDER_REGISTRY[provider_name]
            model = os.getenv(f"{provider_name.upper()}_MODEL", default_model)
            instance = loader(model)

            if instance:
                self._chain.append((provider_name, instance))
                logger.info("Provider loaded: %s (model: %s)", provider_name, model)
            else:
                logger.info("Provider skipped (no API key): %s", provider_name)

        if not self._chain:
            raise RuntimeError("No LLM provider available.")

        logger.info("Fallback chain: %s",
            " -> ".join(name for name, _ in self._chain))

    def invoke(self, prompt: str) -> str:
        last_error = None
        for provider_name, llm in self._chain:
            try:
                result = llm.invoke(prompt)
                return result.content.strip()
            except Exception as e:
                logger.warning("Provider %s failed: %s", provider_name, e)
                last_error = e
                continue
        raise RuntimeError(f"All providers failed. Last error: {last_error}")

Each provider has its own loader function that returns None if the API key is not set:

def _load_gemini(model: str) -> Optional[BaseChatModel]:
    api_key = os.getenv("GEMINI_API_KEY")
    if not api_key:
        return None  # Skip silently — not an error
    try:
        from langchain_google_genai import ChatGoogleGenerativeAI
        return ChatGoogleGenerativeAI(model=model, google_api_key=api_key)
    except Exception as e:
        logger.warning("Gemini load failed: %s", e)
        return None

The registry maps provider names to loader functions and default models:

PROVIDER_REGISTRY = {
    "gemini":       (_load_gemini,       "gemini-3.6-flash"),
    "openai":       (_load_openai,       "gpt-4o-mini"),
    "claude":       (_load_claude,       "claude-haiku-4-5-20251001"),
    "groq":         (_load_groq,         "openai/gpt-oss-20b"),
    "azure_openai": (_load_azure_openai, "gpt-4o-mini"),
    "custom":       (_load_custom,       "llama3.2"),
}

The Custom Endpoint — The Part I Didn't Expect to Need

The most useful feature ended up being one I almost didn't build: the custom OpenAI-compatible endpoint.

def _load_custom(model: str) -> Optional[BaseChatModel]:
    base_url = os.getenv("CUSTOM_LLM_BASE_URL")
    if not base_url:
        return None
    from langchain_openai import ChatOpenAI
    return ChatOpenAI(
        model=model,
        base_url=base_url,
        api_key=os.getenv("CUSTOM_LLM_API_KEY", "custom"),
    )

This works with anything that implements the OpenAI chat completions spec: Ollama, OpenRouter, vLLM, or a third-party API gateway. In my case, I'm using a custom API gateway as primary provider with Gemini as fallback:

# .env
LLM_PROVIDER_ORDER=custom,gemini
CUSTOM_LLM_BASE_URL=https://my-api-gateway.example.com/v1
CUSTOM_LLM_API_KEY=my_key
CUSTOM_LLM_MODEL=claude-opus-4-8
GEMINI_MODEL=gemini-3.6-flash

Startup log confirms the chain:

Provider loaded: custom (model: claude-opus-4-8)
Provider loaded: gemini (model: gemini-3.6-flash)
Provider skipped (no API key): groq
Provider skipped (no API key): openai
Fallback chain: custom -> gemini

What Actually Happened in Production (Accidental Stress Test)

I didn't plan this, but it happened during testing and it validated the entire design.

I submitted a job while Docker Compose was running. The gateway crashed mid-flight (exited with code 137 — OOM killed). Here's the exact log sequence from the worker:

job_received job_id=cmsi1m2dn000001pmnnhwojpf
parse_node attempt=1
Invoking provider: custom
parse_node output={"name": "Test Retry1", ...}
validate_node attempt=1
validate_node is_valid=False reason=name must be two words with each word
    capitalized and contain only letters; 'Retry1' contains a digit
graph_decision result=retry attempt=1

parse_node attempt=2
parse_node output={"name": "Test Retry", ...}
validate_node is_valid=True reason=OK
graph_decision result=valid attempts=2

job_finished job_id=cmsi1m2dn000001pmnnhwojpf is_valid=True attempts=2

callback_connection_error ... attempt=1/5
callback_retry_scheduled delay=1.0s next_attempt=2
callback_connection_error ... attempt=2/5
callback_retry_scheduled delay=2.0s next_attempt=3
callback_connection_error ... attempt=3/5
callback_retry_scheduled delay=4.0s next_attempt=4
callback_sent status_code=200 attempt=4

Three things happened simultaneously that I didn't orchestrate:

1. LangGraph retry loop worked. The validator correctly rejected Test Retry1 because Retry1 contains a digit. The parser received the rejection reason and fixed it on attempt 2. The feedback loop between validator and parser is exactly what makes this more than a simple LLM call.

2. Callback retry with exponential backoff worked. The gateway was down. The worker kept retrying with 1s → 2s → 4s delays. When the gateway came back up, attempt 4 succeeded. No data was lost.

3. The worker kept running. Because LangGraph and the callback retry are separate from the gateway's lifecycle, the worker finished its job correctly even though the service it was trying to reach was down.

Configuration — 5 Providers, Zero Code Changes

This is what I was aiming for. To switch from custom + Gemini to OpenAI + Groq:

# Before
LLM_PROVIDER_ORDER=custom,gemini

# After — no code change, just env update
LLM_PROVIDER_ORDER=openai,groq
OPENAI_API_KEY=sk-...
GROQ_API_KEY=gsk_...

To override the model for a specific provider:

GEMINI_MODEL=gemini-3.6-flash
OPENAI_MODEL=gpt-4o
GROQ_MODEL=openai/gpt-oss-20b

The router reads these at startup and builds the chain. Five providers configured, but only the ones with valid credentials get loaded.

What I'd Do Differently

Health check before adding to chain. Right now, _load_custom succeeds if CUSTOM_LLM_BASE_URL is set — even if the endpoint is unreachable. A lightweight ping at startup would catch misconfigured endpoints before the first real job comes in.

Provider-level metrics. I want to know which provider is being invoked most often and which ones are failing. Right now this lives in logs. It should be structured data I can query.

Circuit breaker per provider. If Gemini returns 5xx three times in a row, stop trying Gemini for the next 60 seconds before retrying. Right now every invoke attempt goes through the full retry before moving to the next provider in the chain.

The Takeaway

Single provider = single point of failure. This is obvious in hindsight, but it took an 11 PM rate limit hit to make me actually fix it.

The pattern is simple: registry of loaders, environment-configured priority, iterate and fallback on exception. About 100 lines of Python. The complexity is low; the resilience gain is significant.

If you're building any backend system that calls an LLM, this abstraction is worth the hour it takes to implement.

AuraFlow AI 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. You can find my work at awaluddin-dev.vercel.app or connect on LinkedIn.

I Rewrote a Fintech Platform Alone — No Handover, No Team, No Docs

Awaluddin — Fri, 26 Jun 2026 08:30:19 +0000

The OJK auditor was on the call. I was sharing my screen. And the app crashed.
No senior engineer to ping. No team lead to escalate to. Just me, a system I had rebuilt from scratch over the past several months, and a government regulator watching in real time.
That moment is burned into my memory — not as the worst day of my career, but as the most clarifying one.

The Situation I Walked Into
I joined a fintech startup as the sole backend developer. What I found on day one: no handover document, no architecture diagram, no README, no one to ask. The previous developer had already left.
The existing system was a PHP Laravel monolith running on CentOS, with a React Native mobile app, MySQL, and two Alibaba Cloud instances — one primary, one for mirroring. It had been running for years. Nobody fully understood how it worked anymore.
The company held a P2P lending license and was under pressure from OJK (Indonesia's financial regulator) and AFPI to bring the system up to compliance standards. The director gave me the brief and one key piece of information: most of the core operations had been running manually.
Within months, the director stopped coming in. I later learned the company was facing serious financial difficulties. By that point, I was the only person in the office — me and one admin.
I kept building.

First Decision: Don't Rewrite Everything at Once
The legacy codebase was beyond patching. A full rewrite was necessary. But rewriting an entire fintech platform alone is not a sprint — it's a months-long commitment with regulatory deadlines attached.
My decision: rewrite incrementally. Start with authentication only. Prove the new stack works. Then expand.
New stack: NestJS on the backend, Next.js for the web frontend, Expo for mobile, Ubuntu replacing CentOS, Docker for containerization.
The auth system was completed. Clean, working, deployable.
Then I got the feedback from OJK: a partial system is not acceptable for audit. The entire platform needed to be functional — not just authentication.
I had made a technically sound decision with a faulty assumption. I had not validated the audit scope with the regulator before committing to the incremental approach.
That was the first real lesson.

What I Built, Alone
Over the following months, working without a team, without guidance, and with a paycheck that arrived increasingly late, I built:

Two-factor authentication (password + OTP via email and phone)
KYC flow with face photo upload
DigiSign integration for digital signatures
Full migration from monolith to microservices
AFPI Data Center integration for regulatory reporting
Loan application flow with approval stages, interest calculation, tenor management, and insurance integration
BNI sandbox integration, tested through to live transaction
Mobile app rebuilt in Expo
CI/CD pipeline with Jenkins and Docker on Ubuntu

All of it, in parallel, without a senior engineer to ask.

The Audit Day
When the OJK audit session came, the system had most of its core features working. But during the live demo, there was a bug. The kind that only appears under pressure, in front of the people you least want to see it.
The audit did not proceed.
I am not going to reframe this as a hidden success. It was a hard moment. The system I had poured months into — alone, under financial stress, in an increasingly unstable company — did not make it through the finish line that day.
But here is what I know: the system ran. It processed real transactions. It integrated with government data centers. It had security layers that the original system never had.
It just wasn't enough, on that day.

What I Actually Learned
Validate assumptions with your stakeholders before committing to an architecture decision.

Incremental migration was the right technical call. But I assumed OJK would accept a partial system without confirming it first. That assumption cost months of work. Now, before I make any significant architectural decision, I ask the hard question upfront: what does "done" mean to the person who will evaluate this?
One database per service is not optional in microservices.

Early in the build, I made a mistake I only recognized later: I created a dedicated Express service for the service bus, but gave each service its own MySQL instance rather than its own database on a shared instance. The RAM cost was immediate and painful. I fixed it — but I learned the principle properly only by feeling the consequence.
Working without a team teaches you things a team never would.

When there is no one else, every architectural decision is yours. Every bug is yours. Every late-night debugging session is yours. It is uncomfortable in ways that are difficult to describe. But the outcome is a developer who can read an unfamiliar codebase, make decisions without guidance, and stay functional when the environment is unstable.

The platform never went live. The company did not survive its challenges.
But every system I have built since — at Telkomsel, at Astra Group — carries the decision-making instincts I developed in that office, alone, with a legacy codebase and a deadline I could not fully control.
That is what the hardest projects give you. Not a success story. A foundation.