16.13-知识库文件流水线

要点

  • 前面 12 篇把文件处理的各个能力拆开讲了——上传、类型校验、解析、存储、异步队列，这篇把它们串成一条完整的流水线
  • 流水线按责任链模式组织，每个步骤只做一件事，可以独立测试、独立替换
  • 文件从上传到最终进入向量数据库，中间经历 6 个阶段，每个阶段都有独立的状态和错误处理
  • 增量更新策略避免重复处理——文件只改了一页，不需要把整份文档重新跑一遍
  • 这篇同时也是整个第 16 章的收束，最后会回顾这个系列讲了哪些能力、它们怎样拼在一起

1. 先看清整条流水线

知识库场景的文件处理和简单的「上传-下载」不一样。用户上传一份 PDF、一个 Word 文档、一张图片，目标不是存起来让别人下载，而是把里面的文字提取出来，切成合适大小的文本块，向量化后存入向量数据库，供后续的 RAG 检索使用。

把这条链路拆开，一共 6 个阶段：

上传 → 类型检测 → 内容解析 → 文本清洗 → 分块 → 向量化 → 存入向量数据库

每个阶段对应前面某篇文章讲过的能力：

阶段对应文章
上传16.01 文件上传接口
类型检测16.02 文件类型校验
大小限制16.03 文件大小限制
PDF / Word / Markdown / HTML 解析16.04 ~ 16.07
图片 OCR16.08 图片 OCR
音频转文本16.09 音频转文本
文件存储16.10 ~ 16.11
异步队列16.12 异步文件处理队列

这篇的任务是把上面这些独立模块组装到一起，形成一条可以从上传一直跑到向量入库的完整流水线。

2. 责任链模式组织处理步骤

如果用一个巨大的 if-else 把所有处理逻辑堆在一起，代码会很快膨胀到不可维护。更合理的做法是用责任链模式——每个处理步骤是一个独立的类，接收上一步的输出，产出自己的输出，然后交给下一步。

先定义一个贯穿整条流水线的 FileContext，每个步骤往里写东西、从里面读东西：

// src/pipeline/types.ts
export type FileContext = {
  fileId: string; knowledgeBaseId: string; originalName: string
  mimeType: string; size: number; storageKey: string
  rawText: string; cleanedText: string
  chunks: Chunk[]; vectors?: Vector[]
  status: 'pending' | 'processing' | 'completed' | 'failed' | 'retrying'
  currentStage: Stage
  error?: { stage: Stage; message: string; retryCount: number }
}
 
export type Stage =
  'upload' | 'detect' | 'parse' | 'clean' | 'chunk' | 'vectorize' | 'store' | 'done'
 
// 每个步骤实现同一个接口——接收上下文，返回处理后的上下文
export interface PipelineStep {
  readonly name: Stage
  process(ctx: FileContext): Promise<FileContext>
}

每个步骤可以独立写测试；新增或替换步骤不影响其他代码——加一种新的解析器只需要加一个类。

3. 各阶段实现

类型检测

这一步在 16.02 已经写过。包装成流水线步骤后，核心就是一个 MIME 类型白名单检查：

class DetectTypeStep implements PipelineStep {
  readonly name = 'detect' as const
  private allowedTypes = new Set(['application/pdf', /* ... */])
  async process(ctx: FileContext): Promise<FileContext> {
    if (!this.allowedTypes.has(ctx.mimeType)) {
      throw new PipelineError(this.name, `不支持的文件类型: ${ctx.mimeType}`)
    }
    return ctx
  }
}

内容解析

根据 MIME 类型分发到不同的解析器——前面 16.04 ~ 16.09 实现了各种解析器，这里做的是把它们统一管理起来：

class ParseContentStep implements PipelineStep {
  readonly name = 'parse' as const
  constructor(private bucket: R2Bucket) {}
 
  async process(ctx: FileContext): Promise<FileContext> {
    const buffer = await (await this.bucket.get(ctx.storageKey))!.arrayBuffer()
    let result: { text: string; pages?: { pageNumber: number; text: string }[] }
 
    switch (ctx.mimeType) {
      case 'application/pdf':  result = await parsePDF(buffer); break
      case 'application/vnd.openxmlformats-officedocument.wordprocessingml.document':
        result = await parseWord(buffer); break
      case 'text/markdown':    result = { text: new TextDecoder().decode(buffer) }; break
      case 'text/html':        result = await parseHTML(buffer); break
      case 'image/png':
      case 'image/jpeg':       result = await performOCR(buffer, ctx.mimeType); break
      case 'audio/mpeg':
      case 'audio/wav':        result = await transcribeAudio(buffer, ctx.mimeType); break
      default: throw new PipelineError(this.name, `无对应的解析器: ${ctx.mimeType}`)
    }
    return { ...ctx, rawText: result.text, pages: result.pages }
  }
}

