"""内容质量评估工具函数 —— Prompt 构建 + LLM 响应解析""" import hashlib import re import logging from typing import Dict, List from src.infra.shared.tools import safe_json_parse from ._const import ContentQualityConst logger = logging.getLogger(__name__) def md5(text: str) -> str: return hashlib.md5(text.encode("utf-8")).hexdigest() # ═══════════════════════════════════════════════════════════════ # Prompt 构建 # ═══════════════════════════════════════════════════════════════ def build_title_quality_prompt(titles: List[Dict]) -> str: """构造标题质量评分 prompt,每行一个标题""" lines = [t["title"] for t in titles] return ContentQualityConst.PROMPT_TITLE_QUALITY.format( title_list="\n".join(lines), ) def build_category_prompt(titles: List[Dict]) -> str: """构造品类识别 prompt,格式: [(id, title), ...] 元组列表""" tuples = [(t["id"], t["title"]) for t in titles] return ContentQualityConst.PROMPT_TITLE_CATEGORY.format( title_list=str(tuples), ) # ═══════════════════════════════════════════════════════════════ # LLM 响应解析 # ═══════════════════════════════════════════════════════════════ def parse_title_quality_response( raw: str | None, batch_titles: List[Dict], ) -> List[Dict]: """解析 prompt1 返回值:LLM 按行输出纯数字分数,按顺序对应输入标题 返回: [{"id": 1, "score": 85}, ...] """ result: List[Dict] = [] if not raw: logger.warning("LLM 标题质量返回为空") return result # 从文本中提取所有数字 scores = re.findall(r"\d+", raw) if not scores: logger.warning("LLM 标题质量未提取到分数: %s", raw[:200]) return result for i, s in enumerate(scores): if i >= len(batch_titles): break score = int(s) if 0 <= score <= 100: result.append({"id": batch_titles[i]["id"], "score": score}) return result def parse_category_response( raw: str | None, batch_titles: List[Dict], ) -> List[Dict]: """解析 prompt2 返回值:{id: category} JSON 对象 返回: [{"id": 1, "category": "知识科普"}, ...] """ result: List[Dict] = [] if not raw: logger.warning("LLM 品类识别返回为空") return result parsed = safe_json_parse(raw) if not isinstance(parsed, dict): logger.warning("LLM 品类识别返回非对象: %s", raw[:200]) return result for t in batch_titles: category = parsed.get(str(t["id"])) if category: result.append({"id": t["id"], "category": str(category)[:32]}) return result __all__ = [ "md5", "build_title_quality_prompt", "build_category_prompt", "parse_title_quality_response", "parse_category_response", ]