如何设计高吞吐的 PDF 批处理系统，加快解析和向量入库？

2026-11-04架构老王

结论

建议用队列连接各处理阶段，分别为下载、Azure DI 解析、分块、LLM 处理、embedding 和入库配置 worker 与并发上限。不同文件可以同时处于不同阶段，也可以针对实际瓶颈单独扩容。

仅凭每月 100,000 个文件、单个文件 1 至 50 MB，还无法确定最优并发数。需要补充峰值到达量、完成时限、页数分布，以及各阶段的耗时和服务配额。

先定义需要达到的速度

“处理得快”可能指单份文件尽快完成，也可能指整批文件在指定时间内完成，需要分别衡量。

按 30 天计算，100,000 个文件平均每天约 3,333 个，每小时约 139 个。如果文件集中到达，或要求在短时间内处理完，峰值吞吐需求会明显高于平均值。

可以先用以下关系估算：

目标吞吐量 R = 待处理文件数 N / 允许处理时间 T
阶段所需在途任务数 C ≈ 目标吞吐量 R × 该阶段平均耗时 L

第二个关系适合用于稳定状态下的初步估算，不能直接作为生产配置。还需要考虑延迟波动、重试和下游容量。如果某阶段处理的是 chunk，应使用 chunk 的到达速率和处理耗时来计算。

文件大小也不能代替解析工作量。应至少记录页数、是否需要 OCR、解析后的 token 数和 chunk 数。

将流程拆成独立阶段

可以采用以下结构：

文件 ID 清单
    ↓
下载队列 → 下载 worker → 对象存储
    ↓
解析队列 → Azure DI 解析 worker → Markdown / 结构化结果
    ↓
分块队列 → 分块 worker → chunks
    ↓
可选的 LLM 处理队列 → LLM worker
    ↓
embedding 队列 → embedding worker
    ↓
写入队列 → 文档存储 + 向量存储
    ↓
发布可检索的文档版本

队列消息只传递文件 ID、对象存储地址和处理版本等元数据。PDF、Markdown 和大批 chunks 放在持久化存储中，避免大文件反复经过消息队列。

每个阶段完成并持久化结果后，再进入下一阶段。后续步骤失败时，可以从已有结果继续处理，减少重复下载、解析和模型调用。

分阶段控制并发和背压

下载、解析、模型调用与存储写入的容量通常不同，应分别设置并发和限流。

下载阶段关注带宽、连接数、超时和临时存储占用。
解析阶段关注提交速率、在途任务数和结果获取方式。
LLM 与 embedding 阶段关注请求速率、token 吞吐量及输入限制。
写入阶段关注批量大小、写入延迟和索引更新负载。

Azure DI 的具体调用方式和限制，需要根据实际使用的 API、部署配置与配额确认。如果接口采用异步任务模式，应持久化任务 ID，并按接口约定获取结果，避免持续占用本地执行资源等待。

当 embedding 队列持续积压时，应限制上游继续提交任务。一味增加下载或解析并发，只会扩大待处理数据量。

检查每个 chunk 是否都需要经过 LLM

如果目标是基于文本向量进行 semantic search，可以先评估这条较短路径：

解析 → 清理文本 → 分块 → embedding → 入库

是否保留额外的 LLM 步骤，取决于它承担的任务。例如，摘要、字段抽取或为片段补充上下文可能需要模型处理；普通文本切分则可以先评估使用规则或 token 边界是否足够。

如果当前对每个 chunk 都调用 LLM，这一步可能产生大量额外请求。应使用代表性文档比较检索效果、处理时间和调用成本，再决定是否保留，不能仅凭架构推断可以删除。

分块时保留标题、页码和章节信息，并按所选模型的输入限制控制长度。表格和跨页内容应单独检查切分效果。

对 embedding 使用有界批量处理

如果所选 embedding 接口支持，可以将多个 chunk 合并到一次请求中，减少请求次数。批次必须同时满足单条输入、总 token 数、输入条数和请求体大小等限制。

下面是并发控制的 Python 示意代码。它假设批次已按实际接口限制构建，且任务与结果已有持久化机制：

import asyncio

async def embed_batches(batches, embed_batch, save_batch, concurrency):
    queue = asyncio.Queue(maxsize=concurrency * 2)

    async def worker():
        while True:
            batch = await queue.get()
            try:
                if batch is None:
                    return
                vectors = await embed_batch(batch)
                await save_batch(batch, vectors)
            finally:
                queue.task_done()

    async with asyncio.TaskGroup() as group:
        for _ in range(concurrency):
            group.create_task(worker())

        for batch in batches:
            await queue.put(batch)

        for _ in range(concurrency):
            await queue.put(None)

        await queue.join()

示例使用 Python 3.11+ 的 asyncio.TaskGroup，仅展示有界队列和固定数量的 worker。concurrency 应由实际配额和压测结果确定；生产环境还需要请求速率限制、token 速率限制、超时与失败恢复。协程数量限制不能代替服务限流。

用幂等和版本管理避免重复处理

队列可能重复投递任务，模型调用成功后也可能发生写入失败。建议为每个阶段生成稳定的幂等键，例如：

阶段任务键 = 文件 ID + 源文件版本 + 阶段名称 + 处理配置版本
chunk ID = 文档版本 + 分块配置版本 + chunk 序号
向量记录键 = chunk ID + embedding 模型版本

如果来源没有可靠的文件版本，可以评估使用内容哈希识别文件变化。不要把可变的文件 ID 当作内容永久不变的依据。

各阶段记录成功状态和产物位置，重复执行时检查已有结果；文档与向量按稳定 ID 执行幂等写入。新版本的数据全部准备就绪后，再将其标记为可检索或切换活动版本，避免搜索结果混用新旧内容。

对超时、限流和暂时性服务错误进行有限次数重试，并按接口约定处理等待时间。损坏文件、加密文件或不支持的输入应进入失败清单，避免无限重试。

根据测量结果确定扩容位置

先选一批覆盖不同大小、页数和扫描质量的文档，记录各阶段耗时、吞吐量、错误率及队列等待时间。队列持续积压的位置可以作为排查瓶颈的线索，还需结合下游限流和错误情况判断。

逐步提高该阶段的并发或批次大小，观察吞吐是否继续上升，以及限流、延迟和失败率是否恶化。完成后，再用实际检索问题验证结果质量，避免处理速度提高了，分块和文本质量却下降。