DEV Community: Elizabeth Adhiambo

What Really Matters In AI Agent Performance

Elizabeth Adhiambo — Mon, 18 May 2026 10:44:17 +0000

Harness Engineering: The Primary Lever for Agent Performance

When an AI agent fails, the first instinct is usually the same: "We probably need a better model."

So teams upgrade the model, tweak the temperature, rewrite prompts for the tenth time, and hope the next run magically stabilizes the system. Most of us have fallen into this loop.

Sometimes performance improves slightly. Most times, the failures remain — instruction drift, inconsistent outputs, tool misuse, looping behaviour, degraded reasoning over long tasks.

The issue is often not the model itself. The issue is the harness.

Three Eras of AI Engineering

The industry has quietly moved through three distinct phases:

Era	Core Question
Prompt Engineering	What should I say?
Context Engineering	What should I send?
Harness Engineering	What system should I build around the model?

The focus is no longer just prompting. The focus is orchestration.

Reliable agents are not produced by prompts alone. They emerge from the surrounding system: tool interfaces, memory handling, routing logic, retry strategies, validation layers, execution environments, and context management.

What the Harness Actually Is

The simplest definition, adopted from LangChain: everything surrounding the LLM that isn't the LLM itself.

Agent = Model + Harness

If you are not training foundation models, most of your engineering work is actually harness engineering. The harness includes:

System prompts - The persistent behavioural rules shaping the agent.

Tools and MCPs - The schemas, descriptions, and execution interfaces the model uses to act.

Memory systems - What gets preserved, retrieved, summarized, or discarded.

Control logic - Retries, checkpoints, validation loops, routing, delegation, and stopping conditions.

Infrastructure - Browsers, sandboxes, terminals, APIs, filesystems, execution environments.

Middleware and guardrails - Deterministic layers wrapped around probabilistic behaviour.

Most production reliability comes from these layers - not from raw model intelligence.

The Four Recurring Failure Bottlenecks

Agent failures are usually systematic, not random. In practice, breakdowns cluster around four recurring bottlenecks.

1. Prompt Failures

Prompt failures are often specification failures. Agents break when prompts contain vague objectives, unclear success criteria, conflicting instructions, missing constraints, or no negative examples.

The signal is usually inconsistent behaviour across runs — the same input produces dramatically different outcomes because the system never defined behaviour precisely enough.

A lot of prompt engineering is really requirements engineering.

2. Tool Failures

Tool quality heavily influences agent reliability. Models understand tools through text: names, schemas, descriptions, parameter definitions. Poorly designed tools create failure even with strong models.

Common issues include overly broad tools, ambiguous parameters, overlapping capabilities, inconsistent outputs, and weak docstrings.

The signal: wrong tool calls, malformed arguments, hallucinated tool usage, skipped retrieval steps. A highly capable model can still fail if the tool layer is poorly designed.

3. Context Failures

One of the biggest misconceptions in agent systems is: "More context automatically improves reasoning." It often does the opposite.

Teams frequently dump full chat histories, verbose tool outputs, entire documents, duplicated summaries, and abandoned reasoning traces into the context window - hoping the model finds what matters. This creates context flooding. The model starts searching for a needle inside an expanding haystack.

High-performing systems instead optimize for context precision, continuously asking:

What does the model actually need for this step?
What can be summarized?
What must remain verbatim?
What should never enter context at all?

More context does not automatically mean better context.

4. Control Logic Failures

Long-running agents degrade without orchestration. Without proper control logic, retries become loops, failures propagate unchecked, reasoning drifts, and outputs degrade over time.

The absence of validation, checkpoints, retry strategies, evaluation loops, human approval gates, and stopping conditions creates unstable systems regardless of model quality. Autonomy without control logic becomes chaos very quickly.

A Practical Demonstration: Same Model, Different Harness

Consider an HR operations agent responsible for generating job descriptions, validating compliance, posting to a job board, reviewing applications, extracting resume data, and ranking candidates.

A monolithic harness uses one giant prompt, one agent, one long execution chain. It may complete parts of the workflow, but reliability quickly degrades: instructions get lost, outputs become inconsistent, validation gets skipped, tool usage becomes noisy, later stages contradict earlier decisions.

A harness with specialized subagents distributes responsibility cleanly:

