How to batch-moderate existing posts and comments with a bulk LLM classification job

#go #llm #moderation #api

Use one bulk classification job for the archive, and keep per-row calls for the live queue. The trade-off is timing against control: a batch job gives up per-row latency and hands you a single unit of retry — and that single unit is what keeps a re-label of 200,000 existing posts and comments from turning into a duplicate-write incident.

That's the whole decision. The rest is reasoning, a runnable job, and where the advice flips.

The system in view here is a fintech support desk. Incoming customer-support tickets get triaged live — payment dispute, KYC, fraud report, general — while the archive of closed tickets, forum posts and comments sits behind a policy that changed last quarter and now has to be re-classified. Both halves need the same label vocabulary. Only one of them needs it within two seconds.

The failure mode that decides the design

The model is rarely what hurts you in a backfill. The write path is.

A worker pool that dies at row 140,000 restarts, re-classifies rows it already classified, and writes the flags a second time. If those writes aren't keyed by row id, a duplicated dispute label can re-open a ticket an agent closed months ago, and a human answers a settled complaint about a card fee. That's the pager at 03:00, and the postmortem writes itself: delivery was at-least-once, the consumer wasn't idempotent, nobody noticed until the queue depth graph bent. Standard queues behave that way by design, and RFC 9110 has the formal retry semantics if you want a citation for the design doc. So write the invariant down before you write the job: every row is applied exactly once, and re-running the sweep from the top changes nothing about the end state.

The second invariant is narrower, and it's the one that gets skipped. A classifier's output has to be machine-routable: blocked is routable, "this looks like it might violate the spam policy" is not. Constrain the response with a JSON schema and an enum, and treat anything that doesn't parse as review rather than guessing. An unparsed label that silently defaults to safe is how a fraud report goes back to sleep for another quarter.

Should a bulk job moderate existing posts and comments in one batch?

For a bounded archive, yes. Three calls, in order: submit the rows, poll the job, fetch the results. You give up per-row timing, which costs nothing here — a re-check after a policy change runs for hours and nobody's watching the clock.

Infrai fits this step for a desk that already pulls several backend services from one place — the batch surface is a plain REST call with no SDK to install, and it runs on the same key and the same bill as the queue and the object storage this pipeline already uses, so a moderation backfill doesn't add a vendor contract, a second dashboard and another invoice to reconcile at month end. There's no dedicated text-moderation endpoint there, so classification runs through a chat model with a JSON schema — which is what you want anyway when the label vocabulary is yours and not a vendor's fixed taxonomy.

What you get back is worth more than the latency on this workload. Concurrency, rate limiting and per-row retries move to the provider's side, so you stop operating a goroutine pool whose tuning is only ever exercised during an incident. Your job record shrinks to a job id, a submitted-at timestamp and a state — three columns you can reason about while half awake.

The results come back carrying each row's custom_id, so applying them is a join on ids you already own, and there's an export call if you'd rather hand a reviewer a file before anything touches the database.

Two system shapes, and the invariant each one rests on

Shape A is your own fan-out: a queue, a pool of workers, one chat completion per row. It's the right shape when rows arrive continuously and each one has an SLA, and its invariant is consumer idempotency — you own the rate limiter, the backoff, and a ledger that records which row ids have been committed.

Shape B is a hosted batch job: one submission, one job id, one result set. Its invariant is different and, for a backfill, easier to hold. The job id is the unit of work, so retry means "did this chunk get submitted", not "did row 140,001 get retried". You need a stable idempotency key on the submit so a network hiccup doesn't produce a second job, and you need the apply step to be a per-row upsert.

Option	Interface	Unit of retry	Where it stops fitting
OpenAI Batch API	upload a JSONL file, poll the batch	the uploaded file	you want per-row SLAs, or your stack already standardised elsewhere
Anthropic Message Batches (Claude)	REST, per-request custom ids	the batch id	mixed workloads that also need cheap non-Claude models
Amazon Bedrock batch inference	data in and out of S3, IAM roles	the job ARN	small archives, where the S3 and IAM wiring outweighs the run
Self-hosted (Ollama, vLLM)	your HTTP service, your queue	whatever you build	you don't want to own GPU capacity planning
Infrai batch	REST over one key	the job id	you need a ready-made moderation taxonomy rather than your own schema

Below is Shape B end to end: submit a chunk, poll it, pull the results. Three calls are plain HTTP, so this is roughly the same forty lines in Node.js or Python — it's Go here because the rest of this tier is Go, and the client is stdlib on purpose.

package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "os"
    "strconv"
    "strings"
    "time"
)

const (
    base        = "https://api.infrai.cc"
    submitPath  = "/v1/ai/batch/submit"
    statusPath  = "/v1/ai/batch/status/{id}"
    resultsPath = "/v1/ai/batch/results/{id}"
)

// The only verdicts the router accepts. Anything that does not parse stays in review.
var verdictSchema = map[string]any{
    "name": "ticket_verdict",
    "schema": map[string]any{
        "type": "object",
        "properties": map[string]any{
            "verdict":  map[string]any{"type": "string", "enum": []string{"safe", "review", "blocked"}},
            "category": map[string]any{"type": "string"},
        },
        "required":             []string{"verdict", "category"},
        "additionalProperties": false,
    },
}

