A8: 文案按A/B/C三套正交叙事生成,避免套路化重复

- constants: 新增 TEXT_NARRATIVE_BY_STRATEGY(A痛点/B场景/C成分),与图片侧同轴
- build_prompt: 加 strategy_narrative 参数并注入 prompt
- text_variants: 全链路透传(含优化轮)
- run_text_generation: 改循环三套,text_count均摊(divmod余前补),跨套去重,打_strategy标记
- TextCandidate: 加 strategy String(4) 字段 + 迁移021(已upgrade head)
- packaging: 打包按strategy精准配对文图(texts_by_strategy映射+三层兜底)
- SSE text_candidate 事件携带 strategy

独立agent交叉验证7改造点全过,边界(text_count<3/无别名/不截断)无must-fix

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/alembic/versions/021_text_candidate_strategy.py

@@ -0,0 +1,30 @@
+"""021 text_candidates 表加 strategy 正交叙事套字段（A8 三套不同角度文案）
+
+Revision ID: 021
+Revises: 020
+Create Date: 2026-06-18
+
+A8(倩倩姐2026-06-18拍板「方向对，按这个写」)：文案按 A/B/C 三套正交叙事
+分别生成(A痛点先行/B场景先行/C成分背书先行)，与图片侧 strategy 同轴。
+text_candidates 加 strategy 字段记录每条属哪套，供前端分组展示 + 打包按套精准
+匹配文图(不再靠脆弱 idx 对应)。旧数据 strategy=NULL 不影响展示。
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "021"
+down_revision = "020"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "text_candidates",
+        sa.Column("strategy", sa.String(length=4), nullable=True,
+                  comment="A/B/C 正交叙事套(A痛点/B场景/C成分),与图片侧同轴"),
+    )
+
+
+def downgrade():
+    op.drop_column("text_candidates", "strategy")

backend/app/models/task.py

@@ -84,6 +84,7 @@ class TextCandidate(Base):
         default=CandidateSource.AI, nullable=False,
     )
     angle_label: Mapped[str | None] = mapped_column(String(64))
+    strategy: Mapped[str | None] = mapped_column(String(4))  # A/B/C 正交叙事套(A痛点/B场景/C成分),与图片侧同轴
     content: Mapped[str | None] = mapped_column(Text)
     score_json: Mapped[str | None] = mapped_column(Text)    # 五维分 JSON
     banned_word_status: Mapped[str] = mapped_column(

backend/app/services/ai_engine/_text_prompt.py

@@ -163,11 +163,13 @@ def _build_benchmark_block(refs: list[dict]) -> str:
     )
 
 
-def build_prompt(product: dict, count: int, extra_rules: str = "") -> str:
+def build_prompt(product: dict, count: int, extra_rules: str = "",
+                 strategy_narrative: str = "") -> str:
     """
     组装文案生成 user_prompt。
     数据层：product 动态注入（name/selling_points/style_tone/text_angles/custom_prompt）
     方法层：已在 COPY_SYSTEM 固定，这里只注入产品数据+随机变量
