看到“New Orleans is using AI to triage 911 calls in case of backlog”这则新闻时很多人第一反应可能是AI 要接管紧急呼叫中心了。这个判断大概率是错的。真实的场景没有这么科幻反而更有工程参考价值。新奥尔良的做法更像是在紧急呼叫量突然变大、人工接警来不及处理的“积压”场景下用 AI 先把每一通电话变成结构化事件帮助接线员快速判断“哪一通现在必须马上处理”。这个思路不是没有风险但它提供了一个很典型的行业样本AI 在公共安全、客服中心、工单系统里真正有价值的落地方式并不是替代人而是在人忙不过来的时候帮人把优先级排对。这篇文章我会从业务视角和技术工程视角一起拆解这件事。先讲清楚 911 呼叫中心里的“triage”和“backlog”到底是什么问题再给出一个这类 AI 分流系统常见的整体架构然后带着你写一个最小可运行的 AI Triage Demo用 FastAPI 接一个模拟大模型的服务把一通电话的转写文本转成“事件类别 紧急等级 地址 摘要”的结构化结果最后重点讨论误判、数据合规、人工复核、降级回滚这些真实工程里躲不开的问题。如果你正在做呼叫中心、客服工单、应急调度或者任何带“优先级排队”的系统这篇文章的思路可以直接迁移过去。1. 这篇文章真正要解决的问题先看业务本质。911 呼叫中心也好普通客服中心也好突然涌进大量请求时核心矛盾不只是“人多不多”而是队列里的任务价值差异极大。有的电话是噪音投诉可以等十分钟有的电话是有人昏迷、火灾、枪击晚接一秒都可能造成严重后果。但传统排队系统大多是先来先服务FIFO或者最多按 IVR 按键做一个非常粗糙的入口分流。一旦队列积压所有电话混在一起接线员只能按顺序接听真正的高危电话可能被淹没在后面。这就是“backlog”最危险的地方。backlog 不只是“排队时间长”而是“队列里最紧急的事件无法被优先识别”。人工接线员要听完报警人描述才能判断事件类型高峰期根本没有时间做二次转接和资源调度。所以新奥尔良这类城市尝试用 AI 做 triage本质不是用 AI 代替接线员去救人而是在队列被填满之前或填满过程中先用 AI 把“这是什么事件、紧急度多高、在哪里、需不需要救护车”这些关键信息抽出来让最高危的事件优先被人工盯上。对工程师来说这个案例真正值得学的是当我们要用大模型或 AI 解决一个现有系统问题时不能只做一个“信息抽取接口”就完事还要考虑它如何接入调度队列、如何设计人工复核机制、如何避免 AI 抽风导致更大的事故。本文会围绕“AI 辅助分流”这个主题从架构到代码再到排错完整走一遍。2. 关键术语与业务模型2.1 triage从医疗分诊到呼叫分流triage 这个词来自医疗系统意思是“根据病情轻重决定处理顺序”。放到 911 场景里它指的是一套“分级决策”一通电话进来后系统要判断这属于医疗急救、火灾、治安事件还是交通事件并给出紧急等级例如 critical、high、medium、low。这里很容易有一个误区triage 不是聊天机器人不是让 AI 去和报警人对话更不是让 AI 自己决定“要不要派警车”。它做的是一件事——把非结构化的语音转写文本变成结构化的调度信息。这个“结构化”是关键。大模型在理解口语、同义表述、乱序描述方面有明显优势但它真正的工程价值在于输出稳定字段让下游系统能自动处理。2.2 backlog积压的真正代价backlog 指的是服务速率跟不上到达速率时队列里堆积未处理任务。对于 911 呼叫中心这个词意味着还有电话在等待被接听而等待时间每增加一秒风险都在上升。过去缓解 backlog 的方法是加人、加话务通道、忙时回拨。但应急场景不是商场大促无法提前预测极端高峰也不允许像外卖平台那样“订单太多就关闭下单”。所以 AI triage 的另一个重要价值是在人类坐席都占线时先自动把电话内容转成事件再按紧急程度重新排队。低优先级电话可以等待高优先级电话则通过队列调度被提前处理。2.3 911、PSAP、CAD 分别是什么为了让读者理解这个业务链路简单交代三个词。911 是美国紧急求助电话负责警察、消防和医疗三类事件的统一接入。PSAPPublic Safety Answering Point是公共安全应答点可以理解成接警中心。CADComputer-Aided Dispatch是计算机辅助调度系统接警中心坐席在 CAD 里看到事件信息并派发警力。AI triage 系统做的事就是位于“电话接入”和“CAD 生成工单”之间把一通语音电话快速变成一个 CAD 可以消费的结构化事件。国内读者可能更熟悉 110、119、120 分开设立的架构但无论架构怎么分核心矛盾是一样的第一通电话的信息质量决定了后续资源调度的效率。2.4 传统文本分类和大模型结构化输出有什么区别很多人会问这种分流一定需要大模型吗传统的 IVR 按键分流、关键词命中、文本分类模型是不是也能做可以做但效果上限不同。传统方案更适合“场景封闭、话术固定”的情况。紧急呼叫则是典型的开放场景报警人可能语无伦次可能口音很重可能同时包含多个事件要素地址可能只说一个路名火势和伤员情况往往是零散描述的。大模型在这种场景下的优势不是“分类准”而是能同时完成事件分类、实体抽取、摘要生成、置信度判断多个任务并直接输出 JSON 结构省去了许多串联子模型的工作。对比项传统 IVR/文本分类大模型结构化输出输入理解依赖关键词和预设菜单能理解口语化、混乱表达字段抽取通常只做分类缺少实体信息可一次输出类别、地址、伤员、状态摘要能力不具备可以生成坐席快速阅读的摘要工程复杂度较低较高需要处理延时、输错、回退适合场景简单的低风险分流复杂呼叫、backlog 下的辅助分流3. 系统整体架构AI 如何接入紧急呼叫链路3.1 总体链路设计一个典型的 AI 辅助 911 分流系统按正常流程可以分成六层通话接入层负责把电话语音送到系统语音网关之后接 ASR 语音识别服务把音频转成带时间戳的文本AI Triage Service 拿到转写文本后调用大模型或规则引擎进行结构化分析结果写入调度队列人工坐席在 CAD 或坐席工作台上看到 AI 给出的“事件类别 紧急等级 摘要 关键场地信息”进行确认或修正最后由 CAD 完成资源调度。这里要强调一点AI 结果不能直接跳到“派警”这一步。从公开报道和行业常识来看911 这类公共安全场景更稳妥的做法是把 AI 定位成“辅助坐席”而不是“自动决策者”。即使某一天 AI 分类准确率非常高也需要保留人工审核环节因为错误分诊的代价不是用户体验扣分而是生命安全风险。3.2 两种运行模式影子模式和辅助模式新系统上线不能一步到位。工程上常见的做法是先用影子模式跑一段时间AI 在后台实时处理真实电话转写产生结果但不直接进入调度流程只和人工坐席的最终结论做对比用来评估准确率、漏报率、误报率。当指标达到预设标准后再切换成辅助模式AI 结果展示在坐席工作台上坐席看到的是“系统建议”可以一键采纳也可以修改。更激进一点的自动模式只适合在低风险事件里使用比如判断为“噪音投诉”且置信度极高时自动打低优先级标签但这类自动动作依然要有完整日志和规则兜底。3.3 系统模块清单从工程实现角度看这个系统至少包含以下模块ASR 接入模块负责音频转写、说话人分离、标点恢复。事件结构化模块用大模型或混合规则抽取事件字段。分级决策模块综合模型输出、规则阈值、置信度决定紧急等级。摘要生成模块生成坐席可读的一句话摘要。队列调度模块根据紧急等级和进入时间重排任务。监控与告警模块记录每个环节的耗时、异常、降级情况。人工审核工作台让坐席快速确认或修改 AI 结果。4. 核心模块拆解与设计要点4.1 语音转写先解决“听不清”的问题紧急呼叫语音转写比普通会议转录更难。报警人可能处于奔跑状态背景有警笛、风声、人群嘈杂声可能一边哭一边说可能有很重的方言或口音可能说一半甚至说不完整。ASR 模型的选型必须尽量兼顾准确率和延迟转写文本质量直接决定后面大模型的分析效果。工程上建议在 ASR 之后加一个文本后处理步骤比如标点恢复、纠错、数字和地址规范化。911 场景里地址常常是数字加路名例如“245 Main Street”ASR 容易把 “Main” 听成 “Mane” 或 “May”所以地址字段要结合地理词典做二次校正。这个细节很容易被忽略却在真实场景中决定了调度车辆能不能找对地方。4.2 事件要素抽取让模型输出固定 JSON大模型的 Prompt 设计决定了输出质量。对 triage 场景Prompt 里要明确指定必须输出的字段事件类别、紧急等级、地址、是否有伤员、事件是否仍在持续、一句话摘要。为了让下游系统稳定解析要求模型只返回 JSON不要附带解释文字。下面给一个简化的 Prompt 设计参考实际项目里你可以把它放到一个专门的 prompt 模板文件中管理你是一个紧急呼叫分流助手。请分析下面的报警电话转写文本输出 JSON字段如下 { event_category: medical/fire/crime/traffic/noise/other, urgency_level: critical/high/medium/low, address: 地址或交叉路口没有则为 null, injured: true/false, ongoing: true/false, summary: 一句话中文摘要 } 判断标准 1. critical正在发生的生命危险、枪击、火灾、昏迷、严重出血。 2. high交通事故、明显需要紧急响应但不属于即刻生命危险。 3. medium需要处理但不紧急。 4. low投诉、咨询、非紧急事件。 注意如果事件描述模糊不要强行猜测请选择低一级的紧急等级并在 summary 中说明不确定点。 报警转写文本 {transcript}这个 Prompt 有几个细节很重要一是明确“不确定时选择低一级等级”减少 AI 的过度自信二是要求字段固化方便代码解析三是给 summary 留出说明空间让坐席能快速看到 AI 的判断依据。4.3 分级决策模型建议 规则兜底 人工确认分级是整个系统里最敏感的一步。这里推荐“三层决策”而不是只信任模型输出。第一层是规则引擎先把绝对关键词命中比如“枪声”“fire”“昏迷”直接拉高优先级第二层是大模型综合语义判断处理规则覆盖不到的长尾表达第三层是人工坐席确认尤其是在生产环境初期。规则层看起来“原始”却是重要兜底。比如大模型服务超时、返回字段不完整、输出不合法 JSON 时系统必须能退回到关键词规则否则一通火警电话可能因为模型故障被当成普通事件。这就是工程里的“稳定优先于聪明”。4.4 摘要生成给坐席省时间坐席在高压力下没有时间读完整转写摘要要回答“发生了什么、在哪里、现在什么状态、需要什么资源”。摘要不应该有太长的形容词要尽量呈现事实。例如一通转写可能是“I’m at the corner of 5th and Main, I see smoke coming out of the building, and I think there’s a person inside” 摘要可以生成“5th 和 Main 路口建筑冒烟可能有人在内建议消防与急救联动确认。”这种摘要看起来简单但在高峰时段能显著降低坐席的认知负担。4.5 队列调度与 backlog 重排AI 完成结构化分析后不能只把结果丢进一个普通消息队列就结束。backlog 场景下调度逻辑必须支持“按紧急等级重新排队”。一个简单做法是队列里每个任务带 urgency_level 和 received_at消费端不按先进先出而是按“优先级 等待时间”计算综合分数。更稳妥的是用多个优先级队列critical 队列优先被消费low 队列甚至可以允许人工临时挂起。这样即使整体 backlog 仍然存在也能保证最危险事件不会排在后面。5. 最小可运行示例AI Triage Demo下面用一个 FastAPI 服务演示整个分流逻辑。为了让你能直接在本地跑起来这个 Demo 不依赖外部大模型服务而是用模拟函数返回结构化结果真实环境中你可以把模拟函数替换成内部大模型 API。5.1 环境准备Python 3.10 或以上版本pip 包管理工具一台可以本地运行 FastAPI 的机器不需要 GPU安装依赖文件requirements.txtfastapi0.100 uvicorn0.23 pydantic2.0安装命令pip install -r requirements.txt5.2 项目目录结构ai-triage-demo/ ├── main.py ├── models.py ├── triage_engine.py ├── requirements.txt └── README.md5.3 定义数据模型文件models.pyfrom enum import Enum from typing import Optional from pydantic import BaseModel, Field class UrgencyLevel(str, Enum): CRITICAL critical HIGH high MEDIUM medium LOW low class CallEvent(BaseModel): call_id: str transcript: str received_at: str 2025-01-01T00:00:00Z class TriageResult(BaseModel): call_id: str event_category: str Field(..., descriptionmedical/fire/crime/traffic/noise/other) urgency_level: UrgencyLevel address: Optional[str] None injured: bool False ongoing: bool False summary: str这里使用 Pydantic 的原因有二一是可以做请求参数校验二是可以用 FastAPI 自动输出 OpenAPI 接口文档。UrgencyLevel用枚举类型约束紧急等级避免脏数据进入下游调度系统。5.4 实现分流引擎文件triage_engine.pyimport os from models import CallEvent, TriageResult, UrgencyLevel def mock_llm_triage(transcript: str) - dict: text transcript.lower() if fire in text or 着火 in text or 冒烟 in text: return { event_category: fire, urgency_level: critical if (person in text or 受伤 in text) else high, address: None, injured: person in text or 受伤 in text, ongoing: True, summary: 疑似火情需确认具体地址与人员情况, } if shooting in text or gun in text or 枪 in text: return { event_category: crime, urgency_level: critical, address: None, injured: False, ongoing: True, summary: 涉及枪击或武器需立即响应, } if unconscious in text or heart in text or 昏迷 in text: return { event_category: medical, urgency_level: critical, address: None, injured: True, ongoing: True, summary: 疑似人员昏迷或心脏问题需急救, } if traffic in text or crash in text or 车祸 in text: return { event_category: traffic, urgency_level: high, address: None, injured: False, ongoing: True, summary: 交通事故需交警与医疗确认, } if noise in text or 吵闹 in text: return { event_category: noise, urgency_level: low, address: None, injured: False, ongoing: False, summary: 噪音投诉按低优先级处理, } return { event_category: other, urgency_level: low, address: None, injured: False, ongoing: False, summary: 事件描述模糊建议人工复核, } def call_llm_triage(transcript: str) - dict: api_key os.getenv(LLM_API_KEY, ) endpoint os.getenv(LLM_ENDPOINT, ) if not api_key or not endpoint: return mock_llm_triage(transcript) # 真实环境请替换为团队内部已授权的大模型接口。 # 调用要点 # 1. 设置超时时间和重试次数。 # 2. 要求模型只返回 JSON并对返回结果做 JSON 合法性校验。 # 3. 校验失败时回退到 mock_llm_triage 或规则引擎。 # 由于不同模型接口差异较大这里不再写死实现。 return mock_llm_triage(transcript) def triage_call(event: CallEvent) - TriageResult: raw call_llm_triage(event.transcript) return TriageResult( call_idevent.call_id, event_categoryraw.get(event_category, other), urgency_levelUrgencyLevel(raw.get(urgency_level, low)), addressraw.get(address), injuredraw.get(injured, False), ongoingraw.get(ongoing, False), summaryraw.get(summary, ), ) def reorder_calls(events: list[CallEvent]) - list[TriageResult]: results [triage_call(event) for event in events] priority_order { UrgencyLevel.CRITICAL: 0, UrgencyLevel.HIGH: 1, UrgencyLevel.MEDIUM: 2, UrgencyLevel.LOW: 3, } results.sort(keylambda r: (priority_order[r.urgency_level], r.call_id)) return resultsmock_llm_triage是一个规则函数用来模拟大模型的结构化输出。真实生产环境里call_llm_triage应该调用真实的模型接口并且在调用异常时回退到规则函数保证系统不会因为模型超时而中断。reorder_calls演示了 backlog 场景下的队列重排不按来电时间排序而是按紧急等级优先。5.5 实现 FastAPI 接口文件main.pyfrom fastapi import FastAPI, HTTPException from models import CallEvent, TriageResult from triage_engine import reorder_calls, triage_call app FastAPI(titleAI Triage Demo) app.post(/api/v1/triage, response_modelTriageResult) def handle_triage(event: CallEvent): try: return triage_call(event) except Exception as exc: raise HTTPException(status_code500, detailstr(exc)) app.post(/api/v1/backlog-reorder, response_modellist[TriageResult]) def handle_backlog_reorder(events: list[CallEvent]): try: return reorder_calls(events) except Exception as exc: raise HTTPException(status_code500, detailstr(exc))接口设计上/api/v1/triage负责单通电话分流/api/v1/backlog-reorder接收一批积压电话返回按紧急等级排序后的结果。真实项目里第二个接口更适合做成异步任务避免一次性计算时间太长这里用同步方式演示核心逻辑。6. 运行结果与效果验证启动服务uvicorn main:app --reload --port 8000用 curl 测试单通电话分流curl -X POST http://127.0.0.1:8000/api/v1/triage \ -H Content-Type: application/json \ -d {call_id:C-1001,transcript:There is a fire on Main Street and someone is injured,received_at:2025-01-01T00:00:00Z}预期输出{ call_id: C-1001, event_category: fire, urgency_level: critical, address: null, injured: true, ongoing: true, summary: 疑似火情需确认具体地址与人员情况 }第一次验证时重点检查三件事第一响应字段是否完整、是否符合TriageResult模型第二紧急等级判断是否符合预期火灾且有人受伤应该被判定为 critical第三如果返回 non-JSON 或者字段缺失服务会不会直接抛 500。这个 Demo 里 Pydantic 会自动校验但真实系统中还要考虑模型返回脏数据的情况。再测试 backlog 批量重排。用一段 Python 脚本直接调用from models import CallEvent from triage_engine import reorder_calls events [ CallEvent(call_idC-1003, transcriptI can hear gunshots outside), CallEvent(call_idC-1001, transcriptThere is a fire on Main Street and someone is injured), CallEvent(call_idC-1002, transcriptA car crashed into a pole, no one injured), CallEvent(call_idC-1004, transcriptMy neighbor is playing loud music), ] results reorder_calls(events) for r in results: print(r.call_id, r.urgency_level.value, r.event_category)预期输出顺序应该是枪击critical和火灾critical排在最前面然后是交通事故high最后是噪音投诉low。如果输出顺序不符合预期优先检查reorder_calls里的priority_order映射是否写错。运行失败时排查步骤可以按这个顺序来先确认pip list里 fastapi、uvicorn、pydantic 是否已经安装再确认启动命令所在目录是否有main.py然后看终端日志里有没有 import 错误最后用浏览器打开http://127.0.0.1:8000/docs看 FastAPI 自动生成的接口文档是否正常展示。如果是 Pydantic 版本过低导致枚举解析报错升级到 pydantic 2.x 即可。7. 常见问题与排查思路问题现象可能原因排查方式解决方案大模型返回非法 JSON接口报 500模型输出附带了解释文字或 Markdown 代码块打印完整响应检查前后是否有额外字符用正则抽取 JSON 片段解析失败后回退规则函数非紧急事件被误判为 critical关键词过于宽泛或 Prompt 缺乏“不确定时降级”约束检查命中规则和 Prompt 输出日志增加置信度字段降低不确定时的默认等级ASR 把地址听错调度车辆找错地方语音转写缺少地理词典校正对比原始音频和转写结果增加地址后处理配合地理编码服务做二次匹配高峰并发时模型调用超时大模型接口响应慢或队列消费无流控查看模型调用耗时 P99 和队列堆积量设置超时与熔断超额流量降级到规则引擎系统离线时所有电话都受影响缺少本地降级机制检查依赖链路是否有单点AI 服务异常时不阻塞主流程自动切换人工模式坐席直接采信 AI 结果不做复核界面设计或培训不到位观察坐席改派率、采纳率在界面中突出“建议”标签要求确认方可派单这些问题的共同点在于AI 只是链路中的一个环节真正决定系统稳定性的往往是链路周围的工程能力比如超时处理、降级策略、日志监控和人工审核。这也是很多人做大模型应用时容易忽略的部分。8. 最佳实践与工程建议8.1 人在回路是底线紧急呼叫场景不能做成无人化闭环。不是说 AI 不够强而是当一个系统要处理生命安全相关事件时责任边界必须清楚。建议把 AI 定位成“辅助决策建议器”坐席看到 AI 结果后需要确认或修改。如果未来要做自动处理也只允许在低风险、高置信度的场景下生效比如噪音投诉自动标记低优先级。8.2 数据合规与隐私保护911 通话包含大量隐私数据包括位置、姓名、健康状况甚至语音本身。系统建设必须遵循最小必要原则只采集分流所需的文本信息不保存与调度无关的语音内容如果要留存语音用于模型迭代需要脱敏和授权所有数据访问要有审计日志。国内类似系统还要遵循本地法律法规不能简单照搬美国方案。8.3 可观测性上线前就要想清楚监控指标。至少需要跟踪AI 分流耗时、坐席采纳率、人工修改率、误报率、漏报率、队列积压量、模型调用失败率。特别是“人工修改率”它直接反映 AI 结果可靠程度。建议把 AI 输出和人工最终结果都落到日志表里定期做差异分析样本量足够后按事件类别和紧急等级分别统计准确率。8.4 降级和回滚AI 分流服务应当是一层“可以去掉的能力”而不是“去不掉的前置依赖”。当模型服务不可用时系统要能回到纯人工模式当模型质量明显下降时要能一键关闭高级判断退回到关键词规则。实现方式可以是配置中心里的开关也可以是发布平台的 feature flag。无论哪种方式关键是要保证紧急呼叫主流程永远不会被 AI 故障卡死。8.5 上线路线图更稳妥的上线路径是先在影子模式跑数周积累一批真实转写文本让 AI 结果和人工坐席的最终结论做离线对比找到明显的错误模式修正 Prompt 和规则后进入辅助模式小范围试点稳定后再逐步扩大覆盖面。不要直接从离线测试跳到全量生产紧急呼叫场景没有一个小的试错成本。9. 总结与后续学习方向回到新奥尔良这个案例AI 给 911 电话做 triage 的真正价值是把“每个电话都必须马上被人工听完”这个硬约束改成“每个电话都能被快速结构化最高危的那一部分先被人工处理”。实现这种能力最关键的并不是模型本身有多聪明而是队列策略、人工确认、规则回退、监控告警这些工程细节是否到位。如果你想沿着这个方向继续深入可以按这几个步骤实践先在自己熟悉的业务里找一个积压队列梳理出排队任务的关键字段然后用大模型或规则函数做结构化抽取把结果接入现有的工单系统再设计一套人工复核和降级机制最后用历史数据回测比较“使用 AI 前后的高优先级事件响应时间”是否有改善。这个流程不限于应急呼叫客服中心、维修工单、值班告警都能复用同一套思路。更底层的方向也值得研究排队理论里 M/M/c 模型如何评估积压风险异步事件驱动架构如何应对高并发语音识别后处理如何和地理信息结合以及如何设计一个能在 200 毫秒内完成分级判断的低延迟推理服务。这些内容每一块都能单独成文但先把本文的 Demo 跑通你会更容易理解这些方向各自解决的是链路中的哪一段问题。