AI应急呼叫原型设计:从大模型到结构化事件卡的技术拆解 📅 发布时间:2026/8/27 11:11:37 👁 浏览次数: AI 版 911 什么时候出现这个问题在 Hacker News 上经常被讨论。抛开“全自动应急中心”的科幻叙事真正值得拆解的是 AI 在紧急呼叫场景里能否辅助接警员完成信息识别、风险分级和调度建议。本文从工程实践视角出发设计一个“AI 应急呼叫处理原型”覆盖链路拆分、数据模型、大模型调用、规则兜底、本地验证和排错路径。原型不能代替真实 911 系统但能帮你理解这类系统要解决的核心问题。1. 先理解 AI 版 911 的技术边界1.1 911 场景不是普通的客服对话很多人会误以为 AI 版 911 就是把智能客服换一个语音入口。真实差别很大。普通客服对话容忍延迟、允许重复提问、问题域相对固定应急呼叫却是在极端情绪、环境嘈杂、信息碎片化的前提下快速完成一次“从语音到事件画像”的转换。一通典型求助电话可能包含以下信息说话人语速快、句子不完整甚至夹杂哭声或背景噪音。地址信息可能被反复确认可能包含地标而不是门牌号。事件类型需要尽快判断比如火灾、医疗急救、治安事件或交通事故。同一通电话里可能有多个受害者需要持续跟踪。接警员不仅要记录还要在几十秒内决定是否需要调度警察、消防还是急救。这些要求决定了 AI 不能只做文本摘要它必须输出结构化、可核验、可路由的事件信息。能力普通客服 AIAI 应急呼叫响应延迟要求秒级甚至可接受等待优先保证前几秒内产出关键信息信息完整性可以反复追问补齐需要在碎片化信息下先给初步结果输出要求对话回复结构化事件卡 风险等级错误代价客服工单错误可补救错误分类可能影响资源调度合规要求一般隐私协议通话录音、处置链路需可追溯1.2 AI 在这条链路中承担什么角色如果目标是“AI 接听 911”角色定义会非常容易引起争议。更稳妥的分工方式是把 AI 放到接警员旁边的“处理引擎”位置。它做四件事第一把语音实时转写成文字。这一步不是纯粹的 ASR还需要处理「嗯」「那个」「喂」等语气词并标记听不清的片段。第二识别事件类型和关键要素。从连续文本里抽取出地点、人数、是否持械、是否意识清醒、是否火灾冒烟等结构化字段。第三给出优先级和调度建议。根据要素组合建议“立即派救护车”“消防到场前先确认起火点”“警察到场但不需要 SWAT”等方案。第四生成辅助记录。接警员可修改、确认、驳回 AI 输出最终记录进入事件档案。这四件事仍然需要人来做最终决策。AI 的输出是建议不是命令。1.3 为什么不能简单当成大模型聊天大模型适合开放域问答但直接拿聊天式输出作为应急系统结果会有几个问题。首先幻觉不可接受。模型可能把听不清楚的地址补成一个貌似合理的地址这在真实调度里会误导车辆。其次延迟不稳定。一次大模型调用可能需要几百毫秒到几秒如果整个链路串行多次大模型调用接警员等不起。第三输出格式不稳定。语言模型返回的自然语言可能漏字段导致下游解析失败。第四责任边界不清。如果系统自动派单但分类错误需要明确责任归属。纯模型决策很难满足这种审计要求。所以工程上常见的做法不是“用大模型一步生成结论”而是“用大模型做语义理解用规则做约束用人工做确认”。这也是本文原型采用混合链路的原因。2. 搭建智能应急响应流的整体架构2.1 从一通电话到一张事件卡在设计架构前先把一次呼叫的数据流转走出来。输入是一条转写后的文本常见内容像这样“有人在中山路附近受伤了好像是被车撞了他已经躺在地上不动了请快点派人来。”系统要对这段文本输出一张事件卡{ event_type: traffic_accident, address: 中山路附近, people_count: 1, unconscious: true, bleeding: false, weapon: false, fire: false, priority: 1, dispatch_suggestion: police_ambulance }这张事件卡就是 AI 版本 911 的核心产物。接警员看到它可以快速确认信息而不是从原始语音里重新听完一遍。2.2 模块划分和消息流转原型可以拆成 5 个模块接入层接收文本请求并做基础校验。理解层调用大模型把文本转成结构化意图。兜底层用关键词和规则修正或补齐模型结果。决策层根据结构化结果计算优先级、建议调度部门。审计层记录原始文本、中间结果、最终结果和处理耗时。消息流转不需要引入重型消息队列用函数调用就可以串联。生产环境会换成异步事件流例如 Kafka 或 RabbitMQ但核心数据流一致。文本输入 - 校验参数 - 调用 LLM 识别事件要素 - 解析 JSON - 规则规则兜底修正 - 计算优先级和调度建议 - 输出事件卡 - 写入审计日志2.3 技术选型参考原型使用 FastAPI 提供 HTTP 接口使用 OpenAI SDK 调用兼容接口的语言模型。原型的 ASR 层并不真实接入而是用文本模拟。生产技术选型可以参考这张表。模块原型选型生产需要考虑的方向API 框架FastAPI高并发、链路追踪、限流语音转写暂不接入文本模拟实时 ASR、方言支持、噪音消除语义理解大模型 JSON 输出自建模型微调、私有化部署规则引擎Python 正则 关键词支持动态规则热更新事件存储SQLite / JSON 文件高可用数据库、冷热分离审计日志本地日志文件加密存储、权限隔离技术选型原则是原型用最少依赖跑通生产再逐步替换组件。3. 开发环境与最小可运行项目3.1 准备 Python 环境和依赖建议使用 Python 3.10 以上版本创建独立虚拟环境避免依赖冲突。项目依赖如下fastapi0.115.6 uvicorn0.32.1 openai1.55.3 pydantic2.10.4 python-dotenv1.0.1安装命令python -m venv venv source venv/bin/activate pip install -r requirements.txtWindows 环境将source venv/bin/activate换成venv\Scripts\activate。注意openai SDK 版本要求 1.x。0.x 版本的访问方式不同容易在后续代码中出现方法不存在的问题。3.2 项目结构最小工程结构如下重点是把模型调用、解析、规则、路由拆开方便单独测试。ai_911/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── models.py │ ├── config.py │ ├── llm_client.py │ ├── rules.py │ └── dispatcher.py ├── requirements.txt ├── .env └── README.md严格遵守模块边界main.py只负责 HTTP 路由不写业务逻辑llm_client.py只负责模型调用rules.py负责规则兜底dispatcher.py负责优先级和部门建议。3.3 核心数据模型在app/models.py中定义请求和响应模型。请求是文本响应是事件卡再附加一条解析状态。from typing import Optional from pydantic import BaseModel, Field class CallTextIn(BaseModel): text: str Field(..., min_length1, max_length2000) call_id: str Field(..., description呼叫唯一标识) language: str Field(zh, description语音原始语言) class EventCard(BaseModel): event_type: str address: str people_count: int 1 unconscious: bool False bleeding: bool False weapon: bool False fire: bool False priority: int Field(ge1, le3, description1表示最高3表示普通) dispatch_suggestion: str class ProcessResponse(BaseModel): call_id: str card: Optional[EventCard] None raw_result: str parse_success: bool False used_rule_fallback: bool False latency_ms: int 0priority使用 1 到 3 表示应急级别。1 为立即响应2 为优先响应3 为普通响应。这个设定在生产环境中应该对应具体的 SLA 指标。3.4 配置管理在.env中管理 API 密钥和模型名。不要把密钥写进代码。OPENAI_API_KEYyour-api-key OPENAI_BASE_URLhttps://api.openai.com/v1 LLM_MODELgpt-4o-miniconfig.py读取这些配置import os from dotenv import load_dotenv load_dotenv() class Settings: openai_api_key: str os.getenv(OPENAI_API_KEY, ) openai_base_url: str os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) model_name: str os.getenv(LLM_MODEL, gpt-4o-mini) settings Settings()如果使用本地部署的兼容模型只需修改OPENAI_BASE_URL指向本地服务地址不需要改代码。4. 实现核心判断逻辑从文本到应急等级4.1 设计 Prompt 模板大模型的输出质量在相当大程度上取决于 Prompt 是否限定输出格式。这个项目不要求模型生成解释性长文本而是要求返回稳定 JSON。模板直接写在llm_client.py中SYSTEM_PROMPT 你是一个应急呼叫信息提取助手。你从接警员转写文本中提取关键事件信息。 只输出 JSON不要输出其他解释。JSON 字段如下 { event_type: fire/medical/traffic_accident/police/other, address: 地点描述, people_count: 人员数量, unconscious: true/false, bleeding: true/false, weapon: true/false, fire: true/false } 判断规则 1. event_type 只能取给定枚举值。 2. address 可以是模糊描述不要强行补全成标准地址。 3. 文本没有提到的布尔字段默认 false。 4. people_count 文本没有提到时默认 1。 5. 遇到“不清楚”“不知道”等内容不要编造按 false 或 1 处理。 这里刻意要求“不要强行补全地址”。真实系统的地址错误可能引发次生问题算法宁可模糊也不能编造。4.2 调用大模型并解析结果llm_client.py使用 OpenAI SDK 发送请求。关键点是开启response_format为json_object尽可能降低非法输出概率。import json from openai import OpenAI from app.config import settings client OpenAI( api_keysettings.openai_api_key, base_urlsettings.openai_base_url, ) def extract_with_llm(text: str): response client.chat.completions.create( modelsettings.model_name, temperature0, response_format{type: json_object}, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: text}, ], ) content response.choices[0].message.content try: data json.loads(content) return data, True except json.JSONDecodeError: return {}, Falsetemperature固定为 0。应急场景需要可复现和稳定输出不需要创造性表达。如果模型接口不支持response_format可以去掉该参数但下游解析代码必须做好失败兜底。4.3 加入规则兜底模型可能拒绝输出、返回空 JSON或者在文本包含极端信息时返回异常结果。规则层此时要接管。rules.py包含两类逻辑关键词修正和缺失字段补齐。import re KEYWORD_RULES { fire: [火, 烧, 冒烟, 着火, 爆炸], traffic_accident: [车祸, 撞车, 被车撞, 交通事故, 翻车], medical: [昏迷, 流血, 心脏病, 呼吸困难, 晕倒, 受伤], police: [打架, 抢劫, 小偷, 刀, 枪, 袭击], } def rule_parse(text: str): detected set() for event_type, keywords in KEYWORD_RULES.items(): for kw in keywords: if kw in text: detected.add(event_type) card { event_type: other, address: , people_count: 1, unconscious: False, bleeding: False, weapon: False, fire: False, } if len(detected) 1: card[event_type] detected.pop() if 昏迷 in text or 不动 in text: card[unconscious] True if 流血 in text or 血 in text: card[bleeding] True if 刀 in text or 枪 in text or 凶器 in text: card[weapon] True if 火 in text or 烟 in text: card[fire] True return card规则层不要试图回答所有情况它只负责“模型不可用时给出保守结果”以及“修正模型明显漏掉的强信号关键词”。4.4 优先级路由决策dispatcher.py根据事件卡计算优先级和建议调度部门。这里的规则要尽量简单、可解释。def dispatch_advice(card: dict): event_type card.get(event_type, other) weapon card.get(weapon, False) unconscious card.get(unconscious, False) bleeding card.get(bleeding, False) fire card.get(fire, False) if fire: return 1, fire_department if weapon: return 1, police if unconscious or bleeding: return 1, ambulance if event_type traffic_accident: return 2, police_ambulance if event_type medical: return 2, ambulance if event_type police: return 2, police return 3, operator_review这层逻辑本质上是决策表。使用硬编码在原型阶段完全足够。生产环境建议把决策表外置到配置中心让业务人员可以在不发布代码的情况下调整规则。5. 运行验证与测试5.1 启动服务在app/main.py中实现 FastAPI 路由import time from fastapi import FastAPI from app.models import CallTextIn, ProcessResponse from app.llm_client import extract_with_llm from app.rules import rule_parse from app.dispatcher import dispatch_advice app FastAPI(titleAI 911 Prototype) app.post(/api/v1/process, response_modelProcessResponse) def process_call(payload: CallTextIn): start time.time() data, success extract_with_llm(payload.text) used_fallback False if not success or not data: data rule_parse(payload.text) used_fallback True card { event_type: data.get(event_type, other), address: data.get(address, ), people_count: data.get(people_count, 1), unconscious: bool(data.get(unconscious, False)), bleeding: bool(data.get(bleeding, False)), weapon: bool(data.get(weapon, False)), fire: bool(data.get(fire, False)), } priority, dispatch dispatch_advice(card) card[priority] priority card[dispatch_suggestion] dispatch latency_ms int((time.time() - start) * 1000) return ProcessResponse( call_idpayload.call_id, cardcard, raw_resultjson.dumps(data, ensure_asciiFalse), parse_successsuccess, used_rule_fallbackused_fallback, latency_mslatency_ms, )启动命令uvicorn app.main:app --host 0.0.0.0 --port 80005.2 用 curl 模拟呼叫文本开启服务后打开另一个终端发送请求curl -X POST http://127.0.0.1:8000/api/v1/process \ -H Content-Type: application/json \ -d { call_id: test-001, text: 有人被车撞了倒在中山路路口不动了有流血请快点来, language: zh }5.3 预期返回正常情况会收到类似下面的响应{ call_id: test-001, card: { event_type: traffic_accident, address: 中山路路口, people_count: 1, unconscious: true, bleeding: true, weapon: false, fire: false, priority: 1, dispatch_suggestion: police_ambulance }, raw_result: {\event_type\: \traffic_accident\, ...}, parse_success: true, used_rule_fallback: false, latency_ms: 820 }关键是确认parse_success为 truecard中布尔字段符合文本含义。5.4 用固定测试集评估不要只靠一两个样例评估。推荐准备一个小的测试集用脚本批量跑并输出指标。test_cases [ {text: 着火了厨房冒烟了, expected_event: fire}, {text: 有人拿着刀在街上走, expected_event: police}, {text: 老人晕倒在家里没呼吸了, expected_event: medical}, ] for case in test_cases: resp requests.post(http://127.0.0.1:8000/api/v1/process, json{ call_id: eval, text: case[text], }) card resp.json().get(card, {}) ok card.get(event_type) case[expected_event] print(case[text], -, card.get(event_type), OK if ok else FAIL)评估时至少记录三项指标事件类型准确率、关键字段准确率、平均延迟。后续每次改 Prompt 都要重新跑测试集防止“修了一个案例坏了一片案例”。注意不要只验证程序能启动还要验证输入、输出、异常分支和日志是否符合预期。特别是模型返回非法 JSON 时系统必须能自动降级到规则层。6. 常见问题与排查链路6.1 模型返回的 JSON 无法解析现象调用接口成功但parse_success始终为 false响应里used_rule_fallback为 true。可能原因模型返回了 Markdown 代码块、解释文本或者字段名和 Prompt 定义不一致。某些情况下模型会输出json包裹的文本直接json.loads会失败。检查方式在日志中打印raw_result观察实际返回内容。处理方案第一优先开启response_format{type: json_object}第二在解析前剥离代码块标记第三保留规则兜底。预防建议把 Prompt 输出约束写清楚并在测试集里加入“包含数字、英文、乱码”的对抗样本。6.2 语音识别错误被当成事实现象文本里地址识别错误模型基于错误文本给出错误事件卡。这类问题的根因不在大模型而在上游 ASR。原型没有接入 ASR所以容易忽略这一点。处理方案生产系统必须将 ASR 结果同时返回置信度低于阈值的片段标记为unclear并在事件卡里增加address_confidence字段。接警员看到低置信度地址时需要人工确认。预防建议在 Prompt 中增加“如果地址描述不完整不要补全字段可以是空字符串”。同时在 UI 上把低置信度文本用不同颜色展示。6.3 响应延迟过高现象单次请求耗时超过 3000ms而真实应急场景可能只有几秒决策窗口。可能原因模型调用本身慢temperature0不一定保证延迟服务所在地区与 API 节点网络延迟较高每次请求都重复加载模型或裸调用。检查方式在日志里打印latency_ms并分别统计 LLM 调用耗时和其余逻辑耗时。处理方案增加缓存层相同类型呼叫只请求一次模型。把关键决策拆成两级先用规则快速给出初判再用模型异步完善。生产环境使用流式输出把模型结果逐步推给前端。如果采用本地模型需要单独评估 GPU 推理性能。6.4 并发和数据安全现象系统上线后并发增加服务出现超时或者审计日志里出现完整手机号、身份证号等敏感信息。并发处理FastAPI 同步接口在高并发下可能阻塞。原型中可以把process_call改成async def并在依赖中加入限流组件。生产环境使用消息队列削峰。数据安全原始呼叫文本属于高敏感数据。日志和数据库必须做脱敏处理模型调用尽量使用私有化部署避免把原始文本发到外部 API。原型代码只做功能演示不能直接把真实呼叫数据接入公共模型服务。7. 从原型到真实的 AI 911 还差什么7.1 真实系统中的工程约束这个原型能跑通但距离真实 911 系统非常远。真实系统需要满足四个约束。第一低延迟和可用性。调度系统不是“能跑就行”而是要求 99.99% 可用性单次处理延迟需要被严格限制。第二责任闭环。AI 给出的任何建议都要有对应人工确认状态。事件卡要记录模型版本、调用参数、规则命中原因、确认人员。第三持续评估。呼叫场景会随季节和地域变化。夏天溺水事件多冬天一氧化碳中毒事件多。模型需要持续迭代和回归。第四隐私与合规。呼叫录音、转写文本、位置信息都属于公民敏感数据。必须设计访问控制、加密存储和审计制度。7.2 可落地的渐进路径不需要等“全自动 AI 911”一步到位。工程上可以分三个阶段推进。第一阶段辅助记录。只做实时转写和字段抽取生成事件卡初稿接警员强制确认。这个阶段不掉以轻心主决策权仍由人承担。第二阶段辅助建议。规则和模型联合给出优先级和调度建议接警员可以一键采纳或驳回。系统记录每次采纳比例作为模型效果指标。第三阶段特定场景自动派单。在低风险、高规则化场景中比如“凌晨的固定烟雾报警”允许自动派单。但必须配置回滚机制一旦识别到异常立即转人工。7.3 最佳实践清单回到文章开头的问题AI 版 911 什么时候真正到来从工程角度看不是某一个时间点而是一系列能力逐步落地的过程。下面的清单可以用于检查和指导项目迭代建立真实脱敏数据集覆盖不同方言、噪音、情绪和事件类型。事件卡字段必须稳定模型输出和数据库表字段一一对应。规则兜底层永远保留不依赖模型可用性。所有模型输出记录版本号方便回放和审计。地址解析不要使用大模型“猜”使用独立地址服务或让接警员确认。延迟指标和准确率指标分开统计不能为了准确率牺牲可用性。上线前准备“模型熔断”方案模型调用失败时系统只保留规则和人工流程。定期用最近一个月的真实脱敏数据回测 Prompt 和决策表防止概念漂移。AI 版 911 的本质不是用 AI 替代接警员而是把接警员从“听写、判断、翻手册”中解放出来让他们把精力放在最需要对人的判断上。这个目标不需要等到技术完全成熟从今天的小型辅助系统就可以开始。