14.18-RAG接口设计

要点

  • RAG 接口需要覆盖三个核心流程：文档管理（上传/删除）、知识库检索、对话集成
  • 文档上传是异步流程——接口设计要考虑轮询和回调
  • 检索接口需要暴露足够的参数（topK、阈值、过滤）让前端灵活控制
  • 对话接口把 RAG 和聊天历史整合——系统 Prompt + 上下文 + 历史 + 用户输入
  • 流式响应让用户体验更好——先返回检索结果，再流式生成回答
  • 错误处理和降级策略要在接口层统一处理

1. 接口概览

RAG 系统需要以下几类接口：

文档管理：
  POST   /api/documents          上传文档（异步处理）
  GET    /api/documents          列出文档
  GET    /api/documents/:id      文档详情
  DELETE /api/documents/:id      删除文档
  GET    /api/documents/:id/status  处理状态查询

检索：
  POST   /api/rag/search         检索（只返回相关 chunk，不生成回答）

对话：
  POST   /api/rag/chat           RAG 对话（检索 + 生成）
  POST   /api/rag/chat/stream    流式对话

知识库管理：
  POST   /api/knowledge-bases    创建知识库
  GET    /api/knowledge-bases    列出知识库
  DELETE /api/knowledge-bases/:id  删除知识库

2. 文档上传接口

文档上传是异步的——解析、切分、嵌入都需要时间。

上传接口

// src/routes/documents.ts
import { Hono } from 'hono'
import { z } from 'zod'
import { zValidator } from '@hono/zod-validator'
 
const documentsApp = new Hono()
 
const uploadSchema = z.object({
  title: z.string().min(1).max(256),
  knowledgeBaseId: z.string(),
  // 文件通过 multipart/form-data 上传
})
 
documentsApp.post('/', zValidator('form', uploadSchema), async (c) => {
  const user = c.get('user')
  const { title, knowledgeBaseId } = c.req.valid('form')
  const file = c.req.valid('form').file
 
  if (!file) {
    return c.json({ error: 'No file provided' }, 400)
  }
 
  // 1. 验证知识库权限
  const kb = await knowledgeBaseService.getById(knowledgeBaseId)
  if (!kb || kb.tenantId !== user.tenantId) {
    return c.json({ error: 'Knowledge base not found' }, 404)
  }
  if (!hasPermission(user, kb, 'write')) {
    return c.json({ error: 'Permission denied' }, 403)
  }
 
  // 2. 保存文件到对象存储
  const fileKey = `docs/${user.tenantId}/${Date.now()}-${file.name}`
  await objectStorage.put(fileKey, await file.arrayBuffer())
 
  // 3. 创建文档记录（状态：pending）
  const document = await documentService.create({
    title,
    knowledgeBaseId,
    fileKey,
    fileName: file.name,
    fileSize: file.size,
    mimeType: file.type,
    status: 'pending',
    userId: user.id,
    tenantId: user.tenantId,
  })
 
  // 4. 推入处理队列（异步）
  await queue.publish('document.process', {
    documentId: document.id,
    fileKey,
    mimeType: file.type,
  })
 
  return c.json({
    id: document.id,
    status: document.status,
    message: '文档已上传，正在处理',
  }, 202)
})

状态查询接口

前端轮询文档处理状态：

documentsApp.get('/:id/status', async (c) => {
  const user = c.get('user')
  const { id } = c.req.param()
 
  const document = await documentService.getById(id)
  if (!document || document.tenantId !== user.tenantId) {
    return c.json({ error: 'Not found' }, 404)
  }
 
  return c.json({
    id: document.id,
    status: document.status,       // pending | processing | completed | failed
    progress: document.progress,   // 0-100
    error: document.error,         // 失败时的错误信息
    chunks: document.chunkCount,   // 处理完成后的 chunk 数量
    createdAt: document.createdAt,
    updatedAt: document.updatedAt,
  })
})

文档状态机：

pending → processing → completed
                    ↘ failed → processing (重试)

删除接口

删除文档时需要同步清理向量数据：

