17.10-文档向量化任务

要点

文档向量化是 RAG 系统的核心任务——把上传的文档变成可检索的向量
流程：解析文档 → 切 chunk → 生成 embedding → 写入向量库
每个步骤都可以异步、分阶段、可重试
长文档需要分批处理——避免单次 API 调用超时
向量化任务需要追踪进度——前端展示「已处理 X / Y 段」

1. 文档向量化的完整流程

用户上传一个文档（PDF、Word、Markdown），系统需要把它变成可检索的向量。完整流程：

上传文档
  ↓
[1] 解析文档 → 提取纯文本
  ↓
[2] 文本清洗 → 去除噪音
  ↓
[3] 切 chunk → 按语义切分段落
  ↓
[4] 生成 embedding → 每个 chunk 转向量
  ↓
[5] 写入向量库 → 存入向量数据库
  ↓
可以检索

每个步骤对应前面章节讲过的能力：

步骤	对应文章
解析文档	16.04 PDF、16.05 Word、16.06 Markdown、16.07 HTML
文本清洗	17.06 任务状态设计
切 chunk	14.04 文本分块
生成 embedding	14.06 向量生成
写入向量库	14.09 pgvector、14.10 Milvus

这篇讲的是把这些步骤串成一个异步任务，在 Queue 消费者里执行。

2. 任务定义

interface VectorizeTask {
  taskId: string
  userId: string
  knowledgeBaseId: string
  fileKey: string        // R2 存储路径
  fileName: string
  mimeType: string
}
 
type VectorizeStage =
  | 'parse'
  | 'clean'
  | 'chunk'
  | 'embed'
  | 'index'
  | 'completed'
  | 'failed'

3. 消费者实现

// src/consumers/vectorize-consumer.ts
import type { MessageBatch } from '@cloudflare/workers-types'
 
export async function vectorizeConsumer(
  batch: MessageBatch<VectorizeTask>,
  env: Env
): Promise<void> {
  for (const msg of batch.messages) {
    const task = msg.body
 
    try {
      // 更新状态为 processing
      await updateTaskStatus(env, task.taskId, 'processing', 'parse')
 
      // 从 R2 读文件
      const object = await env.BUCKET.get(task.fileKey)
      if (!object) throw new Error('File not found in R2')
 
      const buffer = await object.arrayBuffer()
 
      // 阶段一：解析
      await updateProgress(env, task.taskId, 0, 'parse')
      const rawText = await parseDocument(buffer, task.mimeType)
 
      // 阶段二：清洗
      await updateProgress(env, task.taskId, 20, 'clean')
      const cleanedText = cleanText(rawText)
 
      // 阶段三：切 chunk
      await updateProgress(env, task.taskId, 30, 'chunk')
      const chunks = chunkText(cleanedText, { chunkSize: 512, overlap: 64 })
 
      // 阶段四：生成 embedding
      await updateProgress(env, task.taskId, 40, 'embed')
      const vectors = await generateEmbeddings(chunks, env)
 
      // 阶段五：写入向量库
      await updateProgress(env, task.taskId, 80, 'index')
      await indexVectors(task.taskId, chunks, vectors, env)
 
      // 完成
      await updateTaskStatus(env, task.taskId, 'completed', null)
      await updateProgress(env, task.taskId, 100, 'completed')
 
      msg.ack()
    } catch (err) {
      const errMsg = err instanceof Error ? err.message : String(err)
      console.error(`Vectorize task ${task.taskId} failed:`, errMsg)
 
      await updateTaskStatus(env, task.taskId, 'failed', null, errMsg)
 
      const retryCount = msg.attempts - 1
      if (retryCount < 3 && isRetryable(err)) {
        msg.retry({ delaySeconds: 30 * (retryCount + 1) })
      } else {
        msg.ack()
      }
    }
  }
}

4. 文档解析

根据 MIME 类型分发到不同的解析器：

async function parseDocument(buffer: ArrayBuffer, mimeType: string): Promise<string> {
  switch (mimeType) {
    case 'application/pdf':
      return parsePdf(buffer)
    case 'application/vnd.openxmlformats-officedocument.wordprocessingml.document':
      return parseDocx(buffer)
    case 'text/markdown':
      return new TextDecoder().decode(buffer)
    case 'text/html':
      return parseHtml(buffer)
    case 'image/png':
    case 'image/jpeg':
      return await performOCR(buffer)
    case 'audio/mpeg':
    case 'audio/wav':
      return await transcribeAudio(buffer)
    default:
      // 纯文本
      return new TextDecoder().decode(buffer)
  }
}

