Advanced

Human Review Queue

Build a human review queue with a reviewer dashboard for flagged content, approve/reject workflows, priority sorting, and an appeals process for content creators.

Review Queue Backend

# app/review/queue.py
import uuid
import logging
from datetime import datetime
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional

logger = logging.getLogger(__name__)


class ReviewStatus(str, Enum):
    PENDING = "pending"
    IN_REVIEW = "in_review"
    APPROVED = "approved"
    REJECTED = "rejected"
    APPEALED = "appealed"


@dataclass
class ReviewItem:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    content_id: str = ""
    content_type: str = "text"  # "text" or "image"
    content_preview: str = ""
    severity_score: float = 0.0
    categories: list[str] = field(default_factory=list)
    reasons: list[str] = field(default_factory=list)
    status: ReviewStatus = ReviewStatus.PENDING
    reviewer: Optional[str] = None
    reviewer_notes: str = ""
    created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    reviewed_at: Optional[str] = None
    appeal_reason: Optional[str] = None


class ReviewQueue:
    def __init__(self):
        self.items: dict[str, ReviewItem] = {}

    def add(self, content_id: str, content_type: str, preview: str,
            severity: float, categories: list, reasons: list) -> ReviewItem:
        item = ReviewItem(
            content_id=content_id, content_type=content_type,
            content_preview=preview[:500], severity_score=severity,
            categories=categories, reasons=reasons,
        )
        self.items[item.id] = item
        logger.info(f"Queued for review: {item.id} (severity: {severity:.2f})")
        return item

    def get_pending(self, limit: int = 20) -> list[ReviewItem]:
        pending = [i for i in self.items.values() if i.status == ReviewStatus.PENDING]
        return sorted(pending, key=lambda x: x.severity_score, reverse=True)[:limit]

    def claim(self, item_id: str, reviewer: str) -> ReviewItem:
        item = self.items.get(item_id)
        if not item: raise ValueError("Item not found")
        item.status = ReviewStatus.IN_REVIEW
        item.reviewer = reviewer
        return item

    def decide(self, item_id: str, action: str, notes: str = "") -> ReviewItem:
        item = self.items.get(item_id)
        if not item: raise ValueError("Item not found")
        item.status = ReviewStatus.APPROVED if action == "approve" else ReviewStatus.REJECTED
        item.reviewer_notes = notes
        item.reviewed_at = datetime.utcnow().isoformat()
        return item

    def appeal(self, item_id: str, reason: str) -> ReviewItem:
        item = self.items.get(item_id)
        if not item: raise ValueError("Item not found")
        item.status = ReviewStatus.APPEALED
        item.appeal_reason = reason
        return item

    def get_stats(self) -> dict:
        statuses = [i.status.value for i in self.items.values()]
        return {s: statuses.count(s) for s in set(statuses)}

review_queue = ReviewQueue()

Reviewer Dashboard

<!-- frontend/reviewer.html -->
<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <title>Content Review Dashboard</title>
  <style>
    * { margin: 0; padding: 0; box-sizing: border-box; }
    body { font-family: -apple-system, sans-serif; background: #f5f5f5; }
    .container { max-width: 1000px; margin: 0 auto; padding: 20px; }
    h1 { color: #6366f1; margin-bottom: 20px; }
    .stats { display: flex; gap: 16px; margin-bottom: 24px; }
    .stat { background: white; padding: 16px 24px; border-radius: 8px;
      box-shadow: 0 1px 3px rgba(0,0,0,0.1); text-align: center; }
    .stat-num { font-size: 28px; font-weight: 700; color: #6366f1; }
    .queue-item { background: white; border-radius: 8px; padding: 20px;
      margin-bottom: 16px; box-shadow: 0 1px 3px rgba(0,0,0,0.1);
      border-left: 4px solid #f59e0b; }
    .queue-item.high { border-left-color: #ef4444; }
    .severity { font-weight: 700; }
    .severity.high { color: #ef4444; }
    .severity.medium { color: #f59e0b; }
    .actions { display: flex; gap: 8px; margin-top: 12px; }
    .btn { padding: 8px 20px; border: none; border-radius: 6px;
      cursor: pointer; font-weight: 500; }
    .btn-approve { background: #10b981; color: white; }
    .btn-reject { background: #ef4444; color: white; }
    .btn-skip { background: #6b7280; color: white; }
    .preview { background: #f9fafb; padding: 12px; border-radius: 6px;
      margin: 12px 0; font-family: monospace; white-space: pre-wrap; }
    .tags { display: flex; gap: 6px; margin: 8px 0; }
    .tag { padding: 2px 10px; border-radius: 12px; font-size: 12px;
      background: #fee2e2; color: #991b1b; }
  </style>
</head>
<body>
  <div class="container">
    <h1>🔮 Content Review Dashboard</h1>
    <div class="stats" id="stats"></div>
    <div id="queue">Loading...</div>
  </div>
  <script>
    async function loadQueue() {
      const res = await fetch("/api/review/pending");
      const data = await res.json();
      const el = document.getElementById("queue");
      if (!data.items.length) { el.innerHTML = "<p>No items to review</p>"; return; }
      el.innerHTML = data.items.map(item => `
        <div class="queue-item ${item.severity_score > 0.7 ? 'high' : ''}">
          <div>
            <span class="severity ${item.severity_score > 0.7 ? 'high' : 'medium'}">
              Severity: ${(item.severity_score * 100).toFixed(0)}%
            </span>
            <span> | ${item.content_type} | ${item.created_at}</span>
          </div>
          <div class="tags">${item.categories.map(c =>
            `<span class="tag">${c}</span>`).join("")}</div>
          <div class="preview">${item.content_preview}</div>
          <div class="actions">
            <button class="btn btn-approve" onclick="decide('${item.id}','approve')">Approve</button>
            <button class="btn btn-reject" onclick="decide('${item.id}','reject')">Reject</button>
            <button class="btn btn-skip" onclick="loadQueue()">Skip</button>
          </div>
        </div>
      `).join("");
    }
    async function decide(id, action) {
      await fetch(`/api/review/${id}/decide`, {
        method: "POST", headers: {"Content-Type": "application/json"},
        body: JSON.stringify({action, notes: ""})
      });
      loadQueue();
    }
    loadQueue();
  </script>
</body>
</html>

Review API Endpoints

# Add to app/main.py
from app.review.queue import review_queue

@app.get("/api/review/pending")
async def get_pending_reviews(limit: int = 20):
    items = review_queue.get_pending(limit)
    return {"items": [vars(i) for i in items]}

@app.post("/api/review/{item_id}/decide")
async def review_decide(item_id: str, body: dict):
    item = review_queue.decide(item_id, body["action"], body.get("notes", ""))
    return {"item_id": item.id, "status": item.status.value}

@app.get("/api/review/stats")
async def review_stats():
    return review_queue.get_stats()
💡
Priority sorting: Items are sorted by severity score so reviewers see the most dangerous content first. This ensures critical violations are addressed before borderline cases.

Key Takeaways

  • High-severity items appear first in the queue so reviewers tackle critical content immediately.
  • Claim/decide workflow prevents two reviewers from working on the same item.
  • Appeals allow content creators to challenge rejections with a reason.
  • Stats tracking shows pending, approved, rejected, and appealed counts for oversight.