724 lines
28 KiB
Python
724 lines
28 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
OCS × AI 中间件 —— AI 客户端（DeepSeek 视觉版）
|
||
|
||
图片链路（两条，自动选择）：
|
||
1. OCS 5.0+「图片题优化」：OCS 在浏览器里把图片转成 base64 数组，随请求体
|
||
`images` 一起传过来，题目/选项里的图片被替换成 [图片N] 标记
|
||
→ 中间件直接把 base64 图片块按顺序拼进多模态消息（无需下载，免疫防盗链）
|
||
2. 旧版 OCS / 关闭图片题优化时：图片以 URL 文本内嵌在 title / options 里
|
||
→ 中间件自己抽 URL、带 Referer 下载、转 base64 后发给视觉模型
|
||
|
||
DeepSeek 模型说明（2026-10）：
|
||
deepseek-flash 支持图片输入（唯一支持视觉的模型，V4.1-Flash）
|
||
deepseek-v4-pro 只支持文字，传图片会被忽略，图片题会答非所问
|
||
|
||
OCR 方案（UniRec / onnxruntime）已移除，不再需要本地 OCR 模型。
|
||
"""
|
||
import asyncio
|
||
import base64
|
||
import html
|
||
import logging
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Optional
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger("ocs_ai")
|
||
|
||
ANSWER_SEPARATOR = "#"
|
||
LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
|
||
|
||
# 请求图片时带上 Referer / UA：超星等图床对防盗链时也能正常取图
|
||
_IMG_HEADERS = {
|
||
"Referer": "https://mooc1.chaoxing.com/",
|
||
"User-Agent": (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
|
||
),
|
||
"Accept": "image/avif,image/webp,image/png,image/jpeg,image/gif,*/*;q=0.8",
|
||
}
|
||
|
||
# DeepSeek 视觉接口支持的图片格式
|
||
_IMAGE_EXTS = (".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".svg")
|
||
# 没有扩展名也当图片尝试的图床域名关键字
|
||
_IMAGE_HOST_HINTS = (
|
||
"ananas.chaoxing.com",
|
||
"chaoxing.com",
|
||
"zhihuishu.com",
|
||
"icve.com.cn",
|
||
"yuketang.cn",
|
||
"icourse163.org",
|
||
"xuetangx.com",
|
||
"unipus.cn",
|
||
)
|
||
|
||
_URL_PATTERN = r"""https?://[^\s"'<>()（）\[\]{}，。；、！？“”‘’]+"""
|
||
# 按出现顺序切分文本：<img> / markdown 图片 / 裸链接
|
||
_URL_JOINED_RE = re.compile(
|
||
r"""<img[^>]*?\bsrc\s*=\s*["']?(?P<img>[^"'\s>]+)[^>]*>"""
|
||
r"""|!\[[^\]]*\]\(\s*(?P<md>[^)\s]+)\s*\)"""
|
||
r"""|(?P<bare>""" + _URL_PATTERN + r""")""",
|
||
re.I,
|
||
)
|
||
|
||
# 单张图片上限（DeepSeek 接口限制 32MB，这里留点余量）
|
||
_MAX_IMAGE_BYTES = 24 * 1024 * 1024
|
||
# base64 编码后的图片总量上限（请求体限制 48MiB）
|
||
_MAX_TOTAL_ENCODED_BYTES = 40 * 1024 * 1024
|
||
_DOWNLOAD_CONCURRENCY = 6
|
||
|
||
# OCS 5.0「图片题优化」：图片以 base64 数组上传，题目文本中的图片被替换成 [图片N] 标记
|
||
_IMAGE_MARK_RE = re.compile(r"\[图片\s*(\d+)\]")
|
||
_MAX_OCS_IMAGES = 20
|
||
# OCS 会把这些空值占位符替换成字面量字符串，这里统一当空处理
|
||
_EMPTY_LITERALS = ("", "undefined", "null", "none")
|
||
|
||
|
||
@dataclass
|
||
class QuestionInfo:
|
||
title: str
|
||
type: str
|
||
options: str
|
||
# ── 以下字段来自 OCS 5.0「图片题优化」 ──
|
||
images: list[str] = field(default_factory=list) # base64 data URL 数组，顺序对应 [图片1]、[图片2]…
|
||
suggestion_title: str = "" # 标题里图片 URL 已替换成 [图片N]
|
||
suggestion_options: str = "" # 选项里图片 URL 已替换成 [图片N]
|
||
|
||
|
||
@dataclass
|
||
class PreparedImage:
|
||
"""一张题图在本地的准备结果"""
|
||
|
||
url: str
|
||
data_url: Optional[str] = None # 成功下载并编码后的 data URL
|
||
is_image: bool = True # False 表示下载到的其实不是图片（按文字处理）
|
||
|
||
@property
|
||
def ok(self) -> bool:
|
||
return bool(self.data_url)
|
||
|
||
|
||
# ═══════════════ 图片 URL 抽取 / 文本清洗 ═══════════════
|
||
|
||
|
||
def _normalize_url(url: str) -> str:
|
||
url = html.unescape(url or "").strip().strip("\"'`")
|
||
url = url.replace("\\/", "/")
|
||
# 去掉句末常见标点、右括号等（OCS 的图片链接后面常常紧跟中文标点）
|
||
return url.rstrip(".,;:!?)]}>、，。；：！？ ")
|
||
|
||
|
||
def _is_image_url(url: str) -> bool:
|
||
low = (url or "").lower()
|
||
if not low.startswith(("http://", "https://")):
|
||
return False
|
||
path = low.split("#", 1)[0].split("?", 1)[0]
|
||
if path.endswith(_IMAGE_EXTS):
|
||
return True
|
||
m = re.match(r"https?://([^/:]+)", low)
|
||
host = m.group(1) if m else ""
|
||
return any(hint in host for hint in _IMAGE_HOST_HINTS)
|
||
|
||
|
||
def _extract_image_urls(text: str) -> list[str]:
|
||
"""从题目文本中抽取图片 URL（按出现顺序去重）。
|
||
|
||
覆盖 OCS 实际会发来的几种形态：
|
||
* 裸链接： https://p.ananas.chaoxing.com/star3/origin/xx.png
|
||
* HTML 图片： <img src="https://.../xx.png">
|
||
* markdown： 
|
||
* 图床无扩展名：https://p.ananas.chaoxing.com/star3/origin/abc123
|
||
"""
|
||
if not text:
|
||
return []
|
||
text = text.replace("\\/", "/")
|
||
urls: list[str] = []
|
||
for m in _URL_JOINED_RE.finditer(text):
|
||
raw = m.group("img") or m.group("md") or m.group("bare") or ""
|
||
url = _normalize_url(raw)
|
||
if not url.startswith(("http://", "https://")):
|
||
continue
|
||
# <img> / markdown 里的链接默认当图片；裸链接需要用扩展名或图床域名判断
|
||
tagged = bool(m.group("img") or m.group("md"))
|
||
if not tagged and not _is_image_url(url):
|
||
continue
|
||
if url not in urls:
|
||
urls.append(url)
|
||
return urls
|
||
|
||
|
||
def _clean_fragment(text: str) -> str:
|
||
"""清洗文本片段：解码实体、去掉 HTML 标签、保留换行"""
|
||
if not text:
|
||
return ""
|
||
text = html.unescape(text)
|
||
text = re.sub(r"(?i)<br\s*/?>", "\n", text)
|
||
text = re.sub(r"<[^>]+>", "", text)
|
||
text = text.replace("\u200b", "").replace("\xa0", " ")
|
||
text = re.sub(r"[ \t]+", " ", text)
|
||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||
return text.strip()
|
||
|
||
|
||
def _strip_image_markup(text: str) -> str:
|
||
"""纯文字模式：把图片（及图片链接）替换成占位符 [图]"""
|
||
if not text:
|
||
return ""
|
||
text = text.replace("\\/", "/")
|
||
text = re.sub(r"(?is)<img[^>]*>", "[图]", text)
|
||
text = re.sub(r"!\[[^\]]*\]\([^)]*\)", "[图]", text)
|
||
for url in _extract_image_urls(text):
|
||
text = text.replace(url, "[图]")
|
||
text = re.sub(r"(?:\[图\]\s*){2,}", "[图]", text)
|
||
text = re.sub(r"[ \t]+", " ", text)
|
||
return text.strip()
|
||
|
||
|
||
def _iter_segments(text: str, known_urls: set) -> list[tuple[str, str]]:
|
||
"""把文本按图片链接位置切成 [('text', ...), ('image', url), ...]"""
|
||
text = (text or "").replace("\\/", "/")
|
||
segments: list[tuple[str, str]] = []
|
||
pos = 0
|
||
for m in _URL_JOINED_RE.finditer(text):
|
||
raw = m.group("img") or m.group("md") or m.group("bare") or ""
|
||
url = _normalize_url(raw)
|
||
if url not in known_urls:
|
||
continue
|
||
if m.start() > pos:
|
||
segments.append(("text", text[pos:m.start()]))
|
||
segments.append(("image", url))
|
||
pos = m.end()
|
||
if pos < len(text):
|
||
segments.append(("text", text[pos:]))
|
||
return segments
|
||
|
||
|
||
def _parse_option_list(options_raw: str) -> list[str]:
|
||
opts = []
|
||
for line in (options_raw or "").split("\n"):
|
||
line = line.strip()
|
||
if line:
|
||
opts.append(line)
|
||
return opts
|
||
|
||
|
||
def _option_prefix(index: int, option_text: str) -> str:
|
||
"""选项前缀：如果文本里已经带了 "A." 就不要再重复加"""
|
||
letter = LETTERS[index] if index < len(LETTERS) else str(index + 1)
|
||
stripped = (option_text or "").lstrip()
|
||
if re.match(rf"^{letter}\s*[.、．)）:：]", stripped, re.I):
|
||
return ""
|
||
return f"{letter}. "
|
||
|
||
|
||
# ═══════════════ OCS 5.0「图片题优化」payload ═══════════════
|
||
|
||
|
||
def _clean_ocs_text(text) -> str:
|
||
"""OCS 占位符可能被替换成 undefined 字面量，这里统一归一化"""
|
||
if not isinstance(text, str):
|
||
return ""
|
||
text = text.strip()
|
||
if text.lower() in _EMPTY_LITERALS:
|
||
return ""
|
||
return text
|
||
|
||
|
||
def _normalize_ocs_image(value) -> str:
|
||
"""把 OCS 传来的图片项归一化成 data URL；无效项返回空字符串"""
|
||
if not isinstance(value, str):
|
||
return ""
|
||
value = value.strip()
|
||
if not value:
|
||
return ""
|
||
if value.startswith("data:image/"):
|
||
return value
|
||
# OCS 正常都会带 data: 前缀；万一只有裸 base64 就补一个 PNG 头
|
||
probe = value[:200]
|
||
if len(value) > 64 and re.fullmatch(r"[A-Za-z0-9+/=\s]+", probe):
|
||
return "data:image/png;base64," + re.sub(r"\s+", "", value)
|
||
# 个别环境可能给的是图片 URL，交给上面的 URL 下载链路处理
|
||
return ""
|
||
|
||
|
||
def _prepare_ocs_images(raw) -> list[str]:
|
||
"""筛选 / 限流 OCS 传来的 base64 图片数组"""
|
||
if not isinstance(raw, (list, tuple)):
|
||
return []
|
||
images: list[str] = []
|
||
total_encoded = 0
|
||
for item in raw:
|
||
data_url = _normalize_ocs_image(item)
|
||
if not data_url:
|
||
continue
|
||
size = len(data_url)
|
||
if total_encoded + size > _MAX_TOTAL_ENCODED_BYTES:
|
||
logger.warning(f"OCS 图片总量超限，已忽略剩余 {len(raw) - len(images)} 张")
|
||
break
|
||
images.append(data_url)
|
||
total_encoded += size
|
||
if len(images) >= _MAX_OCS_IMAGES:
|
||
logger.warning(f"OCS 图片数量超过 {_MAX_OCS_IMAGES} 张，已截断")
|
||
break
|
||
return images
|
||
|
||
|
||
def _iter_marker_segments(text: str) -> list[tuple[str, int]]:
|
||
"""把含 [图片N] 标记的文本切成 [('text', str) | ('image', N)]"""
|
||
text = text or ""
|
||
segments: list[tuple[str, int]] = []
|
||
pos = 0
|
||
for m in _IMAGE_MARK_RE.finditer(text):
|
||
if m.start() > pos:
|
||
segments.append(("text", text[pos:m.start()]))
|
||
segments.append(("image", int(m.group(1))))
|
||
pos = m.end()
|
||
if pos < len(text):
|
||
segments.append(("text", text[pos:]))
|
||
return segments
|
||
|
||
|
||
# ═══════════════ 图片下载 ═══════════════
|
||
|
||
|
||
def _sniff_image_mime(data: bytes, content_type: str = "") -> str:
|
||
"""按文件头判断图片类型（DeepSeek 只认 JPEG / PNG / GIF / WebP）"""
|
||
if not data:
|
||
return ""
|
||
if data.startswith(b"\x89PNG\r\n\x1a\n"):
|
||
return "image/png"
|
||
if data.startswith(b"\xff\xd8\xff"):
|
||
return "image/jpeg"
|
||
if data[:6] in (b"GIF87a", b"GIF89a"):
|
||
return "image/gif"
|
||
if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
|
||
return "image/webp"
|
||
ct = (content_type or "").split(";")[0].strip().lower()
|
||
if ct in ("image/jpeg", "image/jpg"):
|
||
return "image/jpeg"
|
||
if ct in ("image/png", "image/gif", "image/webp"):
|
||
return ct
|
||
return ""
|
||
|
||
|
||
async def _download_image(url: str, timeout: float) -> PreparedImage:
|
||
"""下载一张图片并编码成 data URL；失败时返回标记，由上层决定回退策略"""
|
||
try:
|
||
async with httpx.AsyncClient(
|
||
timeout=min(max(timeout, 5.0), 25.0),
|
||
headers=_IMG_HEADERS,
|
||
follow_redirects=True,
|
||
) as client:
|
||
resp = await client.get(url)
|
||
if resp.status_code != 200:
|
||
logger.warning(f"图片下载失败 HTTP {resp.status_code}: {url[:90]}")
|
||
return PreparedImage(url, None, True)
|
||
content = resp.content
|
||
if not content:
|
||
logger.warning(f"图片内容为空: {url[:90]}")
|
||
return PreparedImage(url, None, True)
|
||
mime = _sniff_image_mime(content, resp.headers.get("content-type", ""))
|
||
if not mime:
|
||
logger.info(f"链接不是受支持的图片，按文字处理: {url[:90]}")
|
||
return PreparedImage(url, None, False)
|
||
if len(content) > _MAX_IMAGE_BYTES:
|
||
logger.warning(f"图片过大（{len(content) / 1048576:.1f}MB），已跳过: {url[:90]}")
|
||
return PreparedImage(url, None, True)
|
||
b64 = base64.b64encode(content).decode("ascii")
|
||
logger.info(f"图片就绪 {len(content) // 1024}KB {mime}: {url[:70]}")
|
||
return PreparedImage(url, f"data:{mime};base64,{b64}", True)
|
||
except Exception as e:
|
||
logger.warning(f"图片下载异常: {url[:90]} | {e}")
|
||
return PreparedImage(url, None, True)
|
||
|
||
|
||
# ═══════════════ 视觉消息 ═══════════════
|
||
|
||
|
||
def _vision_messages(question: QuestionInfo, images: dict[str, PreparedImage]) -> list[dict]:
|
||
"""构建 OpenAI 兼容的多模态消息：文字 + 图片内容块，保持题目/选项的原始顺序"""
|
||
blocks: list[dict] = []
|
||
known = set(images.keys())
|
||
seen_images: set[str] = set()
|
||
|
||
def push_text(text: str) -> None:
|
||
text = _clean_fragment(text)
|
||
if not text:
|
||
return
|
||
if blocks and blocks[-1].get("type") == "text":
|
||
blocks[-1]["text"] = (blocks[-1]["text"] + " " + text).strip()
|
||
else:
|
||
blocks.append({"type": "text", "text": text})
|
||
|
||
def push_image(url: str) -> None:
|
||
if url in seen_images:
|
||
push_text("[同上图]")
|
||
return
|
||
seen_images.add(url)
|
||
info = images.get(url)
|
||
if info is not None and not info.is_image:
|
||
# 服务器返回的不是图片（比如防盗链拦截页），原样保留链接当文字
|
||
push_text(url)
|
||
return
|
||
if info is None or not info.data_url:
|
||
# 注意：不要把原链接直接交给 DeepSeek 抓取——抓取失败会导致整个请求 400
|
||
push_text("[图片加载失败]")
|
||
return
|
||
blocks.append(
|
||
{"type": "image_url", "image_url": {"url": info.data_url, "detail": "high"}}
|
||
)
|
||
|
||
push_text("题目：")
|
||
for kind, value in _iter_segments(question.title, known):
|
||
if kind == "text":
|
||
push_text(value)
|
||
else:
|
||
push_image(value)
|
||
|
||
options = _parse_option_list(question.options)[:26]
|
||
if options:
|
||
push_text("选项：")
|
||
for i, opt in enumerate(options):
|
||
segments = _iter_segments(opt, known)
|
||
if not segments:
|
||
push_text(f"{LETTERS[i]}. {_clean_fragment(opt)}")
|
||
continue
|
||
first = True
|
||
for kind, value in segments:
|
||
if kind == "text":
|
||
text = _clean_fragment(value)
|
||
if not text:
|
||
continue
|
||
prefix = _option_prefix(i, opt) if first else ""
|
||
push_text(f"{prefix}{text}")
|
||
first = False
|
||
else:
|
||
if first:
|
||
push_text(f"{LETTERS[i]}.")
|
||
first = False
|
||
push_image(value)
|
||
|
||
qtype = (question.type or "").strip()
|
||
push_text(_answer_instruction(qtype))
|
||
|
||
system = (
|
||
"你是一个精确的在线课程答题助手。你能够识别图片中的文字、数学公式、图表和选项内容，"
|
||
"并按要求的格式返回答案。"
|
||
)
|
||
return [
|
||
{"role": "system", "content": system},
|
||
{"role": "user", "content": blocks},
|
||
]
|
||
|
||
|
||
def _answer_instruction(qtype: str) -> str:
|
||
"""按题型生成最终作答要求（两种视觉消息构建方式共用）"""
|
||
qtype = (qtype or "").strip()
|
||
if qtype == "single":
|
||
return (
|
||
"请仔细阅读上面的题目和选项（含图片中的公式、图表、文字），"
|
||
"只回答一个正确选项的字母（A/B/C/D），不要解释。"
|
||
)
|
||
if qtype == "multiple":
|
||
return (
|
||
"请仔细阅读上面的题目和选项（含图片中的公式、图表、文字），"
|
||
f"回答所有正确选项的字母，用 {ANSWER_SEPARATOR} 分隔（例如 A{ANSWER_SEPARATOR}C），不要解释。"
|
||
)
|
||
if qtype == "judgement":
|
||
return "请根据题目（含图片中的内容）判断该说法正确还是错误，只回答「正确」或「错误」，不要解释。"
|
||
if qtype == "completion":
|
||
return "请根据题目（含图片中的内容）填空，直接给出答案，多个空用 # 分隔，不要解释。"
|
||
return "请根据题目（含图片中的内容）只回答正确答案，不要解释。"
|
||
|
||
|
||
def _ocs_vision_messages(question: QuestionInfo, images: list[str]) -> list[dict]:
|
||
"""OCS 5.0「图片题优化」消息：文本里的 [图片N] 标记原位替换成第 N 张 base64 图片"""
|
||
blocks: list[dict] = []
|
||
used_images: set[int] = set()
|
||
|
||
def push_text(text: str) -> None:
|
||
text = _clean_fragment(text)
|
||
if not text:
|
||
return
|
||
if blocks and blocks[-1].get("type") == "text":
|
||
blocks[-1]["text"] = (blocks[-1]["text"] + " " + text).strip()
|
||
else:
|
||
blocks.append({"type": "text", "text": text})
|
||
|
||
def push_image(index: int) -> None:
|
||
if 1 <= index <= len(images):
|
||
used_images.add(index)
|
||
blocks.append(
|
||
{"type": "image_url", "image_url": {"url": images[index - 1], "detail": "high"}}
|
||
)
|
||
else:
|
||
push_text(f"[图片{index}]")
|
||
|
||
def push_rich_text(text: str) -> None:
|
||
for kind, value in _iter_marker_segments(text):
|
||
if kind == "text":
|
||
push_text(value)
|
||
else:
|
||
push_image(value)
|
||
|
||
title_src = _clean_ocs_text(question.suggestion_title) or _strip_image_markup(question.title)
|
||
push_text("题目：")
|
||
push_rich_text(title_src)
|
||
|
||
options_src = _clean_ocs_text(question.suggestion_options) or question.options
|
||
options = _parse_option_list(options_src)[:26]
|
||
if options:
|
||
push_text("选项：")
|
||
for i, opt in enumerate(options):
|
||
segments = _iter_marker_segments(opt)
|
||
if not segments:
|
||
push_text(f"{LETTERS[i]}. {_clean_fragment(opt)}")
|
||
continue
|
||
first = True
|
||
for kind, value in segments:
|
||
if kind == "text":
|
||
text = _clean_fragment(value)
|
||
if not text:
|
||
continue
|
||
prefix = _option_prefix(i, opt) if first else ""
|
||
push_text(f"{prefix}{text}")
|
||
first = False
|
||
else:
|
||
if first:
|
||
push_text(f"{LETTERS[i]}.")
|
||
first = False
|
||
push_image(value)
|
||
|
||
# 兜底：图片数组里的图没有被任何 [图片N] 标记用上（例如题库配置只引用了 ${images}
|
||
# 而没引用 ${suggestion_title}/${suggestion_options}），也要把图片交给模型，不能静默丢弃
|
||
unused = [idx for idx in range(1, len(images) + 1) if idx not in used_images]
|
||
if unused:
|
||
push_text("其余图片：" if used_images else "题目图片：")
|
||
for idx in unused:
|
||
push_image(idx)
|
||
|
||
push_text(_answer_instruction(question.type))
|
||
|
||
system = (
|
||
"你是一个精确的在线课程答题助手。你能够识别图片中的文字、数学公式、图表和选项内容，"
|
||
"并按要求的格式返回答案。"
|
||
)
|
||
return [
|
||
{"role": "system", "content": system},
|
||
{"role": "user", "content": blocks},
|
||
]
|
||
|
||
|
||
# ═══════════════ 纯文字消息（无图 / 未配置视觉模型时） ═══════════════
|
||
|
||
|
||
def _build_text_prompt(question: QuestionInfo, has_images: bool = False) -> str:
|
||
title_clean = _strip_image_markup(question.title)
|
||
options_list = _parse_option_list(question.options)[:26]
|
||
lines = []
|
||
for i, opt in enumerate(options_list):
|
||
clean = _strip_image_markup(opt)
|
||
prefix = _option_prefix(i, opt)
|
||
lines.append(f"{prefix}{clean}" if clean else f"{LETTERS[i]}. [图]")
|
||
options_text = "\n".join(lines)
|
||
if not options_text:
|
||
options_text = "（无）"
|
||
|
||
qtype = (question.type or "").strip()
|
||
if qtype == "single":
|
||
prompt = f"题目：{title_clean}\n选项：\n{options_text}\n\n只回答一个正确选项的字母，务必给出答案。"
|
||
elif qtype == "multiple":
|
||
prompt = (
|
||
f"题目：{title_clean}\n选项：\n{options_text}\n\n"
|
||
f"回答所有正确选项的字母，用{ANSWER_SEPARATOR}分隔，务必给出答案。"
|
||
)
|
||
elif qtype == "judgement":
|
||
prompt = f"判断以下说法，回答「正确」或「错误」，务必给出答案。\n\n{title_clean}"
|
||
elif qtype == "completion":
|
||
prompt = (
|
||
"你是一个答题助手。回答以下填空题，先给出最终答案，然后可以简要说明。\n\n"
|
||
f"{title_clean}\n\n答案："
|
||
)
|
||
else:
|
||
prompt = f"题目：{title_clean}\n选项：\n{options_text}\n\n只回答正确答案，不要解释。"
|
||
|
||
if has_images:
|
||
prompt += "\n\n（注意：题目中含图片，但当前未启用视觉识别，请尽量根据文字信息作答。）"
|
||
return prompt
|
||
|
||
|
||
def _text_messages(question: QuestionInfo, has_images: bool = False) -> list[dict]:
|
||
if question.type == "completion":
|
||
system = "你是一个精确的在线课程答题助手。请按要求格式返回答案。"
|
||
else:
|
||
system = (
|
||
"你是一个精确的在线课程答题助手。请按要求格式返回答案。"
|
||
"即使信息不完整，也请选择最可能的选项。"
|
||
)
|
||
return [
|
||
{"role": "system", "content": system},
|
||
{"role": "user", "content": _build_text_prompt(question, has_images=has_images)},
|
||
]
|
||
|
||
|
||
# ═══════════════ AIClient ═══════════════
|
||
|
||
|
||
class AIClient:
|
||
def __init__(
|
||
self,
|
||
api_key: str,
|
||
api_base: str = "https://api.openai.com/v1",
|
||
model: str = "deepseek-v4-pro",
|
||
vision_model: str = "",
|
||
temperature: float = 0.1,
|
||
max_tokens: int = 500,
|
||
timeout: float = 60.0,
|
||
thinking: bool = True,
|
||
provider: str = "deepseek",
|
||
):
|
||
self.api_key = api_key
|
||
self.api_base = api_base.rstrip("/")
|
||
self.model = model
|
||
self.vision_model = (vision_model or "").strip()
|
||
# 兼容旧配置：千问(qwen)用户过去就是直接把图片发给 model，没单独配置视觉模型
|
||
if not self.vision_model and provider == "qwen":
|
||
self.vision_model = model
|
||
self.temperature = temperature
|
||
self.max_tokens = max_tokens
|
||
self.timeout = timeout
|
||
self.thinking = thinking
|
||
self.provider = provider
|
||
|
||
def _headers(self) -> dict:
|
||
return {
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
def _payload(self, messages: list[dict], max_tokens: int, model: str) -> dict:
|
||
payload = {
|
||
"model": model,
|
||
"messages": messages,
|
||
"temperature": self.temperature,
|
||
"max_tokens": max_tokens,
|
||
}
|
||
if self.thinking:
|
||
if self.provider == "qwen":
|
||
payload["chat_template_kwargs"] = {"enable_thinking": True}
|
||
else:
|
||
payload["thinking"] = {"type": "enabled"}
|
||
return payload
|
||
|
||
async def _post(self, payload: dict) -> dict:
|
||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||
response = await client.post(
|
||
f"{self.api_base}/chat/completions",
|
||
json=payload,
|
||
headers=self._headers(),
|
||
)
|
||
if response.status_code != 200:
|
||
raise Exception(
|
||
f"AI API 请求失败: HTTP {response.status_code} - {response.text[:500]}"
|
||
)
|
||
return response.json()
|
||
|
||
async def _chat(self, messages: list[dict], max_tokens: int = 0, model: str = "") -> str:
|
||
model = model or self.model
|
||
max_tokens = max_tokens or self.max_tokens
|
||
data = await self._post(self._payload(messages, max_tokens, model))
|
||
choice = (data.get("choices") or [{}])[0]
|
||
message = choice.get("message") or {}
|
||
content = (message.get("content") or "").strip()
|
||
finish_reason = choice.get("finish_reason")
|
||
|
||
# 思考模式可能把 max_tokens 全用在 reasoning 上，导致 content 为空：
|
||
# 降低思考强度并放大预算重试一次
|
||
if not content and finish_reason == "length":
|
||
logger.warning("AI 返回内容为空（思考被截断），降低思考强度重试...")
|
||
retry_payload = self._payload(messages, min(max_tokens * 2, 8192), model)
|
||
if self.provider == "qwen":
|
||
retry_payload["chat_template_kwargs"] = {"enable_thinking": True}
|
||
else:
|
||
retry_payload["reasoning_effort"] = "low"
|
||
data = await self._post(retry_payload)
|
||
choice = (data.get("choices") or [{}])[0]
|
||
message = choice.get("message") or {}
|
||
content = (message.get("content") or "").strip()
|
||
|
||
reasoning = (message.get("reasoning_content") or "").strip()
|
||
if reasoning:
|
||
logger.debug(f"AI 思考: {reasoning[:300]}")
|
||
return content
|
||
|
||
async def _prepare_images(self, urls: list[str]) -> dict[str, PreparedImage]:
|
||
semaphore = asyncio.Semaphore(_DOWNLOAD_CONCURRENCY)
|
||
|
||
async def one(url: str):
|
||
async with semaphore:
|
||
return await _download_image(url, self.timeout)
|
||
|
||
results = await asyncio.gather(*(one(u) for u in urls))
|
||
prepared: dict[str, PreparedImage] = {}
|
||
total_encoded = 0
|
||
for item in results:
|
||
# 整体请求体限制 48MiB：base64 后过大就舍弃，避免整个请求被拒
|
||
if item.data_url:
|
||
size = len(item.data_url)
|
||
if total_encoded + size > _MAX_TOTAL_ENCODED_BYTES:
|
||
logger.warning(f"图片总量超限，已跳过: {item.url[:80]}")
|
||
item = PreparedImage(item.url, None, True)
|
||
else:
|
||
total_encoded += size
|
||
prepared[item.url] = item
|
||
return prepared
|
||
|
||
def _max_tokens_for(self, question: QuestionInfo) -> int:
|
||
mt = self.max_tokens
|
||
if self.thinking:
|
||
mt = max(mt, 3000)
|
||
elif question.type == "completion":
|
||
mt = max(mt, 1500)
|
||
return mt
|
||
|
||
async def ask(self, question: QuestionInfo) -> Optional[str]:
|
||
image_urls: list[str] = []
|
||
for text in (question.title, question.options):
|
||
for url in _extract_image_urls(text):
|
||
if url not in image_urls:
|
||
image_urls.append(url)
|
||
|
||
# OCS 5.0「图片题优化」：图片已经是 base64 数组，直接用，无需下载
|
||
ocs_images = _prepare_ocs_images(getattr(question, "images", None))
|
||
mt = self._max_tokens_for(question)
|
||
|
||
if ocs_images and self.vision_model:
|
||
logger.info(
|
||
f"视觉答题（OCS 图片题优化）: {len(ocs_images)} 张 base64 图片 → {self.vision_model}"
|
||
)
|
||
messages = _ocs_vision_messages(question, ocs_images)
|
||
content = await self._chat(messages, max_tokens=mt, model=self.vision_model)
|
||
elif image_urls and self.vision_model:
|
||
logger.info(f"视觉答题: {len(image_urls)} 张图片 → {self.vision_model}")
|
||
prepared = await self._prepare_images(image_urls)
|
||
ok = sum(1 for item in prepared.values() if item.ok)
|
||
logger.info(f"图片准备完成: {ok}/{len(image_urls)} 张可用（其余按文字占位处理）")
|
||
messages = _vision_messages(question, prepared)
|
||
content = await self._chat(messages, max_tokens=mt, model=self.vision_model)
|
||
else:
|
||
if image_urls or ocs_images:
|
||
logger.warning(
|
||
f"检测到 {len(ocs_images) or len(image_urls)} 张图片，但未配置 ai.vision_model，"
|
||
"图片将按 [图] 占位处理；请在 config.yaml 中配置 vision_model: deepseek-flash"
|
||
)
|
||
content = await self._chat(
|
||
_text_messages(question, has_images=bool(image_urls or ocs_images)), max_tokens=mt
|
||
)
|
||
|
||
logger.info(f"AI → '{content[:80] if content else '(空)'}'")
|
||
if not content:
|
||
logger.warning(f"AI返回空。题目: {_strip_image_markup(question.title)[:200]}")
|
||
return content if content else None
|