从 0 到 1 构建城市公共设施智能报修与派单系统：FastAPI + Vue3 + RabbitMQ + Redis Geo + AI 工作流实践

项目定位：面向政府市政部门的城市公共设施智能报修与派单平台，覆盖「市民报修 → AI 解析 → 工单生成 → 智能派单 → 维修处理 → 完工验收 → 用户评价 → 结算统计 → 数据分析」完整业务闭环。

技术栈：FastAPI + Vue 3 + MySQL + Redis + MongoDB + Elasticsearch + RabbitMQ + LangChain + 阿里云百炼大模型

---

一、项目背景与业务痛点

1.1 城市公共设施报修的典型场景

一座中型城市拥有数以万计的公共设施：路灯、井盖、护栏、垃圾桶、交通信号灯、绿化树木、道路路面……任何一个环节出问题，都可能影响市民出行安全和市容市貌。市民发现设施损坏后，通常拨打市政热线，接线员记录后电话通知维修队，维修队派工到人，维修完成后纸质回单，整个流程冗长、信息不对称、缺乏闭环。

1.2 传统人工派单存在的问题

1. 报修入口分散，信息非结构化：市民通过电话、微信、口头转述等多种方式报修，接线员人工记录，容易遗漏关键信息（准确位置、损坏程度、设施编号）。
2. 派单靠经验，效率低下：调度员凭记忆判断哪个维修员离得近、谁手里活少，经常出现"忙的人忙死、闲的人闲死"，或者派了一个不擅长该类型故障的维修员。
3. 维修过程不可视：维修员什么时候出发、是否到场、修了什么、用了什么材料，管理员完全不可知，只能靠电话催问。
4. 验收缺乏标准：是否修好由维修员自己说了算，没有第三方验证机制。
5. 数据无法沉淀：每月手工统计工单量、响应时长、满意度，耗时耗力且不准确，无法支撑管理决策。

1.3 系统要解决的核心问题

基于以上痛点，我确定了这个系统要解决的核心问题：

• 报修智能化：市民拍照+文字描述即可报修，AI 自动识别设施类型和紧急程度
• 派单自动化：基于实时位置、技能、负载、评分多维度自动匹配最优维修员
• 过程透明化：全流程节点记录（接单、签到、维修、完工），三端实时可见
• 验收可量化：AI 对比修前修后照片辅助判断，关键节点人工复核
• 数据可分析：多维度聚合统计，支撑运营优化和绩效考核

---

二、系统整体架构

2.1 架构总览

在架构设计阶段，我将系统拆分为三端 + 一网关 + 四库 + 一队列 + 一 AI 层的经典微服务分层架构。

2.2 各层职责说明

层次	技术选型	职责
前端层	Vue 3 + Vite + Pinia + Element Plus + ECharts	三端 UI，分别面向市民、维修员、管理员
网关层	FastAPI + Uvicorn	统一 API 网关，JWT 认证，请求路由，参数校验
业务层	FastAPI 异步路由	报修创建、智能派单、维修闭环、结算审核、统计分析
AI 层	LangChain + 阿里云百炼 qwen-vl-max	NLP 报修解析、AI 验收对比、派单评分辅助
消息队列	RabbitMQ (aio-pika)	派单异步、超时延迟检查、差评复核、ES 同步
存储层	MySQL + Redis + MongoDB + ES	四库分层，各司其职

2.3 为什么选择这些技术

• FastAPI：原生异步支持（async/await），自动生成 OpenAPI 文档，Pydantic 类型校验，性能接近 Node.js/Go。对于需要同时处理 HTTP 请求、数据库查询、Redis 操作、MQ 发布的项目，异步非阻塞是刚需。
• Vue 3 + Vite：Composition API 让复杂交互逻辑更清晰，Vite 开发体验极快。三端应用共享相同的技术栈和 API 封装模式，降低维护成本。
• Redis：不仅仅是缓存。Redis Geo 支持空间半径查询（GEORADIUS），天然适合"找附近的维修员"场景；分布式锁（SETNX）保证派单互斥；多 DB 分库（DB0 缓存、DB1 Geo、DB2 锁、DB3 计数器）隔离不同用途。
• MongoDB：AI 解析结果、维修记录（耗材不固定）、审计日志（字段可变）都是非结构化或半结构化数据，Schema-less 的 MongoDB 比 MySQL 更合适。
• Elasticsearch：管理后台需要按区域、设施类型、状态、时间多维度聚合统计，ES 的聚合查询（aggs）比 MySQL GROUP BY 灵活得多，加上 IK 中文分词器可以做工单内容全文搜索。
• RabbitMQ：派单、超时、差评复核都是典型的"触发后不必立即完成"的场景，用消息队列解耦可以避免 API 超时，同时利用 TTL + DLX 实现优雅的延迟任务。

---

三、核心业务流程设计

整个系统的核心是一条完整的业务闭环链路。下面我从市民提交报修开始，逐步拆解每一步的工程实现。

3.1 完整业务链路

3.2 关键环节详解

环节 1：市民提交报修

市民在 H5 页面填写故障描述、拍摄现场照片、授权 GPS 定位后提交。后端做三层处理：

• 防重复：对「用户ID + 设施类型 + 坐标哈希」做 SHA256，Redis 缓存 5 分钟，防止用户短时间内重复提交相同报修。
• GPS 反查地址：调用高德逆地理编码 API，将经纬度转为"XX 区 XX 路 XX 号"的可读地址。
• 图片上传：先上传阿里云 OSS，拿到 URL 后再写 MongoDB ticket_attachments 集合，带上水印哈希（SHA256）防止篡改。

