外部文章质量判断

src/domains/content_quality/article_quality_category.py

@@ -1,12 +1,12 @@
-"""品类识别 Service —— 种子同步 + 批量 LLM 识别"""
+"""品类识别 Service —— 种子同步 + 并发批量 LLM 识别"""
 
 import asyncio
 import logging
-
-from tqdm import tqdm
+from typing import Dict
 
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.external.volcengine import fetch_deepseek_completion
+from src.infra.shared.async_tasks import run_tasks_with_asyncio_task_group
 
 from ._const import ContentQualityConst
 from ._mapper import QualityMapper
@@ -19,6 +19,8 @@ class ArticleQualityCategory(ContentQualityConst):
 
     def __init__(self, pool: MysqlManager):
         self.mapper = QualityMapper(pool)
+        self._evaluated = 0
+        self._lock = asyncio.Lock()
 
     async def deal(self, *, limit: int | None = None) -> dict:
         _limit = limit if limit is not None else self.CATEGORY_DEAL_LIMIT
@@ -33,44 +35,55 @@ class ArticleQualityCategory(ContentQualityConst):
             logger.info("无待识别品类")
             return {"seeded": seeded, "evaluated": 0}
 
-        evaluated = 0
-        with tqdm(total=len(rows), desc="品类识别", unit="条") as pbar:
-            for start in range(0, len(rows), self.LLM_BATCH_SIZE):
-                batch = rows[start : start + self.LLM_BATCH_SIZE]
-                batch_ids = [r["id"] for r in batch]
-
-                affected = await self.mapper.mark_category_processing(batch_ids)
-                if affected == 0:
-                    pbar.update(len(batch))
-                    continue
-
-                titles = [{"id": r["id"], "title": r["title"]} for r in batch]
-                prompt = build_category_prompt(titles)
-
-                try:
-                    raw = await self._call_llm(prompt)
-                    if raw is None:
-                        raise RuntimeError("LLM 返回 None")
-
-                    parsed = parse_category_response(raw, titles)
-                    if not parsed:
-                        raise RuntimeError("LLM 响应解析为空")
-
-                    await self.mapper.batch_update_category(parsed)
-                    evaluated += len(parsed)
-
-                except Exception:
-                    logger.exception("品类识别失败 batch_start=%d", start)
-                    await self.mapper.mark_category_batch_failed(batch_ids)
-
-                pbar.update(len(batch))
-                pbar.set_postfix(evaluated=evaluated)
-
-                if start + self.LLM_BATCH_SIZE < len(rows):
-                    await asyncio.sleep(self.LLM_INTERVAL_SEC)
-
-        logger.info("品类识别完成: seeded=%d, evaluated=%d", seeded, evaluated)
-        return {"seeded": seeded, "evaluated": evaluated}
+        batches = []
+        for start in range(0, len(rows), self.LLM_BATCH_SIZE):
+            batch = rows[start : start + self.LLM_BATCH_SIZE]
+            batch_ids = [r["id"] for r in batch]
+            titles = [{"id": r["id"], "title": r["title"]} for r in batch]
+            prompt = build_category_prompt(titles)
+            batches.append({
+                "batch_ids": batch_ids,
+                "titles": titles,
+                "prompt": prompt,
+            })
+
+        # 先一次性标记 PROCESSING，避免并发 CAS 冲突
+        for b in batches:
+            await self.mapper.mark_category_processing(b["batch_ids"])
+
+        result = await run_tasks_with_asyncio_task_group(
+            batches,
+            self._evaluate_one_batch,
+            description="品类识别",
+            unit="batch",
+            max_concurrency=5,
+            show_progress=True,
+        )
+
+        logger.info(
+            "品类识别完成: seeded=%d, batches=%d, evaluated=%d, errors=%d",
+            seeded, len(batches), self._evaluated, len(result["errors"]),
+        )
+        return {"seeded": seeded, "evaluated": self._evaluated}
+
+    async def _evaluate_one_batch(self, batch: Dict) -> None:
+        try:
+            raw = await self._call_llm(batch["prompt"])
+            if raw is None:
+                raise RuntimeError("LLM 返回 None")
+
+            parsed = parse_category_response(raw, batch["titles"])
+            if not parsed:
+                raise RuntimeError("LLM 响应解析为空")
+
+            await self.mapper.batch_update_category(parsed)
+
+            async with self._lock:
+                self._evaluated += len(parsed)
+
+        except Exception:
+            logger.exception("品类识别失败 batch_ids=%s", batch["batch_ids"][:3])
+            await self.mapper.mark_category_batch_failed(batch["batch_ids"])
 
     async def _call_llm(self, prompt: str) -> str | None:
         loop = asyncio.get_running_loop()

src/domains/content_quality/article_quality_title.py

@@ -1,12 +1,12 @@
-"""标题质量评分 Service —— 种子同步 + 批量 LLM 评分"""
+"""标题质量评分 Service —— 种子同步 + 并发批量 LLM 评分"""
 
 import asyncio
 import logging
