如何设置并发推理？

[Aaaattack]

我看代码在使用vlm作为backend时，doc_analyze的调用似乎是没有并发的，如果我希望在单卡上并发推理，有没有什么参数可以设置，或者代码应该怎么修改呢

[dosubot[bot]]

MinerU的VLM backend在单卡上实现并发推理有几种方式：

1. 使用异步Backend (推荐)

切换到 vllm-async-engine 或 lmdeploy-engine，并使用异步版本的 aio_doc_analyze：

from mineru.backend.vlm.vlm_analyze import aio_doc_analyze
import asyncio

async def process_multiple_pdfs(pdf_bytes_list):
    tasks = [
        aio_doc_analyze(
            pdf_bytes=pdf_bytes,
            backend="vllm-async-engine",  # 或 "lmdeploy-engine"
            model_path=model_path
        )
        for pdf_bytes in pdf_bytes_list
    ]
    return await asyncio.gather(*tasks)

results = asyncio.run(process_multiple_pdfs([pdf1, pdf2, pdf3]))

2. 调整Batch Size

对于 transformers backend，batch_size会根据GPU内存自动配置（≥16GB为8，≥8GB为4），也可以手动指定：

result = doc_analyze(
    pdf_bytes=pdf_bytes,
    backend="transformers",
    batch_size=8
)

3. HTTP Client并发

如果使用 http-client backend，可以通过 max_concurrency 参数（默认100）控制并发数。

注意: 当前doc_analyze对单个PDF的处理是将所有页面打包成batch一次性推理，没有实现页面级别的并发。如果需要页面级并发，需要修改底层实现。

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}

   

[dosubot[bot]]

基于之前的讨论，使用 vLLM 作为 backend 异步调用有两种方式：

方式1：使用 vllm-async-engine backend

from mineru.backend.vlm.vlm_analyze import aio_doc_analyze
import asyncio

async def process_pdf(pdf_bytes):
    result = await aio_doc_analyze(
        pdf_bytes=pdf_bytes,
        backend="vllm-async-engine",
        model_path="your_model_path"
    )
    return result

# 多文档并发处理
async def process_multiple(pdf_list):
    tasks = [aio_doc_analyze(pdf_bytes=pdf, backend="vllm-async-engine", model_path="your_model_path") 
             for pdf in pdf_list]
    return await asyncio.gather(*tasks)

results = asyncio.run(process_multiple([pdf1, pdf2, pdf3]))

方式2：使用 http-client 架构

如果更偏向客户端-服务器分离部署：

# 服务端
mineru-openai-server --port 30000

# 客户端可以并发调用
mineru -p input.pdf -o output -b vlm-http-client -u http://127.0.0.1:30000

http-client 模式下 max_concurrency 默认为 100，支持高并发请求。

aio_doc_analyze 是 doc_analyze 的异步版本，内部使用 await predictor.aio_batch_two_step_extract() 实现异步推理。

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}

   

[Aaaattack]

在bash脚本中应该怎么通过mineru命令来使用vllm-async-engine作为backend

[dosubot[bot]]

vllm-async-engine 不能直接通过 mineru CLI 的 -b 参数指定。CLI 只支持以下 backend：pipeline、vlm-http-client、hybrid-http-client、vlm-auto-engine、hybrid-auto-engine。

代码中明确拒绝在同步模式下使用 vllm-async-engine。

推荐方案：使用 http-client 架构

# 终端1：启动服务端
mineru-openai-server --port 30000

# 终端2：客户端调用（支持高并发）
mineru -p input.pdf -o output -b vlm-http-client -u http://127.0.0.1:30000

如果想在单机使用本地推理，可以用 vlm-auto-engine，它会自动选择合适的引擎：

mineru -p input.pdf -o output -b vlm-auto-engine

异步特性主要在 Python API 层面通过 aio_doc_analyze 函数使用，CLI 命令本身是同步执行的。

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}

   

[Aaaattack]

如果我将_process_vlm函数里的for循环中的串行推理改成多线程并发推理（使用http-vllm-client作为backend），会有并发安全吗

[dosubot[bot]]

