14.03-文档上传流程

要点

RAG 索引链路的第一步：把文档从用户手里接过来，送进处理管道
文档上传不只是「接收文件」——它涉及格式校验、大小限制、异步处理、状态追踪
大文档不能同步处理——上传、解析、切分、向量化是一条异步管道
状态管理很重要：pending → parsing → chunking → embedding → indexed → failed
失败重试和断点续传：大文档处理中断后不需要从头来
权限和配额：谁能上传、能上传多少、单文件大小限制

1. 从最朴素的实现开始

先看最简单的文档上传：

// ❌ 同步处理，大文档会超时
app.post('/documents', async (c) => {
  const file = await c.req.parseBody()
  const content = await (file['file'] as File).text()
 
  // 同步走完全部流程：解析 → 切分 → 向量化 → 入库
  const parsed = parseDocument(content)
  const chunks = chunkText(parsed.content)
  const embeddings = await callEmbeddingAPI(chunks.map(c => c.text))
  await vectorDB.upsert(/* ... */)
 
  return c.json({ ok: true })
})

这个实现有几个问题：

大文档会超时——一个 10MB 的 PDF 解析可能要 30 秒，向量化要几分钟
没有状态追踪——用户不知道文档处理到哪一步了
没有错误处理——解析失败、向量化失败，用户不知道原因
没有大小限制——恶意用户可以上传超大文件
没有去重——同一份文档可以重复上传

2. 异步处理管道

正确的做法是：上传和处理分开，上传立即返回，处理在后台异步进行。

用户上传 ─→ 接收文件 ─→ 返回 document_id ─→ 完成
                           │
                           ▼
                       加入处理队列
                           │
                           ▼
              ┌─────── 后台 Worker ───────┐
              │ 1. 解析文档 → parsed      │
              │ 2. 切分文本 → chunks     │
              │ 3. 向量化    → embeddings │
              │ 4. 写入向量库             │
              │ 5. 更新状态               │
              └───────────────────────────┘
                           │
                           ▼
                       状态变更通知

状态机

// src/services/rag/document-states.ts
 
export type DocumentStatus =
  | 'pending'      // 已上传，等待处理
  | 'parsing'      // 解析中
  | 'chunking'     // 切分中
  | 'embedding'    // 向量化中
  | 'indexing'     // 写入向量库中
  | 'indexed'      // 完成
  | 'failed'       // 失败
 
export type Document = {
  id: string
  userId: string
  fileName: string
  fileSize: number
  mimeType: string
  status: DocumentStatus
  error?: string
  chunksCount?: number
  createdAt: string
  updatedAt: string
}
 
// 状态转换规则
const VALID_TRANSITIONS: Record<DocumentStatus, DocumentStatus[]> = {
  pending: ['parsing', 'failed'],
  parsing: ['chunking', 'failed'],
  chunking: ['embedding', 'failed'],
  embedding: ['indexing', 'failed'],
  indexing: ['indexed', 'failed'],
  indexed: [],  // 终态
  failed: ['pending'],  // 可以重试
}
 
export async function transitionStatus(
  docId: string,
  newStatus: DocumentStatus,
  error?: string
) {
  const doc = await getDocument(docId)
  if (!VALID_TRANSITIONS[doc.status].includes(newStatus)) {
    throw new Error(`Invalid transition: ${doc.status} → ${newStatus}`)
  }
 
  await db.query(
    'UPDATE documents SET status = ?, error = ?, updated_at = ? WHERE id = ?',
    [newStatus, error ?? null, new Date().toISOString(), docId]
  )
}

上传接口

// src/routes/documents.ts
import { Hono } from 'hono'
import { nanoid } from 'nanoid'
 
const MAX_FILE_SIZE = 50 * 1024 * 1024  // 50MB
const ALLOWED_MIME_TYPES = [
  'application/pdf',
  'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
  'text/plain',
  'text/markdown',
  'text/html',
]
 
const documents = new Hono()
 
// 上传文档
documents.post('/', async (c) => {
  const user = c.get('user')
 
  // 配额检查
  const quota = await getUserQuota(user.id)
  if (quota.used >= quota.limit) {
    return c.json({ error: '配额已满' }, 403)
  }
 
  // 解析文件
  const body = await c.req.parseBody()
  const file = body['file'] as File
 
  if (!file) {
    return c.json({ error: '缺少文件' }, 400)
  }
 
  // 格式校验
  if (!ALLOWED_MIME_TYPES.includes(file.type)) {
    return c.json({ error: `不支持的文件格式: ${file.type}` }, 400)
  }
 
  // 大小校验
  if (file.size > MAX_FILE_SIZE) {
    return c.json({ error: `文件过大，最大 ${MAX_FILE_SIZE / 1024 / 1024}MB` }, 400)
  }
 
  // 创建文档记录
  const docId = nanoid()
  await db.query(
    `INSERT INTO documents (id, user_id, file_name, file_size, mime_type, status, created_at)
     VALUES (?, ?, ?, ?, ?, 'pending', ?)`,
    [docId, user.id, file.name, file.size, file.type, new Date().toISOString()]
  )
 
  // 存储原始文件（对象存储或本地）
  const fileBuffer = await file.arrayBuffer()
  await storage.put(`documents/${docId}/raw`, new Uint8Array(fileBuffer))
 
  // 加入处理队列（不阻塞上传响应）
  await enqueueProcessing(docId)
 
  return c.json({
    id: docId,
    status: 'pending',
    message: '文档已上传，正在处理中',
  }, 202)
})
 
