异步 API 与任务编排：FastAPI 服务的异步任务协议

异步 API 与任务编排：FastAPI 服务的异步任务协议

学习目标

读完本章后，你将能够：

• 理解 MinerU 的 FastAPI 服务架构
• 掌握异步任务协议（提交→状态查询→结果获取）
• 设计基于 MinerU API 的客户端
• 处理并发请求与任务队列

前置知识

• API 网关设计

---

1. API 服务架构

MinerU 提供两个 API 入口：

两个入口的区别：

• mineru-api：单实例 FastAPI 服务，适合开发和小规模部署
• mineru-router：多实例负载均衡，适合生产环境多 GPU 部署

2. 异步任务协议 v2

MinerU API 采用异步任务协议，协议版本为 2：

mineru/cli/api_protocol.py

API_PROTOCOL_VERSION = 2
DEFAULT_MAX_CONCURRENT_REQUESTS = 3
DEFAULT_PROCESSING_WINDOW_SIZE = 64

2.1 任务状态机

2.2 三个核心接口

# 1. 提交任务
POST /tasks
Content-Type: multipart/form-data
Body: {
    "file": <文件>,
    "options": {解析参数}
}
Response: {"task_id": "uuid", "status": "pending"}

# 2. 查询状态
GET /tasks/{task_id}
Response: {"status": "processing", "queued_ahead": 2}

# 3. 获取结果
GET /tasks/{task_id}/result
Response: {"status": "completed", "result": {...}}

# 兼容老版本的同步接口
POST /file_parse
Response: {"status": "completed", "result": {...}}  # 阻塞等待

3. 源码实现分析

3.1 FastAPI 服务启动

mineru/cli/fast_api.py

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：预加载模型
    await maybe_preload_vlm_model(...)
    yield
    # 关闭时：清理资源
    await shutdown_cached_models()
    shutdown_pdf_render_executor()

app = FastAPI(lifespan=lifespan)
app.add_middleware(GZipMiddleware)

@app.post("/tasks")
async def submit_task(
    file: UploadFile,
    background_tasks: BackgroundTasks,
    request: Request,
):
    task_id = str(uuid.uuid4())
    task_state[task_id] = {"status": TASK_PENDING, ...}
    background_tasks.add_task(process_task, task_id, file)
    return {"task_id": task_id, "status": TASK_PENDING}

@app.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    state = task_state.get(task_id)
    if not state:
        raise HTTPException(404, "Task not found")
    return {"status": state["status"], ...}

@app.get("/tasks/{task_id}/result")
async def get_task_result(task_id: str):
    state = task_state.get(task_id)
    if state["status"] != TASK_COMPLETED:
        raise HTTPException(400, "Task not completed")
    return {"status": TASK_COMPLETED, "result": state["result"]}

3.2 任务处理流程

async def process_task(task_id: str, file: UploadFile):
    # 1. 更新状态为 processing
    task_state[task_id]["status"] = TASK_PROCESSING

    try:
        # 2. 读取文件
        file_bytes = await file.read()

        # 3. 调用解析
        result = await aio_do_parse(
            file_bytes=file_bytes,
            # ... 解析参数
        )

        # 4. 存储结果
        task_state[task_id]["status"] = TASK_COMPLETED
        task_state[task_id]["result"] = result

    except Exception as e:
        task_state[task_id]["status"] = TASK_FAILED
        task_state[task_id]["error"] = str(e)

    finally:
        # 5. 清理临时文件
        ...

3.3 并发控制

# 最大并发请求数
max_concurrent = read_max_concurrent_requests()  # 默认 3

# 处理窗口大小（滑动窗口）
window_size = get_processing_window_size()  # 默认 64 页

4. 客户端实现

MinerU CLI 本身也是一个 API 客户端，伪代码：

class LocalAPIServer:
    def __init__(self, api_url: str):
        self.api_url = api_url

    async def submit(self, file_path: Path) -> SubmitResponse:
        async with httpx.AsyncClient() as client:
            with open(file_path, "rb") as f:
                resp = await client.post(
                    f"{self.api_url}/tasks",
                    files={"file": f},
                )
            return SubmitResponse(**resp.json())

    async def poll_status(self, task_id: str) -> str:
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"{self.api_url}/tasks/{task_id}")
            return resp.json()["status"]

    async def get_result(self, task_id: str):
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"{self.api_url}/tasks/{task_id}/result")
            return resp.json()["result"]

5. 陷阱与对策

陷阱	对策
任务提交后未轮询状态	使用异步轮询或 WebSocket（未来支持）
并发数过高导致 OOM	调整 MAX_CONCURRENT_REQUESTS 配置
大文件上传超时	增加 Uvicorn 的 timeout 配置
任务结果占用过多内存	及时获取结果后清理服务端状态
公共 IP 绑定安全风险	使用 warn_if_public_http_client_policy 检查

6. 设计权衡

同步 vs 异步 API

方案	优势	劣势
同步（/file_parse）	简单，适合短任务	长任务会超时
异步（/tasks）	支持长任务，可查询进度	客户端实现复杂

取舍：MinerU 同时提供两种接口，/file_parse 兼容老版本插件，/tasks 是新协议推荐方式。

---

参考来源

• MinerU 源码：mineru/cli/fast_api.py、mineru/cli/api_client.py、mineru/cli/api_protocol.py
• FastAPI 文档：https://fastapi.tiangolo.com/
• Uvicorn 文档：https://www.uvicorn.org/