这个 switch 把「哪种 MIME 类型交给哪个解析器」集中到了一个地方。后续加新格式只需要加一个 case。

文本清洗

从各种格式中提取出来的原始文本需要做一轮清洗——统一换行符、合并连续空行、去除控制字符、去掉过短的行（可能是页眉页脚）。清洗规则可以根据实际数据持续调整，实际项目中可能还需要处理 PDF 断行拼接、HTML 残留标签等问题。

分块

分块策略直接影响检索质量——块太大，检索精度下降；块太小，上下文丢失：

class ChunkTextStep implements PipelineStep {
  readonly name = 'chunk' as const
  constructor(private chunkSize = 512, private chunkOverlap = 64) {}
 
  async process(ctx: FileContext): Promise<FileContext> {
    const paragraphs = ctx.cleanedText.split(/\n\n+/)
    const chunks: Chunk[] = []
    let current = '', idx = 0
 
    for (const para of paragraphs) {
      if (this.estimateTokens(para) > this.chunkSize) {
        // 单段超长，按句子拆分
        if (current) { chunks.push(makeChunk(current, idx++)); current = '' }
        for (const sub of this.splitLongParagraph(para)) chunks.push(makeChunk(sub, idx++))
      } else if (this.estimateTokens(current + para) <= this.chunkSize) {
        current += '\n\n' + para
      } else {
        chunks.push(makeChunk(current, idx++))
        // overlap——取上一个块末尾几句，和新段落拼在一起
        current = this.getLastSentences(current, this.chunkOverlap) + '\n\n' + para
      }
    }
    if (current.trim()) chunks.push(makeChunk(current, idx++))
    return { ...ctx, chunks }
  }
 
  private estimateTokens(text: string): number {
    const cjk = (text.match(/[一-鿿]/g) || []).length
    return Math.ceil(cjk / 1.5 + (text.length - cjk) / 4)
  }
}

分块策略分三层：先按段落切分保持语义完整；段落之间往一个块里攒，攒到接近 chunkSize 就输出；单段超长时退化为按句子切分。chunkOverlap 控制相邻块之间的重叠 token 数。

向量化与入库

class VectorizeStep implements PipelineStep {
  readonly name = 'vectorize' as const
  constructor(private ai: Ai) {}
  async process(ctx: FileContext): Promise<FileContext> {
    const vectors: Vector[] = []
    for (let i = 0; i < ctx.chunks.length; i += 10) {
      const batch = ctx.chunks.slice(i, i + 10)
      const res = await this.ai.run('@cf/baai/bge-m3', { text: batch.map(c => c.text) })
      batch.forEach((c, j) => vectors.push({ chunkIndex: c.index, embedding: res.data[j] }))
    }
    return { ...ctx, vectors }
  }
}

向量生成后写入 Cloudflare Vectorize，同时把 chunk 文本存入 D1 供检索时返回上下文。Vectorize 单次最多 100 条，分批写入。metadata 里存 chunkText 的前 1000 个字符，检索时直接拿到摘要，不需要回 D1 查全文。

4. 流水线编排

有了各个步骤，接下来需要一个编排器把它们按顺序跑起来：

// src/pipeline/runner.ts
export class PipelineRunner {
  private steps: PipelineStep[] = []
 
  addStep(step: PipelineStep): this { this.steps.push(step); return this }
 
  async run(initialContext: FileContext): Promise<FileContext> {
    let ctx: FileContext = { ...initialContext, status: 'processing' }
 
    for (const step of this.steps) {
      ctx = { ...ctx, currentStage: step.name }
      await this.updateStatus(ctx)
 
      try {
        ctx = await step.process(ctx)
      } catch (error) {
        if (error instanceof PipelineError) {
          const retryCount = ctx.error?.retryCount ?? 0
 
          // 可重试的错误：网络超时、速率限制
          if (this.isRetryable(error) && retryCount < 3) {
            ctx = { ...ctx, status: 'retrying', error: {
              stage: step.name, message: error.message, retryCount: retryCount + 1
            }}
            await this.updateStatus(ctx)
            await this.sleep(1000 * Math.pow(2, retryCount))  // 指数退避
            ctx = await step.process(ctx)
            continue
          }
 
          // 不可重试或重试耗尽
          ctx = { ...ctx, status: 'failed', error: {
            stage: step.name, message: error.message, retryCount
          }}
          await this.updateStatus(ctx)
          return ctx
        }
        throw error
      }
    }
    ctx = { ...ctx, status: 'completed', currentStage: 'done' }
    await this.updateStatus(ctx)
    return ctx
  }
 
