给本地GPU大模型加PostgreSQL排队网关

原文要点

该n8n模板把本地OpenAI兼容模型包装成持久FIFO网关：POST /v1/chat/completions用Header Auth收请求，新任务写入PostgreSQL的llm_queue_jobs表；调度器每5秒回收过期租约，在咨询事务锁下取最老任务逐条推理，默认最多3次尝试，408/425/429/5xx指数退避重试，租约两小时。支持/result、/wait、/cancel、/queue命令、GET /v1/queue/tasks/:taskId状态查询和GET /v1/models。stream:true时返回SSE分块回放而非实时逐token流；支持带shell工具的AI harness自动轮询等待，依赖curl+jq。

这次做成什么

在本站设计的练习中，把一个本地GPU模型（如Ollama或vLLM的OpenAI兼容端点）接入该n8n队列网关，完成建表、提交一个排队请求、轮询取结果、取消任务的全流程验证，理解其异步语义。

准备什么

• 已部署n8n实例并能创建PostgreSQL凭据
• 一个可用的PostgreSQL数据库，具备建表权限
• 本地GPU推理服务提供OpenAI兼容的/chat/completions端点
• 三台Webhook统一配置的Header Auth密钥及TLS入口
• 测试机上安装curl与jq用于发请求和解析

具体怎么做

1. 建库建表

   在n8n中新建PostgreSQL凭据并指向你的数据库，将模板内所有Postgres节点绑定该凭据，然后手动触发一次模板自带的setup触发器，创建llm_queue_jobs表及索引。若不便导入模板，可用下方SQL材料手工建结构相同的队列表作练习。

2. 配置网关与调度参数

   在网关配置节点设置max attempts（默认3）与wait/poll上限（默认5秒间隔、720次约一小时）；在调度器配置节点填入上游chat completions URL（如http://127.0.0.1:11434/v1/chat/completions）、超时、租约分钟数、重试基数秒；若上游需要鉴权，在HTTP Request节点加相应Header。/v1/models返回的模型ID用逗号分隔配置，例如queued-local-model。

3. 配置三个Webhook鉴权

   为POST /v1/chat/completions、GET /v1/queue/tasks/:taskId、GET /v1/models三个Webhook统一配置同一个Header Auth凭据，并确认n8n对外经TLS暴露。激活工作流后复制三个Webhook的完整URL，拼成客户端的base URL。

4. 提交排队请求并确认入队

   用curl以Authorization: Bearer头POST一个OpenAI格式请求体（model、messages、stream:false）。正常会立即收到"Your request is queued as task_...."的确认，而不是模型回答——这是该网关的核心异步语义。

5. 查询状态与取结果

   调度器每5秒跑一轮，持锁取出最老排队任务调上游并持久化结果。用GET /v1/queue/tasks/{taskId}查询状态与队列位置；完成后POST一条内容为"/result {taskId}"的chat请求即可拿回持久化的OpenAI格式响应，不会触发第二次推理。排队中的任务可用"/cancel {taskId}"取消，已运行任务无法在上游侧取消。

6. 验证SSE回放与错误路径

   把stream改为true重发请求，观察Content-Type变为text/event-stream且以data: [DONE]结尾——注意这只是把排队确认/已存结果转成分块回放，不是GPU逐token实时流。再验证错误路径：临时把上游指向一个会返回5xx的地址提交任务，观察按指数退避重试后任务状态里出现错误详情；若是连接不上这类网络级失败，则不会立即重试，要等租约过期后才被回收重排队。

可直接复制的材料

CREATE TABLE IF NOT EXISTS llm_queue_jobs (
  id BIGSERIAL PRIMARY KEY,
  task_id TEXT NOT NULL UNIQUE,
  status TEXT NOT NULL DEFAULT 'queued',
  request_body JSONB NOT NULL,
  result_body JSONB,
  error_detail TEXT,
  attempts INTEGER NOT NULL DEFAULT 0,
  max_attempts INTEGER NOT NULL DEFAULT 3,
  lease_expires_at TIMESTAMPTZ,
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_llm_queue_jobs_fifo ON llm_queue_jobs (created_at, id);
CREATE INDEX IF NOT EXISTS idx_llm_queue_jobs_status ON llm_queue_jobs (status);

{
  "gateway": {
    "maxAttempts": 3,
    "waitIntervalSeconds": 5,
    "waitMaxAttempts": 720,
    "modelIds": "queued-local-model"
  },
  "dispatcher": {
    "upstreamChatCompletionsUrl": "http://127.0.0.1:11434/v1/chat/completions",
    "httpTimeoutSeconds": 300,
    "leaseMinutes": 120,
    "retryBaseSeconds": 10
  }
}

OPENAI_BASE_URL="https://n8n.example.com/webhook/xxxx/v1"
OPENAI_API_KEY="test-only-shared-header-secret"

curl -sS "$OPENAI_BASE_URL/chat/completions" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{"model":"queued-local-model","messages":[{"role":"user","content":"用两句话解释消息队列的FIFO语义。"}],"stream":false,"max_tokens":200}'

curl -sS "$OPENAI_BASE_URL/queue/tasks/task_REPLACE_ME" \
  -H "Authorization: Bearer $OPENAI_API_KEY"

curl -sS "$OPENAI_BASE_URL/chat/completions" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{"model":"queued-local-model","messages":[{"role":"user","content":"/result task_REPLACE_ME"}]}'

输入与输出示例

以上为本站虚构的端到端示意，数字与文本均为练习设定，并非实际调用模板得到的结果。

输入

POST /v1/chat/completions，请求体：{"model":"queued-local-model","messages":[{"role":"user","content":"用两句话解释消息队列的FIFO语义。"}],"stream":false,"max_tokens":200}

对应输出

第一阶段立即返回："Your request is queued as task_9f2c1a. Check it with /result task_9f2c1a."；待调度器后续轮次完成推理后，GET /v1/queue/tasks/task_9f2c1a返回{"task_id":"task_9f2c1a","status":"completed","queue_position":0,"error":null}；随后发送"/result task_9f2c1a"返回{"id":"chatcmpl-queued","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"FIFO队列按先入先出顺序处理消息……"},"finish_reason":"stop"}]}

做完检查这些

• 提交chat请求后立即收到task_开头的排队确认而非模型正文
• GET /v1/queue/tasks/:taskId能返回status与queue_position字段
• /result命令取回的完成响应是标准OpenAI chat.completion结构
• 并发提交两任务时按created_at,id顺序逐条执行而非并行
• stream:true响应以text/event-stream返回且以data: [DONE]结束

容易卡住的地方

• 网关是异步语义：首个completion可能只是排队确认或合成tool call，不能当普通同步API用
• SSE只是协议兼容的回放，不实时转发GPU token；需要长连接实时流的客户端须另建转发层
• 只有一个共享Header Auth凭据和一个共享队列，没有租户隔离
• 上游HTTP节点网络级失败要等租约过期（模板默认两小时，可在调度器配置中调整）才回收重排队，不会立刻重试
• 自动等待模式要求客户端具备shell工具且环境装有curl与jq，否则退回人工命令模式
• 模板不会自动清理历史prompt与结果，需自行定期归档llm_queue_jobs