OpenMed 异步 Python API 实战指南：在 FastAPI 与批量场景中安全使用 asyncio 包装器

OpenMed 异步 Python API 实战指南：在 FastAPI 与批量场景中安全使用 asyncio 包装器

【免费下载链接】openmedLocal-first healthcare AI: clinical NER & HIPAA PII de-identification that runs 100% on-device. 2,200+ medical models, 21 languages, Apple MLX + Python, no cloud, no patient data leaving your network. Apache-2.0项目地址: https://gitcode.com/GitHub_Trending/ope/openmed

OpenMed 为阻塞式 Python API 提供了一等公民的协程包装器（aextract_pii、adeidentify、aanalyze_text、abatch），它们在 asyncio 的 worker 线程池中运行既有同步实现，从而让应用事件循环在模型推理或批量处理期间始终保持响应。本文以 docs/async-api.md 为骨架，结合 openmed/aio.py、openmed/init.py 与 tests/unit/test_async_api.py 的源码实现，系统讲解这些包装器的设计原理、FastAPI 集成方式、批量并发控制与优雅关闭策略，帮助你写出现场可用的异步去标识化服务。

一、异步 API 的设计定位与核心机制

OpenMed 的异步封装遵循一个简单而明确的原则：不重新实现任何推理逻辑，只负责把阻塞调用安全地移出事件循环线程。同步版本的全部能力（PII 实体提取、多种去标识化方法、临床文本分析、批量处理）都被原样保留，异步层只提供await语法糖。

1.1 包装器与同步实现的对应关系

四个顶层懒加载导出与同步实现的映射如下（见 docs/async-api.md 与 openmed/init.py 中的_LAZY_IMPORTS）：

默认模型与默认阈值可直接在 openmed/aio.py 与各包装器签名中核实；extract_pii、deidentify的完整参数契约见 openmed/core/pii.py 与 openmed/core/pii.py。

1.2 每个包装器接受相同参数、返回相同结果

这是异步 API 最核心的契约：参数签名与同步函数完全一致，返回值类型也完全一致，异常原样传播。测试 tests/unit/test_async_api.py 用inspect.signature逐一断言了三个包装器与同步 API 的签名相等，并验证了aextract_pii返回的就是同步的PredictionResult实例：

def test_async_wrappers_match_sync_signatures(): assert inspect.signature(openmed.aextract_pii) == inspect.signature(openmed.extract_pii) assert inspect.signature(openmed.adeidentify) == inspect.signature(openmed.deidentify) assert inspect.signature(openmed.aanalyze_text) == inspect.signature(openmed.analyze_text)

这意味着你可以在同步与异步调用之间自由切换：迁移代码时无需调整任何参数、默认值或结果字段，只需在调用前加上await。

1.3 懒加载：导入 openmed 不会触碰 asyncio

一个值得注意的工程细节是懒加载。import openmed本身既不会导入asyncio，也不会导入openmed.aio模块，更不会创建事件循环——只有当你首次访问某个a*辅助函数时，.aio模块才会被加载（见 docs/async-api.md 与 openmed/init.py 的__getattr__实现）。测试 tests/unit/test_async_api.py 在一个独立子进程中验证了这一点：

probe = ( "import sys; import openmed; " "assert 'asyncio' not in sys.modules; " "assert 'openmed.aio' not in sys.modules" ) subprocess.run([sys.executable, "-c", probe], check=True)

因此，纯同步的程序不会因为import openmed而背上任何 asyncio 开销；只有真正使用异步 API 时才付出这一成本。

二、快速上手：第一个 await 调用

最简单的用法是直接把同步调用替换为带await的异步版本：

import openmed result = await openmed.adeidentify( "Synthetic patient Casey Example called 555-0100.", method="mask", ) print(result.deidentified_text)

该调用在 worker 线程中执行完整的 PII 检测与掩码化流程，返回的DeidentificationResult与同步deidentify完全一致。由于是协程，运行环境需要 iscoroutine 上下文（如asyncio.run、FastAPI 路由或 Jupyter 的await）。

关于method参数的取值，可在 openmed/core/pii.py 查看到完整的DeidentificationMethod字面量类型：

