如果一次 LLM（大语言模型） 调用可能需要 20 秒甚至更长时间，就不要把它放在 HTTP 处理程序中。把这项工作放到由 Redis（内存数据结构存储系统） 支持的内存数据结构存储系统 BullMQ 队列上，立即返回一个 job ID 和 202 Accepted，然后让单独的 worker 执行这个慢调用，同时客户端通过 Server-Sent Events 获取进度更新。
本文其余部分就是这套方案的可用代码。它涵盖带退避的重试、并发与速率限制、死信处理，以及向浏览器发送进度。
为什么请求路径是错误的放置位置
内联 LLM 调用在开发环境中看起来没问题。但在真实流量下，这些问题会出现：
超时会不断累积。你的负载均衡器、反向代理和客户端各自都有自己的空闲超时。一次长时间生成可能超过其中某个超时，即使用户的模型调用实际上成功了，用户也会看到错误。
重试会造成重复。当请求超时时，浏览器或用户会再次尝试。现在你要为两次生成付费，并且可能写入两份结果。
提供商限制会影响所有人。一波流量会把一波调用直接发送给提供商。429 会以失败的形式返回给用户。
部署会终止进行中的工作。重启 API（应用程序编程接口） 进程会丢弃所有仍在运行的生成任务。
使用队列后，API 可以保持快速，而慢的部分拥有自己的重试策略、自己的并发上限和自己的失败处理。
整体结构
客户端发送 POST /summaries。API 添加一个 job 并返回 { jobId }。
一个 worker 进程领取该 job，调用模型并保存结果。
客户端打开 GET /summaries/:jobId/events，并在 job 完成前持续接收进度。
永久失败的 job 会进入死信队列，以便有人查看它们。
安装这些包：
npm install bullmq ioredis express
共享队列设置
// queue.js
import
{
Queue
,
QueueEvents
}
from
'
bullmq
'
;
import
IORedis
from
'
ioredis
'
;
export
function
makeConnection
()
{
// BullMQ workers need maxRetriesPerRequest set to null
return
new
IORedis
(
process
.
env
.
REDIS_URL
,
{
maxRetriesPerRequest
:
null
});
}
export
const
llmQueue
=
new
Queue
(
'
llm-summaries
'
,
{
connection
:
makeConnection
()
});
export
const
deadLetterQueue
=
new
Queue
(
'
llm-dead-letter
'
,
{
connection
:
makeConnection
()
});
export
const
llmEvents
=
new
QueueEvents
(
'
llm-summaries
'
,
{
connection
:
makeConnection
()
});Workers 和 QueueEvents 使用阻塞式 Redis 命令。为每个实例提供自己的连接可以避免它们互相干扰。
入队并立即返回
// api.js
app
.
post
(
'
/summaries
'
,
async
(
req
,
res
)
=>
{
const
{
documentId
}
=
req
.
body
;
const
job
=
await
llmQueue
.
add
(
'
summarize
'
,
{
documentId
},
{
jobId
:
'
summary-
'
+
documentId
,
// same document, same job
attempts
:
5
,
backoff
:
{
type
:
'
exponential
'
,
delay
:
2000
},
removeOnComplete
:
{
age
:
3600
},
removeOnFail
:
{
age
:
24
*
3600
},
}
);
res
.
status
(
202
).
json
({
jobId
:
job
.
id
});
});自定义 jobId 是你所能获得的最廉价的去重方式。如果用户点击按钮三次，只要第一个任务仍然存在，BullMQ 就不会添加具有相同 ID 的第二个任务。根据使工作唯一化的因素来选择 ID，例如文档加上提示词版本。
使用 attempts: 5 和从 2 秒开始的指数退避，重试大约等待 2、4、8 和 16 秒。这种间隔给受到速率限制的提供商留出了恢复时间。
工作进程：并发与速率限制
// worker.js
import
{
Worker
,
UnrecoverableError
}
from
'
bullmq
'
;
import
{
makeConnection
,
deadLetterQueue
}
from
'
./queue.js
'
;
const
worker
=
new
Worker
(
'
llm-summaries
'
,
async
(
job
)
=>
{
await
job
.
updateProgress
({
stage
:
'
loading
'
,
pct
:
10
});
const
doc
=
await
loadDocument
(
job
.
data
.
documentId
);
if
(
!
doc
)
throw
new
UnrecoverableError
(
'
Document not found
'
);
await
job
.
updateProgress
({
stage
:
'
generating
'
,
pct
:
30
});
const
summary
=
await
callModel
(
doc
.
text
,
60000
);
await
job
.
updateProgress
({
stage
:
'
saving
'
,
pct
:
90
});
await
saveSummary
(
job
.
data
.
documentId
,
summary
);
return
{
documentId
:
job
.
data
.
documentId
};
},
{
connection
:
makeConnection
(),
concurrency
:
8
,
limiter
:
{
max
:
50
,
duration
:
60000
},
}
);这两个设置控制的是不同的事情：
concurrency 是一个工作进程同时运行多少个任务。3 个工作进程、并发数为 8 时，最多有 24 个模型调用在途。
limiter 作用于整个队列，跨所有工作进程。这里它允许每分钟 50 个任务。把它设置得略低于你的提供商配额，这样共享该密钥的其他服务仍有空间。
如果你混合了廉价和昂贵的任务，例如分类和长文本生成，请给它们单独的队列。否则一堆慢任务会占满所有槽位，快任务只能排在它们后面等待。
只重试正确的错误
对格式错误的请求重试五次既浪费钱又拖慢反馈。在调用模型的地方把错误分成两组：
async
function
callModel
(
text
,
timeoutMs
)
{
try
{
// llm.summarize is a stand-in for your provider SDK call
return
await
llm
.
summarize
(
text
,
{
signal
:
AbortSignal
.
timeout
(
timeoutMs
)
});
}
catch
(
err
)
{
const
status
=
err
.
status
??
0
;
if
([
400
,
401
,
403
,
404
,
422
].
includes
(
status
))
{
throw
new
UnrecoverableError
(
'
Model rejected request:
'
+
err
.
message
);
}
throw
err
;
// 429, 5xx, network errors and timeouts: let BullMQ retry
}
}UnrecoverableError 告诉 BullMQ 跳过剩余尝试并立即让任务失败。对于每次尝试都会以同样方式失败的任何情况，都使用它。
始终在模型调用上设置你自己的超时。如果提供商挂起且没有任何超时，该任务将永远占用它的并发槽位。
死信处理
BullMQ 没有内置的死信队列，但添加一个只需几行代码。工作进程的 failed 事件会在每次失败尝试时触发，因此检查这是否是最后一次：
worker
.
on
(
'
failed
'
,
async
(
job
,
err
)
=>
{
if
(
!
job
)
return
;
const
outOfAttempts
=
job
.
attemptsMade
>=
(
job
.
opts
.
attempts
??
1
);
const
unrecoverable
=
err
.
name
===
'
UnrecoverableError
'
;
if
(
outOfAttempts
||
unrecoverable
)
{
await
deadLetterQueue
.
add
(
'
dead
'
,
{
originalJobId
:
job
.
id
,
data
:
job
.
data
,
reason
:
err
.
message
,
attempts
:
job
.
attemptsMade
,
failedAt
:
Date
.
now
(),
});
}
});独立的死信队列为你提供了一个统一的位置来进行告警、检查和重放。没有人需要在每个队列的失败集合中逐一搜索。
一个限制：此处理器运行在 worker 进程内部。如果进程在最终失败后立即崩溃，死信写入就不会发生。由于 removeOnFail 的作用，该任务仍会在失败集合中保留 24 小时，因此定时清扫可以补上任何遗漏的内容。
对于重放，添加一个小型管理脚本，从死信队列中读取、修复原因，并以原始 ID 重新添加任务。
将进度发送回客户端
Server-Sent Events 在这里效果很好。它们使用普通 HTTP，浏览器会自动重连，而且你只需要数据单向流动。
app
.
get
(
'
/summaries/:jobId/events
'
,
async
(
req
,
res
)
=>
{
const
{
jobId
}
=
req
.
params
;
res
.
writeHead
(
200
,
{
'
Content-Type
'
:
'
text/event-stream
'
,
'
Cache-Control
'
:
'
no-cache
'
,
Connection
:
'
keep-alive
'
,
});
const
send
=
(
event
,
data
)
=>
res
.
write
(
'
event:
'
+
event
+
'
\n
data:
'
+
JSON
.
stringify
(
data
)
+
'
\n\n
'
);
let
closed
=
false
;
const
onProgress
=
({
jobId
:
id
,
data
})
=>
{
if
(
id
===
jobId
)
send
(
'
progress
'
,
data
);
};
const
onCompleted
=
({
jobId
:
id
})
=>
{
if
(
id
===
jobId
)
{
send
(
'
done
'
,
{
resultUrl
:
'
/summaries/
'
+
jobId
});
cleanup
();
}
};
const
onFailed
=
({
jobId
:
id
,
failedReason
})
=>
{
if
(
id
===
jobId
)
{
send
(
'
failed
'
,
{
reason
:
failedReason
});
cleanup
();
}
};
function
cleanup
()
{
if
(
closed
)
return
;
closed
=
true
;
llmEvents
.
off
(
'
progress
'
,
onProgress
);
llmEvents
.
off
(
'
completed
'
,
onCompleted
);
llmEvents
.
off
(
'
failed
'
,
onFailed
);
res
.
end
();
}
llmEvents
.
on
(
'
progress
'
,
onProgress
);
llmEvents
.
on
(
'
completed
'
,
onCompleted
);
llmEvents
.
on
(
'
failed
'
,
onFailed
);
req
.
on
(
'
close
'
,
cleanup
);
// The job may have finished before the client connected
const
job
=
await
llmQueue
.
getJob
(
jobId
);
const
state
=
job
?
await
job
.
getState
()
:
'
missing
'
;
if
(
state
===
'
completed
'
)
onCompleted
({
jobId
});
else
if
(
state
===
'
failed
'
||
state
===
'
missing
'
)
onFailed
({
jobId
,
failedReason
:
state
});
});人们经常跳过最后的状态检查。快速任务可能在 POST 和浏览器打开流之间就完成了。没有这个检查，客户端会永远等待。
done 事件发送的是一个 URL，而不是结果本身。这样可以将大型输出排除在 Redis 事件负载之外，客户端可以通过正常的认证检查从你的数据库中获取结果。
每个 API 进程只需要一个 QueueEvents 实例。许多打开的流会添加许多监听器，因此请调用 llmEvents.setMaxListeners(0)，或者通过你自己的以任务 ID 为键的查找表将事件发送给客户端。
生产环境检查清单
使用 await worker.close() 处理 SIGTERM，以便在部署前完成正在执行的任务。
将 worker 作为独立于 API 的进程或容器运行，并单独扩展它们。
开启 Redis 持久化。否则，Redis 重启会丢失所有排队的任务。
对死信队列深度和最旧任务的等待时间进行告警，而不仅仅是对错误计数告警。
使 saveSummary 具有幂等性。部分写入后的重试应该覆盖，而不是重复。
在每次模型调用时记录任务 ID、尝试次数和 token 使用量。
何时不需要队列
如果调用通常在一两秒内完成，例如短分类，并且用户无论如何都在屏幕上等待，那么内联流式返回响应会更简单。当调用缓慢、突发发生、重复成本高昂或必须在部署后存活时，请使用队列。
在 Geminate Solutions，我们构建了队列密集型系统，包括一个在生产环境中每分钟处理超过 1000 万次请求的考试平台。大多数问题都来自上述细节，例如重试错误的错误或遗漏快速完成，而不是来自队列库。
转载声明：本文为转载整理，原文版权归原作者及原站点所有。
原文来源：DEV.to
原文作者：Geminate Solutions；如有侵权，请联系管理员删除。
