Priority Ranking System

Overview

The Priority Ranking System takes analyzed emergency calls and assigns them a weight score that determines their position in the dispatcher queue. Higher weight = higher priority.

Core Algorithm

# Source: app/ranking/ranking.py:58-94
def build_ranking(inputs: RankingInputs) -> Dict:
    """
    Collapse all triage signals into a single sortable 'weight'.
    Higher weight = higher priority in the queue.
    """
    
    # 1) Base risk from LIVE distress/emotion
    base_risk = RISK_LEVEL_WEIGHT.get(inputs.risk_level.upper(), 1)
    
    # 2) Category importance (EMS > FIRE > POLICE > OTHER by default)
    cat_weight = CATEGORY_PRIORITY.get(inputs.category.upper(), 1)
    
    # 3) Tags: sum up "dangerous" entities
    tag_weight = 0
    for t in inputs.tags or []:
        tag_weight += TAG_WEIGHTS.get(t.upper(), 0)
    
    # 4) Combine into one scalar.
    weight = (
        base_risk * 20         # CRITICAL vs LOW really matters
        + cat_weight * 10      # EMS vs OTHER matters a lot
        + tag_weight * 1.5     # multiple strong tags stack up
        + inputs.risk_score * 10  # continuous distress signal fine-tunes ordering
    )
    
    created_at = inputs.created_at or datetime.now(timezone.utc).isoformat()
    
    return {
        "weight": float(weight),
        "score": float(inputs.risk_score),
        "risk_level": inputs.risk_level,
        "category": inputs.category,
        "tags": list(inputs.tags or []),
        "created_at": created_at,
    }

Input Data

# Source: app/ranking/ranking.py:49-55
@dataclass
class RankingInputs:
    risk_level: str       # "CRITICAL", "ELEVATED", "NORMAL", "LOW"
    risk_score: float     # 0..1 (from compute_risk_level)
    category: str         # "EMS", "FIRE", "POLICE", "OTHER"
    tags: List[str]       # e.g. ["TRAUMA", "VIOLENCE"]
    created_at: Optional[str] = None  # ISO string

risk_level

string

required

Categorical risk classification derived from the audio distress score:

"CRITICAL": distress ≥ 0.7 (highly distressed)
"ELEVATED": distress ≥ 0.4 (distressed)
"NORMAL": distress ≥ 0.2 (tense)
"LOW": distress < 0.2 (calm)

risk_score

float

required

Continuous distress score (0.0 - 1.0) from the audio analysis track.Used to fine-tune ordering within the same risk level.

Weight Calculation

The final weight is computed as a weighted sum of four components:

1. Base Risk Weight (×20)

# Source: app/ranking/ranking.py:40-46
RISK_LEVEL_WEIGHT: Dict[str, int] = {
    "CRITICAL": 4,
    "ELEVATED": 3,
    "NORMAL": 2,
    "LOW": 1,
}

base_risk = RISK_LEVEL_WEIGHT.get(inputs.risk_level.upper(), 1)
weight += base_risk * 20

The ×20 multiplier ensures that risk level dominates the ranking. A CRITICAL call (4 × 20 = 80 points) will almost always outrank a LOW call (1 × 20 = 20 points).

2. Category Priority (×10)

# Source: app/ranking/ranking.py:10-15
CATEGORY_PRIORITY: Dict[str, int] = {
    "EMS": 5,
    "FIRE": 4,
    "POLICE": 3,
    "OTHER": 1,
}

cat_weight = CATEGORY_PRIORITY.get(inputs.category.upper(), 1)
weight += cat_weight * 10

Priority order: EMS > FIRE > POLICE > OTHER

Medical emergencies (EMS) are weighted higher than other categories by default, since they typically involve immediate life-threatening situations.

3. Tag Weights (×1.5)

# Source: app/ranking/ranking.py:17-38
TAG_WEIGHTS: Dict[str, int] = {
    # life-threatening medical
    "NOT_BREATHING": 10,
    "CARDIAC_ARREST": 10,
    "OVERDOSE": 9,
    "UNCONSCIOUS": 8,
    "SEIZURE": 7,
    "BLEEDING": 7,
    # trauma / violence
    "GUNSHOT": 9,
    "STABBING": 8,
    "TRAUMA": 7,
    "VIOLENCE": 7,
    "ASSAULT": 6,
    # fire / hazards
    "FIRE": 8,
    "SMOKE": 7,
    "EXPLOSION": 7,
    # catch-alls
    "MENTAL_HEALTH": 5,
    "DISTURBANCE": 3,
}

tag_weight = 0
for t in inputs.tags or []:
    tag_weight += TAG_WEIGHTS.get(t.upper(), 0)
weight += tag_weight * 1.5

Tags stack additively. A call with ["CARDIAC_ARREST", "NOT_BREATHING"] gets:

(10 + 10) × 1.5 = 30 points

