feat(M6): 第2环标杆8维分析引擎

8维草案(banana不变量+俊达配方+3链接):标题公式/开头钩子/正文结构/
卖点呈现/情绪语气emoji/目标人群/话题标签/视觉配图节奏。
- _benchmark_prompt.py: 8维vision prompt(不打分,提炼可复用配方)
- benchmark_analyzer.py: 引擎核心(vision读截图+手填亮点→8维JSON,失败兜底)
- benchmarks.py: POST /benchmarks/{id}/analyze 端点(写features_json,流转analyze_status)
端到端实测:8维齐全0缺失,真调LLM(conf0.9非兜底),落库done,质量高(识别收藏钩子/成分背书/诚实标未体现)
This commit is contained in:
yangqianqian
2026-06-16 14:28:17 +08:00
parent c5567614b4
commit c1ac2a6ab9
3 changed files with 159 additions and 1 deletions

View File

@@ -33,10 +33,18 @@ class BannedWordCreate(BaseModel):
def _fmt_benchmark(b: BenchmarkNote) -> dict:
import json
features = None
if b.features_json:
try:
features = json.loads(b.features_json)
except Exception:
features = None
return {
"id": b.id, "product_id": b.product_id,
"screenshot_url": b.screenshot_url,
"highlights": b.highlights, "link_url": b.link_url,
"features": features, "analyze_status": b.analyze_status,
"created_at": b.created_at.isoformat(),
}
@@ -84,7 +92,66 @@ def create_benchmark(
return ok(_fmt_benchmark(b))
# ── 违禁词库 ────────────────────────────────────────────────
@router.post("/benchmarks/{benchmark_id}/analyze")
async def analyze_benchmark_note(
benchmark_id: int,
current_user: Annotated[CurrentUser, Depends(require_admin)] = None,
db: Session = Depends(get_db),
):
"""
第2环对已存标杆笔记跑 8 维拆解。
输入源=screenshot_url 截图(vision) + highlights 手填亮点(参考)。
结果写入 features_jsonanalyze_status 流转 analyzing→done/failed。
key 链路遵基石Bplain_key 只活局部,立即清零。
"""
import json
from app.core.response import raise_business
from app.models.workspace import UserApiKey
from app.utils.fernet_utils import decrypt_key
from app.services.ai_engine.gemini_factory import build_ai_clients
from app.services.ai_engine.benchmark_analyzer import analyze_benchmark
b = db.query(BenchmarkNote).filter(
BenchmarkNote.id == benchmark_id,
BenchmarkNote.workspace_id == current_user.workspace_id,
).first()
if not b:
raise_not_found("标杆笔记不存在")
if not b.screenshot_url and not b.highlights:
raise_business("该标杆无截图也无手填亮点,无法分析")
# 读截图字节screenshot_url 存的是上传后的绝对/相对路径)
screenshot = None
if b.screenshot_url:
try:
with open(b.screenshot_url, "rb") as f:
screenshot = f.read()
except Exception as e:
logger.warning("标杆截图读取失败 id=%s: %s", benchmark_id, e)
api_key_row = db.query(UserApiKey).filter(
UserApiKey.user_id == current_user.user_id,
UserApiKey.workspace_id == current_user.workspace_id,
UserApiKey.provider.in_(["openai", "apiports"]),
).first()
if not api_key_row:
raise_business("未配置 API Key请先在设置中录入")
plain_key = decrypt_key(api_key_row.encrypted_key)
clients = build_ai_clients(plain_key)
plain_key = None
b.analyze_status = "analyzing"; db.commit()
try:
result = await analyze_benchmark(clients, screenshot, b.highlights)
finally:
await clients.aclose()
b.features_json = json.dumps(result, ensure_ascii=False)
b.analyze_status = "done" if result.get("source") != "fallback" else "failed"
db.commit(); db.refresh(b)
logger.info("benchmark analyzed id=%s status=%s source=%s",
benchmark_id, b.analyze_status, result.get("source"))
return ok({**_fmt_benchmark(b), "features": result, "analyze_status": b.analyze_status})
@router.get("/banned-words")
def list_banned_words(
page: int = 1, page_size: int = 50,

View File

@@ -0,0 +1,25 @@
"""
app/services/ai_engine/_benchmark_prompt.py — 第2环标杆8维分析 prompt
8维草案来源倩倩姐2026-06-16banana不变量/可变量方法论 + 俊达拆解中心配方 + 3小红书链接。
方法论核心:分析目的不是打分,是提炼"仿写时必须保留的配方"不变量喂给第5环文案+第6环配图对标。
输入=爆款笔记截图GPT vision 读图)+ 运营手填亮点highlights
"""
BENCHMARK_8DIM_PROMPT = """你是小红书爆款拆解专家。请把这篇爆款笔记拆成"可复用配方",供后续对标仿写。
硬性规则:只分析本次提供的截图与亮点描述,不臆造、不沿用历史案例、不默认任何品类。
看不清或无法判断的维度,填"截图未体现",不要编。
请从以下8个维度拆解返回JSON仅JSON无其他文字
{
"title_formula": "标题用了什么套路(提问式/数字式/悬念式/对比式/利益前置等)+大致字数",
"opening_hook": "开头前1-3句如何抓人痛点直击/反差/利益前置/场景代入等),摘录关键句",
"content_structure": "正文骨架顺序(如 痛点-方案-效果 / 故事-产品-号召 / 测评-对比-结论)",
"selling_point_style": "卖点怎么呈现(几个卖点、轻重排布、是否数据化/场景化、植入方式)",
"emotion_tone": "情绪语气(亲切/专业/俏皮/种草感/姐妹分享等)+ emoji使用规律频率/类型/位置)",
"target_audience": "这篇打给谁(年龄段/使用场景/核心痛点画像),一句话",
"topic_tags": "话题标签策略选了哪类tag、几个、大词与精准小词的配比",
"visual_rhythm": "视觉配图配方(封面风格、构图、色调、光线、配图排布节奏、是否压字/大字报模板)",
"confidence": 0.9,
"source": "vision"
}"""

View File

@@ -0,0 +1,66 @@
"""
app/services/ai_engine/benchmark_analyzer.py — 第2环 标杆8维分析引擎
复用 products.analyze_product_image 的 vision 范式,解耦成 service。
输入:爆款截图 bytes + 手填亮点输出8维配方 JSON写入 benchmark_notes.features_json
key 链路遵基石Bplain_key 只活在调用方局部,本模块只收已构建的 clients。
"""
import json
import logging
import re
from app.services.ai_engine._benchmark_prompt import BENCHMARK_8DIM_PROMPT
logger = logging.getLogger(__name__)
# 8维字段顺序与 prompt JSON key 一致analyze_status 流转 done 时校验齐全)
DIM_KEYS = [
"title_formula", "opening_hook", "content_structure", "selling_point_style",
"emotion_tone", "target_audience", "topic_tags", "visual_rhythm",
]
def parse_benchmark_json(raw: str) -> dict:
"""从模型返回文本提取 JSON容错 markdown ```json 包裹)。"""
cleaned = re.sub(r"```json\s*", "", raw)
cleaned = re.sub(r"```\s*", "", cleaned)
m = re.search(r"\{[\s\S]*\}", cleaned)
if not m:
raise ValueError("标杆分析返回中未找到 JSON")
return json.loads(m.group())
def fallback_benchmark(highlights: str | None = None) -> dict:
"""vision 失败兜底:保证 8 维不空手,标 source=fallback 供前端提示。"""
note = highlights or "截图未体现"
return {
"title_formula": note, "opening_hook": note, "content_structure": note,
"selling_point_style": note, "emotion_tone": note, "target_audience": note,
"topic_tags": note, "visual_rhythm": note,
"confidence": 0.4, "source": "fallback",
}
async def analyze_benchmark(clients, screenshot: bytes | None, highlights: str | None) -> dict:
"""
跑 8 维分析。截图走 vision无截图但有手填亮点则把亮点拼进 prompt 兜底。
返回结构化 8 维 dict含 confidence/source。失败不抛返回 fallback。
"""
prompt = BENCHMARK_8DIM_PROMPT
if highlights:
prompt = f"{prompt}\n\n运营补充的亮点描述(作为分析参考):{highlights}"
images = [screenshot] if screenshot else []
try:
if not images:
# 无截图纯文本兜底gpt_vision_analyze 支持空图走文本)
logger.info("analyze_benchmark: 无截图,走手填亮点文本分析")
raw = await clients.gpt_vision_analyze(prompt, images)
result = parse_benchmark_json(raw)
result["source"] = "vision" if images else "text"
return result
except Exception as e:
logger.warning("analyze_benchmark failed, fallback: %s", e)
result = fallback_benchmark(highlights)
result["warning"] = f"标杆分析失败,已兜底:{e}"
return result