LLMOps: What Changes When You Deploy Generative AI Instead of Traditional ML

A production engineering guide using a social-media Comment Assistant

LLMOps extends MLOps rather than replacing it. Versioning, deployment automation, observability, security and incident response still apply; generative systems add prompt and tool versioning, retrieval quality, variable token economics, open-ended evaluation and new safety risks. A model can return HTTP 200 and fluent text that is still inaccurate, unsafe, off-brand or commercially harmful.

Running example: a Comment Assistant that helps creators draft replies, summarise audience sentiment and suggest post ideas. It must protect creators and commenters, follow platform rules, keep spend bounded, and leave every publishing decision with the creator.

1. Version the Whole Configuration

In an LLM application the deployed artifact is a configuration bundle: model snapshot, system prompt, generation parameters, output schema, tools, retrieval settings and safety policy. Changing any one of them can change quality, safety, latency and cost, so they are released, fingerprinted and rolled back together.

import hashlib
import json
from dataclasses import asdict, dataclass
from typing import Any

@dataclass(frozen=True)
class RuntimeConfig:
    """Everything that can change quality, safety, latency or cost."""
    provider: str
    model: str                  # pinned snapshot ID, never a moving alias
    prompt_version: str
    system_prompt: str
    temperature: float
    max_output_tokens: int
    output_schema: dict[str, Any]  # the schema itself, not just its name
    safety_policy_version: str
    retrieval_index_version: str
    retrieval_top_k: int
    tools: tuple[str, ...] = ()    # canonical JSON tool definitions

    def fingerprint(self) -> str:  # content address: same config, same hash
        payload = json.dumps(asdict(self), sort_keys=True,
                             separators=(",", ":"), ensure_ascii=False)
        return "sha256:" + hashlib.sha256(payload.encode()).hexdigest()

@dataclass(frozen=True)
class Release:
    release_id: str  # human-readable, e.g. "comment-assistant-2026.10.1"
    config: RuntimeConfig

REPLY_SCHEMA = {"type": "object", "properties": {"reply": {"type": "string"}},
                "required": ["reply"], "additionalProperties": False}

The fingerprint is a content address: it hashes the configuration only, so two releases with identical settings share a hash and any real change produces a new one. Keep the human-readable release ID outside the hash. Hash the output schema and tool definitions themselves, not just their version labels, which can be edited without being bumped.

Provider aliases can change behaviour silently. Pin a dated model snapshot where the provider offers one, and record the model ID the provider actually returns on every request.

Store each approved release as an immutable artifact. Promotion requires evaluation, security review, canary results and a known-good rollback target; ordinary users never edit the production system prompt. The fingerprint is an integrity check, not a replacement for access control or signed releases.

2. Evaluating Open-Ended Outputs

A single reference answer is rarely enough. A reply must address the comment, match the creator's voice, be factually supported, comply with policy, avoid sensitive disclosures and fit the schema. Evaluate in layers on a versioned golden set that includes ordinary, adversarial, multilingual, ambiguous, spam and high-risk examples:

Deterministic: schema, length, forbidden actions, URL allowlists, PII detectors and policy rules. Model-assisted: rubric scores for relevance, tone, groundedness and completeness from judges calibrated against human labels. Human: stratified samples, high-risk queues, new languages and judge disagreements. Online: acceptance, edits, regenerations, dismissals, reports and opt-outs, by language and feature.

Word overlap is not relevance (a good paraphrase can share no words), keyword presence is not brand fit, and a forbidden-phrase list is a smoke test, not proof of safety. User feedback is an outcome signal, not a correctness label.

Put product rules in the schema so they run wherever the model is parsed, and keep hard failures separate from soft scores in the release gate:

import statistics
import unicodedata
from collections.abc import Sequence
from dataclasses import dataclass
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

MAX_WORDS = 40

