Building a 311 Service Request Agent: Citizen Complaint Intake and Routing

Why 311 Systems Need AI Agents

Cities across the United States handle millions of 311 service requests every year. Potholes, broken streetlights, noise complaints, missed trash pickups, and graffiti reports all flow through the same intake system. Traditional 311 centers rely on human operators who manually classify each request, look up the responsible department, and enter details into a work-order system. This process is slow during peak hours, inconsistent across operators, and expensive to scale.

An AI agent can handle the intake front-end: understanding what the citizen is reporting, classifying it into the correct service category, routing it to the appropriate department, and providing real-time status updates. The agent does not replace human workers who fix the pothole — it replaces the manual classification and routing layer that sits between the citizen and the field crew.

Designing the Request Classification System

The foundation of a 311 agent is its ability to classify free-text citizen reports into structured service categories. Cities typically have between 50 and 200 distinct service request types organized into departments. We start by defining this taxonomy.

from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
import uuid


class Department(Enum):
    PUBLIC_WORKS = "public_works"
    SANITATION = "sanitation"
    PARKS = "parks_and_recreation"
    TRANSPORTATION = "transportation"
    CODE_ENFORCEMENT = "code_enforcement"
    UTILITIES = "utilities"
    ANIMAL_CONTROL = "animal_control"
    HEALTH = "health_department"


SERVICE_CATEGORIES = {
    "pothole_repair": {
        "department": Department.PUBLIC_WORKS,
        "priority": "medium",
        "sla_hours": 72,
        "required_fields": ["location", "size_estimate"],
    },
    "streetlight_outage": {
        "department": Department.UTILITIES,
        "priority": "medium",
        "sla_hours": 48,
        "required_fields": ["location", "pole_number"],
    },
    "missed_trash_pickup": {
        "department": Department.SANITATION,
        "priority": "high",
        "sla_hours": 24,
        "required_fields": ["location", "pickup_type"],
    },
    "noise_complaint": {
        "department": Department.CODE_ENFORCEMENT,
        "priority": "low",
        "sla_hours": 96,
        "required_fields": ["location", "noise_type", "time_of_occurrence"],
    },
    "graffiti_removal": {
        "department": Department.PUBLIC_WORKS,
        "priority": "low",
        "sla_hours": 120,
        "required_fields": ["location", "surface_type"],
    },
    "stray_animal": {
        "department": Department.ANIMAL_CONTROL,
        "priority": "high",
        "sla_hours": 4,
        "required_fields": ["location", "animal_type", "behavior"],
    },
}

Each category maps to a department, carries a default priority level, defines SLA (service level agreement) hours for resolution, and lists the fields the agent must collect from the citizen before the request can be dispatched.

Building the Agent Core

The agent uses an LLM to interpret the citizen's description and map it to the correct service category. Once classified, it collects any missing required fields through follow-up questions.

from openai import OpenAI
import json

client = OpenAI()

CLASSIFICATION_PROMPT = """You are a 311 service request classifier for a city government.

Given a citizen's description of their issue, classify it into exactly one
of these categories: {categories}

Respond with JSON containing:
- "category": the matching category key
- "confidence": float between 0 and 1
- "extracted_fields": dict of any fields you can extract from the description
- "missing_fields": list of required fields not found in the description

If no category matches with confidence above 0.6, set category to "unknown".
"""


@dataclass
class ServiceRequest:
    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    category: str = ""
    department: Department | None = None
    priority: str = "medium"
    description: str = ""
    location: str = ""
    fields: dict = field(default_factory=dict)
    status: str = "open"
    created_at: datetime = field(default_factory=datetime.utcnow)
    sla_deadline: datetime | None = None


def classify_request(citizen_description: str) -> dict:
    """Classify a citizen's free-text report into a service category."""
    categories_list = ", ".join(SERVICE_CATEGORIES.keys())
    category_details = json.dumps(SERVICE_CATEGORIES, indent=2, default=str)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": CLASSIFICATION_PROMPT.format(
                    categories=category_details
                ),
            },
            {"role": "user", "content": citizen_description},
        ],
        response_format={"type": "json_object"},
        temperature=0.1,
    )

    return json.loads(response.choices[0].message.content)

