Logistics Applicant Evidence Routing in Node.js: Typed JSON for Text/Image Safety Checks

Short answer: for a logistics application that must moderate text and image submissions through an OpenAI-compatible API, make structured output a typed application contract: request a small JSON Schema verdict from chat completions, validate it again locally, and send uncertain or failed checks to review rather than publishing automatically.

This is a better fit for a candidate-scoring workflow than a bare true/false flag. A driver application might contain a resume, a photo of a license, and a short note about route experience. The safety gate decides whether that material is safe to pass to the scoring pipeline. It does not decide whether the person gets an interview.

Keep those decisions separate. The distinction makes evals easier to interpret and gives operations somewhere to put ambiguous cases.

What should a Node.js content moderation flow do with chat completions, JSON Schema, text, and image safety?

The flow has five steps: discover an available model, build a text-only or multimodal user message, request a closed JSON Schema, validate the returned object at the service boundary, and map the result to an operational action. In Node.js, the HTTP payload is the important part; the same contract can be sent from any language that can make HTTPS requests.

For this scenario I use three decisions: allow lets a candidate record move to the next internal stage, review puts it in a queue, and block stops the record from entering downstream workflows. Categories are evidence for a reviewer, not an instruction to the model to take a hiring action. A model should not infer employment policy from a safety prompt.

The image path deserves its own test cases. A license photo can be blurry, an image URL can require authentication, and a text description can disagree with what is visible. Treat an unavailable image as missing evidence and choose review; do not silently turn a failed fetch into allow.

Images fail quietly.

The following example uses Python because the article's code convention is Python, but the request shape is the same one a Node.js service would send. It uses the standard library, keeps the base URL in configuration, and calls only the two protocol paths involved in this flow: GET /v1/models for discovery and POST /v1/chat/completions for classification.

A small, closed contract before the scoring model

The schema below is deliberately narrower than the policy document. It has enough information to route a record and explain a queue item, while leaving the final product policy in application code. additionalProperties: false matters because silently accepting a new field can hide a prompt or model change.

import json
import os
import sys
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


DECISIONS = {"allow", "review", "block"}
CATEGORIES = {
    "identity_document",
    "sexual_content",
    "violence",
    "harassment",
    "self_harm",
    "spam",
}

SCHEMA = {
    "name": "candidate_content_verdict",
    "strict": True,
    "schema": {
        "type": "object",
        "additionalProperties": False,
        "required": ["decision", "categories", "rationale"],
        "properties": {
            "decision": {
                "type": "string",
                "enum": ["allow", "review", "block"],
            },
            "categories": {
                "type": "array",
                "items": {
                    "type": "string",
                    "enum": sorted(CATEGORIES),
                },
                "uniqueItems": True,
            },
            "rationale": {"type": "string"},
        },
    },
}


def request_json(url: str, api_key: str, payload: dict[str, Any]) -> dict[str, Any]:
    request = Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urlopen(request, timeout=20) as response:
        return json.load(response)


def available_models(api_base: str, api_key: str) -> set[str]:
    request = Request(
        f"{api_base.rstrip('/')}/v1/models",
        headers={"Authorization": f"Bearer {api_key}"},
        method="GET",
    )
    with urlopen(request, timeout=10) as response:
        document = json.load(response)
    return {item["id"] for item in document.get("data", [])}


def validate_verdict(value: Any) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ValueError("verdict is not an object")
    if value.get("decision") not in DECISIONS:
        raise ValueError("decision is outside the policy contract")
    categories = value.get("categories")
    if not isinstance(categories, list) or not set(categories) <= CATEGORIES:
        raise ValueError("categories are outside the policy contract")
    if not isinstance(value.get("rationale"), str):
        raise ValueError("rationale is not a string")
    return value


def moderate_candidate(text: str, image_url: str | None = None) -> dict[str, Any]:
    api_base = os.environ["CHAT_API_BASE"]
    api_key = os.environ["CHAT_API_KEY"]
    model = os.environ["CHAT_MODEL"]
    if model not in available_models(api_base, api_key):
        raise ValueError("configured model is not in the current model list")

    content: str | list[dict[str, Any]] = text
    if image_url:
        content = [
            {"type": "text", "text": text},
            {"type": "image_url", "image_url": {"url": image_url}},
        ]

    payload = {
        "model": model,
        "temperature": 0,
        "messages": [
            {
                "role": "system",
                "content": (
                    "Classify submitted candidate material for publication safety. "
                    "Use only the allowed decisions and categories. "
                    "Choose review when the evidence is ambiguous."
                ),
            },
            {"role": "user", "content": content},
        ],
        "response_format": {"type": "json_schema", "json_schema": SCHEMA},
    }
    response = request_json(
        f"{api_base.rstrip('/')}/v1/chat/completions", api_key, payload
    )
    raw = response["choices"][0]["message"].get("content")
    if not raw:
        raise ValueError("response did not contain a verdict")
    return validate_verdict(json.loads(raw))


