Next.js AI SaaS 架构模式：从页面到任务队列

0阅读10分钟

一次线上事故让我重新理解了 AI SaaS 的边界

我负责过一个 AI 写作助手，早期架构很「标准」：前端点击按钮，Server Action 里直接调 OpenAI API，流式返回结果，写到页面上。Demo 阶段一切顺利，内部试用两周也没出问题。

直到一个用户提交了一篇 8000 字的长文要求改写。

模型生成了整整 4 分钟，Server Action 在第 47 秒超时断开。用户看到的是页面卡在加载状态，后台是 OpenAI 还在继续生成但响应无处可去，数据库里这条任务记录停留在 pending 永远不会变成 succeeded。更要命的是，计费系统已经记录了一次完整调用——因为模型侧的 token 确实用掉了。

这次事故暴露的问题不只是超时。它揭示了 AI SaaS 和传统 Web 应用在架构上的根本差异：模型调用是一个不可靠的、长时间运行的、有成本的外部进程，而传统 Web 架构默认所有请求都是毫秒级返回的同步操作。

这篇文章总结的是我在后续重构中摸索出的一套架构模式，核心思路是把 Next.js 的页面、API、任务队列、模型调用、结果存储和可观测性拆成清晰的边界。

AI 任务不是普通 API 请求

传统 CRUD 请求有一个隐含假设：请求处理完毕，响应返回，连接关闭，整个过程在几百毫秒内完成。即使慢一些，也很少超过 5 秒。在这个假设下，HTTP 连接本身就是任务的生命周期容器。

AI 任务打破了这个假设。一次模型调用可能持续数秒到数分钟，中间可能出现 token 限流、模型超时、网络中断，而且每一次调用都有真金白银的成本。Vercel 官方文档明确指出，Serverless Function 的执行时间有硬上限——Hobby 计划 10 秒，Pro 计划 60 秒，Enterprise 计划最高 300 秒（参见 Next.js Route Handlers 文档）。这意味着一个复杂的多步 AI 任务，光模型生成时间就可能超出函数执行上限。

Ably 团队在一篇工程博客中总结得更直接：标准的 HTTP 流式传输假设「用户断开连接等于放弃任务」，但实际场景中，用户可能只是切换了浏览器标签、网络抖动了一下、或者笔记本合盖了。如果任务的存活完全依赖一条 HTTP 连接，这些正常的用户行为都会导致任务丢失。

这就需要把任务的生命周期从 HTTP 连接中解耦出来。

核心架构：五层边界

经过几次重构迭代，我把 AI SaaS 的后端拆成了五层，每层有明确的职责和边界：

流程图画布 · 115%
Mermaid 流程图加载中...

页面层只负责输入收集和结果展示。它不应该知道模型调用的细节，不应该持有流式连接的生命周期，也不应该处理重试逻辑。

API 层（Route Handler）负责鉴权、参数校验、限流和任务入队。它返回一个任务 ID 给前端，然后关闭连接。这个响应应该在 200 毫秒内完成。

任务层把任务写入数据库或队列，标记为 pending，然后立即返回。任务的执行不再依赖任何 HTTP 连接。

Worker 层是一个独立的持久化进程，从队列中取任务，调用模型，写入结果，更新状态。它不受 Serverless 函数超时限制。

存储层用 Postgres 保存任务状态、结果和向量数据（用 pgvector 做语义检索）。任务的状态机包括 pending → running → succeeded / failed / cancelled。

案例一：流式响应的三种实现路径

在决定怎么把模型输出推给前端时，我试过三种方案，踩了不少坑。

方案 A：Server Action 直接流式（早期方案，已废弃）

// ❌ 坏做法：Server Action 直接调模型流式返回
'use server'
 
export async function generateContent(prompt: string) {
  const stream = await openai.chat.completions.create({
    model: 'gpt-4o',
    messages: [{ role: 'user', content: prompt }],
    stream: true,
  })
 
  // Server Action 无法真正流式返回给浏览器
  // 结果会被缓冲，直到全部生成完毕才一次性返回
  let fullText = ''
  for await (const chunk of stream) {
    fullText += chunk.choices[0]?.delta?.content || ''
  }
  return fullText
}