  private isRetryable(error: PipelineError): boolean {
    return /timeout|rate limit|503/i.test(error.message)
  }
  private async updateStatus(ctx: FileContext): Promise<void> {
    // 把当前阶段和状态写入 D1
  }
  private sleep(ms: number): Promise<void> {
    return new Promise(r => setTimeout(r, ms))
  }
}

编排器的核心逻辑：按顺序执行每个步骤，步骤失败时区分可重试和不可重试的错误。可重试的等待后重来，不可重试的直接标记失败。

组装和入队

在 API 路由里把各个步骤组装起来，上传文件后立即返回 202，把处理工作交给队列：

importApp.post('/api/kb/:kbId/import', async (c) => {
  const kbId = c.req.param('kbId')
  const body = await c.req.parseBody()
  const file = body['file'] as File
  if (!file) return c.json({ error: '缺少文件' }, 400)
  if (file.size > 50 * 1024 * 1024) return c.json({ error: '文件过大' }, 413)
 
  const fileId = crypto.randomUUID()
  const storageKey = `kb/${kbId}/${fileId}/${file.name}`
  await c.env.BUCKET.put(storageKey, file.stream(), {
    httpMetadata: { contentType: file.type },
  })
 
  // 记录文件信息
  await c.env.DB.prepare(`
    INSERT INTO kb_files (id, kb_id, name, mime_type, size, storage_key, status, stage)
    VALUES (?, ?, ?, ?, ?, ?, 'pending', 'upload')
  `).bind(fileId, kbId, file.name, file.type, file.size, storageKey).run()
 
  // 发送到队列异步处理
  await c.env.PROCESSING_QUEUE.send(JSON.stringify({
    fileId, knowledgeBaseId: kbId, storageKey,
    mimeType: file.type, originalName: file.name, size: file.size,
  }))
 
  return c.json({ fileId, status: 'pending', message: '文件已接收，后台处理中' }, 202)
})

队列消费者

队列消费者拿到消息后，创建流水线并执行：

// src/queues/file-processing.ts
export default {
  async fetch(batch: MessageBatch, env: AppEnv): Promise<void> {
    const runner = new PipelineRunner()
    runner
      .addStep(new DetectTypeStep())
      .addStep(new ParseContentStep(env.BUCKET))
      .addStep(new CleanTextStep())
      .addStep(new ChunkTextStep(512, 64))
      .addStep(new VectorizeStep(env.AI))
      .addStep(new StoreVectorsStep(env.DB, env.VECTOR_DB))
 
    for (const message of batch.messages) {
      const payload = JSON.parse(message.body as string)
      const ctx: FileContext = { ...payload, rawText: '', cleanedText: '', chunks: [], status: 'pending', currentStage: 'detect' }
      try {
        const result = await runner.run(ctx)
        message.ack()
      } catch (error) {
        message.retry({ delaySeconds: 30 })
      }
    }
  },
}

5. 处理状态跟踪

每个文件在处理过程中，状态会不断更新。前端需要知道「当前处理到哪一步了」。

数据库表结构

CREATE TABLE kb_files (
  id TEXT PRIMARY KEY,
  kb_id TEXT NOT NULL,
  name TEXT NOT NULL,
  mime_type TEXT NOT NULL,
  size INTEGER NOT NULL,
  storage_key TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending',
  stage TEXT NOT NULL DEFAULT 'upload',
  chunk_count INTEGER,
  error_message TEXT,
  retry_count INTEGER DEFAULT 0,
  created_at TEXT NOT NULL DEFAULT (datetime('now')),
  updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);

查询进度

根据 stage 可以估算一个大致的进度百分比：

const stageProgress: Record<Stage, number> = {
  upload: 10, detect: 15, parse: 30, clean: 50,
  chunk: 65, vectorize: 80, store: 95, done: 100,
}
 