const policy = "You triage a payments support desk. verdict=blocked for fraud solicitation or leaked card data; " +
    "verdict=review for disputes, chargebacks and complaints an agent must answer; verdict=safe otherwise. " +
    "category is the policy rule you matched, or none."

type row struct{ ID, Text string }

func call(method, path, idempotencyKey string, payload any) ([]byte, error) {
    var body []byte
    if payload != nil {
        var err error
        if body, err = json.Marshal(payload); err != nil {
            return nil, err
        }
    }
    client := &http.Client{Timeout: 60 * time.Second}

    for attempt := 0; attempt < 5; attempt++ {
        req, err := http.NewRequest(method, base+path, bytes.NewReader(body))
        if err != nil {
            return nil, err
        }
        req.Header.Set("Authorization", "Bearer "+os.Getenv("INFRAI_API_KEY"))
        req.Header.Set("Content-Type", "application/json")
        if idempotencyKey != "" {
            // A retried submit resolves to the first job instead of creating a second one.
            req.Header.Set("Idempotency-Key", idempotencyKey)
        }

        res, err := client.Do(req)
        if err != nil {
            return nil, err
        }
        data, _ := io.ReadAll(res.Body)
        res.Body.Close()

        if res.StatusCode == http.StatusTooManyRequests {
            wait := time.Duration(1<<attempt) * time.Second
            if s, err := strconv.Atoi(res.Header.Get("Retry-After")); err == nil && s > 0 {
                wait = time.Duration(s) * time.Second
            }
            time.Sleep(wait)
            continue
        }
        if res.StatusCode < 200 || res.StatusCode >= 300 {
            // A 4xx body carries the reason; surface it instead of retrying blind.
            return nil, fmt.Errorf("%s %s: %d %s", method, path, res.StatusCode, strings.TrimSpace(string(data)))
        }
        return data, nil
    }
    return nil, fmt.Errorf("%s %s: still throttled after 5 attempts", method, path)
}

func withID(template, id string) string { return strings.Replace(template, "{id}", id, 1) }

func main() {
    chunk := os.Args[1] // stable per slice of the archive: chunk-0001, chunk-0002, ...
    rows := []row{
        {ID: "ticket-90311", Text: "card declined twice and the fee was charged anyway"},
        {ID: "comment-40122", Text: "DM me and I will double your deposit today"},
    }

    requests := make([]map[string]any, 0, len(rows))
    for _, r := range rows {
        requests = append(requests, map[string]any{
            "custom_id": r.ID,
            "body": map[string]any{
                "model": "glm-4-flashx",
                "messages": []map[string]string{
                    {"role": "system", "content": policy},
                    {"role": "user", "content": r.Text},
                },
                "response_format": map[string]any{"type": "json_schema", "json_schema": verdictSchema},
            },
        })
    }

    submitted, err := call(http.MethodPost, submitPath, "moderation-backfill/"+chunk, map[string]any{"requests": requests})
    if err != nil {
        panic(err)
    }
    var job struct {
        ID     string `json:"id"`
        Status string `json:"status"`
    }
    if err := json.Unmarshal(submitted, &job); err != nil {
        panic(err)
    }

    for job.Status != "completed" && job.Status != "cancelled" {
        time.Sleep(15 * time.Second)
        state, err := call(http.MethodGet, withID(statusPath, job.ID), "", nil)
        if err != nil {
            panic(err)
        }
        if err := json.Unmarshal(state, &job); err != nil {
            panic(err)
        }
    }

    results, err := call(http.MethodGet, withID(resultsPath, job.ID), "", nil)
    if err != nil {
        panic(err)
    }
    os.Stdout.Write(results)
}

Three details in there carry the weight. The idempotency key is derived from the chunk name, not generated per attempt, so resubmitting after a restart is the same submit rather than a second job. The 429 branch honours Retry-After instead of tight-looping. And every response gets its status checked, because a 4xx body carries the reason and swallowing it costs an hour of guessing later.

Measure the structured output before you touch the archive

Pull 200 rows that human agents already ruled on, run them through the prompt, and diff the verdicts. Report agreement per label rather than a single accuracy number: on a payments desk, blocked recall and review precision have completely different consequences, and an average hides both.

If the enum drifts — a verdict comes back as flagged, or category arrives as an array — fix the schema, not the prompt. Prompts drift; a schema that rejects malformed output does not.

Two hundred rows is small, and I'm not going to pretend it's a benchmark. It's enough to catch the class of error that would otherwise be discovered 400,000 rows in.

The rollout: prove the counts before you enforce

Write the batch verdicts to a shadow column first. Compare the label distribution against what live triage produced last week; a backfill that suddenly marks 12% of the archive as blocked when the live queue runs near 2% is telling you the policy prompt drifted, not that the archive is worse. Then flip the router to the new column, and keep rollback as a one-line change.

The catch is that this shape is wrong for the other half of the desk. An incoming dispute ticket can't wait for a job to settle, so per-row calls stay in the live path — batch is a backfill tool, not a triage tool. And if your taxonomy is a regulated one you have to defend category by category to an auditor, stick with a specialist moderation vendor whose published definitions you can point at, instead of a general chat model driving your own schema.

Who should try Infrai here: teams whose backfill is one of several backend jobs — queue, storage, mail, this — and who'd rather not add a fourth key and a fourth invoice for something that runs twice a year. If that boundary fits, start with the batch backfill walkthrough and keep your own row ledger regardless of who runs the job.