153 lines
5.6 KiB
Python
153 lines
5.6 KiB
Python
"""AI 3모델 실연동 — GPT(OpenAI) · Claude(Anthropic) · Gemini(Google).
|
|
|
|
각 모델에 동일한 경기 컨텍스트를 주고 구조화된 예측 JSON 을 받는다.
|
|
키가 없으면 ProviderUnavailable 을 던지고(조용한 실패 0), 워커는 모델별로
|
|
독립 처리하여 가능한 것만 갱신한다.
|
|
|
|
반환 표준 dict:
|
|
{outcome, scoreA, scoreB, confidencePct, reasonKo, reasonEn}
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass
|
|
|
|
from ..config import settings
|
|
|
|
log = logging.getLogger("triplepick.ai")
|
|
|
|
OUTCOMES = {"TEAM_A_WIN", "DRAW", "TEAM_B_WIN"}
|
|
|
|
|
|
class ProviderUnavailable(RuntimeError):
|
|
"""API 키 미설정 등으로 해당 모델을 호출할 수 없음."""
|
|
|
|
|
|
@dataclass
|
|
class MatchContext:
|
|
team_a: str # 표시명 (예: Korea Republic)
|
|
team_b: str
|
|
venue: str
|
|
kickoff: str # ISO
|
|
|
|
|
|
def _prompt(ctx: MatchContext) -> str:
|
|
return (
|
|
f"You are a football match analyst. Predict the result of this match.\n"
|
|
f"Match: {ctx.team_a} (Team A, home) vs {ctx.team_b} (Team B, away)\n"
|
|
f"Venue: {ctx.venue}\nKickoff: {ctx.kickoff}\n\n"
|
|
f"Predict the final regulation-time score. "
|
|
f"Respond with a single JSON object and nothing else, with keys:\n"
|
|
f' "scoreA": integer 0-9 (Team A goals),\n'
|
|
f' "scoreB": integer 0-9 (Team B goals),\n'
|
|
f' "outcome": one of "TEAM_A_WIN" | "DRAW" | "TEAM_B_WIN" (must match the score),\n'
|
|
f' "confidencePct": integer 0-100 (confidence in the predicted winner; '
|
|
f"for a draw, confidence in the draw),\n"
|
|
f' "reasonKo": a short one-line rationale in Korean (max ~30 chars),\n'
|
|
f' "reasonEn": a short one-line rationale in English (max ~60 chars).\n'
|
|
)
|
|
|
|
|
|
# JSON Schema (구조화 출력용 — Anthropic/OpenAI 공통)
|
|
_SCHEMA = {
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"properties": {
|
|
"scoreA": {"type": "integer"},
|
|
"scoreB": {"type": "integer"},
|
|
"outcome": {"type": "string", "enum": ["TEAM_A_WIN", "DRAW", "TEAM_B_WIN"]},
|
|
"confidencePct": {"type": "integer"},
|
|
"reasonKo": {"type": "string"},
|
|
"reasonEn": {"type": "string"},
|
|
},
|
|
"required": ["scoreA", "scoreB", "outcome", "confidencePct", "reasonKo", "reasonEn"],
|
|
}
|
|
|
|
|
|
def _normalize(data: dict) -> dict:
|
|
a = max(0, min(9, int(data["scoreA"])))
|
|
b = max(0, min(9, int(data["scoreB"])))
|
|
# outcome 은 스코어와 일관되도록 서버에서 재도출 (모델 불일치 방지)
|
|
outcome = "TEAM_A_WIN" if a > b else "TEAM_B_WIN" if a < b else "DRAW"
|
|
conf = max(0, min(100, int(data.get("confidencePct", 50))))
|
|
return {
|
|
"outcome": outcome,
|
|
"scoreA": a,
|
|
"scoreB": b,
|
|
"confidencePct": conf,
|
|
"reasonKo": str(data.get("reasonKo", "")).strip()[:120],
|
|
"reasonEn": str(data.get("reasonEn", "")).strip()[:160],
|
|
}
|
|
|
|
|
|
# ── GPT (OpenAI) ────────────────────────────────────────────
|
|
async def predict_gpt(ctx: MatchContext) -> dict:
|
|
if not settings.openai_api_key:
|
|
raise ProviderUnavailable("OPENAI_API_KEY 미설정")
|
|
from openai import AsyncOpenAI
|
|
|
|
client = AsyncOpenAI(api_key=settings.openai_api_key)
|
|
resp = await client.chat.completions.create(
|
|
model=settings.openai_model,
|
|
messages=[
|
|
{"role": "system", "content": "You output only valid JSON."},
|
|
{"role": "user", "content": _prompt(ctx)},
|
|
],
|
|
response_format={"type": "json_object"},
|
|
)
|
|
content = resp.choices[0].message.content or "{}"
|
|
return _normalize(json.loads(content))
|
|
|
|
|
|
# ── Claude (Anthropic) ──────────────────────────────────────
|
|
async def predict_claude(ctx: MatchContext) -> dict:
|
|
if not settings.anthropic_api_key:
|
|
raise ProviderUnavailable("ANTHROPIC_API_KEY 미설정")
|
|
from anthropic import AsyncAnthropic
|
|
|
|
client = AsyncAnthropic(api_key=settings.anthropic_api_key)
|
|
|
|
async def _create(**extra): # noqa: ANN003
|
|
return await client.messages.create(
|
|
model=settings.anthropic_model,
|
|
max_tokens=1024,
|
|
messages=[{"role": "user", "content": _prompt(ctx)}],
|
|
**extra,
|
|
)
|
|
|
|
# 1순위: 구조화 출력(output_config.format) + Opus 4.8 어댑티브 thinking.
|
|
# SDK/모델 버전에 따라 미지원이면 평문 JSON 파싱으로 폴백.
|
|
try:
|
|
msg = await _create(
|
|
thinking={"type": "adaptive"},
|
|
output_config={"format": {"type": "json_schema", "schema": _SCHEMA}},
|
|
)
|
|
except TypeError:
|
|
msg = await _create()
|
|
text = "".join(b.text for b in msg.content if getattr(b, "type", "") == "text")
|
|
return _normalize(json.loads(text))
|
|
|
|
|
|
# ── Gemini (Google) ─────────────────────────────────────────
|
|
async def predict_gemini(ctx: MatchContext) -> dict:
|
|
if not settings.google_api_key:
|
|
raise ProviderUnavailable("GOOGLE_API_KEY 미설정")
|
|
from google import genai
|
|
from google.genai import types
|
|
|
|
client = genai.Client(api_key=settings.google_api_key)
|
|
resp = await client.aio.models.generate_content(
|
|
model=settings.google_model,
|
|
contents=_prompt(ctx),
|
|
config=types.GenerateContentConfig(response_mime_type="application/json"),
|
|
)
|
|
return _normalize(json.loads(resp.text or "{}"))
|
|
|
|
|
|
PROVIDERS = {
|
|
"GPT": predict_gpt,
|
|
"Claude": predict_claude,
|
|
"Gemini": predict_gemini,
|
|
}
|