// 查询文档状态
documents.get('/:id', async (c) => {
  const doc = await getDocument(c.req.param('id'))
  if (!doc) return c.json({ error: 'Not found' }, 404)
 
  // 权限检查：只能查自己的文档
  if (doc.userId !== c.get('user').id) {
    return c.json({ error: 'Forbidden' }, 403)
  }
 
  return c.json(doc)
})
 
// 重试失败的文档
documents.post('/:id/retry', async (c) => {
  const doc = await getDocument(c.req.param('id'))
  if (!doc || doc.status !== 'failed') {
    return c.json({ error: 'Cannot retry' }, 400)
  }
 
  await transitionStatus(doc.id, 'pending')
  await enqueueProcessing(doc.id)
 
  return c.json({ ok: true })
})

3. 后台处理 Worker

上传接口只负责接收文件和创建记录。实际处理在后台 Worker 中完成。

// src/workers/document-processor.ts
 
export async function processDocument(docId: string) {
  const doc = await getDocument(docId)
  if (!doc) return
 
  try {
    // 1. 解析
    await transitionStatus(docId, 'parsing')
    const rawFile = await storage.get(`documents/${docId}/raw`)
    const parsed = await parseDocument(rawFile, doc.mimeType)
 
    // 保存解析结果（方便调试和重处理）
    await storage.put(
      `documents/${docId}/parsed.json`,
      new TextEncoder().encode(JSON.stringify(parsed))
    )
 
    // 2. 切分
    await transitionStatus(docId, 'chunking')
    const chunks = chunkText(parsed.content, {
      chunkSize: 1000,
      overlap: 200,
    })
 
    await storage.put(
      `documents/${docId}/chunks.json`,
      new TextEncoder().encode(JSON.stringify(chunks))
    )
 
    // 3. 向量化（分批处理，避免 API 限流）
    await transitionStatus(docId, 'embedding')
    const BATCH_SIZE = 50
    const allEmbeddings: number[][] = []
 
    for (let i = 0; i < chunks.length; i += BATCH_SIZE) {
      const batch = chunks.slice(i, i + BATCH_SIZE)
      const embeddings = await callEmbeddingAPI(batch.map((c) => c.text))
      allEmbeddings.push(...embeddings)
    }
 
    // 4. 写入向量库
    await transitionStatus(docId, 'indexing')
    const vectors = chunks.map((chunk, i) => ({
      id: `${docId}-chunk-${i}`,
      vector: allEmbeddings[i],
      metadata: {
        text: chunk.text,
        documentId: docId,
        documentTitle: parsed.title,
        chunkIndex: i,
        source: doc.fileName,
        userId: doc.userId,
      },
    }))
 
    await vectorDB.upsert(vectors)
 
    // 5. 完成
    await db.query(
      'UPDATE documents SET status = ?, chunks_count = ?, updated_at = ? WHERE id = ?',
      ['indexed', chunks.length, new Date().toISOString(), docId]
    )
 
  } catch (err) {
    const message = err instanceof Error ? err.message : 'Unknown error'
    await transitionStatus(docId, 'failed', message)
    console.error(`Document processing failed: ${docId}`, err)
  }
}

队列实现

队列可以用多种方式实现：

// 方案一：数据库轮询（简单，适合小规模）
async function enqueueProcessing(docId: string) {
  await db.query(
    'INSERT INTO processing_queue (document_id, status, created_at) VALUES (?, ?, ?)',
    [docId, 'pending', new Date().toISOString()]
  )
}
 
// Worker 定时轮询
setInterval(async () => {
  const pending = await db.query(
    "SELECT document_id FROM processing_queue WHERE status = 'pending' ORDER BY created_at LIMIT 1"
  )
  if (pending.length > 0) {
    await db.query(
      "UPDATE processing_queue SET status = 'processing' WHERE document_id = ?",
      [pending[0].document_id]
    )
    await processDocument(pending[0].document_id)
    await db.query(
      "UPDATE processing_queue SET status = 'done' WHERE document_id = ?",
      [pending[0].document_id]
    )
  }
}, 5000)  // 每 5 秒轮询一次
 
// 方案二：Redis / BullMQ（适合生产环境）
import { Queue, Worker } from 'bullmq'
 
const ragQueue = new Queue('rag-processing', { connection: { host: 'localhost', port: 6379 } })
 
async function enqueueProcessing(docId: string) {
  await ragQueue.add('process-document', { docId })
}
 
const worker = new Worker('rag-processing', async (job) => {
  await processDocument(job.data.docId)
}, { connection: { host: 'localhost', port: 6379 } })

4. 大文档处理：断点续传

