diff --git a/backend/alembic/versions/021_text_candidate_strategy.py b/backend/alembic/versions/021_text_candidate_strategy.py new file mode 100644 index 0000000..8f4b626 --- /dev/null +++ b/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") diff --git a/backend/app/models/task.py b/backend/app/models/task.py index 09d4989..ddd3cf1 100644 --- a/backend/app/models/task.py +++ b/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( diff --git a/backend/app/services/ai_engine/_text_prompt.py b/backend/app/services/ai_engine/_text_prompt.py index f3b5d9b..fc19e59 100644 --- a/backend/app/services/ai_engine/_text_prompt.py +++ b/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, diff --git a/backend/app/services/ai_engine/constants.py b/backend/app/services/ai_engine/constants.py index fcf6e88..76b428c 100644 --- a/backend/app/services/ai_engine/constants.py +++ b/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" diff --git a/backend/app/services/ai_engine/text_variants.py b/backend/app/services/ai_engine/text_variants.py index 2bb3420..1d824d0 100644 --- a/backend/app/services/ai_engine/text_variants.py +++ b/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/超时）：优化是锦上添花，原始候选已够用，不再耗时重试 diff --git a/backend/app/workers/packaging_task.py b/backend/app/workers/packaging_task.py index 53541f0..53de92c 100644 --- a/backend/app/workers/packaging_task.py +++ b/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", ""), diff --git a/backend/app/workers/pipeline_io.py b/backend/app/workers/pipeline_io.py index 0fbcfbb..a06308b 100644 --- a/backend/app/workers/pipeline_io.py +++ b/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 diff --git a/plans/核销-第12环-归档.md b/plans/核销-第12环-归档.md index 25d17c1..73f2e6b 100644 --- a/plans/核销-第12环-归档.md +++ b/plans/核销-第12环-归档.md @@ -27,7 +27,7 @@ - [ ] **A5**〔human〕下载交付包：approved 任务点"下载交付包"，≤60s 触发下载 delivery-{id}.zip(>1KB)；`unzip -l` 含 note_01/文案.txt/发布清单/合规说明。 - [x] **A6**〔auto〕zip 结构完整：✅2026-06-18 task74端到端验:zip解出note_01/02/03各7项(6图+文案.txt),发布清单+合规说明在root,18张图全非空真写入(磁盘文件真读到)。 - [x] **A7**〔auto〕文案.txt 格式：✅2026-06-18 task74验:note_02/文案.txt以"【标题】懒人三分钟搞定,倍分子上脸气色回来了"开头(非{,JSON已解析),品牌词"倍分子"已植入。 -- [x] **A8**〔human+auto〕多套打包：✅2026-06-18 task74(6图×3套A/B/C)端到端真跑:选1条文案→建package→build_delivery_package→`grep note_0[1-3]`=3套各6图,每套独立note_0N夹+文案.txt。独立agent交叉验证7条全过(分组/去重/老数据兼容/越界/异常/契约/红线)。⚠️攒批问倩倩姐:当前只生成选1条文案,三套图共用同文案→交付包"3套"仅图风格(A/B/C)不同、文案全同,品牌词"倍分子"已正确植入。是否要三套各配不同角度文案? +- [x] **A8**〔human+auto〕多套打包+三套不同角度文案：✅2026-06-18 task74(6图×3套A/B/C)端到端真跑:选1条文案→建package→build_delivery_package→`grep note_0[1-3]`=3套各6图,每套独立note_0N夹+文案.txt。✅2026-06-18 补全"三套各配不同角度文案"(倩倩姐拍板"方向对按这个写"):①constants加TEXT_NARRATIVE_BY_STRATEGY(A痛点/B场景/C成分,与图侧同轴)②build_prompt加strategy_narrative参数并注入prompt③text_variants全链路透传(含优化轮)④run_text_generation改循环A/B/C三套,text_count三套均摊(divmod余前补),跨套previous_copies去重,每条打_strategy⑤TextCandidate加strategy String(4)+迁移021(已upgrade head,SHOW COLUMNS确认)⑥打包按strategy精准配对(texts_by_strategy映射+三层兜底)⑦SSE带strategy。worker已restart重载。独立agent二次交叉验证7改造点全✅+边界(text_count<3跳n=0/无别名/count每套不截断)全过,结论"可进一条龙无must-fix"。品牌词"倍分子"植入文案+图片文字层(不P瓶身)。 - [x] **A9**〔auto〕产品 JSON 导出：✅2026-06-18 ws3实测:export_products返3条,selling_points类型=list(['烟酰胺改善暗沉提亮','水解珍珠锁水匀肤']非JSON字符串);端点真注册路由表+顺序在/products/{product_id}之前(idx9<13不被吞)。 - [ ] **A10**〔auto〕标杆 CSV 导出：⏸️逻辑已审(CSV带UTF-8 BOM+JOIN Product填product_name+json/csv双格式),但ws3无标杆数据未触发真导出;待录标杆后验或Z阶段一条龙带验。 - [ ] **A11**〔auto〕导出权限：⚠️审计注——require_write_permission 当前只查 workspace_members 不分角色(workspace_guard.py:65-88)；未登录调 → code=40101，workspace 成员 → 200。【两端点均挂require_write_permission,独立审核确认】