df1856d793
- 产品编辑入口统一走 ProductFormFull(卖点/风格/人群/品牌词全字段); 修复开任务页 <form> 套 <form> 致"编辑产品"报错、改不了、跳回首个产品 - dashboard 入口卡片对齐实际路由: 系统管理(/config) 与 工作配置(/settings) 分开; settings ?tab=products 直达改用挂载后读 URL, 消除 hydration mismatch - 新增用户管理(users API/admin service/改密页) + alembic 022/023/024 - 上线部署: Dockerfile / docker-compose.prod+https / nginx https / .env.example - A8 三套正交叙事(痛点/场景/成分背书) + beige 调色去AI化 + 飞轮 text_import 高权重信号 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
257 lines
14 KiB
Python
257 lines
14 KiB
Python
"""
|
||
text_variants.py — 文案双轨主入口（≤100行）
|
||
轨A: generate_text_variants — 调 LLM 出 N 角度 JSON
|
||
轨B: text_import_handler — 导入外部文案进候选池
|
||
|
||
prompt 组装/解析见 _text_prompt.py；评分/去重见 text_scoring.py
|
||
"""
|
||
from __future__ import annotations
|
||
import asyncio
|
||
import logging
|
||
import os
|
||
from typing import Any
|
||
|
||
from .constants import MAX_OPTIMIZE_ROUNDS
|
||
from ._text_prompt import COPY_SYSTEM, build_prompt, parse_json_array, build_local_drafts
|
||
from .text_scoring import score_copy, dedupe_copies
|
||
from .llm_scorer import llm_score_copy, ScoringUnavailableError
|
||
from .banned_word_checker import check_and_fix, build_entries_from_db, CheckResult
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
async def _call_llm(client: Any, prompt: str, max_tokens: int = 8192) -> str:
|
||
"""统一 LLM 调用，client 由 worker 注入，隔离 key。
|
||
G1坑修复：AIClients 没有 .chat.completions，正确方法是 .chat_complete()
|
||
S8: 503/429 指数退避重试（最多3次，2^attempt 秒），其他异常直接降级返 ''。
|
||
max_tokens 由调用方按批量缩放：opus 会尽量填满输出空间，8192 token 的生成
|
||
单批 >60s 必撞 apiports 网关上限返 503(task46 实测每请求恰挂 ~61s)。实测单条
|
||
max_tokens=1500~2500 仅 16~18s。故按条数动态收，墙钟压进 60s 网关窗口内。
|
||
"""
|
||
import httpx
|
||
|
||
# 倩倩姐2026-06-13拍板"加大重试+拉长退避"：apiports负载波动时单条opus也会被
|
||
# 拖过60s返503，短退避(1/2/4s)赶不开高负载窗口。故重试5次、退避拉长到最长30s，
|
||
# 给中转站负载回落留时间。墙钟换稳定(MVP免费阶段可接受)。
|
||
max_attempts = 5
|
||
backoff = [5, 10, 20, 30] # 第1~4次重试前等待秒数，拉长跨过apiports高负载窗口
|
||
for attempt in range(max_attempts):
|
||
try:
|
||
return await client.chat_complete(
|
||
messages=[
|
||
{"role": "system", "content": COPY_SYSTEM},
|
||
{"role": "user", "content": prompt},
|
||
],
|
||
model=client._model,
|
||
max_tokens=max_tokens,
|
||
temperature=0.75,
|
||
)
|
||
except httpx.HTTPStatusError as exc:
|
||
status = exc.response.status_code if exc.response is not None else 0
|
||
if status in (503, 429) and attempt < max_attempts - 1:
|
||
wait = backoff[min(attempt, len(backoff) - 1)]
|
||
logger.warning(
|
||
"LLM 返回 %s，第%d/%d次重试，等待 %ds: %s",
|
||
status, attempt + 1, max_attempts - 1, wait, exc,
|
||
)
|
||
await asyncio.sleep(wait)
|
||
continue
|
||
logger.error("LLM HTTP错误(不可重试或已达上限): %s: %s", type(exc).__name__, exc)
|
||
return ""
|
||
except Exception as exc:
|
||
# 其他异常（超时/网络断开等）不重试，直接降级
|
||
logger.error("LLM 调用失败: %s: %s", type(exc).__name__, exc)
|
||
return ""
|
||
return ""
|
||
|
||
|
||
# apiports 网关单次响应有 ~60s 上限，claude 一次生成 >4 条长文案会超时返 503。
|
||
# 故分批：每批最多 4 条，串行调用合并。批大小可经 TEXT_BATCH_SIZE 调。
|
||
TEXT_BATCH_SIZE = int(os.environ.get("TEXT_BATCH_SIZE", "4"))
|
||
|
||
# 评分有限并发：评分无状态、逐条独立可并发，但仍限并发数防 apiports 限流(生成层串行
|
||
# 不动，task45 实测全并发雪崩)。默认3=保守值(评分 max_tokens仅1500，远轻于生成8192，
|
||
# 且评分已带 503/429 重试兜底)，可经 SCORE_CONCURRENCY 调。
|
||
_SCORE_CONCURRENCY = int(os.environ.get("SCORE_CONCURRENCY", "3"))
|
||
|
||
|
||
def _build_cross_set_avoidance(previous: list[dict]) -> str:
|
||
"""从已生成的其他套文案提炼回避约束，喂进生成 prompt 防三套撞车。
|
||
覆盖三个最易同质化的点：已用标签、背书人物桥段、收尾句式。"""
|
||
if not previous:
|
||
return ""
|
||
used_tags: list[str] = []
|
||
used_titles: list[str] = []
|
||
for c in previous:
|
||
used_titles.append(c.get("title", ""))
|
||
for t in (c.get("tags") or []):
|
||
tag = str(t).replace("🛒", "").strip()
|
||
if tag and tag not in used_tags:
|
||
used_tags.append(tag)
|
||
lines = ["【跨套去重·硬约束（本条务必与已发布的其它套拉开，禁止换皮重复）】"]
|
||
if used_titles:
|
||
lines.append(f"- 已用标题（禁止同主题/同句式）：{ '；'.join(t for t in used_titles if t) }")
|
||
if used_tags:
|
||
lines.append(f"- 已用标签（本条至少换掉一半，改用不同人群/场景/成分组合的精准长尾，扩大触达不抢同批流量）：{ '、'.join(used_tags) }")
|
||
lines.append("- 标签禁止与已用标签共用同一核心搜索词（别套已用的核心词，本条标签就别再都堆这个词，换成本套独有的人群/场景/卖点关键词分流）。")
|
||
lines.append("- 背书角色/桥段禁止与上面套雷同：若别套用了'闺蜜种草'，本条换成自证/家人/同事/自己反复试错等不同来源；")
|
||
lines.append("- 卖点罗列的措辞顺序与 emoji 禁止与别套一致；同一个具体种草桥段全局只许出现一次。")
|
||
return "\n".join(lines)
|
||
|
||
|
||
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, strategy_narrative=strategy_narrative,
|
||
), batch_max_tokens)
|
||
parsed = parse_json_array(raw)
|
||
if parsed:
|
||
return parsed
|
||
logger.warning("文案批(%d条)第%d次解析失败%s", batch_n, attempt + 1,
|
||
"，重试" if attempt == 0 else "，放弃本批")
|
||
return []
|
||
|
||
|
||
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 实测)。故改串行，墙钟换稳定。"""
|
||
sizes: list[int] = []
|
||
remaining = count
|
||
while remaining > 0:
|
||
n = min(TEXT_BATCH_SIZE, remaining)
|
||
sizes.append(n)
|
||
remaining -= n
|
||
collected: list[dict] = []
|
||
for n in sizes:
|
||
r = await _generate_one_batch(llm_client, product, n, extra, strategy_narrative)
|
||
collected.extend(r)
|
||
return collected
|
||
|
||
|
||
async def generate_text_variants(
|
||
llm_client: Any,
|
||
product: dict,
|
||
count: int,
|
||
previous_copies: list[dict] | None = None,
|
||
banned_word_rows: list[dict] | None = None,
|
||
flywheel_context: str = "",
|
||
strategy_narrative: str = "",
|
||
) -> list[dict]:
|
||
"""轨A：一次出 count 条不同角度文案，三层兜底，自动优化循环。
|
||
strategy_narrative：本套正交叙事主线(A痛点/B场景/C成分)，由调用方按套传入，
|
||
贯穿首批生成与优化轮，确保同套内文案同一叙事不串味。"""
|
||
banned_entries = build_entries_from_db(banned_word_rows or [])
|
||
extra = flywheel_context
|
||
# 跨套去重：把已生成套的标签/背书桥段提炼成回避约束喂进生成 prompt，
|
||
# 防止三套撞车（仅事后 dedupe 删不掉"同用闺蜜背书/同套标签"的同质化）
|
||
avoid = _build_cross_set_avoidance(previous_copies or [])
|
||
if avoid:
|
||
extra = f"{extra}\n{avoid}" if extra else avoid
|
||
|
||
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
|
||
|
||
# 评分有限并发：逐条独立无状态可并发，sem 限并发数防 apiports 限流(生成层串行不动)。
|
||
# 每条语义与原串行完全一致：违禁词机械检查(同步,放 sem 外不占并发额度)→AI评分→
|
||
# 评分不可用则标记不计合格不炸整批；auto_fixed 回写修正文案。gather 保序=候选顺序不变。
|
||
sem = asyncio.Semaphore(_SCORE_CONCURRENCY)
|
||
|
||
async def _score_one(c: dict) -> dict:
|
||
ban: CheckResult = check_and_fix(
|
||
f"{c.get('title','')} {c.get('content','')}",
|
||
banned_entries or None,
|
||
)
|
||
try:
|
||
async with sem:
|
||
scored = await llm_score_copy(llm_client, c, product, [e.word for e in banned_entries])
|
||
except ScoringUnavailableError as exc:
|
||
# B1：评分两通道都挂→不打机械分糊弄，这条标"评分不可用·不计合格"，不炸整批
|
||
logger.warning("文案评分不可用，标记不计合格(非质量问题): %s", exc)
|
||
c.update({"source": "ai", "score": 0,
|
||
"score_detail": [{"item": "评分不可用", "score": 0, "max": 100,
|
||
"note": "评分通道(apiports+codeproxy)暂不可用，非文案质量问题"}],
|
||
"passed": False, "banned_word_status": ban.status,
|
||
"scoring_unavailable": True, "verdict": "", "summary": "评分服务暂不可用"})
|
||
return c
|
||
c.update({"source": "ai", "score": scored["score"], "score_detail": scored["score_detail"],
|
||
"passed": scored["passed"], "banned_word_status": ban.status,
|
||
"verdict": scored.get("verdict", ""), "summary": scored.get("summary", "")})
|
||
if ban.status == "auto_fixed" and ban.fixed_text:
|
||
c["content"] = ban.fixed_text
|
||
return c
|
||
|
||
candidates: list[dict] = list(await asyncio.gather(*(_score_one(c) for c in copies)))
|
||
scoring_failures = sum(1 for c in candidates if c.get("scoring_unavailable"))
|
||
|
||
# 整批评分全因通道不可用而失败 → 抛错让上层透传"评分服务不可用"，区别于"质量不合格"
|
||
if copies and scoring_failures == len(copies):
|
||
raise ScoringUnavailableError(
|
||
f"本套 {len(copies)} 条文案评分全部失败：评分通道(apiports+codeproxy)均不可用")
|
||
|
||
failed = [c for c in candidates if not c["passed"] and c["banned_word_status"] != "hard_block"]
|
||
# 优化轮默认关闭：apiports 60s 网关限制下优化轮的 _call_llm 常需白等 60s 才 503，
|
||
# 严重拖慢出文案（实测 +100s+）。质量优化等北哥 prompt 方案到位再开（架构已留位）。
|
||
optimize_enabled = os.environ.get("TEXT_OPTIMIZE_ENABLED", "false").lower() == "true"
|
||
rounds = MAX_OPTIMIZE_ROUNDS if optimize_enabled else 0
|
||
for _ in range(rounds):
|
||
if not failed:
|
||
break
|
||
# 优化轮也受 60s 网关上限约束：一次最多重生成 TEXT_BATCH_SIZE 条
|
||
batch_failed = failed[:TEXT_BATCH_SIZE]
|
||
hint = "\n".join(
|
||
f"标题「{c['title']}」{c['score']}分，需改进：" +
|
||
"；".join(d["note"] for d in c.get("score_detail", []) if d["score"] < d["max"] * 0.72)
|
||
for c in batch_failed
|
||
)
|
||
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/超时）：优化是锦上添花，原始候选已够用，不再耗时重试
|
||
logger.warning("文案优化轮 LLM 失败，沿用原始候选不再重试")
|
||
break
|
||
for nc in parse_json_array(raw2):
|
||
try:
|
||
sc2 = await llm_score_copy(llm_client, nc, product, [e.word for e in banned_entries])
|
||
except ScoringUnavailableError as exc:
|
||
logger.warning("优化轮评分不可用，跳过该条不计入: %s", exc)
|
||
continue
|
||
nc.update({"source": "ai", "score": sc2["score"], "score_detail": sc2["score_detail"],
|
||
"passed": sc2["passed"], "banned_word_status": "pass",
|
||
"verdict": sc2.get("verdict", ""), "summary": sc2.get("summary", "")})
|
||
candidates.append(nc)
|
||
failed = [c for c in candidates if not c["passed"]]
|
||
|
||
return dedupe_copies(candidates, previous_copies or [])[:count]
|
||
|
||
|
||
def text_import_handler(
|
||
raw_text: str,
|
||
product: dict,
|
||
banned_word_rows: list[dict] | None = None,
|
||
) -> dict:
|
||
"""轨B：导入外部文案（豆包等）直接进候选池，source=import"""
|
||
banned_entries = build_entries_from_db(banned_word_rows or [])
|
||
lines = raw_text.strip().splitlines()
|
||
title = lines[0].strip() if lines else ""
|
||
content = "\n".join(lines[1:]).strip() if len(lines) > 1 else raw_text.strip()
|
||
candidate: dict = {"title": title, "content": content, "tags": [], "angle": "import",
|
||
"buyingPoint": "", "coverTitle": title, "imageBrief": "", "source": "import"}
|
||
ban = check_and_fix(f"{title} {content}", banned_entries or None)
|
||
# 轨B(导入外部文案)走机械 score_copy 而非 AI 评委：导入的是用户自带成品，评分仅作
|
||
# 参考展示不卡发布；且本函数同步、改 await 会扩大到调用方。AI 评委只用于轨A生成链路。
|
||
scored = score_copy(candidate, product, [e.word for e in banned_entries])
|
||
candidate.update({"score": scored["score"], "score_detail": scored["score_detail"],
|
||
"passed": scored["passed"], "banned_word_status": ban.status})
|
||
return candidate
|