Build a Robotaxi Fleet Operations & Safety Monitoring Workflow with LangGraph

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Introduction

On August 14, 2026, the autonomous-mobility story took its biggest commercial step yet: Uber and Pony.ai are preparing to put more than 2,000 robotaxis on European roads. The latest AI news coverage has been tracking autonomous fleets from pilots to scale all year, and 2,000 vehicles in commercial service is the scale where operations stop being a demo and become a discipline. A fleet that size generates continuous telemetry, dispatch decisions, service events, and safety conditions — far more than human operators can track raw. It is a multi-agent problem in the most literal sense: dispatch agents, telemetry agents, safety monitors, and incident responders all need to coordinate in real time, with hard gates between autonomous action and human escalation.

This dispatch builds the operations layer that scale demands: a LangGraph fleet operations workflow, fleet-ops, that ingests vehicle telemetry, monitors each vehicle against a safety envelope, routes service events to the right response lane, escalates safety-critical conditions to a human operations center, and keeps a complete audit trail of every fleet decision. The same orchestration discipline we document across the AI workflows library — declare state, monitor against envelopes, isolate failures, escalate to humans — is exactly what a commercial robotaxi fleet needs.

The operations problem at 2,000 vehicles

A single robotaxi is an engineering marvel; 2,000 are an operations problem. Every vehicle streams position, speed, battery, sensor health, and safety state, continuously. Every minute, the fleet generates dispatch decisions — which vehicle takes which trip, when it returns to charge, when it enters service. Every day, it generates service events — sensor cleaning, tire pressure, software updates, edge-case disengagements. And occasionally, it generates something safety-critical — an envelope violation, a sensor failure in traffic, a control anomaly — where the response time and the decision quality both matter enormously.

The human answer to that volume is a fleet operations center with dashboards and alarms. The 2026 answer is an operations layer that does the routine work — monitoring, triage, dispatch routing — automatically, and escalates the safety-critical work to humans with full context. That division of labor is the same pattern that runs through every agentic-operations system: agents handle the volume and the routine; humans own the judgment calls. The workflow in this dispatch is the reference implementation for a robotaxi fleet, and the pattern transfers to any large autonomous fleet — delivery robots, warehouse vehicles, aerial drones.

Architecture overview

graph TD
  subgraph Telemetry[Vehicle Telemetry]
    T1[Fleet Stream] --> T2[Telemetry Normalizer]
    T2 --> T3[(Fleet Store)]
  end
  T3 --> M1[Safety Envelope Monitor]
  M1 --> C1{Event Class}
  C1 -->|Routine| L1[Maintenance Lane]
  C1 -->|Anomaly| L2[Diagnostics Lane]
  C1 -->|Safety-Critical| H1[Human Ops Center]
  L1 --> D1[Schedule]
  L2 --> D2[Diagnose & Repair]
  H1 --> D3[Manual Override]
  D1 --> A1[(Fleet Audit Log)]
  D2 --> A1
  D3 --> A1

The pipeline has five stages. Stage one — the telemetry normalizer ingests the fleet stream into a consistent per-vehicle state. Stage two — the safety envelope monitor checks each vehicle against its operating boundaries. Stage three — the event classifier routes events into three lanes: routine maintenance, telemetry anomalies, and safety-critical escalations. Stage four — each lane executes its response: scheduling for maintenance, diagnostics for anomalies, and human-operator handoff for anything safety-critical. Stage five — every decision lands in the fleet audit log. The design goal: routine operations run without humans; safety decisions never do.

Part 1 — The fleet schema

.env

FLEET_DB_URL=postgresql://ops:secret@pg-fleet.internal/fleet_ops
SPEED_ENVELOPE_PCT=1.05
BATTERY_MIN_PCT=15
SENSOR_HEALTH_MIN=0.9
OPS_CENTER_CHANNEL=#fleet-ops-critical
TELEMETRY_INTERVAL_SEC=5

schemas.py

from pydantic import BaseModel, Field
from typing import List, Literal
from datetime import datetime

class VehicleState(BaseModel):
    vehicle_id: str
    lat: float
    lon: float
    speed_kmh: float
    speed_limit_kmh: float
    battery_pct: float
    sensor_health: float
    control_status: Literal["autonomous", "manual", "degraded", "fault"]
    last_seen: datetime

class FleetEvent(BaseModel):
    event_id: str
    vehicle_id: str
    event_type: Literal["routine", "anomaly", "safety"]
    description: str
    severity: float                       # 0..1
    raw: dict = Field(default_factory=dict)
    created_at: datetime

class DispatchDecision(BaseModel):
    decision_id: str
    vehicle_id: str
    lane: Literal["maintenance", "diagnostics", "human_ops"]
    action: str
    requires_human: bool
    status: Literal["pending", "executed", "escalated"]
    created_at: datetime