问题在于 Server Action 的返回值是一次性的 RPC 响应，它不支持流式传输到客户端。Vercel AI SDK 的 GitHub issue #617 讨论过这个限制——社区多次呼吁支持 Server Action 流式返回，但截至 AI SDK 6.x，Route Handler 仍然是流式场景的推荐入口。

方案 B：Route Handler + SSE（改进方案）

// ✅ 好做法：Route Handler + ReadableStream 流式返回
// app/api/generate/route.ts
export const runtime = 'edge'
 
export async function POST(req: Request) {
  const { messages } = await req.json()
 
  // 鉴权和限流放在 Route Handler
  const session = await getSession(req)
  if (!session) return new Response('Unauthorized', { status: 401 })
 
  const result = await streamText({
    model: openai('gpt-4o'),
    messages,
  })
 
  return result.toDataStreamResponse()
}

客户端用 Vercel AI SDK 的 useChat hook 消费：

// ✅ 好做法：客户端用 useChat 消费流式响应
'use client'
import { useChat } from 'ai/react'
 
function ChatPanel() {
  const { messages, input, handleInputChange, handleSubmit, isLoading } = useChat({
    api: '/api/generate',
  })
 
  return (
    <div>
      {messages.map((m) => (
        <div key={m.id}>{m.content}</div>
      ))}
      <form onSubmit={handleSubmit}>
        <input value={input} onChange={handleInputChange} />
      </form>
    </div>
  )
}

这种方案适合「即时对话」场景——用户等待结果、实时看到输出、完成后交互结束。HTTP 连接的生命周期和任务的生命周期一致，不需要额外的状态管理。

方案 C：任务队列 + 轮询（生产方案）

// ✅ 好做法：Route Handler 只负责入队，返回任务 ID
// app/api/tasks/route.ts
export async function POST(req: Request) {
  const { prompt, type } = await req.json()
  const session = await getSession(req)
 
  // 参数校验
  if (!prompt || prompt.length > 10000) {
    return Response.json({ error: 'INVALID_PROMPT' }, { status: 400 })
  }
 
  // 创建任务记录
  const task = await db.task.create({
    data: {
      userId: session.userId,
      type,
      prompt,
      status: 'pending',
    },
  })
 
  // 推入队列（不等待执行完成）
  await taskQueue.enqueue('ai-generate', {
    taskId: task.id,
    userId: session.userId,
  })
 
  // 200ms 内返回任务 ID
  return Response.json({ taskId: task.id })
}

Worker 进程独立执行：

// ✅ 好做法：Worker 独立进程，不受 Serverless 超时限制
// worker/ai-generate.ts
export async function processJob(payload: { taskId: string; userId: string }) {
  // 更新状态为 running
  await db.task.update({
    where: { id: payload.taskId },
    data: { status: 'running', startedAt: new Date() },
  })
 
  try {
    const result = await streamText({
      model: openai('gpt-4o'),
      messages: [{ role: 'user', content: payload.prompt }],
      // 长文本任务不需要流式返回给 HTTP
      // 直接等完整结果
    })
 
    const fullText = await result.text
 
    // 写入结果
    await db.task.update({
      where: { id: payload.taskId },
      data: {
        status: 'succeeded',
        result: fullText,
        tokenUsage: result.usage,
        completedAt: new Date(),
      },
    })
  } catch (error) {
    // 结构化错误，方便重试判断
    await db.task.update({
      where: { id: payload.taskId },
      data: {
        status: 'failed',
        error: {
          code: error.code || 'UNKNOWN',
          message: error.message,
          retryable: isRetryable(error),
        },
      },
    })
  }
}

前端轮询任务状态：

// ✅ 好做法：前端轮询任务状态，支持断线恢复
function useTaskPolling(taskId: string) {
  const [task, setTask] = useState(null)
 
  useEffect(() => {
    const poll = async () => {
      const res = await fetch(`/api/tasks/${taskId}`)
      const data = await res.json()
      setTask(data)
 
      // 任务未完成，继续轮询
      if (data.status === 'pending' || data.status === 'running') {
        setTimeout(poll, 2000)
      }
    }
    poll()
  }, [taskId])
 
  return task
}