不推荐直接改成多线程并发，存在几个并发安全问题：

1. ModelSingleton 不是线程安全的：_models 字典作为全局缓存，没有使用 Lock 保护，多线程同时访问会有竞态条件

2. pypdfium2 不支持多线程：这个底层 PDF 解析库在多线程环境下会触发 weakref.finalize 回调异常，导致段错误

3. 官方明确不支持进程内多线程并发：会导致数据混乱或崩溃

推荐方案：

• 多进程：用 multiprocessing.Pool 替代多线程，每个进程独立处理一个 PDF，建议并发数 2-3 个
• 分布式架构：部署 mineru-openai-server，然后启动多个客户端进程并发请求

虽然 _process_vlm 里每个循环的文件写入器是独立的，但上述底层依赖的线程安全问题无法规避。

To reply, just mention @dosu.

---

^{How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other}

   

[xllrun]

使用 http-client 架构，我在4090上运行mineru-openai-server --port 30000，2个并发比1个并发快10s，3个并发直接卡了，是怎么做到支持100高并发请求的呢

[myhloli]

使用http client连接openai server时，单个文档会有数十个http并发请求，多个文档并发请求时，http请求数量会远超过vllm server的承受上限导致卡顿。3.0版本后推荐基于fastapi实现文件级别并发而不是http请求级别的并发。

[xllrun]

我的步骤是启动了vlm-http-client模型： mineru-openai-server --model 模型路径 --port 30000 ；然后启动了mineru-api ,理应我是通过基于fastapi实现文件级别并发而不是http请求级别的并发的呀，使用的测试脚本是：
import asyncio
import aiohttp
from pathlib import Path

========== 配置 ==========

API_URL = "http://127.0.0.1:8000/file_parse"

待并发解析的文件列表

FILE_LIST = [
r"/mnt/d/xll/02_programme/mineru03/data/测试1.pdf",
r"/mnt/d/xll/02_programme/mineru03/data/测试2.pdf",
r"/mnt/d/xll/02_programme/mineru03/data/测试3.pdff",

# 追加更多文件

]

最大并发数（和服务端配置对齐：MAX_CONCURRENT_REQUESTS）

MAX_CONCURRENT = 4
TIMEOUT = aiohttp.ClientTimeout(total=600) # 10分钟超时

=========================

固定请求表单参数

FORM_DATA = {
"backend": "vlm-http-client",
"server_url":"http://0.0.0.0:30000",
"return_md": "true",
"return_middle_json": "false",
"return_model_output": "false",
"return_content_list": "false",
"return_images": "false",
"response_format_zip": "false",
"return_original_file": "false",
"start_page_id": 0,
"end_page_id": 99999
}

限制并发信号量

sem = asyncio.Semaphore(MAX_CONCURRENT)

async def parse_single_file(session: aiohttp.ClientSession, file_path: str):
async with sem:
file_name = Path(file_path).name
print(f"开始解析: {file_name}")
try:
data = aiohttp.FormData()
# 添加表单固定字段
for k, v in FORM_DATA.items():
data.add_field(k, str(v))
# 上传文件
data.add_field("files", open(file_path, "rb"), filename=file_name)
print(data)
async with session.post(API_URL, data=data, timeout=TIMEOUT) as resp:
if resp.status == 200:
res = await resp.json()
print(f"✅ {file_name} 解析完成 | TaskID: {res['task_id']}")
return {"file": file_name, "success": True, "data": res}
else:
text = await resp.text()
print(f"❌ {file_name} 失败 | 状态码:{resp.status} | {text}")
return {"file": file_name, "success": False, "error": text}
except Exception as e:
print(f"❌ {file_name} 异常: {str(e)}")
return {"file": file_name, "success": False, "error": str(e)}

async def main():
async with aiohttp.ClientSession() as session:
tasks = [parse_single_file(session, fp) for fp in FILE_LIST if Path(fp).exists()]
print(tasks)
results = await asyncio.gather(*tasks)
print("\n===== 全部并发任务结束 =====")

if name == "main":
import time
start = time.time()
asyncio.run(main())
end = time.time()
print(end -start)

[xllrun]

想请教下正确的使用方法