每个解析器的实现参考第 16 章的对应文章。

5. 文本清洗

function cleanText(text: string): string {
  return text
    // 移除连续空行
    .replace(/\n{3,}/g, '\n\n')
    // 移除行首行尾空白
    .split('\n')
    .map((line) => line.trim())
    .join('\n')
    // 修复 PDF 断行
    .replace(/(\w)-\n(\w)/g, '$1$2')
    // 移除页码
    .replace(/---+\s*\d+\s*---+/g, '')
    .replace(/\[\d+\]/g, '')
    // 移除控制字符
    .replace(/[\x00-\x08\x0B\x0C\x0E-\x1F]/g, '')
}

6. 切 chunk

interface Chunk {
  id: string
  index: number
  content: string
  metadata: {
    heading?: string
    page?: number
    startOffset: number
    endOffset: number
  }
}
 
function chunkText(
  text: string,
  opts: { chunkSize: number; overlap: number }
): Chunk[] {
  const paragraphs = text.split(/\n\n+/)
  const chunks: Chunk[] = []
  let current = ''
  let offset = 0
  let index = 0
 
  for (const para of paragraphs) {
    if (current.length + para.length > opts.chunkSize && current.length > 0) {
      // 当前 chunk 已满
      chunks.push({
        id: crypto.randomUUID(),
        index: index++,
        content: current.trim(),
        metadata: { startOffset: offset - current.length, endOffset: offset },
      })
 
      // overlap
      current = current.slice(-opts.overlap) + '\n\n' + para
    } else {
      current += (current ? '\n\n' : '') + para
    }
 
    offset += para.length + 2
  }
 
  if (current.trim()) {
    chunks.push({
      id: crypto.randomUUID(),
      index,
      content: current.trim(),
      metadata: { startOffset: offset - current.length, endOffset: offset },
    })
  }
 
  return chunks
}

7. 生成 embedding

批量生成 embedding，避免单次 API 调用超时：

async function generateEmbeddings(
  chunks: Chunk[],
  env: Env
): Promise<number[][]> {
  const BATCH_SIZE = 10  // 每次处理 10 个 chunk
  const vectors: number[][] = []
 
  for (let i = 0; i < chunks.length; i += BATCH_SIZE) {
    const batch = chunks.slice(i, i + BATCH_SIZE)
    const texts = batch.map((c) => c.content)
 
    // Workers AI
    const result = await env.AI.run('@cf/baai/bge-base-en-v1.5', {
      text: texts,
    })
 
    vectors.push(...result.data.map((d: any) => d.embedding ?? d))
 
    // 更新进度
    const progress = 40 + Math.round(((i + BATCH_SIZE) / chunks.length) * 40)
    // await updateProgress(env, taskId, progress, 'embed')
  }
 
  return vectors
}

批量调用的好处：减少 API 调用次数、降低延迟。坏处：单次失败整批重试。可以折中——每批 5-10 个，失败时只重试当前批。

8. 写入向量库

async function indexVectors(
  taskId: string,
  chunks: Chunk[],
  vectors: number[][],
  env: Env
): Promise<void> {
  const points = chunks.map((chunk, i) => ({
    id: chunk.id,
    vector: vectors[i],
    payload: {
      taskId,
      chunkId: chunk.id,
      content: chunk.content,
      heading: chunk.metadata.heading,
      page: chunk.metadata.page,
    },
  }))
 
  // Cloudflare Vectorize
  const BATCH_SIZE = 100
  for (let i = 0; i < points.length; i += BATCH_SIZE) {
    const batch = points.slice(i, i + BATCH_SIZE)
 
    await env.VECTORIZE.upsert(batch.map((p) => ({
      id: p.id,
      values: p.vector,
      metadata: p.payload,
    })))
  }
}

Vectorize 单次最多 100 条，分批写入。

9. 长文档处理

大文档（100+ 页 PDF、几万字）需要特殊处理：

分段解析

PDF 逐页解析，每解析 N 页更新一次进度：

async function parseLargePdf(buffer: ArrayBuffer, taskId: string, env: Env): Promise<string> {
  const totalPages = estimatePageCount(buffer)
  let text = ''
 
  for (let page = 0; page < totalPages; page++) {
    text += await parsePage(buffer, page)
 
    // 每 10 页更新进度
    if (page % 10 === 0) {
      const progress = Math.round((page / totalPages) * 20)  // 解析占 20%
      await updateProgress(env, taskId, progress, 'parse')
    }
  }
 
  return text
}

