JobManagerMixin Architecture Documentation¶
Overview¶
The JobManagerMixin is a sophisticated architectural pattern that bridges synchronous web views with asynchronous job execution systems. It was designed specifically for AI-Parrot's needs but maintains flexibility for any Python web framework.
Architecture Diagram¶
┌─────────────────────────────────────────────────────────────┐
│ Client / User │
└───────────────────────┬─────────────────────────────────────┘
│
│ HTTP Request (POST/GET)
▼
┌─────────────────────────────────────────────────────────────┐
│ Web Framework Layer │
│ (FastAPI / Django / Flask / Custom) │
└───────────────────────┬─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ View with JobManagerMixin │
│ ┌───────────────────────────────────────────────────┐ │
│ │ @as_job decorated methods │ │
│ │ ├─ Method 1 (queue="embeddings") │ │
│ │ ├─ Method 2 (queue="rag") │ │
│ │ └─ Method 3 (queue="agents") │ │
│ └───────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ GET handler │ │
│ │ ├─ Check for job_id parameter │ │
│ │ ├─ Return job status/result │ │
│ │ └─ Delegate to parent GET if no job_id │ │
│ └───────────────────────────────────────────────────┘ │
└───────────────────────┬─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Job Manager │
│ (RQ / Celery / Dramatiq / Custom) │
│ ┌─────────────────────────────────────────┐ │
│ │ Job Queue │ │
│ │ ├─ embeddings queue │ │
│ │ ├─ rag queue │ │
│ │ └─ agents queue │ │
│ └─────────────────────────────────────────┘ │
└───────────────────────┬─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Worker Processes │
│ Multiple workers consuming from queues │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Worker 1 │ │ Worker 2 │ │ Worker 3 │ │
│ │ (CPU) │ │ (GPU) │ │ (I/O) │ │
│ └────────────┘ └────────────┘ └────────────┘ │
└───────────────────────┬─────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ AI-Parrot Components │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ LLM Clients │ │ Vector Store │ │ Agents │ │
│ │ - OpenAI │ │ - PgVector │ │ - Registry │ │
│ │ - Claude │ │ - Embeddings│ │ - Tools │ │
│ │ - Gemini │ │ │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
Component Breakdown¶
1. JobManagerMixin Class¶
The core mixin that provides:
Responsibilities:
- Decorator factory (@as_job) for marking methods as async
- Job enqueueing and metadata handling
- GET method override for status checking
- Response formatting and error handling
Key Methods:
@staticmethod
def as_job(queue, timeout, result_ttl, return_job_id):
"""Transform any method into an async job"""
def get(request, *args, **kwargs):
"""Handle job status requests or delegate to parent"""
def _handle_job_status_request(job_id, request):
"""Fetch and format job status/result"""
2. Decorator Pattern¶
The @as_job decorator implements the following flow:
# Original method
def process_documents(self, request):
# Heavy computation
return result
# Decorated method
@JobManagerMixin.as_job(queue="processing")
def process_documents(self, request):
# Same implementation
return result
# What happens when called:
# 1. Wrapper intercepts the call
# 2. Enqueues job to job_manager
# 3. Returns job_id immediately
# 4. Actual execution happens in worker
Decorator Flow:
Method Call
↓
Wrapper intercepts
↓
Extract instance & arguments
↓
Enqueue to job_manager
↓
Create job metadata
↓
Return response with job_id
↓
(Actual execution in worker process)
3. Job Lifecycle State Machine¶
┌──────────┐
│ QUEUED │ ← Job just enqueued
└────┬─────┘
│
▼
┌──────────┐
│ STARTED │ ← Worker picked up job
└────┬─────┘
│
├──────→ ┌───────────┐
│ │ FAILED │ ← Exception occurred
│ └───────────┘
│
▼
┌──────────┐
│ FINISHED │ ← Job completed successfully
└──────────┘
4. Request Flow Patterns¶
Pattern A: Create Async Job¶
POST /api/documents/process
{
"action": "vectorize",
"documents": [...]
}
↓
View.vectorize_documents() called
↓
@as_job decorator intercepts
↓
Job enqueued to "embeddings" queue
↓
Response with job_id
{
"success": true,
"job_id": "job_abc123",
"status_url": "/api/documents/process?job_id=job_abc123"
}
Pattern B: Check Job Status¶
GET /api/documents/process?job_id=job_abc123
↓
View.get() called
↓
Mixin detects job_id parameter
↓
Fetch job from job_manager
↓
Return status/result
{
"success": true,
"job_id": "job_abc123",
"status": "finished",
"result": {...}
}
Integration Patterns¶
Pattern 1: FastAPI Integration¶
class FastAPIAdapter(JobManagerMixin):
"""Async-native FastAPI integration."""
def __init__(self):
self.job_manager = rq.Queue(connection=redis_conn)
super().__init__()
# Usage in routes
@app.post("/task")
async def create_task():
return adapter.async_method()
Benefits: - Natural async/await support - Type hints work seamlessly - Automatic OpenAPI documentation
Pattern 2: Django Integration¶
class DjangoView(JobManagerMixin, APIView):
"""Django REST Framework integration."""
def __init__(self):
self.job_manager = celery_app
super().__init__()
# Celery handles distributed task execution
Benefits: - Leverages Django's middleware - ORM integration for job metadata - Built-in authentication/permissions
Pattern 3: AI-Parrot Native Integration¶
class AIParrotView(JobManagerMixin):
"""Full AI-Parrot stack integration."""
def __init__(self, llm, vector_store, agents, tools):
self.llm = llm # Claude/GPT/Gemini
self.vector_store = vector_store # PgVector
self.agents = agents # Agent Registry
self.tools = tools # Tool Manager
self.job_manager = get_job_manager()
super().__init__()
@JobManagerMixin.as_job(queue="embeddings", timeout=7200)
def vectorize_documents(self, request):
# Use all AI-Parrot components
docs = self.loader.load(request.data['files'])
embeddings = self.vector_store.embed(docs)
self.vector_store.store(embeddings)
return {"success": True}
Design Patterns Used¶
1. Mixin Pattern¶
- Adds functionality to any base class
- Multiple inheritance friendly
- No modification of base class needed
2. Decorator Pattern¶
- Wraps methods with async behavior
- Preserves original function metadata
- Configurable via parameters
3. Template Method Pattern¶
get()method provides structure- Subclasses can override specific parts
- Default behavior with extension points
4. Strategy Pattern¶
- Job manager is pluggable (RQ, Celery, etc.)
- Framework-agnostic design
- Adapter pattern for different backends
Advantages¶
1. Separation of Concerns¶
- API endpoint logic separate from execution
- View handles routing, worker handles computation
- Clean boundaries between components
2. Scalability¶
- Workers scale independently
- Queue-based load distribution
- Different queues for different resource types
3. Resilience¶
- Job failures don't crash web server
- Retries handled by job manager
- Timeouts prevent resource exhaustion
4. User Experience¶
- Immediate response (no blocking)
- Progressive status updates
- Asynchronous notifications possible
5. Resource Optimization¶
- CPU-intensive tasks don't block I/O
- GPU tasks queued separately
- Memory-intensive operations isolated
AI-Parrot Specific Use Cases¶
Use Case 1: Document Ingestion Pipeline¶
@JobManagerMixin.as_job(queue="document_processing", timeout=7200)
def ingest_documents(self, request):
"""
Multi-stage pipeline:
1. Load various document formats (PDF, DOCX, etc.)
2. Chunk documents intelligently
3. Generate embeddings with Huggingface
4. Store in PgVector with metadata
"""
# Long-running operation perfect for async execution
Why async? - Processing 1000s of documents takes hours - Embedding generation is CPU-intensive - User shouldn't wait for completion - Can process in parallel across workers
Use Case 2: RAG Query with Context¶
@JobManagerMixin.as_job(queue="rag_queries", timeout=300)
def rag_query(self, request):
"""
RAG pipeline:
1. Embed user query
2. Similarity search in PgVector
3. Retrieve top-k contexts
4. Build prompt with context
5. Call LLM (Claude/GPT)
6. Return response with sources
"""
Why async? - Vector search can be slow with large DBs - LLM API calls have latency - Want to track query history - Can aggregate multiple sources
Use Case 3: Multi-Agent Orchestration¶
@JobManagerMixin.as_job(queue="agent_workflows", timeout=1800)
def execute_agent_workflow(self, request):
"""
Orchestrate multiple agents:
1. Research agent gathers information
2. Analysis agent processes data
3. Writer agent creates content
4. Reviewer agent validates output
Each agent may call multiple tools and LLMs
"""
Why async? - Multi-step workflows take time - Agent-to-agent communication - Tool calls (web search, calculations) - Multiple LLM interactions
Performance Considerations¶
Queue Selection Strategy¶
# Fast queue for quick operations (<30s)
@as_job(queue="fast", timeout=30)
def quick_lookup(self, request):
pass
# Medium queue for typical operations (30s-5min)
@as_job(queue="medium", timeout=300)
def standard_processing(self, request):
pass
# Slow queue for heavy operations (5min-2hr)
@as_job(queue="slow", timeout=7200)
def batch_processing(self, request):
pass
# GPU queue for ML inference
@as_job(queue="gpu", timeout=300)
def run_inference(self, request):
pass
Worker Configuration¶
# CPU workers for embeddings
rq worker embeddings --burst
# GPU workers for inference
CUDA_VISIBLE_DEVICES=0 rq worker gpu_inference
# I/O workers for document loading
rq worker document_loading --worker-class rq.Worker
Testing Strategy¶
Unit Tests¶
def test_decorator_marks_method():
"""Test @as_job marks methods correctly"""
def test_job_enqueuing():
"""Test jobs are enqueued with correct params"""
def test_status_checking():
"""Test job status retrieval"""
Integration Tests¶
def test_full_lifecycle():
"""Test create → enqueue → execute → retrieve"""
def test_error_handling():
"""Test job failure handling"""
def test_concurrent_jobs():
"""Test multiple jobs execute correctly"""
End-to-End Tests¶
def test_ai_parrot_workflow():
"""Test complete AI-Parrot pipeline"""
# Upload docs → vectorize → query → get results
Future Enhancements¶
1. Job Prioritization¶
2. Job Dependencies¶
@as_job(queue="processing", depends_on=[job_id_1, job_id_2])
def aggregate_results(self, request):
pass
3. Webhooks¶
4. Progress Tracking¶
@as_job(queue="processing")
def process_large_dataset(self, request):
for i, item in enumerate(dataset):
process(item)
self.update_progress(i / len(dataset) * 100)
Conclusion¶
The JobManagerMixin provides a robust, scalable architecture for integrating asynchronous job execution into web applications. Its design specifically addresses AI-Parrot's needs for:
- Long-running ML operations
- Document processing pipelines
- Multi-agent orchestration
- Vector database operations
While maintaining flexibility for general-purpose use across different frameworks and job queue backends.