documentsApp.delete('/:id', async (c) => {
  const user = c.get('user')
  const { id } = c.req.param()
 
  const document = await documentService.getById(id)
  if (!document || document.tenantId !== user.tenantId) {
    return c.json({ error: 'Not found' }, 404)
  }
  if (!hasPermission(user, document, 'delete')) {
    return c.json({ error: 'Permission denied' }, 403)
  }
 
  // 1. 删除向量数据
  await vectorStore.deleteDocument(document.knowledgeBaseId, id)
 
  // 2. 删除文件
  await objectStorage.delete(document.fileKey)
 
  // 3. 更新文档状态
  await documentService.updateStatus(id, 'deleted')
 
  return c.json({ success: true })
})

3. 检索接口

检索接口只返回相关 chunk——不生成回答。用于前端需要展示引用来源或让用户选择上下文的场景。

const ragApp = new Hono()
 
const searchSchema = z.object({
  query: z.string().min(1).max(2000),
  knowledgeBaseIds: z.array(z.string()).optional(),  // 指定知识库，不传则搜索用户所有可访问的
  topK: z.number().int().min(1).max(50).optional().default(5),
  threshold: z.number().min(0).max(1).optional().default(0.7),
  enableRerank: z.boolean().optional().default(false),
  filters: z.object({
    documentIds: z.array(z.string()).optional(),
    dateRange: z.object({
      start: z.string().datetime().optional(),
      end: z.string().datetime().optional(),
    }).optional(),
  }).optional(),
})
 
ragApp.post('/search', zValidator('json', searchSchema), async (c) => {
  const user = c.get('user')
  const params = c.req.valid('json')
 
  // 1. 查询 embedding
  const queryVector = await embeddingService.embed(params.query)
 
  // 2. 构建过滤条件
  const filter = buildSearchFilter(user, params)
 
  // 3. 向量检索
  let results = await vectorStore.search(
    params.knowledgeBaseIds,
    queryVector,
    {
      topK: params.enableRerank ? params.topK * 5 : params.topK,
      filter,
      threshold: params.threshold,
    }
  )
 
  // 4. Rerank（如果启用）
  if (params.enableRerank) {
    results = await rerankService.rerank(params.query, results, {
      topK: params.topK,
    })
  }
 
  // 5. 格式化返回
  return c.json({
    results: results.map((r) => ({
      id: r.id,
      content: r.content,
      score: r.score,
      metadata: {
        documentId: r.metadata.documentId,
        documentTitle: r.metadata.documentTitle,
        sectionTitle: r.metadata.sectionTitle,
        chunkIndex: r.metadata.chunkIndex,
      },
    })),
    total: results.length,
  })
})
 
function buildSearchFilter(user: User, params: z.infer<typeof searchSchema>): Filter {
  const conditions: FilterCondition[] = [
    // 租户隔离
    { key: 'tenant_id', match: { value: user.tenantId } },
  ]
 
  // 知识库过滤
  if (params.knowledgeBaseIds?.length) {
    conditions.push({
      key: 'knowledge_base_id',
      match: { any: params.knowledgeBaseIds },
    })
  } else {
    // 默认：用户可访问的所有知识库
    conditions.push({
      key: 'knowledge_base_id',
      match: { any: user.accessibleKnowledgeBases },
    })
  }
 
  // 文档过滤
  if (params.filters?.documentIds?.length) {
    conditions.push({
      key: 'document_id',
      match: { any: params.filters.documentIds },
    })
  }
 
  // 时间范围
  if (params.filters?.dateRange?.start) {
    conditions.push({
      key: 'created_at',
      range: { gte: params.filters.dateRange.start },
    })
  }
  if (params.filters?.dateRange?.end) {
    conditions.push({
      key: 'created_at',
      range: { lte: params.filters.dateRange.end },
    })
  }
 
  return { must: conditions }
}

4. 对话接口

对话接口整合了检索和生成。

const chatSchema = z.object({
  message: z.string().min(1).max(5000),
  knowledgeBaseIds: z.array(z.string()).optional(),
  topK: z.number().int().min(1).max(20).optional().default(5),
  enableRerank: z.boolean().optional().default(true),
  history: z.array(z.object({
    role: z.enum(['user', 'assistant']),
    content: z.string(),
  })).optional().default([]),
})
 