class ReplyDraft(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
    reply: str = Field(min_length=1, max_length=500)
    needs_human_review: bool = False
    review_reason: str | None = Field(default=None, max_length=200)

    @field_validator("reply")
    @classmethod
    def product_rules(cls, v: str) -> str:
        if len(v.split()) > MAX_WORDS:
            raise ValueError(f"reply exceeds {MAX_WORDS} words")
        if any(unicodedata.category(ch) == "Cc" and ch != "\n" for ch in v):
            raise ValueError("control characters are not allowed")  # incl. NUL "\x00"
        return v

    @model_validator(mode="after")
    def reason_required(self) -> "ReplyDraft":
        if self.needs_human_review and not self.review_reason:
            raise ValueError("review_reason is required when review is requested")
        return self

@dataclass(frozen=True)
class EvalResult:
    case_id: str
    hard_failures: tuple[str, ...]  # any one blocks promotion
    scores: dict[str, float]        # calibrated rubric scores in [0, 1]

def release_gate(results: Sequence[EvalResult], baseline: dict[str, float],
                 max_drop: float = 0.02) -> list[str]:
    """Return blocking problems; an empty list means the release may be promoted."""
    problems = [f"{r.case_id}: {f}" for r in results for f in r.hard_failures]
    for metric, base in baseline.items():
        mean = statistics.fmean(r.scores[metric] for r in results)
        if mean < base - max_drop:
            problems.append(f"{metric} regressed: {mean:.3f} vs baseline {base:.3f}")
    return problems

Rejecting control characters by Unicode category catches the real NUL byte. A check written as "\\x00" in reply searches for the four printable characters \x00 instead: it rejects harmless text and lets the real byte through.

A hard failure blocks promotion regardless of averages. Compare scores per slice (language, risk category) as well as overall, because an aggregate can hide a regression in a small, high-risk segment.

3. Safety, Guardrails and Prompt Injection

Comments, captions, imported posts, retrieved documents and tool outputs are untrusted. A malicious comment can try to override instructions, extract secrets, trigger tools or produce harassment, and regex filters for known phrases are easily evaded with paraphrase, Unicode tricks, encoding or other languages. Filters are defence in depth; the strongest controls are structural:

Pass untrusted text as data, not instructions: JSON-encode it in its own message (section 8) so it cannot close a delimiter such as </comment> and escape. Never let the model choose tenant IDs, account IDs or permissions.

Grant least privilege: no unnecessary tools, authorisation enforced outside the model, and creator confirmation before publishing, deleting, messaging or any other external action. Never expose secrets, system prompts, internal tool schemas or other users' data.

Run input and output moderation as independent services with versioned categories and thresholds, and fail closed when they are unavailable. Never pass blocked text downstream, even inside an error object.

import asyncio
from dataclasses import dataclass
from enum import StrEnum
from typing import Protocol

class Decision(StrEnum):
    ALLOW = "allow"
    REVIEW = "review"
    BLOCK = "block"

SEVERITY = {Decision.ALLOW: 0, Decision.REVIEW: 1, Decision.BLOCK: 2}

@dataclass(frozen=True)
class SafetyResult:
    decision: Decision
    reason_code: str

class Moderator(Protocol):  # wraps a maintained moderation service
    async def classify(self, text: str) -> SafetyResult: ...

async def moderate(moderator: Moderator, text: str, timeout_s: float = 2.0) -> SafetyResult:
    try:
        async with asyncio.timeout(timeout_s):
            return await moderator.classify(text)
    except Exception:  # timeout, outage or malformed reply: fail closed
        return SafetyResult(Decision.BLOCK, "moderation_unavailable")

def combine(*results: SafetyResult, sensitive_topic: bool = False) -> SafetyResult:
    """Deterministic precedence: the most severe result wins, with its reason code."""
    worst = max(results, key=lambda r: SEVERITY[r.decision])
    if worst.decision is Decision.ALLOW and sensitive_topic:
        return SafetyResult(Decision.REVIEW, "sensitive_topic")
    return worst

Precedence is deterministic and keeps the specific reason code (for example policy_x) instead of collapsing everything into a generic label, which makes review queues and dashboards actionable.

4. Latency, Tokens and Cost

LLM latency includes queueing, time to first token, generation, retrieval and tool work, retries and application overhead. Track time to first token and end-to-end latency separately, at p50, p95 and p99, by feature, model and region. Take token counts and cost from provider usage metadata; character or word counts are not tokens.

Metrics: request rate, concurrency, queue depth, timeouts, provider 4xx/5xx, retries and circuit-breaker state; input, cached and output tokens and cost per request, tenant, feature and day; budget use and quota rejections. Keep metric labels low-cardinality: user, comment and request IDs belong in traces, not Prometheus labels.

Providers report usage differently. OpenAI counts cached tokens inside prompt_tokens; Anthropic's input_tokens excludes them. A rule like “cached input cannot exceed total input” therefore rejects valid Anthropic usage, and a three-price model cannot represent cache writes, which some providers bill at a premium. Normalise into disjoint buckets:

from dataclasses import dataclass
from datetime import date
from decimal import Decimal

@dataclass(frozen=True)
class Usage:  # disjoint buckets; each provider adapter converts into this
    uncached_input: int
    cache_write: int
    cache_read: int
    output: int  # includes reasoning tokens

def usage_from_chat_completions(u) -> Usage:
    # OpenAI-style: prompt_tokens INCLUDES cached tokens.
    details = getattr(u, "prompt_tokens_details", None)
    cached = (details.cached_tokens if details else 0) or 0
    return Usage(u.prompt_tokens - cached, 0, cached, u.completion_tokens)

def usage_from_anthropic(u) -> Usage:
    # Anthropic: input_tokens EXCLUDES cache reads and cache writes.
    return Usage(u.input_tokens, u.cache_creation_input_tokens or 0,
                 u.cache_read_input_tokens or 0, u.output_tokens)

@dataclass(frozen=True)
class Price:  # USD per million tokens, from an effective-dated registry
    effective_from: date
    input: Decimal
    cache_write: Decimal
    cache_read: Decimal
    output: Decimal

def estimate_cost(u: Usage, p: Price) -> Decimal:
    if min(u.uncached_input, u.cache_write, u.cache_read, u.output) < 0:
        raise ValueError("token counts cannot be negative")
    total = (u.uncached_input * p.input + u.cache_write * p.cache_write
             + u.cache_read * p.cache_read + u.output * p.output)
    return (total / 1_000_000).quantize(Decimal("0.00000001"))

Keep prices in an effective-dated registry and reconcile estimates with invoices. A budget named “daily” needs a real per-day, per-tenant aggregation, enforced before dispatch. Retries need bounded attempts, backoff with jitter, deadlines and idempotency.

5. Retrieval Quality

RAG may supply creator-approved voice examples, community guidelines, product FAQs, campaign details or policy references. Measure retrieval separately from generation: a fluent answer cannot repair missing, stale, unauthorised or irrelevant context. Lexical overlap ignores synonyms, negation and ranking, and a “grounded-term ratio” rewards word reuse even for unsupported claims, so use ranking metrics on a labelled set:

import math
from collections.abc import Mapping, Sequence

def recall_at_k(retrieved: Sequence[str], relevant: set[str], k: int) -> float:
    if not relevant:
        raise ValueError("each labelled query needs at least one relevant document")
    return len(set(retrieved[:k]) & relevant) / len(relevant)

def ndcg_at_k(retrieved: Sequence[str], gains: Mapping[str, float], k: int) -> float:
    ranked = list(dict.fromkeys(retrieved))[:k]  # duplicates must not earn gain twice
    dcg = sum(gains.get(d, 0.0) / math.log2(i + 2) for i, d in enumerate(ranked))
    ideal = sorted(gains.values(), reverse=True)[:k]
    idcg = sum(g / math.log2(i + 2) for i, g in enumerate(ideal))
    return dcg / idcg if idcg else 0.0

Also monitor no-result, duplicate-chunk and stale-source rates and citation validity (check each cited claim against its source span), and version the embedding model, chunker, index, reranker and corpus snapshot.

Enforce tenant and permission filters inside the retrieval query, before ranking; filtering afterwards can leak content through logs and prompts. Judge freshness from authoritative effective and expiry dates for every required source, not the age of the newest document. If a required policy source is missing or stale, abstain or route to a human rather than answer from model memory, and treat injected instructions in retrieved documents like any other untrusted input.

6. Business Impact and User Trust

Useful signals include adoption on eligible comments, accepted-without-edit rate, edit categories, regenerations, dismissals, response time, reports, opt-outs and creator satisfaction. Compute acceptance over drafts actually shown, not all comments received. Simulated data where assistant users are made faster proves nothing about causation; measure impact with randomised assignment or a defensible quasi-experiment.

Randomise at creator or account level to avoid spillover, and pre-register the primary metric, guardrails, sample size and power, duration and stopping rules. Account for creator-level clustering in the analysis.

Segment by language, creator size, content category, surface and accessibility needs, and never optimise acceptance alone: bland, unsafe or misleading drafts can still be accepted.

7. Production System Architecture

Flow: platform webhooks → webhook gateway (signature over the raw body, timestamp window, event-ID dedupe, rate limits) → durable queue with dead-letter handling → tenant authorisation and consent checks → content minimisation and PII handling → tenant-scoped retrieval → LLM call from a versioned release → input/output moderation and schema validation → draft store with audit metadata → creator review (approve, edit, reject) → publishing adapter (OAuth scopes, platform limits, idempotency) → platform API. Cross-cutting: secret manager, encryption, telemetry, budgets, release registry, evaluation pipeline and incident response.

Derive tenant and actor from verified token claims and check access to the connected account on every operation, never from headers, request bodies or model output.

Acknowledge webhooks only after durable acceptance; process asynchronously with bounded retries and idempotent handlers, since delivery is at-least-once.

Store drafts separately from published content, require explicit approval and re-check authorisation at publish time. Degrade gracefully: deadlines, circuit breakers, concurrency limits, and a provider fallback only after it has passed evaluation.

8. Reference Service (FastAPI)

The service below shows the production boundary: verified identity, account authorisation, idempotency before spending tokens, schema-enforced output, layered moderation, and persisted drafts that wait for creator approval. It never publishes. Manage dependencies with a lock file (for example uv lock or pip-compile --generate-hashes) and update them regularly; versions pinned in an article go stale quickly.

Database schema (PostgreSQL)

CREATE TABLE account_members (  -- server-side source of truth for account access
  tenant_id  UUID NOT NULL,
  account_id TEXT NOT NULL,
  actor_id   TEXT NOT NULL,
  role       TEXT NOT NULL CHECK (role IN ('owner', 'editor', 'viewer')),
  PRIMARY KEY (tenant_id, account_id, actor_id)
);

CREATE TABLE comment_drafts (
  draft_id           UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id          UUID NOT NULL,
  creator_account_id TEXT NOT NULL,
  source_comment_id  TEXT NOT NULL,
  status             TEXT NOT NULL CHECK (status IN
                       ('pending_review', 'needs_review', 'approved', 'rejected', 'published')),
  reply_text         TEXT NOT NULL,
  release_id         TEXT NOT NULL,
  config_fingerprint TEXT NOT NULL,
  provider_model     TEXT NOT NULL,  -- model ID returned by the provider
  safety_decision    TEXT NOT NULL CHECK (safety_decision IN ('allow', 'review')),
  safety_reason      TEXT NOT NULL,
  created_by         TEXT NOT NULL,
  created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
  approved_by        TEXT,
  approved_at        TIMESTAMPTZ,
  published_at       TIMESTAMPTZ,
  CHECK (status NOT IN ('approved', 'published')
         OR (approved_by IS NOT NULL AND approved_at IS NOT NULL)),
  CHECK (status <> 'published' OR published_at IS NOT NULL),
  UNIQUE (tenant_id, creator_account_id, source_comment_id, release_id)
);
CREATE INDEX comment_drafts_tenant_created ON comment_drafts (tenant_id, created_at DESC);

account_members is the server-side source of truth for who may act on which creator account. CHECK constraints make it impossible to store an approved or published draft without an approver, or a published draft without a publish time.

Application

import asyncio
import json
import logging
from contextlib import asynccontextmanager
from dataclasses import asdict, dataclass
from typing import Annotated
import asyncpg
import jwt
from fastapi import Depends, FastAPI, HTTPException, Request, Response
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from openai import APIConnectionError, APIStatusError, APITimeoutError, AsyncOpenAI
from pydantic import BaseModel, ConfigDict, Field, SecretStr, ValidationError
from pydantic_settings import BaseSettings
from app.drafts import ReplyDraft                    # section 2
from app.release import load_release                 # section 1
from app.safety import Decision, SafetyResult, build_moderator, combine, moderate  # section 3
from app.usage import usage_from_chat_completions    # section 4

log = logging.getLogger("comment_assistant")

class Settings(BaseSettings):  # injected from the secret manager; fails fast if missing
    database_url: SecretStr
    llm_base_url: str
    llm_api_key: SecretStr
    oidc_issuer: str
    oidc_audience: str
    oidc_jwks_url: str
    release_id: str

S = Settings()
RELEASE = load_release(S.release_id)  # immutable Release from the registry (section 1)
CFG = RELEASE.config
FINGERPRINT = CFG.fingerprint()
MODERATOR = build_moderator()  # adapter for your approved moderation service

MEMBER_SQL = """SELECT 1 FROM account_members
                WHERE tenant_id = $1 AND account_id = $2 AND actor_id = $3"""
FIND_SQL = """SELECT draft_id, status, reply_text FROM comment_drafts
              WHERE tenant_id = $1 AND creator_account_id = $2
                AND source_comment_id = $3 AND release_id = $4"""
INSERT_SQL = """INSERT INTO comment_drafts (tenant_id, creator_account_id, source_comment_id,
                  status, reply_text, release_id, config_fingerprint, provider_model,
                  safety_decision, safety_reason, created_by)
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
                ON CONFLICT (tenant_id, creator_account_id, source_comment_id, release_id)
                DO NOTHING RETURNING draft_id, status, reply_text"""

@dataclass(frozen=True)
class Principal:
    tenant_id: str
    actor_id: str

class GenerateRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")
    creator_account_id: str = Field(min_length=1, max_length=128)
    source_comment_id: str = Field(min_length=1, max_length=256)
    comment_text: str = Field(min_length=1, max_length=4000)

class DraftResponse(BaseModel):
    draft_id: str
    status: str
    reply: str
    release_id: str
    config_fingerprint: str

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.db = await asyncpg.create_pool(
        S.database_url.get_secret_value(), min_size=2, max_size=10, command_timeout=5)
    app.state.llm = AsyncOpenAI(base_url=S.llm_base_url, api_key=S.llm_api_key.get_secret_value(),
                                timeout=20.0, max_retries=2)  # SDK retries 408/429/5xx with backoff
    app.state.jwks = jwt.PyJWKClient(S.oidc_jwks_url, cache_keys=True)
    try:
        yield
    finally:
        await app.state.llm.close()
        await app.state.db.close()

app = FastAPI(title="Comment Assistant API", lifespan=lifespan)
bearer = HTTPBearer()

async def current_principal(
    request: Request, cred: Annotated[HTTPAuthorizationCredentials, Depends(bearer)],
) -> Principal:
    token = cred.credentials
    try:  # JWKS lookup may hit the network: keep it off the event loop
        key = await asyncio.to_thread(request.app.state.jwks.get_signing_key_from_jwt, token)
        claims = jwt.decode(token, key.key, algorithms=["RS256"], audience=S.oidc_audience,
                            issuer=S.oidc_issuer, options={"require": ["exp", "sub", "tenant_id"]})
    except jwt.PyJWTError:
        raise HTTPException(401, "invalid token") from None
    return Principal(tenant_id=claims["tenant_id"], actor_id=claims["sub"])  # never from headers

async def generate_reply(llm: AsyncOpenAI, comment: str):
    completion = await llm.chat.completions.create(
        model=CFG.model,
        temperature=CFG.temperature,
        max_completion_tokens=CFG.max_output_tokens,
        response_format={"type": "json_schema", "json_schema": {
            "name": "reply", "strict": True, "schema": CFG.output_schema}},
        messages=[
            {"role": "system", "content": CFG.system_prompt},
            # JSON-encode untrusted text: it cannot close a delimiter and "escape".
            {"role": "user", "content": json.dumps({"untrusted_comment": comment},
                                                   ensure_ascii=False)},
        ],
    )
    choice = completion.choices[0]
    if choice.finish_reason != "stop" or choice.message.refusal:
        raise ValueError(f"unusable completion ({choice.finish_reason})")  # e.g. truncated
    return ReplyDraft.model_validate_json(choice.message.content or ""), completion

def blocked(result: SafetyResult) -> HTTPException:
    code = 503 if result.reason_code == "moderation_unavailable" else 422
    return HTTPException(code, result.reason_code)

@app.post("/v1/drafts", response_model=DraftResponse, status_code=201)
async def create_draft(body: GenerateRequest, request: Request, response: Response,
                       who: Annotated[Principal, Depends(current_principal)]) -> DraftResponse:
    db = request.app.state.db
    key = (who.tenant_id, body.creator_account_id, body.source_comment_id, RELEASE.release_id)
    async with db.acquire() as conn:
        if not await conn.fetchval(MEMBER_SQL, who.tenant_id, body.creator_account_id,
                                   who.actor_id):
            raise HTTPException(403, "no access to this account")
        row = await conn.fetchrow(FIND_SQL, *key)
    if row is not None:  # idempotent replay: no second model call, no second charge
        response.status_code = 200
        return to_response(row)

    safety_in = await moderate(MODERATOR, body.comment_text)
    if safety_in.decision is Decision.BLOCK:
        raise blocked(safety_in)
    try:
        draft, completion = await generate_reply(request.app.state.llm, body.comment_text)
    except APITimeoutError:
        raise HTTPException(504, "model timeout") from None
    except (APIConnectionError, APIStatusError) as exc:
        log.warning("provider_error", extra={"error": type(exc).__name__})
        raise HTTPException(502, "model provider failure") from None
    except (ValueError, ValidationError) as exc:
        log.warning("invalid_model_output", extra={"error": type(exc).__name__})
        raise HTTPException(502, "invalid model output") from None
    usage = usage_from_chat_completions(completion.usage)
    log.info("llm_usage", extra={"model": completion.model, "release": RELEASE.release_id,
                                 **asdict(usage)})  # metadata only, never comment text

    final = combine(safety_in, await moderate(MODERATOR, draft.reply),
                    sensitive_topic=draft.needs_human_review)
    if final.decision is Decision.BLOCK:
        raise blocked(final)  # unsafe text is neither stored nor returned
    status = "pending_review" if final.decision is Decision.ALLOW else "needs_review"
    async with db.acquire() as conn:
        row = await conn.fetchrow(INSERT_SQL, *key[:3], status, draft.reply, key[3],
                                  FINGERPRINT, completion.model, final.decision.value,
                                  final.reason_code, who.actor_id)
        if row is None:  # a concurrent request won the race; return its draft
            row = await conn.fetchrow(FIND_SQL, *key)
            response.status_code = 200
    return to_response(row)

def to_response(row) -> DraftResponse:
    return DraftResponse(draft_id=str(row["draft_id"]), status=row["status"],
                         reply=row["reply_text"], release_id=RELEASE.release_id,
                         config_fingerprint=FINGERPRINT)

Identity: tokens are verified with the provider's JWKS (issuer, audience, expiry and required claims); a shared static token compared with != plus caller-supplied tenant headers is not authentication. The JWKS fetch runs in a thread so a key refresh cannot block the event loop.

Cost and idempotency: an existing draft is returned before moderation or generation runs, so client retries never pay twice. A concurrent duplicate is caught by ON CONFLICT and receives the winner's draft.

Output integrity: strict json_schema output replaces json_object mode, which guarantees valid JSON but not the required fields. max_completion_tokens replaces the deprecated max_tokens (some OpenAI-compatible servers still expect the older name). A truncated (finish_reason="length") or refused completion is rejected explicitly, and the stored model ID is the one the provider returned.

Safety: “review” outcomes are kept as needs_review drafts instead of being treated as allowed; blocked text is neither stored nor returned; a moderation outage returns 503. Logs carry usage and error types, never comment text.

Still required before launch: webhook verification, per-tenant rate limits, audit events, retention and deletion workflows, readiness probes that check the database and moderation service, and a publish adapter with its own authorisation.

9. Reliability, Deployment and SLOs

Set per-request deadlines and bounded concurrency; use provider-aware retries with backoff, jitter and Retry-After, without stacking application retries on top of SDK retries.

Use idempotency keys for inbound events and publish actions, persist state transitions, and never assume exactly-once delivery. Separate liveness from readiness: a live process can still have an unavailable database or moderation service.

Autoscale on queue depth and concurrency, deploy with rolling or canary releases, and rehearse restore, queue replay, dead-letter reprocessing, credential rotation, provider and regional outages, and rollback.

Example SLOs (set from measured baselines): 99.9% monthly availability for draft creation; p95 non-streaming draft latency under 5 seconds; 100% of drafts stored with release and fingerprint; zero cross-tenant access; 100% of publish operations tied to an approval record. Define separate SLOs for moderation and publishing.

10. Security, Privacy and Platform Compliance

Use OAuth with least-privilege scopes for connected accounts; encrypt refresh tokens with KMS-backed envelope encryption and support rotation and revocation.

Verify webhook signatures and follow each platform's API terms, rate limits, data-use and retention rules, and automation policies. Do not ingest direct messages without a valid purpose, user authorisation and platform permission.

Enforce tenant isolation everywhere data lives: database queries, caches, vector indexes, object storage, logs and background jobs. Apply role-based access, audit admin actions, and scan dependencies and containers.

Define retention and deletion per data category and propagate deletions to caches, indexes and backups under documented expiry. Never train on creator content without consent, contracts, privacy notices and provider terms that permit it.

Conclusion

LLMOps expands the production surface of MLOps. For a Comment Assistant, the unit of deployment is the whole versioned system of prompts, models, retrieval, policies, permissions and integrations. Production quality means being able to say which release produced a draft, rejecting unsafe content, protecting tenant data, staying within latency and spend budgets, recovering from dependency failures, and leaving publishing under explicit creator control. The code here is tested reference material, not a turnkey product.

Further reading: NIST AI Risk Management Framework (nist.gov/itl/ai-risk-management-framework); OWASP Top 10 for LLM Applications (owasp.org); OpenTelemetry (opentelemetry.io/docs); FastAPI (fastapi.tiangolo.com); PostgreSQL (postgresql.org/docs).

Back to all blogs