For AI Engineers: Learn how to create custom AI services that integrate seamlessly with the credit system and provide excellent user experiences.
🚀 Quick Start: Build and deploy an AI service in under 30 minutes!
Step 1: Create Your Service Class
All AI services inherit from BaseService
and use the @register_service
decorator:
Text Sentiment Analysis Service
from services.base import BaseService
from services.decorators import register_service
from django.contrib.auth import get_user_model
User = get_user_model()
@register_service("text_sentiment_analysis")
class TextSentimentAnalysisService(BaseService):
"""Advanced text sentiment analysis service."""
def execute_service(self, user: User, text: str = "", **kwargs):
"""Analyze sentiment of text using AI algorithms."""
if not text or not text.strip():
raise ValueError("Text input is required and cannot be empty")
# AI sentiment analysis logic here
positive_words = {'excellent', 'amazing', 'wonderful', 'great', 'love'}
negative_words = {'terrible', 'awful', 'horrible', 'bad', 'hate'}
words = text.lower().split()
total_words = len(words)
positive_count = sum(1 for word in words if word in positive_words)
negative_count = sum(1 for word in words if word in negative_words)
sentiment_score = (positive_count - negative_count) / max(total_words, 1)
if sentiment_score > 0.1:
sentiment_label = "positive"
elif sentiment_score < -0.1:
sentiment_label = "negative"
else:
sentiment_label = "neutral"
confidence = min(0.9, max(0.3, (positive_count + negative_count) / max(total_words, 1) * 2))
return {
'sentiment': {
'label': sentiment_label,
'score': round(sentiment_score, 3),
'confidence': round(confidence, 3)
},
'analysis': {
'total_words': total_words,
'positive_words_found': positive_count,
'negative_words_found': negative_count
},
'metadata': {
'service_name': 'text_sentiment_analysis',
'version': '1.0'
}
}
Keyword Extraction Service
@register_service("text_keyword_extractor")
class TextKeywordExtractorService(BaseService):
"""Extract important keywords from text."""
def execute_service(self, user: User, text: str = "", max_keywords: int = 10, **kwargs):
"""Extract keywords using frequency analysis."""
if not text or not text.strip():
raise ValueError("Text input is required and cannot be empty")
# Stop words to filter out
stop_words = {
'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of',
'with', 'by', 'is', 'are', 'was', 'were', 'this', 'that', 'it', 'they'
}
# Process and filter text
words = text.lower().replace('.', '').replace(',', '').split()
filtered_words = [word for word in words if word not in stop_words and len(word) > 2]
# Count frequencies
word_freq = {}
for word in filtered_words:
word_freq[word] = word_freq.get(word, 0) + 1
# Get top keywords
top_keywords = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:max_keywords]
return {
'keywords': [
{
'word': word,
'frequency': freq,
'relevance_score': round(freq / len(filtered_words), 3)
}
for word, freq in top_keywords
],
'analysis': {
'total_words': len(words),
'unique_words': len(set(words)),
'filtered_words': len(filtered_words)
},
'metadata': {
'service_name': 'text_keyword_extractor',
'max_keywords_requested': max_keywords,
'keywords_found': len(top_keywords)
}
}
Image Analysis Service
@register_service("image_metadata_extractor")
class ImageMetadataExtractorService(BaseService):
"""Extract metadata from image data."""
def execute_service(self, user: User, image_data=None, **kwargs):
"""Extract metadata from image."""
if not image_data:
raise ValueError("Image data is required for processing")
# Analyze image data
data_size = len(image_data)
# Detect format from headers
if isinstance(image_data, bytes):
if image_data.startswith(b'\xFF\xD8\xFF'):
format_hint = 'jpeg'
elif image_data.startswith(b'\x89PNG\r\n\x1a\n'):
format_hint = 'png'
elif image_data.startswith(b'GIF8'):
format_hint = 'gif'
else:
format_hint = 'unknown'
else:
format_hint = 'base64_encoded'
# Estimate size category
if data_size < 50000:
size_category = 'small'
estimated_dimensions = '400x300'
elif data_size < 500000:
size_category = 'medium'
estimated_dimensions = '800x600'
else:
size_category = 'large'
estimated_dimensions = '1920x1080'
return {
'metadata': {
'format': format_hint,
'file_size_bytes': data_size,
'size_category': size_category,
'estimated_dimensions': estimated_dimensions
},
'analysis': {
'compression_ratio': round(data_size / 1024, 2),
'quality_estimate': 'high' if data_size > 100000 else 'medium'
}
}
External API Integration Service
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
@register_service("external_ai_processor")
class ExternalAIProcessorService(BaseService):
"""Service that integrates with external AI APIs."""
def __init__(self, service_name: str):
super().__init__(service_name)
# Configure session with retries
self.session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("http://", adapter)
self.session.mount("https://", adapter)
def execute_service(self, user: User, prompt: str = "", **kwargs):
"""Process data using external AI service."""
if not prompt:
raise ValueError("Prompt is required for AI processing")
try:
response = self.session.post(
"https://api.openai.com/v1/completions", # Example API
headers={
"Authorization": f"Bearer {settings.OPENAI_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": "text-davinci-003",
"prompt": prompt,
"max_tokens": 150,
"temperature": 0.7
},
timeout=30
)
response.raise_for_status()
result = response.json()
return {
'response': result.get('choices', [{}])[0].get('text', '').strip(),
'usage': result.get('usage', {}),
'model': result.get('model', 'text-davinci-003'),
'metadata': {
'external_api': 'openai',
'prompt_length': len(prompt),
'response_time_ms': response.elapsed.total_seconds() * 1000
}
}
except requests.exceptions.RequestException as e:
raise RuntimeError(f"External AI service error: {str(e)}")
except Exception as e:
raise RuntimeError(f"Processing failed: {str(e)}")
Step 2: Register Your Service in Database
Create a Service record in the Django admin or using a management command:
# In Django shell or management command
from credits.models import Service
Service.objects.create(
name="text_sentiment_analysis", # Must match @register_service name
description="Advanced AI-powered sentiment analysis for text content",
credit_cost=1.0, # How many credits this service costs
is_active=True
)
Step 3: Use Your Service
Once registered, your service is automatically available through the API and web interface:
# Using the service programmatically
from services.decorators import create_service_instance
service = create_service_instance("text_sentiment_analysis")
result = service.run(user, text="This product is amazing!")
# API usage (cURL)
curl -X POST "https://api.{{ request.get_host }}/v1/services/1/use" \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Content-Type: application/json" \
-d '{"input_data": {"text": "This product is amazing!"}}'
🎉 That's it! Your AI service is now live and integrated with the credit system, user management, and API.
Best Practices for AI Services
Input Validation
- Always validate inputs early with clear error messages
- Set reasonable limits (text length, file size, etc.)
- Use type hints for better code clarity
Error Handling
- Handle external API failures gracefully
- Don't expose internal implementation details
- Log errors for debugging while returning user-friendly messages
Performance Optimization
- Cache expensive operations using Django's cache framework
- Use connection pooling for external APIs
- Consider async processing for long-running tasks
Credit Cost Guidelines
- Simple operations (keyword extraction): 0.5-1.0 credits
- Text analysis (sentiment, classification): 1.0-2.0 credits
- Image processing: 2.0-5.0 credits
- External AI API calls: 5.0-10.0 credits