Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi xây dựng hệ thống UGC (User-Generated Content) moderation cho game của mình — từ việc sử dụng API chính thức OpenAI với chi phí $150/tháng, đến việc chuyển hoàn toàn sang HolySheep AI và tiết kiệm được 87% chi phí. Đây là playbook đầy đủ mà tôi ước có được khi bắt đầu dự án.
Tại sao đội ngũ game cần hệ thống UGC Moderation riêng?
Khi game của bạn cho phép người chơi tạo nội dung (skin, map, dialogue), đây là những rủi ro thực sự:
- Spam và quảng cáo lậu — Bot đăng tải hàng loạt link casino, cờ bạc
- Nội dung nhạy cảm — Hình ảnh khiêu dâm, bạo lực trong user-generated assets
- Vi phạm bản quyền — Sprite, soundtrack sao chép từ game khác
- Toxic behavior — Chat rác, lời lẽ thù địch trong custom dialogue
Với 50,000+ người dùng hoạt động hàng ngày, đội ngũ moderation thủ công của tôi mất 8 giờ/ngày chỉ để review. Chi phí API chính thức: $4.50/1000 hình ảnh + $0.50/1000 ký tự. Sau 3 tháng, hóa đơn OpenAI đã vượt $450.
Vì sao chọn HolySheep thay vì API chính thức?
Khi so sánh chi phí, tôi phát hiện HolySheep cung cấp cùng model nhưng với giá thấp hơn đáng kể:
| Model | API chính thức | HolySheep AI | Tiết kiệm |
|---|---|---|---|
| GPT-4o (Vision) | $0.00765/ảnh | $0.0012/ảnh | 84% |
| Kimi (128K context) | $0.015/1K token | $0.002/1K token | 87% |
| DeepSeek V3.2 | $0.27/1M token | $0.42/1M token | Không rẻ hơn |
Điểm cộng lớn nhất: HolySheep hỗ trợ WeChat Pay, Alipay, Visa/Mastercard — phương thức thanh toán quen thuộc với developer châu Á. Đăng ký lần đầu còn nhận tín dụng miễn phí $5 để test trước khi cam kết.
Kiến trúc hệ thống UGC Moderation
Đây là kiến trúc tôi đã triển khai thực tế cho game mobile RPG với 50K DAU:
┌─────────────────────────────────────────────────────────────────┐
│ Game Client (Unity/UE5) │
└─────────────────────────────┬───────────────────────────────────┘
│ POST /api/ugc/submit
▼
┌─────────────────────────────────────────────────────────────────┐
│ API Gateway (Nginx + Lua) │
│ - Rate limiting: 100 req/phút/user │
│ - JWT authentication │
│ - Request deduplication (Redis) │
└─────────────────────────────┬───────────────────────────────────┘
│
┌───────────────┼───────────────┐
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Image Queue │ │ Text Queue │ │ Audio Queue │
│ (Kafka) │ │ (Kafka) │ │ (Kafka) │
└──────┬──────┘ └──────┬──────┘ └──────┬──────┘
│ │ │
▼ ▼ ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ GPT-4o │ │ Kimi │ │ Whisper + │
│ Moderation │ │ Regulation │ │ Claude │
│ Worker │ │ Lookup │ │ Analysis │
└─────────────┘ └─────────────┘ └─────────────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────────────────────────────┐
│ Results + Flagged Content Store │
│ (PostgreSQL + Redis Cache + Dashboard) │
└─────────────────────────────────────────────────────────────┘Code mẫu: Image Moderation với GPT-4o qua HolySheep
Đây là module xử lý hình ảnh hoàn chỉnh — tôi đã test và deploy thành công trên production:
#!/usr/bin/env python3
"""
UGC Image Moderation Worker - HolySheep AI Integration
Production-ready với retry, timeout, và graceful degradation
"""
import base64
import hashlib
import hmac
import json
import time
import asyncio
import aiohttp
from dataclasses import dataclass
from enum import Enum
from typing import Optional, List
import redis.asyncio as redis
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
============================================================
CẤU HÌNH HOLYSHEEP - THAY THẾ BẰNG API KEY CỦA BẠN
============================================================
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY", # ← Thay bằng key thật
"model": "gpt-4o",
"timeout": 30, # giây
"max_retries": 3,
"retry_delay": 1.0, # giây
}
Redis cache cho kết quả đã duyệt
REDIS_URL = "redis://localhost:6379/0"
class ModerationResult(Enum):
APPROVED = "approved"
REJECTED = "rejected"
NEEDS_MANUAL_REVIEW = "needs_manual_review"
ERROR = "error"
@dataclass
class ImageModerationResponse:
status: ModerationResult
confidence: float
violations: List[str]
categories: dict
processing_time_ms: float
cached: bool = False
class HolySheepImageModerator:
"""
Image moderation sử dụng GPT-4o Vision qua HolySheep API
- Tự động retry khi fail
- Cache kết quả để tiết kiệm chi phí
- Rate limiting-aware
"""
# Prompt hệ thống cho moderation
MODERATION_SYSTEM_PROMPT = """Bạn là AI moderation chuyên kiểm tra nội dung game UGC.
QUY TẮC NGHIÊM NGẶT - Reject nếu phát hiện:
1. Nội dung khiêu dâm, gợi dục (dù ẩn danh)
2. Bạo lực đẫm máu, tra tấn
3. Hình ảnh thực tế chụp lén (real person photos)
4. Logo thương hiệu, bản quyền (Nike, Disney, game khác)
5. Biểu tượng phân biệt chủng tộc, tôn giáo
6. Quảng cáo cờ bạc, casino, phishing links
CÁC TRƯỜNG HỢP CHO PHÉP:
- Anime/illustration style (dù có suggestiveness nhẹ)
- Fan art gốc (không copy nguyên design)
- Nội dung text trong game (emoji, meme hợp lệ)
Trả lời JSON với format:
{
"status": "approved|rejected|needs_manual_review",
"confidence": 0.0-1.0,
"violations": ["mô tả từng vi phạm nếu có"],
"categories": {
"nsfw": true/false,
"violence": true/false,
"copyright": true/false,
"hate_symbols": true/false
}
}"""
def __init__(self, config: dict = None):
self.config = config or HOLYSHEEP_CONFIG
self.redis_client: Optional[redis.Redis] = None
self.session: Optional[aiohttp.ClientSession] = None
async def initialize(self):
"""Khởi tạo connection pools"""
self.redis_client = await redis.from_url(REDIS_URL)
self.session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=self.config["timeout"])
)
async def close(self):
"""Cleanup connections"""
if self.session:
await self.session.close()
if self.redis_client:
await self.redis_client.close()
def _get_cache_key(self, image_data: bytes) -> str:
"""Generate cache key từ image hash"""
img_hash = hashlib.sha256(image_data).hexdigest()[:16]
return f"ugc:img:mod:{img_hash}"
async def _check_cache(self, image_data: bytes) -> Optional[dict]:
"""Kiểm tra cache trước khi gọi API"""
cache_key = self._get_cache_key(image_data)
cached = await self.redis_client.get(cache_key)
if cached:
result = json.loads(cached)
result["cached"] = True
return result
return None
async def _save_to_cache(self, image_data: bytes, result: dict, ttl: int = 86400):
"""Lưu kết quả vào cache (mặc định 24h)"""
cache_key = self._get_cache_key(image_data)
await self.redis_client.setex(cache_key, ttl, json.dumps(result))
async def _call_holysheep_api(self, image_base64: str) -> dict:
"""
Gọi HolySheep API - SỬ DỤNG ENDPOINT CHÍNH XÁC
"""
headers = {
"Authorization": f"Bearer {self.config['api_key']}",
"Content-Type": "application/json"
}
payload = {
"model": self.config["model"],
"messages": [
{
"role": "system",
"content": self.MODERATION_SYSTEM_PROMPT
},
{
"role": "user",
"content": [
{
"type": "text",
"text": "Kiểm tra hình ảnh UGC này và trả lời JSON:"
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
}
],
"max_tokens": 500,
"temperature": 0.1 # Low temperature cho consistent output
}
# ⚠️ QUAN TRỌNG: base_url phải là api.holysheep.ai/v1
url = f"{self.config['base_url']}/chat/completions"
async with self.session.post(url, headers=headers, json=payload) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(f"HolySheep API error {response.status}: {error_text}")
result = await response.json()
return result
def _parse_api_response(self, api_response: dict) -> dict:
"""Parse response từ HolySheep thành structured result"""
try:
content = api_response["choices"][0]["message"]["content"]
# GPT có thể trả JSON với markdown code block
if "```json" in content:
content = content.split("``json")[1].split("``")[0]
elif "```" in content:
content = content.split("``")[1].split("``")[0]
result = json.loads(content.strip())
return result
except (KeyError, json.JSONDecodeError) as e:
logger.error(f"Failed to parse API response: {e}")
raise Exception("Invalid response format from moderation model")
async def moderate_image(self, image_data: bytes) -> ImageModerationResponse:
"""
Main moderation function với đầy đủ error handling
"""
start_time = time.time()
# 1. Check cache trước
cached_result = await self._check_cache(image_data)
if cached_result:
logger.info(f"Cache hit - Image approved in {cached_result.get('processing_time_ms', 0):.0f}ms")
return ImageModerationResponse(
status=ModerationResult(cached_result["status"]),
confidence=cached_result["confidence"],
violations=cached_result.get("violations", []),
categories=cached_result["categories"],
processing_time_ms=cached_result.get("processing_time_ms", 0),
cached=True
)
# 2. Encode image và gọi API với retry
image_base64 = base64.b64encode(image_data).decode("utf-8")
last_error = None
for attempt in range(self.config["max_retries"]):
try:
api_response = await self._call_holysheep_api(image_base64)
parsed = self._parse_api_response(api_response)
processing_time = (time.time() - start_time) * 1000
# 3. Save to cache
await self._save_to_cache(image_data, {
**parsed,
"processing_time_ms": processing_time
})
return ImageModerationResponse(
status=ModerationResult(parsed["status"]),
confidence=parsed["confidence"],
violations=parsed.get("violations", []),
categories=parsed["categories"],
processing_time_ms=processing_time,
cached=False
)
except Exception as e:
last_error = e
logger.warning(f"Attempt {attempt + 1} failed: {e}")
if attempt < self.config["max_retries"] - 1:
await asyncio.sleep(self.config["retry_delay"] * (attempt + 1))
# Fallback: fail open nhưng log để manual review
logger.error(f"All retries exhausted: {last_error}")
return ImageModerationResponse(
status=ModerationResult.NEEDS_MANUAL_REVIEW,
confidence=0.0,
violations=["System error - requires manual review"],
categories={},
processing_time_ms=(time.time() - start_time) * 1000
)
============================================================
USAGE EXAMPLE - CÁCH SỬ DỤNG TRONG PRODUCTION
============================================================
async def main():
moderator = HolySheepImageModerator()
await moderator.initialize()
try:
# Đọc image từ file hoặc HTTP request
with open("test_ugc_image.jpg", "rb") as f:
image_data = f.read()
result = await moderator.moderate_image(image_data)
print(f"Status: {result.status.value}")
print(f"Confidence: {result.confidence:.2%}")
print(f"Processing: {result.processing_time_ms:.0f}ms")
print(f"Cached: {result.cached}")
print(f"Violations: {result.violations}")
if result.status == ModerationResult.APPROVED:
print("✅ Image passed - enable user upload")
elif result.status == ModerationResult.REJECTED:
print("🚫 Image rejected - notify user")
else:
print("⚠️ Manual review required")
finally:
await moderator.close()
if __name__ == "__main__":
asyncio.run(main())Code mẫu: Long-text Regulation Retrieval với Kimi
Với nội dung text dài (custom dialogue, story mode), tôi dùng Kimi 128K context để lookup regulations và kiểm tra compliance:
#!/usr/bin/env python3
"""
UGC Text Moderation với Kimi - Regulation Lookup
Xử lý long-form content: dialogue, story, chat logs
"""
import asyncio
import aiohttp
import json
import time
from typing import List, Optional, Dict
from dataclasses import dataclass
from enum import Enum
============================================================
HOLYSHEEP KIMI CONFIGURATION
============================================================
KIMI_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY", # ← API key từ HolySheep
"model": "moonshot-v1-128k", # 128K context window
"timeout": 60,
}
Game regulations database (có thể load từ file/DB)
GAME_REGULATIONS = """
=== GAME UGC REGULATIONS V1.3 ===
BẮT BUỘC TUÂN THỦ:
1. Không được phép:
- Sử dụng từ ngữ tục tĩu, phản cảm
- Liên quan đến chủ đề chính trị, tôn giáo nhạy cảm
- Quảng cáo sản phẩm/brand khác
- Tiết lộ thông tin cá nhân (doxing)
- Lời lẽ kích động thù địch giữa các nhóm
2. Giới hạn độ dài:
- Dialogue: tối đa 2000 ký tự
- Story chapter: tối đa 10000 ký tự
- Chat message: tối đa 500 ký tự
3. Character limit cho user content:
- Username: 3-20 ký tự, không emoji
- Guild name: 3-30 ký tự
- Item description: tối đa 200 ký tự
=== VIETNAM GAME RATING BOARD ===
- Game rated 12+: Không bạo lực quá mức, không nội dung nhạy cảm
- Game rated 16+: Cho phép drama nhưng giới hạn ngôn từ
"""
class TextViolationType(Enum):
PROFANITY = "profanity"
POLITICAL = "political"
ADVERTISING = "advertising"
PERSONAL_INFO = "personal_info"
HATE_SPEECH = "hate_speech"
LENGTH_EXCEEDED = "length_exceeded"
NONE = "none"
@dataclass
class TextModerationResult:
is_safe: bool
violations: List[Dict]
content_type: str
confidence: float
suggestions: List[str]
processing_time_ms: float
class KimiTextModerator:
"""
Long-text moderation sử dụng Kimi 128K context
- Hiểu được ngữ cảnh dài (không chỉ keyword matching)
- Áp dụng game-specific regulations
- Context-aware violation detection
"""
SYSTEM_PROMPT = """Bạn là AI moderation chuyên kiểm tra nội dung text trong game.
NHIỆM VỤ:
1. Đọc kỹ content được gửi
2. So sánh với regulations đã cho
3. Xác định các vi phạm (nếu có)
4. Đề xuất cách sửa nếu cần
LUÔN TRẢ LỜI JSON:
{
"is_safe": true/false,
"violations": [
{
"type": "profanity|political|advertising|personal_info|hate_speech|length_exceeded",
"description": "Mô tả cụ thể vi phạm",
"severity": "low|medium|high",
"position": "Vị trí trong text (nếu áp dụng)"
}
],
"confidence": 0.0-1.0,
"suggestions": ["Gợi ý sửa đổi nếu có violation"]
}"""
def __init__(self, config: dict = None):
self.config = config or KIMI_CONFIG
self.session: Optional[aiohttp.ClientSession] = None
async def initialize(self):
self.session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=self.config["timeout"])
)
async def close(self):
if self.session:
await self.session.close()
async def moderate_text(
self,
content: str,
content_type: str = "dialogue"
) -> TextModerationResult:
"""
Moderate text content với Kimi
content_type: "dialogue", "story", "chat", "username", "guild_name", "item_desc"
"""
start_time = time.time()
headers = {
"Authorization": f"Bearer {self.config['api_key']}",
"Content-Type": "application/json"
}
# Xác định giới hạn độ dài theo content_type
length_limits = {
"dialogue": 2000,
"story": 10000,
"chat": 500,
"username": 20,
"guild_name": 30,
"item_desc": 200
}
max_length = length_limits.get(content_type, 5000)
payload = {
"model": self.config["model"],
"messages": [
{"role": "system", "content": self.SYSTEM_PROMPT},
{
"role": "user",
"content": f"""KIỂM TRA NỘI DUNG SAU:
Loại content: {content_type}
Giới hạn độ dài: {max_length} ký tự
REGULATIONS:
{GAME_REGULATIONS}
NỘI DUNG CẦN KIỂM TRA:
---
{content}
---
TRẢ LỜI JSON:"""
}
],
"temperature": 0.1
}
url = f"{self.config['base_url']}/chat/completions"
async with self.session.post(url, headers=headers, json=payload) as response:
if response.status != 200:
error = await response.text()
raise Exception(f"Kimi API error {response.status}: {error}")
result = await response.json()
content_response = result["choices"][0]["message"]["content"]
# Parse JSON response
try:
if "```json" in content_response:
content_response = content_response.split("``json")[1].split("``")[0]
parsed = json.loads(content_response.strip())
except json.JSONDecodeError:
raise Exception(f"Invalid JSON from Kimi: {content_response[:200]}")
return TextModerationResult(
is_safe=parsed.get("is_safe", False),
violations=parsed.get("violations", []),
content_type=content_type,
confidence=parsed.get("confidence", 0.0),
suggestions=parsed.get("suggestions", []),
processing_time_ms=(time.time() - start_time) * 1000
)
============================================================
PRODUCTION INTEGRATION EXAMPLE
============================================================
async def example_usage():
moderator = KimiTextModerator()
await moderator.initialize()
try:
# Test với dialogue content
test_dialogue = """
[Player 1]: Này, mày có thấy server này lag không?
[Player 2]: Ừ, mình cũng gặp vấn đề từ chiều. Có vẻ như lag khi đánh boss.
[Player 1]: Mình nghĩ là do mấy thằng spam bot kia. Mod ơi vào xử lý đi!
[Player 3]: Ê mấy ku, mình vừa tìm được trick farm gold nhanh, inbox mình nha!
"""
result = await moderator.moderate_text(test_dialogue, content_type="dialogue")
print(f"✅ Safe: {result.is_safe}")
print(f"⏱️ Time: {result.processing_time_ms:.0f}ms")
print(f"📊 Confidence: {result.confidence:.0%}")
if result.violations:
print(f"🚫 Violations found: {len(result.violations)}")
for v in result.violations:
print(f" - [{v['severity']}] {v['type']}: {v['description']}")
else:
print("✅ No violations - content approved")
finally:
await moderator.close()
if __name__ == "__main__":
asyncio.run(example_usage())Rate Limiting Strategy cho High-Traffic Game
Với game có lượng user lớn, rate limiting là critical để tránh bị block và kiểm soát chi phí:
#!/usr/bin/env python3
"""
Rate Limiter với Token Bucket Algorithm
Designed cho game có spike traffic (event launch, update)
"""
import time
import asyncio
from typing import Dict, Optional
from dataclasses import dataclass
import hashlib
import redis.asyncio as redis
@dataclass
class RateLimitConfig:
"""Cấu hình rate limit cho từng endpoint"""
requests_per_minute: int
requests_per_hour: int
burst_size: int # Số request có thể burst trong thời gian ngắn
# Chi phí (tokens) cho từng loại request - tính theo HolySheep pricing
# GPT-4o Vision: ~100 tokens/request, Kimi: ~50 tokens/request
token_cost: float
class RateLimiter:
"""
Token Bucket Rate Limiter với Redis backend
- Distributed: hoạt động across multiple servers
- Per-user và per-endpoint limits
- Automatic cost tracking
"""
# Cấu hình rate limit mặc định
DEFAULT_LIMITS = {
"image_moderation": RateLimitConfig(
requests_per_minute=60,
requests_per_hour=1000,
burst_size=20,
token_cost=0.0012 # $0.0012/ảnh trên HolySheep
),
"text_moderation": RateLimitConfig(
requests_per_minute=120,
requests_per_hour=5000,
burst_size=50,
token_cost=0.0002 # $0.0002/1K tokens
),
"bulk_moderation": RateLimitConfig(
requests_per_minute=10,
requests_per_hour=100,
burst_size=5,
token_cost=0.0010
)
}
def __init__(self, redis_url: str = "redis://localhost:6379/1"):
self.redis = redis.Redis.from_url(redis_url, decode_responses=True)
self.limits = self.DEFAULT_LIMITS.copy()
def _get_key(self, user_id: str, endpoint: str, window: str) -> str:
"""Generate Redis key cho rate limit tracking"""
return f"ratelimit:{endpoint}:{user_id}:{window}"
async def check_rate_limit(
self,
user_id: str,
endpoint: str,
cost: Optional[float] = None
) -> Dict:
"""
Kiểm tra và cập nhật rate limit
Returns: {
"allowed": bool,
"remaining_minute": int,
"remaining_hour": int,
"retry_after": int, # seconds
"cost_estimate": float
}
"""
config = self.limits.get(endpoint)
if not config:
return {"allowed": True, "error": "Unknown endpoint"}
cost = cost or config.token_cost
now = time.time()
minute_key = self._get_key(user_id, endpoint, "minute")
hour_key = self._get_key(user_id, endpoint, "hour")
# Lua script for atomic check-and-increment
lua_script = """
local minute_key = KEYS[1]
local hour_key = KEYS[2]
local limit_minute = tonumber(ARGV[1])
local limit_hour = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local window_minute = 60
local window_hour = 3600
-- Get current counts
local minute_count = redis.call('GET', minute_key) or '0'
local hour_count = redis.call('GET', hour_key) or '0'
minute_count = tonumber(minute_count)
hour_count = tonumber(hour_count)
-- Check minute limit
if minute_count >= limit_minute then
local ttl = redis.call('TTL', minute_key)
return {0, minute_count, hour_count, ttl}
end
-- Check hour limit
if hour_count >= limit_hour then
local ttl = redis.call('TTL', hour_key)
return {0, minute_count, hour_count, ttl}
end
-- Increment counters
local minute_incr = redis.call('INCR', minute_key)
if minute_incr == 1 then
redis.call('EXPIRE', minute_key, window_minute)
end
local hour_incr = redis.call('INCR', hour_key)
if hour_incr == 1 then
redis.call('EXPIRE', hour_key, window_hour)
end
return {1, minute_incr, hour_incr, 0}
"""
result = await self.redis.eval(
lua_script, 2, minute_key, hour_key,
config.requests_per_minute,
config.requests_per_hour,
now
)
allowed = bool(result[0])
retry_after = int(result[3]) if result[3] > 0 else 0
# Track estimated cost
if allowed:
cost_key = f"cost:{user_id}:{time.strftime('%Y%m%d')}"
await self.redis.incrbyfloat(cost_key, cost)
await self.redis.expire(cost_key, 86400 * 2) # Keep for 2 days
return {
"allowed": allowed,
"remaining_minute": config.requests_per_minute - int(result[1]),
"remaining_hour": config.requests_per_hour - int(result[2]),
"retry_after": retry_after,
"cost_estimate": cost
}
async def get_daily_cost(self, user_id: str) -> float:
"""Lấy chi phí ước tính trong ngày"""
today_key = f"cost:{user_id}:{time.strftime('%Y%m%d')}"
cost = await self.redis.get(today_key)
return float(cost) if cost else 0.0
async def get_monthly_projection(self, user_id: str) -> float:
"""Project chi phí tháng dựa trên trend hiện tại"""
# Implementation: aggregate last 7 days and project
pass
============================================================
MIDDLEWARE INTEGRATION EXAMPLE
============================================================
class UGCAPIMiddleware:
"""
FastAPI middleware cho UGC moderation endpoints
"""
def __init__(self, rate_limiter: RateLimiter):
self.rate_limiter = rate_limiter
async def __call__(self, request, call_next):
user_id = request.headers.get("X-User-ID")
endpoint = request.url.path
# Skip rate limiting cho health check
if endpoint == "/health":
return await call_next(request)
result = await self.rate_limiter.check_rate_limit(user_id, endpoint)
if not result["allowed"]:
return {
"status_code": 429,
"body": {
"error": "Rate limit exceeded",
"retry_after": result["retry_after"],
"remaining_minute": 0,
"remaining_hour": 0
}
}
# Add rate limit headers
response = await call_next(request)
response.headers["X-RateLimit-Remaining-Minute"] = str(result["remaining_minute"])
response.headers["X-RateLimit-Remaining-Hour"] = str(result["remaining_hour"])
return response
============================================================
PRODUCTION USAGE
============================================================
async def production_example():
limiter = RateLimiter()
# Test rate limit
user_id = "user_12345"
result = await limiter.check_rate_limit(user_id, "image_moderation")
print(f"Allowed: {result['allowed']}")
print(f"Remaining (min): {result['remaining_minute']}")