AjaxAI Documentation

Installation

Install the AjaxAI SDK to start processing AI requests at scale:

pip install ajaxai-sdk

Requirements: Python 3.8+

Quick Start

from ajaxai import create_batch_job, AjaxAiRequestItem

# Create job and add requests
job = create_batch_job(api_key="your_api_key")
request = AjaxAiRequestItem(request_id="req_1").add_text("Hello world")
job.add_request(request)

# Submit and get results
job.submit()
for result in job.get_results(output_model=None):
    print(result.response.text)

Get your API key from the Settings page.

Core Concepts

Jobs

A job is a container for batch processing multiple AI requests efficiently. Jobs handle:

Automatic batching and queue management
Model configuration and system prompts
Retry logic for failed requests
Result aggregation and streaming

Requests

A request is an individual AI task with content and optional metadata. Requests support:

Multiple content types (text, images)
Structured output with Pydantic models
Custom metadata for tracking
Flexible content building with helper methods

Results

Results contain the AI response plus metadata:

Structured response data
Usage metrics (tokens, costs)
Request metadata and summaries
Error information when applicable

API Reference

create_batch_job()

Create a new batch processing job.

create_batch_job(
    api_key: str = None,
    model: str = "gemini-2.0-flash",
    temperature: float = 0.5,
    max_output_tokens: int = 1024,
    system_instruction: str = "",
    job_type: str = "",
    metadata: dict = None,
    max_batch_size: int = 20
) -> AjaxAiBatchJob

Parameters:

api_key: Your AjaxAI API key (or set AJAXAI_API_KEY env var)
model: AI model to use (see Models)
temperature: Creativity level (0.0-1.0)
max_output_tokens: Maximum response length
system_instruction: System prompt for all requests
job_type: Job type for callback routing
metadata: Custom job metadata
max_batch_size: Requests per batch (1-100)

Returns: AjaxAiBatchJob instance

AjaxAiRequestItem()

Create an AI request with content and configuration.

AjaxAiRequestItem(
    request_id: str,
    output_model: Type[BaseModel] = None,
    content_parts: List[Dict] = None
)

Parameters:

request_id: Unique identifier for this request
output_model: Pydantic model for structured output
content_parts: Content list (auto-initialized if None)

Methods:

.add_text(text: str) - Add text content
.add_image(image: str) - Add image (URL, path, or GCS URI)
.add_request_metadata(metadata: BaseModel) - Add tracking metadata

get_batch_job()

Retrieve an existing job by ID.

get_batch_job(api_key: str, job_id: str) -> AjaxAiBatchJob

AjaxAiBatchJob Methods

.add_request()

Add a request to the job queue.

job.add_request(request: AjaxAiRequestItem) -> AddRequestResponse

Requests are automatically batched and sent to the API when the batch size limit is reached or after a timeout.

.submit()

Submit the job for processing.

job.submit() -> SubmitJobResponse

Flushes any pending requests and starts AI processing. Returns immediately - use monitoring to track progress.

.get_state()

Get the current job status.

job.get_state() -> str

Possible states:

created - Job created, no requests added
populating - Adding requests
ready - Ready for submission
processing - Being processed by AI
succeeded - Completed successfully
failed - Processing failed
cancelled - Job cancelled

.get_results()

Stream job results as they become available.

job.get_results(output_model: BaseModel = None) -> Generator[BatchRequestResult]

Parameters:

output_model: Pydantic model for response parsing

Returns: Generator yielding BatchRequestResult objects with:

request: Original request data
response: AI response (structured if output_model provided)
metadata: Request metadata
summary: Request summary with status and metrics
usage: Token usage statistics

.get_metrics()

Get job performance statistics.

job.get_metrics() -> JobPerformanceMetrics

Returns: Metrics object with:

total_requests: Total requests in job
processed_requests: Completed requests
successful_requests: Successful requests
failed_requests: Failed requests
total_tokens: Total tokens used
duration_seconds: Processing time

Models

Gemini Models

gemini-2.0-flash - Fast, general-purpose (default)
gemini-2.0-flash-lite - Lightweight, fastest
gemini-1.5-flash-002 - Balanced speed and quality
gemini-1.5-pro-002 - Advanced reasoning, slower

Claude Models

claude-3.5-sonnet-v2 - Excellent for writing and analysis
claude-3.5-haiku - Fast Claude model

Model Selection Guide

Speed: gemini-2.0-flash-lite > gemini-2.0-flash > claude-3.5-haiku
Quality: gemini-1.5-pro-002 > claude-3.5-sonnet-v2 > gemini-1.5-flash-002
Writing: claude-3.5-sonnet-v2 is best for creative and analytical writing
Reasoning: gemini-1.5-pro-002 excels at complex reasoning tasks

