Building a Video Metadata Moderation Pipeline with Python and the Claude API

Every two hours our ingest cron wakes up and pulls a few thousand freshly trending videos from nine Asia-Pacific regions — Japan, Korea, Taiwan, Singapore, Vietnam, Thailand, Hong Kong and a couple more. By the time the rows land in SQLite, the metadata is already a problem. Titles arrive in Japanese, Korean, Traditional and Simplified Chinese, Thai and Vietnamese. A chunk of them are clickbait reuploads. Some carry adult keywords stuffed into the description to game search. A few are outright scams with a Telegram link where the channel name should be. At TopVideoHub we aggregate this firehose into a searchable index, and if we let the raw metadata through untouched, our SQLite FTS5 search starts surfacing garbage within a day.

We ran keyword blocklists for a year. They break constantly on CJK text because word boundaries do not exist the way they do in English, and because spammers rotate homoglyphs and full-width characters faster than we can add rules. This post is how we replaced ~4,000 lines of regex with a Claude-backed moderation pipeline that classifies, language-tags and scores every incoming video before it touches the search index.

What the pipeline actually has to decide

Moderation here is not a single yes/no. For each video we need four things out of one pass:

• A decision: approve, review (send to a human queue), or reject.
• A language tag: a BCP-47 code like ja, ko, zh-Hant, th, vi. This drives which FTS5 analyzer path and which UI locale we route the row to.
• Content categories: so the same call that moderates also feeds our category facets instead of a second model run.
• A confidence score: low-confidence approvals still go to the human queue.

The volume matters for the design. A single ingest cycle on our busiest node processes 3,000–4,000 rows. At that scale you do not reach for the biggest model on every row. We use claude-haiku-4-5 for the classification pass — it is fast, cheap, and more than capable of the structured judgement we need — and only escalate genuinely ambiguous rows to claude-sonnet-4-6.

Why an LLM instead of a bigger blocklist

The honest answer is CJK tokenization. Our search stack is PHP 8.4 in front of SQLite FTS5 with a trigram tokenizer, behind LiteSpeed and Cloudflare. Trigram search is excellent for retrieval across languages that do not use spaces, but it is useless as a moderation signal — it has no idea that a title means "free crypto giveaway" in Vietnamese. Keyword lists have the opposite problem: they only match what you already thought of, and full-width/half-width variants (Ａ vs A), Cyrillic look-alikes and zenkaku spaces defeat naive matching immediately.

An LLM reads intent. It does not care that the spammer wrote the URL with Unicode fraktur characters; it still sees a URL. That single property — semantic understanding that survives obfuscation — is why we switched.

Designing the moderation contract with tool use

The most important decision was to never parse free text out of the model. We force structured output through a tool definition, set tool_choice to that tool, and read the arguments. This gives us a stable schema, enum-constrained fields, and no fragile JSON-string extraction.

import anthropic

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from env

MODERATION_TOOL = {
    "name": "record_moderation",
    "description": "Record the moderation verdict for one video's metadata.",
    "input_schema": {
        "type": "object",
        "properties": {
            "decision": {
                "type": "string",
                "enum": ["approve", "review", "reject"],
            },
            "language": {
                "type": "string",
                "description": "BCP-47 code of the dominant title language, e.g. ja, ko, zh-Hant, th, vi",
            },
            "categories": {
                "type": "array",
                "items": {"type": "string"},
                "description": "1-3 content categories such as music, gaming, news, vlog",
            },
            "confidence": {
                "type": "number",
                "description": "0.0-1.0 confidence in the decision",
            },
            "reason": {"type": "string", "description": "One short sentence"},
        },
        "required": ["decision", "language", "categories", "confidence", "reason"],
    },
}

SYSTEM_PROMPT = (
    "You moderate metadata for an Asia-Pacific video aggregator. "
    "Reject scams, adult content, and keyword-stuffed spam. "
    "Send to review anything ambiguous or where confidence is below 0.7. "
    "Judge intent, not surface characters: obfuscated URLs, homoglyphs and "
    "full-width tricks still count as the thing they imitate. "
    "Detect the dominant language of the TITLE, not the description."
)

Notice the schema does the validation work for us. decision can only ever be one of three strings; the SDK will surface a model that tries to invent a fourth as a plain missing-enum case we can log, rather than a silent bad write.

Calling Claude and reading the verdict