三种方案的对比：

维度Server Action 直接调用Route Handler + SSE任务队列 + 轮询
适用场景Demo/原型即时对话（< 60s）长任务（> 60s）、多步骤
超时风险极高（10-60s 限制）中等（Edge 可放宽）无（Worker 持久运行）
断线恢复不支持需额外实现 stream resumption天然支持（查数据库）
成本可控性差（无重试控制）中（可 abort）好（结构化重试）
实现复杂度低中高

案例二：Token 计费与任务状态的一致性

早期版本里，计费是在模型调用完成后才记录的。问题在于：如果 Worker 在模型调用中途崩溃（OOM、进程被杀），token 已经消耗了但计费记录没有写入。月底对账时发现模型账单和平台计费数据差了 15%。

// ❌ 坏做法：模型调用完成后才记录计费
async function processJob(payload) {
  const result = await streamText({ model: openai('gpt-4o'), messages })
  const text = await result.text
  const usage = result.usage
 
  // 如果上面崩溃了，下面的计费永远不会执行
  await db.billing.create({
    data: {
      userId: payload.userId,
      tokens: usage.totalTokens,
      cost: calculateCost(usage),
    },
  })
}

修复方案是把计费拆成两步：调用前预扣，调用后结算差额。

// ✅ 好做法：预扣 + 结算，确保计费不丢
async function processJob(payload) {
  // 1. 预估 token 数，预扣费用
  const estimatedTokens = estimateTokens(payload.prompt)
  const preDebit = await db.billing.create({
    data: {
      userId: payload.userId,
      taskId: payload.taskId,
      type: 'pre_debit',
      tokens: estimatedTokens,
      cost: calculateCost({ totalTokens: estimatedTokens }),
      status: 'pending',
    },
  })
 
  try {
    // 2. 调用模型
    const result = await streamText({ model: openai('gpt-4o'), messages })
    const text = await result.text
    const actualUsage = result.usage
 
    // 3. 用实际 token 数结算
    await db.billing.update({
      where: { id: preDebit.id },
      data: {
        type: 'settle',
        tokens: actualUsage.totalTokens,
        cost: calculateCost(actualUsage),
        status: 'settled',
        settledAt: new Date(),
      },
    })
 
    // 4. 更新任务结果
    await db.task.update({
      where: { id: payload.taskId },
      data: { status: 'succeeded', result: text, tokenUsage: actualUsage },
    })
  } catch (error) {
    // 5. 失败时退还预扣
    await db.billing.update({
      where: { id: preDebit.id },
      data: { status: 'refunded', refundedAt: new Date() },
    })
 
    await db.task.update({
      where: { id: payload.taskId },
      data: { status: 'failed', error: serializeError(error) },
    })
  }
}

这里的关键是：计费记录的生命周期不应该和模型调用的生命周期绑定。预扣机制让即使用户在模型生成过程中取消任务、Worker 崩溃、或者网络中断，至少有一笔「已记录的调用尝试」留在数据库里，可以对账。

计费方式数据丢失风险用户体验对账难度
调用后记录高（崩溃=丢失）余额实时扣减难（需和模型厂商账单交叉比对）
预扣 + 结算低（预扣一定存在）先冻结后扣减，多退少补易（平台记录即真相源）
按次预付无需先充值最简单（但不灵活）

案例三：pgvector 语义检索与任务上下文管理

SaaS 产品做到一定规模后，用户会期望 AI 「记住」之前的对话和生成的内容。这就需要一个上下文存储层。

我最初的方案是把用户所有历史消息拼进 prompt，模型上下文窗口轻松被打满。GPT-4o 的 128K token 窗口看起来很大，但一个活跃用户一个月产生的对话内容轻松超过 200K token。