Authentication

API Keys

Get your API key from the Settings page.

Environment Variable (Recommended)

import os
os.environ["AJAXAI_API_KEY"] = "your_api_key"
job = create_batch_job()

Direct Parameter

job = create_batch_job(api_key="your_api_key")

Security Best Practices

Never commit API keys to version control
Use environment variables in production
Rotate keys regularly via the dashboard
Monitor usage for unexpected activity

Error Handling

Exception Types

from ajaxai.classes.exceptions import (
    AjaxAiApiError,           # Base API error
    AjaxAiAuthorizationError, # Invalid API key (401/403)
    AjaxAiRateLimitError,     # Rate limit exceeded (429)
    AjaxAiJobNotFoundError,   # Job not found (404)
    AjaxAiServerError         # Server error (5xx)
)

Error Handling Patterns

Basic Error Handling

try:
    job = create_batch_job(api_key="invalid_key")
    job.submit()
except AjaxAiAuthorizationError:
    print("Invalid API key")
except AjaxAiRateLimitError:
    print("Rate limit exceeded")
except AjaxAiApiError as e:
    print(f"API error: {e}")

Retry Logic

import time
from ajaxai.classes.exceptions import AjaxAiRateLimitError

max_retries = 3
for attempt in range(max_retries):
    try:
        response = job.submit()
        break
    except AjaxAiRateLimitError:
        if attempt == max_retries - 1:
            raise
        wait_time = 2 ** attempt  # Exponential backoff
        time.sleep(wait_time)

Monitoring

Dashboard Monitoring

Monitor jobs visually in the AjaxAI Dashboard:

Real-time status updates
Progress tracking
Error logs and debugging
Performance analytics
Result downloads

Programmatic Monitoring

Automatic Polling

# Start background monitoring
client = AjaxAiClient(api_key="your_api_key")
client.start_polling()

# Submit jobs - callbacks will trigger when complete
job.submit()

# Stop when done
client.stop_polling()

Manual Status Checks

# Check overall progress
status = job.get_state()
metrics = job.get_metrics()
print(f"Status: {status}")
print(f"Progress: {metrics.processed_requests}/{metrics.total_requests}")

# Check individual requests
request_status = job.get_request_state("req_1")

Structured Outputs

Basic Structured Output

from pydantic import BaseModel
from typing import List

class ProductDescription(BaseModel):
    title: str
    description: str
    features: List[str]
    price_range: str

request = AjaxAiRequestItem(
    request_id="product_1",
    output_model=ProductDescription
).add_text("Generate a product description for wireless headphones")

job.add_request(request)
job.submit()

# Get typed responses
for result in job.get_results(output_model=ProductDescription):
    product: ProductDescription = result.response
    print(f"Title: {product.title}")
    print(f"Features: {', '.join(product.features)}")

Advanced Schema Design

Nested Models

class ContactInfo(BaseModel):
    email: str
    phone: Optional[str] = None

class LeadAnalysis(BaseModel):
    score: int  # 1-100
    contact: ContactInfo
    interests: List[str]
    next_action: str

request = AjaxAiRequestItem(
    request_id="lead_1",
    output_model=LeadAnalysis
).add_text("Analyze this lead: John Doe, loves tech gadgets...")

Validation and Constraints

from pydantic import Field, validator

class ReviewAnalysis(BaseModel):
    sentiment: str = Field(..., regex="^(positive|negative|neutral)$")
    rating: int = Field(..., ge=1, le=5)
    summary: str = Field(..., max_length=500)
    
    @validator('rating')
    def rating_must_match_sentiment(cls, v, values):
        sentiment = values.get('sentiment')
        if sentiment == 'positive' and v < 3:
            raise ValueError('Positive sentiment requires rating >= 3')
        return v

Callbacks

Basic Callback Setup

Implement callbacks to receive immediate notifications when requests complete.

from ajaxai import create_batch_job, AjaxAiRequestItem
from ajaxai.classes.events import BatchRequestEvent

def my_callback_function(event: BatchRequestEvent):
    if event.event_type == "request_succeeded":
        print(f"Request {event.request_id} succeeded with response: {event.result.response.text}")
    elif event.event_type == "request_failed":
        print(f"Request {event.request_id} failed: {event.error}")

job = create_batch_job(api_key="your_api_key")
job.on_request_complete(my_callback_function)

# Add requests and submit as usual
request = AjaxAiRequestItem(request_id="cb_req_1").add_text("Callback test")
job.add_request(request)
job.submit()
# No need to call get_results() if using callbacks for real-time processing

Callbacks are asynchronous and run in a separate thread, allowing your main application to continue processing.

Structured Callback Processing

