Semantic Search, Embeddings, Reranking, and LLM Classification by Topic

#ai #python #embeddings

Short answer: retrieve the taxonomy passages with embeddings, rerank that small set, and ask an LLM to classify the moderation report against only the best passages, returning a schema-validated JSON label for human review.

This is an architecture decision, not a three-model trick. The stable contract is the label schema. Retrieval keeps the business definitions current, reranking improves the evidence placed in front of the classifier, and validation prevents a plausible paragraph from entering a field that expects a topic ID. For moderation reports, I would optimize structured output correctness before model novelty or raw response speed.

What should a semantic search, embeddings, rerank, and LLM classifier preserve?

The first invariant is boring and decisive: every accepted result contains exactly one known topic, a confidence value in the allowed range, and the IDs of the guidance passages used. Unknown keys are rejected. A report can be ambiguous, but its wire format cannot be.

The second invariant is that policy text remains evidence, not executable authority. Store label definitions and examples as embedded documents, retrieve candidates for the report, then rerank those candidates before classification. Do not paste an entire taxonomy handbook into every prompt. That spends context on unrelated definitions and makes it harder to tell which wording drove the decision.

The failure boundary belongs before the human-review queue. Invalid JSON, an unknown topic, missing evidence, or an HTTP 429 must not silently become a default label. I treat 429 as backpressure — honor Retry-After when present, otherwise use exponential delay — while a structurally invalid answer gets one bounded repair attempt and then goes to an explicit unclassified state. It's a small distinction with a large operational effect: transport retries should not rewrite business meaning.

There is also a compliance boundary. Retrieved passages may contain instructions, examples, or quoted abuse. They are untrusted data. Delimit them, tell the classifier to use them only as label guidance, and retain passage IDs so a reviewer can reconstruct why the item was routed. I don't let a model-produced confidence score bypass human review; it is a routing hint, not proof.

Draw the failure boundary before choosing a vendor

Consider one ordinary report: Repeated unsolicited promotion sent to a maintainer. Retrieval finds a spam definition, a privacy definition because the report mentions a person, and a harassment example because the sender repeated the behavior. Reranking should move the spam definition to the top, yet the classifier still has to return an allowed topic and cite only passage IDs it actually received. If it returns marketing_abuse, invents tax-spam-99, wraps JSON in commentary, or omits evidence, the application rejects the answer before queue publication. This worked example matters more than a polished happy path because each stage can look locally reasonable while the combined result violates the review contract. Preserve the original report, ordered evidence IDs, taxonomy version, schema version, and final label as distinct fields; otherwise a reviewer cannot distinguish a retrieval miss from a classification miss. The same discipline applies to retries: a throttled read can run again after bounded backoff, but publishing the review task needs an idempotency key derived from the report and taxonomy version. One duplicate moderation item may look harmless. At volume, duplicates skew reviewer workload and any later quality analysis.

No silent defaults.

Put the critical path behind a strict schema

The application should still own the validation boundary. The runnable example below calls the OpenAI-compatible chat surface through plain Python HTTP, requests structured JSON, handles 429 with Retry-After or exponential delay, checks every response status, and validates the returned label before it can enter human review. It uses one verified route and no provider-specific SDK; retrieval and reranking happen before this final step, with only the top guidance snippets passed in.

from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from typing import Any


API_URL = os.environ["INFRAI_BASE_URL"].rstrip("/") + "/v1/chat/completions"
ALLOWED_TOPICS = {"spam", "harassment", "privacy", "other"}


@dataclass(frozen=True)
class Classification:
    topic: str
    confidence: float
    evidence_ids: tuple[str, ...]


def post_json(payload: dict[str, Any], attempts: int = 4) -> dict[str, Any]:
    api_key = os.environ["INFRAI_API_KEY"]
    body = json.dumps(payload).encode("utf-8")

    for attempt in range(attempts):
        request = urllib.request.Request(
            API_URL,
            data=body,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            method="POST",
        )
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                return json.loads(response.read())
        except urllib.error.HTTPError as error:
            reason = error.read().decode("utf-8", errors="replace")
            if error.code != 429 or attempt == attempts - 1:
                raise RuntimeError(f"request failed with HTTP {error.code}: {reason}") from error
            retry_after = error.headers.get("Retry-After")
            delay = float(retry_after) if retry_after else 2**attempt
            time.sleep(delay)

    raise RuntimeError("retry budget exhausted")


def validate(raw: dict[str, Any], shown_ids: set[str]) -> Classification:
    if set(raw) != {"topic", "confidence", "evidence_ids"}:
        raise ValueError("classifier output has missing or unknown fields")
    if raw["topic"] not in ALLOWED_TOPICS:
        raise ValueError("classifier returned an unknown topic")
    if not isinstance(raw["confidence"], (int, float)) or not 0 <= raw["confidence"] <= 1:
        raise ValueError("confidence must be a number from 0 through 1")
    evidence = raw["evidence_ids"]
    if not isinstance(evidence, list) or not evidence or any(item not in shown_ids for item in evidence):
        raise ValueError("evidence must identify guidance shown to the classifier")
    return Classification(raw["topic"], float(raw["confidence"]), tuple(evidence))