// ❌ 坏做法：把所有历史消息塞进 prompt
async function buildPrompt(userId: string, currentMessage: string) {
  const history = await db.message.findMany({
    where: { userId },
    orderBy: { createdAt: 'desc' },
    // 没有截断，全量拉取
  })
 
  const messages = [
    ...history.map((m) => ({ role: m.role, content: m.content })),
    { role: 'user', content: currentMessage },
  ]
 
  return messages
}

修复方案是引入 pgvector 做语义检索：只拉取和当前问题最相关的历史片段，而不是全量历史。

// ✅ 好做法：pgvector 语义检索，只拉取相关上下文
import { openaiEmbeddings } from '@ai-sdk/openai'
import { embed } from 'ai'
 
async function buildContextualPrompt(
  userId: string,
  currentMessage: string,
  maxTokens = 4000
) {
  // 1. 把当前消息转成向量
  const { embedding } = await embed({
    model: openaiEmbeddings('text-embedding-3-small'),
    value: currentMessage,
  })
 
  // 2. 用 pgvector 做余弦相似度检索
  const relevantChunks = await db.$queryRaw`
    SELECT content, role, created_at,
      1 - (embedding <=> ${embedding}::vector) as similarity
    FROM message_embeddings
    WHERE user_id = ${userId}
      AND 1 - (embedding <=> ${embedding}::vector) > 0.7
    ORDER BY embedding <=> ${embedding}::vector
    LIMIT 10
  `
 
  // 3. 按相关性排序，拼入 system prompt
  const contextBlock = relevantChunks
    .map((c) => `[${c.role}] ${c.content}`)
    .join('\n---\n')
 
  return [
    {
      role: 'system',
      content: `以下是用户的历史相关内容，仅供参考：\n${contextBlock}`,
    },
    { role: 'user', content: currentMessage },
  ]
}

消息写入时同步生成 embedding：

// ✅ 好做法：消息写入时同步生成 embedding
async function saveMessageWithEmbedding(
  userId: string,
  role: string,
  content: string
) {
  // 保存原始消息
  const message = await db.message.create({
    data: { userId, role, content },
  })
 
  // 生成 embedding 并存入向量表
  const { embedding } = await embed({
    model: openaiEmbeddings('text-embedding-3-small'),
    value: content,
  })
 
  await db.$executeRaw`
    INSERT INTO message_embeddings (message_id, user_id, content, embedding)
    VALUES (${message.id}, ${userId}, ${content}, ${embedding}::vector)
  `
}
上下文策略Token 利用率相关性实现复杂度
全量历史差（快速撞上限）高（但噪音也多）低
滑动窗口（最近 N 条）中中（丢失早期上下文）低
语义检索（pgvector）好（只取相关片段）高中
摘要 + 语义检索最好最高高

好/坏做法对比汇总

1. API 层设计

// ❌ 坏做法：Route Handler 里直接调模型
export async function POST(req: Request) {
  const { prompt } = await req.json()
  // 没有鉴权
  // 没有限流
  // 模型调用可能要等 2 分钟
  const result = await generateText({
    model: openai('gpt-4o'),
    prompt,
  })
  return Response.json({ result })
}
 
// ✅ 好做法：Route Handler 只做鉴权+入队，立即返回
export async function POST(req: Request) {
  const session = await getSession(req)
  if (!session) return new Response('Unauthorized', { status: 401 })
 
  // 限流检查
  const rateLimit = await checkRateLimit(session.userId, 'generate', {
    max: 20,
    window: '1m',
  })
  if (!rateLimit.allowed) {
    return Response.json(
      { error: 'RATE_LIMITED', retryAfter: rateLimit.retryAfter },
      { status: 429 }
    )
  }
 
  const { prompt } = await req.json()
  const task = await createTask({ userId: session.userId, prompt })
  await taskQueue.enqueue('ai-generate', { taskId: task.id })
 
  return Response.json({ taskId: task.id }, { status: 202 })
}

2. 任务状态管理

// ❌ 坏做法：状态只有「完成」和「未完成」
type TaskStatus = 'done' | 'not_done'
 
// ✅ 好做法：结构化状态机，覆盖所有生命周期
type TaskStatus = 'pending' | 'running' | 'succeeded' | 'failed' | 'cancelled'
 
