|
@@ -1,7 +1,6 @@
|
|
|
"""内容质量评估工具函数 —— Prompt 构建 + LLM 响应解析"""
|
|
"""内容质量评估工具函数 —— Prompt 构建 + LLM 响应解析"""
|
|
|
|
|
|
|
|
import hashlib
|
|
import hashlib
|
|
|
-import re
|
|
|
|
|
import logging
|
|
import logging
|
|
|
from typing import Dict, List
|
|
from typing import Dict, List
|
|
|
|
|
|
|
@@ -22,10 +21,10 @@ def md5(text: str) -> str:
|
|
|
|
|
|
|
|
|
|
|
|
|
def build_title_quality_prompt(titles: List[Dict]) -> str:
|
|
def build_title_quality_prompt(titles: List[Dict]) -> str:
|
|
|
- """构造标题质量评分 prompt,每行一个标题"""
|
|
|
|
|
- lines = [t["title"] for t in titles]
|
|
|
|
|
|
|
+ """构造标题质量评分 prompt,格式: [(id, title), ...] 元组列表"""
|
|
|
|
|
+ tuples = [(t["id"], t["title"]) for t in titles]
|
|
|
return ContentQualityConst.PROMPT_TITLE_QUALITY.format(
|
|
return ContentQualityConst.PROMPT_TITLE_QUALITY.format(
|
|
|
- title_list="\n".join(lines),
|
|
|
|
|
|
|
+ title_list=str(tuples),
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
@@ -46,7 +45,10 @@ def parse_title_quality_response(
|
|
|
raw: str | None,
|
|
raw: str | None,
|
|
|
batch_titles: List[Dict],
|
|
batch_titles: List[Dict],
|
|
|
) -> List[Dict]:
|
|
) -> List[Dict]:
|
|
|
- """解析 prompt1 返回值:LLM 按行输出纯数字分数,按顺序对应输入标题
|
|
|
|
|
|
|
+ """解析标题质量评分返回值:{id: score} JSON 对象,key 为标题 id
|
|
|
|
|
+
|
|
|
|
|
+ 与品类识别 prompt 对齐——ID 显式绑定,消除位置依赖导致的分数错位。
|
|
|
|
|
+ 如果某条标题在 LLM 响应中缺失 key 或值非法,则静默跳过(不影响同批次其他标题)。
|
|
|
|
|
|
|
|
返回: [{"id": 1, "score": 85}, ...]
|
|
返回: [{"id": 1, "score": 85}, ...]
|
|
|
"""
|
|
"""
|
|
@@ -55,18 +57,21 @@ def parse_title_quality_response(
|
|
|
logger.warning("LLM 标题质量返回为空")
|
|
logger.warning("LLM 标题质量返回为空")
|
|
|
return result
|
|
return result
|
|
|
|
|
|
|
|
- # 从文本中提取所有数字
|
|
|
|
|
- scores = re.findall(r"\d+", raw)
|
|
|
|
|
- if not scores:
|
|
|
|
|
- logger.warning("LLM 标题质量未提取到分数: %s", raw[:200])
|
|
|
|
|
|
|
+ parsed = safe_json_parse(raw)
|
|
|
|
|
+ if not isinstance(parsed, dict):
|
|
|
|
|
+ logger.warning("LLM 标题质量返回非 JSON 对象: %s", raw[:200])
|
|
|
return result
|
|
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})
|
|
|
|
|
|
|
+ for t in batch_titles:
|
|
|
|
|
+ score_val = parsed.get(str(t["id"]))
|
|
|
|
|
+ if score_val is None:
|
|
|
|
|
+ continue
|
|
|
|
|
+ try:
|
|
|
|
|
+ score = int(score_val)
|
|
|
|
|
+ if 0 <= score <= 100:
|
|
|
|
|
+ result.append({"id": t["id"], "score": score})
|
|
|
|
|
+ except (ValueError, TypeError):
|
|
|
|
|
+ pass
|
|
|
|
|
|
|
|
return result
|
|
return result
|
|
|
|
|
|