diff --git a/frontend/src/pages/matches/hooks/useMatchPredict.ts b/frontend/src/pages/matches/hooks/useMatchPredict.ts index 798d99a..42735db 100644 --- a/frontend/src/pages/matches/hooks/useMatchPredict.ts +++ b/frontend/src/pages/matches/hooks/useMatchPredict.ts @@ -11,6 +11,8 @@ import { useRef, useState } from 'react' import { http } from '../../../lib/http' import type { Match, Prediction } from '../types' +const sleep = (ms: number) => new Promise(r => setTimeout(r, ms)) + /** 把后端/网络错误翻译成用户可读文案 */ function readablePredictError(e: unknown): string { if (e instanceof Error) { @@ -59,17 +61,43 @@ export function useMatchPredict({ onError }: UseMatchPredictOptions) { onError(null) setPrediction(null) setPredictionFor(m) - // LLM 多专家预测耗时可达数分钟,给足超时(与 nginx 代理 300s 对齐) + + // P1-async: 预测改为异步,POST 立即返回 job_id,轮询结果避免网关超时(Cloudflare 100s → 524) const controller = new AbortController() predictAbort.current = controller - const timer = setTimeout(() => controller.abort(), 300_000) + const overallTimer = setTimeout(() => controller.abort(), 300_000) + try { - const data = await http.post('/predict', { match_id: m.id, mode: 'multi' }, { - timeoutMs: 300_000, - signal: controller.signal, - }) + // 1) 发起预测,拿到 job_id + const started = await http.post<{ job_id: string; poll_url: string }>( + '/predict', + { match_id: m.id, mode: 'multi' }, + { timeoutMs: 10_000, signal: controller.signal }, + ) 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) { if (seq !== predictSeq.current) return onError( @@ -78,7 +106,7 @@ export function useMatchPredict({ onError }: UseMatchPredictOptions) { : readablePredictError(e), ) } finally { - clearTimeout(timer) + clearTimeout(overallTimer) if (seq === predictSeq.current) setPredictingId(null) } } diff --git a/src/api/routes/predict.py b/src/api/routes/predict.py index a599c75..23bbb3f 100644 --- a/src/api/routes/predict.py +++ b/src/api/routes/predict.py @@ -4,11 +4,13 @@ - 限流: 每分钟 10 次 / IP(内存实现) - P1-D: 全局 LLM 并发限制(默认 4),防止过多并发 LLM 调用压垮服务 - DB 连接: 短 session 模式,LLM 调用期间不持有连接 + - P1-async: 预测改为异步(后台任务 + 轮询),避免网关超时(Cloudflare 100s) """ from __future__ import annotations import asyncio import logging +import uuid from fastapi import APIRouter, Depends, HTTPException, Query, Request from sqlalchemy import select @@ -28,6 +30,9 @@ router = APIRouter(prefix="/api/v1", tags=["predict"]) # 此处在路由层限制单实例全 LLM 调用(所有模式汇总),默认 4。 _GLOBAL_LLM_SEMAPHORE = asyncio.Semaphore(4) +# P1-async: 预测任务内存存储(job_id → 结果/异常)。单进程部署足够,无需入库。 +_predict_jobs: dict[str, dict] = {} + async def _predict_with_concurrency(req: PredictRequest) -> PredictResult: """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): - """对一场比赛调 LLM 预测。mode=multi(默认,5专家+终裁)、single 或 baseline。 + """对一场比赛调 LLM 预测(异步)。mode=multi(默认,5专家+终裁)、single 或 baseline。 公开接口,仅做限流保护(不要求登录)。 - - DB 连接优化: - 1. 短 read session 检查比赛存在性/状态 - 2. 释放连接后调用 LLM(可能几十秒) - 3. 短 write session 保存 Prediction + P1-async: 立即返回 job_id,预测在后台执行,前端轮询 GET /predict/jobs/{job_id}。 + 避免多专家预测耗时 60-180s 触发网关超时(Cloudflare 100s → HTTP 524)。 """ - # 1. 短 read session: 检查比赛(连接立即释放) + # 1. 短 read session: 检查比赛存在性/状态 async with short_read() as session: m = await session.get(Match, req.match_id) if m is None: @@ -59,56 +92,21 @@ async def predict(req: PredictRequest, request: Request): if m.match_status == "finished": raise HTTPException(400, "该比赛已完赛,不再支持预测") - # 2. 预测调用(不持有任何 DB 连接,受全局 LLM 并发限制) - try: - result = await _predict_with_concurrency(req) - except ValueError as e: - msg = str(e) - if "已结算" in msg: - 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, "预测失败,请查看服务器日志") + # 2. P1-async: 启动后台任务,立即返回 job_id + job_id = str(uuid.uuid4()) + _predict_jobs[job_id] = {"status": "running"} + asyncio.create_task(_run_predict_async(job_id, req)) + logger.info("predict job started: %s match=%s mode=%s", job_id, req.match_id, req.mode) + return {"job_id": job_id, "status": "running", "poll_url": f"/api/v1/predict/jobs/{job_id}"} - # D2: 三种模式统一返回 PredictResult —— 字段映射单一化,无 dict 分支。 - # P3-2:baseline 已在服务层(predict_baseline)落库并回填真实 prediction_id, - # 路由层不再需要特殊的 _persist_baseline,与 single/multi 路径统一。 - prediction_id = result.prediction_id - # 3. 结果映射(无 DB 访问) - logger.info( - "预测完成 match=%s mode=%s pred=%s:%s (%s)", - req.match_id, req.mode, - result.pred_home_goals, result.pred_away_goals, result.pred_1x2, - ) - - 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("/predict/jobs/{job_id}") +async def get_predict_job(job_id: str): + """P1-async: 轮询预测任务状态。""" + job = _predict_jobs.get(job_id) + if job is None: + raise HTTPException(404, f"预测任务不存在: {job_id}") + return job @router.get("/predictions", response_model=list[PredictionOut], dependencies=[Depends(require_admin)])