Subagent	Responsibility
JD Agent	Generate the job description
Compliance Agent	Validate legal/compliance requirements
Screening Agent	Extract structured resume data
Ranking Agent	Score candidates
Evaluation Agent	Verify outputs before progression

The model itself remains unchanged. What changes is the orchestration layer.

The result: more stable execution, cleaner context separation, better tool usage, lower instruction drift, easier debugging, and better recoverability. The performance improvement comes from architecture, not intelligence.

Why Harnessed Systems Perform Better

Once agents are treated as systems rather than prompts, several benefits emerge.

Reliability - The same task produces consistent behaviour across runs. The goal in production is not occasional brilliance; it is dependable execution.

Observability - Failures become diagnosable. You can inspect prompts, tool calls, retries, state transitions, context evolution, and evaluation results. Observability turns agent debugging from guesswork into engineering.

Recoverability - Strong systems recover from failure instead of collapsing from it. Good harnesses retry intelligently, validate outputs, checkpoint progress, and escalate when confidence drops. Resilient systems are not the ones that never fail — they are the ones that recover gracefully.

Scalability - As workflows grow more complex, architecture absorbs the complexity instead of endlessly expanding prompts. Responsibility becomes distributed across agents, workflows, memory layers, evaluators, and routing systems.

Model independence - Well-engineered harnesses extract more value from existing models. In many cases, a strong harness running on a smaller model will outperform a weak harness running on a larger one.

When the Model Actually Is the Bottleneck

Model capability still matters. There are legitimate cases where the model is the limiting factor: deep multi-step reasoning, specialized domain knowledge, advanced planning, large-scale code synthesis, highly constrained inference tasks.

But those situations should be demonstrated, not assumed.

If the failure can be explained by vague instructions, poor tool design, context flooding, missing retries, absent validation, or weak orchestration — then the problem is architectural before it is intellectual.

In production systems:

The harness is usually the product. The model is a component inside it.

Github Repo

Building Async Backends with Django and Celery

Elizabeth Adhiambo — Fri, 03 Oct 2025 17:05:46 +0000

Performance is everything in modern web applications. Users expect fast responses, instant feedback, and seamless experiences even when your backend is running heavy workloads.

So what happens when your app needs to handle tasks that can't complete in milliseconds? Think of:

Generating AI-powered summaries
Processing large CSV files for bulk imports
Sending thousands of emails or notifications
Running complex image or video processing

If you try to handle these in the request/response cycle, your app slows down or worse, times out.

This is where background processing with Django and Celery comes in. In this article, we'll break down how to build a production-ready async backend that can handle heavy lifting without blocking user requests.

Why Background Processing Matters

Imagine this scenario: your user uploads a 100MB CSV with 50,000 records, and you try to process it in the same request.

The request takes minutes to complete
Your server thread is blocked
The user gets frustrated or abandons the app

Instead, with background processing:

The request is accepted quickly
The task is queued in the background
The user gets an instant response (202 Accepted – Your job is being processed)
The heavy lifting happens elsewhere (workers)
The user can poll for results or receive notifications

The Architecture

Here's a simplified flow:

Client → Django View → Celery Task Queue → Worker → Result Backend → Client (WebSocket)

Client → Django View: User submits a request (e.g., upload a file, request AI summary)
Django View → Task Queue: The request is validated and pushed to Celery via a broker like Redis
Worker: Celery workers pick up tasks from the queue and process them asynchronously
Result Backend: Task results and status are stored in Redis or your database
Client: Receives real-time updates via WebSocket (no polling needed!)

Project Setup & Configuration

Installing Dependencies

pip install celery redis channels channels-redis django-rest-framework
pip install google-generativeai PyPDF2 pdfplumber python-docx

Celery Configuration

The key to a robust Celery setup is proper configuration with rate limiting, time limits, and periodic tasks:

# core/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'core.settings')
app = Celery('ai_document_processor')

app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

# Periodic tasks for maintenance
app.conf.beat_schedule = {
    'cleanup-failed-jobs': {
        'task': 'documents.tasks.cleanup_failed_jobs',
        'schedule': 3600.0,  # Every hour
    },
}

