116 lines
4.3 KiB
Python
116 lines
4.3 KiB
Python
"""AI-backed sentence explanations for the mobile clients."""
|
|
|
|
from dataclasses import dataclass
|
|
import logging
|
|
from urllib.parse import urlsplit
|
|
|
|
import requests
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AITutorError(RuntimeError):
|
|
"""Raised when the configured AI service cannot answer a request."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AITutor:
|
|
endpoint: str
|
|
api_key: str
|
|
model: str
|
|
timeout_seconds: float
|
|
|
|
@property
|
|
def available(self) -> bool:
|
|
return bool(self.endpoint and self.model)
|
|
|
|
@property
|
|
def request_url(self) -> str:
|
|
"""Resolve either a full chat URL or an OpenAI-compatible base URL."""
|
|
parsed = urlsplit(self.endpoint)
|
|
path = parsed.path.rstrip("/")
|
|
if parsed.query or path.endswith("/chat/completions"):
|
|
return self.endpoint
|
|
if path.endswith("/v1") or path.endswith("/v1/openai"):
|
|
return self.endpoint + "/chat/completions"
|
|
if not path:
|
|
return self.endpoint + "/v1/chat/completions"
|
|
return self.endpoint
|
|
|
|
def explain(
|
|
self,
|
|
sentence_text: str,
|
|
*,
|
|
course_title: str = "",
|
|
sentence_index: int | None = None,
|
|
language: str = "",
|
|
) -> str:
|
|
if not self.available:
|
|
raise AITutorError("AI explanation is not configured.")
|
|
|
|
context = course_title.strip() or "口语训练课程"
|
|
position = f"第 {sentence_index + 1} 句" if sentence_index is not None else "当前句子"
|
|
language_note = f"句子语言：{language.strip()}。" if language.strip() else ""
|
|
system_prompt = (
|
|
"你是一位耐心的外语口语老师。请用简体中文讲解用户提供的影视/课程句子。"
|
|
"先给出自然的中文翻译，再解释关键词、常用搭配、语法要点，并简要说明这句话适合怎样跟读。"
|
|
"回答使用简明清晰的分段短文，总长度不超过 600 字，不要编造视频里没有的信息。"
|
|
)
|
|
user_prompt = (
|
|
f"课程：{context}\n位置：{position}\n{language_note}句子：{sentence_text}"
|
|
)
|
|
headers = {"Content-Type": "application/json"}
|
|
if self.api_key:
|
|
headers["Authorization"] = f"Bearer {self.api_key}"
|
|
try:
|
|
response = requests.post(
|
|
self.request_url,
|
|
headers=headers,
|
|
json={
|
|
"model": self.model,
|
|
"messages": [
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": user_prompt},
|
|
],
|
|
"temperature": 0.3,
|
|
"stream": False,
|
|
},
|
|
timeout=self.timeout_seconds,
|
|
)
|
|
except requests.RequestException as exc:
|
|
logger.exception(
|
|
"AI explanation request failed: endpoint=%s model=%s",
|
|
self.request_url,
|
|
self.model,
|
|
)
|
|
raise AITutorError("AI service request failed.") from exc
|
|
|
|
if response.status_code < 200 or response.status_code >= 300:
|
|
logger.warning(
|
|
"AI explanation returned HTTP %s: endpoint=%s model=%s body=%s",
|
|
response.status_code,
|
|
self.request_url,
|
|
self.model,
|
|
response.text[:1000],
|
|
)
|
|
raise AITutorError(f"AI service returned HTTP {response.status_code}.")
|
|
try:
|
|
body = response.json()
|
|
explanation = body["choices"][0]["message"]["content"]
|
|
except (KeyError, IndexError, TypeError, ValueError) as exc:
|
|
logger.warning(
|
|
"AI explanation returned an unexpected body: endpoint=%s body=%s",
|
|
self.request_url,
|
|
response.text[:1000],
|
|
)
|
|
raise AITutorError("AI service returned an unexpected response.") from exc
|
|
if not isinstance(explanation, str) or not explanation.strip():
|
|
logger.warning(
|
|
"AI explanation returned empty content: endpoint=%s body=%s",
|
|
self.request_url,
|
|
response.text[:1000],
|
|
)
|
|
raise AITutorError("AI service returned an empty explanation.")
|
|
return explanation.strip()
|