ragApp.post('/chat', zValidator('json', chatSchema), async (c) => {
  const user = c.get('user')
  const params = c.req.valid('json')
 
  // 1. 检索上下文
  const queryVector = await embeddingService.embed(params.message)
  const filter = buildSearchFilter(user, params)
 
  let chunks = await vectorStore.search(
    params.knowledgeBaseIds ?? user.accessibleKnowledgeBases,
    queryVector,
    { topK: params.enableRerank ? params.topK * 5 : params.topK, filter }
  )
 
  if (params.enableRerank) {
    chunks = await rerankService.rerank(params.message, chunks, { topK: params.topK })
  }
 
  // 2. 构建上下文
  const context = contextBuilder.build(chunks)
 
  // 3. 构建消息
  const messages: Message[] = [
    {
      role: 'system',
      content: buildSystemPrompt(context),
    },
    // 对话历史
    ...params.history.map((h) => ({
      role: h.role as 'user' | 'assistant',
      content: h.content,
    })),
    // 当前问题
    {
      role: 'user',
      content: params.message,
    },
  ]
 
  // 4. 调用 LLM
  const response = await llm.chat({ messages, temperature: 0.3 })
 
  // 5. 提取引用
  const { cleanAnswer, citations } = extractCitations(response.content)
  const references = buildReferences(citations, chunks)
 
  return c.json({
    answer: cleanAnswer,
    citations,
    references,
    sources: chunks.map((c) => ({
      id: c.id,
      documentTitle: c.metadata.documentTitle,
      sectionTitle: c.metadata.sectionTitle,
    })),
  })
})
 
function buildSystemPrompt(context: string): string {
  return `你是一个知识库助手。请根据以下参考资料回答用户的问题。
 
## 回答要求
 
1. 只基于参考资料回答，不要编造信息
2. 在每个事实陈述后面用 [编号] 标注来源
3. 如果参考资料中没有相关信息，明确说明「根据已有资料无法回答」
4. 保持回答简洁、准确
 
## 参考资料
 
${context}`
}

5. 流式对话接口

流式响应让用户体验更好——先显示检索到的来源，再流式生成回答。

ragApp.post('/chat/stream', zValidator('json', chatSchema), async (c) => {
  const user = c.get('user')
  const params = c.req.valid('json')
 
  const stream = new ReadableStream({
    async start(controller) {
      const encoder = new TextEncoder()
 
      try {
        // 1. 检索
        const queryVector = await embeddingService.embed(params.message)
        const filter = buildSearchFilter(user, params)
        let chunks = await vectorStore.search(
          params.knowledgeBaseIds ?? user.accessibleKnowledgeBases,
          queryVector,
          { topK: params.topK * 3, filter }
        )
 
        if (params.enableRerank) {
          chunks = await rerankService.rerank(params.message, chunks, { topK: params.topK })
        }
 
        // 2. 先发送来源信息
        controller.enqueue(encoder.encode(
          `data: ${JSON.stringify({
            type: 'sources',
            sources: chunks.map((c) => ({
              id: c.id,
              documentTitle: c.metadata.documentTitle,
              sectionTitle: c.metadata.sectionTitle,
              score: c.score,
            })),
          })}\n\n`
        ))
 
        // 3. 构建上下文和消息
        const context = contextBuilder.build(chunks)
        const messages: Message[] = [
          { role: 'system', content: buildSystemPrompt(context) },
          ...params.history.map((h) => ({ role: h.role as 'user' | 'assistant', content: h.content })),
          { role: 'user', content: params.message },
        ]
 
        // 4. 流式调用 LLM
        const llmStream = await llm.stream({ messages, temperature: 0.3 })
 
        let fullAnswer = ''
        for await (const chunk of llmStream) {
          const text = chunk.content ?? ''
          fullAnswer += text
 
          controller.enqueue(encoder.encode(
            `data: ${JSON.stringify({ type: 'token', content: text })}\n\n`
          ))
        }
 
        // 5. 发送引用信息
        const { citations } = extractCitations(fullAnswer)
        controller.enqueue(encoder.encode(
          `data: ${JSON.stringify({ type: 'citations', citations })}\n\n`
        ))
 
        // 6. 结束
        controller.enqueue(encoder.encode('data: [DONE]\n\n'))
        controller.close()
      } catch (err) {
        controller.enqueue(encoder.encode(
          `data: ${JSON.stringify({ type: 'error', error: String(err) })}\n\n`
        ))
        controller.close()
      }
    },
  })
 
  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
    },
  })
})

前端消费流式响应：

const response = await fetch('/api/rag/chat/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ message: '退款多久到账？' }),
})
 
const reader = response.body!.getReader()
const decoder = new TextDecoder()
 
