Overview

The pipeline validates every ad copy variant through up to five sequential gates before it reaches Jon for review. Each gate evaluates different quality dimensions using a different AI model, providing diverse perspectives that catch issues a single model would miss.

Quick Start

Dry run (Gate 0 only, no API calls)

make copy-qa-dry

Or directly:

python scripts/copy_qa_pipeline.py \
  --input ./reports/ad-copy \
  --platform meta \
  --language en \
  --region us \
  --gate0-only

Full pipeline (requires API keys)

make copy-qa

Or directly:

python scripts/copy_qa_pipeline.py \
  --input ./reports/ad-copy \
  --platform meta \
  --language en \
  --region us \
  --output-dir ./reports/ad-copy/qa

Generate + QA in one step

python scripts/ad_copy_generator.py generate \
  --platform meta \
  --product "Thick-Cut Filet" \
  --region eu \
  --language de \
  --variants 5 \
  --output-dir ./reports/ad-copy

QA runs automatically after generation. Add --skip-qa to bypass.

Architecture

Gate 0 - Platform Compliance

Gate 0 runs first on every variant. If it fails, no AI gates run. This saves money on obviously non-compliant copy.

Checks performed

Character limits by platform

Result type

@dataclass
class Gate0Result:
    passed: bool
    issues: list[Gate0Issue]
    checks_run: int
    checks_passed: int

@dataclass
class Gate0Issue:
    field: str       # Which field failed (e.g., "primary_text")
    check: str       # Check name (e.g., "char_limit")
    severity: str    # HIGH, MEDIUM, LOW
    problem: str     # Human-readable description
    suggestion: str  # Fix suggestion

Gate 1 - Gemini 2.5 Flash

Evaluates brand alignment and creative quality.

Dimensions (0-100 each)

Score formula: overall = brand_voice*0.30 + cta_clarity*0.25 + audience_match*0.25 + cultural_fit*0.20

