列车事件文本结构化：基于FastAPI的轻量解析服务实现

列车进站信息这类文本看起来只是一句话但对自动化和信息展示系统来说最大的工作量往往在于把它变成结构化字段。以“沪太标杆Z198进石家庄北站”为例人工读起来很轻松车次是 Z198动作是进站车站是石家庄北站前面还有“沪太”这个方向/线路标记。要让程序自动识别出这些信息再把结果交给大屏、客服系统或推送服务就需要一套可重复调用的解析接口。本文围绕这个场景实现一个本地就能跑的列车事件解析服务带接口、带批量任务不依赖显卡纯 CPU 就能启动。需要先说明边界本文不做列车动态时刻预测也不提供任何实时车次数据源更不会教你去抓取非公开接口。重点是把“文本变结构化”这层能力做好。真实列车是否停靠某个站、几点进站必须以车站公告和铁路官方发布为准。下面我们把这套服务从环境准备到接口调用完整跑一遍。1. 核心能力速览这个示例项目的核心能力并不复杂但它覆盖了后续接业务时最常用的几个模块。先看一张速览表。能力项说明项目类型列车运行信息文本解析与 HTTP 接口服务核心输入类似“沪太标杆Z198进石家庄北站”的原始文本核心输出车次、车站、事件类型、线路别名、修饰标记等结构化 JSON主要功能单条解析、批量解析、健康检查、可扩展事件关键词硬件门槛普通 CPU 即可运行不需要 GPU显存占用不使用深度学习后端时不涉及显存如果接入大模型则由模型推理环境决定启动方式Python 虚拟环境 uvicorn 命令行启动接口能力GET /health、POST /api/parse、POST /api/batch批量任务支持 /api/batch 同步批量也可以脚本化处理本地文本文件技术栈Python 3.10、FastAPI、Pydantic、正则表达式适合场景车站通知内容结构化、列车信息大屏、客服辅助、事件推送测试从表里可以看到这不是一个“大模型 高显存”项目而是一个偏向工程落地的轻量服务。它的优点是启动快、依赖少、方便接入现有系统。缺点是规则式解析对自然语言变化的容忍度有限复杂文本需要考虑引入分词或大模型。这个示例适合三类读者第一类是想把列车信息文本接口化的服务端开发第二类是在做车站大屏、列车到发消息推送的测试工程第三类是刚开始接触 FastAPI想找一个真实场景练手的人。如果你关心的是实时列车的运行位置、正晚点数据这类数据应该单独对接正规数据源不由本文的解析器负责。2. 适用场景与使用边界以“Z198进石家庄北站”为例这个服务能解决的是“信息入口统一化”的问题。很多系统会从不同渠道拿到列车信息文本有时来自人工录入有时来自通知文件有时来自车站广播的转写结果。这些文本如果直接丢给前端很难做搜索、聚合和告警。解析服务做的就是把它们统一成字段方便下游使用。这套解析方法适合以下场景。第一消息归档场景每天产生大量类似“G1次列车到达北京南站”“K1234通过郑州站”的文本手工录入费时费力自动解析后可以写入数据库。第二大屏展示场景不需要展示原文而是显示车次、状态、站名这时结构化输出是刚性需求。第三事件触发场景解析出“进站”后可以继续触发短信通知、控制室弹窗或语音播报。不适合的场景同样要讲清楚。第一不适合做实时列车位置服务文本里根本没有到发时间缺乏计算准点率的数据基础。第二不适合直接从非官方渠道持续采集列车动态数据接口稳定性、数据版权、服务条款都存在风险合规边界不清晰。第三不适合处理大量无规律口语比如“那趟去石家庄北的Z198应该快进站了吧”规则解析会失败这种场景需要大模型和更完整的上下文。使用边界上必须强调授权和数据合规。如果你做的系统会展示真实铁路信息使用前要确认数据来源是否获得授权。发布到公网时建议加访问控制。生产环境中任何解析结果都不能单独作为行车依据一定要与官方运行图或调度系统校验。3. 环境准备与前置条件先准备一个干净的本地目录和 Python 环境。本文示例基于 Python 3.10 或更高版本建议使用虚拟环境避免污染系统 Python。3.1 检查 Python 环境在终端执行python --version如果输出版本低于 3.10需要先安装新版 Python。Windows 用户在安装时记得勾选 Add Python to PATH。Linux/macOS 用户通常自带 Python 3不过版本可能比较旧需要按系统方式更新。3.2 创建项目目录mkdir rail-event-parser cd rail-event-parser3.3 创建虚拟环境并安装依赖python -m venv venv source venv/bin/activateWindows 下激活命令不同venv\Scripts\activate激活成功以后终端行首会出现(venv)。然后创建requirements.txt并写入依赖。fastapi uvicorn[standard] pydantic requests安装依赖pip install -r requirements.txt这里不锁版本号是为了避免示例代码被将来升级彻底卡住。如果你需要固定版本可以根据实际安装结果把版本号写死。依赖安装完成后不需要装 CUDA也不需要配置显卡驱动。4. 安装部署与启动方式项目的目录结构建议这样组织。rail-event-parser/ ├── main.py ├── parser.py ├── batch_process.py ├── data/ │ └── input_texts.txt └── requirements.txtparser.py负责文本解析main.py负责提供 HTTP 接口batch_process.py负责本地批量文件处理。三个文件职责分开后续替换算法或扩展接口都方便。4.1 编写解析器parser.py使用正则和关键词表完成车次、车站和事件识别。正则方案虽然朴素但对近几年的列车车次格式、常见车站名结构已经足够演示。import re from typing import Any, Dict TRAIN_NO_RE re.compile(r[A-Z]\d{1,4}) STATION_RE re.compile( r(?:进|到达|通过|出发|开往|终到|到)([\u4e00-\u9fa5]?)(?:站|$) ) EVENT_MAP [ (到达, 到达), (通过, 通过), (出发, 出发), (开往, 开往), (终到, 终到), (进站, 进站), ] def _find_event(text: str) - str: for keyword, event_type in EVENT_MAP: if keyword in text: return event_type if re.search(r进[\u4e00-\u9fa5]站, text): return 进站 return def parse_rail_event(text: str) - Dict[str, Any]: raw_text text.strip() m TRAIN_NO_RE.search(raw_text) train_no m.group(0) if m else alias alias_tag False if m: prefix raw_text[: m.start()] if 标杆 in prefix: alias_tag True alias prefix.replace(标杆, ) station_m STATION_RE.search(raw_text) station if station_m: station station_m.group(1) 站 event_type _find_event(raw_text) return { raw_text: raw_text, train_no: train_no, line_alias: alias, is_benchmark_mark: alias_tag, station: station, event_type: event_type, parser: rule, }这段代码的关键点有四个。第一车次用[A-Z]\d{1,4}匹配能覆盖 G、D、C、Z、K、T 等常见车次前缀第二站名匹配放在动作词之后并自动补“站”字第三进站事件专门做了补充判断兼容“进石家庄北站”这种把站名直接放在“进”后面的写法第四修饰标记is_benchmark_mark用于判断原文中是否出现“标杆”这类车辆运行标记。4.2 编写 FastAPI 接口main.py提供两个主要接口单条解析和批量解析。批量解析在演示阶段做成同步接口数据量不大的时候够用。from fastapi import FastAPI from pydantic import BaseModel, Field from parser import parse_rail_event app FastAPI(titleRail Event Parser, version0.1.0) class ParseRequest(BaseModel): text: str Field( ..., description列车运行信息原始文本例如沪太标杆Z198进石家庄北站, ) class BatchRequest(BaseModel): texts: list[str] Field( ..., description多条原始文本, ) app.get(/health) def health(): return {status: ok} app.post(/api/parse) def parse(req: ParseRequest): return {code: 0, data: parse_rail_event(req.text)} app.post(/api/batch) def parse_batch(req: BatchRequest): results [] for text in req.texts: item {text: text, result: parse_rail_event(text)} results.append(item) return {code: 0, count: len(results), data: results}这个接口文件很短但包含了生产接口的基本形态请求用 Pydantic 模型做校验响应统一返回code字段便于前端判断。后续如果要加入鉴权、日志、限流都可以在这个结构上扩展。4.3 启动服务确认当前目录是rail-event-parser然后执行uvicorn main:app --host 127.0.0.1 --port 8000启动成功后终端会显示类似Uvicorn running on http://127.0.0.1:8000的日志。默认监听 127.0.0.1也就是只有本机能访问。如果需要在局域网内调试可以把--host改成0.0.0.0但要注意访问控制。5. 功能测试与效果验证服务启动后按照“单条解析 - 批量解析 - 错误样例对照”的顺序测试效果。5.1 单条解析测试先测试最核心的一条输入沪太标杆Z198进石家庄北站。用 curl 发起请求curl -X POST http://127.0.0.1:8000/api/parse \ -H Content-Type: application/json \ -d {text:沪太标杆Z198进石家庄北站}预期返回类似{ code: 0, data: { raw_text: 沪太标杆Z198进石家庄北站, train_no: Z198, line_alias: 沪太, is_benchmark_mark: true, station: 石家庄北站, event_type: 进站, parser: rule } }判断成功的标准是train_no为Z198station为石家庄北站event_type为进站line_alias为沪太。这四个字段只要有一个为空就要检查正则是否匹配。5.2 批量解析测试在data/input_texts.txt中写入待解析文本每行一条。可以是这种形式沪太标杆Z198进石家庄北站 G1次列车到达北京南站 K1234通过郑州站 D123出发站为上海虹桥站执行本地批量脚本python batch_process.py --input data/input_texts.txt --output data/output_result.json查看输出文件cat data/output_result.json批量解析成功后输出 JSON 里每条记录都会包含解析结果和 status 字段。这样以后可以把原始文本文件当作消息队列输入解析完再统一入库。5.3 规则扩展测试规则解析器的上限取决于关键词表。增加车站、事件词时不要在代码里堆一堆 if而是把关键词集中管理。更稳妥的做法是准备一个地名库先把站名库加载到内存再结合动作词搜索可以有效减少误匹配。例如遇到“Z198石家庄北站进站”这种把站名放在动作前的文本现有规则会失败。解决办法是在_find_event前增加“先定位站名再定位事件”的两段式解析。实际生产中建议先写 20 到 50 条代表性的原始文本作为测试集再逐步补充规则。5.4 结果校验方法文本解析结果不能只看一两条。建议把测试集分成两组。第一组是“正常句式”例如“沪太标杆Z198进石家庄北站”第二组是“干扰句式”例如“Z198次是否进站”这种问句或“石家庄北站Z198进站”这种倒装句。运行后用人工比对准确率。规则解析的目标是固定句式上达到高准确率在不固定句式上做到不崩溃。6. 接口 API 与批量任务如果要把这个服务接进自己的工具可以直接使用 HTTP API。下面对核心接口做拆解。6.1 健康检查接口curl http://127.0.0.1:8000/health返回{ status: ok }这个接口适合部署后用监控系统定时探测。如果服务进程异常请求会超时或直接拒绝连接。6.2 单条解析接口请求参数字段类型是否必填说明textstring是原始列车信息文本返回结构{ code: 0, data: { raw_text: 沪太标杆Z198进石家庄北站, train_no: Z198, line_alias: 沪太, is_benchmark_mark: true, station: 石家庄北站, event_type: 进站, parser: rule } }6.3 Python 调用示例Python 调用接口可以直接用 requests 库。import requests url http://127.0.0.1:8000/api/parse payload {text: 沪太标杆Z198进石家庄北站} response requests.post(url, jsonpayload, timeout10) print(response.json())如果返回的code不是 0需要检查文本是否为空、请求头是否正确。如果网络不通先确认 uvicorn 进程是否还在运行。6.4 批量解析 API批量接口接收一个texts数组。curl -X POST http://127.0.0.1:8000/api/batch \ -H Content-Type: application/json \ -d {texts:[沪太标杆Z198进石家庄北站,G1次列车到达北京南站,K1234通过郑州站]}这种方式适合几千条以内的同步解析。一旦到达万级或十万级建议改为异步任务服务先把原始文本写入消息队列后台 worker 消费解析再把结果写入数据库或文件。本文的batch_process.py就是 worker 的简化版实际生产可以把循环体替换成从 Redis Stream 或 RabbitMQ 消费。6.5 失败重试设计批量解析不是一个“永远成功”的过程。单条文本可能触发表述不规范导致字段为空。外部数据源如果一条条取回来再解析网络错误也会中断任务。设计时要记录每条解析的状态。成功记录写结果失败记录写原始文本和错误原因方便重跑。7. 资源占用与性能观察先强调一点不要看文字推断显存文本解析服务不启动深度学习模型时没有显存占用。如果你在服务器上同时用 NVIDIA 显卡跑大模型那么显存占用要看的是大模型推理进程不是这个解析服务。本地观察资源占用可以分成三步。第一步启动服务前记录一次内存基线第二步连续请求/api/parse第三步观察进程内存是否持续上涨。如果内存只增不减可能是有大量对象没有释放需要用 profiling 工具检查。如果服务器是 Linux可以用top或htop观察 python 进程。如果要更精细一点可以用 Python 内置工具。import psutil import os import time process psutil.Process(os.getpid()) while True: memory_mb process.memory_info().rss / 1024 / 1024 cpu_percent process.cpu_percent(interval1) print(fmemory{memory_mb:.2f}MB cpu{cpu_percent:.1f}%) time.sleep(1)需要先安装 psutilpip install psutil通过这个脚本可以判断规则解析在高并发下 CPU 使用率稳定内存占用接近启动后的基线这是理想状态。如果出现内存持续上涨优先排查是否有大量日志对象、请求体被全局缓存。影响性能的主要因素有两个。第一是请求文本长度但列车文本通常很短影响有限。第二是并发量FastAPI 本身是异步框架但解析函数是同步的计算逻辑它不会阻塞事件循环太久不过高并发时建议用run_in_executor把同步解析丢到线程池执行。你还可以做压测ab -n 1000 -c 20 -p post.json -T application/json http://127.0.0.1:8000/api/parsepost.json里写{ text: 沪太标杆Z198进石家庄北站 }压测前先小并发摸底再逐步