> ## Documentation Index
> Fetch the complete documentation index at: https://docs.hitl.sh/llms.txt
> Use this file to discover all available pages before exploring further.
# Code Templates & Examples
> Production-ready code templates and boilerplate examples for integrating HITL.sh into your applications. Copy, customize, and deploy quickly.
Get started faster with production-ready code templates for common HITL integration patterns. These templates include error handling, best practices, and can be customized for your specific use case.
> [!NOTE]
>
> 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.
> [tab] Python (Flask)
>
> ```python theme={null}
> """
> 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)
> ```
>
> **Environment Setup:**
>
> ```bash theme={null}
> # .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
> ```
>
> **Dashboard Template (templates/dashboard.html):**
>
> ```html theme={null}
>
>
>
> Content Moderation Dashboard
>
>
>
> Content Moderation Dashboard
>
> {% if stats %}
>
>
>
{{ stats.total_requests }}
>
Total Requests
>
>
>
{{ stats.pending }}
>
Pending Review
>
>
>
{{ stats.completed }}
>
Completed
>
>
>
{{ "%.1f"|format(stats.approval_rate) }}%
>
Approval Rate
>
>
Unable to load statistics
> {% endif %}
>
> Test Content Moderation
>
>
>
>
>
> ```
> [tab] Node.js (Express)
>
> ```javascript theme={null}
> /**
> * 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(`
>
>
>
> Content Moderation Dashboard
>
>
>
> Content Moderation Dashboard
>
> ${stats ? `
>
>
>
${stats.total_requests}
>
Total Requests
>
>
>
${stats.pending}
>
Pending Review
>
>
>
${stats.completed}
>
Completed
>
>
>
${stats.approval_rate.toFixed(1)}%
>
Approval Rate
>
>
Unable to load statistics
'}
>
> Test Content Moderation
>
>
>
>
>
> `);
> });
>
> 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`);
> });
> ```
>
> **Package.json:**
>
> ```json theme={null}
> {
> "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"
> }
> }
> ```
>
> **Environment Setup:**
>
> ```bash theme={null}
> # .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.
> [tab] Python
>
> ```python theme={null}
> """
> 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.
> [tab] Go
>
> ```go theme={null}
> 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(`
>
>
>
> Data Verification Service
>
>
> HITL Data Verification Service
>
> Business Verification
>
>
>
>
>
> `))
> })
>
> port := os.Getenv("PORT")
> if port == "" {
> port = "8080"
> }
>
> log.Printf("Data verification service running on port %s", port)
> log.Fatal(http.ListenAndServe(":"+port, nil))
> }
> ```
>
> **Environment Setup:**
>
> ```bash theme={null}
> # .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.
> [tab] Python
>
> ```python theme={null}
> """
> 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)
> ```
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.