-
-from tqdm import tqdm
+from typing import Dict, List
 
 from src.infra.database.mysql.manager import MysqlManager
 from src.infra.external.volcengine import fetch_deepseek_completion
+from src.infra.shared.async_tasks import run_tasks_with_asyncio_task_group
 
 from ._const import ContentQualityConst
 from ._mapper import QualityMapper
@@ -19,6 +19,8 @@ class ArticleQualityTitle(ContentQualityConst):
 
     def __init__(self, pool: MysqlManager):
         self.mapper = QualityMapper(pool)
+        self._evaluated = 0
+        self._lock = asyncio.Lock()
 
     async def deal(self, *, limit: int | None = None) -> dict:
         _limit = limit if limit is not None else self.TITLE_DEAL_LIMIT
@@ -33,50 +35,62 @@ class ArticleQualityTitle(ContentQualityConst):
             logger.info("无待评分标题")
             return {"seeded": seeded, "evaluated": 0}
 
-        evaluated = 0
-        batches = (len(rows) + self.LLM_BATCH_SIZE - 1) // self.LLM_BATCH_SIZE
-        with tqdm(total=len(rows), desc="标题评分", unit="条") as pbar:
-            for start in range(0, len(rows), self.LLM_BATCH_SIZE):
-                batch = rows[start : start + self.LLM_BATCH_SIZE]
-                batch_ids = [r["id"] for r in batch]
-
-                affected = await self.mapper.mark_title_processing(batch_ids)
-                if affected == 0:
-                    pbar.update(len(batch))
-                    continue
-
-                titles = [{"id": r["id"], "title": r["title"]} for r in batch]
-                prompt = build_title_quality_prompt(titles)
-
-                try:
-                    raw = await self._call_llm(prompt)
-                    if raw is None:
-                        raise RuntimeError("LLM 返回 None")
-
-                    parsed = parse_title_quality_response(raw, titles)
-                    if not parsed:
-                        raise RuntimeError("LLM 响应解析为空")
+        # 拆批
+        batches = []
+        for start in range(0, len(rows), self.LLM_BATCH_SIZE):
+            batch = rows[start : start + self.LLM_BATCH_SIZE]
+            batch_ids = [r["id"] for r in batch]
+            titles = [{"id": r["id"], "title": r["title"]} for r in batch]
+            prompt = build_title_quality_prompt(titles)
+            batches.append({
+                "batch_ids": batch_ids,
+                "titles": titles,
+                "prompt": prompt,
+            })
+
+        # 并发执行（先标记 PROCESSING，避免并发冲突）
+        for b in batches:
+            await self.mapper.mark_title_processing(b["batch_ids"])
+
+        result = await run_tasks_with_asyncio_task_group(
+            batches,
+            self._evaluate_one_batch,
+            description="标题评分",
+            unit="batch",
+            max_concurrency=5,
+            show_progress=True,
+        )
+
+        # 回写 relation.quality_score
+        if self._evaluated > 0:
+            backfilled = await self.mapper.backfill_relation_quality_score()
+            logger.info("relation 质量分回写: %d 行", backfilled)
 
-                    await self.mapper.batch_update_title_score(parsed)
-                    evaluated += len(parsed)
+        logger.info(
+            "标题评分完成: seeded=%d, batches=%d, evaluated=%d, errors=%d",
+            seeded, len(batches), self._evaluated, len(result["errors"]),
+        )
+        return {"seeded": seeded, "evaluated": self._evaluated}
 
-                except Exception:
-                    logger.exception("标题评分失败 batch_start=%d", start)
-                    await self.mapper.mark_title_batch_failed(batch_ids)
+    async def _evaluate_one_batch(self, batch: Dict) -> None:
+        """单个批次：调 LLM → 解析 → 回写"""
+        try:
+            raw = await self._call_llm(batch["prompt"])
+            if raw is None:
+                raise RuntimeError("LLM 返回 None")
 
-                pbar.update(len(batch))
-                pbar.set_postfix(evaluated=evaluated)
+            parsed = parse_title_quality_response(raw, batch["titles"])
+            if not parsed:
+                raise RuntimeError("LLM 响应解析为空")
 
-                if start + self.LLM_BATCH_SIZE < len(rows):
-                    await asyncio.sleep(self.LLM_INTERVAL_SEC)
+            await self.mapper.batch_update_title_score(parsed)
 
-        # 回写 relation.quality_score，让详情/账号拉取按质量分排序
-        if evaluated > 0:
-            backfilled = await self.mapper.backfill_relation_quality_score()
-            logger.info("relation 质量分回写: %d 行", backfilled)
+            async with self._lock:
+                self._evaluated += len(parsed)
 
-        logger.info("标题评分完成: seeded=%d, evaluated=%d", seeded, evaluated)
-        return {"seeded": seeded, "evaluated": evaluated}
+        except Exception:
+            logger.exception("标题评分失败 batch_ids=%s", batch["batch_ids"][:3])
+            await self.mapper.mark_title_batch_failed(batch["batch_ids"])
 
     async def _call_llm(self, prompt: str) -> str | None:
         loop = asyncio.get_running_loop()