+    strategy_narrative：三套正交叙事主线（A痛点/B场景/C成分），由调用方按套传入
     """
     name     = product.get("name", "产品")
     selling  = "、".join(product.get("selling_points") or ["核心卖点待录入"])
@@ -191,6 +193,7 @@ def build_prompt(product: dict, count: int, extra_rules: str = "") -> str:
         f"产品：{name}",
         f"核心卖点（必须翻译成用户能感知的生活化利益，禁止直接列功效词；翻译范例：'烟酰胺'→'熬夜后第二天脸不那么黄了'，'高保湿'→'涂上去一整天都没搓泥拔干'）：{selling}",
         f"风格调性：{style}",
+        strategy_narrative,
         angle_hint,
         brand_rule,
         custom,

backend/app/services/ai_engine/constants.py

@@ -125,8 +125,28 @@ NARRATIVE_BY_STRATEGY = {
     ),
 }
 
-# ── 生图通道 ──────────────────────────────────────────────
-IMAGE_RETRY_ATTEMPTS = 3
+# ── 文案3套正交叙事策略（倩倩姐2026-06-18过目版，与图片侧同A/B/C轴）──────
+# 注入 build_prompt，让三套文案各走不同叙事主线，避免套路化重复
+# 与 NARRATIVE_BY_STRATEGY（图片侧）同根：套A文案痛点先行↔套A图也痛点先行，同套内文图一致
+TEXT_NARRATIVE_BY_STRATEGY = {
+    "A": (
+        "【本条叙事主线·痛点先行】开篇直戳用户困扰（脸黄显疲惫/素颜不敢出门/早八顶着黄脸），"
+        "情绪强、句子短促带感叹，痛点贯穿全文到种草，结尾用'别再顶着黄脸早八'这类痛点收束促单。"
+        "基调：紧迫感、强对比、情绪共鸣。"
+    ),
+    "B": (
+        "【本条叙事主线·场景先行】用真实生活场景开篇（早八来不及/通勤手忙脚乱/赶时间出门），"
+        "轻松代入感，突出'快/省时/伪素颜自由'，点到平价性价比但不堆砌。"
+        "基调：轻松、生活化、像朋友随手分享。"
+    ),
+    "C": (
+        "【本条叙事主线·成分背书先行】用成分原理或测评视角开篇（核心成分为什么有用/亲测对比），"
+        "专业可信，带使用前后时间线对比，像真实用户实证背书，结尾用'成分党闭眼入'收束。"
+        "基调：专业、可信、真实测评感。"
+    ),
+}
+
+
 IMAGE_RETRY_BACKOFF_BASE = 2.0  # 指数退避底数（秒）
 IMAGE_SIZE_DEFAULT = "1024x1536"
 

backend/app/services/ai_engine/text_variants.py

@@ -70,12 +70,15 @@ async def _call_llm(client: Any, prompt: str, max_tokens: int = 8192) -> str:
 TEXT_BATCH_SIZE = int(os.environ.get("TEXT_BATCH_SIZE", "4"))
 
 
-async def _generate_one_batch(llm_client: Any, product: dict, batch_n: int, extra: str) -> list[dict]:
+async def _generate_one_batch(llm_client: Any, product: dict, batch_n: int, extra: str,
+                              strategy_narrative: str = "") -> list[dict]:
     """生成一批 batch_n 条，含解析重试（最多2次）。失败返回空列表。
     max_tokens 按条数缩放(每条约 1800 token，封顶 8192)，压进 apiports 60s 网关窗口。"""
     batch_max_tokens = min(8192, max(1800, batch_n * 1800))
     for attempt in range(2):
-        raw = await _call_llm(llm_client, build_prompt(product, batch_n, extra_rules=extra), batch_max_tokens)
+        raw = await _call_llm(llm_client, build_prompt(
+            product, batch_n, extra_rules=extra, strategy_narrative=strategy_narrative,
+        ), batch_max_tokens)
         parsed = parse_json_array(raw)
         if parsed:
             return parsed
@@ -84,7 +87,8 @@ async def _generate_one_batch(llm_client: Any, product: dict, batch_n: int, extr
     return []
 
 
-async def _generate_in_batches(llm_client: Any, product: dict, count: int, extra: str) -> list[dict]:
+async def _generate_in_batches(llm_client: Any, product: dict, count: int, extra: str,
+                               strategy_narrative: str = "") -> list[dict]:
     """把 count 条按 TEXT_BATCH_SIZE 分批，串行调用合并。
     串行而非并发：opus 单批就慢(~300s)且 apiports 限并发，多批 gather 会触发
     大面积 503 雪崩(task45 实测)。故改串行，墙钟换稳定。"""
@@ -96,7 +100,7 @@ async def _generate_in_batches(llm_client: Any, product: dict, count: int, extra
         remaining -= n
     collected: list[dict] = []
     for n in sizes:
-        r = await _generate_one_batch(llm_client, product, n, extra)
+        r = await _generate_one_batch(llm_client, product, n, extra, strategy_narrative)
         collected.extend(r)
     return collected
 
@@ -108,12 +112,16 @@ async def generate_text_variants(
     previous_copies: list[dict] | None = None,
     banned_word_rows: list[dict] | None = None,
     flywheel_context: str = "",
+    strategy_narrative: str = "",
 ) -> list[dict]:
-    """轨A：一次出 count 条不同角度文案，三层兜底，自动优化循环"""
+    """轨A：一次出 count 条不同角度文案，三层兜底，自动优化循环。
+    strategy_narrative：本套正交叙事主线(A痛点/B场景/C成分)，由调用方按套传入，
+    贯穿首批生成与优化轮，确保同套内文案同一叙事不串味。"""
     banned_entries = build_entries_from_db(banned_word_rows or [])
     extra = flywheel_context
 
-    copies: list[dict] = await _generate_in_batches(llm_client, product, count, extra)
+    copies: list[dict] = await _generate_in_batches(
+        llm_client, product, count, extra, strategy_narrative)
     if not copies:
         copies = list(build_local_drafts(product, count))  # generator → list
 
@@ -149,6 +157,7 @@ async def generate_text_variants(
         raw2 = await _call_llm(llm_client, build_prompt(
             product, len(batch_failed),
             extra_rules=f"以下文案未达标，请重新生成并改进：\n{hint}\n不要重复已有标题和角度。",
+            strategy_narrative=strategy_narrative,
         ), min(8192, max(1800, len(batch_failed) * 1800)))
         if not raw2:
             # LLM 失败（如 503/超时）：优化是锦上添花，原始候选已够用，不再耗时重试

backend/app/workers/packaging_task.py

@@ -80,12 +80,27 @@ def build_delivery_package(self, package_id: int) -> dict:
             if prev is None or ic.id > prev.id:
                 slot[ic.seq] = ic  # 同 seq 留最新
 
+        # 文案按 strategy 建映射，供图片组按套精准配对（A8：文图同套对齐，不靠脆弱 idx）。
+        # 同套多条选中取第 1 条；老数据 strategy=None 的归入 fallback 列表。
+        texts_by_strategy: dict = {}
+        texts_no_strategy: list = []
+        for tc in selected_texts:
+            if tc.strategy:
+                texts_by_strategy.setdefault(tc.strategy, tc)
+            else:
+                texts_no_strategy.append(tc)
+        _text_fallback = iter(texts_no_strategy or selected_texts)
+
         notes = []
         for idx, (_strategy, slot) in enumerate(groups.items()):
             images_data = [_read_image(slot[k]) for k in sorted(slot)]
-            # 文案配对：选中文案数≥套数则一套一条；否则各套共用第 1 条
-            # (图均以第 1 条文案为语境生成，共用合理；多选则尊重运营按套选的文案)
-            tc = selected_texts[idx] if idx < len(selected_texts) else selected_texts[0]
+            # 配对优先级：①同 strategy 文案精准对齐 ②无同套则按顺序取无套文案
+            # ③再兜底用第 idx 条/第 1 条，确保每组图都有文案不漏。
+            tc = texts_by_strategy.get(_strategy)
+            if tc is None:
+                tc = next(_text_fallback, None)
+            if tc is None:
+                tc = selected_texts[idx] if idx < len(selected_texts) else selected_texts[0]
             text_data = json.loads(tc.content or "{}")
             notes.append({
                 "title": text_data.get("title", ""),

backend/app/workers/pipeline_io.py

@@ -46,7 +46,7 @@ def run_text_generation(db, clients, task, product_dict: dict, flywheel_fragment
     from app.models.flywheel import AiCallLog
     from app.models.workspace import UserApiKey
     from app.constants.enums import CandidateSource, BannedWordStatus
-    from app.services.ai_engine.constants import QUALITY_PASS_SCORE
+    from app.services.ai_engine.constants import QUALITY_PASS_SCORE, TEXT_NARRATIVE_BY_STRATEGY
 
     banned_rows = db.query(BannedWord).filter(
         BannedWord.workspace_id == workspace_id
@@ -61,20 +61,37 @@ def run_text_generation(db, clients, task, product_dict: dict, flywheel_fragment
     ).first()
     key_id = key_row.id if key_row else None
 
+    # A8：文案按 A/B/C 三套正交叙事分别生成，每套不同角度避免套路化重复，
+    # 与图片侧同轴(A痛点/B场景/C成分)。text_count 均摊三套(余数前补)，
+    # 逐套把已生成的喂作 previous_copies 做跨套去重。
+    _strategies = ("A", "B", "C")
+    _base, _rem = divmod(task.text_count, 3)
+    _per = {s: _base + (1 if i < _rem else 0) for i, s in enumerate(_strategies)}
+
     t0 = time.monotonic()
     llm_success = True
-    try:
-        candidates_raw = asyncio.run(generate_text_variants(
-            llm_client=clients,
-            product=product_dict,
-            count=task.text_count,
-            banned_word_rows=banned_dicts,
-            flywheel_context=flywheel_fragment,
-        ))
-    except Exception as exc:
-        llm_success = False
-        logger.error("generate_text_variants 失败: %s", exc)
-        candidates_raw = []
+    candidates_raw: list = []
+    for s in _strategies:
+        n = _per[s]
+        if n <= 0:
+            continue
+        try:
+            part = asyncio.run(generate_text_variants(
+                llm_client=clients,
+                product=product_dict,
+                count=n,
+                previous_copies=candidates_raw,
+                banned_word_rows=banned_dicts,
+                flywheel_context=flywheel_fragment,
+                strategy_narrative=TEXT_NARRATIVE_BY_STRATEGY.get(s, ""),
+            ))
+        except Exception as exc:
+            llm_success = False
+            logger.error("generate_text_variants(套%s) 失败: %s", s, exc)
+            part = []
+        for c in part:
+            c["_strategy"] = s
+        candidates_raw.extend(part)
     latency_ms = int((time.monotonic() - t0) * 1000)
 
     # 写 ai_call_logs（留痕，不含明文key）
@@ -113,6 +130,7 @@ def run_text_generation(db, clients, task, product_dict: dict, flywheel_fragment
             workspace_id=workspace_id,
             task_id=task.id,
             source=CandidateSource.AI,
+            strategy=c.get("_strategy"),
             angle_label=c.get("angle_label") or c.get("angle", ""),
             content=json.dumps(c, ensure_ascii=False),
             score_json=json.dumps(c.get("score_detail", []), ensure_ascii=False),
@@ -124,6 +142,7 @@ def run_text_generation(db, clients, task, product_dict: dict, flywheel_fragment
         seq += 1
         push_fn(task.id, workspace_id, "text_candidate", {
             "candidate_id": tc.id, "angle_label": tc.angle_label,
+            "strategy": tc.strategy,
             "content": c.get("content", ""), "score": score,
         }, seq)
         seq += 1