AI 驱动的异常检测：配送超时、路线偏离与轨迹异常的自动识别
一、深度引言与场景痛点：10 万个订单中，哪些正在"出事"？
在一个拥有 10 万活跃骑手的配送平台上，客服部门每天收到约 2000 通投诉电话。其中 60% 的问题如果在发生后的 30 秒内被自动检测到，是可以提前干预的——比如配送员走错路了、订单在取货点停留时间异常长、配送路线严重偏离规划路径。
但问题在于：10 万个订单同时进行，靠人工监控是不可能的。必须有一套自动化系统，从海量的实时轨迹数据中，快速识别出异常信号。
二、底层机制与原理深度剖析
异常检测的三大场景
三、生产级代码实现与最佳实践
# 配送异常检测系统
from dataclasses import dataclass
from math import radians, sin, cos, sqrt, atan2
class DeliveryAnomalyDetector:
"""配送异常检测器
三个核心检测模块：
1. 配送超时预测
2. 路线偏离检测
3. 异常行为识别
"""
def __init__(self):
# 阈值配置（实际应从配置中心读取，支持动态调整）
self.config = {
"timeout_threshold": 0.7, # 超时概率阈值
"deviation_distance": 500, # 偏离距离阈值（米）
"deviation_duration": 300, # 偏离持续时间阈值（秒）
"stay_too_long": 600, # 停留过长时间阈值（秒）
"loop_detection_radius": 200, # 回环检测半径（米）
}
def check_all(self, order: dict, trajectory: list[dict],
planned_route: list[dict]) -> dict:
"""综合检测
Returns:
{"is_anomalous": bool, "alerts": […]}
"""
alerts = []
# 检测 1：配送超时
timeout_alert = self._check_timeout(order, trajectory)
if timeout_alert:
alerts.append(timeout_alert)
# 检测 2：路线偏离
deviation_alert = self._check_deviation(
trajectory, planned_route
)
if deviation_alert:
alerts.append(deviation_alert)
# 检测 3：异常行为
behavior_alerts = self._check_behavior(trajectory)
alerts.extend(behavior_alerts)
return {
"order_id": order["id"],
"is_anomalous": len(alerts) > 0,
"alert_count": len(alerts),
"alerts": alerts,
}
def _check_timeout(self, order: dict, trajectory: list[dict]) -> dict:
"""超时预测
核心逻辑：
1. 计算已用时间
2. 估算剩余路程时间
3. 对比承诺送达时间
4. 如果超时概率 > 70%，触发预警
"""
now = time.time()
created_at = order.get("created_at", now)
promised_at = order.get("promised_delivery_at", now + 1800)
# 已用时间（秒）
elapsed = now – created_at
# 估算剩余时间
if not trajectory:
return None
current_pos = trajectory[-1]
delivery_pos = {
"lat": order["delivery_lat"],
"lng": order["delivery_lng"],
}
# 剩余直线距离
remaining_dist = self._haversine(
current_pos["lat"], current_pos["lng"],
delivery_pos["lat"], delivery_pos["lng"]
)
# 估算剩余时间（假设平均速度 20 km/h）
avg_speed = 20 / 3.6 # 转换为 m/s
estimated_remaining = remaining_dist / avg_speed if avg_speed > 0 else 9999
# 预测送达时间
predicted_arrival = now + estimated_remaining
# 超时概率 = (预测时间 – 承诺时间) / 缓冲时间
buffer_time = promised_at – created_at
if buffer_time <= 0:
return None
overtime_gap = predicted_arrival – promised_at
timeout_prob = max(0, min(1, overtime_gap / (buffer_time * 0.3)))
if timeout_prob > self.config["timeout_threshold"]:
return {
"type": "TIMEOUT_RISK",
"severity": "HIGH" if timeout_prob > 0.9 else "MEDIUM",
"probability": round(timeout_prob * 100, 1),
"estimated_delay": round(max(0, overtime_gap / 60), 1),
"message": (
f"订单 {order['id']} 存在超时风险，"
f"概率 {timeout_prob*100:.0f}%，"
f"预计延迟 {max(0, overtime_gap/60):.0f} 分钟"
),
}
return None
def _check_deviation(self, trajectory: list[dict],
planned_route: list[dict]) -> dict:
"""路线偏离检测
计算当前轨迹点与规划路线的最近距离。
如果偏离超过阈值且持续一段时间，触发告警。
"""
if not trajectory or not planned_route:
return None
# 取最近 N 个轨迹点（防止单点抖动误报）
recent_points = trajectory[-5:]
deviation_count = 0
max_deviation = 0
for point in recent_points:
# 计算该点到规划路线的最近距离
min_dist = float('inf')
for segment in zip(planned_route, planned_route[1:]):
dist = self._point_to_segment_distance(
point, segment[0], segment[1]
)
min_dist = min(min_dist, dist)
if min_dist > self.config["deviation_distance"]:
deviation_count += 1
max_deviation = max(max_deviation, min_dist)
# 如果最近 5 个点中有 3 个以上偏离，判定为路线偏离
if deviation_count >= 3:
return {
"type": "ROUTE_DEVIATION",
"severity": "HIGH",
"max_deviation_meters": round(max_deviation),
"deviated_points": deviation_count,
"message": (
f"检测到配送路线偏离，"
f"最大偏离距离 {max_deviation:.0f} 米"
),
}
return None
def _check_behavior(self, trajectory: list[dict]) -> list[dict]:
"""异常行为识别"""
alerts = []
if len(trajectory) < 5:
return alerts
# 1. 停留超时检测
stay_alert = self._check_long_stay(trajectory)
if stay_alert:
alerts.append(stay_alert)
# 2. 轨迹回环检测
loop_alert = self._check_trajectory_loop(trajectory)
if loop_alert:
alerts.append(loop_alert)
return alerts
def _check_long_stay(self, trajectory: list[dict]) -> dict:
"""检测长时间停留
如果连续多个位置点距离很近（< 20m），判定为停留。
"""
recent = trajectory[-10:] if len(trajectory) >= 10 else trajectory
if len(recent) < 3:
return None
# 检查最近几个点是否都在很小的范围内
center = recent[0]
all_nearby = True
for point in recent[1:]:
dist = self._haversine(
center["lat"], center["lng"],
point["lat"], point["lng"]
)
if dist > 20: # 20 米以外就不是停留
all_nearby = False
break
if all_nearby:
# 计算停留时长
duration = recent[-1]["timestamp"] – recent[0]["timestamp"]
if duration > self.config["stay_too_long"]:
return {
"type": "LONG_STAY",
"severity": "MEDIUM",
"duration_minutes": round(duration / 60, 1),
"message": f"骑手已停留 {duration/60:.0f} 分钟",
}
return None
def _check_trajectory_loop(self, trajectory: list[dict]) -> dict:
"""检测轨迹回环（可能走回头路或绕路）"""
if len(trajectory) < 10:
return None
# 简化方法：检查当前位置是否和 5 分钟前的位置很近
# 如果距离 < 200m，且中间经过了较长距离，可能存在回环
current = trajectory[-1]
five_min_ago = trajectory[-10] if len(trajectory) >= 10 else trajectory[0]
dist = self._haversine(
current["lat"], current["lng"],
five_min_ago["lat"], five_min_ago["lng"]
)
# 检查中间路程
total_moved = 0
for i in range(len(trajectory) – 11, len(trajectory) – 1):
total_moved += self._haversine(
trajectory[i]["lat"], trajectory[i]["lng"],
trajectory[i+1]["lat"], trajectory[i+1]["lng"]
)
# 如果实际移动距离远大于直线距离，可能是绕路
if total_moved > 1000 and dist < self.config["loop_detection_radius"]:
return {
"type": "TRAJECTORY_LOOP",
"severity": "MEDIUM",
"extra_distance": round(total_moved – dist),
"message": f"检测到疑似绕路，多余距离 {total_moved-dist:.0f} 米",
}
return None
def _haversine(self, lat1, lon1, lat2, lon2) -> float:
R = 6371000
dlat = radians(lat2 – lat1)
dlon = radians(lon2 – lon1)
a = sin(dlat/2)**2 + cos(radians(lat1))*cos(radians(lat2))*sin(dlon/2)**2
return R * 2 * atan2(sqrt(a), sqrt(1-a))
def _point_to_segment_distance(self, point, seg_start, seg_end) -> float:
"""计算点到线段的最短距离"""
# 使用向量投影法
px, py = point["lng"], point["lat"]
x1, y1 = seg_start["lng"], seg_start["lat"]
x2, y2 = seg_end["lng"], seg_end["lat"]
# 线段长度平方
dx, dy = x2 – x1, y2 – y1
seg_len_sq = dx * dx + dy * dy
if seg_len_sq == 0:
return self._haversine(py, px, y1, x1)
# 投影参数 t
t = max(0, min(1, ((px – x1) * dx + (py – y1) * dy) / seg_len_sq))
# 投影点坐标
proj_x = x1 + t * dx
proj_y = y1 + t * dy
return self._haversine(py, px, proj_y, proj_x)
四、边界分析与架构权衡
误报 vs 漏报
异常检测系统面临的核心权衡：降低误报率（不要频繁骚扰运营）vs 降低漏报率（不错过真正的异常）。
策略：
- 线下高灵敏度（不漏），线上多级确认（降误报）
- 轻度异常仅记录不告警，中度和重度异常触发自动干预
延迟容忍度
不同异常的响应时间要求不同：
- 路线偏离：需要 30 秒内检测（尽快纠正）
- 配送超时：可以 2 分钟内检测（有时间缓冲）
- 异常停留：可以 5 分钟内检测（可能是等红灯）
五、总结
异常检测系统的核心价值不是"事后分析出了什么问题"，而是在问题演变成投诉之前自动发现并干预。这套系统的三个关键原则：
在构建这类系统时，最有价值的不是算法本身，而是对业务异常模式的理解——知道什么场景下容易出问题、出问题后的典型模式是什么。这需要和一线运营同学反复沟通。