环节 2：AI 解析报修内容

这是系统的第一个 AI 应用点。将市民的文字描述和图片 URL 传入 LangChain + 百炼 qwen-vl-max 视觉模型，输出结构化分类结果。具体设计见第九节。

环节 3-4：智能派单 + 超时检查

这是系统的核心亮点。工单创建后发布到 RabbitMQ dispatch 队列，消费者异步执行派单逻辑。派单完成后发布一条 10 分钟 TTL 的延迟消息到 dispatch_timeout.delay 队列——如果 10 分钟后无人接单，消息过期自动路由到 dispatch_timeout 队列，触发强制重新派单。具体设计见第七节和第八节。

环节 6：到场签到

维修员到达现场后点击"签到"，前端获取 GPS 坐标传给后端。后端用 Haversine 公式计算签到位置与设施位置的距离，阈值 50 米以内视为有效签到，记录到 MongoDB repair_records 的 gps_checkin 字段。超过阈值则提示"请到达设施附近后再签到"。

环节 7-8：完工提交 + AI 验收

维修员上传修前照片和修后照片，填写耗材和工时。后端调用 AI 验收服务，视觉模型对比修前修后照片判断是否合格。AI 判定通过 → 工单进入 verifying 状态等待市民确认；AI 判定不通过 → 工单退回 repairing 状态。AI 不是最终决策者，只是一个高效的前置过滤器。

环节 10-11：评价与差评复核

市民评价 2 星及以下时，系统自动发布消息到 RabbitMQ review_queue，触发管理员复核流程。管理员可以查看维修记录、AI 验收结果、照片对比后做出裁定（支持申诉）。

环节 12：结算与统计

工单关闭后自动生成结算单。结算逻辑从 MongoDB repair_records 读取耗材明细，从 MySQL audit_rules 读取对应设施品类的计价规则（基价、加班费率、紧急倍率、夜班补贴），计算劳务费 + 材料费 = 总结算金额。所有工单数据同步到 Elasticsearch，支撑管理后台的数据看板。

---

四、三端前端设计

4.1 技术选型理由

三端统一使用 Vue 3 + Vite + Pinia + Vue Router，理由如下：

• Vue 3 Composition API：三端都有大量表单、列表、状态切换场景，Composition API 的 setup 语法让逻辑复用（如定位、图片上传、状态轮询）变得非常自然。
• Vite：HMR 极快，三端独立构建互不影响。
• Pinia：相比 Vuex，API 更简洁，TypeScript 支持更好。在本项目中，市民端管理用户信息 + 工单列表，维修员端管理待接单列表 + 实时位置，后台管理登录态 + 数据看板缓存。
• Element Plus（仅后台）：管理后台是典型的 PC 中后台场景，表格、表单、弹窗、分页、日期选择器大量使用，Element Plus 是最成熟的选择。
• ECharts（仅后台）：驾驶舱的实时指标卡片、工单趋势折线图、片区分布柱状图、设施品类饼图，全部基于 ECharts 实现。

4.2 市民端

路由结构：

/login          → 登录（用户名密码 / 短信验证码）
/register       → 注册
/home           → 首页报修表单（文字 + 拍照 + GPS 定位）
/ticket/:id     → 工单详情（进度时间轴 + 维修员信息）
/evaluation/:id → 评价页（星级 + 标签 + 文字）
/my-tickets     → 我的工单列表（分页 + 状态筛选）

关键交互：

• 报修表单集成了高德 SDK 定位，降级方案用浏览器原生 navigator.geolocation
• 工单详情页展示状态时间轴（提交 → AI 解析 → 派单 → 接单 → 签到 → 维修 → 验收 → 完结），每个节点有对应的时间戳
• 评价页支持快捷标签（“响应快”“态度好”“技术专业”）+ 自定义文字

4.3 维修员端

路由结构：

/login          → 登录
/home           → 工作台（今日统计 + 快捷入口）
/queue          → 接单大厅（按距离排序的待接工单列表）
/ticket/:id     → 工单详情 + 操作按钮
/complete/:id   → 完工提交（修前/修后照片 + 耗材 + 工时）
/my-tickets     → 我的工单历史
/notifications  → 派单通知列表

关键交互：

• 接单大厅：后端返回按距离排序的工单列表（基于维修员实时位置 + Redis Geo），维修员可以看到每个工单的距离、设施类型、紧急程度
• 实时位置上报：登录后启动定时器，每 30 秒将 GPS 坐标上报到 PUT /worker/location，后端写入 Redis Geo
• 定位工具：封装了 utils/geolocation.js，高德 SDK 优先 + 原生 GPS 降级，签到场景精度阈值 50 米

4.4 管理后台

页面结构：

/login          → 管理员登录
/dashboard      → 数据驾驶舱（实时指标 + ECharts 图表）
/tickets        → 工单管理（ES 全文搜索 + 状态筛选 + 强制指派）
/workers        → 人员管理（维修员 CRUD + 技能标签配置）
/facilities     → 设施管理（设施档案 CRUD）
/settlements    → 结算审核（待审/已审/驳回）
/audit          → 审计日志（操作人/操作类型/变更前后快照）
/config         → 配置管理（结算规则 CRUD）

关键交互：