# Task execution limits and rate control
app.conf.task_annotations = {
    'documents.tasks.process_document': {
        'rate_limit': '10/m',      # 10 tasks per minute
        'time_limit': 1800,        # 30 minutes max
        'soft_time_limit': 1500,   # Warning at 25 minutes
    },
}

Key Configuration Highlights:

Rate Limiting: Prevents overwhelming external APIs
Time Limits: Prevents runaway tasks from consuming resources
Periodic Tasks: Automated cleanup via Celery Beat
Auto-discovery: Finds tasks across all Django apps

Django Settings

# settings.py
REDIS_URL = config('REDIS_URL', default='redis://localhost:6379/0')

# Celery
CELERY_BROKER_URL = REDIS_URL
CELERY_RESULT_BACKEND = REDIS_URL
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'

# Channels for WebSockets
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {'hosts': [REDIS_URL]},
    },
}

# AI Configuration
GEMINI_API_KEY = config('GEMINI_API_KEY')
GEMINI_MODEL = 'gemini-pro'

Data Models

We use four key models to track the entire processing lifecycle:

Document: Stores file metadata and processing status
ProcessingJob: Tracks Celery task execution with progress updates
DocumentAnalysis: Stores AI-generated results (summary, sentiment, topics, etc.)
ProcessingLog: Detailed logs for debugging

Each model uses UUIDs for better security and includes timestamps, status fields, and relationships that ensure data integrity through OneToOne and ForeignKey constraints.

The Processing Pipeline

Main Task Orchestration

Here's the heart of the system - the main Celery task that coordinates everything:

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def process_document(self, document_id: str) -> dict:
    """Process document through the complete pipeline"""
    try:
        document = Document.objects.get(id=document_id)

        # Create processing job
        processing_job = ProcessingJob.objects.create(
            document=document,
            celery_task_id=self.request.id,
            status='started'
        )

        # Step 1: Extract text (20% progress)
        self.update_state(state='PROGRESS', meta={'progress': 20})
        extracted_text = extract_text_from_document(document)
        send_websocket_update(document_id, 'processing_update', {
            'progress': 20,
            'message': 'Text extraction completed'
        })

        # Step 2: AI Analysis (50% progress)
        self.update_state(state='PROGRESS', meta={'progress': 50})
        analysis_results = analyze_document_with_gemini(document, extracted_text)
        send_websocket_update(document_id, 'processing_update', {
            'progress': 90,
            'message': 'AI analysis completed'
        })

        # Step 3: Finalize (100% progress)
        document.status = 'completed'
        document.save()

        send_websocket_update(document_id, 'processing_complete', {
            'status': 'completed'
        })

        return {'status': 'success', 'document_id': document_id}

    except Exception as exc:
        # Handle errors and retry with exponential backoff
        if self.request.retries < self.max_retries:
            raise self.retry(countdown=60 * (2 ** self.request.retries))
        raise exc

Key Features:

Progress Tracking: Updates state at each step (20%, 50%, 100%)
WebSocket Updates: Real-time notifications to connected clients
Retry Logic: Exponential backoff for transient failures
Error Handling: Comprehensive exception management

Sending Real-Time Updates

The critical link between Celery workers and WebSocket clients:

def send_websocket_update(document_id: str, event_type: str, data: dict):
    """Send updates to WebSocket clients"""
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        f'document_{document_id}',
        {
            'type': event_type,
            'document_id': document_id,
            'timestamp': timezone.now().isoformat(),
            **data
        }
    )

This bridges the gap between synchronous Celery tasks and asynchronous WebSocket connections, enabling real-time updates.

Text Extraction with Fallbacks

def extract_text_from_pdf(file_path: str) -> str:
    """Extract text with fallback strategy"""
    try:
        # Primary: pdfplumber (more reliable)
        with pdfplumber.open(file_path) as pdf:
            return '\n'.join([page.extract_text() for page in pdf.pages])
    except Exception:
        # Fallback: PyPDF2
        return extract_with_pypdf2(file_path)

Multiple extraction methods ensure robustness - if one library fails, we automatically try another.

AI Analysis

