Files
pod_trend_agent/graph/llms/mock_backend.py
T

191 lines
8.9 KiB
Python

"""Mock LLM 后端:启发式兜底(无 key 也能端到端跑通)。
逻辑:黑名单硬拦 -> 常识风险词标 review -> 动态风格/配色推导 -> 分类。
这是生产环境 LLM 不可用时的安全降级路径,保证流水线永远能产出可用结果。
"""
from typing import Any, Dict, List
from ..classify import classify, prompt_suggestion
from ..style_rules import derive_style_palette, derive_composition
def _dedup_limit(items: List[str], limit: int) -> List[str]:
"""去重(大小写不敏感)并限量,保留首次出现顺序。"""
seen = set()
out: List[str] = []
for it in items:
it = (it or "").strip()
if not it:
continue
low = it.lower()
if low in seen:
continue
seen.add(low)
out.append(it)
if len(out) >= limit:
break
return out
# 常识风险词(兜底用;真实判定交给 LLM)。同时被 seed_node 复用为「种子护栏」,
# 避免真人/IP/平台词作为 Google Trends 相关查询种子浪费抓取。
COMMON_RISK_WORDS = [
"disney", "marvel", "nike", "adidas", "apple", "iphone", "mcdonalds",
"mcdonald", "starbucks", "coca", "pepsi", "pokemon", "mario", "hello kitty",
"sanrio", "sonic", "minions", "barbie", "harry potter", "batman", "spiderman",
"star wars", "fortnite", "roblox", "minecraft", "tiktok", "netflix", "pearl jam",
"nirvana", "taylor swift", "trump", "biden", "kardashian", "lebron", "kanye",
"kick", "gta", "ufc", "westmeath", "lottery", "prison break", "margot robbie",
"dana white", "euro", "spotify", "youtube", "instagram", "xbox", "playstation",
"noah kahan", "gina carano", "camry", "hurricanes", "eras tour",
"springsteen", "reiner", "eliza lopes", "camilla", "h&m", "truck accident attorney",
]
# 原创短标语池(mock 模式的 slogan;纯原创、无版权无品牌)
# 原创短标语池(mock 模式的 slogan;纯原创、无版权无品牌)
# 英语通用 + 各国语言(JP=日语短标语),按国家动态注入
_STABLE_SLOGANS_EN = [
"good vibes", "stay cozy", "happy place", "be kind", "dream big",
"keep smiling", "sunshine", "peace love", "stay wild", "pet the cat",
"coffee first", "tiny paws", "warm hugs", "soft life", "grow slowly",
"lucky charm", "sweet dreams", "go outside", "mindful", "lazy days",
]
_STABLE_SLOGANS_JP = [
"ゆめいっぱい", "やさしい気持ち", "ずっと元気", "おだやかな日", "きょうもハッピー",
"ねこが好き", "いっしょにね", "ぽかぽか", "はるの風", "なつのおもいで",
"きらきら", "わくわく", "のんびり", "しあわせ", "えがお",
]
def _stable_slogan(topic: str, country: str = "") -> str:
"""按主题哈希稳定选一条原创标语(同一主题缓存一致;mock 兜底用)。
country=JP → 日语短标语;其他国家 → 英语。"""
import hashlib
pool = _STABLE_SLOGANS_JP if str(country).upper() == "JP" else _STABLE_SLOGANS_EN
h = int(hashlib.md5((topic or "").encode("utf-8")).hexdigest(), 16)
return pool[h % len(pool)]
class MockBackend:
name = "mock"
def screen(
self,
topics: List[str],
country: str,
aesthetic_hint: str,
system_prompt: str,
blacklist: List[str],
batch_size: int = 12,
) -> List[Dict[str, Any]]:
bl = [b.lower() for b in (blacklist or [])]
out: List[Dict[str, Any]] = []
for t in topics:
tl = t.lower()
hits = [b for b in bl if b and b in tl]
blocked = bool(hits)
soft_hits = [w for w in COMMON_RISK_WORDS if w in tl]
if blocked:
risk_level = "blocked"
elif soft_hits:
risk_level = "review"
else:
risk_level = "safe"
cat = classify(t)
art_style, palette = derive_style_palette(t, country, category=cat)
# motif:从分类模板取核心描述,去掉配色/白底尾巴,保持干净可复用
motif = prompt_suggestion(t, cat).split(" --no ")[0].split(",")[0].strip()
composition = derive_composition(t, cat)
negative = ("no real people, no likeness of any person, no copyrighted characters, "
"no brand logos, no trademarks, no celebrity, no readable text unless safe")
slogan = _stable_slogan(t, country) # 按国家语言(JP→日语短标语,其余英语)
out.append({
"topic": t,
"safe_for_print": not blocked,
"risk_level": risk_level,
"risk_reasons": [f"命中黑名单: {hits}"] if hits
else (["疑似受保护实体,需人工复核"] if soft_hits else []),
"suitable_for_print": not blocked,
"design_category": cat,
"concept": f"(启发式兜底)围绕「{t}」做原创{art_style}风格印花",
"motif": motif,
"art_style": art_style,
"color_palette": palette,
"composition": composition,
"slogan": slogan,
"negative_prompt": negative,
"confidence": 0.55 if risk_level == "safe" else 0.4,
})
return out
def generate_seeds(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""规则生成种子词(零 API 成本):借用月份主题、临近节日、trending 派生、历史热点。"""
month_themes = context.get("month_themes", []) or []
upcoming = context.get("upcoming_holidays", []) or []
trending = context.get("trending_seeds", []) or []
history = context.get("history_hotspots", []) or []
style: List[str] = []
related: List[str] = []
# 月份主题 + 临近节日 → 风格种子(带美学倾向)
style += list(month_themes)
style += [f"{h.lower()} aesthetic" for h in upcoming]
style += trending[:4]
# related:历史 safe 热点 + 剩余 trending(行业交叉验证)
related += history[:6]
related += trending[4:8]
max_seeds = int(context.get("max_seeds") or 0)
max_style = int(context.get("max_style_seeds", 10) or 10)
max_related = int(context.get("max_related_seeds", 10) or 10)
if max_seeds > 0:
# 不再按类型分:总量均分到 style/related
max_style = max_related = (max_seeds + 1) // 2
return {
"style_seeds": _dedup_limit(style, max_style),
"related_seeds": _dedup_limit(related, max_related),
}
def generate_pinterest_terms(self, context: Dict[str, Any]) -> Dict[str, Any]:
"""规则生成 Pinterest 搜索词(零 API 成本):从种子词池随机取 + 两两组合增加多样性。"""
import random
seeds = [str(s).strip() for s in (context.get("seeds") or []) if str(s).strip()]
used = {str(u).strip().lower() for u in (context.get("used_terms") or [])}
count = int(context.get("count", 10))
pool = [s for s in seeds if s.lower() not in used]
random.shuffle(pool)
terms = pool[:count]
# 不足时用「种子词 + 风格词」组合补足(视觉导向,避免与已用重复)
style_tail = ["aesthetic", "style", "inspiration", "design", "vibe", "art"]
i = 0
while len(terms) < count and pool:
combo = f"{pool[i % len(pool)]} {style_tail[(i // len(pool)) % len(style_tail)]}"
if combo.lower() not in used and combo not in terms:
terms.append(combo)
i += 1
return {"search_terms": terms}
def analyze_pinterest_images(self, image_paths, term="", country=""):
"""规则生成设计简报(零 API 成本):按搜索词启发式推导风格/配色/构图。"""
from ..classify import classify, prompt_suggestion
cat = classify(term)
art_style, palette = derive_style_palette(term, country, category=cat)
motif = prompt_suggestion(term, cat).split(" --no ")[0].split(",")[0].strip()
composition = derive_composition(term, cat)
negative = ("no real people, no likeness of any person, no copyrighted characters, "
"no brand logos, no trademarks, no celebrity, no readable text unless safe")
n = max(1, len(image_paths or []))
paths = list(image_paths or [])
return [{
"topic": term,
"concept": f"(启发式兜底)围绕「{term}」做原创{art_style}风格印花",
"motif": motif,
"art_style": art_style,
"color_palette": palette,
"composition": composition,
"negative_prompt": negative,
# 生图参考:每条简报对应其来源爬取图(mock 按图逐张产出简报,顺序一一对应)
"ref_images": [str(paths[i])] if i < len(paths) else [],
"source": "pinterest",
} for i in range(n)]