Combine callbacks with structured outputs for robust, type-safe event handling.

from pydantic import BaseModel
from ajaxai import create_batch_job, AjaxAiRequestItem
from ajaxai.classes.events import BatchRequestEvent
from typing import List

class DocumentSummary(BaseModel):
    title: str
    summary_text: str
    keywords: List[str]

def process_document_summary(event: BatchRequestEvent):
    if event.event_type == "request_succeeded" and event.result.response_model is not None:
        summary: DocumentSummary = event.result.response_model
        print(f"Summarized document '{summary.title}': {summary.summary_text}")
    elif event.event_type == "request_failed":
        print(f"Failed to summarize {event.request_id}: {event.error}")

job = create_batch_job(api_key="your_api_key")
job.on_request_complete(process_document_summary)

request = AjaxAiRequestItem(
    request_id="doc_summary_1",
    output_model=DocumentSummary
).add_text("Summarize this long document about AI and machine learning...")
job.add_request(request)
job.submit()

Error Handling in Callbacks

It's crucial to handle exceptions within your callback functions to prevent them from crashing the callback listener.

from ajaxai.classes.events import BatchRequestEvent

def resilient_callback(event: BatchRequestEvent):
    try:
        if event.event_type == "request_succeeded":
            # Simulate a processing error
            if "error_trigger" in event.request.request_id:
                raise ValueError("Simulated processing error in callback")
            print(f"Successfully processed {event.request_id}")
        elif event.event_type == "request_failed":
            print(f"Request {event.request_id} failed: {event.error}")
    except Exception as e:
        print(f"An error occurred in callback for request {event.request_id}: {e}")
        # Log the error to your monitoring system
        # send_error_to_sentry(e, request_id=event.request_id)

job = create_batch_job(api_key="your_api_key")
job.on_request_complete(resilient_callback)

job.add_request(AjaxAiRequestItem(request_id="success_req").add_text("This should succeed"))
job.add_request(AjaxAiRequestItem(request_id="error_trigger_req").add_text("This will trigger a callback error"))
job.submit()

Exceptions raised inside callbacks do not affect the job's overall completion status, but they will be logged by the AjaxAI SDK and should be handled to ensure your application remains stable.

Examples

Content Generation Pipeline

Automate blog post creation, product descriptions, or marketing copy.

from ajaxai import create_batch_job, AjaxAiRequestItem
from pydantic import BaseModel
from typing import List

class BlogPost(BaseModel):
    title: str
    sections: List[str]
    keywords: List[str]

job = create_batch_job(api_key="your_api_key", model="gemini-1.5-pro-002", temperature=0.7)

prompts = [
    "Generate a blog post about the benefits of cloud computing for small businesses.",
    "Create a detailed product description for a smart home security camera.",
    "Write a marketing email promoting a new line of eco-friendly cleaning products."
]

for i, prompt in enumerate(prompts):
    request = AjaxAiRequestItem(request_id=f"content_gen_{i+1}", output_model=BlogPost).add_text(prompt)
    job.add_request(request)

job.submit()