The call itself is small. We pin tool_choice so the model must emit the tool, then pull the first tool_use block. We also attach cache_control to the system prompt — it is identical on every one of the thousands of calls per cycle, so prompt caching cuts the input cost dramatically after the first request.

from dataclasses import dataclass

@dataclass
class Verdict:
    decision: str
    language: str
    categories: list[str]
    confidence: float
    reason: str

def moderate(title: str, description: str, channel: str) -> Verdict:
    resp = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=400,
        system=[{
            "type": "text",
            "text": SYSTEM_PROMPT,
            "cache_control": {"type": "ephemeral"},
        }],
        tools=[MODERATION_TOOL],
        tool_choice={"type": "tool", "name": "record_moderation"},
        messages=[{
            "role": "user",
            "content": (
                f"Channel: {channel}\n"
                f"Title: {title}\n"
                f"Description: {description[:1500]}"
            ),
        }],
    )
    for block in resp.content:
        if block.type == "tool_use":
            data = block.input
            return Verdict(**data)
    raise RuntimeError("model returned no tool_use block")

We truncate the description to 1,500 characters. Spam descriptions are frequently 5,000+ characters of repeated keywords, and none of that extra volume improves the decision — it only costs tokens. The first 1,500 characters carry all the signal the model needs.

Running it at ingest scale

One synchronous call per row would make the ingest cron take longer than the two-hour window between runs. We fan the batch out with asyncio and an AsyncAnthropic client, bounded by a semaphore so we stay under the account rate limit. A bounded gather with per-row error isolation is the whole trick — one row that trips a content filter or times out must not sink the batch.

import asyncio
import anthropic

aclient = anthropic.AsyncAnthropic()
SEM = asyncio.Semaphore(8)  # tune to your rate limit

async def moderate_one(row: dict) -> dict:
    async with SEM:
        try:
            resp = await aclient.messages.create(
                model="claude-haiku-4-5",
                max_tokens=400,
                system=[{
                    "type": "text",
                    "text": SYSTEM_PROMPT,
                    "cache_control": {"type": "ephemeral"},
                }],
                tools=[MODERATION_TOOL],
                tool_choice={"type": "tool", "name": "record_moderation"},
                messages=[{
                    "role": "user",
                    "content": f"Channel: {row['channel']}\n"
                               f"Title: {row['title']}\n"
                               f"Description: {row['description'][:1500]}",
                }],
            )
            verdict = next(b.input for b in resp.content if b.type == "tool_use")
            return {"video_id": row["video_id"], **verdict}
        except Exception as exc:  # isolate: bad row -> manual queue
            return {"video_id": row["video_id"], "decision": "review",
                    "language": "und", "categories": [], "confidence": 0.0,
                    "reason": f"pipeline error: {exc}"}

async def moderate_batch(rows: list[dict]) -> list[dict]:
    return await asyncio.gather(*(moderate_one(r) for r in rows))

if __name__ == "__main__":
    import json, sys
    incoming = json.load(sys.stdin)
    results = asyncio.run(moderate_batch(incoming))
    json.dump(results, sys.stdout, ensure_ascii=False)

With a semaphore of 8 and prompt caching warm, a 3,500-row batch clears in a few minutes. The failure path is deliberate: anything that throws becomes a review verdict with language und, so a broken API call degrades to "a human looks at it" rather than "the row silently vanishes" or "garbage gets published."

Persisting verdicts into SQLite FTS5

The Python worker emits JSON on stdout; a small PHP CLI script on the ingest host consumes it and writes to SQLite. Only approve rows go into the FTS5 index. review rows go into a moderation queue table, reject rows are dropped but logged so we can audit false positives later. Keeping the search index write-path in PHP means it shares the exact PDO connection, pragmas and trigram tokenizer config the rest of the site uses — no second source of truth for how a video becomes searchable.

<?php
declare(strict_types=1);

$db = new PDO('sqlite:/var/data/topvideohub.sqlite');
$db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$db->exec('PRAGMA journal_mode = WAL');

$db->exec("CREATE VIRTUAL TABLE IF NOT EXISTS video_fts USING fts5(
    video_id UNINDEXED,
    title,
    description,
    lang UNINDEXED,
    tokenize = 'trigram'
)");

$db->exec('CREATE TABLE IF NOT EXISTS moderation_queue(
    video_id TEXT PRIMARY KEY, reason TEXT, confidence REAL, created_at INTEGER
)');

$verdicts = json_decode(stream_get_contents(STDIN), true, 512, JSON_THROW_ON_ERROR);