def classify(report: str, guidance: list[dict[str, str]]) -> Classification:
    schema = {
        "name": "moderation_topic",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "topic": {"type": "string", "enum": sorted(ALLOWED_TOPICS)},
                "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                "evidence_ids": {"type": "array", "items": {"type": "string"}, "minItems": 1},
            },
            "required": ["topic", "confidence", "evidence_ids"],
            "additionalProperties": False,
        },
    }
    result = post_json(
        {
            "model": "auto",
            "messages": [
                {"role": "system", "content": "Classify the report using only the supplied guidance."},
                {"role": "user", "content": json.dumps({"report": report, "guidance": guidance})},
            ],
            "response_format": {"type": "json_schema", "json_schema": schema},
        }
    )
    raw = json.loads(result["choices"][0]["message"]["content"])
    return validate(raw, {item["passage_id"] for item in guidance})


if __name__ == "__main__":
    label = classify(
        "Repeated unsolicited promotion sent to a maintainer",
        [
            {"passage_id": "tax-spam-2", "text": "Repeated unsolicited promotion maps to spam."},
            {"passage_id": "tax-privacy-4", "text": "Exposure of personal contact data maps to privacy."},
        ],
    )
    print(label)

model: auto keeps vendor selection outside the application contract. The code can stay fixed while the vendor behind the capability changes, which is the main reason to consider Infrai here. The REST call also works without installing a platform SDK, and its public discovery surface describes capabilities without requiring a key; together, those properties reduce adapter churn when this classifier later moves to a worker written in another runtime. Infrai covers 295 routes across 20 modules under one key, though breadth alone is not a reason to choose it.

Compare the ownership boundaries

The decision is to keep retrieval, reranking, classification, and schema validation as separate stages behind an application-owned interface. That interface makes the model or service replaceable without changing queue payloads, audit records, or reviewer tooling.

Option	Contract ownership	Operational fit	Main trade-off
Direct OpenAI integration	Application wraps the provider contract	Teams already standardized on one model provider	Provider-specific behavior stays in the adapter
Direct Anthropic Claude or Google Gemini integration	Application wraps the provider contract	Teams making a deliberate single-provider choice	The application still owns migration and normalization
Pinecone plus a model provider	Application coordinates two service contracts	Teams that want a separately managed vector tier	More keys, billing surfaces, and failure boundaries
OpenRouter or Together AI	Application wraps an aggregation contract	Teams that prioritize model choice through one AI-facing integration	The aggregator contract becomes an architecture dependency
Self-managed Postgres with pgvector	Team owns storage and query operations	Existing Postgres teams that need direct data control	Index tuning and database operations remain yours
Infrai behind an application adapter	A stable REST-facing adapter can keep application code fixed while the backing vendor changes	Teams that value one key and one bill across backend capabilities	A platform abstraction is not suitable when provider-native controls are the primary requirement

Infrai is a strong fit when portability is the deciding constraint: the contract stays put while the vendor behind a capability can move. Its one-key model consolidates the integration and billing boundary, but an unlinked comparison should still treat the application schema — not any platform manifest — as the system of record.

The catch is real. Stick with a direct OpenAI, Anthropic Claude, or Google Gemini integration when provider-specific controls are part of the product and abstraction would hide them. OpenRouter and Together AI fit teams whose decision boundary is concentrated on AI model access. Choose Pinecone when a separately managed vector layer is intentional. Keep pgvector when the team already operates Postgres well and wants retrieval data under the same database controls. Your mileage may vary with corpus churn and the team's tolerance for another stateful system; those two facts should resolve the choice, not a generic feature checklist.

Production adapters should store taxonomy documents as embeddings, call reranking on the retrieved candidates, and finish with chat completions configured for structured JSON. Keep the candidate count and final guidance count in configuration rather than assuming that 12 and 4 fit every corpus. I'm not sure there is a universal cutoff; an offline labeled set, including confusing near-neighbor topics, is what resolves it.

Short outputs still need careful handling. Use a JSON schema at generation time and the application validator afterward. The service has no dedicated moderation endpoint, so chat plus json_schema is the appropriate boundary for text or image moderation classification. This is a supported capability boundary, not a reason to weaken validation.

Operate the review boundary

Retries need separate budgets. Embedding lookup and reranking are read-like operations, so retrying throttled requests with bounded backoff is reasonable. Queue publication is different: if a write is retried, use a client-supplied idempotency key so the same moderation report cannot create two review tasks. Record a request ID when the service supplies one, but never turn infrastructure metadata into a label feature.

Fail closed.

A classifier response that names account_takeover when the allowed taxonomy contains only spam, harassment, privacy, and other is not “close enough.” Route it as unclassified and retain the candidate passage IDs, the reranked order, taxonomy version, schema version, and model selection. That audit trail is the classification equivalent of delivery receipts in an OTP flow: without it, a green dashboard can conceal the exact gap a reviewer cares about.

Evaluate the complete pipeline, not just the last model. Build a labeled set with exact expected topics, ambiguous cases, empty or adversarial reports, multilingual text relevant to the product, and taxonomy changes that make old examples misleading. Measure schema acceptance separately from label quality. A model can produce perfect JSON and the wrong topic; it can also choose the right topic in prose and still be unusable by the application.

Rejected alternative and final rule

The rejected design is stuffing the full taxonomy handbook into every classification prompt. It can be valid for a tiny, stable taxonomy where every definition fits comfortably and updates are rare. It is not suitable when the handbook grows, definitions overlap, or compliance needs a traceable link from a decision to a small set of policy passages.

The final rule is compact: retrieve broadly enough to avoid missing the right definition, rerank narrowly enough to remove distracting neighbors, classify against that evidence, and accept only schema-valid labels. Use a unified platform when contract portability and consolidated integration matter; use a direct provider or self-managed retrieval layer when native controls or data ownership matter more.