for result in job.get_results(output_model=BlogPost):
    if result.response_model:
        blog_post: BlogPost = result.response_model
        print(f"--- Generated Content for {result.request.request_id} ---")
        print(f"Title: {blog_post.title}")
        print("Sections:")
        for section in blog_post.sections:
            print(f"- {section}")
        print(f"Keywords: {', '.join(blog_post.keywords)}
")
    else:
        print(f"Failed to generate content for {result.request.request_id}: {result.summary.status_message}")

Customer Support Analysis

Process customer inquiries, feedback, and support tickets to extract sentiment, topics, and urgency.

from ajaxai import create_batch_job, AjaxAiRequestItem
from pydantic import BaseModel

class SupportTicketAnalysis(BaseModel):
    sentiment: str # positive, negative, neutral
    topic: str
    urgency_score: int # 1-5, 5 being most urgent
    recommended_action: str

support_tickets = [
    {"id": "T101", "text": "My internet is completely down! I can't work."},
    {"id": "T102", "text": "The new feature is great, but the UI is a bit clunky."},
    {"id": "T103", "text": "I have a question about my billing cycle next month."},
]

job = create_batch_job(api_key="your_api_key", model="claude-3.5-sonnet-v2", temperature=0.3)

for ticket in support_tickets:
    request_text = f"Analyze the following customer support ticket for sentiment, topic, urgency (1-5), and recommended action:

{ticket['text']}"
    request = AjaxAiRequestItem(request_id=ticket['id'], output_model=SupportTicketAnalysis).add_text(request_text)
    job.add_request(request)

job.submit()

for result in job.get_results(output_model=SupportTicketAnalysis):
    if result.response_model:
        analysis: SupportTicketAnalysis = result.response_model
        print(f"--- Analysis for Ticket {result.request.request_id} ---")
        print(f"Sentiment: {analysis.sentiment}")
        print(f"Topic: {analysis.topic}")
        print(f"Urgency: {analysis.urgency_score}")
        print(f"Action: {analysis.recommended_action}
")
    else:
        print(f"Failed to analyze ticket {result.request.request_id}: {result.summary.status_message}")

Product Catalog Enhancement

Automatically generate SEO-friendly descriptions, categorize products, or extract key attributes from raw product data.

from ajaxai import create_batch_job, AjaxAiRequestItem
from pydantic import BaseModel, Field
from typing import List, Optional

class ProductDetails(BaseModel):
    product_name: str
    short_description: str = Field(..., max_length=150)
    long_description: str
    features: List[str]
    category: str
    keywords: List[str]
    
product_data = [
    {"id": "P001", "raw_text": "Wireless earbuds, noise canceling, 24hr battery, Bluetooth 5.2, for gym/travel."},
    {"id": "P002", "raw_text": "Smartwatch with heart rate monitor, GPS, waterproof, compatible with iOS/Android."},
    {"id": "P003", "raw_text": "Ergonomic office chair, lumbar support, adjustable height, mesh back, black color."},
]

job = create_batch_job(api_key="your_api_key", model="gemini-1.5-flash-002", temperature=0.6)

for product in product_data:
    prompt = f"Generate product details including short description (max 150 chars), long description, features, category, and keywords for: {product['raw_text']}"
    request = AjaxAiRequestItem(request_id=product['id'], output_model=ProductDetails).add_text(prompt)
    job.add_request(request)

job.submit()

for result in job.get_results(output_model=ProductDetails):
    if result.response_model:
        details: ProductDetails = result.response_model
        print(f"--- Product Details for {result.request.request_id} ({details.product_name}) ---")
        print(f"Short Desc: {details.short_description}")
        print(f"Category: {details.category}")
        print(f"Features: {', '.join(details.features)}
")
    else:
        print(f"Failed to enhance product {result.request.request_id}: {result.summary.status_message}")

Best Practices

Performance Optimization

Batching: Maximize max_batch_size (up to 100) for fewer API calls and better throughput.
Model Selection: Use "Flash" models for speed when advanced reasoning isn't strictly necessary.
Parallel Jobs: For very large datasets, consider running multiple jobs concurrently (respecting rate limits).
Asynchronous Processing: Leverage callbacks or get_results() streaming for non-blocking operations.

Resource Management

API Keys: Store securely as environment variables, not in code.
Error Handling: Implement robust try-except blocks and retry logic for transient API errors.
Cost Monitoring: Regularly check dashboard metrics and set up budget alerts.
Data Minimization: Send only necessary data to the AI to reduce token usage and improve privacy.

Error Recovery

Idempotency: Design your request IDs to be unique and consistent for retries.
State Tracking: Persist job IDs and request states to resume processing after interruptions.
Logging: Comprehensive logging helps diagnose issues quickly, especially for failed requests and callbacks.
Output Validation: Use Pydantic models with validation to catch malformed AI responses early.

FAQ

How long do jobs take to process?

Processing time depends on:

Model selected: Flash models are faster than Pro models
Request complexity: Simple text is faster than multimodal content
Queue position: Jobs are processed in order
Batch size: Larger batches process more efficiently

Typical processing times:

Simple text requests: 1-5 seconds per request
Complex analysis: 5-30 seconds per request
Large batches: 10-50% faster per request

What's the maximum batch size?

Requests per job: No hard limit (tested with 100,000+)
Requests per batch: 1-100 (default: 20)
Content size: 32MB per request
Total job size: 1GB recommended maximum

How much does it cost?

Pricing is based on token usage:

Input tokens: Text you send to the AI
Output tokens: Text the AI generates
Model multiplier: Pro models cost more than Flash models

Check current pricing on the Dashboard.

How long are results stored?

Results: 30 days after job completion
Job metadata: 90 days
Usage logs: 1 year for billing purposes

Download important results promptly or use callbacks to process them automatically.

What if my callback fails?

Callback failures don't affect job processing:

Jobs complete normally even if callbacks fail
Callback errors are logged for debugging
You can always retrieve results manually via get_results()
Failed callbacks can be retried by retrieving the job

Recover from callback failure:

# Recover from callback failure
job = get_batch_job(api_key="your_api_key", job_id="failed_callback_job")
if job.get_state() == "succeeded":
    # Manually process results
    for result in job.get_results(output_model=None):
        process_result(result)

Ready to get started?

Set up your Google Cloud integration and start processing AI requests at scale.

Configure Settings Quick Start Guide