长时间任务的进度与 AI 回答流式输出

导出、导入、转码与 AI 任务：跟着实际工作前进的进度条、一完成就出现的结果，以及边写边出现的回答。

导出、视频转码、生成报表、导入数据和 AI 任务，可能要十秒到十分钟。一个没有任何反馈的转圈圈，会让人刷新、再按一次，或干脆放弃。显示真实的进度，并在结果准备好的那一刻呈现，就能把人留在页面上。AI 功能也一样适用：即使任务在后台运行，也能把助理的回答边写边流式输出出来。

用户得到什么

跟着实际任务前进的进度条：“正在处理 10,000 行中的第 3,200 行”。
任务一完成，下载链接或结果就出现，即使在另一个标签页。
AI 的回答一个字一个字出现，不必等很久才一次出现。
可以放心离开页面：完成时会通知他们。

用到的功能

用户验证与 sendToUser，任务的更新只送给启动它的人，而且送到每个标签页。
私有频道（private-job-<id>），用在多人一起追踪同一个任务时。
任何语言的 worker：Node.js、PHP、Python、Go、Ruby、Java 和 .NET 的官方 Pusher 服务器库都能从队列 worker 发布。
推送通知，在用户关掉 App 后告诉他们长时间的任务已完成。

步骤 1：启动任务，取得 ID

app.post('/exports', requireLogin, async (req, res) => {
  const job = await queue.add('export', { userId: req.user.id, filters: req.body });
  res.status(202).json({ jobId: job.id });
});

步骤 2：从 worker 回报进度

在有意义的阶段回报进度，不要每一行都回报：

worker.process('export', async (job) => {
  const user = String(job.data.userId);
  const rows = await db.orders.query(job.data.filters);
  for (let i = 0; i < rows.length; i += 1000) {
    await writeChunk(rows.slice(i, i + 1000));
    await pusher.sendToUser(user, 'job-progress', { jobId: job.id, done: Math.min(i + 1000, rows.length), total: rows.length });
  }
  const url = await uploadAndSign();
  await pusher.sendToUser(user, 'job-done', { jobId: job.id, url });
});

用 pusher 包的 Python worker 写法相同：

pusher_client.send_to_user(str(user_id), "job-progress", {"jobId": job_id, "done": done, "total": total})

步骤 3：在浏览器里显示

用户在连接上登录后（请看实时站内通知）：

const { jobId } = await fetch('/exports', { method: 'POST', body: form }).then((r) => r.json());

pusher.user.bind('job-progress', (p) => {
  if (p.jobId === jobId) progress.set(p.done / p.total);
});
pusher.user.bind('job-done', (p) => {
  if (p.jobId === jobId) showDownload(p.url);
});

流式输出 AI 回答

当 AI 模型在后台任务里撰写回答时，把输出分成小批转给用户，这样一个很长的回答是几十条消息，而不是几千条：

let buffer = '';
let last = Date.now();
for await (const chunk of llm.stream(prompt)) {
  buffer += chunk.text;
  if (Date.now() - last > 150 || buffer.length > 2000) {
    await pusher.sendToUser(user, 'ai-delta', { id: answerId, text: buffer });
    buffer = '';
    last = Date.now();
  }
}
await pusher.sendToUser(user, 'ai-done', { id: answerId, text: buffer });

pusher.user.bind('ai-delta', ({ id, text }) => answers.get(id).append(text));
pusher.user.bind('ai-done', ({ id, text }) => answers.get(id).append(text).finish());

也把完整的回答存在你的服务器上，刷新后才还在。

小提醒

每次更新都是一条消息。每 1–5% 或每 150 毫秒更新一次就很顺，一个任务也只用几十条消息。
在启动任务之前先订阅，或在页面打开时查找一次任务状态，就不会漏掉在页面连接之前送出的更新。
如果用户已经关掉标签页，任务完成时照推送通知指南发送推送。