大文档处理可能耗时几分钟。如果中途失败，不应该从头来。

// 处理进度追踪
type ProcessingProgress = {
  documentId: string
  totalChunks: number
  processedChunks: number
  currentStep: 'parsing' | 'chunking' | 'embedding' | 'indexing'
}
 
async function processDocumentWithResume(docId: string) {
  const doc = await getDocument(docId)
  const progress = await getProgress(docId)
 
  // 从上次中断的地方继续
  if (progress && progress.currentStep === 'embedding') {
    // 已经解析和切分过了，直接从向量化继续
    const chunks = JSON.parse(
      new TextDecoder().decode(await storage.get(`documents/${docId}/chunks.json`))
    )
 
    await transitionStatus(docId, 'embedding')
    const embeddings = []
 
    // 从上次中断的位置继续
    for (let i = progress.processedChunks; i < chunks.length; i += BATCH_SIZE) {
      const batch = chunks.slice(i, i + BATCH_SIZE)
      const batchEmbeddings = await callEmbeddingAPI(batch.map((c) => c.text))
      embeddings.push(...batchEmbeddings)
 
      // 更新进度
      await updateProgress(docId, { processedChunks: i + batch.length })
    }
 
    // ... 继续写入向量库
  } else {
    // 从头开始
    await processDocument(docId)
  }
}

5. 批量上传

用户可能需要一次性上传多个文档。

// 批量上传接口
documents.post('/batch', async (c) => {
  const user = c.get('user')
  const body = await c.req.parseBody()
 
  const files = Object.entries(body)
    .filter(([key]) => key.startsWith('file'))
    .map(([, file]) => file as File)
 
  if (files.length === 0) {
    return c.json({ error: '没有文件' }, 400)
  }
 
  if (files.length > 20) {
    return c.json({ error: '单次最多上传 20 个文件' }, 400)
  }
 
  const results = []
  for (const file of files) {
    // 复用单个上传的逻辑
    const docId = nanoid()
    await db.query(/* ... */)
    await storage.put(`documents/${docId}/raw`, new Uint8Array(await file.arrayBuffer()))
    await enqueueProcessing(docId)
    results.push({ id: docId, fileName: file.name, status: 'pending' })
  }
 
  return c.json({ documents: results }, 202)
})

6. 删除文档

删除文档时需要同时清理向量库中的数据。

// 删除文档
documents.delete('/:id', async (c) => {
  const doc = await getDocument(c.req.param('id'))
  if (!doc) return c.json({ error: 'Not found' }, 404)
  if (doc.userId !== c.get('user').id) return c.json({ error: 'Forbidden' }, 403)
 
  // 1. 从向量库删除相关向量
  // 方式一：按 metadata 过滤删除（如果向量库支持）
  await vectorDB.deleteByFilter({ documentId: doc.id })
 
  // 方式二：如果向量库不支持按 metadata 删除，需要记录所有 chunk ID
  // const chunkIds = await db.query('SELECT id FROM chunks WHERE document_id = ?', [doc.id])
  // await vectorDB.deleteByIds(chunkIds.map(c => c.id))
 
  // 2. 删除存储的文件
  await storage.delete(`documents/${doc.id}/raw`)
  await storage.delete(`documents/${doc.id}/parsed.json`)
  await storage.delete(`documents/${doc.id}/chunks.json`)
 
  // 3. 删除数据库记录
  await db.query('DELETE FROM documents WHERE id = ?', [doc.id])
 
  return c.json({ ok: true })
})

7. 前端状态轮询

前端上传后需要轮询状态，直到文档处理完成。

// 前端轮询逻辑
async function uploadAndWaitForProcessing(file: File) {
  // 1. 上传
  const formData = new FormData()
  formData.append('file', file)
  const uploadRes = await fetch('/documents', { method: 'POST', body: formData })
  const { id } = await uploadRes.json()
 
  // 2. 轮询状态
  const maxWait = 300_000  // 最多等 5 分钟
  const pollInterval = 2000
  const startTime = Date.now()
 
  while (Date.now() - startTime < maxWait) {
    const statusRes = await fetch(`/documents/${id}`)
    const doc = await statusRes.json()
 
    if (doc.status === 'indexed') {
      return { success: true, document: doc }
    }
    if (doc.status === 'failed') {
      return { success: false, error: doc.error }
    }
 
    // 更新 UI 进度
    updateProgress(doc.status)
 
    await new Promise((r) => setTimeout(r, pollInterval))
  }
 
  return { success: false, error: '处理超时' }
}

更好的方式是 WebSocket 或 SSE 推送状态变更，避免轮询。

总结

回顾这一节的要点：

文档上传应该异步化——上传立即返回，处理在后台进行
状态机管理文档生命周期：pending → parsing → chunking → embedding → indexing → indexed / failed
处理管道需要断点续传——大文档中断后不需要从头来
批量上传、删除、重试都需要考虑
删除文档时需要同步清理向量库中的数据
前端可以通过轮询或 WebSocket 追踪处理状态

下一篇讲文档解析——怎么把 PDF、Word、HTML、Markdown 转成纯文本。