importApp.get('/api/kb/:kbId/files/:fileId/status', async (c) => {
  const file = await c.env.DB.prepare(
    'SELECT * FROM kb_files WHERE id = ?'
  ).bind(c.req.param('fileId')).first()
 
  if (!file) return c.json({ error: '文件不存在' }, 404)
 
  return c.json({
    fileId: file.id,
    status: file.status,
    stage: file.stage,
    progress: file.status === 'failed' ? -1 : stageProgress[file.stage],
    chunkCount: file.chunk_count,
    error: file.error_message,
  })
})

progress 不是精确的进度（每个阶段耗时差异很大），但比一个不确定的转圈图标更有用。如果要做得更精确，可以在 ParseContentStep 里按页更新进度，在 VectorizeStep 里按块更新进度——代价是数据库写入次数增加。

SSE 实时推送

如果想做到实时更新，可以用 Server-Sent Events 推送状态变化：

importApp.get('/api/kb/files/:fileId/progress', async (c) => {
  const encoder = new TextEncoder()
  const stream = new ReadableStream({
    async start(controller) {
      let lastStage = ''
      for (let i = 0; i < 150; i++) {  // 最多监听 5 分钟
        const file = await c.env.DB.prepare(
          'SELECT status, stage, chunk_count, error_message FROM kb_files WHERE id = ?'
        ).bind(c.req.param('fileId')).first()
        if (!file) break
 
        if (file.stage !== lastStage) {
          controller.enqueue(encoder.encode(
            `data: ${JSON.stringify({ status: file.status, stage: file.stage })}\n\n`
          ))
          lastStage = file.stage
        }
        if (file.status === 'completed' || file.status === 'failed') break
        await new Promise(r => setTimeout(r, 2000))
      }
      controller.close()
    },
  })
  return new Response(stream, {
    headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache' },
  })
})

只在阶段变化时推送，减少无意义的数据。终态时自动结束流。

6. 增量更新

知识库里的文件不是一次性导入就不动了。用户可能更新了一份文档的某几页，把整份文档从头跑一遍流水线，既浪费算力，也浪费时间。

增量更新的核心思路是对比差异，只重新处理变化的块。

基于内容哈希的差异检测

export async function detectChanges(
  db: D1Database, fileId: string,
  newChunks: { index: number; text: string }[]
) {
  const oldChunks = await db.prepare(
    'SELECT chunk_index, content_hash FROM kb_file_chunks WHERE file_id = ?'
  ).bind(fileId).all()
 
  const oldMap = new Map(oldChunks.results.map(c => [c.chunk_index, c.content_hash]))
  const newMap = new Map(newChunks.map(c => [c.index, hashText(c.text)]))
 
  const added: typeof newChunks = []
  const modified: typeof newChunks = []
  const removed: number[] = []
  const unchanged: { oldIndex: number; newIndex: number }[] = []
 
  for (const [index, hash] of newMap) {
    const oldHash = oldMap.get(index)
    if (!oldHash) added.push(newChunks.find(c => c.index === index)!)
    else if (oldHash !== hash) modified.push(newChunks.find(c => c.index === index)!)
    else unchanged.push({ oldIndex: index, newIndex: index })
  }
  for (const [index] of oldMap) {
    if (!newMap.has(index)) removed.push(index)
  }
 
  return { added, modified, removed, unchanged }
}

增量向量更新

检测到差异后，只需要对新增和修改的块重新做向量化：

// 保留未变化的块对应的向量
for (const { oldIndex, newIndex } of changes.unchanged) {
  const existing = existingVectors.find(v => v.chunkIndex === oldIndex)
  if (existing) vectors.push({ chunkIndex: newIndex, embedding: existing.embedding })
}
// 只对新增和修改的块重新生成向量
for (const chunk of [...changes.added, ...changes.modified]) {
  const res = await this.ai.run('@cf/baai/bge-m3', { text: chunk.text })
  vectors.push({ chunkIndex: chunk.index, embedding: res.data[0] })
}
// 删除被移除的块对应的向量
for (const oldIndex of changes.removed) {
  await this.deleteVector(ctx.fileId, oldIndex)
}

这样做的好处在文件规模大时尤为明显。一份 200 页的文档只改了 3 页，重新向量化的工作量从 200 块降到 3 块。

7. 错误处理策略

流水线里任何一步都可能出错。不同类型的错误需要不同的处理方式：

