From b413104587276656bea5bfcdd786f77ed5bc133e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BB=BB=E6=B1=89=E7=86=99?= <3218485270@qq.com> Date: Thu, 3 Sep 2026 11:44:10 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=BB=93=E6=9E=84=E5=8C=96=E8=BE=93?= =?UTF-8?q?=E5=87=BA=E6=8C=89=E5=AE=98=E6=96=B9=E8=A7=84=E8=8C=83=20+=20?= =?UTF-8?q?=E4=B8=A4=E6=AE=B5=E5=BC=8F=E9=87=8D=E8=AF=95=20+=20DeepSeek=20?= =?UTF-8?q?=E5=A4=8D=E6=A3=80=E5=BC=80=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 豆包: response_format.json_schema(strict)；DeepSeek: 切 Responses API text.format.json_schema （官方 chat/completions 通道不支持 json_schema），不支持时自动降级 json_object（[output] schema） - 重试两段式：传输错误 cfg.retries 次；解析失败/空正文有独立 3 次专用重试，仍失败落 parse_fail - 移除三票复核（decide_final/_verify_clean/verify_clean），DeepSeek 单次判定即终稿 - 新增 [run] deepseek_recheck 开关（默认 no=仅豆包初筛；yes=追加 DeepSeek 复检） - GUI 不再覆盖 config 提示词；prompt 相对/前导斜杠路径按 exe 目录解析 - 报表去「票型/复核」列并同步说明；README/config.example.ini 同步 - 测试新增输出格式、重试、开关与提示词路径用例（50 passed） --- README.md | 32 +- config.example.ini | 12 +- src/violation_detector/cli.py | 13 +- src/violation_detector/config.py | 64 +++- src/violation_detector/gui.py | 27 +- src/violation_detector/pipeline.py | 96 +---- src/violation_detector/providers.py | 527 +++++++++++++++++++++++----- src/violation_detector/report.py | 20 +- tests/test_config.py | 81 ++++- tests/test_pipeline.py | 37 +- tests/test_providers.py | 204 ++++++++++- tests/test_report.py | 8 +- 12 files changed, 832 insertions(+), 289 deletions(-) diff --git a/README.md b/README.md index c78dce6..4d3509b 100644 --- a/README.md +++ b/README.md @@ -4,10 +4,17 @@ ``` 全部图片 ──▶ 豆包（火山方舟 Ark）初筛 ──┬─ 判为违规 ──▶ 直接定稿（便宜，拦下大头） - └─ 无违规/违规不明/异常 ──▶ DeepSeek 复检 - └─ 仍判无违规 ──▶ 追加两票复核（3票取多数） + └─ 无违规/违规不明/异常 ──▶ DeepSeek 复检（可开关） + （单次判定即终稿，已移除三票复核） ``` +DeepSeek 复检默认关闭（`[run] deepseek_recheck=no`），即默认只跑豆包初筛； +需要第二级复检时把该开关设为 `yes`（或 `mode=deepseek` 全量走 DeepSeek）。 + +AI 输出启用官方结构化输出：豆包走 `response_format.json_schema`（strict），DeepSeek 走 +Responses API 的 `text.format.json_schema`（其 chat/completions 通道不支持 schema）； +接口不支持时自动降级 `json_object`（见 `[output] schema`），保证单次判定稳定可靠。 + ## 输出结构 指定输出目录后，自动建立时间戳文件夹，按分类归档图片，Excel 放在一级目录： @@ -58,7 +65,6 @@ uv run --with pyinstaller --with aiohttp --with openpyxl \ -o, --output DIR 输出目录（默认为图片文件夹） --mode cascade（默认）/ doubao / deepseek - --no-verify 跳过 DeepSeek 无违规两票复核 --prompt FILE 自定义提示词（默认内置 prompts.txt） --workers-ark/--workers-ds/--max-tokens 调优项（默认即可） ``` @@ -77,11 +83,15 @@ api_key = sk-... model = deepseek-v4-flash-vision-exp max_tokens = 5000 [run] -mode = cascade # cascade / doubao（仅豆包）/ deepseek（仅 DeepSeek） -verify_clean = yes # DeepSeek 判无违规后是否追加两票复核（yes/no） +mode = cascade # cascade / doubao（仅豆包）/ deepseek（仅 DeepSeek） +deepseek_recheck = no # 仅 cascade 生效：no=只豆包初筛（默认）/ yes=追加 DeepSeek 复检 +[output] +schema = auto # auto（默认）/ on / off；auto=启用官方结构化输出，接口不支持自动降级 json_object ``` 运行模式：config.ini `[run] mode` 全局生效（GUI 也跟随）；CLI 的 `--mode` 参数可临时覆盖。 +DeepSeek 复检开关：`[run] deepseek_recheck`（默认 `no`）。`no` 时 cascade 等效“仅豆包”， +DeepSeek 相关的 API Key 也不再是必填项；需要 DeepSeek 复检时设 `yes`（或直接 `mode=deepseek` 全量走 DeepSeek）。 ## 项目结构 @@ -89,7 +99,7 @@ verify_clean = yes # DeepSeek 判无违规后是否追加两票复核（yes/no ├── src/violation_detector/ │ ├── config.py # 配置（自动生成 ini、缺项检测、冻结环境路径） │ ├── providers.py # 豆包/DeepSeek 异步客户端，统一结果结构 -│ ├── pipeline.py # 级联调度、断点缓存(runs/)、两票复核、成本汇总 +│ ├── pipeline.py # 级联调度、断点缓存(runs/)、成本汇总 │ ├── report.py # 时间戳输出目录、分类归档、Excel 报表 │ ├── cli.py / gui.py ├── app_gui.py / app_cli.py # 启动器 + PyInstaller 入口 @@ -116,9 +126,11 @@ uv run --with pytest --with aiohttp --with openpyxl --no-project \ ## 检测规则与已知特性 - 分类按提示词 17 类标准输出唯一分类，自动归一化写法差异。 -- DeepSeek：temperature=0 + JSON 模式、max_tokens=5000（全量梯度实测零失败最小档）。 -- 无违规两票复核：3 票一致才稳；2:1 保留多数但标记“人工复核”；分歧从严取违规类。 -- 边界图存在运行间抖动（同为 temp=0 重测全量一致率约 68%），与 max_tokens 无关； - 报表“票型/复核”列已标出分歧图，建议人工终审。 +- AI 输出：按官方启用结构化输出——豆包=response_format.json_schema(strict)， + DeepSeek=Responses API 的 text.format.json_schema（约束「文字/图形属性 · 侵权/违规逻辑 · + 违规分类」三字段）；接口不支持时自动降级 json_object（[output] schema）。 + DeepSeek 走 chat/completions 兜底时 temperature=0、max_tokens=5000。 +- 复检单次即终稿：DeepSeek 对豆包放行/存疑图各跑一遍即定稿，不再追加多票复核。 +- 边界图仍可能存在运行间抖动（单模型单次判定所致），成本较低可整批重跑比对，存疑图建议人工终审。 - 级联成本：DeepSeek 只处理豆包放行的少数存疑图（单张约 1-1.6 分钱， 空闲时段半价；高峰=工作日 9-12、14-18 点）。 diff --git a/config.example.ini b/config.example.ini index 5b84837..7190c77 100644 --- a/config.example.ini +++ b/config.example.ini @@ -24,8 +24,14 @@ file = [cascade] recheck = 无违规,违规不明,检测异常 -# 运行模式：cascade=豆包初筛+DeepSeek复检（默认）/ doubao=仅豆包 / deepseek=仅DeepSeek -# 无违规两票复核：DeepSeek 判为无违规的图片是否再追加两票判定（yes/no，默认 yes） +# 运行模式：cascade=豆包初筛+（可选）DeepSeek复检 / doubao=仅豆包 / deepseek=仅DeepSeek +# DeepSeek 复检开关（仅 cascade 模式生效）：no=只豆包初筛（默认）/ yes=对无违规等存疑图再用 DeepSeek 复检 [run] mode = cascade -verify_clean = yes +deepseek_recheck = no + +# AI 输出模式：auto=启用官方结构化输出（豆包 response_format.json_schema / +# DeepSeek Responses text.format.json_schema），接口不支持时自动降级 json_object（默认） +# on=强制结构化输出 / off=仅 json_object（旧行为） +[output] +schema = auto diff --git a/src/violation_detector/cli.py b/src/violation_detector/cli.py index 8480206..1b3dd54 100644 --- a/src/violation_detector/cli.py +++ b/src/violation_detector/cli.py @@ -6,7 +6,7 @@ import logging import sys from . import __version__ -from .config import DEFAULT_CONFIG, load_config, missing_fields +from .config import DEFAULT_CONFIG, effective_mode, load_config, missing_fields from .logutil import log_file, setup_logging from .pipeline import run_detection from .report import build_report, organize_output @@ -24,12 +24,12 @@ def build_parser() -> argparse.ArgumentParser: p.add_argument("--prompt", default=None, help="提示词文件路径（覆盖配置）") p.add_argument("--mode", choices=["cascade", "doubao", "deepseek"], default=None, help="检测模式（默认取 config.ini [run] mode，未配置则 cascade）：" - "cascade=豆包初筛+DeepSeek复检，doubao=仅豆包，deepseek=仅 DeepSeek 全量") + "cascade=豆包初筛+DeepSeek复检（受 [run] deepseek_recheck 控制），" + "doubao=仅豆包，deepseek=仅 DeepSeek 全量") p.add_argument("--workers-ark", type=int, default=None, help="豆包并发数") p.add_argument("--workers-ds", type=int, default=None, help="DeepSeek 并发数") p.add_argument("--max-tokens", type=int, default=None, help="DeepSeek max_tokens（默认 5000，实测零失败最小档）") - p.add_argument("--no-verify", action="store_true", help="跳过 DeepSeek 无违规两票复核") p.add_argument("--version", action="version", version=f"%(prog)s {__version__}") return p @@ -47,9 +47,7 @@ def main(argv=None) -> int: cfg.deepseek_workers = args.workers_ds if args.max_tokens: cfg.max_tokens = args.max_tokens - mode = args.mode or cfg.mode - # 复核开关：CLI --no-verify 强制关；否则跟随 config.ini [run] verify_clean - verify = cfg.verify_clean and not args.no_verify + mode = effective_mode(cfg, args.mode or cfg.mode) missing = missing_fields(cfg, mode) if missing: @@ -59,8 +57,7 @@ def main(argv=None) -> int: return 2 try: - rows, summary = asyncio.run(run_detection( - args.folder, cfg, mode=mode, verify=verify)) + rows, summary = asyncio.run(run_detection(args.folder, cfg, mode=mode)) except FileNotFoundError as e: logger.error("路径或文件不存在：%s", e) return 1 diff --git a/src/violation_detector/config.py b/src/violation_detector/config.py index 924c8d3..e1568f4 100644 --- a/src/violation_detector/config.py +++ b/src/violation_detector/config.py @@ -56,11 +56,17 @@ file = [cascade] recheck = 无违规,违规不明,检测异常 -# 运行模式：cascade=豆包初筛+DeepSeek复检（默认）/ doubao=仅豆包 / deepseek=仅DeepSeek -# 无违规两票复核：DeepSeek 判为无违规的图片是否再追加两票判定（yes/no，默认 yes） +# 运行模式：cascade=豆包初筛+（可选）DeepSeek复检 / doubao=仅豆包 / deepseek=仅DeepSeek +# DeepSeek 复检开关（仅 cascade 模式生效）：no=只豆包初筛（默认）/ yes=对无违规等存疑图再用 DeepSeek 复检 [run] mode = cascade -verify_clean = yes +deepseek_recheck = no + +# AI 输出模式：auto=启用官方结构化输出（豆包 response_format.json_schema / +# DeepSeek Responses text.format.json_schema），接口不支持时自动降级 +# json_object（默认）；on=强制结构化输出 / off=仅 json_object（旧行为） +[output] +schema = auto """ # 环境变量优先于配置文件 @@ -76,6 +82,7 @@ PRICE = {"miss": (3.0, 1.5), "cached": (0.10, 0.05), "output": (9.0, 4.5)} VALID_MODES = ("cascade", "doubao", "deepseek") +VALID_SCHEMA_MODES = ("auto", "on", "off") def normalize_mode(mode: str) -> str: @@ -83,6 +90,11 @@ def normalize_mode(mode: str) -> str: return mode if mode in VALID_MODES else "cascade" +def normalize_schema_mode(mode: str) -> str: + """非法输出模式回退为 auto。""" + return mode if mode in VALID_SCHEMA_MODES else "auto" + + @dataclass class AppConfig: # 豆包（火山方舟） @@ -101,23 +113,35 @@ class AppConfig: recheck_categories: list = field(default_factory=lambda: ["无违规", "违规不明", "检测异常"]) # 运行 mode: str = "cascade" - verify_clean: bool = True + deepseek_recheck: bool = False # 是否用 DeepSeek 复检（cascade 模式生效，默认关） + output_schema_mode: str = "auto" # auto / on / off retries: int = 3 @property def prompt_path(self) -> Path: - """提示词路径：空=内置；相对路径依次在工作目录/资源目录中查找。""" + """提示词路径：空=内置；含盘符的绝对路径原样返回。 + + 相对路径依次在：exe/项目目录（app_dir）→ 当前工作目录 → 内置资源目录 中查找； + 容错处理无盘符的行首 `/` 或 `\\`（如 `/prompts.txt` 视为 exe 同目录的 prompts.txt）。""" raw = (self.prompt_file or "").strip() if not raw: return DEFAULT_PROMPT p = Path(raw) - if p.is_absolute(): + if p.drive or (p.is_absolute() and not raw.startswith(("/", "\\"))): return p - for base in (app_dir(), resource_dir()): - cand = base / p + rel = raw.lstrip("/\\") or p.name # 去掉行首分隔符，视作相对路径 + for base in (app_dir(), Path.cwd(), resource_dir()): + cand = base / rel if cand.exists(): return cand - return app_dir() / p + return app_dir() / rel + + +def effective_mode(cfg: AppConfig, mode: str) -> str: + """DeepSeek 复检开关关闭时，cascade 等效 doubao（仅豆包初筛）。""" + if mode == "cascade" and not cfg.deepseek_recheck: + return "doubao" + return mode def ensure_config(path: str | None = None) -> Path: @@ -158,12 +182,12 @@ def load_config(path: str | None = None) -> AppConfig: return None def get_bool(section, key): - if not parser.has_option(section, key): - return None - try: - return parser.getboolean(section, key) - except ValueError: - return None + if parser.has_option(section, key): + try: + return parser.getboolean(section, key) + except ValueError: + return None + return None cfg.ark_api_key = get("ark", "api_key") or cfg.ark_api_key cfg.ark_model = get("ark", "model_id") or cfg.ark_model @@ -179,8 +203,10 @@ def load_config(path: str | None = None) -> AppConfig: if raw: cfg.recheck_categories = [s.strip() for s in raw.split(",") if s.strip()] cfg.mode = normalize_mode(get("run", "mode") or cfg.mode) - if get_bool("run", "verify_clean") is not None: - cfg.verify_clean = get_bool("run", "verify_clean") + if get_bool("run", "deepseek_recheck") is not None: + cfg.deepseek_recheck = get_bool("run", "deepseek_recheck") + cfg.output_schema_mode = normalize_schema_mode( + get("output", "schema") or cfg.output_schema_mode) for attr, env in ENV_OVERRIDES.items(): val = os.environ.get(env) @@ -201,7 +227,9 @@ def save_config(cfg: AppConfig, path: str | None = None): "max_tokens": str(cfg.max_tokens)} parser["prompt"] = {"file": cfg.prompt_file or ""} parser["cascade"] = {"recheck": ",".join(cfg.recheck_categories)} - parser["run"] = {"mode": cfg.mode, "verify_clean": "yes" if cfg.verify_clean else "no"} + parser["run"] = {"mode": cfg.mode, + "deepseek_recheck": "yes" if cfg.deepseek_recheck else "no"} + parser["output"] = {"schema": cfg.output_schema_mode} with ini.open("w", encoding="utf-8") as f: parser.write(f) diff --git a/src/violation_detector/gui.py b/src/violation_detector/gui.py index 965af2a..5421bfc 100644 --- a/src/violation_detector/gui.py +++ b/src/violation_detector/gui.py @@ -1,7 +1,8 @@ # -*- coding: utf-8 -*- """Tk GUI 入口：python -m violation_detector.gui 或 violation-detector-gui。 -面向使用者的极简界面：只选文件夹和输出目录，级联模式与两票复核固定开启， +面向使用者的极简界面：只选文件夹和输出目录，检测流程由 config.ini 控制 +（豆包初筛；[run] deepseek_recheck=yes 时才追加 DeepSeek 复检）， 模型参数全部走 config.ini（缺失时弹出首次配置对话框）。 检测在后台线程运行 asyncio 事件循环；日志与进度经线程安全队列由主线程刷新。 """ @@ -14,7 +15,7 @@ import tkinter as tk from tkinter import filedialog, messagebox, scrolledtext, ttk from . import __version__ -from .config import load_config, missing_fields, save_config +from .config import effective_mode, load_config, missing_fields, save_config from .logutil import QueueHandler, log_file, setup_logging from .pipeline import run_detection from .report import build_report, organize_output @@ -105,11 +106,11 @@ class App: ttk.Button(grid, text="浏览…", command=self.pick_output).grid(row=1, column=2) ttk.Label(grid, text="提示词").grid(row=2, column=0, sticky="w", pady=4) - self.var_prompt = tk.StringVar(value="") + self.var_prompt = tk.StringVar(value=self.cfg.prompt_file or "") self.ent_prompt = ttk.Entry(grid, textvariable=self.var_prompt) self.ent_prompt.grid(row=2, column=1, sticky="ew", padx=6) ttk.Button(grid, text="浏览…", command=self.pick_prompt).grid(row=2, column=2) - ttk.Label(frm, text="提示词留空使用内置版本；检测流程固定为：豆包初筛 → 无违规图 DeepSeek 复检（含两票复核）", + ttk.Label(frm, text="提示词框留空=自动用 config.ini [prompt] file（无则用内置）；浏览选择=本次临时覆盖。检测流程：豆包初筛，deepseek_recheck=yes 时再 DeepSeek 复检", foreground="#666666").pack(anchor="w", pady=(2, 6)) # ---- 控制区 ---- @@ -218,14 +219,18 @@ class App: messagebox.showwarning("提示", "请选择有效的图片文件夹") return custom_prompt = self.var_prompt.get().strip() - if custom_prompt and not os.path.isfile(custom_prompt): - messagebox.showwarning("提示", "自定义提示词文件不存在") - return out_dir = self.var_output.get().strip() or folder - mode = self.cfg.mode # 模式来自 config.ini [run] mode，界面不暴露 - # 配置检查：缺失则弹首次配置对话框 + # 配置检查：缺失则弹首次配置对话框（模式来自 config.ini [run] mode） cfg = load_config() + if custom_prompt: + cfg.prompt_file = custom_prompt # 手动填写/浏览 → 本次临时覆盖 + # 未手动填写则沿用 config.ini [prompt] file（cfg 已由 load_config 载入；留空=内置提示词） + prompt_path = cfg.prompt_path + if not os.path.isfile(prompt_path): + messagebox.showwarning("提示", f"提示词文件不存在：{prompt_path}") + return + mode = effective_mode(cfg, cfg.mode) # deepseek_recheck 关时 cascade → doubao if missing_fields(cfg, mode): logger.info("检测到 config.ini 配置不完整（模式 %s），请补齐必填项。", mode) dlg = FirstRunDialog(self.root, cfg, mode) @@ -236,8 +241,8 @@ class App: setattr(cfg, attr, val) save_config(cfg) logger.info("配置已保存到 config.ini") - cfg.prompt_file = custom_prompt # 空 = 内置提示词 self.cfg = cfg + logger.info("使用提示词：%s", prompt_path) self.btn_run.configure(state="disabled") self.btn_stop.configure(state="normal") @@ -257,7 +262,7 @@ class App: try: rows, summary = loop.run_until_complete(run_detection( folder, cfg, progress=progress, - stop=self.stop_event.is_set, mode=mode, verify=cfg.verify_clean)) + stop=self.stop_event.is_set, mode=mode)) logger.info("正在整理输出（分类归档 + 生成报表）…") ts_dir = organize_output(rows, folder, out_dir) xlsx = build_report(rows, str(ts_dir), diff --git a/src/violation_detector/pipeline.py b/src/violation_detector/pipeline.py index 05458ff..94d1c71 100644 --- a/src/violation_detector/pipeline.py +++ b/src/violation_detector/pipeline.py @@ -1,12 +1,12 @@ # -*- coding: utf-8 -*- """两阶段级联检测管道： -阶段一（豆包/Ark）全量初筛 -> 阶段二（DeepSeek）仅复检初筛为 -「无违规/违规不明/检测异常」的图片；DeepSeek 判为无违规的再追加两票复核。 +阶段一（豆包/Ark）全量初筛 -> 阶段二（DeepSeek）复检初筛为 +「无违规/违规不明/检测异常」的图片；DeepSeek 单次判定即终稿（已移除三票复核）。 mode: cascade 默认，豆包初筛 + DeepSeek 复检 doubao 仅豆包 - deepseek 仅 DeepSeek（全量直接进 DeepSeek，含无违规两票复核） + deepseek 仅 DeepSeek（全量直接进 DeepSeek） 所有过程与异常通过 logging 输出（文件记录 DEBUG 级全量与完整堆栈）。 """ @@ -47,29 +47,6 @@ def _cat_num(cat): return int(m.group(1)) if m else 99 -def decide_final(runs): - """按票型决定最终结论。runs 为 status=ok 的结果列表（首票在前）。 - 全票无违规→无违规；无违规占多数→保留但标记复核； - 违规票占多数/并列→从严取违规类（并列取分类序号最小）。""" - cats = [r["category"] for r in runs] - n = len(cats) - clean = cats.count("无违规") - if clean == n: - return runs[0], f"{n}票一致：无违规" - if clean * 2 > n: - r = next(r for r in runs if r["category"] == "无违规") - viol = sorted(set(c for c in cats if c != "无违规"), key=_cat_num) - return r, f"票型{clean}:{n - clean}（{' / '.join(cats)}），少数违规票：{'、'.join(viol)}，建议人工复核" - viol = [c for c in cats if c != "无违规"] - cnt = {} - for c in viol: - cnt[c] = cnt.get(c, 0) + 1 - top = max(cnt.values()) - best = sorted([c for c, v in cnt.items() if v == top], key=_cat_num)[0] - r = next(r for r in runs if r["category"] == best) - return r, f"票型（{' / '.join(cats)}），从严取「{best}」，建议人工复核" - - def _cache_path(stage: str, folder: str) -> Path: name = re.sub(r"[^\w\u4e00-\u9fff-]", "_", Path(folder).resolve().name) or "root" RUNS_DIR.mkdir(exist_ok=True) @@ -160,56 +137,13 @@ async def _run_stage(detect, session, sem, cfg, prompt, folder, files, cache_pat return cache -async def _verify_clean(session, sem, prompt, folder, files, stage2, cfg, - progress, stop, cache_path): - """对阶段二判为无违规的图片各追加两票独立复核。""" - verify = [f for f in files - if f in stage2 and stage2[f].get("status") == "ok" - and stage2[f].get("category") == "无违规" - and "votes" not in stage2[f]] - if not verify: - return - logger.info("复核阶段：%d 张 DeepSeek 无违规图片，各追加 2 票", len(verify)) - - async def run_verify(fname): - out = [] - for _ in range(2): - if stop and stop(): - break - out.append(await detect_deepseek(session, sem, cfg, prompt, - os.path.join(folder, fname), stop)) - return fname, out - - tasks = [asyncio.ensure_future(run_verify(f)) for f in verify] - vdone = 0 - for fut in asyncio.as_completed(tasks): - try: - fname, extra = await fut - except Exception: # noqa: BLE001 - logger.exception("复核任务抛出未捕获异常，跳过该图") - continue - runs = [stage2[fname]] + [r for r in extra if r.get("status") == "ok"] - if len(runs) < 2: - stage2[fname]["vote_note"] = "复核调用失败，维持单票结论" - logger.error("复核调用失败：%s", fname) - else: - final, note = decide_final(runs) - final = dict(final) - final["votes"] = [r["category"] for r in runs] - final["vote_note"] = note - stage2[fname] = final - vdone += 1 - logger.info(" 复核 (%d/%d) %s -> %s｜%s", vdone, len(verify), fname, - stage2[fname]["category"], stage2[fname].get("vote_note", "")) - if progress: - progress(vdone, len(verify)) - _save_cache(cache_path, stage2) - return stage2 - - async def run_detection(folder, cfg: AppConfig, progress=None, stop=None, - mode="cascade", verify=True): + mode="cascade"): """执行级联检测，返回 (rows, summary)。rows 已按文件名自然排序。""" + # DeepSeek 复检开关关闭时，cascade 等效 doubao（仅豆包初筛） + if mode == "cascade" and not cfg.deepseek_recheck: + logger.info("DeepSeek 复检开关关闭（[run] deepseek_recheck=no），本次仅豆包初筛") + mode = "doubao" t0 = time.monotonic() prompt = load_prompt(cfg.prompt_path) files = list_images(folder) @@ -256,19 +190,17 @@ async def run_detection(folder, cfg: AppConfig, progress=None, stop=None, stage2 = await _run_stage(detect_deepseek, session, sem2, cfg, prompt, folder, files, ds_cache_path, progress, stop) - # ---- 无违规两票复核 ---- - if stage2 and verify and not (stop and stop()): - sem2 = asyncio.Semaphore(cfg.deepseek_workers) - await _verify_clean(session, sem2, prompt, folder, files, stage2, cfg, - progress, stop, ds_cache_path) - # ---- 合并结果 ---- rows = [] for f in files: r = dict(stage2.get(f) or stage1.get(f) or {}) r["file"] = f - r["channel"] = ("DeepSeek复检" if f in stage2 else - "豆包初筛" if stage1 else "DeepSeek") + if f in stage2 and stage1: + r["channel"] = "DeepSeek复检" + elif f in stage1: + r["channel"] = "豆包初筛" + else: + r["channel"] = "DeepSeek" r["remark"] = STATUS_LABEL.get(r.get("status", ""), "") rows.append(r) diff --git a/src/violation_detector/providers.py b/src/violation_detector/providers.py index 24e963e..217023c 100644 --- a/src/violation_detector/providers.py +++ b/src/violation_detector/providers.py @@ -1,7 +1,34 @@ # -*- coding: utf-8 -*- -"""视觉模型提供方：豆包（火山方舟）与 DeepSeek，均为 OpenAI 兼容 chat/completions。 +"""视觉模型提供方：豆包（火山方舟）与 DeepSeek。 -统一的返回结构： +按各家官方文档实现结构化输出（没有叫顶层 output_schema 的参数）： + +- 豆包（火山方舟 Ark，OpenAI 兼容 chat/completions）： + response_format = {"type": "json_schema", + "json_schema": {"name": ..., "strict": true, "schema": 三字段 JSON Schema}} + 参考：BytePlus ModelArk "Structured output (beta)"。 + +- DeepSeek（官方原生 api.deepseek.com）： + Chat Completions 只支持 response_format.type ∈ {text, json_object}，不支持 json_schema； + 真正的 schema 结构化输出在 Responses API（POST /responses）上： + text.format = {"type": "json_schema", "name": ..., "schema": ...} + 因此 DeepSeek 需要在 schema 模式下切到 /responses，图片用 input_image 内容部件传入。 + +输出模式（[output] schema，默认 auto）： + auto 启用上方结构化输出；某接口/模型不支持时（400/404/422/501 等）自动降级为 + json_object（deepseek 退回 chat/completions json_object），并缓存探测结果。 + on 强制结构化输出，不支持时按失败处理（不静默降级）。 + off 一律只用 json_object（旧行为）。 + +重试机制（两段式）： + 阶段一 重试 HTTP 错误码 / 网络异常，最多 cfg.retries 次； + 结构化输出被 4xx 拒绝时改 json_object 立即重发（不占次数）。 + 阶段二 一旦收到 200 但正文为空或解析不出三字段（parse_fail），转入“专用解析重试”， + 独立再给 cfg.retries 次（默认 3）；不再占用阶段一里 HTTP 错误码的次数。 + 解析段耗尽仍失败 → 落 parse_fail（违规不明，便于人工终审）； + 仅传输错误耗尽 → 落 error（检测异常）。 + +统一返回结构： {"file", "provider", "status"(ok/parse_fail/error), "category", "attr", "logic", "raw", "usage": {"prompt", "completion", "cached"}, "finish_reason"} """ @@ -36,6 +63,38 @@ CANON = { 16: "16. 欧区国家禁售", 17: "17. 冒犯和争议性内容", } +# 违规分类取值枚举：17 类标准名称 + 无违规 + 违规不明（与 prompts.txt 一致） +SCHEMA_CATEGORIES = [CANON[n] for n in sorted(CANON)] + ["无违规", "违规不明"] + +# 判定结果的输出 Schema（约束三个中文字段）。 +# 豆包放在 response_format.json_schema.schema；DeepSeek 放在 /responses 的 +# text.format.schema。字段说明与顺序会影响部分接口的生成。 +OUTPUT_SCHEMA = { + "type": "object", + "properties": { + "文字/图形属性": { + "type": "string", + "description": "图案/文字/设计元素的客观描述，不含主观判断", + }, + "侵权/违规逻辑": { + "type": "string", + "description": "判定为某违规分类（或无违规）的推理依据，逻辑严谨、依据明确", + }, + "违规分类": { + "type": "string", + "enum": SCHEMA_CATEGORIES, + "description": "仅可从给定的 17 类标准及 无违规/违规不明 中选择唯一一项", + }, + }, + "required": ["文字/图形属性", "侵权/违规逻辑", "违规分类"], + "additionalProperties": False, +} + +SCHEMA_NAME = "violation_check" + +# 已探测到「不支持结构化输出」的接口，key = provider|url|model +_schema_ok: dict[str, bool] = {} + def encode_image(path: str) -> str: ext = os.path.splitext(path)[1].lower() @@ -75,12 +134,222 @@ def parse_judgment(raw: str): def _norm_usage(u: dict) -> dict: + """chat/completions 的 usage → 统一结构。""" u = u or {} return {"prompt": u.get("prompt_tokens"), "completion": u.get("completion_tokens"), "cached": (u.get("prompt_tokens_details") or {}).get("cached_tokens", 0)} -def _result(fname, provider, content, data, path_for_err=""): +def _norm_usage_responses(u: dict) -> dict: + """Responses API 的 usage → 统一结构。""" + u = u or {} + return {"prompt": u.get("input_tokens"), "completion": u.get("output_tokens"), + "cached": (u.get("input_tokens_details") or {}).get("cached_tokens", 0)} + + +# ---------- 输出模式判定与 payload 组装 ---------- + +def _schema_key(provider: str, base_url: str, model: str) -> str: + return f"{provider}|{base_url.rstrip('/')}|{model}" + + +def _use_schema(cfg: AppConfig, key: str) -> bool: + """按配置决定该接口本轮是否启用结构化输出。""" + if cfg.output_schema_mode == "off": + return False + if cfg.output_schema_mode == "on": + return True + return _schema_ok.get(key, True) # auto：未探测到不支持前先用结构化输出 + + +def build_chat_payload(model, messages, extras: dict, use_schema: bool) -> dict: + """OpenAI 兼容 chat/completions 请求体（豆包用）： + 结构化输出走 response_format.json_schema（strict）；否则 json_object。""" + payload = {"model": model, "messages": messages} + payload.update(extras) + if use_schema: + payload["response_format"] = { + "type": "json_schema", + "json_schema": {"name": SCHEMA_NAME, "strict": True, "schema": OUTPUT_SCHEMA}, + } + else: + payload["response_format"] = {"type": "json_object"} + return payload + + +def build_responses_payload(model, input_items, extras: dict, use_schema: bool = True) -> dict: + """DeepSeek Responses API 请求体：结构化输出走 text.format.json_schema。""" + payload = {"model": model, "input": input_items} + payload.update(extras) + if use_schema: + payload["text"] = {"format": {"type": "json_schema", "name": SCHEMA_NAME, + "schema": OUTPUT_SCHEMA}} + else: + payload["text"] = {"format": {"type": "json_object"}} + return payload + + +def _schema_client_error(status: int, body: str) -> bool: + """结构化输出请求被拒的 4xx/5xx：这类值得降级 json_object 再试一次。""" + return status in (400, 422, 501) + + +def _schema_explicitly_unsupported(body: str) -> bool: + """错误信息是否明确指向「接口不支持 json_schema/output_schema」：决定整场直接关闭。""" + low = (body or "").lower() + if "output_schema" in low or "json_schema" in low: + return True + if "schema" not in low: + return False + tokens = ("not support", "unsupported", "does not support", "doesn't support", + "unrecognized", "unrecognised", "not recognise", "not recognize", + "unknown", "not found", "不存在", "不支持", "未知参数", + "不支持的参数", "无法识别") + return any(t in low for t in tokens) + + +def _stop_result(fname, provider): + return {"file": fname, "provider": provider, "status": "error", "attr": "", + "logic": "", "category": "检测异常", "raw": "用户停止", "usage": {}} + + +def _give_up(fname, provider, last_err): + return {"file": fname, "provider": provider, "status": "error", "attr": "", + "logic": "", "category": "检测异常", "raw": last_err, "usage": {}} + + +async def _http_post(session, sem, url, headers, payload): + """单次 POST。返回 (status, data, note)： + status==200 → data 为响应 json，note 空； + 0 bool: + """chat/completions 的正文是否可接受（非空且能解析出三字段）。""" + return bool(content and content.strip()) and parse_judgment(content) is not None + + +async def _chat_detect(session, sem, cfg, path, stop, + provider, label, base_url, api_key, model, messages, + extras): + """向一家 OpenAI 兼容 chat/completions 服务发一次图片检测请求。 + + 两段式重试：错误码/网络异常先在“传输段”重试（cfg.retries 次，schema 被 4xx 拒则 + 改 json_object 不占次数）；一旦 200 但空正文/解析失败即转入“专用解析重试段” + （独立 cfg.retries 次）。传输段耗尽→error；解析段耗尽→parse_fail（违规不明）。 + """ + fname = os.path.basename(path) + url = base_url.rstrip("/") + "/chat/completions" + headers = {"Authorization": f"Bearer {api_key}", + "Content-Type": "application/json"} + if provider == "doubao": + headers["x-is-encrypted"] = "true" + key = _schema_key(provider, base_url, model) + use_schema = _use_schema(cfg, key) + schema_rejected = False # 是否已因结构化输出被拒而降级 json_object + + def _build(): + return build_chat_payload(model, messages, extras, use_schema) + + def _confirm_schema(): + if schema_rejected and cfg.output_schema_mode == "auto": + _schema_ok[key] = False + logger.info("%s 检测 %s：确认接口不支持结构化输出，本场剩余图片改用 json_object", + label, fname) + + # ---- 阶段一：传输/错误码重试 ---- + last_err = "" + attempt = 0 + saw_parse_issue = False + while attempt < cfg.retries: + attempt += 1 + if stop and stop(): + logger.info("%s 检测 %s：用户停止", label, fname) + return _stop_result(fname, provider) + status, data, note = await _http_post(session, sem, url, headers, _build()) + if status == 200: + content = (data["choices"][0]["message"]["content"] or "") + if _chat_content_usable(content): + _confirm_schema() + return _result(fname, provider, content, data) + # 空正文 / 解析失败 → 交给阶段二独立重试 + logger.warning("%s 检测 %s 第%d/%d次尝试：返回为空或无法解析，" + "进入独立解析重试", label, fname, attempt, cfg.retries) + saw_parse_issue = True + break + # 非 200 / 网络异常 + last_err = f"HTTP {status}: {note[:300]}" if status > 0 else "网络/协议异常（详见日志）" + logger.warning("%s 检测 %s 第%d/%d次尝试失败：%s", + label, fname, attempt, cfg.retries, last_err) + if status > 0 and use_schema and _schema_client_error(status, note): + if _schema_explicitly_unsupported(note): + _schema_ok[key] = False + logger.warning("%s 接口不支持结构化输出，本场剩余图片改用 json_object", label) + else: + logger.info("%s 检测 %s：结构化输出请求被拒，本张改 json_object 重试", label, fname) + schema_rejected = True + use_schema = False + attempt -= 1 # 降级重发不占用本次传输次数 + continue + if attempt < cfg.retries: + logger.info("%s 检测 %s：%.0f 秒后重试", label, fname, 3 * attempt) + await asyncio.sleep(3 * attempt) + if not saw_parse_issue: + logger.error("%s 检测 %s 最终失败：%s", label, fname, last_err) + return _give_up(fname, provider, last_err) + + # ---- 阶段二：专用解析重试（独立 cfg.retries 次） ---- + last_data = None + p = 0 + while p < cfg.retries: + p += 1 + if stop and stop(): + logger.info("%s 检测 %s：用户停止（解析重试中）", label, fname) + return _stop_result(fname, provider) + status, data, note = await _http_post(session, sem, url, headers, _build()) + if status == 200: + content = (data["choices"][0]["message"]["content"] or "") + if _chat_content_usable(content): + _confirm_schema() + return _result(fname, provider, content, data) + last_data = data + logger.warning(" 解析重试 (%d/%d) %s 仍为空/无法解析", p, cfg.retries, fname) + else: + last_err = f"HTTP {status}: {note[:300]}" if status > 0 else "网络/协议异常（详见日志）" + logger.warning(" 解析重试 (%d/%d) %s 遇 %s", p, cfg.retries, fname, last_err) + if status > 0 and use_schema and _schema_client_error(status, note): + if _schema_explicitly_unsupported(note): + _schema_ok[key] = False + logger.warning("%s 接口不支持结构化输出，改用 json_object", label) + else: + logger.info("%s 检测 %s：结构化输出被拒，本张改 json_object 重试", label, fname) + use_schema = False + p -= 1 # 降级重发不占用解析次数 + continue + if p < cfg.retries: + logger.info(" 解析重试 %.0f 秒后再次尝试", 3 * p) + await asyncio.sleep(3 * p) + if last_data is not None: + content = last_data["choices"][0]["message"]["content"] or "" + logger.error("%s 检测 %s：含专用解析重试仍无法解析，按 parse_fail 记录", label, fname) + return _result(fname, provider, content, last_data) + logger.error("%s 检测 %s 解析重试中最终失败：%s", label, fname, last_err) + return _give_up(fname, provider, last_err) + + +def _result(fname, provider, content, data): ch = (data.get("choices") or [{}])[0] parsed = parse_judgment(content) base = {"file": fname, "provider": provider, "finish_reason": ch.get("finish_reason"), @@ -95,105 +364,173 @@ def _result(fname, provider, content, data, path_for_err=""): return base -async def detect_ark(session, sem, cfg: AppConfig, prompt: str, path: str, stop=None): - """豆包（火山方舟）视觉检测：图片在前、文本在后，强制 JSON 输出。""" +# ---------- DeepSeek Responses API 引擎（结构化输出） ---------- + +def _extract_output_text(data: dict) -> str: + """Responses API 响应里拼接所有 assistant output_text 内容。""" + parts = [] + for item in data.get("output") or []: + if item.get("type") == "message" and item.get("role") == "assistant": + for part in item.get("content") or []: + if part.get("type") == "output_text": + parts.append(part.get("text") or "") + return "".join(parts) + + +def _responses_usable(data) -> bool: + """Responses 响应是否可接受：status=completed、正文非空且能解析出三字段。""" + if data.get("status") != "completed": + return False + content = _extract_output_text(data) + return bool(content and content.strip()) and parse_judgment(content) is not None + + +def _result_responses(fname, data): + content = _extract_output_text(data) + parsed = parse_judgment(content) + base = {"file": fname, "provider": "deepseek", + "finish_reason": "completed" if data.get("status") == "completed" + else (data.get("status") or ""), + "usage": _norm_usage_responses(data.get("usage")), "raw": content} + if parsed: + attr, logic, cat = parsed + base.update({"status": "ok", "attr": attr, "logic": logic, + "category": norm_category(cat), "raw": content}) + else: + base.update({"status": "parse_fail", "attr": "", "logic": "", + "category": "违规不明", "raw": content or ""}) + return base + + +async def _responses_detect(session, sem, cfg, prompt, path, stop, key): + """DeepSeek Responses API（POST /responses）结构化输出。 + + 两段式重试同 chat 引擎。返回约定： + 成功 → 结果字典；接口不支持（400/404/422/501）→ None（调用方决定降级）； + 解析段耗尽 → parse_fail；传输段耗尽 → error。""" fname = os.path.basename(path) - url = cfg.ark_base_url.rstrip("/") + "/chat/completions" - headers = {"Authorization": f"Bearer {cfg.ark_api_key}", - "Content-Type": "application/json", - "x-is-encrypted": "true"} - payload = { - "model": cfg.ark_model, - "messages": [{ - "role": "user", - "content": [ - {"type": "image_url", "image_url": {"url": encode_image(path)}}, - {"type": "text", "text": prompt}, - ], - }], - "response_format": {"type": "json_object"}, - } - - last_err = "" - for attempt in range(1, cfg.retries + 1): - if stop and stop(): - logger.info("豆包检测 %s：用户停止", fname) - return {"file": fname, "provider": "doubao", "status": "error", "attr": "", - "logic": "", "category": "检测异常", "raw": "用户停止", "usage": {}} - try: - async with sem: - async with session.post(url, json=payload, headers=headers) as r: - if r.status == 200: - data = await r.json() - content = data["choices"][0]["message"]["content"] - return _result(fname, "doubao", content, data) - body = (await r.text())[:2000] - last_err = f"HTTP {r.status}: {body[:300]}" - logger.warning("豆包检测 %s 第%d/%d次尝试失败：HTTP %s\n响应体：%s", - fname, attempt, cfg.retries, r.status, body) - except Exception: # noqa: BLE001 网络层异常：记录完整堆栈后重试 - last_err = "网络/协议异常（详见日志）" - logger.exception("豆包检测 %s 第%d/%d次尝试抛出异常", fname, attempt, cfg.retries) - if attempt < cfg.retries: - logger.info("豆包检测 %s：%.0f 秒后重试", fname, 3 * attempt) - await asyncio.sleep(3 * attempt) - logger.error("豆包检测 %s 最终失败：%s", fname, last_err) - return {"file": fname, "provider": "doubao", "status": "error", "attr": "", - "logic": "", "category": "检测异常", "raw": last_err, "usage": {}} - - -async def detect_deepseek(session, sem, cfg: AppConfig, prompt: str, path: str, stop=None): - """DeepSeek 视觉检测：temperature=0 + JSON 模式。 - 注意该模型为推理模型，reasoning 消耗 completion 预算，max_tokens 须留足。""" - fname = os.path.basename(path) - url = cfg.deepseek_base_url.rstrip("/") + "/chat/completions" + url = cfg.deepseek_base_url.rstrip("/") + "/responses" headers = {"Authorization": f"Bearer {cfg.deepseek_api_key}", "Content-Type": "application/json"} - payload = { - "model": cfg.deepseek_model, - "messages": [{ - "role": "user", - "content": [ - {"type": "text", "text": prompt}, - {"type": "image_url", "image_url": {"url": encode_image(path)}}, - ], - }], - "temperature": 0, - "response_format": {"type": "json_object"}, - "max_tokens": cfg.max_tokens, - } + input_item = {"role": "user", "content": [ + {"type": "input_text", "text": prompt}, + {"type": "input_image", "image_url": encode_image(path)}, + ]} + def _build(): + return build_responses_payload( + cfg.deepseek_model, [input_item], + {"temperature": 0, "max_output_tokens": cfg.max_tokens}, True) + + # ---- 阶段一：传输/错误码重试 ---- last_err = "" - for attempt in range(1, cfg.retries + 1): + attempt = 0 + saw_parse_issue = False + while attempt < cfg.retries: + attempt += 1 if stop and stop(): logger.info("DeepSeek 检测 %s：用户停止", fname) - return {"file": fname, "provider": "deepseek", "status": "error", "attr": "", - "logic": "", "category": "检测异常", "raw": "用户停止", "usage": {}} - try: - async with sem: - async with session.post(url, json=payload, headers=headers) as r: - if r.status == 200: - data = await r.json() - content = data["choices"][0]["message"]["content"] - if content and content.strip(): - return _result(fname, "deepseek", content, data) - # 推理烧尽预算会得到空正文，按可重试错误处理 - fr = data["choices"][0].get("finish_reason") - last_err = f"empty content (finish_reason={fr}, max_tokens={cfg.max_tokens})" - logger.warning("DeepSeek 检测 %s 第%d/%d次尝试返回空正文" - "（finish_reason=%s，推理可能耗尽 max_tokens=%s）", - fname, attempt, cfg.retries, fr, cfg.max_tokens) - else: - body = (await r.text())[:2000] - last_err = f"HTTP {r.status}: {body[:300]}" - logger.warning("DeepSeek 检测 %s 第%d/%d次尝试失败：HTTP %s\n响应体：%s", - fname, attempt, cfg.retries, r.status, body) - except Exception: # noqa: BLE001 网络层异常：记录完整堆栈后重试 - last_err = "网络/协议异常（详见日志）" - logger.exception("DeepSeek 检测 %s 第%d/%d次尝试抛出异常", fname, attempt, cfg.retries) + return _stop_result(fname, "deepseek") + status, data, note = await _http_post(session, sem, url, headers, _build()) + if status == 200: + if _responses_usable(data): + return _result_responses(fname, data) + # completed 但空/解析失败，或服务端未完成 → 交给阶段二独立重试 + reason = (data.get("incomplete_details") or {}).get("reason") if data.get( + "status") != "completed" else None + logger.warning("DeepSeek 检测 %s 第%d/%d次尝试：%s（%s），进入独立解析重试", + fname, attempt, cfg.retries, data.get("status"), + reason or "空正文/无法解析") + saw_parse_issue = True + break + last_err = f"HTTP {status}: {note[:300]}" if status > 0 else "网络/协议异常（详见日志）" + logger.warning("DeepSeek 检测 %s 第%d/%d次尝试失败：%s", + fname, attempt, cfg.retries, last_err) + if status > 0 and status in (400, 404, 422, 501): + # 端点上 /responses 或 json_schema 不受支持 → 标记并交还调用方降级 + _schema_ok[key] = False + logger.warning("DeepSeek /responses 结构化输出不可用，后续改用 chat/completions json_object") + return None if attempt < cfg.retries: logger.info("DeepSeek 检测 %s：%.0f 秒后重试", fname, 3 * attempt) await asyncio.sleep(3 * attempt) - logger.error("DeepSeek 检测 %s 最终失败：%s", fname, last_err) - return {"file": fname, "provider": "deepseek", "status": "error", "attr": "", - "logic": "", "category": "检测异常", "raw": last_err, "usage": {}} + if not saw_parse_issue: + logger.error("DeepSeek 检测 %s 最终失败：%s", fname, last_err) + return _give_up(fname, "deepseek", last_err) + + # ---- 阶段二：专用解析重试（独立 cfg.retries 次） ---- + last_data = None + p = 0 + while p < cfg.retries: + p += 1 + if stop and stop(): + logger.info("DeepSeek 检测 %s：用户停止（解析重试中）", fname) + return _stop_result(fname, "deepseek") + status, data, note = await _http_post(session, sem, url, headers, _build()) + if status == 200: + if _responses_usable(data): + return _result_responses(fname, data) + last_data = data + logger.warning(" 解析重试 (%d/%d) %s 仍为空/无法解析", p, cfg.retries, fname) + else: + last_err = f"HTTP {status}: {note[:300]}" if status > 0 else "网络/协议异常（详见日志）" + logger.warning(" 解析重试 (%d/%d) %s 遇 %s", p, cfg.retries, fname, last_err) + if status > 0 and status in (400, 404, 422, 501): + _schema_ok[key] = False + logger.warning("DeepSeek /responses 结构化输出不可用，退回 chat/completions") + return None + if p < cfg.retries: + logger.info(" 解析重试 %.0f 秒后再次尝试", 3 * p) + await asyncio.sleep(3 * p) + if last_data is not None: + logger.error("DeepSeek 检测 %s：含专用解析重试仍无法解析，按 parse_fail 记录", fname) + return _result_responses(fname, last_data) + logger.error("DeepSeek 检测 %s 解析重试中最终失败：%s", fname, last_err) + return _give_up(fname, "deepseek", last_err) + + +# ---------- 对外检测入口 ---------- + +async def detect_ark(session, sem, cfg: AppConfig, prompt: str, path: str, stop=None): + """豆包（火山方舟）视觉检测：图片在前、文本在后。 + 结构化输出（[output] schema）时用 response_format json_schema(strict)。""" + messages = [{"role": "user", "content": [ + {"type": "image_url", "image_url": {"url": encode_image(path)}}, + {"type": "text", "text": prompt}, + ]}] + return await _chat_detect( + session, sem, cfg, path, stop, + provider="doubao", label="豆包", + base_url=cfg.ark_base_url, api_key=cfg.ark_api_key, + model=cfg.ark_model, messages=messages, extras={}) + + +async def detect_deepseek(session, sem, cfg: AppConfig, prompt: str, path: str, stop=None): + """DeepSeek 视觉检测。 + + [output] schema 启用（auto/on）时走官方 Responses API（/responses + text.format + json_schema，这是 DeepSeek 唯一支持 schema 强约束的通道）；auto 下接口不可用则 + 退回 chat/completions + json_object。关闭（off）时直接用 json_object。""" + key = _schema_key("deepseek", cfg.deepseek_base_url, cfg.deepseek_model) + if _use_schema(cfg, key): + res = await _responses_detect(session, sem, cfg, prompt, path, stop, key) + if res is not None: + return res + fname = os.path.basename(path) + if cfg.output_schema_mode == "on": + logger.error("DeepSeek 不支持结构化输出（[output] schema=on 强制开启，不降级）") + return _give_up(fname, "deepseek", + "接口不支持结构化输出（[output] schema=on 强制开启，不降级）") + logger.info("DeepSeek 检测 %s：退回 chat/completions json_object", fname) + + # chat/completions + json_object（官方 chat 通道唯一支持的模式） + messages = [{"role": "user", "content": [ + {"type": "text", "text": prompt}, + {"type": "image_url", "image_url": {"url": encode_image(path)}}, + ]}] + return await _chat_detect( + session, sem, cfg, path, stop, + provider="deepseek", label="DeepSeek", + base_url=cfg.deepseek_base_url, api_key=cfg.deepseek_api_key, + model=cfg.deepseek_model, messages=messages, + extras={"temperature": 0, "max_tokens": cfg.max_tokens}) diff --git a/src/violation_detector/report.py b/src/violation_detector/report.py index 353f2e3..bedc6cf 100644 --- a/src/violation_detector/report.py +++ b/src/violation_detector/report.py @@ -95,9 +95,9 @@ def build_report(rows, out_dir, meta=None): ws = wb.active ws.title = "检测结果" headers = ["序号", "文件名", "违规分类", "判定通道", "文字/图形属性", - "侵权/违规逻辑", "票型/复核", "备注"] + "侵权/违规逻辑", "备注"] last_col = len(headers) + 1 - widths = {"B": 6, "C": 42, "D": 26, "E": 13, "F": 32, "G": 50, "H": 30, "I": 20} + widths = {"B": 6, "C": 42, "D": 26, "E": 13, "F": 32, "G": 50, "H": 20} ws.sheet_view.showGridLines = False ws.column_dimensions["A"].width = 3 @@ -130,7 +130,6 @@ def build_report(rows, out_dir, meta=None): rn = 5 + i values = [i + 1, r.get("file", ""), r.get("category", ""), r.get("channel", ""), r.get("attr", ""), r.get("logic", ""), - r.get("vote_note", "单票判定（未触发复核）"), r.get("remark", STATUS_LABEL.get(r.get("status", ""), ""))] fill = PatternFill("solid", fgColor=NEUTRAL_100 if i % 2 else "FFFFFF") for ci, v in enumerate(values, start=2): @@ -141,19 +140,20 @@ def build_report(rows, out_dir, meta=None): ws.cell(row=rn, column=2).alignment = Alignment(horizontal="right", vertical="center") ws.cell(row=rn, column=4).font = Font(name=FONT_NAME, size=11, color=_cat_color(values[2])) - vnote = ws.cell(row=rn, column=8) - if "人工复核" in str(vnote.value): - vnote.font = Font(name=FONT_NAME, size=11, color=ACCENT_WARNING) last_row = 4 + len(rows) _auto_heights(ws, widths, 5, last_row) ws.freeze_panes = "C5" # 底注 now = datetime.now().strftime("%Y-%m-%d %H:%M") - note1 = (f"检测模式：{meta.get('mode', 'cascade')}（豆包初筛 + DeepSeek 复检无违规）｜" - f"图片总数：{len(rows)}｜生成时间：{now}") - note2 = ("依据《违规检测提示词》17 类标准判定；DeepSeek temperature=0 + JSON 模式、" - "max_tokens 可配置；判定为无违规的图片自动追加两票复核（3 票取多数，分歧从严取违规并标记人工复核）。") + mode_txt = {"cascade": "豆包初筛 + DeepSeek 复检（单次判定即终稿）", + "doubao": "仅豆包初筛", + "deepseek": "仅 DeepSeek 全量检测"}.get(meta.get("mode", "cascade"), + meta.get("mode", "cascade")) + note1 = f"检测模式：{mode_txt}｜图片总数：{len(rows)}｜生成时间：{now}" + note2 = ("依据《违规检测提示词》17 类标准判定；AI 输出启用官方结构化输出" + "（豆包 response_format.json_schema / DeepSeek Responses json_schema，" + "接口不支持自动降级 json_object）；DeepSeek 复检单次判定即终稿。") for off, note in ((2, note1), (3, note2)): c = ws.cell(row=last_row + off, column=2, value=note) c.font = Font(name=FONT_NAME, size=9, color=NEUTRAL_600) diff --git a/tests/test_config.py b/tests/test_config.py index d664c28..0e208b3 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,7 +1,9 @@ # -*- coding: utf-8 -*- """config 模块测试：模板生成、缺项检测、读写回环、提示词路径。""" +import violation_detector.config as config_mod from violation_detector.config import ( - AppConfig, ensure_config, load_config, missing_fields, save_config, + AppConfig, effective_mode, ensure_config, load_config, missing_fields, + save_config, ) @@ -63,18 +65,57 @@ def test_mode_from_ini(tmp_path): assert load_config(str(ini)).mode == "cascade" -def test_verify_clean_from_ini(tmp_path): +def test_output_schema_from_ini(tmp_path): ini = tmp_path / "config.ini" - ini.write_text("[run]\nverify_clean = no\n", encoding="utf-8") - assert load_config(str(ini)).verify_clean is False - ini.write_text("[run]\nverify_clean = yes\n", encoding="utf-8") - assert load_config(str(ini)).verify_clean is True - # 非法值保持默认开启 - ini.write_text("[run]\nverify_clean = 也许\n", encoding="utf-8") - assert load_config(str(ini)).verify_clean is True - # 未配置保持默认开启 + ini.write_text("[output]\nschema = on\n", encoding="utf-8") + assert load_config(str(ini)).output_schema_mode == "on" + ini.write_text("[output]\nschema = off\n", encoding="utf-8") + assert load_config(str(ini)).output_schema_mode == "off" + ini.write_text("[output]\nschema = auto\n", encoding="utf-8") + assert load_config(str(ini)).output_schema_mode == "auto" + # 非法值回退 auto；未配置默认 auto + ini.write_text("[output]\nschema = 乱来\n", encoding="utf-8") + assert load_config(str(ini)).output_schema_mode == "auto" ini.write_text("[ark]\n", encoding="utf-8") - assert load_config(str(ini)).verify_clean is True + assert load_config(str(ini)).output_schema_mode == "auto" + + +def test_output_schema_roundtrip(tmp_path): + ini = tmp_path / "config.ini" + save_config(AppConfig(output_schema_mode="off"), str(ini)) + assert load_config(str(ini)).output_schema_mode == "off" + assert AppConfig().output_schema_mode == "auto" + + +def test_deepseek_recheck_default_and_ini(tmp_path): + # 默认关 + assert AppConfig().deepseek_recheck is False + ini = tmp_path / "config.ini" + ini.write_text("[run]\ndeepseek_recheck = yes\n", encoding="utf-8") + assert load_config(str(ini)).deepseek_recheck is True + ini.write_text("[run]\ndeepseek_recheck = no\n", encoding="utf-8") + assert load_config(str(ini)).deepseek_recheck is False + # 非法值/未配置保持默认关 + ini.write_text("[run]\ndeepseek_recheck = 也许\n", encoding="utf-8") + assert load_config(str(ini)).deepseek_recheck is False + ini.write_text("[ark]\n", encoding="utf-8") + assert load_config(str(ini)).deepseek_recheck is False + + +def test_effective_mode_respects_recheck_switch(): + on = AppConfig(mode="cascade", deepseek_recheck=True) + off = AppConfig(mode="cascade", deepseek_recheck=False) + assert effective_mode(on, "cascade") == "cascade" + assert effective_mode(off, "cascade") == "doubao" + # 开关只影响 cascade；doubao/deepseek 不受影响 + assert effective_mode(off, "doubao") == "doubao" + assert effective_mode(off, "deepseek") == "deepseek" + + +def test_deepseek_recheck_roundtrip(tmp_path): + ini = tmp_path / "config.ini" + save_config(AppConfig(deepseek_recheck=True), str(ini)) + assert load_config(str(ini)).deepseek_recheck is True def test_save_and_load_roundtrip(tmp_path): @@ -104,3 +145,21 @@ def test_prompt_path_custom(tmp_path): p.write_text("x", encoding="utf-8") cfg = AppConfig(prompt_file=str(p)) assert cfg.prompt_path == p + + +def test_prompt_path_relative_prefers_app_dir(tmp_path, monkeypatch): + (tmp_path / "prompts.txt").write_text("x", encoding="utf-8") + monkeypatch.setattr(config_mod, "app_dir", lambda: tmp_path) + monkeypatch.setattr(config_mod, "resource_dir", lambda: tmp_path) + cfg = AppConfig(prompt_file="prompts.txt") + assert cfg.prompt_path == tmp_path / "prompts.txt" + + +def test_prompt_path_leading_slash_treated_relative(tmp_path, monkeypatch): + # 无盘符的 /prompts.txt 应视为 exe 同目录的相对文件（Windows 容错） + (tmp_path / "prompts.txt").write_text("x", encoding="utf-8") + monkeypatch.setattr(config_mod, "app_dir", lambda: tmp_path) + monkeypatch.setattr(config_mod, "resource_dir", lambda: tmp_path) + for raw in ("/prompts.txt", "\\prompts.txt"): + cfg = AppConfig(prompt_file=raw) + assert cfg.prompt_path == tmp_path / "prompts.txt", raw diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index fd3036f..0b93ecd 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- -"""pipeline 模块测试：排序、投票裁决、成本计算、缓存路径。""" +"""pipeline 模块测试：排序、成本计算、缓存路径。""" from violation_detector.pipeline import ( - _cache_path, _cat_num, deepseek_cost, decide_final, natural_key, + _cache_path, _cat_num, deepseek_cost, natural_key, ) @@ -19,42 +19,11 @@ def test_natural_key_sort(): def test_cat_num(): assert _cat_num("12. 侵权 - 其他") == 12 assert _cat_num("1. 烟草") == 1 - # 无编号的结论统一排到最后（无违规/违规不变动分类，不影响裁决） + # 无编号的结论统一排到最后（无违规/违规不明不变动分类） assert _cat_num("无违规") == 99 assert _cat_num("违规不明") == 99 -def test_decide_final_unanimous_clean(): - final, note = decide_final([R("无违规"), R("无违规"), R("无违规")]) - assert final["category"] == "无违规" - assert "3票一致" in note - - -def test_decide_final_clean_majority_keeps_but_flags(): - final, note = decide_final([R("无违规"), R("无违规"), R("12. 侵权 - 除人物外的其他侵权")]) - assert final["category"] == "无违规" - assert "2:1" in note and "人工复核" in note - - -def test_decide_final_violation_majority_wins(): - final, note = decide_final( - [R("无违规"), R("12. 侵权 - 除人物外的其他侵权"), R("12. 侵权 - 除人物外的其他侵权")]) - assert final["category"].startswith("12.") - assert "人工复核" in note - - -def test_decide_final_three_way_split_takes_lowest_number(): - final, note = decide_final([R("无违规"), R("13. 侵权 - 人物相关"), R("5. 负向敏感")]) - assert final["category"].startswith("5.") - assert "从严" in note - - -def test_decide_final_two_violations_differ_majority_none(): - # 两张违规票不同、一张无违规：违规方 2 票 > 1，并列取序号最小 - final, _ = decide_final([R("无违规"), R("13. 侵权 - 人物相关"), R("12. 侵权 - 其他")]) - assert final["category"].startswith("12.") - - def test_deepseek_cost_known_usage(): # miss 407 + cached 1792 + completion 2282 results = {"a": R("无违规")} diff --git a/tests/test_providers.py b/tests/test_providers.py index 01104f7..05be6c0 100644 --- a/tests/test_providers.py +++ b/tests/test_providers.py @@ -1,10 +1,17 @@ # -*- coding: utf-8 -*- -"""providers 模块测试：判定 JSON 解析、分类归一化、图片编码、usage 规整。""" +"""providers 模块测试：判定 JSON 解析、分类归一化、图片编码、usage 规整、output_schema、重试。""" +import asyncio import base64 import pytest -from violation_detector.providers import CANON, _norm_usage, encode_image, norm_category, parse_judgment +from violation_detector.config import AppConfig +from violation_detector.providers import ( + CANON, OUTPUT_SCHEMA, SCHEMA_CATEGORIES, SCHEMA_NAME, _chat_detect, + _extract_output_text, _norm_usage, _result_responses, + _schema_explicitly_unsupported, build_chat_payload, + build_responses_payload, encode_image, norm_category, parse_judgment, +) GOOD = ('{"文字/图形属性": "黑色T恤印字母", "侵权/违规逻辑": "未授权使用商标", ' '"违规分类": "12. 侵权 - 除人物外的其他侵权"}') @@ -67,3 +74,196 @@ def test_norm_usage_handles_missing_details(): "prompt": 100, "completion": 50, "cached": 0} assert _norm_usage({}) == {"prompt": None, "completion": None, "cached": 0} assert _norm_usage({"prompt_tokens": 100, "prompt_tokens_details": {"cached_tokens": 80}})["cached"] == 80 + + +# ---------- output_schema ---------- + +def test_output_schema_shape(): + assert OUTPUT_SCHEMA["type"] == "object" + assert set(OUTPUT_SCHEMA["required"]) == {"文字/图形属性", "侵权/违规逻辑", "违规分类"} + assert OUTPUT_SCHEMA["additionalProperties"] is False + assert set(OUTPUT_SCHEMA["properties"]) == set(OUTPUT_SCHEMA["required"]) + + +def test_schema_categories_covers_canon_and_specials(): + expect = [CANON[n] for n in sorted(CANON)] + ["无违规", "违规不明"] + assert SCHEMA_CATEGORIES == expect + + +def test_build_chat_payload_schema_vs_json(): + # 豆包：结构化输出 = response_format.json_schema(strict)；降级 = json_object + messages = [{"role": "user", "content": []}] + s = build_chat_payload("m", messages, {"temperature": 0}, True) + assert s["model"] == "m" and s["temperature"] == 0 + rf = s["response_format"] + assert rf["type"] == "json_schema" + assert rf["json_schema"]["name"] == SCHEMA_NAME + assert rf["json_schema"]["strict"] is True + assert rf["json_schema"]["schema"] is OUTPUT_SCHEMA + j = build_chat_payload("m", messages, {}, False) + assert j["response_format"] == {"type": "json_object"} + + +def test_build_responses_payload_schema(): + # DeepSeek：结构化输出 = text.format.json_schema（Responses API） + items = [{"role": "user", "content": [{"type": "input_text", "text": "p"}]}] + p = build_responses_payload("m", items, {"max_output_tokens": 5000}, True) + assert p["model"] == "m" and p["max_output_tokens"] == 5000 + fmt = p["text"]["format"] + assert fmt["type"] == "json_schema" + assert fmt["name"] == SCHEMA_NAME + assert fmt["schema"] is OUTPUT_SCHEMA + + +def test_extract_output_text_concats_output_text(): + data = {"output": [ + {"type": "reasoning", "role": "assistant", + "content": [{"type": "reasoning_text", "text": "思考"}], + "status": "completed"}, + {"type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": '{"a": 1}'}], + "status": "completed"}, + ]} + assert _extract_output_text(data) == '{"a": 1}' + + +def test_result_responses_ok(): + data = { + "status": "completed", + "usage": {"input_tokens": 10, "output_tokens": 20, + "input_tokens_details": {"cached_tokens": 4}}, + "output": [{"type": "message", "role": "assistant", "content": [ + {"type": "output_text", "text": '{"文字/图形属性": "T恤", ' + '"侵权/违规逻辑": "无", "违规分类": "无违规"}'}]}], + } + r = _result_responses("a.jpg", data) + assert r["status"] == "ok" and r["category"] == "无违规" + assert r["usage"] == {"prompt": 10, "completion": 20, "cached": 4} + assert r["finish_reason"] == "completed" + + +def test_schema_explicitly_unsupported(): + assert _schema_explicitly_unsupported( + "Unrecognized request argument supplied: response_format.json_schema") + assert _schema_explicitly_unsupported("model not support json_schema") + assert _schema_explicitly_unsupported("不支持的参数 json_schema") + assert not _schema_explicitly_unsupported("HTTP 400 some other problem") + + +# ---------- 重试机制（错误码 / 空正文 / 解析失败） ---------- + +RETRY_GOOD = '{"文字/图形属性": "T恤", "侵权/违规逻辑": "无", "违规分类": "无违规"}' +RETRY_BAD = '{"文字/图形属性": "T恤"}' # 缺字段，解析失败 + + +def _chat_ok(content): + return {"choices": [{"message": {"content": content}, "finish_reason": "stop"}], + "usage": {}} + + +class _FakeResp: + def __init__(self, status, data=None, body=""): + self.status = status + self._data = data + self._body = body + + async def json(self): + return self._data + + async def text(self): + return self._body + + +class _FakeCM: + def __init__(self, resp): + self._resp = resp + + async def __aenter__(self): + return self._resp + + async def __aexit__(self, *exc): + return False + + +class _FakeSession: + def __init__(self, responses): + self.responses = list(responses) + self.posts = 0 + + def post(self, url, json, headers): + resp = self.responses[min(self.posts, len(self.responses) - 1)] + self.posts += 1 + return _FakeCM(resp) + + +def _run_chat(session, retries=3): + cfg = AppConfig(output_schema_mode="off", retries=retries) + return asyncio.run(_chat_detect( + session, asyncio.Semaphore(1), cfg, "a.jpg", stop=lambda: False, + provider="deepseek", label="DeepSeek", + base_url="https://fake.deepseek.com", api_key="k", model="m", + messages=[{"role": "user", "content": "p"}], + extras={"temperature": 0, "max_tokens": 50})) + + +async def _no_sleep(_): + return None + + +def test_parse_fail_retries_then_recovers(monkeypatch): + # 第 1 次返回不可解析 → 触发重试；第 2 次成功 + monkeypatch.setattr(asyncio, "sleep", _no_sleep) + sess = _FakeSession([_FakeResp(200, _chat_ok(RETRY_BAD)), + _FakeResp(200, _chat_ok(RETRY_GOOD))]) + res = _run_chat(sess) + assert res["status"] == "ok" and res["category"] == "无违规" + assert sess.posts == 2 + + +def test_parse_fail_exhausts_to_parse_fail(monkeypatch): + # 一直解析失败 → 第 1 次进入解析段，独立再重试 3 次后落 parse_fail（违规不明） + monkeypatch.setattr(asyncio, "sleep", _no_sleep) + sess = _FakeSession([_FakeResp(200, _chat_ok(RETRY_BAD))]) + res = _run_chat(sess, retries=3) + assert res["status"] == "parse_fail" and res["category"] == "违规不明" + assert sess.posts == 1 + 3 # 首判 + 3 次独立解析重试 + + +def test_parse_fail_dedicated_budget_ignores_http_errors(monkeypatch): + # 先 1 次 HTTP 500，再解析失败：HTTP 错误不消耗解析段次数， + # 解析仍独立获得 cfg.retries 次专用重试 + monkeypatch.setattr(asyncio, "sleep", _no_sleep) + sess = _FakeSession([_FakeResp(500, body="boom"), + _FakeResp(200, _chat_ok(RETRY_BAD))]) + res = _run_chat(sess, retries=3) + assert res["status"] == "parse_fail" and res["category"] == "违规不明" + assert sess.posts == 2 + 3 # 1×HTTP500 + 首判解析失败 + 3×解析专用重试 + + +def test_error_code_retries_then_recovers(monkeypatch): + # HTTP 500 → 退避重试；随后 200 成功 + monkeypatch.setattr(asyncio, "sleep", _no_sleep) + sess = _FakeSession([_FakeResp(500, body="boom"), + _FakeResp(200, _chat_ok(RETRY_GOOD))]) + res = _run_chat(sess) + assert res["status"] == "ok" and res["category"] == "无违规" + assert sess.posts == 2 + + +def test_error_code_exhausts_to_error(monkeypatch): + # 一直 HTTP 500 → 重试耗尽落 error（检测异常） + monkeypatch.setattr(asyncio, "sleep", _no_sleep) + sess = _FakeSession([_FakeResp(500, body="boom")]) + res = _run_chat(sess, retries=3) + assert res["status"] == "error" and res["category"] == "检测异常" + assert sess.posts == 3 + + +def test_empty_content_retries(monkeypatch): + # 空正文 → 重试；随后成功 + monkeypatch.setattr(asyncio, "sleep", _no_sleep) + sess = _FakeSession([_FakeResp(200, _chat_ok("")), + _FakeResp(200, _chat_ok(RETRY_GOOD))]) + res = _run_chat(sess) + assert res["status"] == "ok" and res["category"] == "无违规" + assert sess.posts == 2 diff --git a/tests/test_report.py b/tests/test_report.py index 0b3a21c..61876a1 100644 --- a/tests/test_report.py +++ b/tests/test_report.py @@ -7,7 +7,7 @@ from violation_detector.report import build_report, organize_output, sanitize_di def ROW(cat, fname="img.jpg", channel="豆包初筛"): return {"file": fname, "category": cat, "channel": channel, "attr": "属性", - "logic": "逻辑", "status": "ok", "vote_note": "", "remark": ""} + "logic": "逻辑", "status": "ok", "remark": ""} def test_sanitize_dirname(): @@ -48,7 +48,6 @@ def test_build_report_structure(tmp_path): rows = [ROW("12. 侵权 - 除人物外的其他侵权", "a.jpg"), ROW("无违规", "b.jpg", channel="DeepSeek复检"), ROW("无违规", "b.jpg")] - rows[2]["vote_note"] = "票型2:1（无违规 / 无违规 / 12. 侵权），建议人工复核" out = build_report(rows, str(tmp_path), {"mode": "cascade", "ds_calls": 3, "ds_peak_cost": 0.1, "ds_idle_cost": 0.05}) assert out.endswith(".xlsx") @@ -56,13 +55,12 @@ def test_build_report_structure(tmp_path): wb = load_workbook(out) ws = wb["检测结果"] assert ws["B2"].value.startswith("商品图合规检测工具") - headers = [ws.cell(row=4, column=c).value for c in range(2, 10)] + headers = [ws.cell(row=4, column=c).value for c in range(2, 9)] assert headers == ["序号", "文件名", "违规分类", "判定通道", "文字/图形属性", - "侵权/违规逻辑", "票型/复核", "备注"] + "侵权/违规逻辑", "备注"] assert ws.cell(row=5, column=2).value == 1 assert ws.cell(row=6, column=5).value == "DeepSeek复检" assert ws.cell(row=5, column=5).value == "豆包初筛" - assert "人工复核" in ws.cell(row=7, column=8).value ws2 = wb["统计汇总"] cats = [ws2.cell(row=r, column=2).value for r in (5, 6)]