$insertFts = $db->prepare(
    'INSERT INTO video_fts(video_id, title, description, lang) VALUES(?, ?, ?, ?)'
);
$queue = $db->prepare(
    'INSERT OR REPLACE INTO moderation_queue(video_id, reason, confidence, created_at)
     VALUES(?, ?, ?, ?)'
);

$db->beginTransaction();
foreach ($verdicts as $v) {
    $approved = $v['decision'] === 'approve' && (float) $v['confidence'] >= 0.7;
    if ($approved) {
        $meta = getVideoMeta($db, $v['video_id']); // title/description from ingest table
        $insertFts->execute([$v['video_id'], $meta['title'], $meta['description'], $v['language']]);
    } elseif ($v['decision'] !== 'reject') {
        $queue->execute([$v['video_id'], $v['reason'], $v['confidence'], time()]);
    }
}
$db->commit();

The lang column is UNINDEXED because we never full-text search on it — we only filter by it. Storing it alongside the FTS row lets our PHP search layer scope a query to a single language cheaply (WHERE lang = 'ja') without a join back to the ingest table, which matters when LiteSpeed is serving a cached-miss search page under Cloudflare.

Getting CJK language detection right

Language detection is where naive approaches quietly fail. A Japanese title routinely mixes kanji (shared with Chinese), hiragana, katakana and Latin characters. Traditional vs Simplified Chinese cannot be told apart by codepoint ranges alone for many characters. Off-the-shelf langdetect-style libraries flip between zh and ja on short titles constantly.

We lean on Claude for this because it reads the whole title in context, but we still add guardrails:

• We ask for the TITLE language explicitly, not the description. Descriptions are frequently machine-translated to English and would mislabel the row.
• We accept und (undetermined) as a valid answer. Forcing a guess on a title that is purely an emoji string or a bare URL produces worse routing than admitting we do not know.
• We normalize before storage: zh with Traditional characters becomes zh-Hant, which is what Taiwan and Hong Kong traffic expects, so the locale switcher and hreflang tags line up.

The payoff is concrete: with correct language tags we can render the right UI locale, emit correct hreflang metadata, and — because the trigram tokenizer is language-agnostic — keep one FTS index while still scoping searches per language.

Caching, deduplication and cost control

Trending feeds repeat. The same viral video shows up across five regions in one cycle, and again in the next cycle two hours later. Calling the model twice for identical metadata is wasted money, so we short-circuit before the API:

• Content-hash cache: hash title + description, and if we moderated that hash in the last 7 days, reuse the verdict. On a typical cycle this eliminates 30–40% of calls outright.
• Prompt caching: the cache_control block on the system prompt means the identical instructions are not re-billed at full rate across the batch.
• Model tiering: Haiku handles the base pass. We only re-run the small slice of review-with-mid-confidence rows through Sonnet, which is where the harder judgement calls live.
• Description truncation: 1,500 characters caps the input tokens on the pathological spam rows that would otherwise dominate the bill.

Together these keep the per-cycle cost low enough that moderating every single incoming video — not a sample — is affordable.

What we measure after shipping it

A moderation pipeline you cannot audit is a liability, so every verdict is logged with its reason and confidence even when the row is rejected. That log is what lets us answer the only questions that matter:

• False-positive rate: how often did we reject something a human would have kept? We sample rejects weekly.
• Human queue depth: if review volume climbs, the confidence threshold or the prompt needs tuning.
• Language tag accuracy: spot-checked against the locale users actually engage with.
• Cost per 1,000 videos: tracked per cycle so a prompt change that quietly doubles output tokens gets caught immediately.

Since cutting over, keyword-stuffed spam in our search results dropped to near zero, the human moderation queue holds steady at a few dozen genuinely ambiguous rows per day instead of thousands, and we deleted the entire regex blocklist codebase.

Conclusion

The pattern that made this work is small and reusable: force structured output with a tool definition, fan the batch out with a bounded async client, degrade errors to a human-review verdict instead of dropping or publishing them, and cache aggressively on both the content hash and the prompt. The LLM is doing exactly one thing keyword lists never could — reading intent through obfuscation across languages that do not use word boundaries. Everything else is plumbing you already know how to build. If you run any kind of multi-language aggregator, moving moderation from pattern matching to semantic judgement is the single highest-leverage change available, and with a Haiku-class model on the base pass it is cheap enough to run on every row, not a sample.