while (true) {
  const { done, value } = await reader.read()
  if (done) break
 
  const text = decoder.decode(value)
  const lines = text.split('\n').filter((l) => l.startsWith('data: '))
 
  for (const line of lines) {
    const data = line.replace('data: ', '')
    if (data === '[DONE]') continue
 
    const event = JSON.parse(data)
 
    switch (event.type) {
      case 'sources':
        // 显示参考来源
        renderSources(event.sources)
        break
      case 'token':
        // 追加回答文本
        appendAnswer(event.content)
        break
      case 'citations':
        // 标注引用
        renderCitations(event.citations)
        break
      case 'error':
        showError(event.error)
        break
    }
  }
}

6. 错误处理和降级

接口层统一处理错误和降级：

// src/middleware/rag-error-handler.ts
import { createMiddleware } from 'hono/factory'
 
export const ragErrorHandler = createMiddleware(async (c, next) => {
  try {
    await next()
  } catch (err) {
    const error = err as Error
 
    // 向量数据库不可用 → 返回无结果
    if (isVectorDBError(error)) {
      return c.json({
        answer: '抱歉，知识库暂时不可用，请稍后重试。',
        sources: [],
        error: 'Knowledge base temporarily unavailable',
      }, 200)  // 返回 200 而不是 500，前端可以展示友好提示
    }
 
    // LLM 不可用 → 只返回检索结果
    if (isLLMError(error)) {
      const query = c.get('ragQuery')
      const queryVector = await embeddingService.embed(query)
      const results = await vectorStore.search(queryVector, { topK: 5 })
 
      return c.json({
        answer: null,  // 没有生成回答
        sources: results,
        error: 'Answer generation temporarily unavailable',
      }, 200)
    }
 
    // Embedding 服务不可用 → 503
    if (isEmbeddingError(error)) {
      return c.json({ error: 'Embedding service unavailable' }, 503)
    }
 
    // 其他错误 → 500
    return c.json({ error: 'Internal server error' }, 500)
  }
})

超时处理

RAG 接口涉及多个外部服务——embedding、向量数据库、rerank、LLM。任何一个超时都需要处理：

async function searchWithTimeout<T>(
  fn: () => Promise<T>,
  timeoutMs: number,
  fallback: T
): Promise<T> {
  return Promise.race([
    fn(),
    new Promise<T>((_, reject) =>
      setTimeout(() => reject(new Error('Timeout')), timeoutMs)
    ),
  ]).catch(() => fallback)
}
 
// 使用
const chunks = await searchWithTimeout(
  () => vectorStore.search(queryVector, { topK: 20 }),
  3000,  // 3 秒超时
  []     // 超时返回空数组
)

7. 接口聚合

把所有 RAG 相关的接口注册到主应用：

// src/app.ts
import { Hono } from 'hono'
import { ragPermissionMiddleware } from './middleware/rag-permission'
import { ragErrorHandler } from './middleware/rag-error-handler'
import { documentsApp } from './routes/documents'
import { ragApp } from './routes/rag'
 
const app = new Hono()
 
// 全局中间件
app.use('/api/*', authMiddleware)
app.use('/api/rag/*', ragPermissionMiddleware)
app.use('/api/rag/*', ragErrorHandler)
 
// 路由
app.route('/api/documents', documentsApp)
app.route('/api/rag', ragApp)
 
export default app

总结

回顾这一节的要点：

  • RAG 接口覆盖三类：文档管理（异步上传）、检索（纯检索）、对话（检索+生成）
  • 文档上传是异步流程——接口返回 202，前端轮询状态
  • 检索接口暴露 topK、阈值、过滤条件等参数
  • 对话接口整合检索 + 上下文拼接 + LLM 生成 + 引用提取
  • 流式响应先返回来源，再流式生成——用户体验更好
  • 错误处理和降级策略：向量数据库不可用时返回友好提示，LLM 不可用时只返回检索结果
  • 超时处理：每个外部服务调用都设超时，超时后降级
  • 权限中间件统一注入，错误处理中间件统一兜底

至此，第 14 章 RAG 的全部内容讲完了。从基本概念到文档解析、切分、Embedding、向量数据库、检索策略、Rerank、上下文拼接、引用返回、权限控制、评估指标、接口设计——覆盖了构建一个 RAG 系统的所有关键环节。

下一章回到 Cloudflare Vectorize 的具体实践——一个轻量的向量数据库方案。