VehicleState is the normalized per-vehicle snapshot — position, speed, battery, sensor health, control status. FleetEvent is the classifier's output with severity and a typed event class. DispatchDecision is the routing record, and requires_human is the field that enforces the safety gate: a safety event's decision is created with requires_human=True by construction, so the workflow cannot auto-execute it. The schema discipline mirrors the MCP directory guidance: small stable types, explicit dimensions, and the safety property encoded in the data model, not just in a prompt.

Part 2 — The envelope monitor and event classifier

tools.py

import httpx, os, json, math

def envelope_ok(v: VehicleState) -> bool:
    """Check the vehicle against its safety envelope."""
    if v.speed_kmh > v.speed_limit_kmh * float(os.environ["SPEED_ENVELOPE_PCT"]):
        return False
    if v.battery_pct < float(os.environ["BATTERY_MIN_PCT"]):
        return False
    if v.sensor_health < float(os.environ["SENSOR_HEALTH_MIN"]):
        return False
    if v.control_status in ("degraded", "fault"):
        return False
    return True

def classify_event(v: VehicleState, ok: bool) -> FleetEvent:
    """Classify a vehicle's state into a fleet event."""
    if not ok or v.control_status == "fault":
        return FleetEvent(event_id=f"ev-{v.vehicle_id}", vehicle_id=v.vehicle_id,
                          event_type="safety", severity=1.0,
                          description=f"safety envelope violation: {v.control_status}",
                          created_at=datetime.utcnow())
    if v.control_status == "degraded" or v.sensor_health < 0.95:
        return FleetEvent(event_id=f"ev-{v.vehicle_id}", vehicle_id=v.vehicle_id,
                          event_type="anomaly", severity=0.6,
                          description="sensor or control degradation",
                          created_at=datetime.utcnow())
    if v.battery_pct < 25 or v.speed_kmh < 5:
        return FleetEvent(event_id=f"ev-{v.vehicle_id}", vehicle_id=v.vehicle_id,
                          event_type="routine", severity=0.3,
                          description="maintenance or charging window",
                          created_at=datetime.utcnow())
    return None

def dispatch(ev: FleetEvent) -> DispatchDecision:
    """Route an event to its response lane."""
    if ev.event_type == "safety":
        return DispatchDecision(decision_id=f"d-{ev.event_id}", vehicle_id=ev.vehicle_id,
                                lane="human_ops", action="escalate to ops center",
                                requires_human=True, created_at=datetime.utcnow())
    if ev.event_type == "anomaly":
        return DispatchDecision(decision_id=f"d-{ev.event_id}", vehicle_id=ev.vehicle_id,
                                lane="diagnostics", action="run remote diagnostics",
                                requires_human=False, created_at=datetime.utcnow())
    return DispatchDecision(decision_id=f"d-{ev.event_id}", vehicle_id=ev.vehicle_id,
                            lane="maintenance", action="schedule maintenance",
                            requires_human=False, created_at=datetime.utcnow())

The envelope monitor is the safety backbone: five checks against operating boundaries, any violation returning False. The classifier maps state to events — and note the asymmetry: a fault or envelope violation is always a safety event with severity 1.0, never downgraded by a model's interpretation. The dispatcher then encodes the hard gate: safety events produce decisions with requires_human=True by construction, so the workflow physically cannot auto-execute them. That is the difference between a safety policy stated in a slide deck and one encoded in the data model.

Part 3 — The LangGraph fleet-ops workflow

graph.py

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class FleetState(TypedDict):
    vehicles: List[VehicleState]
    events: List[FleetEvent]
    decisions: List[DispatchDecision]
    escalations: List[Dict]
    executed: List[str]

def ingest(s: FleetState) -> FleetState:
    s["events"] = []
    for v in s["vehicles"]:
        ev = classify_event(v, envelope_ok(v))
        if ev:
            s["events"].append(ev)
    return s

def route(s: FleetState) -> FleetState:
    s["decisions"] = [dispatch(ev) for ev in s["events"]]
    return s

def execute_routine(s: FleetState) -> FleetState:
    for d in s["decisions"]:
        if not d.requires_human:
            s["executed"].append(execute_lane(d))   # schedule / diagnose
            log_fleet(d, "executed")
    return s

def escalate(s: FleetState) -> FleetState:
    s["escalations"] = []
    for d in s["decisions"]:
        if d.requires_human:
            s["escalations"].append(handoff_ops_center(d))  # page with full context
            log_fleet(d, "escalated")
    return s

