FST ITN-ZH API开发：构建自定义文本处理服务
1. 简介与背景
中文逆文本标准化（Inverse Text Normalization, ITN）是语音识别、自然语言处理和智能对话系统中的关键预处理环节。其核心任务是将口语化或非标准的中文表达转换为结构化的标准格式，例如将“一百二十三”转为“123”，或将“早上八点半”转为“8:30a.m.”。
本文基于FST ITN-ZH开源项目，介绍如何通过二次开发构建一个可扩展的自定义文本处理服务API。该项目由开发者“科哥”完成WebUI封装，并支持本地部署与批量处理功能。我们将在此基础上进行API化改造，使其不仅限于图形界面使用，还能集成到后端服务中，满足自动化、高并发的生产需求。
本技术方案适用于： - 智能客服系统的语义前处理 - 语音识别结果的后处理模块 - 大规模日志数据清洗 - 自动化文档结构化提取
2. 系统架构与运行机制
2.1 整体架构设计
该系统采用分层架构模式，主要包括以下组件：
+------------------+ | WebUI / API | +--------+---------+ | v +--------+---------+ | 控制逻辑层 | ← 配置解析、参数控制、流程调度 +--------+---------+ | v +--------+---------+ | FST 转换引擎 | ← 基于有限状态机的核心ITN模型 +--------+---------+ | v +--------+---------+ | 输入输出管理 | ← 文件读写、缓存、日志记录 +------------------+其中，FST（Finite State Transducer）作为底层转换引擎，利用加权有限状态自动机实现高效、准确的规则匹配与转换。
2.2 启动流程分析
原始启动命令如下：
/bin/bash /root/run.sh该脚本通常包含以下操作： 1. 激活Python虚拟环境 2. 安装依赖包（首次运行） 3. 加载FST模型文件 4. 启动Gradio Web服务，默认监听7860端口
我们可通过修改此脚本，分离出独立的服务入口，便于后续API封装。
3. API化改造实践
3.1 技术选型与框架选择
为了实现高性能、易集成的API服务，我们选择FastAPI作为主框架，原因如下： - 支持异步处理，提升I/O效率 - 自动生成OpenAPI文档（Swagger UI） - 类型提示驱动，减少接口错误 - 与Pydantic无缝集成，便于数据校验
同时保留原有FST引擎不变，确保转换逻辑一致性。
3.2 核心代码实现
主要依赖安装
pip install fastapi uvicorn python-multipartAPI服务主程序
# api_server.py from fastapi import FastAPI, File, UploadFile, Form from fastapi.responses import JSONResponse import subprocess import json import os import time app = FastAPI(title="FST ITN-ZH API", version="1.0") # 模拟调用原生ITN处理函数（实际应加载FST模块） def itn_transform(text: str, config: dict) -> str: # 此处应替换为真实FST调用逻辑 result = subprocess.run( ["python", "-c", f""" import sys; sys.path.append('/root/itn-zh'); from itn import inverse_text_normalization; print(inverse_text_normalization('{text}', {config})) """], capture_output=True, text=True ) return result.stdout.strip() @app.post("/api/v1/convert") async def convert_text( text: str = Form(...), convert_digits: bool = Form(True), convert_single_digit: bool = Form(False), full_convert_wan: bool = Form(False) ): """ 单条文本转换接口 """ config = { "convert_digits": convert_digits, "convert_single_digit": convert_single_digit, "full_convert_wan": full_convert_wan } try: output = itn_transform(text, config) return JSONResponse({ "success": True, "input": text, "output": output, "timestamp": int(time.time()) }) except Exception as e: return JSONResponse({ "success": False, "error": str(e) }, status_code=500) @app.post("/api/v1/batch") async def batch_convert(file: UploadFile = File(...)): """ 批量文件转换接口 """ contents = await file.read() lines = contents.decode('utf-8').splitlines() results = [] for line in lines: if line.strip(): output = itn_transform(line.strip(), {}) results.append({"input": line.strip(), "output": output}) return JSONResponse({ "success": True, "total": len(results), "results": results })启动API服务
新增启动脚本run_api.sh：
#!/bin/bash cd /root/itn-zh source venv/bin/activate uvicorn api_server:app --host 0.0.0.0 --port 8000 --reload此时可通过http://<ip>:8000/docs访问自动生成的API文档界面。
4. 高级配置与参数控制
4.1 可配置项映射
原WebUI中的高级设置需在API中以参数形式暴露：
| 参数名 | 类型 | 默认值 | 说明 |
|---|---|---|---|
convert_digits | boolean | true | 是否转换独立数字（如“幸运一百”→“幸运100”） |
convert_single_digit | boolean | false | 是否转换单个数字（如“零和九”→“0和9”） |
full_convert_wan | boolean | false | 是否完全展开“万”单位（如“六百万”→“6000000”） |
这些参数直接影响转换粒度，可根据业务场景灵活调整。
4.2 缓存优化建议
对于高频重复输入（如常见时间表达），建议引入Redis缓存层：
import redis r = redis.Redis(host='localhost', port=6379, db=0) def cached_itn_transform(text, config): key = f"itn:{text}:{hash(str(config))}" cached = r.get(key) if cached: return cached.decode('utf-8') result = itn_transform(text, config) r.setex(key, 3600, result) # 缓存1小时 return result可显著降低模型重复计算开销。
5. 批量处理与性能优化
5.1 异步任务队列设计
当面对大规模批量请求时，同步处理可能导致超时。推荐引入Celery + RabbitMQ/Redis构建异步任务系统：
from celery import Celery celery_app = Celery('itn_tasks', broker='redis://localhost:6379/0') @celery_app.task def async_batch_process(file_path: str): with open(file_path, 'r') as f: lines = f.readlines() results = [itn_transform(line.strip(), {}) for line in lines] output_path = f"/data/output/{int(time.time())}.json" with open(output_path, 'w') as f: json.dump(results, f, ensure_ascii=False, indent=2) return {"status": "completed", "output_file": output_path}前端提交任务后返回任务ID，客户端轮询获取结果。
5.2 性能基准测试建议
建议在正式上线前进行压力测试，使用locust工具模拟并发请求：
# locustfile.py from locust import HttpUser, task, between class ITNUser(HttpUser): wait_time = between(1, 3) @task def convert_common_text(self): self.client.post("/api/v1/convert", data={ "text": "二零二四年三月十五日下午四点二十分" })目标：在100并发下平均响应时间 < 200ms。
6. 安全性与部署建议
6.1 接口安全加固
- 添加API Key认证机制
- 限制请求频率（如每分钟最多100次）
- 对上传文件做大小限制（如≤10MB）
- 过滤恶意输入（防止命令注入）
示例认证中间件：
from fastapi.security import APIKeyHeader api_key_header = APIKeyHeader(name="X-API-Key") @app.middleware("http") async def validate_api_key(request, call_next): try: api_key = await api_key_header(request) if api_key != "your-secret-key": return JSONResponse({"success": False, "error": "Invalid API Key"}, 401) except: return JSONResponse({"success": False, "error": "Missing API Key"}, 401) return await call_next(request)6.2 Docker容器化部署
创建Dockerfile实现一键部署：
FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . EXPOSE 8000 CMD ["uvicorn", "api_server:app", "--host", "0.0.0.0", "--port", "8000"]配合docker-compose.yml统一管理服务栈。
7. 总结
本文围绕FST ITN-ZH中文逆文本标准化系统，完成了从WebUI工具到可编程API服务的技术升级。主要内容包括：
- 架构解析：梳理了原始系统的运行逻辑与组件关系；
- API封装：基于FastAPI实现了RESTful接口，支持单条与批量转换；
- 参数控制：将WebUI中的高级设置映射为可编程接口参数；
- 性能优化：提出缓存、异步任务、并发测试等工程化改进方案；
- 安全部署：提供认证、限流、容器化等生产级部署建议。
通过本次改造，FST ITN-ZH不再局限于本地交互式使用，而是成为一个可嵌入各类NLP流水线的标准化服务模块，极大提升了其在企业级应用中的可用性与灵活性。
未来可进一步拓展方向包括： - 支持多语言ITN统一网关 - 提供WebSocket长连接实时转换 - 集成模型热更新机制
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