Tag Stacking

Multiple high-severity tags compound the urgency. This ensures complex emergencies (e.g., shooting + major bleeding) rank higher than single-issue calls.

4. Continuous Risk Score (×10)

weight += inputs.risk_score * 10

The raw distress score (0.0 - 1.0) adds up to 10 points. This fine-tunes ordering between calls with the same risk level and category. Example: Two CRITICAL EMS calls:

Call A: risk_score = 0.95 → +9.5 points
Call B: risk_score = 0.73 → +7.3 points

Call A ranks slightly higher.

Weight Examples

Example 1: Cardiac Arrest (Maximum Priority)

inputs = RankingInputs(
    risk_level="CRITICAL",
    risk_score=0.95,
    category="EMS",
    tags=["CARDIAC_ARREST", "NOT_BREATHING"]
)

Calculation:

base_risk = 4 × 20 = 80
cat_weight = 5 × 10 = 50
tag_weight = (10 + 10) × 1.5 = 30
risk_score = 0.95 × 10 = 9.5

weight = 80 + 50 + 30 + 9.5 = 169.5

Example 2: Noise Complaint (Low Priority)

inputs = RankingInputs(
    risk_level="LOW",
    risk_score=0.12,
    category="POLICE",
    tags=["NOISE_COMPLAINT"]
)

Calculation:

base_risk = 1 × 20 = 20
cat_weight = 3 × 10 = 30
tag_weight = 0 × 1.5 = 0       # NOISE_COMPLAINT not in TAG_WEIGHTS
risk_score = 0.12 × 10 = 1.2

weight = 20 + 30 + 0 + 1.2 = 51.2

Example 3: House Fire (High Priority)

inputs = RankingInputs(
    risk_level="CRITICAL",
    risk_score=0.88,
    category="FIRE",
    tags=["FIRE", "SMOKE", "TRAPPED"]
)

Calculation:

base_risk = 4 × 20 = 80
cat_weight = 4 × 10 = 40
tag_weight = (8 + 7 + 0) × 1.5 = 22.5   # TRAPPED not in TAG_WEIGHTS
risk_score = 0.88 × 10 = 8.8

weight = 80 + 40 + 22.5 + 8.8 = 151.3

Output Structure

{
    "weight": 169.5,
    "score": 0.95,
    "risk_level": "CRITICAL",
    "category": "EMS",
    "tags": ["CARDIAC_ARREST", "NOT_BREATHING"],
    "created_at": "2026-03-03T23:15:42.123Z"
}

weight

float

The sortable priority score. Higher values appear first in the dispatch queue.

score

float

Original continuous distress score (0.0 - 1.0) for reference.

risk_level

string

Categorical risk level: "CRITICAL", "ELEVATED", "NORMAL", or "LOW"

Tuning the Algorithm

The ranking weights are exposed as tunable constants at the top of the file:

# Source: app/ranking/ranking.py:8-46
# --- knobs that can be tweaked -----------------------------------------------

CATEGORY_PRIORITY: Dict[str, int] = {
    "EMS": 5,
    "FIRE": 4,
    "POLICE": 3,
    "OTHER": 1,
}

TAG_WEIGHTS: Dict[str, int] = {
    "NOT_BREATHING": 10,
    "CARDIAC_ARREST": 10,
    "OVERDOSE": 9,
    # ... 30+ more tags
}

RISK_LEVEL_WEIGHT: Dict[str, int] = {
    "CRITICAL": 4,
    "ELEVATED": 3,
    "NORMAL": 2,
    "LOW": 1,
}

Adjusting Multipliers: The formula uses base_risk * 20, cat_weight * 10, and tag_weight * 1.5. These are “vibes-based” defaults. Tweak them based on real-world dispatcher feedback.

Adding New Tags

To recognize a new emergency tag:

TAG_WEIGHTS: Dict[str, int] = {
    # ... existing tags
    "HYPOTHERMIA": 7,           # Add new medical tag
    "ACTIVE_THREAT": 9,         # Add new security tag
}

The service classifier must also be updated to detect these tags from transcripts.

Queue Ordering

Calls are sorted descending by weight:

# Source: app/ranking/service.py (conceptual)
def get_queue() -> List[dict]:
    return sorted(_queue, key=lambda c: c["ranking"]["weight"], reverse=True)

Result: Dispatcher always sees highest-priority calls first.

Edge Cases

Ties

If two calls have identical weights, the created_at timestamp is used as a tiebreaker (older calls first).

Missing Tags

If a tag isn’t in TAG_WEIGHTS, it contributes 0 points:

tag_weight += TAG_WEIGHTS.get(t.upper(), 0)  # Returns 0 if not found

This ensures the system gracefully handles new/unknown tags without crashing.

Multi-Service Emergencies

Some calls need multiple services (e.g., active shooter = EMS + POLICE). The classifier assigns a primary category, and tags indicate the secondary needs:

{
  "category": "POLICE",
  "tags": ["ACTIVE_SHOOTER", "TRAUMA", "VIOLENCE"],
  # TRAUMA tag indicates EMS also needed
}

Real-World Behavior

Cardiac arrest vs. house fire

Cardiac arrest:

CRITICAL (4 × 20 = 80) + EMS (5 × 10 = 50) + tags (20 × 1.5 = 30) = ~170

House fire:

CRITICAL (4 × 20 = 80) + FIRE (4 × 10 = 40) + tags (15 × 1.5 = 22) = ~150

Result: Cardiac arrest ranks slightly higher due to higher category priority (EMS > FIRE).

High distress but unclear emergency

If the caller is highly distressed but the transcript is unclear:

risk_level = "CRITICAL" (80 points from audio)
category = "OTHER" (10 points, low confidence)
tags = [] (0 points, no keywords detected)

Weight: ~90-100 pointsThis still ranks above most NORMAL/LOW calls, ensuring dispatcher attention.

Multiple life-threatening tags

A call with shooting + major bleeding + unconscious:

tags = ["GUNSHOT", "MAJOR_BLEEDING", "UNCONSCIOUS"]
tag_weight = (9 + 7 + 8) × 1.5 = 36 points

This compounding effect ensures complex emergencies get prioritized.

Integration Example

Complete flow from CallPacket to ranked queue entry:

from app.schemas.call_packet import CallPacket
from app.ranking.ranking import build_ranking, RankingInputs
from app.agents.service_classify import classify_service_and_tags

# 1. Get analyzed call packet
packet: CallPacket = ...  # from pipeline

# 2. Classify service and tags
classification = classify_service_and_tags(
    transcript=packet.nlp.transcript,
    distress=packet.audio.distress_score
)

# 3. Determine risk level from distress score
if packet.audio.distress_score >= 0.7:
    risk_level = "CRITICAL"
elif packet.audio.distress_score >= 0.4:
    risk_level = "ELEVATED"
elif packet.audio.distress_score >= 0.2:
    risk_level = "NORMAL"
else:
    risk_level = "LOW"

# 4. Build ranking
ranking = build_ranking(RankingInputs(
    risk_level=risk_level,
    risk_score=packet.audio.distress_score,
    category=classification["category"],
    tags=classification["tags"],
    created_at=call_timestamp
))

# 5. Add to queue
queue_entry = {
    "call_id": packet.call_id,
    "packet": packet.dict(),
    "classification": classification,
    "ranking": ranking
}
add_to_queue(queue_entry)

CallPacket Structure

See the data structure that feeds into ranking

Streaming Pipeline

Learn how distress scores and tags are computed

Get Started

Core Concepts

Guides

Priority Ranking System

Overview

Core Algorithm

Input Data

Weight Calculation

1. Base Risk Weight (×20)

2. Category Priority (×10)

3. Tag Weights (×1.5)

Tag Stacking

4. Continuous Risk Score (×10)

Weight Examples

Example 1: Cardiac Arrest (Maximum Priority)

Example 2: Noise Complaint (Low Priority)

Example 3: House Fire (High Priority)

Output Structure

Tuning the Algorithm

Adding New Tags

Queue Ordering

Edge Cases

Ties

Missing Tags

Multi-Service Emergencies

Real-World Behavior

Integration Example

CallPacket Structure

Streaming Pipeline

Build docs developers (and LLMs) love

Get Started

Core Concepts

Guides

​Overview

​Core Algorithm

​Input Data

​Weight Calculation

​1. Base Risk Weight (×20)

​2. Category Priority (×10)

​3. Tag Weights (×1.5)

Tag Stacking

​4. Continuous Risk Score (×10)

​Weight Examples

​Example 1: Cardiac Arrest (Maximum Priority)

​Example 2: Noise Complaint (Low Priority)

​Example 3: House Fire (High Priority)

​Output Structure

​Tuning the Algorithm

​Adding New Tags

​Queue Ordering

​Edge Cases

​Ties

​Missing Tags

​Multi-Service Emergencies

​Real-World Behavior

​Integration Example

​Related Concepts

CallPacket Structure

Streaming Pipeline

Build docs developers (and LLMs) love

Overview

Core Algorithm

Input Data

Weight Calculation

1. Base Risk Weight (×20)

2. Category Priority (×10)

3. Tag Weights (×1.5)

4. Continuous Risk Score (×10)

Weight Examples

Example 1: Cardiac Arrest (Maximum Priority)

Example 2: Noise Complaint (Low Priority)

Example 3: House Fire (High Priority)

Output Structure

Tuning the Algorithm

Adding New Tags

Queue Ordering

Edge Cases

Ties

Missing Tags

Multi-Service Emergencies

Real-World Behavior

Integration Example

Related Concepts