g = StateGraph(FleetState)
g.add_node("ingest", ingest)
g.add_node("route", route)
g.add_node("execute_routine", execute_routine)
g.add_node("escalate", escalate)
g.set_entry_point("ingest")
g.add_edge("ingest", "route")
g.add_edge("route", "execute_routine")
g.add_edge("route", "escalate")
g.add_edge("execute_routine", END)
g.add_edge("escalate", END)
app = g.compile()

main.py

if __name__ == "__main__":
    result = app.invoke({
        "vehicles": [
            VehicleState(vehicle_id="RT-1042", lat=48.85, lon=2.35, speed_kmh=38,
                         speed_limit_kmh=50, battery_pct=71, sensor_health=0.97,
                         control_status="autonomous", last_seen=datetime.utcnow()),
            VehicleState(vehicle_id="RT-1187", lat=48.86, lon=2.34, speed_kmh=52,
                         speed_limit_kmh=50, battery_pct=64, sensor_health=0.96,
                         control_status="autonomous", last_seen=datetime.utcnow()),
        ],
    })
    for d in result["decisions"]:
        print(d.vehicle_id, d.lane, d.action, "| human:", d.requires_human)
    print("Escalations:", len(result["escalations"]))

Run it and the workflow processes the fleet in one pass: RT-1042 inside its envelope flows to a routine maintenance decision; RT-1187 breaking the speed envelope produces a safety event that routes to the human ops center with requires_human=True — paged with full context, never auto-resolved. The routine runs without humans; the safety decision never does.

Retry rules: telemetry ingestion retries 3 times with exponential backoff on transport errors, and a vehicle that stops reporting is itself a safety event — missing telemetry escalates, it never silently drops. Envelope checks are deterministic and never retried; the same state must always produce the same verdict, because the audit trail depends on it. Lane execution for routine events retries twice on transient failures; safety escalations never retry into auto-resolution — a page that fails to deliver is re-paged until a human acknowledges. These match the AI workflows library standard: transient errors retry cheaply, safety events escalate, and humans are never replaced by a retry loop.

Part 4 — The fleet dashboard and production checklist

The fleet audit log is the operational record and the regulatory surface: every dispatch decision, escalation, and action with the vehicle, timestamp, and reasoning. For a commercial robotaxi fleet, that record is what regulators and insurers will ask for first. Build the workflow with the log as a first-class output, and the operations story writes itself.

1. Monitor the envelope before you dispatch anything. Envelope checks are the safety backbone; dispatch is downstream of them, never upstream.
2. Encode the safety gate in the data model. requires_human=True on safety decisions by construction — not by policy text. The workflow cannot auto-execute what the schema forbids.
3. Escalate with full context. The human ops center gets the vehicle, the violation, the telemetry snapshot, and the reasoning — not a bare alert.
4. Treat missing telemetry as a safety event. A vehicle that stops reporting is a vehicle you cannot supervise; silence escalates.
5. Audit every decision. The fleet log is your operational record and your regulatory surface. It must be complete and reproducible.
6. Start with a pilot fleet, expand the envelope. The same staged rollout discipline runs through every workflow guide we publish — and for physical fleets, the stakes make the staging non-negotiable.

Frequently Asked Questions

Q: What did Uber and Pony.ai announce in August 2026?

A: Uber and Pony.ai are preparing to put more than 2,000 robotaxis on European roads, per August 14, 2026 reporting — a major step for autonomous mobility at commercial scale.

Q: Why does a robotaxi fleet need an operations workflow?

A: A 2,000-vehicle fleet generates continuous telemetry, dispatch decisions, and safety events. An operations layer coordinates those in real time, separates routine events from safety-critical ones, and escalates appropriately.

Q: What is a safety envelope?

A: A per-vehicle boundary on operating parameters — speed relative to limits, distance from obstacles, sensor health, and control status. A vehicle inside its envelope operates autonomously; a violation escalates.

Q: How does the workflow dispatch service events?

A: Events route to lanes by type: routine maintenance goes to the scheduling lane, telemetry anomalies to the diagnostics lane, and anything safety-critical to the human operations center.

Q: What is the hard safety gate?

A: Safety-critical events — envelope violations, sensor failures, or control anomalies — always escalate to a human operator. The workflow never auto-resolves a safety-critical event, only routine ones.

Closing thoughts

Two thousand robotaxis on European roads is the scale where autonomous mobility becomes an operations discipline, and the fleet-ops workflow is the reference pattern for that discipline: envelope monitoring, event classification, lane routing, hard safety gates, and a complete audit trail. The routine runs without humans; the safety decisions never do. Whether your fleet is robotaxis, delivery robots, or warehouse vehicles, the pattern transfers — and the audit log is the proof that it ran correctly. Track the autonomous-fleet rollout on latest AI news and study the operations patterns in the AI workflows library before your fleet scales.