All templates include comprehensive error handling, retry logic, and are production-ready. Simply replace API keys and customize for your needs.
Content Moderation System
A complete content moderation system with webhook integration.- Python (Flask)
- Node.js (Express)
"""
HITL Content Moderation System
A complete Flask application for moderating user-generated content
"""
from flask import Flask, request, jsonify, render_template
import requests
import hmac
import hashlib
import json
import os
from datetime import datetime
import logging
app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
# Configuration
HITL_API_KEY = os.getenv('HITL_API_KEY')
HITL_BASE_URL = 'https://api.hitl.sh/v1'
WEBHOOK_SECRET = os.getenv('WEBHOOK_SECRET')
MODERATION_LOOP_ID = os.getenv('MODERATION_LOOP_ID')
class HITLModerationClient:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = HITL_BASE_URL
self.headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
}
def create_moderation_request(self, content, user_id, content_type='comment', priority='medium'):
"""Submit content for moderation"""
request_data = {
'loop_id': MODERATION_LOOP_ID,
'request_text': f'''Please moderate this {content_type}:
Content: {content}
User ID: {user_id}
Guidelines:
- Check for spam, harassment, inappropriate language
- Consider community standards
- Flag any policy violations''',
'response_type': 'single_select',
'response_config': {
'options': [
'✅ Approve - Content follows guidelines',
'⚠️ Approve with Warning - Minor issues, notify user',
'❌ Reject - Remove content, warn user',
'🚨 Reject and Suspend - Serious violation, suspend user'
]
},
'priority': priority,
'processing_type': 'time-sensitive',
'timeout_minutes': 30,
'metadata': {
'user_id': user_id,
'content_type': content_type,
'original_content': content[:200], # Store excerpt
'timestamp': datetime.now().isoformat()
}
}
try:
response = requests.post(
f'{self.base_url}/requests',
headers=self.headers,
json=request_data,
timeout=30
)
response.raise_for_status()
result = response.json()
logging.info(f"Moderation request created: {result['data']['request']['id']}")
return result['data']['request']
except requests.exceptions.RequestException as e:
logging.error(f"Failed to create moderation request: {e}")
return None
def get_moderation_stats(self):
"""Get moderation statistics"""
try:
response = requests.get(
f'{self.base_url}/requests?limit=100',
headers=self.headers,
timeout=30
)
response.raise_for_status()
requests_data = response.json()['data']['requests']
stats = {
'total_requests': len(requests_data),
'pending': len([r for r in requests_data if r['status'] == 'pending']),
'completed': len([r for r in requests_data if r['status'] == 'completed']),
'approval_rate': 0
}
completed_requests = [r for r in requests_data if r['status'] == 'completed']
if completed_requests:
approved = len([r for r in completed_requests if 'Approve' in r.get('response_data', '')])
stats['approval_rate'] = (approved / len(completed_requests)) * 100
return stats
except requests.exceptions.RequestException as e:
logging.error(f"Failed to get stats: {e}")
return None
# Initialize HITL client
hitl_client = HITLModerationClient(HITL_API_KEY)
def verify_webhook_signature(payload, signature, secret):
"""Verify webhook signature"""
if not signature or not signature.startswith('sha256='):
return False
expected_signature = hmac.new(
secret.encode('utf-8'),
payload.encode('utf-8'),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f'sha256={expected_signature}', signature)
@app.route('/')
def dashboard():
"""Moderation dashboard"""
stats = hitl_client.get_moderation_stats()
return render_template('dashboard.html', stats=stats)
@app.route('/api/moderate', methods=['POST'])
def moderate_content():
"""Submit content for moderation"""
try:
data = request.json
# Validate required fields
required_fields = ['content', 'user_id']
if not all(field in data for field in required_fields):
return jsonify({'error': 'Missing required fields'}), 400
# Create moderation request
result = hitl_client.create_moderation_request(
content=data['content'],
user_id=data['user_id'],
content_type=data.get('content_type', 'comment'),
priority=data.get('priority', 'medium')
)
if result:
return jsonify({
'success': True,
'request_id': result['id'],
'status': 'submitted',
'message': 'Content submitted for moderation'
})
else:
return jsonify({'error': 'Failed to submit for moderation'}), 500
except Exception as e:
logging.error(f"Error in moderate_content: {e}")
return jsonify({'error': 'Internal server error'}), 500
@app.route('/webhook/hitl', methods=['POST'])
def handle_webhook():
"""Handle HITL webhooks"""
try:
# Verify signature
payload = request.get_data(as_text=True)
signature = request.headers.get('X-HITL-Signature-256')
if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
logging.warning('Invalid webhook signature')
return jsonify({'error': 'Invalid signature'}), 401
# Parse webhook data
webhook_data = json.loads(payload)
event_type = webhook_data['event']
data = webhook_data['data']
if event_type == 'request.completed':
handle_moderation_completed(data)
elif event_type == 'request.timeout':
handle_moderation_timeout(data)
return jsonify({'status': 'success'})
except Exception as e:
logging.error(f"Webhook error: {e}")
return jsonify({'error': 'Webhook processing failed'}), 500
def handle_moderation_completed(data):
"""Process completed moderation request"""
request_data = data['request']
reviewer = data['reviewer']
# Extract metadata
metadata = request_data.get('metadata', {})
user_id = metadata.get('user_id')
content_type = metadata.get('content_type', 'content')
# Get moderation decision
decision = request_data['response_data']
logging.info(f"Moderation completed for user {user_id}: {decision}")
# Apply moderation decision
if 'Approve' in decision:
if 'Warning' in decision:
send_user_warning(user_id, 'content_warning')
logging.info(f"Content approved with warning for user {user_id}")
else:
logging.info(f"Content approved for user {user_id}")
elif 'Reject' in decision:
if 'Suspend' in decision:
suspend_user(user_id, reason='policy_violation')
remove_content(user_id, content_type)
logging.warning(f"User {user_id} suspended for policy violation")
else:
remove_content(user_id, content_type)
send_user_warning(user_id, 'content_removed')
logging.info(f"Content rejected for user {user_id}")
def handle_moderation_timeout(data):
"""Handle timed out moderation requests"""
request_data = data['request']
metadata = request_data.get('metadata', {})
user_id = metadata.get('user_id')
# Apply conservative default (remove content)
remove_content(user_id, metadata.get('content_type', 'content'))
logging.warning(f"Moderation timeout for user {user_id} - content removed by default")
def send_user_warning(user_id, warning_type):
"""Send warning to user (implement your notification system)"""
# Implement your user notification system here
logging.info(f"Sending {warning_type} warning to user {user_id}")
def remove_content(user_id, content_type):
"""Remove user content (implement your content management system)"""
# Implement your content removal logic here
logging.info(f"Removing {content_type} for user {user_id}")
def suspend_user(user_id, reason):
"""Suspend user account (implement your user management system)"""
# Implement your user suspension logic here
logging.warning(f"Suspending user {user_id} for {reason}")
@app.route('/api/stats')
def get_stats():
"""Get moderation statistics"""
stats = hitl_client.get_moderation_stats()
if stats:
return jsonify(stats)
else:
return jsonify({'error': 'Failed to fetch stats'}), 500
if __name__ == '__main__':
# Ensure required environment variables are set
required_env = ['HITL_API_KEY', 'WEBHOOK_SECRET', 'MODERATION_LOOP_ID']
missing_env = [var for var in required_env if not os.getenv(var)]
if missing_env:
logging.error(f"Missing required environment variables: {missing_env}")
exit(1)
app.run(debug=False, host='0.0.0.0', port=5000)
# .env file
HITL_API_KEY=your_hitl_api_key_here
WEBHOOK_SECRET=your_webhook_secret_here
MODERATION_LOOP_ID=your_loop_id_here
# Install dependencies
pip install flask requests
# Run the application
python app.py
<!DOCTYPE html>
<html>
<head>
<title>Content Moderation Dashboard</title>
<style>
body { font-family: Arial, sans-serif; margin: 20px; }
.stats { display: flex; gap: 20px; margin-bottom: 30px; }
.stat-card {
background: #f5f5f5;
padding: 20px;
border-radius: 8px;
text-align: center;
}
.stat-value { font-size: 2em; font-weight: bold; color: #333; }
.stat-label { color: #666; margin-top: 5px; }
</style>
</head>
<body>
<h1>Content Moderation Dashboard</h1>
{% if stats %}
<div class="stats">
<div class="stat-card">
<div class="stat-value">{{ stats.total_requests }}</div>
<div class="stat-label">Total Requests</div>
</div>
<div class="stat-card">
<div class="stat-value">{{ stats.pending }}</div>
<div class="stat-label">Pending Review</div>
</div>
<div class="stat-card">
<div class="stat-value">{{ stats.completed }}</div>
<div class="stat-label">Completed</div>
</div>
<div class="stat-card">
<div class="stat-value">{{ "%.1f"|format(stats.approval_rate) }}%</div>
<div class="stat-label">Approval Rate</div>
</div>
</div>
{% else %}
<p>Unable to load statistics</p>
{% endif %}
<h2>Test Content Moderation</h2>
<form id="moderationForm">
<textarea id="content" placeholder="Enter content to moderate..." rows="4" cols="50"></textarea><br><br>
<input type="text" id="userId" placeholder="User ID" required><br><br>
<select id="priority">
<option value="low">Low Priority</option>
<option value="medium" selected>Medium Priority</option>
<option value="high">High Priority</option>
</select><br><br>
<button type="submit">Submit for Moderation</button>
</form>
<script>
document.getElementById('moderationForm').onsubmit = function(e) {
e.preventDefault();
const data = {
content: document.getElementById('content').value,
user_id: document.getElementById('userId').value,
priority: document.getElementById('priority').value
};
fetch('/api/moderate', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify(data)
})
.then(response => response.json())
.then(result => {
if (result.success) {
alert('Content submitted for moderation! Request ID: ' + result.request_id);
document.getElementById('content').value = '';
} else {
alert('Error: ' + result.error);
}
});
};
</script>
</body>
</html>
/**
* HITL Content Moderation System
* Complete Express.js application for content moderation
*/
const express = require('express');
const axios = require('axios');
const crypto = require('crypto');
const path = require('path');
const app = express();
app.use(express.json());
app.use(express.static('public'));
// Configuration
const HITL_API_KEY = process.env.HITL_API_KEY;
const HITL_BASE_URL = 'https://api.hitl.sh/v1';
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;
const MODERATION_LOOP_ID = process.env.MODERATION_LOOP_ID;
class HITLModerationClient {
constructor(apiKey) {
this.apiKey = apiKey;
this.baseURL = HITL_BASE_URL;
this.headers = {
'Authorization': `Bearer ${apiKey}`,
'Content-Type': 'application/json'
};
}
async createModerationRequest(content, userId, contentType = 'comment', priority = 'medium') {
const requestData = {
loop_id: MODERATION_LOOP_ID,
request_text: `Please moderate this ${contentType}:
Content: ${content}
User ID: ${userId}
Guidelines:
- Check for spam, harassment, inappropriate language
- Consider community standards
- Flag any policy violations`,
response_type: 'single_select',
response_config: {
options: [
'✅ Approve - Content follows guidelines',
'⚠️ Approve with Warning - Minor issues, notify user',
'❌ Reject - Remove content, warn user',
'🚨 Reject and Suspend - Serious violation, suspend user'
]
},
priority,
processing_type: 'time-sensitive',
timeout_minutes: 30,
metadata: {
user_id: userId,
content_type: contentType,
original_content: content.substring(0, 200),
timestamp: new Date().toISOString()
}
};
try {
const response = await axios.post(`${this.baseURL}/requests`, requestData, {
headers: this.headers,
timeout: 30000
});
console.log(`Moderation request created: ${response.data.data.request.id}`);
return response.data.data.request;
} catch (error) {
console.error('Failed to create moderation request:', error.response?.data || error.message);
return null;
}
}
async getModerationStats() {
try {
const response = await axios.get(`${this.baseURL}/requests?limit=100`, {
headers: this.headers,
timeout: 30000
});
const requests = response.data.data.requests;
const stats = {
total_requests: requests.length,
pending: requests.filter(r => r.status === 'pending').length,
completed: requests.filter(r => r.status === 'completed').length,
approval_rate: 0
};
const completedRequests = requests.filter(r => r.status === 'completed');
if (completedRequests.length > 0) {
const approved = completedRequests.filter(r =>
r.response_data && r.response_data.includes('Approve')
).length;
stats.approval_rate = (approved / completedRequests.length) * 100;
}
return stats;
} catch (error) {
console.error('Failed to get stats:', error.response?.data || error.message);
return null;
}
}
}
// Initialize HITL client
const hitlClient = new HITLModerationClient(HITL_API_KEY);
function verifyWebhookSignature(payload, signature, secret) {
if (!signature || !signature.startsWith('sha256=')) {
return false;
}
const expectedSignature = crypto
.createHmac('sha256', secret)
.update(payload, 'utf8')
.digest('hex');
return crypto.timingSafeEqual(
Buffer.from(`sha256=${expectedSignature}`),
Buffer.from(signature)
);
}
// Routes
app.get('/', async (req, res) => {
const stats = await hitlClient.getModerationStats();
res.send(`
<!DOCTYPE html>
<html>
<head>
<title>Content Moderation Dashboard</title>
<style>
body { font-family: Arial, sans-serif; margin: 20px; }
.stats { display: flex; gap: 20px; margin-bottom: 30px; }
.stat-card {
background: #f5f5f5;
padding: 20px;
border-radius: 8px;
text-align: center;
}
.stat-value { font-size: 2em; font-weight: bold; color: #333; }
.stat-label { color: #666; margin-top: 5px; }
</style>
</head>
<body>
<h1>Content Moderation Dashboard</h1>
${stats ? `
<div class="stats">
<div class="stat-card">
<div class="stat-value">${stats.total_requests}</div>
<div class="stat-label">Total Requests</div>
</div>
<div class="stat-card">
<div class="stat-value">${stats.pending}</div>
<div class="stat-label">Pending Review</div>
</div>
<div class="stat-card">
<div class="stat-value">${stats.completed}</div>
<div class="stat-label">Completed</div>
</div>
<div class="stat-card">
<div class="stat-value">${stats.approval_rate.toFixed(1)}%</div>
<div class="stat-label">Approval Rate</div>
</div>
</div>
` : '<p>Unable to load statistics</p>'}
<h2>Test Content Moderation</h2>
<form id="moderationForm">
<textarea id="content" placeholder="Enter content to moderate..." rows="4" cols="50"></textarea><br><br>
<input type="text" id="userId" placeholder="User ID" required><br><br>
<select id="priority">
<option value="low">Low Priority</option>
<option value="medium" selected>Medium Priority</option>
<option value="high">High Priority</option>
</select><br><br>
<button type="submit">Submit for Moderation</button>
</form>
<script>
document.getElementById('moderationForm').onsubmit = function(e) {
e.preventDefault();
const data = {
content: document.getElementById('content').value,
user_id: document.getElementById('userId').value,
priority: document.getElementById('priority').value
};
fetch('/api/moderate', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify(data)
})
.then(response => response.json())
.then(result => {
if (result.success) {
alert('Content submitted for moderation! Request ID: ' + result.request_id);
document.getElementById('content').value = '';
} else {
alert('Error: ' + result.error);
}
});
};
</script>
</body>
</html>
`);
});
app.post('/api/moderate', async (req, res) => {
try {
const { content, user_id, content_type = 'comment', priority = 'medium' } = req.body;
// Validate required fields
if (!content || !user_id) {
return res.status(400).json({ error: 'Missing required fields: content, user_id' });
}
// Create moderation request
const result = await hitlClient.createModerationRequest(content, user_id, content_type, priority);
if (result) {
res.json({
success: true,
request_id: result.id,
status: 'submitted',
message: 'Content submitted for moderation'
});
} else {
res.status(500).json({ error: 'Failed to submit for moderation' });
}
} catch (error) {
console.error('Error in moderate_content:', error);
res.status(500).json({ error: 'Internal server error' });
}
});
app.post('/webhook/hitl', (req, res) => {
try {
// Verify signature
const payload = JSON.stringify(req.body);
const signature = req.headers['x-hitl-signature-256'];
if (!verifyWebhookSignature(payload, signature, WEBHOOK_SECRET)) {
console.warn('Invalid webhook signature');
return res.status(401).json({ error: 'Invalid signature' });
}
const { event, data } = req.body;
if (event === 'request.completed') {
handleModerationCompleted(data);
} else if (event === 'request.timeout') {
handleModerationTimeout(data);
}
res.json({ status: 'success' });
} catch (error) {
console.error('Webhook error:', error);
res.status(500).json({ error: 'Webhook processing failed' });
}
});
function handleModerationCompleted(data) {
const request = data.request;
const reviewer = data.reviewer;
const metadata = request.metadata || {};
const userId = metadata.user_id;
const contentType = metadata.content_type || 'content';
const decision = request.response_data;
console.log(`Moderation completed for user ${userId}: ${decision}`);
// Apply moderation decision
if (decision.includes('Approve')) {
if (decision.includes('Warning')) {
sendUserWarning(userId, 'content_warning');
console.log(`Content approved with warning for user ${userId}`);
} else {
console.log(`Content approved for user ${userId}`);
}
} else if (decision.includes('Reject')) {
if (decision.includes('Suspend')) {
suspendUser(userId, 'policy_violation');
removeContent(userId, contentType);
console.warn(`User ${userId} suspended for policy violation`);
} else {
removeContent(userId, contentType);
sendUserWarning(userId, 'content_removed');
console.log(`Content rejected for user ${userId}`);
}
}
}
function handleModerationTimeout(data) {
const request = data.request;
const metadata = request.metadata || {};
const userId = metadata.user_id;
// Apply conservative default (remove content)
removeContent(userId, metadata.content_type || 'content');
console.warn(`Moderation timeout for user ${userId} - content removed by default`);
}
function sendUserWarning(userId, warningType) {
// Implement your user notification system here
console.log(`Sending ${warningType} warning to user ${userId}`);
}
function removeContent(userId, contentType) {
// Implement your content removal logic here
console.log(`Removing ${contentType} for user ${userId}`);
}
function suspendUser(userId, reason) {
// Implement your user suspension logic here
console.warn(`Suspending user ${userId} for ${reason}`);
}
app.get('/api/stats', async (req, res) => {
const stats = await hitlClient.getModerationStats();
if (stats) {
res.json(stats);
} else {
res.status(500).json({ error: 'Failed to fetch stats' });
}
});
// Start server
const PORT = process.env.PORT || 3000;
// Ensure required environment variables
const requiredEnv = ['HITL_API_KEY', 'WEBHOOK_SECRET', 'MODERATION_LOOP_ID'];
const missingEnv = requiredEnv.filter(key => !process.env[key]);
if (missingEnv.length > 0) {
console.error(`Missing required environment variables: ${missingEnv.join(', ')}`);
process.exit(1);
}
app.listen(PORT, () => {
console.log(`Content moderation server running on port ${PORT}`);
console.log(`Dashboard: http://localhost:${PORT}`);
console.log(`Webhook endpoint: http://localhost:${PORT}/webhook/hitl`);
});
{
"name": "hitl-content-moderation",
"version": "1.0.0",
"description": "HITL Content Moderation System",
"main": "server.js",
"scripts": {
"start": "node server.js",
"dev": "nodemon server.js"
},
"dependencies": {
"express": "^4.18.2",
"axios": "^1.6.0"
},
"devDependencies": {
"nodemon": "^3.0.1"
}
}
# .env file
HITL_API_KEY=your_hitl_api_key_here
WEBHOOK_SECRET=your_webhook_secret_here
MODERATION_LOOP_ID=your_loop_id_here
PORT=3000
# Install dependencies
npm install
# Run the application
npm start
AI Quality Assurance System
Review AI-generated content before publication.- Python
"""
HITL AI Quality Assurance System
Review AI-generated content with human experts before publication
"""
import requests
import json
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
@dataclass
class AIContent:
content: str
content_type: str # 'article', 'code', 'creative'
ai_model: str
topic: str
target_audience: str
metadata: Dict
class AIQualityReviewSystem:
def __init__(self, api_key: str, review_loop_id: str):
self.api_key = api_key
self.review_loop_id = review_loop_id
self.base_url = 'https://api.hitl.sh/v1'
self.headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
}
def create_quality_review(self, ai_content: AIContent, priority: str = 'medium') -> Optional[Dict]:
"""Submit AI-generated content for human quality review"""
# Customize review request based on content type
if ai_content.content_type == 'article':
request_config = self._get_article_review_config()
elif ai_content.content_type == 'code':
request_config = self._get_code_review_config()
elif ai_content.content_type == 'creative':
request_config = self._get_creative_review_config()
else:
request_config = self._get_general_review_config()
request_data = {
'loop_id': self.review_loop_id,
'request_text': f'''Quality Review Request
Content Type: {ai_content.content_type}
AI Model: {ai_content.ai_model}
Topic: {ai_content.topic}
Target Audience: {ai_content.target_audience}
Content to Review:
{ai_content.content}
Please evaluate this AI-generated content for publication readiness.''',
**request_config,
'priority': priority,
'processing_type': 'deferred',
'timeout_minutes': 120, # Give reviewers time for thorough review
'metadata': {
'content_type': ai_content.content_type,
'ai_model': ai_content.ai_model,
'topic': ai_content.topic,
'target_audience': ai_content.target_audience,
'submission_time': datetime.now().isoformat(),
**ai_content.metadata
}
}
try:
response = requests.post(
f'{self.base_url}/requests',
headers=self.headers,
json=request_data,
timeout=30
)
response.raise_for_status()
result = response.json()
logging.info(f"Quality review request created: {result['data']['request']['id']}")
return result['data']['request']
except requests.exceptions.RequestException as e:
logging.error(f"Failed to create quality review: {e}")
return None
def _get_article_review_config(self) -> Dict:
return {
'response_type': 'multi_select',
'response_config': {
'options': [
'✅ Publish as-is - High quality, ready for publication',
'📝 Minor Edits - Good content, needs small improvements',
'🔄 Major Revision - Significant issues, needs rework',
'❌ Not Suitable - Poor quality, consider regeneration',
'✂️ Factual Errors - Contains inaccurate information',
'🎯 Off-Topic - Doesn\'t match intended topic/audience',
'📖 Poor Structure - Needs better organization',
'🔤 Language Issues - Grammar, style, or tone problems'
],
'min_selections': 1,
'max_selections': 4
}
}
def _get_code_review_config(self) -> Dict:
return {
'response_type': 'multi_select',
'response_config': {
'options': [
'✅ Code Quality Excellent - Ready to use',
'🐛 Bug Risk - Potential issues identified',
'🔒 Security Concerns - Security vulnerabilities found',
'⚡ Performance Issues - Inefficient implementation',
'📚 Missing Documentation - Needs better comments/docs',
'🎨 Style Issues - Doesn\'t follow coding standards',
'🧪 Needs Tests - Missing or insufficient test coverage',
'♻️ Refactoring Needed - Code structure improvements required'
],
'min_selections': 1,
'max_selections': 5
}
}
def _get_creative_review_config(self) -> Dict:
return {
'response_type': 'rating',
'response_config': {
'min_value': 1,
'max_value': 5,
'labels': {
'1': 'Poor - Major issues, not usable',
'2': 'Below Average - Significant problems',
'3': 'Average - Acceptable with improvements',
'4': 'Good - Minor tweaks needed',
'5': 'Excellent - Ready to publish'
}
}
}
def _get_general_review_config(self) -> Dict:
return {
'response_type': 'text',
'response_config': {
'min_length': 100,
'max_length': 1000,
'placeholder': 'Provide detailed quality assessment and improvement suggestions...',
'guidelines': 'Please evaluate accuracy, clarity, usefulness, and overall quality. Provide specific suggestions for improvement.'
}
}
async def batch_review_content(self, content_list: List[AIContent]) -> List[Optional[Dict]]:
"""Submit multiple pieces of content for review concurrently"""
async def submit_single_review(session, content):
request_data = {
'loop_id': self.review_loop_id,
'request_text': f'''Quality Review Request
Content Type: {content.content_type}
AI Model: {content.ai_model}
Topic: {content.topic}
Content: {content.content[:500]}{'...' if len(content.content) > 500 else ''}''',
'response_type': 'rating',
'response_config': {
'min_value': 1,
'max_value': 5,
'labels': {'1': 'Poor', '3': 'Average', '5': 'Excellent'}
},
'priority': 'medium',
'metadata': {
'content_type': content.content_type,
'ai_model': content.ai_model,
'batch_submission': True
}
}
try:
async with session.post(
f'{self.base_url}/requests',
json=request_data,
headers=self.headers
) as response:
if response.status == 201:
result = await response.json()
return result['data']['request']
else:
logging.error(f"Failed to submit review: {response.status}")
return None
except Exception as e:
logging.error(f"Error submitting review: {e}")
return None
async with aiohttp.ClientSession() as session:
tasks = [submit_single_review(session, content) for content in content_list]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out exceptions and return successful submissions
return [r for r in results if isinstance(r, dict)]
def get_review_results(self, request_ids: List[str]) -> List[Dict]:
"""Get results for multiple review requests"""
results = []
for request_id in request_ids:
try:
response = requests.get(
f'{self.base_url}/requests/{request_id}',
headers=self.headers,
timeout=10
)
response.raise_for_status()
request_data = response.json()['data']['request']
if request_data['status'] == 'completed':
results.append({
'request_id': request_id,
'status': 'completed',
'response_data': request_data['response_data'],
'response_time': request_data.get('response_time_seconds'),
'metadata': request_data.get('metadata', {})
})
else:
results.append({
'request_id': request_id,
'status': request_data['status'],
'response_data': None
})
except Exception as e:
logging.error(f"Error getting results for {request_id}: {e}")
results.append({
'request_id': request_id,
'status': 'error',
'error': str(e)
})
return results
# Example usage and integration
def main():
# Initialize the review system
api_key = "your_hitl_api_key_here"
review_loop_id = "your_review_loop_id_here"
qa_system = AIQualityReviewSystem(api_key, review_loop_id)
# Example AI-generated content
ai_article = AIContent(
content="""
Machine Learning in Healthcare: Transforming Patient Care
Artificial intelligence is revolutionizing healthcare by enabling more accurate diagnoses,
personalized treatment plans, and efficient hospital operations. Recent advances in deep
learning have shown promising results in medical imaging, drug discovery, and patient
monitoring systems.
Key benefits include:
1. Early disease detection through pattern recognition
2. Reduced diagnostic errors
3. Optimized treatment protocols
4. Enhanced patient outcomes
However, challenges remain in data privacy, algorithm bias, and regulatory compliance.
Healthcare organizations must carefully balance innovation with patient safety and ethical considerations.
""",
content_type='article',
ai_model='GPT-4',
topic='Healthcare Technology',
target_audience='Healthcare Professionals',
metadata={
'word_count': 150,
'reading_level': 'professional',
'intended_publication': 'medical_journal'
}
)
# Submit for review
print("Submitting article for quality review...")
review_request = qa_system.create_quality_review(ai_article, priority='high')
if review_request:
print(f"Review request created: {review_request['id']}")
print(f"Status: {review_request['status']}")
print(f"Timeout: {review_request.get('timeout_at')}")
else:
print("Failed to create review request")
# Example batch processing
print("\nTesting batch review...")
content_batch = [
AIContent("Sample code snippet...", "code", "Codex", "Python", "developers", {}),
AIContent("Creative story opening...", "creative", "GPT-4", "Fiction", "general", {}),
AIContent("Technical documentation...", "article", "GPT-4", "API Docs", "developers", {})
]
# Run batch review asynchronously
batch_results = asyncio.run(qa_system.batch_review_content(content_batch))
print(f"Batch submitted: {len(batch_results)} reviews created")
if __name__ == "__main__":
main()
Data Verification Service
Verify user-submitted information with human reviewers.- Go
package main
import (
"bytes"
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"os"
"time"
)
// DataVerificationService handles verification of user-submitted data
type DataVerificationService struct {
APIKey string
BaseURL string
LoopID string
HTTPClient *http.Client
}
// BusinessData represents business information to be verified
type BusinessData struct {
Name string `json:"name"`
Address string `json:"address"`
Phone string `json:"phone"`
Email string `json:"email"`
Website string `json:"website"`
Description string `json:"description"`
Category string `json:"category"`
}
// VerificationRequest represents a HITL verification request
type VerificationRequest struct {
LoopID string `json:"loop_id"`
RequestText string `json:"request_text"`
ResponseType string `json:"response_type"`
ResponseConfig interface{} `json:"response_config"`
Priority string `json:"priority"`
Metadata interface{} `json:"metadata"`
}
// VerificationResponse represents the API response
type VerificationResponse struct {
Error bool `json:"error"`
Msg string `json:"msg"`
Data struct {
Request struct {
ID string `json:"id"`
Status string `json:"status"`
} `json:"request"`
} `json:"data"`
}
// NewDataVerificationService creates a new verification service
func NewDataVerificationService(apiKey, loopID string) *DataVerificationService {
return &DataVerificationService{
APIKey: apiKey,
BaseURL: "https://api.hitl.sh/v1",
LoopID: loopID,
HTTPClient: &http.Client{
Timeout: 30 * time.Second,
},
}
}
// VerifyBusinessData submits business data for human verification
func (dvs *DataVerificationService) VerifyBusinessData(business BusinessData, priority string) (*VerificationResponse, error) {
requestText := fmt.Sprintf(`Business Data Verification Request
Please verify the following business information for accuracy:
Business Name: %s
Address: %s
Phone Number: %s
Email: %s
Website: %s
Category: %s
Description: %s
Verification Tasks:
1. Confirm the business name matches official records
2. Verify the address exists and is correct
3. Check phone number format and validity
4. Validate email address format
5. Verify website is active and belongs to business
6. Confirm business category is appropriate
7. Review description for accuracy and appropriateness
Please select all items that you have successfully verified.`,
business.Name,
business.Address,
business.Phone,
business.Email,
business.Website,
business.Category,
business.Description,
)
responseConfig := map[string]interface{}{
"options": []string{
"✅ Business Name Verified",
"🏢 Address Verified",
"📞 Phone Number Verified",
"📧 Email Address Verified",
"🌐 Website Verified",
"🏷️ Category Appropriate",
"📝 Description Accurate",
"⚠️ Minor Issues Found",
"❌ Significant Problems Detected",
},
"min_selections": 1,
"max_selections": 9,
}
metadata := map[string]interface{}{
"verification_type": "business_data",
"business_name": business.Name,
"submitted_at": time.Now().Format(time.RFC3339),
"data_fields": map[string]bool{
"name": business.Name != "",
"address": business.Address != "",
"phone": business.Phone != "",
"email": business.Email != "",
"website": business.Website != "",
"description": business.Description != "",
},
}
request := VerificationRequest{
LoopID: dvs.LoopID,
RequestText: requestText,
ResponseType: "multi_select",
ResponseConfig: responseConfig,
Priority: priority,
Metadata: metadata,
}
return dvs.submitRequest(request)
}
// VerifyUserProfile verifies user profile information
func (dvs *DataVerificationService) VerifyUserProfile(userID, profileData string, priority string) (*VerificationResponse, error) {
requestText := fmt.Sprintf(`User Profile Verification
User ID: %s
Profile Data:
%s
Please review this user profile for:
1. Appropriate content (no offensive material)
2. Authentic information (not obviously fake)
3. Complete profile information
4. Professional presentation (if business profile)
5. Compliance with community guidelines
Rate the overall profile quality and identify any issues.`,
userID,
profileData,
)
responseConfig := map[string]interface{}{
"scale_max": 5,
// scale_min defaults to 1
// scale_step defaults to 1
}
metadata := map[string]interface{}{
"verification_type": "user_profile",
"user_id": userID,
"submitted_at": time.Now().Format(time.RFC3339),
}
request := VerificationRequest{
LoopID: dvs.LoopID,
RequestText: requestText,
ResponseType: "rating",
ResponseConfig: responseConfig,
Priority: priority,
Metadata: metadata,
}
return dvs.submitRequest(request)
}
// VerifyDocumentContent verifies uploaded document content
func (dvs *DataVerificationService) VerifyDocumentContent(documentType, content string, priority string) (*VerificationResponse, error) {
requestText := fmt.Sprintf(`Document Verification Request
Document Type: %s
Content:
%s
Please verify this document for:
1. Authenticity and legitimacy
2. Complete required information
3. Proper formatting and structure
4. Compliance with regulations (if applicable)
5. No fraudulent or suspicious content
Select the appropriate verification status.`,
documentType,
content,
)
responseConfig := map[string]interface{}{
"options": []string{
"✅ Verified - Document is authentic and complete",
"⚠️ Verified with Notes - Minor issues noted",
"🔍 Needs Review - Requires additional verification",
"❌ Rejected - Document has serious issues",
"🚨 Fraudulent - Suspicious or fake document",
},
}
metadata := map[string]interface{}{
"verification_type": "document",
"document_type": documentType,
"submitted_at": time.Now().Format(time.RFC3339),
"content_length": len(content),
}
request := VerificationRequest{
LoopID: dvs.LoopID,
RequestText: requestText,
ResponseType: "single_select",
ResponseConfig: responseConfig,
Priority: priority,
Metadata: metadata,
}
return dvs.submitRequest(request)
}
// submitRequest sends a verification request to HITL API
func (dvs *DataVerificationService) submitRequest(req VerificationRequest) (*VerificationResponse, error) {
jsonData, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
}
httpReq, err := http.NewRequest("POST", dvs.BaseURL+"/requests", bytes.NewBuffer(jsonData))
if err != nil {
return nil, fmt.Errorf("failed to create HTTP request: %w", err)
}
httpReq.Header.Set("Authorization", "Bearer "+dvs.APIKey)
httpReq.Header.Set("Content-Type", "application/json")
resp, err := dvs.HTTPClient.Do(httpReq)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
}
var verificationResp VerificationResponse
if err := json.Unmarshal(body, &verificationResp); err != nil {
return nil, fmt.Errorf("failed to unmarshal response: %w", err)
}
if resp.StatusCode != http.StatusCreated {
return &verificationResp, fmt.Errorf("API error: %s", verificationResp.Msg)
}
return &verificationResp, nil
}
// GetVerificationResult checks the status of a verification request
func (dvs *DataVerificationService) GetVerificationResult(requestID string) (map[string]interface{}, error) {
req, err := http.NewRequest("GET", dvs.BaseURL+"/requests/"+requestID, nil)
if err != nil {
return nil, fmt.Errorf("failed to create request: %w", err)
}
req.Header.Set("Authorization", "Bearer "+dvs.APIKey)
resp, err := dvs.HTTPClient.Do(req)
if err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}
defer resp.Body.Close()
var result map[string]interface{}
if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
return nil, fmt.Errorf("failed to decode response: %w", err)
}
return result, nil
}
// BatchVerifyBusinesses verifies multiple businesses concurrently
func (dvs *DataVerificationService) BatchVerifyBusinesses(businesses []BusinessData) []chan *VerificationResponse {
results := make([]chan *VerificationResponse, len(businesses))
for i, business := range businesses {
results[i] = make(chan *VerificationResponse, 1)
go func(idx int, biz BusinessData) {
defer close(results[idx])
result, err := dvs.VerifyBusinessData(biz, "medium")
if err != nil {
log.Printf("Failed to verify business %s: %v", biz.Name, err)
results[idx] <- &VerificationResponse{Error: true, Msg: err.Error()}
} else {
results[idx] <- result
}
}(i, business)
// Rate limiting - don't send too many requests at once
time.Sleep(200 * time.Millisecond)
}
return results
}
// Example HTTP server for verification API
func main() {
apiKey := os.Getenv("HITL_API_KEY")
loopID := os.Getenv("VERIFICATION_LOOP_ID")
if apiKey == "" || loopID == "" {
log.Fatal("Please set HITL_API_KEY and VERIFICATION_LOOP_ID environment variables")
}
service := NewDataVerificationService(apiKey, loopID)
http.HandleFunc("/verify/business", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
var business BusinessData
if err := json.NewDecoder(r.Body).Decode(&business); err != nil {
http.Error(w, "Invalid JSON", http.StatusBadRequest)
return
}
result, err := service.VerifyBusinessData(business, "medium")
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(result)
})
http.HandleFunc("/verify/profile", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
var req struct {
UserID string `json:"user_id"`
ProfileData string `json:"profile_data"`
Priority string `json:"priority"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "Invalid JSON", http.StatusBadRequest)
return
}
if req.Priority == "" {
req.Priority = "medium"
}
result, err := service.VerifyUserProfile(req.UserID, req.ProfileData, req.Priority)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(result)
})
http.HandleFunc("/verify/document", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
var req struct {
DocumentType string `json:"document_type"`
Content string `json:"content"`
Priority string `json:"priority"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "Invalid JSON", http.StatusBadRequest)
return
}
if req.Priority == "" {
req.Priority = "medium"
}
result, err := service.VerifyDocumentContent(req.DocumentType, req.Content, req.Priority)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(result)
})
http.HandleFunc("/result/", func(w http.ResponseWriter, r *http.Request) {
requestID := r.URL.Path[len("/result/"):]
if requestID == "" {
http.Error(w, "Missing request ID", http.StatusBadRequest)
return
}
result, err := service.GetVerificationResult(requestID)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(result)
})
// Example usage
http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html")
w.Write([]byte(`
<!DOCTYPE html>
<html>
<head>
<title>Data Verification Service</title>
</head>
<body>
<h1>HITL Data Verification Service</h1>
<h2>Business Verification</h2>
<form id="businessForm">
<input type="text" placeholder="Business Name" name="name" required><br>
<input type="text" placeholder="Address" name="address" required><br>
<input type="text" placeholder="Phone" name="phone"><br>
<input type="email" placeholder="Email" name="email"><br>
<input type="url" placeholder="Website" name="website"><br>
<input type="text" placeholder="Category" name="category"><br>
<textarea placeholder="Description" name="description"></textarea><br>
<button type="submit">Verify Business</button>
</form>
<script>
document.getElementById('businessForm').onsubmit = function(e) {
e.preventDefault();
const formData = new FormData(e.target);
const data = Object.fromEntries(formData);
fetch('/verify/business', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify(data)
})
.then(response => response.json())
.then(result => {
alert('Verification request submitted: ' + result.data.request.id);
})
.catch(error => console.error('Error:', error));
};
</script>
</body>
</html>
`))
})
port := os.Getenv("PORT")
if port == "" {
port = "8080"
}
log.Printf("Data verification service running on port %s", port)
log.Fatal(http.ListenAndServe(":"+port, nil))
}
# .env file
HITL_API_KEY=your_hitl_api_key_here
VERIFICATION_LOOP_ID=your_loop_id_here
PORT=8080
# Build and run
go mod init hitl-verification
go mod tidy
go run main.go
Customer Support Escalation Bot
Automatically escalate complex customer issues to human agents.- Python
"""
Customer Support Escalation System
Automatically escalate complex support tickets to human agents via HITL
"""
import requests
import json
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime
import re
@dataclass
class SupportTicket:
ticket_id: str
customer_email: str
subject: str
message: str
category: str
priority: str
created_at: datetime
metadata: Dict
class SupportEscalationSystem:
def __init__(self, api_key: str, escalation_loop_id: str):
self.api_key = api_key
self.escalation_loop_id = escalation_loop_id
self.base_url = 'https://api.hitl.sh/v1'
self.headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
}
# Define escalation triggers
self.escalation_keywords = [
'angry', 'furious', 'unacceptable', 'terrible', 'worst',
'lawsuit', 'legal action', 'attorney', 'refund', 'cancel',
'disappointed', 'frustrated', 'urgent', 'emergency'
]
self.technical_keywords = [
'api', 'integration', 'webhook', 'database', 'server',
'bug', 'error', 'crash', 'down', 'not working'
]
def should_escalate(self, ticket: SupportTicket) -> tuple[bool, str]:
"""Determine if ticket should be escalated and why"""
escalation_reasons = []
# Check priority
if ticket.priority in ['high', 'critical']:
escalation_reasons.append('High priority ticket')
# Check for angry customer language
message_lower = ticket.message.lower()
found_keywords = [kw for kw in self.escalation_keywords if kw in message_lower]
if found_keywords:
escalation_reasons.append(f'Customer sentiment indicators: {", ".join(found_keywords)}')
# Check for technical issues
tech_keywords = [kw for kw in self.technical_keywords if kw in message_lower]
if tech_keywords:
escalation_reasons.append(f'Technical issue detected: {", ".join(tech_keywords)}')
# Check for length (complex issues are often longer)
if len(ticket.message) > 1000:
escalation_reasons.append('Detailed/complex issue (long message)')
# Check for multiple question marks (indicates confusion/frustration)
if message_lower.count('?') >= 3:
escalation_reasons.append('Multiple questions indicate confusion')
# Check for CAPS (indicates shouting/anger)
caps_words = re.findall(r'\b[A-Z]{3,}\b', ticket.message)
if len(caps_words) >= 2:
escalation_reasons.append('Customer appears frustrated (excessive caps)')
should_escalate = len(escalation_reasons) > 0
reason = '; '.join(escalation_reasons) if escalation_reasons else ''
return should_escalate, reason
def escalate_ticket(self, ticket: SupportTicket, escalation_reason: str) -> Optional[Dict]:
"""Escalate support ticket to human agent"""
request_text = f'''Customer Support Escalation
Ticket ID: {ticket.ticket_id}
Customer: {ticket.customer_email}
Subject: {ticket.subject}
Priority: {ticket.priority}
Category: {ticket.category}
Created: {ticket.created_at.strftime('%Y-%m-%d %H:%M:%S')}
Escalation Reason: {escalation_reason}
Customer Message:
{ticket.message}
Previous Interactions:
{self._format_ticket_history(ticket)}
Please provide a comprehensive response that addresses the customer's concerns and resolves their issue.'''
# Determine response type based on ticket category
if ticket.category == 'technical':
response_config = self._get_technical_response_config()
elif ticket.category == 'billing':
response_config = self._get_billing_response_config()
elif ticket.category == 'complaint':
response_config = self._get_complaint_response_config()
else:
response_config = self._get_general_response_config()
request_data = {
'loop_id': self.escalation_loop_id,
'request_text': request_text,
**response_config,
'priority': self._map_priority(ticket.priority),
'processing_type': 'time-sensitive',
'timeout_minutes': 60 if ticket.priority == 'critical' else 120,
'metadata': {
'ticket_id': ticket.ticket_id,
'customer_email': ticket.customer_email,
'category': ticket.category,
'escalation_reason': escalation_reason,
'escalated_at': datetime.now().isoformat(),
'original_priority': ticket.priority
}
}
try:
response = requests.post(
f'{self.base_url}/requests',
headers=self.headers,
json=request_data,
timeout=30
)
response.raise_for_status()
result = response.json()
return result['data']['request']
except Exception as e:
print(f"Failed to escalate ticket {ticket.ticket_id}: {e}")
return None
def _get_technical_response_config(self) -> Dict:
return {
'response_type': 'text',
'response_config': {
'min_length': 150,
'max_length': 1500,
'placeholder': 'Provide detailed technical solution and next steps...',
'guidelines': 'Include specific troubleshooting steps, code examples if applicable, and timeline for resolution.'
}
}
def _get_billing_response_config(self) -> Dict:
return {
'response_type': 'single_select',
'response_config': {
'options': [
'✅ Issue Resolved - Provided solution to customer',
'💳 Refund Approved - Processing refund as requested',
'🔄 Account Adjustment - Made billing correction',
'📞 Escalate to Manager - Requires management approval',
'📋 More Info Needed - Requested additional details',
'❌ No Action - Issue cannot be resolved as requested'
]
}
}
def _get_complaint_response_config(self) -> Dict:
return {
'response_type': 'multi_select',
'response_config': {
'options': [
'🤝 Apologized to Customer',
'🔧 Provided Solution/Workaround',
'💰 Offered Compensation/Credit',
'📞 Scheduled Follow-up Call',
'📈 Escalated to Management',
'📝 Created Internal Bug Report',
'🎓 Provided Additional Training/Resources',
'✅ Customer Satisfied with Resolution'
],
'min_selections': 1,
'max_selections': 5
}
}
def _get_general_response_config(self) -> Dict:
return {
'response_type': 'text',
'response_config': {
'min_length': 100,
'max_length': 1000,
'placeholder': 'Provide comprehensive response to customer...',
'guidelines': 'Be empathetic, provide clear solutions, and ensure customer satisfaction.'
}
}
def _map_priority(self, original_priority: str) -> str:
"""Map ticket priority to HITL request priority"""
priority_map = {
'low': 'medium', # Still important since it's escalated
'medium': 'high', # Escalated medium becomes high
'high': 'critical', # High priority tickets are critical
'critical': 'critical'
}
return priority_map.get(original_priority, 'medium')
def _format_ticket_history(self, ticket: SupportTicket) -> str:
"""Format ticket history from metadata"""
history = ticket.metadata.get('history', [])
if not history:
return "No previous interactions"
formatted_history = []
for interaction in history[-3:]: # Last 3 interactions
formatted_history.append(
f"- {interaction.get('timestamp', 'Unknown time')}: {interaction.get('type', 'Unknown')} - {interaction.get('summary', 'No details')}"
)
return '\n'.join(formatted_history) if formatted_history else "No previous interactions"
def process_support_queue(self, tickets: List[SupportTicket]) -> Dict:
"""Process a queue of support tickets and escalate as needed"""
results = {
'processed': 0,
'escalated': 0,
'not_escalated': 0,
'errors': 0,
'escalations': []
}
for ticket in tickets:
results['processed'] += 1
try:
should_escalate, reason = self.should_escalate(ticket)
if should_escalate:
escalation_result = self.escalate_ticket(ticket, reason)
if escalation_result:
results['escalated'] += 1
results['escalations'].append({
'ticket_id': ticket.ticket_id,
'request_id': escalation_result['id'],
'reason': reason,
'priority': escalation_result['priority']
})
else:
results['errors'] += 1
else:
results['not_escalated'] += 1
print(f"Ticket {ticket.ticket_id} does not require escalation")
except Exception as e:
print(f"Error processing ticket {ticket.ticket_id}: {e}")
results['errors'] += 1
return results
# Example Flask webhook handler for processing escalation responses
from flask import Flask, request, jsonify
import hmac
import hashlib
app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret_here"
escalation_system = SupportEscalationSystem("your_api_key", "your_loop_id")
def verify_webhook_signature(payload, signature, secret):
expected_signature = hmac.new(
secret.encode('utf-8'),
payload.encode('utf-8'),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f'sha256={expected_signature}', signature)
@app.route('/webhook/support-escalation', methods=['POST'])
def handle_escalation_response():
# Verify signature
payload = request.get_data(as_text=True)
signature = request.headers.get('X-HITL-Signature-256')
if not verify_webhook_signature(payload, signature, WEBHOOK_SECRET):
return jsonify({'error': 'Invalid signature'}), 401
webhook_data = json.loads(payload)
if webhook_data['event'] == 'request.completed':
data = webhook_data['data']
request_data = data['request']
reviewer = data['reviewer']
# Get ticket information from metadata
metadata = request_data.get('metadata', {})
ticket_id = metadata.get('ticket_id')
customer_email = metadata.get('customer_email')
response_data = request_data['response_data']
print(f"Support escalation completed for ticket {ticket_id}")
print(f"Reviewer: {reviewer['email']}")
print(f"Response: {response_data}")
# Send response back to customer
send_customer_response(ticket_id, customer_email, response_data, reviewer['email'])
# Update internal ticketing system
update_ticket_status(ticket_id, 'resolved', response_data)
return jsonify({'status': 'success'})
def send_customer_response(ticket_id, customer_email, response, agent_name):
"""Send the agent's response back to the customer"""
print(f"Sending response to {customer_email} for ticket {ticket_id}")
print(f"Agent: {agent_name}")
print(f"Response: {response}")
# Implement your email sending logic here
# This could integrate with SendGrid, Mailgun, or your existing email system
def update_ticket_status(ticket_id, status, resolution):
"""Update the ticket in your internal system"""
print(f"Updating ticket {ticket_id} status to {status}")
print(f"Resolution: {resolution}")
# Implement your ticketing system update logic here
# This could integrate with Zendesk, Freshdesk, or your custom system
# Example usage
if __name__ == "__main__":
# Example support tickets
sample_tickets = [
SupportTicket(
ticket_id="SUPP-001",
customer_email="angry.customer@example.com",
subject="Your API is TERRIBLE and doesn't work!",
message="I've been trying to integrate with your API for WEEKS and nothing works! This is unacceptable! I want a refund immediately or I'm contacting my lawyer!",
category="technical",
priority="high",
created_at=datetime.now(),
metadata={"history": []}
),
SupportTicket(
ticket_id="SUPP-002",
customer_email="confused.user@example.com",
subject="How do I set up webhooks?",
message="Hi, I'm trying to set up webhooks but I'm not sure how to do it. Can you help?",
category="technical",
priority="medium",
created_at=datetime.now(),
metadata={"history": []}
)
]
# Process tickets
results = escalation_system.process_support_queue(sample_tickets)
print(f"Processing results: {results}")
# Start webhook server
app.run(debug=True, port=5000)