The prompt includes 2 good + 2 bad few-shot examples per language/platform combination (Fix #3) and the terminology glossary for the target language.

Gate 2 - GPT-4.1

Evaluates linguistic quality and persuasive effectiveness.

Dimensions (0-100 each)

Score formula: overall = fluency*0.35 + persuasion*0.35 + platform_fit*0.30

Gate 3 - Claude Sonnet 4.6

Evaluates semantic depth and cross-field consistency. For non-English copy, includes a back-translation check (Fix #4).

Dimensions (0-100 each)

Score formula: overall = semantic_fidelity*0.35 + register_consistency*0.35 + competitive_differentiation*0.30

Back-translation (Fix #4)

For non-English copy, the Gate 3 prompt instructs Claude to:

1. Translate the ad copy back to English
2. Compare with the original brief/seed copy
3. Flag any semantic drift (meaning changed, emphasis shifted, CTA weakened)
4. Factor drift into the semantic_fidelity score

This happens within the same API call - no extra cost.

Gate 4 - Claude Opus 4.6 (Conditional)

Opus arbitration fires only when reviewers disagree or scores fall in a borderline zone. It reviews all three previous gate results and renders a binding verdict.

Arbitration triggers

Gate 4 fires when ALL of these conditions are met:

1. Borderline zone: Any Gate 1/2/3 score is between 80-89, OR the spread between highest and lowest reviewer exceeds 15 points
2. Combined score range: The combined score before arbitration is between 75-94

If the combined score is already >= 95 (clearly excellent) or < 75 (clearly bad), Opus is skipped.

Verdict effects

The Opus score is double-weighted in the final aggregation formula.

Scoring and Pass/Fail Logic

Score aggregation

Without arbitration (most variants):

combined = round((Gate1 + Gate2 + Gate3) / 3)

With Opus arbitration:

combined = round((Gate1 + Gate2 + Gate3 + Gate4*2) / 5)

Thresholds

Blocking rules (override score thresholds)

Severity classification

Score formula enforcement (Fix #6)

Every AI gate prompt requires the model to show its arithmetic using the weighted formula. After receiving the response, the pipeline recalculates the overall score from the dimension scores. If the model's overall deviates by more than 2 points from the formula, the pipeline overrides it with the calculated value and sets _score_override: true in the result.

Six TEEI Gap Fixes

Issues identified in the TEEI translation QA system that were fixed from the start in this pipeline.

Integration Points

ad_copy_generator.py (post-generation QA)

After generate_copy_variants() produces variants, QA runs automatically:

from lib.copy_qa_pipeline import validate_batch

qa_result = validate_batch(
    variants,
    platform=platform,
    language=language,
    region=region,
    skip_ai=args.skip_qa,
)

The QA summary is included in the output JSON. Add --skip-qa to bypass QA entirely. QA is wrapped in try/except - if it fails, variants are still output without QA scores.

creative_refresh.py (replacing _score_variant)

The creative refresh loop uses the QA pipeline to score replacement copy:

from lib.copy_qa_pipeline import validate_batch

qa = validate_batch(
    replacements,
    platform="meta",
    language=language or "en",
    region=region or "us",
    skip_ai=skip_qa,
)

Variants scoring below the threshold (combined_score < 85) are filtered out. The existing --min-score 7 flag (on a 1-10 scale) maps to ~85 on the 0-100 QA scale.

copy_quality_scorer.py (foundation for Gate 0)

Gate 0 imports existing functions from copy_quality_scorer.py: CHAR_LIMITS, check_banned_words(), check_char_limits(). No code was duplicated.

CLI Usage

copy_qa_pipeline.py

usage: copy_qa_pipeline.py [-h] --input INPUT [--platform {meta,google,klaviyo}]
                           [--language LANGUAGE] [--region REGION]
                           [--output-dir OUTPUT_DIR] [--gate0-only]

options:
  --input          Directory containing ad copy JSON files, or a single JSON file
  --platform       Ad platform: meta, google, klaviyo (default: meta)
  --language       Language code: en, de, it, es (default: en)
  --region         Region code: us, eu (default: us)
  --output-dir     Directory to save QA results (default: ./reports/ad-copy/qa)
  --gate0-only     Run Gate 0 only - no API calls, no cost

Variant field names by platform

Output

The pipeline produces two files in the output directory:

1. qa_results_YYYYMMDD_HHMMSS.json - Full structured results (all gate scores, issues, blocking reasons)
2. qa_report_YYYYMMDD_HHMMSS.md - Human-readable markdown report

Makefile targets

make copy-qa       # Full pipeline (Meta, English, US)
make copy-qa-dry   # Gate 0 only (no API calls, $0)

Terminology Glossary

Universal terms (never translate)

Approved translations

Banned alternatives (never use)

Cost Breakdown

Per variant

Per creative refresh cycle (15 variants)

Monthly cost (weekly refresh)

API Keys

All keys are optional. If a key is missing, the corresponding gate logs an error and returns a zero score (it does not crash the pipeline). For --gate0-only / make copy-qa-dry, no API keys are needed.

File Reference

scripts/
  copy_qa_pipeline.py              # Orchestrator (790 LOC)
                                   # - BatchQAResult, VariantQAResult dataclasses
                                   # - deduplicate_issues() (Fix #2)
                                   # - check_terminology_consistency() (Fix #5)
                                   # - _aggregate_scores(), _determine_status()
                                   # - validate_variant(), validate_batch()
                                   # - generate_report() (JSON + markdown)
                                   # - CLI entry point (main)
  lib/
    __init__.py
    qa_gate0_compliance.py         # Gate 0: platform + language + glossary (~250 LOC)
    qa_gemini_client.py            # Gate 1: Gemini reviewer (~200 LOC)
    qa_openai_client.py            # Gate 2: GPT reviewer (~200 LOC)
    qa_anthropic_client.py         # Gate 3+4: Sonnet + Opus (~300 LOC)
    qa_retry_utils.py              # Retry infra (~276 LOC)
    qa_prompts.py                  # Prompt builders + few-shot (~400 LOC)
    qa_language_detect.py          # Stopword-based language detection (~228 LOC)
    qa_glossary.py                 # Locked terminology per language (~218 LOC)

Testing

All tests mock external API calls. No real API keys are needed to run the test suite.

# Run all tests
make test

# Run only QA pipeline tests
python -m pytest tests/test_copy_qa_pipeline.py tests/test_qa_*.py -v

Test coverage by module

Data Types Reference

BatchQAResult

@dataclass
class BatchQAResult:
    variants: list[VariantQAResult]   # One result per input variant
    summary: dict                      # Aggregate counts
    terminology_report: dict           # Cross-variant term consistency
    generated_at: str                  # ISO timestamp

    # Properties:
    .passed        -> list[VariantQAResult]  # status PASSED
    .needs_review  -> list[VariantQAResult]  # status NEEDS_REVIEW
    .failed        -> list[VariantQAResult]  # status FAILED

VariantQAResult

@dataclass
class VariantQAResult:
    variant_index: int             # Position in input batch
    variant: dict                  # The original variant dict
    status: str                    # PASSED, NEEDS_REVIEW, FAILED
    combined_score: int            # 0-100 aggregate score
    gate0: Gate0Result | None      # Platform compliance result
    gate1: dict | None             # Gemini result
    gate2: dict | None             # GPT result
    gate3: dict | None             # Sonnet result
    gate4: dict | None             # Opus result (if arbitrated)
    issues: list[dict]             # Deduplicated issues from all gates
    blocking_reasons: list[str]    # Why it was blocked (if FAILED)
    arbitrated: bool               # Whether Gate 4 was triggered

Summary dict

{
    "total": 5,
    "passed": 3,
    "needs_review": 1,
    "failed": 1,
    "arbitrated": 2,
    "avg_score": 82,
    "terminology_inconsistencies": 0
}

Issue dict (after deduplication)

{
    "field": "primary_text",
    "severity": "MEDIUM",
    "category": "fluency",
    "problem": "Awkward phrasing",
    "suggestion": "Rephrase...",
    "_sources": ["gate1", "gate2"],
    "_flagged_by": 2,
    "_unanimous": false
}