if __name__ == "__main__":
    text = sys.argv[1] if len(sys.argv) > 1 else "Candidate application note"
    image = sys.argv[2] if len(sys.argv) > 2 else None
    try:
        print(json.dumps(moderate_candidate(text, image), indent=2))
    except (HTTPError, URLError, TimeoutError, ValueError) as error:
        print(json.dumps({"decision": "review", "error": str(error)}))

The exception path is intentionally conservative. It is a routing policy, not a claim that a network error says anything about the candidate. I treat HTTP 429 as a retry signal with bounded backoff and server guidance such as Retry-After; after the retry budget is exhausted, the record still goes to review. The same rule applies to a missing model, malformed JSON, and an image that cannot be read.

One detail is easy to miss in a port to Node.js: model discovery and chat creation are different methods. A model list uses GET /v1/models; a completion uses POST /v1/chat/completions. The route names are part of the compatible API contract. Do not invent a REST-shaped moderation endpoint just because it sounds tidy.

How do you evaluate structured output correctness for candidate text and images?

Schema validation answers “does this object have the right shape?” It does not answer “was the policy applied correctly?” Those are separate test layers, and a moderation service needs both.

Start with a labeled set from the logistics workflow. Include ordinary route notes, documents with sensitive words used in a benign sentence, explicit harassment, ambiguous jokes, image-only submissions, text-image pairs, and inaccessible image URLs. Add candidate material that looks like a safety violation but is actually a quotation or a description of a past incident. The point is to test context, not to reward keyword matching.

For each case, store the expected decision and permitted categories before running the model. Then compare false allows with false blocks. A single accuracy number hides the operational choice: a false allow may expose people to harmful material, while a false block can delay a legitimate candidate and create a queue burden. In this application, structured output correctness includes valid enums, no unknown fields, stable category names, and a decision that agrees with the labeled policy. For example, take a candidate note that quotes an abusive message received during a delivery dispute and includes a photograph of a damaged loading bay. A text-only check may see harassment language, while the image adds no safety signal at all. The expected result might be review, because the record needs human context, not because either input is automatically disallowed. Test the note alone, the image alone, both together, and a broken image reference; those four cases reveal different bugs in input assembly, modality support, and fallback routing.

I version the system instruction, schema, model ID, and eval set together. If one changes, rerun the set. I'm not sure a new model is safer merely because it produces valid JSON; validity is a mechanical property, while policy quality needs labeled evidence.

Token use belongs in the same notebook as the eval results, but it should not choose the safety threshold. First find configurations that meet the policy target. Then compare latency and token use among those configurations. This keeps cost pressure from quietly redefining what “safe enough” means.

Failure handling is part of the safety boundary

A queue is not an embarrassment. It is where the system puts uncertainty that a model cannot responsibly resolve.

Use an internal record containing the input reference, schema version, prompt version, model ID, decision, categories, request identifier, and failure reason. Keep image URLs and candidate data under the retention rules for the application; logging the entire payload to debug one malformed response can create a second privacy problem.

Test the negative paths explicitly: empty text, an expired image URL, a response with an extra property, invalid JSON, an authentication rejection, a timeout, and HTTP 429. Every one should produce a controlled review outcome or a bounded retry followed by review. No exception handler should fall through to the code that publishes or scores a candidate.

There is a useful boundary here. The moderation result can gate the scoring pipeline, but it should not be mixed into the scoring prompt as an unverified instruction. Pass a validated object to the next stage, and make the scorer consume only fields that the application has accepted. That prevents a rationale string from becoming an accidental command.

A practical ship decision for this architecture

This pattern fits a team that needs one HTTP contract across Node.js services, Python eval notebooks, and self-hosted or managed backends. It is especially useful when a dedicated moderation endpoint is unavailable and the team is willing to own the policy set, queue, and validation layer. An OpenAI-compatible API can reduce client-library coupling, but compatibility of the wire shape does not prove equal model behavior.

The catch is that chat-based classification is not suitable when your organization requires a separately calibrated moderation system, a certified decision process, or a provider-specific data boundary that this architecture cannot satisfy. Stick with a dedicated service when its independently measured policy performance is a requirement. Keep a direct integration when governance, retention, or existing eval evidence depends on it. The local verdict type still helps in those cases because it keeps downstream application code independent from the response format.

Here is the decision table I would put beside the design document:

No row wins by default. The right row is the one whose evidence, ownership, and failure behavior match the product's risk tolerance.

Before enabling enforcement, run the full labeled set against the exact model and schema you will deploy, verify text-only and multimodal cases, inspect review volume, and exercise each failure path. Confirm the runtime can discover the selected model in the target environment. Then make the operational checklist prose: pin the contract, validate twice, retry only transient transport failures, retain enough evidence for an internal audit, and send uncertainty to review. That is the release decision.

References

• https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events
• https://python.langchain.com/docs/integrations/chat/openai/