Low temperature is critical here. Classification should be deterministic — the same pothole report should always route to public works, not occasionally to transportation.

Routing and SLA Management

Once classified, the agent creates a formal service request, assigns it to the correct department, and calculates the SLA deadline.

from datetime import timedelta


def create_service_request(
    description: str, classification: dict
) -> ServiceRequest:
    """Create a routed service request from classification results."""
    category_key = classification["category"]
    category_config = SERVICE_CATEGORIES.get(category_key)

    if not category_config:
        return ServiceRequest(
            description=description,
            status="needs_manual_review",
        )

    now = datetime.utcnow()
    request = ServiceRequest(
        category=category_key,
        department=category_config["department"],
        priority=category_config["priority"],
        description=description,
        fields=classification.get("extracted_fields", {}),
        sla_deadline=now + timedelta(hours=category_config["sla_hours"]),
    )

    # Check for priority escalation triggers
    request = check_priority_escalation(request)
    return request


def check_priority_escalation(request: ServiceRequest) -> ServiceRequest:
    """Escalate priority based on safety-critical keywords."""
    safety_keywords = [
        "dangerous", "hazard", "injury", "child",
        "flooding", "gas leak", "exposed wire",
    ]
    desc_lower = request.description.lower()

    if any(keyword in desc_lower for keyword in safety_keywords):
        request.priority = "critical"
        request.sla_deadline = request.created_at + timedelta(hours=2)

    return request

The escalation logic is important for public safety. A pothole report that mentions "dangerous" or "injury" should not wait 72 hours in the queue. The agent automatically promotes it to critical priority with a 2-hour SLA.

Status Tracking and Follow-Up

Citizens expect to know what happened with their request. The agent provides status lookup and automated follow-up.

# In-memory store for demo; use a database in production
REQUEST_STORE: dict[str, ServiceRequest] = {}


def track_status(request_id: str) -> dict:
    """Look up current status of a service request."""
    request = REQUEST_STORE.get(request_id)
    if not request:
        return {"error": "Request not found", "request_id": request_id}

    hours_remaining = None
    if request.sla_deadline:
        delta = request.sla_deadline - datetime.utcnow()
        hours_remaining = max(0, delta.total_seconds() / 3600)

    return {
        "request_id": request.id,
        "category": request.category,
        "department": request.department.value if request.department else None,
        "status": request.status,
        "priority": request.priority,
        "sla_hours_remaining": round(hours_remaining, 1) if hours_remaining else None,
        "created_at": request.created_at.isoformat(),
    }

FAQ

How does the agent handle requests that do not fit any predefined category?

When the classification confidence falls below 0.6 or the LLM returns "unknown," the agent creates the request with a status of needs_manual_review and routes it to a general intake queue. A human operator reviews it, classifies it manually, and the system learns from that correction over time. The goal is not 100% automation — it is automating the 80% of requests that fit known patterns so operators can focus on the ambiguous 20%.

What happens when a citizen reports multiple issues in one message?

The agent should detect multi-issue reports during classification and split them into separate service requests. For example, "There is a pothole on Main Street and the streetlight on the corner is out" produces two requests: one for pothole repair routed to public works, and one for streetlight outage routed to utilities. Each gets its own tracking ID and SLA.

How do you prevent duplicate 311 requests for the same issue?

The agent performs geographic and temporal deduplication. Before creating a new request, it searches existing open requests within a configurable radius (e.g., 50 meters) for the same category. If a match is found, the agent adds the new report as a "me too" confirmation on the existing request, which can escalate its priority without creating duplicate work orders.

---

#GovernmentAI #311Services #CitizenServices #RequestRouting #PublicSector #AgenticAI #LearnAI #AIEngineering