The Prioritization Pattern is an orchestration approach that enables AI agents to intelligently rank tasks, allocate limited resources, and determine optimal execution order based on multiple criteria such as urgency, impact, effort, deadlines, and strategic importance. Unlike simple first-in-first-out (FIFO) queues or static scheduling, Prioritization provides a dynamic framework for making informed decisions about what to work on next when faced with competing demands and resource constraints.
This pattern transforms chaotic task management into structured, goal-aligned execution by continuously assessing priority based on changing conditions, preventing resource starvation, ensuring deadline adherence, and optimizing for business value. It's particularly valuable for support systems, project management, incident response, resource allocation, and any scenario where multiple tasks compete for limited capacity.
---
title: Prioritization — Multi-Criteria Scoring
---
%%{init: {'look':'handDrawn','theme':'base','themeVariables':{'background':'#f5ecd9','primaryColor':'#ede0bd','primaryBorderColor':'#6b4423','primaryTextColor':'#3e2723','lineColor':'#6b4423','clusterBkg':'#efe5cd','clusterBorder':'#c5b393','fontFamily':'Caveat, Patrick Hand, cursive'}}}%%
flowchart LR
T([tasks])
Sc[score:<br/>urgency · impact · effort]
Rk[rank]
Sch[schedule]
Ex[execute]
Rb{rebalance?}
T --> Sc --> Rk --> Sch --> Ex --> Rb
Rb -- "drift" --> Sc
Rb -- "ok" --> Ex
Traditional approaches have significant limitations:
- FIFO queues: First task in gets worked on first, regardless of urgency or importance
- Manual prioritization: Subjective, inconsistent, doesn't scale, can't adapt quickly
- Static priorities: Set once and never change, even as conditions evolve
- No resource awareness: Overcommit or underutilize available capacity
- Starvation risk: Low-priority tasks may wait indefinitely
Prioritization solves these by:
- Multi-criteria scoring: Consider urgency, impact, effort, strategic alignment, and deadlines
- Dynamic rebalancing: Priorities adjust as tasks age, deadlines approach, or conditions change
- Resource optimization: Allocate capacity to maximize throughput and value delivery
- Deadline awareness: Ensure time-sensitive tasks complete on schedule
- Starvation prevention: Gradually increase priority of waiting tasks (aging)
- Transparent ranking: Clear, explainable prioritization decisions
Queue: [Ticket A, Ticket B, Ticket C, Ticket D, Ticket E]
→ Process in arrival order
→ Critical production outage (Ticket D) waits behind password reset (Ticket A)
→ Important feature request (Ticket C) with deadline missed
→ VIP customer (Ticket E) frustrated by long wait
→ Simple 5-minute fix (Ticket B) blocks 2-hour investigation
Incoming: [A: Password reset, B: UI bug, C: Feature request (deadline tomorrow),
D: Production outage, E: VIP customer issue]
Priority Assessment:
┌────┬─────────────────────┬─────────┬────────┬────────┬────────┬───────────┐
│ ID │ Task │ Urgency │ Impact │ Effort │ Aging │ Priority │
├────┼─────────────────────┼─────────┼────────┼────────┼────────┼───────────┤
│ D │ Production outage │ 10 │ 10 │ 8 │ 0 │ 9.5 │
│ E │ VIP customer issue │ 9 │ 8 │ 5 │ 1 │ 8.2 │
│ C │ Feature (deadline) │ 8 │ 7 │ 7 │ 2 │ 7.8 │
│ B │ UI bug │ 6 │ 5 │ 3 │ 2 │ 5.5 │
│ A │ Password reset │ 5 │ 2 │ 2 │ 1 │ 3.8 │
└────┴─────────────────────┴─────────┴────────┴────────┴────────┴───────────┘
Execution Order: D → E → C → B → A
✓ Critical issues addressed first
✓ Deadline-sensitive tasks prioritized
✓ VIP customers served quickly
✓ Resource allocation optimized
The Prioritization pattern operates through five interconnected phases:
┌────────────────────────────────────────────────────────────────┐
│ 1. TASK ASSESSMENT │
│ Evaluate each task across multiple dimensions (urgency, │
│ impact, effort, strategic value, deadline, customer tier) │
└─────────────────────────┬──────────────────────────────────────┘
↓
┌────────────────────────────────────────────────────────────────┐
│ 2. PRIORITY SCORING │
│ Calculate weighted priority scores combining all criteria │
│ Priority = w1×urgency + w2×impact + w3×(1/effort) + aging │
└─────────────────────────┬──────────────────────────────────────┘
↓
┌────────────────────────────────────────────────────────────────┐
│ 3. QUEUE RANKING │
│ Order tasks by priority, considering deadlines and │
│ dependencies (Earliest Deadline First, Critical Path) │
└─────────────────────────┬──────────────────────────────────────┘
↓
┌────────────────────────────────────────────────────────────────┐
│ 4. RESOURCE ALLOCATION │
│ Assign available resources to highest-priority ready tasks │
│ considering capacity constraints and skill requirements │
└─────────────────────────┬──────────────────────────────────────┘
↓
┌────────────────────────────────────────────────────────────────┐
│ 5. DYNAMIC REBALANCING │
│ Continuously reassess priorities as tasks age, deadlines │
│ approach, new tasks arrive, or conditions change │
└────────────────────────────────────────────────────────────────┘
↓ ↑
└──────────────────────────────────────────────┘
(Continuous loop)
- Assessment: Evaluate each task's urgency, business impact, effort required, deadline proximity, and strategic importance
- Scoring: Calculate weighted priority scores using configurable weights for each criterion
- Ranking: Order the queue by priority, applying scheduling algorithms (EDF, SJF, weighted fair queuing)
- Allocation: Assign work to available resources, respecting capacity limits and skill requirements
- Rebalancing: Recalculate priorities as time passes, conditions change, or new information emerges
- Customer support systems: Prioritize tickets by urgency, customer tier, SLA requirements
- Incident response: Triage and address critical production issues before minor bugs
- Project task management: Balance urgent deadlines with important strategic work
- Resource allocation: Distribute limited compute, API quota, or human attention optimally
- Request queuing: Handle API rate limits, job scheduling, batch processing priorities
- Development workflows: Prioritize bug fixes, features, technical debt, and research
- Emergency services: Medical triage, disaster response, resource deployment
- Content moderation: Flag and review high-risk content before low-risk items
- Research agenda: Decide which experiments or studies to pursue given time/budget constraints
- Single task at a time: No need to prioritize if only doing one thing
- All tasks equal priority: If everything truly has the same importance (rare)
- Strict FIFO requirements: Some systems require first-in-first-out by design
- Instantaneous decisions: When overhead of prioritization exceeds task duration
- No resource constraints: Unlimited capacity means everything can run in parallel
Use Prioritization when:
- You have more tasks than capacity to handle simultaneously
- Tasks have different levels of urgency or importance
- There are deadlines or SLAs to meet
- Some tasks provide higher business value than others
- Resource optimization matters (time, compute, people, budget)
- You need to prevent low-priority tasks from starving
Don't use Prioritization when:
- Processing one task at a time in order received
- All tasks genuinely equal in every dimension
- No resource constraints or competing demands
- Prioritization overhead > task execution time
- System design requires strict ordering guarantees
Calculates task priority based on multiple weighted criteria:
Priority = (
w_urgency × urgency_score +
w_impact × impact_score +
w_effort × (1 / effort_estimate) + # Prefer quick wins
w_deadline × deadline_pressure +
w_strategic × strategic_value +
aging_bonus # Prevent starvation
)Common criteria:
- Urgency: How time-sensitive is the task? (1-10)
- Impact: How many users/systems affected? Business value? (1-10)
- Effort: Time/resources required (lower effort = higher priority for quick wins)
- Deadline: Days/hours until due (approaching deadlines increase priority)
- Strategic alignment: Fits company goals, OKRs, roadmap? (1-10)
- Customer tier: Free, paid, enterprise, VIP? (affects weight)
- Aging: How long has task been waiting? (increases over time)
Different strategies for ordering the queue:
Simple Priority Queue:
- Order by priority score (highest first)
- Static weights, recalculated periodically
Earliest Deadline First (EDF):
- Prioritize tasks with nearest deadlines
- Optimal for meeting all deadlines when possible
- Risk: May sacrifice important tasks for urgent unimportant ones
Shortest Job First (SJF):
- Execute quick tasks first to maximize throughput
- Improves average completion time
- Risk: Starvation of long tasks
Weighted Fair Queuing:
- Allocate capacity proportionally to priority weights
- Ensures all priority levels get some attention
- Prevents complete starvation
Multi-Level Feedback Queue:
- Tasks start at high priority
- Decrease priority if not completed quickly
- Prevents monopolization by long-running tasks
Tracks and allocates available capacity:
ResourceManager:
- available_capacity: Current free resources
- reserved_capacity: Allocated to in-progress tasks
- total_capacity: Maximum system capacity
- allocation_strategy: How to distribute resources
- constraints: Skill requirements, dependencies, quotasFunctions:
- Check resource availability before starting tasks
- Reserve resources for assigned work
- Release resources when tasks complete
- Prevent overallocation and thrashing
Prevents low-priority tasks from waiting indefinitely:
aging_bonus = wait_time × aging_factor
# Example: Task waiting 10 minutes gets +2.0 priority boost
aging_bonus = 10 minutes × 0.2 = +2.0
# Eventually, even low-priority tasks rise to the topStrategies:
- Linear aging: Constant boost per time unit
- Exponential aging: Accelerating boost as wait time increases
- Threshold aging: Big boost after waiting X time
- Adaptive aging: Adjust rate based on queue depth
Decides if higher-priority tasks should interrupt running work:
PreemptionPolicy:
- enabled: bool # Allow preemption?
- threshold: float # Priority gap required to preempt
- checkpoints: bool # Can tasks save state and resume?
- cost: float # Penalty for context switchingUse preemption when:
- Critical task arrives (production outage)
- Significant priority gap (critical vs. low)
- Running task is interruptible (has checkpoints)
- Benefit > context switch cost
Basic weighted priority calculation:
from dataclasses import dataclass
from typing import List
import heapq
@dataclass
class Task:
id: str
urgency: int # 1-10
impact: int # 1-10
effort: int # hours
wait_time: float = 0.0
def priority(self) -> float:
"""Calculate priority score"""
# Weighted scoring
score = (
0.4 * self.urgency +
0.3 * self.impact +
0.2 * (10 / self.effort) + # Inverse effort (quick wins)
0.1 * self.wait_time # Aging bonus
)
return score
class PriorityQueue:
def __init__(self):
self.tasks = []
def add(self, task: Task):
# Use negative priority for max-heap behavior
heapq.heappush(self.tasks, (-task.priority(), task))
def pop_highest(self) -> Task:
_, task = heapq.heappop(self.tasks)
return task
# Usage
queue = PriorityQueue()
queue.add(Task(id="A", urgency=9, impact=8, effort=2))
queue.add(Task(id="B", urgency=5, impact=10, effort=8))
next_task = queue.pop_highest() # Returns highest priorityEarliest Deadline First (EDF) algorithm:
from datetime import datetime, timedelta
from dataclasses import dataclass
import heapq
@dataclass
class DeadlineTask:
id: str
deadline: datetime
effort_hours: float
priority_base: float
def deadline_pressure(self) -> float:
"""How urgent based on deadline proximity"""
time_remaining = (self.deadline - datetime.now()).total_seconds() / 3600
if time_remaining <= 0:
return 1000.0 # Overdue - critical
elif time_remaining < self.effort_hours:
return 100.0 # Deadline at risk
elif time_remaining < 2 * self.effort_hours:
return 50.0 # Getting close
else:
return 10.0 / time_remaining # Normal priority
def priority(self) -> float:
return self.priority_base + self.deadline_pressure()
class EDFScheduler:
def __init__(self):
self.queue = []
def schedule(self, task: DeadlineTask):
# EDF: prioritize by deadline, with base priority modifier
heapq.heappush(self.queue, (task.deadline, -task.priority(), task))
def next_task(self) -> DeadlineTask:
_, _, task = heapq.heappop(self.queue)
return taskConsider resource availability and constraints:
@dataclass
class ResourceTask:
id: str
priority: float
required_capacity: float # 0.0 - 1.0
estimated_duration: float # minutes
skills_needed: List[str]
class ResourceScheduler:
def __init__(self, total_capacity: float):
self.total_capacity = total_capacity
self.available_capacity = total_capacity
self.running_tasks = []
self.queue = []
def can_start(self, task: ResourceTask) -> bool:
"""Check if resources available"""
return self.available_capacity >= task.required_capacity
def allocate(self, task: ResourceTask):
"""Reserve resources for task"""
self.available_capacity -= task.required_capacity
self.running_tasks.append(task)
def release(self, task: ResourceTask):
"""Free resources after completion"""
self.available_capacity += task.required_capacity
self.running_tasks.remove(task)
def schedule_next(self):
"""Start highest-priority task that fits"""
# Sort queue by priority
self.queue.sort(key=lambda t: t.priority, reverse=True)
for task in self.queue:
if self.can_start(task):
self.queue.remove(task)
self.allocate(task)
return task
return None # No task fits current capacityPrevent starvation with priority aging:
import time
from typing import Dict
class AgingPriorityQueue:
def __init__(self, aging_factor: float = 0.1):
self.tasks: Dict[str, Task] = {}
self.arrival_times: Dict[str, float] = {}
self.aging_factor = aging_factor
def add(self, task: Task):
"""Add task and record arrival time"""
self.tasks[task.id] = task
self.arrival_times[task.id] = time.time()
def get_priority(self, task_id: str) -> float:
"""Calculate current priority with aging"""
task = self.tasks[task_id]
wait_time = time.time() - self.arrival_times[task_id]
# Base priority + aging bonus
aged_priority = task.priority() + (wait_time * self.aging_factor)
return aged_priority
def next_task(self) -> Task:
"""Get highest priority task considering aging"""
if not self.tasks:
return None
# Find task with highest aged priority
best_id = max(self.tasks.keys(), key=self.get_priority)
task = self.tasks.pop(best_id)
del self.arrival_times[best_id]
return task
def rebalance(self):
"""Recalculate all priorities with current aging"""
# Called periodically or when new tasks arrive
pass- Maximize throughput: Complete more high-value work in less time
- Minimize waste: Avoid spending resources on low-impact tasks
- Balance workload: Distribute capacity across priority levels
- Improve efficiency: Quick wins and high-impact work get attention
- Meet SLAs: Ensure time-sensitive commitments are honored
- Early warning: Detect deadline risks before they become failures
- Proactive scheduling: Allocate time for approaching deadlines
- Prevent escalations: Address urgent issues before they become crises
- Strategic focus: Prioritize work aligned with business objectives
- Value maximization: Deliver highest-impact outcomes first
- Clear trade-offs: Explicit decisions about what to defer
- Measurable impact: Track value delivered per resource spent
- Aging mechanism: Low-priority tasks eventually rise to the top
- Guaranteed progress: All tasks complete eventually
- Balanced attention: Weighted fair queuing ensures all levels served
- Predictable wait times: Estimate when tasks will be addressed
- Clear rankings: Understand why one task prioritized over another
- Auditable decisions: Log priority scores and criteria
- Stakeholder visibility: Show what's being worked on and why
- Continuous optimization: Data-driven tuning of weights and algorithms
Issue: Calculating priorities, rebalancing, and managing queues adds computational cost
Impact: Reduced throughput if prioritization time > task execution time
Mitigation:
- Use simple scoring for fast decisions
- Cache priority calculations
- Rebalance periodically rather than continuously
- Use approximate algorithms for large queues
Issue: Low-priority tasks may wait indefinitely if high-priority work keeps arriving
Impact: Some tasks never complete, SLAs missed, customer frustration
Mitigation:
- Implement aging/priority boost over time
- Set maximum wait time thresholds
- Use weighted fair queuing for guaranteed progress
- Reserve capacity for low-priority work
Issue: Balancing urgency, impact, effort, deadlines, and other factors is non-trivial
Impact: Difficult to tune weights, unexpected prioritization, gaming the system
Mitigation:
- Start with simple scoring (2-3 criteria)
- Use machine learning to learn optimal weights from historical data
- A/B test different prioritization strategies
- Collect feedback on prioritization quality
Issue: Low-priority task blocks high-priority task (e.g., holds shared resource)
Impact: High-priority work delayed despite capacity availability
Mitigation:
- Priority inheritance: Boost blocking task's priority temporarily
- Minimize resource contention
- Use preemption when safe and beneficial
- Explicit dependency tracking
Issue: Users may artificially inflate urgency/impact to jump the queue
Impact: Prioritization loses effectiveness, important work deprioritized
Mitigation:
- Require justification for high urgency/impact
- Manager approval for priority escalations
- Monitor for priority inflation patterns
- Separate user-reported vs. system-calculated priority
# Explicit, measurable criteria
PRIORITY_CRITERIA = {
"urgency": {
"critical": 10, # Production down, security breach
"high": 8, # Major feature broken, deadline today
"medium": 5, # Important but not time-critical
"low": 2, # Nice to have, no deadline
},
"impact": {
"all_users": 10, # Affects entire user base
"large_segment": 7, # 20%+ of users
"small_segment": 4, # < 20% of users
"individual": 1, # Single user
},
"customer_tier": {
"enterprise": 10, # Paying customers, SLAs
"paid": 7, # Standard paid tier
"trial": 4, # Evaluation users
"free": 2, # Free tier
}
}# Different profiles for different objectives
WEIGHT_PROFILES = {
"maximize_throughput": {
# Favor quick wins
"urgency": 0.2,
"impact": 0.3,
"effort": 0.4, # High weight on low effort
"aging": 0.1,
},
"deadline_focused": {
# Ensure SLAs met
"urgency": 0.4,
"deadline_pressure": 0.4,
"impact": 0.1,
"aging": 0.1,
},
"value_maximization": {
# Highest business impact
"impact": 0.5,
"strategic_value": 0.3,
"urgency": 0.1,
"aging": 0.1,
}
}def calculate_aging_bonus(wait_time_minutes: float, base_priority: float) -> float:
"""
Gradually increase priority as tasks wait
Strategy: Linear aging with acceleration
- First 30 min: slow growth
- 30-120 min: linear growth
- 120+ min: exponential growth
"""
if wait_time_minutes < 30:
return wait_time_minutes * 0.01 # Minimal boost
elif wait_time_minutes < 120:
return 0.3 + (wait_time_minutes - 30) * 0.05 # Linear
else:
# Exponential boost after 2 hours
excess = wait_time_minutes - 120
return 4.8 + (excess ** 1.2) * 0.01
# Ensure even lowest-priority task eventually reaches top
# Example: Low priority (2.0) after 3 hours wait → ~10.0 priorityclass QueueHealthMonitor:
def check_health(self, queue: PriorityQueue) -> Dict[str, Any]:
"""Monitor for issues"""
metrics = {
"queue_depth": len(queue.tasks),
"oldest_task_age": self.get_oldest_age(queue),
"average_wait_time": self.get_average_wait(queue),
"starvation_risk": self.count_long_waiting(queue, threshold_minutes=180),
"deadline_violations": self.count_overdue(queue),
}
alerts = []
# Alert if tasks waiting too long
if metrics["oldest_task_age"] > 240: # 4 hours
alerts.append(f"Task waiting {metrics['oldest_task_age']} minutes - starvation risk!")
# Alert if queue growing unbounded
if metrics["queue_depth"] > 1000:
alerts.append(f"Queue depth {metrics['queue_depth']} - need more capacity!")
# Alert on deadline violations
if metrics["deadline_violations"] > 0:
alerts.append(f"{metrics['deadline_violations']} tasks overdue!")
return {"metrics": metrics, "alerts": alerts}class MultiQueuePrioritizer:
"""Separate queues for different priority classes"""
def __init__(self):
self.critical_queue = PriorityQueue() # P0: Production issues
self.high_queue = PriorityQueue() # P1: Important deadlines
self.medium_queue = PriorityQueue() # P2: Normal work
self.low_queue = PriorityQueue() # P3: Backlog
def add_task(self, task: Task):
"""Route to appropriate queue"""
if task.is_critical():
self.critical_queue.add(task)
elif task.priority() >= 8:
self.high_queue.add(task)
elif task.priority() >= 5:
self.medium_queue.add(task)
else:
self.low_queue.add(task)
def next_task(self) -> Task:
"""Weighted fair selection across queues"""
# 60% from critical, 25% high, 10% medium, 5% low
# Ensures all queues get attention
rand = random.random()
if rand < 0.60 and not self.critical_queue.empty():
return self.critical_queue.pop()
elif rand < 0.85 and not self.high_queue.empty():
return self.high_queue.pop()
elif rand < 0.95 and not self.medium_queue.empty():
return self.medium_queue.pop()
else:
return self.low_queue.pop()@dataclass
class PriorityDecision:
timestamp: datetime
task_id: str
priority_score: float
criteria_breakdown: Dict[str, float]
position_in_queue: int
estimated_wait_time: float
def log_priority_decision(task: Task, queue: PriorityQueue):
"""Record prioritization for later analysis"""
decision = PriorityDecision(
timestamp=datetime.now(),
task_id=task.id,
priority_score=task.priority(),
criteria_breakdown={
"urgency": task.urgency * WEIGHTS["urgency"],
"impact": task.impact * WEIGHTS["impact"],
"effort_factor": (10 / task.effort) * WEIGHTS["effort"],
"aging_bonus": task.aging_bonus(),
},
position_in_queue=queue.get_position(task),
estimated_wait_time=queue.estimate_wait_time(task),
)
# Store in database or log file for analysis
logger.info(f"Priority decision: {decision}")Track these metrics to evaluate prioritization effectiveness:
- Queue depth: Number of waiting tasks (trend over time)
- Average wait time: Time from submission to start
- Max wait time: Longest any task has waited
- Throughput: Tasks completed per hour/day
- Starvation count: Tasks waiting > threshold time
- Priority distribution: Histogram of priority scores
- Priority accuracy: Do high-priority tasks actually matter more?
- Priority drift: How often do priorities change significantly?
- Gaming attempts: Suspicious priority inflation patterns
- SLA compliance: % of tasks completed within deadline
- Average slack time: Time buffer between completion and deadline
- Deadline violations: Count and severity of missed deadlines
- At-risk tasks: Count of tasks unlikely to meet deadline
- Utilization rate: % of capacity actively used
- Allocation efficiency: Time spent on high-value vs. low-value work
- Context switch cost: Overhead from preemption/rebalancing
- Idle time: Capacity available but no ready tasks
- Value delivered: Sum of business value for completed tasks
- Customer satisfaction: Related to priority handling
- Cost per task: Resource cost divided by tasks completed
- ROI on prioritization: Value gain vs. prioritization overhead
Incoming Support Tickets:
Ticket A: "Can't reset password"
- Urgency: 6 (blocking user but not critical)
- Impact: 1 (single user)
- Effort: 5 min
- Customer: Free tier
- Priority Score: 4.2
Ticket B: "Payment processing down"
- Urgency: 10 (revenue-impacting)
- Impact: 10 (all paid users affected)
- Effort: 30 min
- Customer: All paid
- Priority Score: 9.8
Ticket C: "Feature request: Dark mode"
- Urgency: 2 (nice to have)
- Impact: 5 (many users want it)
- Effort: 4 hours
- Customer: Mixed
- Priority Score: 3.5
- Wait time: 2 days → Aging bonus +2.4
- Adjusted Priority: 5.9
Ticket D: "Dashboard charts not loading for enterprise client"
- Urgency: 8 (SLA at risk)
- Impact: 8 (affects major customer)
- Effort: 15 min
- Customer: Enterprise (SLA)
- Priority Score: 8.9
Prioritized Queue:
1. Ticket B (9.8) - Payment processing down [START IMMEDIATELY]
2. Ticket D (8.9) - Enterprise dashboard issue [NEXT]
3. Ticket C (5.9) - Dark mode feature [After aging boost]
4. Ticket A (4.2) - Password reset [When capacity available]
Outcome:
✓ Revenue-critical issue resolved first
✓ Enterprise SLA maintained
✓ Old feature request not starved (aging helped)
✓ Simple password reset handled when convenient
Development Backlog:
Task A: "Critical security vulnerability fix"
- Urgency: 10 (active exploit in the wild)
- Impact: 10 (all users at risk)
- Effort: 2 hours
- Deadline: Today
- Strategic value: 10
- Priority: 10.0
Task B: "Implement new payment gateway"
- Urgency: 7 (deadline next week)
- Impact: 8 (enables new revenue stream)
- Effort: 16 hours (2 days)
- Deadline: 7 days
- Strategic value: 9
- Priority: 7.8
Task C: "Refactor legacy authentication code"
- Urgency: 3 (technical debt, not urgent)
- Impact: 4 (improves maintainability)
- Effort: 8 hours (1 day)
- Deadline: None
- Strategic value: 5
- Wait time: 45 days → Aging +4.5
- Priority: 7.5
Task D: "Fix minor UI alignment issue"
- Urgency: 2 (cosmetic)
- Impact: 2 (low visibility)
- Effort: 30 min
- Deadline: None
- Strategic value: 1
- Priority: 2.5
Resource Constraints:
- 2 developers available
- 8 hours/day capacity
Schedule:
Day 1:
Dev 1: Task A (2h) → Task C (6h) [High priority + aging boost]
Dev 2: Task B (8h) [Parallel execution]
Day 2:
Dev 1: Task C (2h remaining) → Task D (0.5h)
Dev 2: Task B (8h remaining)
Outcome:
✓ Security vulnerability patched same day
✓ Payment gateway on track for deadline
✓ Technical debt addressed (didn't starve due to aging)
✓ Minor UI fix completed when time allowed
✓ Parallel execution optimized throughput
Active Incidents:
Incident 1: "Database connection pool exhausted"
- Severity: P1 (Critical)
- Affected: 80% of users experiencing errors
- Impact: Revenue loss $5K/hour
- Estimated fix time: 30 min
- Deadline pressure: HIGH (ongoing)
- Priority: 9.5
Incident 2: "Slow page load on marketing site"
- Severity: P3 (Low)
- Affected: 5% of visitors notice
- Impact: Potential lead loss
- Estimated fix time: 2 hours
- Deadline pressure: LOW
- Priority: 3.2
Incident 3: "Email notifications delayed"
- Severity: P2 (Medium)
- Affected: All users (non-blocking)
- Impact: User experience degradation
- Estimated fix time: 1 hour
- Deadline pressure: MEDIUM (SLA: 4 hours)
- Priority: 6.5
Incident 4: "API rate limiting too aggressive"
- Severity: P2 (Medium)
- Affected: 10 enterprise API customers
- Impact: Customer escalations, contract risk
- Estimated fix time: 15 min (config change)
- Deadline pressure: MEDIUM (SLA: 2 hours)
- Priority: 7.8
Dynamic Rebalancing:
T=0: Prioritize Incident 1 (9.5) - Critical production issue
T=30min: Incident 1 resolved
Next: Incident 4 (7.8) - Quick win, high customer impact
T=45min: Incident 4 resolved
Next: Incident 3 (6.5) - SLA deadline approaching
T=1h45min: Incident 3 resolved
Next: Incident 2 (3.2 → 4.7 with aging)
T=3h45min: Incident 2 resolved
Outcome:
✓ Production restored in 30 min
✓ Enterprise customers unblocked quickly
✓ All SLAs met
✓ No incidents starved
✓ Efficient use of on-call engineer time
Use machine learning to predict task priority and adjust proactively:
class PredictivePrioritizer:
"""ML-based priority prediction"""
def __init__(self):
self.model = self.train_priority_model()
def train_priority_model(self):
"""Train on historical data: features → actual priority"""
# Features: task metadata, time, context
# Labels: final priority score after completion
# Learn patterns like "similar tasks escalated" or "VIP customers"
pass
def predict_priority(self, task: Task) -> float:
"""Predict priority using learned patterns"""
features = self.extract_features(task)
predicted_priority = self.model.predict(features)
return predicted_priority
def adjust_for_trends(self, task: Task) -> float:
"""Factor in current trends and anomalies"""
base_priority = task.priority()
predicted = self.predict_priority(task)
# Blend rule-based and ML-based priorities
return 0.7 * base_priority + 0.3 * predictedOptimize across multiple constraints simultaneously:
from ortools.sat.python import cp_model
class ConstraintBasedScheduler:
"""Use constraint programming for optimal scheduling"""
def schedule_tasks(self, tasks: List[Task], resources: List[Resource],
constraints: List[Constraint]) -> Schedule:
"""
Find optimal schedule satisfying all constraints:
- Resource capacity limits
- Task dependencies
- Deadline requirements
- Priority optimization
"""
model = cp_model.CpModel()
# Variables: start time for each task
starts = {t.id: model.NewIntVar(0, HORIZON, f'start_{t.id}')
for t in tasks}
# Constraints
for task in tasks:
# Resource capacity
model.Add(resource_usage <= capacity)
# Dependencies
for dep in task.dependencies:
model.Add(starts[task.id] >= starts[dep] + duration[dep])
# Deadlines
if task.deadline:
model.Add(starts[task.id] + task.duration <= task.deadline)
# Objective: Maximize weighted priority completion
model.Maximize(sum(priority[t] * completed[t] for t in tasks))
# Solve
solver = cp_model.CpSolver()
status = solver.Solve(model)
return self.extract_schedule(solver, starts)Balance competing objectives (throughput vs. fairness vs. deadlines):
class MultiObjectivePrioritizer:
"""Balance multiple goals using Pareto optimization"""
def __init__(self, objectives: Dict[str, float]):
"""
objectives: weights for different goals
{
"throughput": 0.3, # Maximize tasks completed
"value": 0.4, # Maximize business value
"fairness": 0.2, # Minimize max wait time
"deadline_adherence": 0.1 # Meet deadlines
}
"""
self.objectives = objectives
def prioritize(self, tasks: List[Task]) -> List[Task]:
"""
Find Pareto-optimal schedule balancing all objectives
"""
# Score each task on each objective
scores = {task: self.multi_objective_score(task) for task in tasks}
# Sort by composite score
return sorted(tasks, key=lambda t: scores[t], reverse=True)
def multi_objective_score(self, task: Task) -> float:
"""Weighted combination of objectives"""
return (
self.objectives["throughput"] * task.throughput_contribution() +
self.objectives["value"] * task.business_value() +
self.objectives["fairness"] * task.fairness_score() +
self.objectives["deadline_adherence"] * task.deadline_score()
)Automatically tune priority weights based on outcomes:
class AdaptivePrioritizer:
"""Learn optimal weights from feedback"""
def __init__(self):
self.weights = {"urgency": 0.3, "impact": 0.3, "effort": 0.2, "aging": 0.2}
self.performance_history = []
def update_weights(self, completed_tasks: List[Task], metrics: Dict):
"""
Adjust weights based on outcomes:
- Did high-priority tasks actually matter?
- Were deadlines met?
- Was throughput optimal?
- Customer satisfaction scores
"""
# Reinforcement learning approach
if metrics["sla_compliance"] < 0.95:
# Missed deadlines → increase urgency/deadline weight
self.weights["urgency"] += 0.05
self.weights["impact"] -= 0.05
if metrics["starvation_count"] > 5:
# Tasks starving → increase aging weight
self.weights["aging"] += 0.05
self.weights["urgency"] -= 0.05
if metrics["value_per_hour"] < target:
# Low value delivery → increase impact weight
self.weights["impact"] += 0.05
self.weights["effort"] -= 0.05
# Normalize weights to sum to 1.0
total = sum(self.weights.values())
self.weights = {k: v/total for k, v in self.weights.items()}
self.performance_history.append({
"timestamp": datetime.now(),
"weights": self.weights.copy(),
"metrics": metrics
})| Pattern | Focus | Scheduling | Adaptation | When to Use |
|---|---|---|---|---|
| Prioritization | Task ranking | Multi-criteria priority | Dynamic rebalancing | Competing demands, limited resources |
| Goal Management | Goal decomposition | Dependency-based | Replanning on failure | Complex multi-step objectives |
| Agent Communication | Coordination | Negotiated allocation | Message-driven | Multi-agent systems |
| Planning | Strategy creation | Upfront plan | Replanning cycles | Known problem space |
| Load Balancing | Resource distribution | Round-robin, least-loaded | Reactive to load | Stateless parallel work |
Key Differences:
Prioritization vs. Goal Management:
- Prioritization: Flat list of tasks, ranked by priority
- Goal Management: Hierarchical goals with dependencies
Prioritization vs. Planning:
- Prioritization: Ongoing, continuous decision-making
- Planning: Upfront strategy, then execute plan
Prioritization vs. Load Balancing:
- Prioritization: Tasks have different values/priorities
- Load Balancing: All requests equal, distribute evenly
Problem: 20 priority levels makes decisions complex and meaningless
Solution: Use 3-5 levels (Critical, High, Medium, Low, Deferred)
# Bad: Too granular
priority_levels = range(1, 21) # 20 levels
# Good: Clear categories
class PriorityLevel(Enum):
CRITICAL = 10 # P0: Production down
HIGH = 7 # P1: Important deadline
MEDIUM = 5 # P2: Normal work
LOW = 2 # P3: Nice to have
DEFERRED = 0 # P4: BacklogProblem: High-priority task blocked by low-priority prerequisite
Solution: Use priority inheritance or boost blocking tasks
def adjust_for_dependencies(task: Task, queue: PriorityQueue) -> float:
"""Boost priority of blocking tasks"""
if task.is_blocking(high_priority_tasks=queue.get_high_priority()):
# Inherit priority from highest task blocked
max_blocked_priority = max(t.priority() for t in task.blocks)
return max(task.priority(), max_blocked_priority)
return task.priority()Problem: Same priority weights used for incidents vs. feature development
Solution: Use context-specific weight profiles
# Different weights for different contexts
CONTEXTS = {
"incident_response": {
"urgency": 0.6, # Time matters most
"impact": 0.3,
"effort": 0.1,
},
"feature_development": {
"impact": 0.4, # Value matters most
"strategic_value": 0.3,
"effort": 0.2, # Prefer quick wins
"urgency": 0.1,
},
"support_tickets": {
"urgency": 0.3,
"customer_tier": 0.3, # VIP customers matter
"impact": 0.2,
"effort": 0.2,
}
}Problem: Low-priority tasks wait indefinitely despite aging
Solution: Set hard limits on wait time
def enforce_max_wait_time(queue: PriorityQueue, max_wait_minutes: int = 480):
"""Force-promote tasks waiting too long"""
now = time.time()
for task in queue.tasks:
wait_time = (now - task.arrival_time) / 60 # minutes
if wait_time >= max_wait_minutes:
# Override priority - force to front of queue
task.override_priority = 999.0
logger.warning(f"Task {task.id} forced to front after {wait_time:.0f} min wait")Problem: Constantly recalculating priorities and reordering queue
Solution: Batch updates, rate-limit rebalancing
class RateLimitedRebalancer:
"""Prevent excessive rebalancing"""
def __init__(self, min_interval_seconds: int = 60):
self.last_rebalance = 0
self.min_interval = min_interval_seconds
self.pending_changes = []
def request_rebalance(self, reason: str):
"""Queue rebalance request"""
self.pending_changes.append((time.time(), reason))
def maybe_rebalance(self, queue: PriorityQueue):
"""Rebalance if enough time passed and changes pending"""
now = time.time()
if now - self.last_rebalance < self.min_interval:
return # Too soon
if not self.pending_changes:
return # No changes
# Perform rebalance
queue.recalculate_priorities()
self.last_rebalance = now
self.pending_changes.clear()Problem: Everyone marks their tasks as urgent to jump the queue
Solution: Validate priorities, require justification, use system-calculated scores
class PriorityValidator:
"""Prevent priority gaming"""
def validate_priority(self, task: Task, user_claimed: int) -> int:
"""
Cross-check user-claimed priority with objective criteria
"""
# Calculate system priority based on objective data
system_priority = self.calculate_objective_priority(task)
# Allow user to boost within reason
max_boost = 2.0
if user_claimed > system_priority + max_boost:
logger.warning(f"Priority inflation detected: Task {task.id}, "
f"claimed {user_claimed} vs system {system_priority}")
# Require justification for large deviations
if not task.priority_justification:
return system_priority # Cap at system priority
# Flag for manager review
self.flag_for_review(task, user_claimed, system_priority)
return min(user_claimed, system_priority + max_boost)The Prioritization pattern is essential for making intelligent decisions about resource allocation and task execution order in resource-constrained environments. By systematically assessing urgency, impact, effort, deadlines, and strategic value, systems can maximize throughput, meet SLAs, deliver business value, and ensure fairness across competing demands.
Use Prioritization when:
- Tasks compete for limited resources (time, compute, people)
- Different tasks have different values and urgency levels
- Deadlines and SLAs must be met consistently
- You need to optimize for throughput, value, or fairness
- Starvation prevention is important
- Transparent, explainable prioritization decisions required
Implementation checklist:
- ✅ Define clear, measurable priority criteria (urgency, impact, effort, deadlines)
- ✅ Implement multi-criteria scoring with tunable weights
- ✅ Add aging mechanism to prevent starvation
- ✅ Monitor queue health (depth, wait times, deadline risks)
- ✅ Use appropriate scheduling algorithm (EDF for deadlines, WFQ for fairness)
- ✅ Log priority decisions for analysis and tuning
- ✅ Implement resource-aware allocation
- ✅ Set maximum wait time guarantees
- ✅ Validate and guard against priority gaming
- ✅ Measure and optimize based on business metrics
Key Takeaways:
- 🎯 Prioritization enables optimal resource allocation under constraints
- 📊 Multi-criteria scoring provides explainable, tunable decisions
- ⏰ Deadline-aware scheduling ensures SLA compliance
- ⚖️ Aging mechanisms prevent starvation of low-priority work
- 🔄 Dynamic rebalancing adapts to changing conditions
- 📈 Continuous monitoring and tuning improves outcomes over time
- 🎲 Guard against gaming and priority inflation
- 🔍 Transparency and logging enable data-driven optimization
Prioritization transforms chaotic task queues into strategic, value-optimized execution—ensuring that resources flow to where they matter most, deadlines are met, and no task is left behind indefinitely.
If you're behind a corporate SSL-inspecting proxy, run examples with:
AGENTIC_DISABLE_SSL=1 bash run.sh