错误类型示例处理策略
文件不合法不支持的类型、空文件直接标记失败，不重试
解析失败PDF 加密、Word 损坏标记失败，提示用户检查文件
外部 API 超时AI Embedding 超时指数退避重试，最多 3 次
速率限制Vectorize 写入限流等待后重试
存储满R2 配额用完标记失败，通知管理员

在 PipelineRunner 里，通过 isRetryable 方法来区分可重试和不可重试的错误。不可重试的错误直接标记失败，不会浪费 Queue 的重试配额。

对于解析失败的文件，可以考虑一种降级策略：如果 PDF 解析只失败了某几页，可以把成功的页保留，只标记失败的页。这样至少部分内容可以进入知识库，用户也可以针对性地补传缺失的页面。

8. 性能考量

流水线跑起来后，几个地方需要关注性能。

解析阶段的耗时差异 — 一份 10 页的 PDF 可能只要 2 秒，500 页的 PDF 可能需要 30 秒以上。Queue 的 maxBatchSize 和 maxConcurrency 要根据耗时来调整：

// wrangler.jsonc
[[queues.consumers]]
queue = "file-processing"
maxBatchSize = 3          // 大文件场景调小
maxConcurrency = 1        // 串行处理，避免内存峰值
maxRetries = 3
deadLetterQueue = "file-processing-dlq"

向量化的批量控制 — Embedding 接口有速率限制。50 个文件 × 100 块 = 5000 次向量调用。前面代码里已经用了每 10 段一批的方式，如果仍然触发限流，可以在批次之间加延迟：

if (i + 10 < ctx.chunks.length) {
  await new Promise(r => setTimeout(r, 200))
}

大文件的内存管理 — Workers 的内存限制是 128MB。解析完成后，原始的 ArrayBuffer 可以被 GC 回收，只需要保留提取出的文本。避免把 buffer 和解析结果同时保存在上下文里。

9. 本系列总结回顾

这是第 16 章的最后一篇。回头看一下文件处理这个章节都覆盖了哪些东西。

基础能力层

  • 16.01 ~ 16.03：文件上传接口、类型校验、大小限制。解决的是「文件怎么安全地接收进来」
  • 16.10 ~ 16.11：文件存储服务和对象存储接入。解决「文件存在哪里」

内容解析层

  • 16.04 ~ 16.07：PDF、Word、Markdown、HTML 四种文本格式的解析
  • 16.08：图片 OCR——从图片中提取文字
  • 16.09：音频转文本——把语音内容变成可检索的文字

异步与工程化层

  • 16.12：异步文件处理队列——用 Cloudflare Queue 把耗时操作从请求周期中剥离
  • 16.13（本文）：把上面所有能力串成一条知识库文件流水线

这些模块单独看都不复杂。每个解析器就是一个函数，存储就是 R2 的 put/get，队列就是 send/consume。需要花心思的是把它们组织到一起，让它们在面对不同文件格式、不同大小、不同失败场景时都能稳定运行。

从整个 Hono 后端系列来看，文件处理这一章用到的能力在前面都已经铺垫过：

  • 路由和中间件（04 ~ 05 章）——文件上传接口的入口
  • 类型安全（06 章）——请求参数的校验
  • 数据库（09 章）——D1 存储文件状态和处理记录
  • AI 能力（11 ~ 15 章）——Embedding 向量化、OCR、音频转文本
  • 队列和工作流（17 章）——异步处理的基础设施

一篇一篇分开讲，是为了每个主题能聚焦。到了实际项目里，这些能力是同时发挥作用的。这篇流水线就是把它们重新拼到一起的那张图。

总结

这篇把前面 12 篇文件处理的内容组装成了一条完整的知识库文件流水线。几个核心设计：

  • 责任链模式 — 每个处理步骤独立成类，接收上一步输出，传给下一步。新增格式或替换实现只改一个步骤，不影响整条链路
  • 分阶段状态跟踪 — 文件在数据库里有明确的状态和当前阶段，前端可以查询进度或通过 SSE 接收实时推送
  • 增量更新 — 通过内容哈希对比，只重新处理变化的块，避免全量重跑
  • 错误分级处理 — 区分可重试和不可重试的错误，可重试的用指数退避，不可重试的直接标记失败
  • 异步队列驱动 — 上传接口秒级返回，耗时的解析和向量化在后台完成

这个系列从文件上传开始，到完整流水线收尾，中间覆盖了类型校验、多种格式解析、对象存储、异步队列。单独看每一篇解决的是一个具体问题，放在一起是一套可以在知识库场景下直接复用的文件处理方案。