Building a Public Transit Information Agent: Route Planning, Delays, and Accessibility

Why Transit Agencies Need AI Agents

Public transit systems are information-heavy. A mid-size city might operate 40 bus routes, a light rail line, and paratransit services — each with different schedules for weekdays, weekends, and holidays. Add real-time delays, service alerts, detours, and accessibility concerns, and the information landscape becomes overwhelming for riders.

Existing transit apps show maps and schedules but cannot answer questions like "I use a wheelchair and need to get from downtown to the hospital by 2 PM — what are my options?" That requires combining schedule data, real-time vehicle positions, accessibility information, and trip planning logic into a conversational interface. This is exactly what an AI agent can do.

Understanding GTFS: The Transit Data Standard

Nearly every public transit agency publishes data in GTFS (General Transit Feed Specification) format. GTFS is a standardized set of CSV files that describe routes, stops, schedules, and geographic shapes. The agent needs to ingest and query this data.

import csv
from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class Stop:
    stop_id: str
    stop_name: str
    latitude: float
    longitude: float
    wheelchair_accessible: bool = False
    shelter: bool = False


@dataclass
class Route:
    route_id: str
    route_name: str
    route_type: str  # bus, rail, ferry, etc.
    stops: list[Stop] = field(default_factory=list)


def load_stops(gtfs_path: Path) -> dict[str, Stop]:
    """Load stops from GTFS stops.txt file."""
    stops = {}
    with open(gtfs_path / "stops.txt", newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            stops[row["stop_id"]] = Stop(
                stop_id=row["stop_id"],
                stop_name=row["stop_name"],
                latitude=float(row["stop_lat"]),
                longitude=float(row["stop_lon"]),
                wheelchair_accessible=row.get("wheelchair_boarding") == "1",
            )
    return stops


def load_routes(gtfs_path: Path) -> dict[str, Route]:
    """Load routes from GTFS routes.txt file."""
    route_types = {
        "0": "light_rail", "1": "subway", "2": "rail",
        "3": "bus", "4": "ferry", "5": "cable_car",
    }
    routes = {}
    with open(gtfs_path / "routes.txt", newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            routes[row["route_id"]] = Route(
                route_id=row["route_id"],
                route_name=row.get("route_long_name", row.get("route_short_name", "")),
                route_type=route_types.get(row["route_type"], "bus"),
            )
    return routes

Real-Time Delay Integration

Static schedules are only half the picture. The agent must also consume GTFS-Realtime feeds that provide live vehicle positions and service alerts.

import httpx
from datetime import datetime, timedelta
from dataclasses import dataclass


@dataclass
class ServiceAlert:
    alert_id: str
    route_ids: list[str]
    header: str
    description: str
    severity: str  # info, warning, severe
    start_time: datetime | None = None
    end_time: datetime | None = None


@dataclass
class TripDelay:
    route_id: str
    trip_id: str
    stop_id: str
    delay_seconds: int
    timestamp: datetime


class RealtimeTransitClient:
    """Client for consuming GTFS-Realtime feeds."""

    def __init__(self, alerts_url: str, updates_url: str):
        self.alerts_url = alerts_url
        self.updates_url = updates_url
        self._alerts_cache: list[ServiceAlert] = []
        self._delays_cache: dict[str, TripDelay] = {}
        self._last_fetch: datetime | None = None

    async def fetch_alerts(self) -> list[ServiceAlert]:
        """Fetch current service alerts from the transit agency."""
        async with httpx.AsyncClient() as client:
            response = await client.get(self.alerts_url)
            data = response.json()

        alerts = []
        for entry in data.get("entity", []):
            alert_data = entry.get("alert", {})
            route_ids = [
                s["route_id"]
                for s in alert_data.get("informed_entity", [])
                if "route_id" in s
            ]
            alerts.append(ServiceAlert(
                alert_id=entry["id"],
                route_ids=route_ids,
                header=alert_data.get("header_text", {})
                    .get("translation", [{}])[0].get("text", ""),
                description=alert_data.get("description_text", {})
                    .get("translation", [{}])[0].get("text", ""),
                severity=alert_data.get("severity_level", "info"),
            ))

        self._alerts_cache = alerts
        return alerts

    def get_alerts_for_route(self, route_id: str) -> list[ServiceAlert]:
        """Get active alerts affecting a specific route."""
        return [a for a in self._alerts_cache if route_id in a.route_ids]

Building the Trip Planner

The trip planner combines static schedule data with real-time delays to suggest routes. Accessibility filtering is built in from the start, not bolted on as an afterthought.

@dataclass
class TripOption:
    departure_time: str
    arrival_time: str
    duration_minutes: int
    transfers: int
    routes_used: list[str]
    walking_minutes: int
    wheelchair_accessible: bool
    alerts: list[str]
    adjusted_for_delays: bool = False


def plan_trip(
    origin: str,
    destination: str,
    departure_after: datetime,
    require_accessible: bool = False,
    max_transfers: int = 2,
    routes: dict = None,
    stops: dict = None,
    realtime: RealtimeTransitClient = None,
) -> list[TripOption]:
    """Plan a transit trip with optional accessibility filtering."""

    # Step 1: Find nearest stops to origin and destination
    origin_stops = find_nearest_stops(origin, stops, radius_meters=400)
    dest_stops = find_nearest_stops(destination, stops, radius_meters=400)

    if require_accessible:
        origin_stops = [s for s in origin_stops if s.wheelchair_accessible]
        dest_stops = [s for s in dest_stops if s.wheelchair_accessible]

        if not origin_stops or not dest_stops:
            return []  # No accessible stops nearby

    # Step 2: Find connecting routes (simplified graph search)
    options = []
    for o_stop in origin_stops[:3]:
        for d_stop in dest_stops[:3]:
            route_options = find_routes_between(
                o_stop, d_stop, departure_after,
                max_transfers=max_transfers,
            )
            for route_opt in route_options:
                # Step 3: Apply real-time delays
                alerts = []
                if realtime:
                    for rid in route_opt["route_ids"]:
                        route_alerts = realtime.get_alerts_for_route(rid)
                        alerts.extend([a.header for a in route_alerts])

                options.append(TripOption(
                    departure_time=route_opt["depart"],
                    arrival_time=route_opt["arrive"],
                    duration_minutes=route_opt["duration"],
                    transfers=route_opt["transfers"],
                    routes_used=route_opt["route_ids"],
                    walking_minutes=route_opt["walk_min"],
                    wheelchair_accessible=route_opt["accessible"],
                    alerts=alerts,
                ))

    # Sort by arrival time
    options.sort(key=lambda x: x.arrival_time)
    return options[:5]


def find_nearest_stops(
    location: str, stops: dict, radius_meters: int = 400
) -> list[Stop]:
    """Find stops within walking distance of a location.
    In production, geocode the location string first."""
    # Placeholder: would use geocoding + haversine distance
    return list(stops.values())[:5]


def find_routes_between(origin, dest, depart_after, max_transfers=2):
    """Graph search through the transit network.
    In production, use a proper routing engine like OpenTripPlanner."""
    return []

Conversational Agent Layer

The conversational layer wraps the trip planner and real-time data into a natural dialogue.

from openai import OpenAI

client = OpenAI()

TRANSIT_AGENT_PROMPT = """You are a public transit information agent.
You help riders plan trips, check delays, and find accessible routes.

You have access to these tools:
- plan_trip(origin, destination, time, accessible): Plan a transit trip
- get_alerts(route_id): Get service alerts for a route
- find_stop(name): Find a transit stop by name

When a rider asks about accessibility, ALWAYS filter for wheelchair-accessible
stops and routes. Never suggest a route that includes an inaccessible stop
to a rider who needs accessibility.

When there are active delays or alerts on a suggested route, proactively
mention them — do not wait for the rider to ask.
"""

The accessibility-first design is essential for government services. The Americans with Disabilities Act requires that transit information systems provide equivalent access. Building accessibility filtering into the agent's core logic — rather than as an optional add-on — ensures compliance and serves all riders.

FAQ

How does the agent handle transit systems that do not publish GTFS data?

For agencies without GTFS feeds, the agent can ingest schedule data from other formats (spreadsheets, PDFs, or web scraping) and normalize it into the same internal data structures. The GTFS loader is just one ingest pathway. The trip planning and real-time layers work against the internal Stop, Route, and TripOption models regardless of the original data source. However, real-time delay information will be unavailable without a live feed, so the agent should clearly inform riders that arrival times are based on published schedules only.

How do you keep the schedule data current when agencies update routes seasonally?

Transit agencies typically publish new GTFS feeds quarterly or when service changes take effect. The agent runs an automated pipeline that checks the agency's GTFS feed URL on a daily schedule, downloads any updates, validates the data, and rebuilds the internal route graph. A version log tracks which GTFS feed is currently active. If a rider asks about a trip on a date that falls after an announced service change, the agent loads the future-effective schedule for planning.

Can the agent suggest multimodal trips combining transit with bikeshare or rideshare?

Yes. The agent extends the trip planner to include first-mile and last-mile options. If the nearest transit stop is more than a 10-minute walk, the agent checks bikeshare station availability via the GBFS (General Bikeshare Feed Specification) standard or suggests a rideshare connection. The trip option clearly labels each segment — "Walk 3 min to bikeshare station, ride 7 min to Metro station, take Blue Line 4 stops, walk 2 min to destination" — so the rider understands the full journey.

---

#GovernmentAI #PublicTransit #GTFS #RoutePlanning #Accessibility #AgenticAI #LearnAI #AIEngineering