interface TaskError {
  code: string           // 'RATE_LIMITED' | 'MODEL_TIMEOUT' | 'INVALID_INPUT'
  message: string
  retryable: boolean     // 是否值得自动重试
  retryCount: number     // 已重试次数
  maxRetries: number     // 最大重试次数
}
 
interface Task {
  id: string
  userId: string
  status: TaskStatus
  error: TaskError | null
  tokenUsage: { prompt: number; completion: number; total: number } | null
  startedAt: Date | null
  completedAt: Date | null
  createdAt: Date
}

3. 错误处理

// ❌ 坏做法：catch 了错误但只存了 message
catch (error) {
  await db.task.update({
    data: { status: 'failed', error: error.message },
  })
}
 
// ✅ 好做法：结构化错误，区分可重试和不可重试
catch (error) {
  const taskError: TaskError = {
    code: mapErrorCode(error),
    message: error.message,
    retryable: isRetryableError(error),
    retryCount: currentRetry,
    maxRetries: 3,
  }
 
  await db.task.update({
    data: { status: 'failed', error: taskError },
  })
 
  // 可重试的错误重新入队，带退避策略
  if (taskError.retryable && currentRetry < 3) {
    await taskQueue.enqueue('ai-generate', {
      taskId: task.id,
      retryCount: currentRetry + 1,
    }, {
      delay: Math.pow(2, currentRetry) * 1000, // 指数退避
    })
  }
}

生产环境检查清单

上线前必查

  • 所有 AI 任务通过队列异步执行，不在 Route Handler / Server Action 中直接调用模型
  • 任务状态机完整覆盖 pending → running → succeeded / failed / cancelled
  • 失败错误结构化存储，包含 code、retryable、retryCount
  • 计费采用预扣 + 结算机制，防止崩溃导致计费丢失
  • Route Handler 设置了 runtime = 'edge'（除非需要 Node.js 特有 API）
  • 所有模型调用有限流保护（每用户每分钟/每小时上限）
  • 长文本上下文使用语义检索（pgvector），不全量塞 prompt

运营阶段必查

  • 任务执行日志包含 taskId、userId、model、tokenUsage、duration
  • 可重试错误有指数退避重试策略，最大重试次数不超过 3
  • 取消任务时同时发送 abort 信号给模型 API，避免空跑计费
  • 数据库连接池大小与 Worker 并发数匹配，不会互相饿死
  • 有告警规则覆盖：任务失败率 > 5%、平均耗时 > 阈值、计费差异 > 10%
  • 向量索引定期重建或增量更新，避免检索质量退化

什么时候不需要这么复杂

这套架构听起来很重。如果产品还在验证 PMF 阶段、日活不到 100、或者 AI 功能只是一个锦上添花的辅助，直接在 Server Action 里调模型完全够用。

判断标准是：当你的 AI 功能开始产生不可接受的账单差异、用户开始投诉任务丢失、或者你开始手动对账时，就是拆边界的时候。

架构不是一次性设计出来的，是被线上事故逼出来的。

参考资料

  1. Next.js Route Handlers 官方文档 — Serverless Function 执行时间限制和 Route Handler 使用方式
  2. Vercel AI SDK Builder's Guide — Vercel 官方 AI SDK 生产指南，涵盖流式、工具调用和多模型支持
  3. Vercel AI SDK in production: when DefaultChatTransport needs a session layer — Ably 团队关于 AI SDK 生产化 session 层和 stream resumption 的工程博客
  4. Durable Transports for your AI SDK — Electric 团队关于持久化传输层的架构方案
  5. Next.js, Background Jobs & PostgreSQL: Production in 2026 — Render 关于 Next.js 后台任务与 PostgreSQL 生产部署的架构指南
  6. SaaS Architecture Patterns with Next.js — 多租户 SaaS 架构模式，涵盖 RLS、SSE 和缓存策略
  7. Streaming AI Responses: SSE vs ReadableStream vs Vercel AI SDK — 三种流式方案对比与选型建议
  8. Next.js Server Actions vs Route Handlers — Server Actions 与 Route Handlers 的使用场景区分

评论 0

我
0 / 1000