def analyze_document_with_gemini(document, text_content: str) -> dict:
    """Run multiple AI analyses"""
    model = genai.GenerativeModel('gemini-pro')
    analysis = DocumentAnalysis.objects.create(document=document)

    # Run different analysis types
    analysis_types = {
        'summary': 'Provide a 2-3 sentence summary',
        'key_points': 'Extract key points as JSON list',
        'sentiment': 'Analyze sentiment: positive/negative/neutral',
        'topics': 'Identify main topics as JSON list'
    }

    for analysis_type, prompt in analysis_types.items():
        response = model.generate_content(f"{prompt}\n\n{text_content[:4000]}")
        # Store results in analysis object
        setattr(analysis, f'{analysis_type}_completed', True)
        time.sleep(1)  # Respect API rate limits

    analysis.save()
    return analysis

Best Practices:

Rate limiting with sleep between calls
Token tracking for cost management
Flexible JSON parsing with fallbacks
Error isolation - one failure doesn't break others

Real-Time Updates with WebSockets

Server-Side Consumer

class DocumentProcessingConsumer(AsyncWebsocketConsumer):
    """Real-time processing updates"""

    async def connect(self):
        self.document_id = self.scope['url_route']['kwargs']['document_id']
        self.room_group_name = f'document_{self.document_id}'

        # Verify user has access
        if await self.check_document_access():
            await self.channel_layer.group_add(
                self.room_group_name,
                self.channel_name
            )
            await self.accept()
            await self.send_document_status()

    async def processing_update(self, event):
        """Forward processing updates to client"""
        await self.send(text_data=json.dumps({
            'type': 'processing_update',
            'progress': event['progress'],
            'message': event['message']
        }))

WebSocket Features:

Authentication checks before accepting connections
Room-based broadcasting for multiple clients
Automatic status updates on connection
Type-specific event handlers

Client-Side Integration

const socket = new WebSocket(`ws://${location.host}/ws/documents/${docId}/`);

socket.onmessage = function(e) {
    const data = JSON.parse(e.data);

    if (data.type === 'processing_update') {
        updateProgressBar(data.progress);
        showMessage(data.message);
    } else if (data.type === 'processing_complete') {
        showSuccess('Processing completed!');
    }
};

REST API Design

Document Processing Endpoints

class DocumentViewSet(viewsets.ModelViewSet):
    permission_classes = [IsAuthenticated]

    def perform_create(self, serializer):
        """Upload and automatically start processing"""
        document = serializer.save(user=self.request.user)
        process_document.delay(str(document.id))

    @action(detail=True, methods=['post'])
    def reprocess(self, request, pk=None):
        """Restart failed processing"""
        document = self.get_object()
        document.status = 'uploaded'
        document.save()
        process_document.delay(str(document.id))
        return Response({'message': 'Reprocessing started'})

    @action(detail=True, methods=['post'])
    def cancel_processing(self, request, pk=None):
        """Cancel active processing"""
        job = self.get_object().processing_job
        from celery import current_app
        current_app.control.revoke(job.celery_task_id, terminate=True)
        return Response({'message': 'Processing cancelled'})

API Highlights:

RESTful design with standard HTTP methods
Automatic processing on upload
Reprocessing for failed documents
Task cancellation support

Production Best Practices

1. Task Rate Limiting

app.conf.task_annotations = {
    'documents.tasks.process_document': {
        'rate_limit': '10/m',  # Prevent queue flooding
    },
}

2. Monitoring with Flower

pip install flower
celery -A core flower --port=5555
# Access dashboard at http://localhost:5555

Flower provides:

Real-time task monitoring
Worker statistics
Task history and results
Performance graphs

3. Horizontal Scaling

# Run multiple workers
celery -A core worker --concurrency=4 -n worker1@%h
celery -A core worker --concurrency=4 -n worker2@%h

# Separate queues by priority
celery -A core worker -Q high_priority,default -n worker1@%h
celery -A core worker -Q low_priority -n worker2@%h

4. Task Priority Queues

# Configure routing
app.conf.task_routes = {
    'documents.tasks.process_document': {'queue': 'high_priority'},
    'documents.tasks.cleanup_old_logs': {'queue': 'low_priority'},
}

# Send to specific queue
process_document.apply_async(
    args=[document_id],
    queue='high_priority',
    priority=9
)

5. Error Tracking

import sentry_sdk
from sentry_sdk.integrations.celery import CeleryIntegration

sentry_sdk.init(
    dsn="your-sentry-dsn",
    integrations=[CeleryIntegration()],
)

