diff --git a/src/api/routes/predict.py b/src/api/routes/predict.py index 3204df3..a599c75 100644 --- a/src/api/routes/predict.py +++ b/src/api/routes/predict.py @@ -2,10 +2,12 @@ 安全改进: - 限流: 每分钟 10 次 / IP(内存实现) + - P1-D: 全局 LLM 并发限制(默认 4),防止过多并发 LLM 调用压垮服务 - DB 连接: 短 session 模式,LLM 调用期间不持有连接 """ from __future__ import annotations +import asyncio import logging from fastapi import APIRouter, Depends, HTTPException, Query, Request @@ -22,6 +24,21 @@ 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) + + +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, + ) + @router.post("/predict", response_model=PredictOut, dependencies=[Depends(rate_limit_predict)]) async def predict(req: PredictRequest, request: Request): @@ -42,14 +59,9 @@ async def predict(req: PredictRequest, request: Request): if m.match_status == "finished": raise HTTPException(400, "该比赛已完赛,不再支持预测") - # 2. 预测调用(不持有任何 DB 连接) + # 2. 预测调用(不持有任何 DB 连接,受全局 LLM 并发限制) try: - result = await predict_match( - req.match_id, - model=req.model, - prompt_version=req.prompt_version, - mode=req.mode, - ) + result = await _predict_with_concurrency(req) except ValueError as e: msg = str(e) if "已结算" in msg: diff --git a/src/api/schemas.py b/src/api/schemas.py index 4ec0e2a..2c55eaf 100644 --- a/src/api/schemas.py +++ b/src/api/schemas.py @@ -36,7 +36,7 @@ class MatchListOut(BaseModel): class PredictRequest(BaseModel): match_id: int - provider: str | None = None + # P1-D: 删除未接线的 provider 字段(符合"名不副实则删除");provider 由服务端配置决定。 model: str | None = None prompt_version: str | None = None mode: str = Field( diff --git a/tests/test_p1_d_concurrency.py b/tests/test_p1_d_concurrency.py new file mode 100644 index 0000000..ebd6500 --- /dev/null +++ b/tests/test_p1_d_concurrency.py @@ -0,0 +1,71 @@ +"""P1-D 回归测试: 全局 LLM 并发限制 + provider 字段已删除。 + +运行: pytest tests/test_p1_d_concurrency.py -v +""" +import asyncio +from unittest.mock import patch + +import pytest + +from src.api.routes import predict as predict_mod + + +class TestGlobalLLMConcurrency: + """P1-D: 全局 LLM 并发限制(默认 4)。""" + + def test_semaphore_exists_with_limit(self): + """路由模块必须存在 _GLOBAL_LLM_SEMAPHORE 且 value <= 4。""" + assert hasattr(predict_mod, "_GLOBAL_LLM_SEMAPHORE") + sem = predict_mod._GLOBAL_LLM_SEMAPHORE + assert isinstance(sem, asyncio.Semaphore) + assert sem._value == 4, f"期望并发限制 4,实际 {sem._value}" + + def test_predict_with_concurrency_limits_parallel(self): + """P1-D: 并发调用 _predict_with_concurrency 不得超过信号量限制。""" + max_concurrent = 0 + current = 0 + lock = asyncio.Lock() + + async def fake_predict(match_id, **kwargs): + nonlocal current, max_concurrent + async with lock: + current += 1 + max_concurrent = max(max_concurrent, current) + await asyncio.sleep(0.05) # 模拟 LLM 调用 + async with lock: + current -= 1 + return type("R", (), {"prediction_id": 1, "provider": "p", "model": "m", + "prompt_version": "v1", "pred_home_goals": 1.0, + "pred_away_goals": 0.0, "pred_1x2": "1", + "subjective_confidence": 0.5, "reasoning": "", + "status": "success", "context": "", + "latency_ms": 0, "raw": {}})() + + req = type("Req", (), {"match_id": 1, "model": None, "prompt_version": None, "mode": "single"})() + + async def run(): + with patch.object(predict_mod, "predict_match", fake_predict): + # 启动 10 个并发请求 + tasks = [predict_mod._predict_with_concurrency(req) for _ in range(10)] + await asyncio.gather(*tasks) + + asyncio.run(run()) + # 最大并发不得超过信号量限制(4) + assert max_concurrent <= 4, f"并发 {max_concurrent} 超过限制 4" + + +class TestProviderFieldRemoved: + """P1-D: PredictRequest 的 provider 字段必须已删除(未接线)。""" + + def test_predict_request_no_provider(self): + from src.api.schemas import PredictRequest + + fields = set(PredictRequest.model_fields.keys()) + assert "provider" not in fields, f"PredictRequest 应已删除 provider 字段,现有: {fields}" + + def test_predict_request_still_has_core_fields(self): + from src.api.schemas import PredictRequest + + fields = set(PredictRequest.model_fields.keys()) + for required in ("match_id", "model", "prompt_version", "mode"): + assert required in fields, f"缺少核心字段 {required}"