• 驾驶舱：4 个实时指标卡片（今日工单、在岗人数、待派单、待验收）+ 工单趋势折线图（近 6 个月）+ TOP10 故障设施柱状图 + 片区分布饼图
• 工单搜索：基于 Elasticsearch + IK 中文分词，支持按描述关键词、区域、设施类型、状态、时间范围组合搜索
• 强制指派：管理员可手动将工单指派给指定维修员（走 force_dispatch 流程，释放旧锁重新派单）

---

五、后端服务设计

5.1 为什么选择 FastAPI

在技术选型阶段，我对比了 Django、Flask、FastAPI 三个 Python 后端框架：

维度	Django	Flask	FastAPI
异步支持	3.x 才开始	需扩展	原生 async/await
类型校验	ORM 层面	需手动	Pydantic 自动
API 文档	需 drf-yasg	需 flasgger	自动生成 OpenAPI
性能	中等	中等	接近 Node.js/Go
学习曲线	陡峭	平缓	平缓

这个项目需要同时操作 MySQL、Redis、MongoDB、ES、RabbitMQ 五种外部依赖，异步 I/O 是刚需——如果使用同步框架，一个慢查询就会阻塞整个 Worker。FastAPI 的 async/await 让所有 I/O 操作天然并发。

5.2 Pydantic 的请求校验与响应模型

FastAPI 基于 Pydantic 做请求参数校验和响应序列化，这在本项目中非常实用。举一个创建报修的 Schema 示例：

from pydantic import BaseModel, Field
from typing import Optional

class TicketCreateRequest(BaseModel):
    description: str = Field(..., min_length=10, max_length=500, description="故障描述")
    facility_type: str = Field(..., description="设施品类")
    location_lng: float = Field(..., ge=-180, le=180, description="经度")
    location_lat: float = Field(..., ge=-90, le=90, description="纬度")
    image_urls: Optional[list[str]] = Field(default=[], description="图片URL列表")

class TicketResponse(BaseModel):
    ticket_id: str
    status: str
    ai_category: Optional[str]
    emergency_level: int
    created_at: str

    class Config:
        from_attributes = True  # 支持从 ORM 对象直接序列化

Pydantic 的 Field 校验器可以在请求进入业务逻辑之前拦截非法参数（如经度 > 180），减少防御性代码。from_attributes = True 让 ORM 对象可以直接传给 Response Schema，省去手动 to_dict() 的样板代码。

5.3 后端模块划分

我将后端按业务领域拆分为以下模块：

backend/app/
├── api/v1/                    # 路由层
│   ├── auth.py                # 统一认证（用户名/短信登录）
│   ├── citizen/
│   │   ├── tickets.py         # 市民工单 CRUD
│   │   └── evaluations.py     # 市民评价
│   ├── worker/
│   │   ├── auth.py            # 维修员认证
│   │   ├── tickets.py         # 接单大厅 / 接单 / 签到 / 完工
│   │   ├── notifications.py   # 派单通知
│   │   └── performance.py     # 绩效查询
│   ├── admin/
│   │   ├── dashboard.py       # 驾驶舱数据
│   │   ├── tickets.py         # 工单管理 / 搜索 / 强制指派
│   │   ├── workers.py         # 人员管理
│   │   ├── facilities.py      # 设施管理
│   │   ├── settlements.py     # 结算审核
│   │   ├── audit_logs.py      # 审计日志
│   │   └── config.py          # 配置管理
│   └── utils.py               # 逆地理编码 / 图片上传
├── models/
│   ├── mysql/                 # MySQL ORM 模型（7 张表）
│   └── mongo/                 # MongoDB 文档模型（5 个集合）
├── services/
│   ├── citizen/               # 报修服务
│   ├── worker/                # 派单服务 / 维修服务
│   ├── ai/                    # NLP / 验收 / 派单评分
│   ├── mq/                    # RabbitMQ 发布/消费
│   ├── map/                   # 高德地图 API
│   └── storage/               # OSS 图片存储
├── core/                      # JWT / 密码哈希 / 依赖注入
├── middleware/                # 认证中间件 / CORS
└── config/                    # Settings 配置类 / Redis 客户端 / MongoDB 连接

模块划分遵循单一职责原则：路由层只做参数提取和响应返回，业务逻辑全部下沉到 services/，数据访问通过 models/ 的 ORM/Document 模型完成。这样即使未来将某个服务拆成独立微服务，迁移成本也很低。

---

六、数据库与存储设计

6.1 四库分层职责

多数据库不是"为了用而用"，而是每种数据的特点天然适合不同的存储引擎。下表总结了这个项目中四种存储的明确边界：

存储	数据特点	存储内容	选型理由
MySQL	强结构化、需要 ACID 事务	用户、工单、设施、维修员、结算、评价、计价规则	关系型数据 + 状态流转必须保证一致性
Redis	热数据、低延迟、支持 Geo	工单缓存、维修员实时位置、分布式锁、在线集合、计数器	Geo 空间查询 + 高性能缓存 + 原子操作
MongoDB	非结构化、Schema 可变、JSON 原生	AI 解析日志、维修记录（耗材）、图片附件元数据、审计日志、通知	Schema-less，适合 AI 输出和半结构化记录
Elasticsearch	全文检索、聚合统计	工单索引、设施索引、维修员绩效索引	IK 分词 + 聚合查询，支撑搜索和分析

6.2 MySQL 核心表设计

users 用户表：

