"""预测路由。 安全改进: - 限流: 每分钟 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 from sqlalchemy.orm import selectinload from src.api.deps import get_predict_rate_limit_remaining, rate_limit_predict, require_admin from src.api.schemas import PredictOut, PredictRequest, PredictionOut from src.db.base import AsyncSession, get_db_read, short_read from src.db.models import Match, Prediction from src.llm.predict import predict_match, PredictResult logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/v1", tags=["predict"]) # P1-D: 全局 LLM 并发限制。与 orchestrator 内的 match 级 Semaphore(8) 并存, # 此处在路由层限制单实例全 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 并发限制下执行预测。""" async with _GLOBAL_LLM_SEMAPHORE: return await predict_match( req.match_id, model=req.model, prompt_version=req.prompt_version, mode=req.mode, ) 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。 公开接口,仅做限流保护(不要求登录)。 P1-async: 立即返回 job_id,预测在后台执行,前端轮询 GET /predict/jobs/{job_id}。 避免多专家预测耗时 60-180s 触发网关超时(Cloudflare 100s → HTTP 524)。 """ # 1. 短 read session: 检查比赛存在性/状态 async with short_read() as session: m = await session.get(Match, req.match_id) if m is None: raise HTTPException(404, "match not found") if m.match_status == "finished": raise HTTPException(400, "该比赛已完赛,不再支持预测") # 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}"} @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)]) async def list_predictions( match_id: int | None = None, limit: int = Query(50, ge=1, le=200), db: AsyncSession = Depends(get_db_read), ): stmt = select(Prediction).options(selectinload(Prediction.match)) if match_id: stmt = stmt.where(Prediction.match_id == match_id) stmt = stmt.order_by(Prediction.created_at.desc()).limit(limit) rows = (await db.execute(stmt)).scalars().all() return [ PredictionOut( id=p.id, match_id=p.match_id, provider=p.provider, model=p.model, prompt_version=p.prompt_version, mode=p.mode or "single", pred_home_goals=p.pred_home_goals, pred_away_goals=p.pred_away_goals, alt_pred_home_goals=p.alt_pred_home_goals, alt_pred_away_goals=p.alt_pred_away_goals, pred_1x2=p.pred_1x2, subjective_confidence=p.subjective_confidence, reasoning=p.reasoning, status=p.status or "success", agent_outputs=p.agent_outputs, agent_weights=p.agent_weights, created_at=p.created_at, actual_home_goals=p.actual_home_goals, actual_away_goals=p.actual_away_goals, settled=p.settled, match=_match_dict(p.match) if p.match else None, ) for p in rows ] def _match_dict(m) -> dict | None: if m is None: return None return { "id": m.id, "league_code": m.league.code if m.league else None, "season": m.season, "home_team": m.home_team.name if m.home_team else "?", "away_team": m.away_team.name if m.away_team else "?", "home_team_zh": m.home_team.name_zh if m.home_team else None, "away_team_zh": m.away_team.name_zh if m.away_team else None, "match_date": m.match_date.isoformat() if m.match_date else None, "match_status": m.match_status, "home_goals": m.home_goals, "away_goals": m.away_goals, "match_stage": m.match_stage, } @router.get("/predictions/{prediction_id}", response_model=PredictionOut, dependencies=[Depends(require_admin)]) async def get_prediction(prediction_id: int, db: AsyncSession = Depends(get_db_read)): p = await db.get(Prediction, prediction_id) if p is None: raise HTTPException(404, "prediction not found") return PredictionOut( id=p.id, match_id=p.match_id, provider=p.provider, model=p.model, prompt_version=p.prompt_version, mode=p.mode or "single", pred_home_goals=p.pred_home_goals, pred_away_goals=p.pred_away_goals, alt_pred_home_goals=p.alt_pred_home_goals, alt_pred_away_goals=p.alt_pred_away_goals, pred_1x2=p.pred_1x2, subjective_confidence=p.subjective_confidence, reasoning=p.reasoning, status=p.status or "success", agent_outputs=p.agent_outputs, agent_weights=p.agent_weights, created_at=p.created_at, actual_home_goals=p.actual_home_goals, actual_away_goals=p.actual_away_goals, settled=p.settled, )