shangfangjian
f3160e3062
P2-01: 移除 lifespan create_all,改为仅验证连接
新增 /health/ready 就绪检查
P0-04: LLM 输出严格 Pydantic 校验
- Agent 输出越界/非法 → parse_error
- 预测输出自动修正 1X2 与比分一致性
P0-01: injuries cutoff 修复
- get_injuries_for_match 增加 as_of 参数
- injuries_slice 使用 as_of 过滤 retrieved_at
- 防止回测时未来采集数据泄漏
P1-12: 批量入库优化
- 预加载 teams 到内存 dict
- 预加载 existing matches 到内存 set
- 消灭 N+1 查询
131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
"""LLM 输出严格校验。
|
|
|
|
所有 LLM JSON 输出必须经过 Pydantic 校验 + 语义一致性检查后才能落库。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
from src.db.models import Prediction
|
|
|
|
|
|
class AgentReportSchema(BaseModel):
|
|
"""单个专家 Agent 输出的校验 schema。"""
|
|
|
|
data_sufficiency: str = "medium"
|
|
analysis: str = ""
|
|
home_edge: float | None = Field(None, ge=-1.0, le=1.0)
|
|
confidence: float | None = Field(None, ge=0.0, le=1.0)
|
|
key_evidence: list[str] = Field(default_factory=list)
|
|
exp_home_goals: float | None = Field(None, ge=0.0, le=10.0)
|
|
exp_away_goals: float | None = Field(None, ge=0.0, le=10.0)
|
|
probable_score: str | None = None
|
|
|
|
@field_validator("data_sufficiency")
|
|
@classmethod
|
|
def validate_sufficiency(cls, v: str) -> str:
|
|
allowed = {"high", "medium", "low", "none"}
|
|
return v.lower() if v.lower() in allowed else "medium"
|
|
|
|
@field_validator("key_evidence", mode="before")
|
|
@classmethod
|
|
def normalize_evidence(cls, v) -> list[str]:
|
|
if v is None:
|
|
return []
|
|
if isinstance(v, str):
|
|
return [v]
|
|
if isinstance(v, list):
|
|
return [str(e)[:120] for e in v[:5]]
|
|
return []
|
|
|
|
@field_validator("analysis")
|
|
@classmethod
|
|
def truncate_analysis(cls, v: str) -> str:
|
|
return str(v)[:600]
|
|
|
|
|
|
class PredictionOutputSchema(BaseModel):
|
|
"""最终预测输出的校验 schema。"""
|
|
|
|
pred_home_goals: float = Field(ge=0.0, le=10.0)
|
|
pred_away_goals: float = Field(ge=0.0, le=10.0)
|
|
pred_1x2: str
|
|
confidence: float = Field(ge=0.0, le=1.0)
|
|
reasoning: str = ""
|
|
|
|
@field_validator("pred_1x2")
|
|
@classmethod
|
|
def validate_1x2(cls, v: str) -> str:
|
|
if v not in ("1", "X", "2"):
|
|
raise ValueError(f"pred_1x2 must be '1', 'X', or '2', got '{v}'")
|
|
return v
|
|
|
|
@model_validator(mode="after")
|
|
def check_consistency(self) -> "PredictionOutputSchema":
|
|
"""验证比分与胜平负一致。"""
|
|
expected = _score_to_1x2(self.pred_home_goals, self.pred_away_goals)
|
|
if expected and self.pred_1x2 != expected:
|
|
# 自动修正而非拒绝(LLM 常见小错误)
|
|
self.pred_1x2 = expected
|
|
return self
|
|
|
|
|
|
def _score_to_1x2(home: float, away: float) -> str | None:
|
|
"""从比分推导胜平负。"""
|
|
if home > away:
|
|
return "1"
|
|
if home == away:
|
|
return "X"
|
|
if home < away:
|
|
return "2"
|
|
return None
|
|
|
|
|
|
def validate_agent_output(raw: dict) -> AgentReportSchema:
|
|
"""校验并规范化单个 Agent 输出。"""
|
|
return AgentReportSchema(
|
|
data_sufficiency=raw.get("data_sufficiency", "medium"),
|
|
analysis=raw.get("analysis", ""),
|
|
home_edge=_safe_float(raw.get("home_edge")),
|
|
confidence=_safe_float(raw.get("confidence")),
|
|
key_evidence=raw.get("key_evidence", []),
|
|
exp_home_goals=_safe_float(raw.get("exp_home_goals")),
|
|
exp_away_goals=_safe_float(raw.get("exp_away_goals")),
|
|
probable_score=_format_score(raw.get("probable_score")),
|
|
)
|
|
|
|
|
|
def validate_prediction_output(raw: dict) -> PredictionOutputSchema:
|
|
"""校验最终预测输出。"""
|
|
return PredictionOutputSchema(
|
|
pred_home_goals=float(raw.get("pred_home_goals", 0)),
|
|
pred_away_goals=float(raw.get("pred_away_goals", 0)),
|
|
pred_1x2=raw.get("1x2") or raw.get("pred_1x2", "X"),
|
|
confidence=float(raw.get("confidence", 0.5)),
|
|
reasoning=str(raw.get("reasoning", ""))[:1000],
|
|
)
|
|
|
|
|
|
def _safe_float(v) -> float | None:
|
|
"""安全转 float,失败返回 None。"""
|
|
if v is None:
|
|
return None
|
|
try:
|
|
f = float(v)
|
|
if not (f == f): # NaN check
|
|
return None
|
|
return f
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _format_score(v) -> str | None:
|
|
"""格式化比分输出。"""
|
|
if v is None:
|
|
return None
|
|
if isinstance(v, str):
|
|
return v
|
|
if isinstance(v, dict):
|
|
return f"{v.get('home', '?')}-{v.get('away', '?')}"
|
|
return None
|