Beyond AI: Other Real-Time Use Cases

WebSocket + Celery architecture is perfect for:

Live Data Processing: Real-time analytics, log processing, data transformations
File Uploads: Show upload progress, virus scanning, format conversion
Report Generation: PDF/Excel generation with progress updates
Batch Operations: Bulk updates, mass emails, data migrations
Integration Workflows: Multi-step API integrations with status updates
Machine Learning: Model training progress, inference batching
Image/Video Processing: Encoding, resizing, filtering with live previews

Wrapping Up

By combining Django, Celery, and Django Channels, you can:

Keep your app responsive under heavy load
Provide real-time feedback to users (no polling!)
Handle long-running tasks safely in the background
Scale horizontally by adding more workers
Build modern, real-time user experiences

This architecture powers everything from AI-driven features to bulk data processing anywhere you need to keep users informed while work happens asynchronously.

Quick Reference Commands

# Development
redis-server
celery -A core worker --loglevel=info
python manage.py runserver

# Production
celery -A core worker --loglevel=info --concurrency=4
celery -A core beat --loglevel=info  # Scheduled tasks
celery -A core flower                 # Monitoring
daphne -b 0.0.0.0 -p 8000 core.asgi:application

# Debugging
celery -A core inspect active      # View active tasks
celery -A core inspect stats       # Worker statistics
celery -A core purge              # Clear all tasks

Resources

Repository: The complete code is available at ai-document-processor

Why I Chose pass for Password Management: A Complete Guide

Elizabeth Adhiambo — Tue, 15 Aug 2023 20:47:40 +0000

Managing and remembering a plethora of passwords for different services is a universal challenge. It certainly was for me. Take, for instance, the frequent pull requests I made, requiring GitHub's access token each time. My approach?

Saving the token in an empty file or — and I can't believe I'm admitting this — emailing it to myself. Every time I needed that token, I'd dive into my emails or search through my files, copy the token, then paste it where required. It was an exhausting routine, especially during days packed with multiple pushes. That's when I discovered pass — the game-changer that transformed my password management hustle.

In this article, I'll walk you through a step-by-step process to make managing your passwords as seamless as it now is for me.

Why pass?

Unix Philosophy: pass integrates seamlessly with the Unix environment, making it intuitive for those familiar with Unix-based systems.
GPG Encryption: Passwords are securely stored using GPG (GNU Privacy Guard).
Git Integration: Changes to passwords can be tracked using Git, providing version control.

Setting Up pass

GPG Keys: You'll need a GPG key pair. If you don't have one yet, here's how you can generate it:

gpg --gen-key

After generating your GPG Key, you will be prompted to enter your name and email address. Additionally, you will be asked to set a passphrase.

2. Install pass:

The next step is to install pass in your system.

sudo apt install pass   # For Debian/Ubuntu systems
brew install pass       # For macOS

3. Initialize your Password store

The next step is to initialize the password store.

pass init <YOUR_GPG_KEY>

If you forgot to copy you GPG Key your can use this command to retrieve it:

gpg --list-keys

Your key should see something like this on your terminal:

/home/my_computer/.gnupg/pubring.kbx
----------------------------
pub   rsa3072 00-00-00 [SC] [expires: 0000-00-00]
      EG8H723F2CF2T052W6BFJJE3E904L6C87Q8C1A9 #This is your GPG Key
uid           [ultimate] name <youremail@email.com>
sub   rsa3072 00-00-00 [E] [expires: 0000-00-00]

Insert/Store your passwords

To store your password, let's say you want to store your github access key:

pass insert github

Retrieving your passwords

To retrieve your password you can either:

Show the password in the terminal

pass github

Copy the password in your clipboard for 45 Seconds

pass -c github

You can also list your stored passwords using:

pass ls

More pass features

1. Password Generation

Pass can generate complex passwords for you

pass generate [your_service_path] [length]

2. Keep your password history

you can use git to keep a history of your password changes

pass git init
pass git push

Remove passwords

You can also remove stored passwords by:

pass rm github

Conclusion

pass offers an elegant solution to password management. If you're looking for a change or simply aiming to improve your password management strategy, give pass a shot. You might be as thrilled as I was.

For more information or advanced features, refer to the official pass documentation.

Feel free to share other tips you found useful for password management!