fix(P1-async): 预测改为异步 POST+轮询,避免网关超时(Cloudflare 524)

后端:POST /predict 立即返回 job_id,后台 asyncio.create_task 执行;
新增 GET /predict/jobs/{job_id} 轮询状态(running/success/failed)。

前端: useMatchPredict 改为 POST 拿 job_id → 每 3s 轮询直到终态;
整体超时 5 分钟不变。避免多专家预测 60-180s 触发 Cloudflare 100s 超时(HTTP 524)。
This commit is contained in:
shangfangjian
2026-09-22 11:00:40 +08:00
parent 1a9dc63edd
commit 197641b9f7
2 changed files with 90 additions and 64 deletions
@@ -11,6 +11,8 @@ import { useRef, useState } from 'react'
import { http } from '../../../lib/http' import { http } from '../../../lib/http'
import type { Match, Prediction } from '../types' import type { Match, Prediction } from '../types'
const sleep = (ms: number) => new Promise<void>(r => setTimeout(r, ms))
/** 把后端/网络错误翻译成用户可读文案 */ /** 把后端/网络错误翻译成用户可读文案 */
function readablePredictError(e: unknown): string { function readablePredictError(e: unknown): string {
if (e instanceof Error) { if (e instanceof Error) {
@@ -59,17 +61,43 @@ export function useMatchPredict({ onError }: UseMatchPredictOptions) {
onError(null) onError(null)
setPrediction(null) setPrediction(null)
setPredictionFor(m) setPredictionFor(m)
// LLM 多专家预测耗时可达数分钟,给足超时(与 nginx 代理 300s 对齐)
// P1-async: 预测改为异步,POST 立即返回 job_id,轮询结果避免网关超时(Cloudflare 100s → 524)
const controller = new AbortController() const controller = new AbortController()
predictAbort.current = controller predictAbort.current = controller
const timer = setTimeout(() => controller.abort(), 300_000) const overallTimer = setTimeout(() => controller.abort(), 300_000)
try { try {
const data = await http.post<Prediction>('/predict', { match_id: m.id, mode: 'multi' }, { // 1) 发起预测,拿到 job_id
timeoutMs: 300_000, const started = await http.post<{ job_id: string; poll_url: string }>(
signal: controller.signal, '/predict',
}) { match_id: m.id, mode: 'multi' },
{ timeoutMs: 10_000, signal: controller.signal },
)
if (seq !== predictSeq.current) return if (seq !== predictSeq.current) return
setPrediction(data)
// 2) 轮询直到终态(success/failed)或整体超时
const deadline = Date.now() + 300_000
while (Date.now() < deadline) {
if (controller.signal.aborted) throw new DOMException('aborted', 'AbortError')
await sleep(3000)
const job = await http.get<{ status: string; result?: Prediction; error?: string }>(
`/predict/jobs/${started.job_id}`,
{ timeoutMs: 5000, signal: controller.signal },
)
if (seq !== predictSeq.current) return
if (job.status === 'success') {
setPrediction(job.result ?? null)
return
}
if (job.status === 'failed') {
onError(job.error || '预测失败')
return
}
// status === 'running' → 继续轮询
}
// 整体超时
onError('预测超时(5 分钟),请稍后重试')
} catch (e) { } catch (e) {
if (seq !== predictSeq.current) return if (seq !== predictSeq.current) return
onError( onError(
@@ -78,7 +106,7 @@ export function useMatchPredict({ onError }: UseMatchPredictOptions) {
: readablePredictError(e), : readablePredictError(e),
) )
} finally { } finally {
clearTimeout(timer) clearTimeout(overallTimer)
if (seq === predictSeq.current) setPredictingId(null) if (seq === predictSeq.current) setPredictingId(null)
} }
} }
+54 -56
View File
@@ -4,11 +4,13 @@
- 限流: 每分钟 10 次 / IP(内存实现) - 限流: 每分钟 10 次 / IP(内存实现)
- P1-D: 全局 LLM 并发限制(默认 4),防止过多并发 LLM 调用压垮服务 - P1-D: 全局 LLM 并发限制(默认 4),防止过多并发 LLM 调用压垮服务
- DB 连接: 短 session 模式,LLM 调用期间不持有连接 - DB 连接: 短 session 模式,LLM 调用期间不持有连接
- P1-async: 预测改为异步(后台任务 + 轮询),避免网关超时(Cloudflare 100s)
""" """
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import logging import logging
import uuid
from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi import APIRouter, Depends, HTTPException, Query, Request
from sqlalchemy import select from sqlalchemy import select
@@ -28,6 +30,9 @@ router = APIRouter(prefix="/api/v1", tags=["predict"])
# 此处在路由层限制单实例全 LLM 调用(所有模式汇总),默认 4。 # 此处在路由层限制单实例全 LLM 调用(所有模式汇总),默认 4。
_GLOBAL_LLM_SEMAPHORE = asyncio.Semaphore(4) _GLOBAL_LLM_SEMAPHORE = asyncio.Semaphore(4)
# P1-async: 预测任务内存存储(job_id → 结果/异常)。单进程部署足够,无需入库。
_predict_jobs: dict[str, dict] = {}
async def _predict_with_concurrency(req: PredictRequest) -> PredictResult: async def _predict_with_concurrency(req: PredictRequest) -> PredictResult:
"""P1-D: 在全局 LLM 并发限制下执行预测。""" """P1-D: 在全局 LLM 并发限制下执行预测。"""
@@ -40,18 +45,46 @@ async def _predict_with_concurrency(req: PredictRequest) -> PredictResult:
) )
@router.post("/predict", response_model=PredictOut, dependencies=[Depends(rate_limit_predict)]) async def _run_predict_async(job_id: str, req: PredictRequest) -> None:
"""P1-async: 后台执行预测,结果写入 _predict_jobs。"""
try:
result = await _predict_with_concurrency(req)
_predict_jobs[job_id] = {
"status": "success",
"result": {
"prediction_id": result.prediction_id,
"provider": result.provider,
"model": result.model,
"prompt_version": result.prompt_version,
"mode": req.mode,
"pred_home_goals": result.pred_home_goals,
"pred_away_goals": result.pred_away_goals,
"alt_pred_home_goals": result.alt_pred_home_goals,
"alt_pred_away_goals": result.alt_pred_away_goals,
"pred_1x2": result.pred_1x2,
"subjective_confidence": result.subjective_confidence,
"reasoning": result.reasoning,
"status": result.status,
"agent_outputs": result.agent_outputs,
"agent_weights": result.agent_weights,
"context": result.context,
"latency_ms": result.latency_ms,
},
}
except Exception as e:
logger.exception("predict job %s failed", job_id)
_predict_jobs[job_id] = {"status": "failed", "error": str(e)}
@router.post("/predict", dependencies=[Depends(rate_limit_predict)])
async def predict(req: PredictRequest, request: Request): async def predict(req: PredictRequest, request: Request):
"""对一场比赛调 LLM 预测。mode=multi(默认,5专家+终裁)、single 或 baseline。 """对一场比赛调 LLM 预测(异步)。mode=multi(默认,5专家+终裁)、single 或 baseline。
公开接口,仅做限流保护(不要求登录)。 公开接口,仅做限流保护(不要求登录)。
P1-async: 立即返回 job_id,预测在后台执行,前端轮询 GET /predict/jobs/{job_id}
DB 连接优化: 避免多专家预测耗时 60-180s 触发网关超时(Cloudflare 100s → HTTP 524)。
1. 短 read session 检查比赛存在性/状态
2. 释放连接后调用 LLM(可能几十秒)
3. 短 write session 保存 Prediction
""" """
# 1. 短 read session: 检查比赛(连接立即释放) # 1. 短 read session: 检查比赛存在性/状态
async with short_read() as session: async with short_read() as session:
m = await session.get(Match, req.match_id) m = await session.get(Match, req.match_id)
if m is None: if m is None:
@@ -59,56 +92,21 @@ async def predict(req: PredictRequest, request: Request):
if m.match_status == "finished": if m.match_status == "finished":
raise HTTPException(400, "该比赛已完赛,不再支持预测") raise HTTPException(400, "该比赛已完赛,不再支持预测")
# 2. 预测调用(不持有任何 DB 连接,受全局 LLM 并发限制) # 2. P1-async: 启动后台任务,立即返回 job_id
try: job_id = str(uuid.uuid4())
result = await _predict_with_concurrency(req) _predict_jobs[job_id] = {"status": "running"}
except ValueError as e: asyncio.create_task(_run_predict_async(job_id, req))
msg = str(e) logger.info("predict job started: %s match=%s mode=%s", job_id, req.match_id, req.mode)
if "已结算" in msg: return {"job_id": job_id, "status": "running", "poll_url": f"/api/v1/predict/jobs/{job_id}"}
raise HTTPException(409, msg)
logger.warning("predict validation error: %s", e)
raise HTTPException(404, "比赛不存在")
except RuntimeError as e:
logger.error("predict LLM error: %s", e)
raise HTTPException(502, "LLM 预测失败,请查看服务器日志")
except Exception as e:
logger.exception("predict unexpected error")
raise HTTPException(500, "预测失败,请查看服务器日志")
# D2: 三种模式统一返回 PredictResult —— 字段映射单一化,无 dict 分支。
# P3-2:baseline 已在服务层(predict_baseline)落库并回填真实 prediction_id,
# 路由层不再需要特殊的 _persist_baseline,与 single/multi 路径统一。
prediction_id = result.prediction_id
# 3. 结果映射(无 DB 访问) @router.get("/predict/jobs/{job_id}")
logger.info( async def get_predict_job(job_id: str):
"预测完成 match=%s mode=%s pred=%s:%s (%s)", """P1-async: 轮询预测任务状态。"""
req.match_id, req.mode, job = _predict_jobs.get(job_id)
result.pred_home_goals, result.pred_away_goals, result.pred_1x2, if job is None:
) raise HTTPException(404, f"预测任务不存在: {job_id}")
return job
return PredictOut(
prediction_id=prediction_id,
provider=result.provider,
model=result.model,
prompt_version=result.prompt_version,
mode=req.mode,
pred_home_goals=result.pred_home_goals,
pred_away_goals=result.pred_away_goals,
alt_pred_home_goals=result.alt_pred_home_goals,
alt_pred_away_goals=result.alt_pred_away_goals,
pred_1x2=result.pred_1x2,
subjective_confidence=result.subjective_confidence,
reasoning=result.reasoning,
status=result.status,
agent_outputs=result.agent_outputs,
agent_weights=result.agent_weights,
context=result.context,
latency_ms=result.latency_ms,
prompt_tokens=result.prompt_tokens,
completion_tokens=result.completion_tokens,
rate_limit_remaining=get_predict_rate_limit_remaining(request),
)
@router.get("/predictions", response_model=list[PredictionOut], dependencies=[Depends(require_admin)]) @router.get("/predictions", response_model=list[PredictionOut], dependencies=[Depends(require_admin)])