CREATE TABLE users (
    user_id     VARCHAR(32) PRIMARY KEY,       -- USR 前缀
    username    VARCHAR(64) UNIQUE NOT NULL,
    password_hash VARCHAR(256) NOT NULL,        -- bcrypt
    phone       VARCHAR(16),
    role        ENUM('citizen','worker','admin'),
    district    VARCHAR(64),                    -- 所属行政区
    is_active   BOOLEAN DEFAULT TRUE,
    created_at  DATETIME DEFAULT CURRENT_TIMESTAMP
);

tickets 工单主表：

CREATE TABLE tickets (
    ticket_id       VARCHAR(32) PRIMARY KEY,     -- TK 前缀
    user_id         VARCHAR(32) NOT NULL,
    facility_code   VARCHAR(32),
    facility_type   VARCHAR(32),                 -- 路灯故障/道路破损/井盖异常...
    district        VARCHAR(64),                 -- 冗余字段，避免 JOIN
    status          ENUM('pending','accepting','dispatching',
                         'repairing','verifying','closed','cancelled'),
    description     TEXT,
    address         VARCHAR(255),                -- GPS 反查地址
    location_lng    DOUBLE,
    location_lat    DOUBLE,
    emergency_level TINYINT DEFAULT 0,           -- 0普通 1紧急
    assigned_worker_id VARCHAR(32),
    ai_category     VARCHAR(64),                 -- AI 解析分类
    ai_confidence   FLOAT,
    created_at      DATETIME,
    accepted_at     DATETIME,
    started_at      DATETIME,
    completed_at    DATETIME,
    closed_at       DATETIME,
    INDEX idx_user (user_id),
    INDEX idx_worker (assigned_worker_id),
    INDEX idx_status_created (status, created_at),  -- 联合索引，高频分页查询
    INDEX idx_district (district)
);

设计要点：

• district 冗余存储避免每次查询都 JOIN 设施表
• status + created_at 联合索引支撑"按状态分页查询工单列表"这一最高频场景
• 时间轴字段（accepted_at / started_at / completed_at / closed_at）不仅用于展示，还用于计算响应时长、维修时长等 SLA 指标

workers 维修员表：

CREATE TABLE workers (
    worker_id       VARCHAR(32) PRIMARY KEY,     -- W 前缀，关联 users.user_id
    name            VARCHAR(64),
    skills          JSON,                        -- ["路灯","井盖","综合维修"]
    max_daily_orders INT DEFAULT 10,
    district        VARCHAR(64),
    night_duty      BOOLEAN DEFAULT FALSE,       -- 是否值夜班
    star_rating     FLOAT DEFAULT 5.0,
    total_orders    INT DEFAULT 0,
    avg_response_minutes FLOAT DEFAULT 0,
    is_active       BOOLEAN DEFAULT TRUE
);

skills 使用 JSON 类型存储，灵活支持一个维修员拥有多个技能标签，派单时做技能匹配。

6.3 Redis 的四库分库设计

Redis 在这个项目中承担了远不止缓存的角色。我将其分为 4 个 DB 隔离不同用途：

DB	用途	关键 Key	数据结构
DB0 (cache)	热点缓存	ticket:{id}:info, worker:{id}:profile, tickets:accepting	Hash, ZSET, String
DB1 (geo)	空间坐标	workers:geo	Geo (Sorted Set)
DB2 (lock)	分布式锁	lock:ticket:{id}	String (SETNX)
DB3 (counter)	计数器	worker:{id}:daily_order, id_seq:{prefix}	String (INCR)

Redis Geo 的核心操作：

# 维修员登录/位置上报时写入坐标
GEOADD workers:geo 116.397428 39.90923 worker_001

# 派单时：搜索设施周围 5km 内的维修员，返回距离和坐标，按距离升序
GEORADIUS workers:geo 116.404 39.915 5 km WITHDIST WITHCOORD ASC COUNT 20

# 派单成功后移除（避免同一维修员被重复指派）
ZREM workers:geo worker_001

6.4 MongoDB 的文档模型

MongoDB 存储所有"结构可变"的数据。以 repair_records 为例：

{
    "_id": ObjectId("..."),
    "ticket_id": "TK20260706001",
    "worker_id": "W001",
    "materials": [
        {"name": "LED灯板24W", "qty": 2, "unit": "个", "unit_cost": 85.00},
        {"name": "LED驱动电源", "qty": 1, "unit": "个", "unit_cost": 120.00}
    ],
    "labor_hours": 1.5,
    "work_notes": "更换灯板和驱动电源，线路检查正常",
    "before_photos": ["https://oss.../before1.jpg", "https://oss.../before2.jpg"],
    "after_photos": ["https://oss.../after1.jpg"],
    "gps_checkin": {"lng": 116.404, "lat": 39.915, "timestamp": "2026-07-06T10:30:00+08:00"},
    "ai_verify_result": {
        "verified": true,
        "confidence": 0.94,
        "diff_summary": "修后灯板已更换，灯光明亮，故障已排除"
    }
}

不同设施类型的耗材完全不同（修路灯用灯板+驱动，修道路用沥青+乳化剂），MongoDB 的 Schema-less 特性天然适配这种场景。

6.5 Elasticsearch 的索引设计

工单索引 city_repair_tickets 的 Mapping：

{
    "mappings": {
        "properties": {
            "ticket_id": {"type": "keyword"},
            "description": {"type": "text", "analyzer": "ik_max_word"},
            "facility_type": {"type": "keyword"},
            "district": {"type": "keyword"},
            "status": {"type": "keyword"},
            "emergency_level": {"type": "integer"},
            "created_at": {"type": "date"},
            "location": {"type": "geo_point"}
        }
    }
}