DeidentificationMethod = Literal[ "mask", # 默认：用占位符掩码（如 [NAME]） "aadhaar_mask", # Aadhaar 号专用掩码 "remove", # 直接移除实体 "replace", # 用指定文本替换 "hash", # 哈希化 "shift_dates", # 日期偏移（需配置偏移参数） "format_preserve", # 保留格式的替换 ]

异步包装器adeidentify的参数默认值（method="mask"、confidence_threshold=0.7、use_safety_sweep=True）与同步deidentify保持一致，因此可以直接await openmed.adeidentify(text)获得与同步一致的行为。

三、FastAPI 集成：不阻塞服务器事件循环

当推理保持本地（设备端/内网）运行、且调用方希望避免阻塞服务器事件循环时，这些包装器非常适合放在异步路由里（见 docs/async-api.md 的 FastAPI 示例）：

from fastapi import FastAPI from pydantic import BaseModel import openmed app = FastAPI() class RedactionRequest(BaseModel): text: str @app.post("/redact") async def redact(request: RedactionRequest) -> dict[str, str]: result = await openmed.adeidentify( request.text, method="mask", use_safety_sweep=True, ) return {"text": result.deidentified_text}

在把上面的路由投入生产之前，文档明确给出了三条硬性纪律：

1. 绝不记录敏感内容：不要记录请求文本、模型输出，或包含源值的异常。异常与日志是 PHI 泄漏的高发通道；这一点与仓库整体的 no-PHI 日志约束一脉相承（相关文档见 docs/operations/no-phi-telemetry.md）。
2. 复用预热好的 loader：在持续流量下，应为每个进程复用同一个ModelLoader（通过loader=参数传入），避免反复加载模型权重。
3. 施加应用级并发上限：为请求设置并发限制，防止无界请求造成 worker 压力（具体手段见下文「批量并发」与「取消与关闭」两节）。

四、批量并发：abatch 的顺序保持与并发上限

对多个相互独立的输入，abatch是推荐的并发工具。它有两个核心保证：按输入顺序返回结果、可选并发上限。

4.1 基本用法

from openmed import abatch, aextract_pii results = await abatch(aextract_pii, ["Synthetic note one", "Synthetic note two"])

results中的每一项与输入一一对应，顺序不变。abatch的operation参数既可以是异步包装器（如aextract_pii），也可以是任意同步可调用对象——同步操作会被自动通过asyncio.to_thread调度到 worker 线程（见 openmed/aio.py），因此这个辅助函数同样适用于原有同步 API 的批量并发。

4.2 设置并发上限 max_concurrency

当模型会话较大或输入集合很大时，建议传入max_concurrency对同时调度的操作数量施加硬性边界：

results = await abatch( aextract_pii, ["Synthetic note one", "Synthetic note two"], max_concurrency=2, )

从实现看（openmed/aio.py），max_concurrency的取值必须是正整数（bool也会被拒绝），否则抛出ValueError("max_concurrency must be positive")；当限制值大于等于输入数量时，退化为一次性并发。真正有界时，abatch使用固定数量的 worker 协程轮流取任务（next_index索引游标），而不是为每个输入创建独立事件循环任务——测试 tests/unit/test_async_api.py 验证了 50 个输入在max_concurrency=3下，峰值事件循环任务数不超过 4。

4.3 输入物化与敏感值保护

abatch有两个隐藏的行为细节：

• 输入迭代在事件循环线程之外完成：values = await asyncio.to_thread(tuple, items)会把惰性迭代器在 worker 线程中物化成元组，避免生成器/迭代器在事件循环线程中产生阻塞（见 openmed/aio.py）。测试 tests/unit/test_async_api.py 断言迭代器确实运行在非调用线程。
• 迭代失败不会泄露敏感值：如果输入迭代过程抛异常，abatch统一包装为ValueError("items could not be read")，原始异常内容（可能包含敏感文本）不会被透传。测试 tests/unit/test_async_api.py 用含敏感标记的RuntimeError验证了这一点。

这两点对医疗文本尤其重要：任何异常路径都不应回显患者数据。

五、取消与优雅关闭：协作式预算的正确姿势

5.1 取消的边界

asyncio的Task.cancel()语义在 OpenMed 的异步包装器面前有一个明确的边界：取消正在等待的任务只会停止等待结果，却无法强制停止已经在 worker 线程中运行的同步函数（见 docs/async-api.md 的「Cancellation and shutdown」一节）。这是因为 Python 线程无法被外部强制中断——被取消后，worker 线程中的推理仍会跑完，只是其结果不再被接收。

因此文档给出的策略是：不要依赖取消来做资源回收，而要用有界的工作量 + 优雅关闭。

5.2 用 RequestBudget 给请求上「保险丝」

OpenMed 提供了RequestBudget（见 openmed/core/budget.py）作为每请求的协作式资源预算，它是「有界工作」的标准实现。两个独立维度：

• max_wall_time（秒）：墙钟时间上限，用time.perf_counter测量；
• max_input_chars（字符数）：输入长度上限，在进入模型推理前就拒绝超长输入。

预算的检查是协作式的：BudgetClock.check()只在安全的检查点（如 pipeline 阶段之间、batch 项之间）被调用，超限时干净地抛出BudgetExceededError——不杀线程、不破坏部分状态。extract_pii内部在入口即调用budget.check_input_length(len(text), checkpoint="extract_pii.input_guard")（见 openmed/core/pii.py），超长输入在推理前就被拦截。

隐私方面，预算对象与BudgetExceededError从不捕获原始输入文本或 PHI——错误只携带计数、限额与检查点名称（见 openmed/core/budget.py 的模块文档）。coerce_budget同时接受RequestBudget实例、包含max_wall_time/max_input_chars键的映射或None。

在异步包装器中传入预算，即可实现对长任务的软性超时控制：

import asyncio import openmed from openmed.core.budget import RequestBudget async def redact_with_budget(text: str) -> str: budget = RequestBudget(max_wall_time=30.0, max_input_chars=200_000) result = await openmed.adeidentify(text, method="mask", budget=budget) return result.deidentified_text

5.3 优雅关闭：让在途调用完成

关闭进程时，不要依赖asyncio.run()的取消语义去「掐断」推理。正确做法是：

• 停止接收新请求；
• 等待已提交的在途 worker 调用自然完成（它们是有界的）；
• 待所有调用返回（或达到预算上限被BudgetExceededError终止）后再退出解释器。

由于每个请求都被RequestBudget或应用级并发限制约束为有界工作，整个关闭过程的时间上限是可控的。

六、源码级原理：_run_sync 与懒加载解析

最后，用两张源码地图收束全文，方便你继续深入阅读。

异步调度的核心链路（openmed/aio.py）：

def _resolve_sync_export(name: str) -> Callable[..., Any]: import openmed return getattr(openmed, name) def _call_sync(name: str, args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any: return _resolve_sync_export(name)(*args, **kwargs) async def _run_sync(name: str, *args: Any, **kwargs: Any) -> Any: return await asyncio.to_thread(_call_sync, name, args, kwargs)

可以看到：同步导出是在worker 线程内才被解析的（_call_sync内部调用getattr(openmed, name)），这保证懒加载解析本身也不会阻塞事件循环线程。测试 tests/unit/test_async_api.py 专门断言了解析动作运行在非调用线程。

懒加载导出的注册表（openmed/init.py）：

_LAZY_IMPORTS = { "aanalyze_text": ".aio", "abatch": ".aio", "adeidentify": ".aio", "aextract_pii": ".aio", ... }

四个异步入口全部指向.aio模块，并在 openmed/init.py 的__all__中作为公开导出；aio模块自身的__all__（openmed/aio.py）也正是这四个函数。后续如需确认某个参数的行为，直接对照同步实现（openmed/core/pii.py 中的extract_pii/deidentify，openmed/init.py 中的analyze_text）即可——异步层不会引入任何参数语义差异。

七、小结与使用清单

OpenMed 的异步 API 是对同步实现的一次「零语义损耗」封装：同样的参数、同样的返回类型、同样的异常；不同之处只在于阻塞调用被调度到了 asyncio worker 线程池。落地使用时请遵守以下清单：

1. 使用await openmed.adeidentify(...)/aextract_pii(...)/aanalyze_text(...)替代同步调用，签名无需改动；
2. 在 FastAPI 异步路由中直接await，并通过loader=复用预热模型、通过应用级信号量/限流控制并发；
3. 批量场景用abatch(operation, items, max_concurrency=N)，结果顺序有保证，max_concurrency必须是正整数；
4. 永远不要记录请求文本、模型输出或含源值的异常；
5. 取消只停等待、不停线程——用RequestBudget(max_wall_time=..., max_input_chars=...)让每次推理有界，并在关闭进程时等待在途 worker 调用完成；
6. 涉及去标识化方法选择时，参考DeidentificationMethod的 7 种取值（mask/aadhaar_mask/remove/replace/hash/shift_dates/format_preserve）按合规需求选用。

这些辅助函数只负责卸载阻塞工作，不替代临床决策——请始终与同步 API 使用相同的本地模型与隐私配置。

【免费下载链接】openmedLocal-first healthcare AI: clinical NER & HIPAA PII de-identification that runs 100% on-device. 2,200+ medical models, 21 languages, Apple MLX + Python, no cloud, no patient data leaving your network. Apache-2.0项目地址: https://gitcode.com/GitHub_Trending/ope/openmed