分批 embedding

大文档切出几百个 chunk，分批生成 embedding：

async function generateEmbeddingsWithProgress(
  chunks: Chunk[],
  taskId: string,
  env: Env
): Promise<number[][]> {
  const BATCH_SIZE = 10
  const vectors: number[][] = []
 
  for (let i = 0; i < chunks.length; i += BATCH_SIZE) {
    const batch = chunks.slice(i, i + BATCH_SIZE)
    const result = await env.AI.run('@cf/baai/bge-base-en-v1.5', {
      text: batch.map((c) => c.content),
    })
 
    vectors.push(...result.data.map((d: any) => d.embedding ?? d))
 
    // 更新进度
    const progress = 40 + Math.round(((i + BATCH_SIZE) / chunks.length) * 40)
    await updateProgress(env, taskId, progress, 'embed')
  }
 
  return vectors
}

10. 任务状态追踪

CREATE TABLE vectorize_tasks (
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  knowledge_base_id TEXT NOT NULL,
  file_key TEXT NOT NULL,
  file_name TEXT NOT NULL,
  mime_type TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending',
  stage TEXT,
  progress INTEGER NOT NULL DEFAULT 0,
  chunk_count INTEGER,
  error_message TEXT,
  created_at TEXT NOT NULL,
  updated_at TEXT NOT NULL
);

async function updateTaskStatus(
  env: Env,
  taskId: string,
  status: string,
  stage: string | null,
  errorMessage?: string
): Promise<void> {
  const updates = ['status = ?', 'updated_at = ?']
  const values: unknown[] = [status, new Date().toISOString()]
 
  if (stage !== null) {
    updates.push('stage = ?')
    values.push(stage)
  }
  if (errorMessage) {
    updates.push('error_message = ?')
    values.push(errorMessage)
  }
 
  values.push(taskId)
 
  await env.DB.prepare(`
    UPDATE tasks SET ${updates.join(', ')} WHERE id = ?
  `).bind(...values).run()
}

11. 完整流程串联

// 1. HTTP 接口接收上传
app.post('/api/knowledge-base/:kbId/documents', async (c) => {
  const kbId = c.req.param('kbId')
  const formData = await c.req.formData()
  const file = formData.get('file') as File
 
  const taskId = crypto.randomUUID()
  const fileKey = `documents/${taskId}/${file.name}`
 
  // 存 R2
  await c.env.BUCKET.put(fileKey, file.stream(), {
    httpMetadata: { contentType: file.type },
  })
 
  // 写 D1
  await c.env.DB.prepare(`
    INSERT INTO vectorize_tasks (id, user_id, knowledge_base_id, file_key, file_name, mime_type, status, created_at, updated_at)
    VALUES (?, ?, ?, ?, ?, ?, 'pending', ?, ?)
  `).bind(taskId, c.get('user').id, kbId, fileKey, file.name, file.type, new Date().toISOString(), new Date().toISOString()).run()
 
  // 入队
  await c.env.VECTORIZE_QUEUE.send({
    taskId,
    userId: c.get('user').id,
    knowledgeBaseId: kbId,
    fileKey,
    fileName: file.name,
    mimeType: file.type,
  })
 
  return c.json({ taskId, status: 'pending' }, 202)
})
 
// 2. Queue 消费者处理
export default {
  async queue(batch: MessageBatch<VectorizeTask>, env: Env) {
    await vectorizeConsumer(batch, env)
  },
}
 
// 3. 前端查询进度
app.get('/api/tasks/:taskId/progress', async (c) => {
  const task = await c.env.DB.prepare(`
    SELECT status, stage, progress, chunk_count, error_message
    FROM vectorize_tasks WHERE id = ?
  `).bind(c.req.param('taskId')).first()
 
  return c.json(task)
})

总结

回顾这一节的要点：

文档向量化是 RAG 系统的核心任务——解析 → 清洗 → 切 chunk → embedding → 写入向量库
每个步骤在 Queue 消费者里异步执行，分阶段可重试
根据 MIME 类型分发到不同解析器——PDF、Word、Markdown、HTML、OCR、音频
长文档需要分段处理——逐页解析、分批 embedding，避免超时
进度追踪：KV 存实时进度，D1 存最终状态，前端轮询或 SSE
批量 embedding 减少 API 调用，但失败时整批重试
写入向量库分批进行——Vectorize 单次最多 100 条

下一篇讲批量 AI 生成任务——批量调 LLM 生成内容。