description 使用 IK 中文分词器，市民搜索"路灯不亮"能匹配到"路灯故障"的工单。facility_type、district、status 使用 keyword 类型做精确聚合。

---

七、智能派单设计

智能派单是这个系统最核心也最复杂的模块。我从"最近派单"逐步演进到"多因子评分派单"。

7.1 派单流程

7.2 派单策略详解

第一层：空间筛选（Redis Geo）

radius_km = 10 if emergency else 5  # 紧急工单扩大搜索半径
nearby_workers = await redis_geo.georadius(
    "workers:geo", facility_lng, facility_lat,
    radius_km, unit="km", withdist=True, withcoord=True,
    sort="ASC", count=20
)

第二层：硬约束过滤

for worker in nearby_workers:
    # 约束1：日单量未达上限
    if today_orders >= worker.max_daily_orders: continue
    # 约束2：夜班时段（22:00-06:00）必须开启夜班
    if is_night and not worker.night_duty: continue
    # 约束3：技能标签匹配设施类型
    if not skill_matched(worker.skills, facility_type): continue
    filtered.append(worker)

如果硬约束过滤后候选人为空，启用降级策略：日单上限放宽至 1.5 倍，技能匹配放宽。

第三层：高德驾车距离修正

Redis Geo 返回的是直线距离，实际维修员需要走道路。我用高德驾车路径规划 API 修正距离，并缓存 5 分钟：

route = await driving_distance(worker_lng, worker_lat, facility_lng, facility_lat)
worker["distance_km"] = route["distance_km"]   # 替换直线距离
worker["driving_duration_min"] = route["duration_min"]
await redis_cache.setex(f"amap:driving:{worker_id}:{ticket_id}", 300, json.dumps(route))

第四层：多维评分

# 简化版：只按距离排序，每公里扣 20 分
score = max(0, 100 - distance_km * 20)

# 完整版（启用 LLM_ENABLE_DISPATCH_SCORING 时）：
# score = distance_weight(40%) + load_weight(30%) + rating_weight(20%) + response_weight(10%)

当前简化版已能满足大部分场景——最近的维修员通常是最高效的选择。完整版通过 LangChain 调用 LLM 做综合判断，LLM 失败时自动回退到简化版本地算法。

第五层：分布式锁 + 事务写入

# Redis 分布式锁（SETNX，300 秒自动过期防死锁）
acquired = await redis_lock.set(f"lock:ticket:{ticket_id}", worker_id, nx=True, ex=300)
if not acquired: return  # 已被其他并发流程锁定

# MySQL 事务写入
ticket.assigned_worker_id = worker_id
ticket.status = "dispatching"
await db.commit()

# Redis 状态同步
await redis_cache.hset(f"ticket:{ticket_id}:info", mapping={...})
await redis_cache.zrem("tickets:accepting", ticket_id)  # 从接单大厅移除
await redis_geo.zrem("workers:geo", worker_id)           # 从 Geo 移除

7.3 超时重新派单

派单完成后发布一条 10 分钟 TTL 的延迟消息。10 分钟后如果工单状态仍是 dispatching（无人接单），触发 force_dispatch 流程：

async def force_dispatch(ticket_id, db, admin_id):
    # 释放旧锁
    await redis_lock.delete(f"lock:ticket:{ticket_id}")
    # 重新执行完整派单流程
    result = await execute_dispatch(ticket_id, db)
    # 写入审计日志
    await mongo_db.audit_logs.insert_one({...})
    return result

---

八、RabbitMQ 异步流程设计

8.1 为什么需要消息队列

以下场景不适合在 HTTP 请求-响应周期内同步完成：

• 派单：涉及 Redis Geo 查询 + 高德 API + MySQL 事务，耗时可能超过 2 秒
• 超时检查：需要 10 分钟后触发，HTTP 无法实现延迟执行
• 差评复核：低优先级任务，不需要立即处理
• ES 同步：失败需要重试，不能丢失

RabbitMQ 让这些流程从"同步阻塞"变为"异步解耦"，核心收益是 API 响应速度大幅提升 + 可靠性保障。

8.2 队列拓扑设计

8.3 队列清单

队列	用途	特殊机制
dispatch	智能派单	普通队列，立即消费
dispatch_timeout.delay	超时延迟	per-message TTL + DLX → dispatch_timeout
dispatch_timeout	超时检查	由延迟队列 DLX 路由过来
review_queue	差评复核	普通队列
es_sync	ES 同步	DLX → city_repair.dlx，消费者内重试
es_sync.delay	ES 同步延迟重试	per-message TTL + DLX → es_sync
es_sync.dlq	ES 同步死信	重试耗尽后人工处理

8.4 延迟队列的实现：TTL + DLX

RabbitMQ 没有内置的延迟队列，但可以通过 per-message TTL + Dead Letter Exchange 优雅实现：

# 声明延迟队列：消息过期后自动转发到目标队列
timeout_delay_queue = await channel.declare_queue(
    "dispatch_timeout.delay",
    durable=True,
    arguments={
        "x-dead-letter-exchange": "city_repair.direct",    # 过期后投递到的交换器
        "x-dead-letter-routing-key": "dispatch_timeout",   # 过期后投递到的队列
    },
)

# 发布带 TTL 的消息
message = aio_pika.Message(
    body=json.dumps({"ticket_id": ticket_id, "action": "timeout_check"}).encode(),
    expiration=str(10 * 60 * 1000),  # 10 分钟 TTL（毫秒）
    delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
)
await exchange.publish(message, routing_key="dispatch_timeout.delay")

8.5 幂等性保证

消息可能被重复投递（网络抖动、消费者崩溃重启），必须保证幂等。我的做法是：

• 派单消息：消费前检查工单状态，dispatching 状态说明已派单，跳过
• ES 同步消息：ES 的 index 操作本身幂等（相同 _id 的文档会覆盖）
• 超时检查消息：执行前检查工单是否仍处于 dispatching 状态

8.6 ES 同步的可靠性设计

ES 同步是最容易出问题的环节（ES 可能暂时不可用），我设计了指数退避重试 + 死信队列：

async def _handle_es_sync(message: aio_pika.IncomingMessage):
    body = json.loads(message.body.decode())
    ticket_id = body["ticket_id"]
    retry_count = int(message.headers.get("x-retry-count", 0))

    try:
        await sync_ticket_to_es(ticket_id)
        await message.ack()
    except Exception:
        if retry_count < settings.ES_SYNC_MAX_RETRIES:
            # 指数退避：2s → 4s → 8s → 16s → 32s
            delay = min(settings.ES_SYNC_BASE_DELAY_SEC * (2 ** retry_count),
                        settings.ES_SYNC_MAX_DELAY_SEC)
            # 发布到延迟队列
            retry_msg = aio_pika.Message(
                body=message.body,
                headers={"x-retry-count": retry_count + 1},
                expiration=str(delay * 1000),
            )
            await exchange.publish(retry_msg, routing_key="es_sync.delay")
            await message.ack()
        else:
            # 重试耗尽 → 死信队列
            await message.reject(requeue=False)  # DLX 自动处理

---

九、AI 工作流设计

9.1 总体定位

AI 在这个系统中承担的是辅助决策角色，而不是替代人工。每个 AI 输出都会记录到 MongoDB ai_analysis_logs，包含输入、输出、置信度、模型版本，便于事后审计和回溯。

9.2 三大 AI 应用场景

9.3 场景一：报修内容解析

输入：

{
    "text": "解放路和人民路交叉口的路灯不亮了，已经三天了，晚上很黑很危险",
    "image_urls": ["https://oss.../lamp_broken.jpg"],
    "lng": 116.404,
    "lat": 39.915
}

输出（Pydantic Schema）：

class NLPOutput(BaseModel):
    category: str              # "路灯故障"
    sub_category: str          # "灯具不亮"
    emergency_level: int       # 0 普通
    confidence: float          # 0.92
    repair_knowledge: list[str] # ["检查供电线路...", "LED驱动电源..."]
    tools_needed: list[str]    # ["绝缘手套", "万用表", "螺丝刀套装"]
    safety_tips: list[str]     # ["作业前必须断电并验电..."]
    parts_needed: list[str]    # ["LED灯板24W", "LED驱动电源"]

降级策略：当 LLM API Key 未配置或调用失败时，自动回退到内置关键词匹配知识库。知识库覆盖了路灯故障、道路破损、井盖异常、护栏损坏、环卫设施、交通信号设施、公共绿化 7 大类，每类都包含标准的维修知识、工具、安全提示——确保即使 AI 不可用，维修员仍能看到有用的指导信息。

9.4 场景二：智能派单辅助

当前默认使用简化版本地算法（只按距离排序），但架构上已预留了 LLM 评分的能力。开启 LLM_ENABLE_DISPATCH_SCORING=true 后，会将候选维修员列表 + 权重配置传入 LLM 做综合判断。

# LLM 评分模式（开启后）
model = provider.get_model()
structured_model = model.with_structured_output(DispatchScoreOutput)
result = await structured_model.ainvoke(messages)
# 失败自动回退
except Exception:
    result = _simple_score_candidates(candidates)

9.5 场景三：AI 辅助验收

这是三个场景中技术含量最高的——它需要视觉模型对比修前和修后照片。

关键设计：

• 使用 qwen-vl-max-latest（支持图片理解的多模态模型）
• 构造多模态消息：修前照片 + 修后照片 + 维修描述
• 输出结构化结果：{verified: bool, confidence: float, diff_summary: str}
• AI 判定通过 → 进入 verifying 状态等市民确认
• AI 判定不通过 → 退回 repairing 状态
• AI 不可用时默认通过（verified=True, confidence=0.92），不阻塞流程

9.6 AI 的安全边界

• AI 输出不直接改变工单状态，必须经过规则校验（如验收必须修前修后照片都有）
• 所有 AI 结果写入 MongoDB 审计日志，完整保留输入输出
• 关键决策节点（差评复核、结算审核）仍需要人工确认
• LLM 调用失败有多层降级策略，保证业务不中断

---

十、接口设计示例

10.1 创建报修

POST /api/v1/citizen/tickets
Authorization: Bearer <jwt_token>

请求体：

{
    "description": "解放路和人民路交叉口路灯不亮，已经三天了",
    "facility_type": "路灯故障",
    "location_lng": 116.404,
    "location_lat": 39.915,
    "image_urls": ["https://oss-cn-beijing.aliyuncs.com/.../photo1.jpg"]
}

响应：

{
    "code": 200,
    "data": {
        "ticket_id": "TK20260706001",
        "status": "accepting",
        "ai_category": "路灯故障",
        "ai_confidence": 0.92,
        "emergency_level": 0,
        "address": "北京市朝阳区解放路与人民路交叉口",
        "created_at": "2026-07-06T10:25:00+08:00"
    }
}

说明：创建成功后异步触发 NLP 解析 → 派单流程，市民立即拿到工单 ID 和预估分类。

10.2 查询工单详情（三端通用）

GET /api/v1/citizen/tickets/TK20260706001
GET /api/v1/worker/tickets/TK20260706001
GET /api/v1/admin/tickets/TK20260706001

响应（聚合了 MySQL + MongoDB + Redis 数据）：

{
    "code": 200,
    "data": {
        "ticket_id": "TK20260706001",
        "status": "repairing",
        "facility_type": "路灯故障",
        "description": "...",
        "address": "北京市朝阳区...",
        "emergency_level": 0,
        "timeline": [
            {"event": "市民提交报修", "time": "2026-07-06T10:25:00"},
            {"event": "AI解析完成", "time": "2026-07-06T10:25:03"},
            {"event": "已派单给维修员张三", "time": "2026-07-06T10:25:05"},
            {"event": "维修员已接单", "time": "2026-07-06T10:28:00"},
            {"event": "维修员已到场签到", "time": "2026-07-06T10:45:00"}
        ],
        "worker": {
            "name": "张三",
            "phone": "138****5678",
            "star_rating": 4.8,
            "skills": ["路灯", "交通信号设施"]
        },
        "ai_analysis": {
            "category": "路灯故障",
            "repair_knowledge": ["..."]
        }
    }
}

10.3 维修员接单

PUT /api/v1/worker/tickets/TK20260706001/accept
Authorization: Bearer <worker_jwt>

响应：

{
    "code": 200,
    "data": {
        "ticket_id": "TK20260706001",
        "status": "repairing",
        "accepted_at": "2026-07-06T10:28:00+08:00"
    }
}

说明：接单时需要 Redis 分布式锁 + MySQL 乐观锁双保险，防止多人同时抢单。

10.4 提交维修结果

PUT /api/v1/worker/tickets/TK20260706001/complete
Authorization: Bearer <worker_jwt>

请求体：

{
    "before_photos": ["https://oss.../before1.jpg"],
    "after_photos": ["https://oss.../after1.jpg"],
    "materials": [
        {"name": "LED灯板24W", "qty": 2, "unit": "个", "unit_cost": 85.00}
    ],
    "labor_hours": 1.5,
    "work_notes": "更换灯板和驱动电源，线路检查正常"
}

响应：

{
    "code": 200,
    "data": {
        "ticket_id": "TK20260706001",
        "status": "verifying",
        "ai_verify": {
            "verified": true,
            "confidence": 0.94,
            "diff_summary": "修后灯板已更换，灯光明亮，故障已排除"
        }
    }
}

10.5 管理员审核结算

PUT /api/v1/admin/settlements/ST20260706001/audit
Authorization: Bearer <admin_jwt>

请求体：

{
    "action": "approved",
    "comment": "材料用量合理，审核通过"
}

10.6 数据看板统计

GET /api/v1/admin/dashboard/realtime
GET /api/v1/admin/dashboard/analytics?months=6

响应（analytics）：

{
    "code": 200,
    "data": {
        "total_tickets": 12340,
        "trend": [
            {"month": "2026-01", "count": 1890},
            {"month": "2026-02", "count": 1750}
        ],
        "top_facilities": [
            {"type": "路灯故障", "count": 4520},
            {"type": "道路破损", "count": 3210}
        ],
        "district_distribution": [
            {"district": "朝阳区", "count": 2890},
            {"district": "海淀区", "count": 2340}
        ],
        "avg_response_minutes": 8.5,
        "satisfaction_rate": 0.94
    }
}

---

十一、关键难点与解决方案

11.1 多角色流程状态复杂

难点：系统涉及市民、维修员、管理员三个角色，工单状态有 7 个（pending → accepting → dispatching → repairing → verifying → closed / cancelled），每个状态允许的操作不同。

解决方案：

• 在 MySQL tickets 表使用 ENUM 约束状态值，数据库层面防止非法状态
• 每个 API 路由层校验当前状态是否允许该操作（如只有 dispatching 状态的工单才能接单）
• 状态变更通过 services/ 层的专用函数完成，确保状态 + 时间戳 + 审计日志原子写入

11.2 派单公平性与效率平衡

难点：纯距离优先可能导致某个维修员一直接到远处的单子（因为他在 Geo 集合里），而纯负载均衡可能派一个很远的维修员。

解决方案：

• 使用高德驾车距离替换 Geo 直线距离，让距离计算更真实
• 日单量上限作为硬约束（不是软权重），避免"累死一个人"
• 派单后从 Redis Geo 移除该维修员，直到他完成当前工单后重新上报位置

11.3 AI 输出不稳定

难点：LLM 可能返回非预期的格式、幻觉内容，或直接超时。

解决方案：

• LangChain 的 with_structured_output() 约束输出格式为 Pydantic Schema，自动重试格式不匹配的响应
• 内置关键词匹配降级知识库，覆盖 7 大类设施，确保 LLM 不可用时也能给出合理分类
• 紧急程度归一化函数：兼容数字 0/1、中文"紧急/高"、英文"urgent/high"等多种输入
• AI 验收默认通过（verified=True），不阻塞流程

11.4 工单状态一致性问题

难点：工单状态同时存在于 MySQL（主存储）、Redis（热缓存）、Elasticsearch（搜索索引），三处可能不一致。

解决方案：

• MySQL 是唯一真相源（Single Source of Truth）
• Redis 缓存设置 14 天 TTL，作为加速层，缓存未命中时回源 MySQL
• ES 通过 RabbitMQ 异步同步，有重试 + 死信机制保证最终一致性
• 管理后台搜索时优先查 ES，ES 不可用时降级查 MySQL

11.5 消息队列重复消费

难点：RabbitMQ 的 at-least-once 语义意味着消息可能被重复投递。

解决方案：

• 消费者处理前检查业务状态（派单消费者检查工单是否已处于 dispatching）
• ES 同步天然幂等（相同 _id 覆盖写入）
• 超时检查使用 Redis 分布式锁避免重复执行

11.6 多数据库数据边界问题

难点：MySQL、Redis、MongoDB、Elasticsearch 四库共存，数据边界不清晰会导致维护噩梦。

解决方案：建立清晰的规则——

• 强结构化 + 需要事务 → MySQL
• 热数据 + 需要快速访问 → Redis（来自 MySQL 的缓存副本）
• Schema 可变 + JSON 原生 → MongoDB
• 全文搜索 + 聚合统计 → Elasticsearch（来自 MySQL 的搜索副本）

跨库查询通过应用层组装，不搞跨库 JOIN。

11.7 管理后台统计性能问题

难点：按区域、设施类型、时间范围多维度聚合统计，MySQL GROUP BY 大表性能差。

解决方案：

• 实时指标（今日工单数、在岗人数）用 Redis 计数器 + MySQL CURDATE() 查询
• 聚合统计（趋势、分布、TOP N）走 Elasticsearch 的聚合查询
• ES 同步通过 RabbitMQ 异步完成，不阻塞主流程

---

十二、项目亮点总结

回顾整个项目，我认为以下几个技术点是最有工程价值的：

1. 三端协同架构：市民端、维修员端、管理后台共享同一套 API，通过 JWT 的 role 字段做接口级权限控制，代码复用率高。

2. AI + 规则引擎结合：AI 负责"理解"（NLP 解析、视觉验收），规则引擎负责"执行"（状态流转、硬约束过滤）。AI 输出不直接改变系统状态，关键节点保留人工复核。

3. Redis Geo 空间派单：利用 Redis 内置的 Geo 数据结构实现"找附近的维修员"，比用 MySQL 计算 Haversine 距离快两个数量级，且天然支持排序和分页。

4. RabbitMQ 异步解耦：派单、超时检查、ES 同步、差评复核全部通过消息队列异步化，API 响应时间从秒级降到毫秒级。延迟队列（TTL + DLX）优雅实现了超时自动派单。

5. 四库分层存储：MySQL（ACID 事务）、Redis（Geo + 缓存 + 锁）、MongoDB（非结构化文档）、Elasticsearch（搜索 + 聚合），每种存储做它最擅长的事。

6. FastAPI 异步接口：原生 async/await 让所有 I/O 操作（数据库查询、Redis 操作、MQ 发布）都非阻塞，单进程即可承载高并发。

7. 数据闭环：从报修到评价，全流程数据可追踪、可审计、可分析，管理后台的驾驶舱让运营决策有数据支撑。

---

十三、后续优化方向

这个系统还有不少可以深挖的方向：

1. WebSocket 实时推送：目前工单状态变更需要前端轮询，改为 WebSocket 推送后，维修员可以实时收到新工单通知（类似外卖骑手接单体验）。

2. 更完整的规则引擎：当前硬约束是硬编码的，可以引入 Drools 或自研轻量规则引擎，让管理员可以可视化配置派单策略。

3. GIS 地图服务：管理后台接入高德/百度地图 SDK，实现工单热力图、维修员轨迹回放、片区工单密度可视化。

4. AI 图片识别设施损坏程度：目前 AI 验收只是对比修前修后，未来可以让 AI 直接评估损坏程度（如路面坑洞的面积和深度），输出量化的损坏等级。

5. 维修员路径规划：维修员一天可能要去多个点位，引入路径规划算法（TSP 问题），帮维修员规划最优路线。

6. 工单 SLA 体系：为不同设施类型和紧急程度定义 SLA（如紧急工单 30 分钟响应、普通工单 2 小时响应），超时自动告警升级。

7. 更完善的权限系统：引入 RBAC 或 ABAC，支持多级管理员、片区经理、维修队长等更细粒度的角色。

8. 可观测性：接入 OpenTelemetry 做分布式链路追踪，Prometheus + Grafana 做指标监控，ELK 做日志聚合。

9. 容器化部署与 CI/CD：编写 Docker Compose / Kubernetes 部署配置，搭配 GitHub Actions / GitLab CI 做自动化测试和部署。

---

结语

这个项目从零开始搭建，到最终形成一个覆盖「报修 → AI 解析 → 派单 → 维修 → 验收 → 评价 → 结算 → 分析」完整闭环的系统，技术栈横跨前端三端、后端微服务、四类数据库、消息队列、AI 大模型、空间计算等多个领域。

做这个项目最大的体会是：架构设计的核心不是"用了多少技术"，而是"每个技术放在它最合适的位置上"。MySQL 做事务、Redis 做 Geo 和缓存、MongoDB 存文档、ES 做搜索和聚合、RabbitMQ 做异步解耦——每一层都有明确的职责边界，组合在一起才是一个健壮的系统。

希望这篇文章能对正在做类似智慧城市、O2O 派单、工单管理系统的同学有所启发。