194 lines
8.0 KiB
Python
194 lines
8.0 KiB
Python
"""节点 0/6：动态种子词（seed）。
|
||
|
||
在 fetch 之前运行：收集「trending 派生 + 历史 safe 热点 + 月份/节日」上下文，
|
||
按 seed_provider 策略（static / mock / LLM）生成/合并种子词，注入 country_config 的
|
||
style.seeds / related.seed_keywords，供后续 fetch 的 related_queries 展开使用。
|
||
|
||
关键机制：动态种子词按 (国家, provider, 日期) 缓存（.cache/seeds/）。
|
||
同一天内多次运行使用同一套种子词 → related_queries 的 24h 缓存稳定命中，
|
||
避免「history 每次跑完都变 → 种子词震荡 → Google 反复全量重抓 → 429 限流」。
|
||
|
||
带 with_fallback：任何异常都降级为"仅用 yaml 静态种子"，不阻塞整图。
|
||
"""
|
||
import datetime
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
from typing import Any, Dict
|
||
|
||
from graph.llms import get_backend
|
||
from graph.llms.mock_backend import COMMON_RISK_WORDS
|
||
from graph.paths import runtime_root
|
||
from graph.scoring import filter_person_names, filter_query_noise
|
||
from graph.seeds import get_seed_strategy
|
||
from graph.seeds.holidays import build_holiday_context
|
||
from graph.sources.google_trends_source import fetch_trending
|
||
from graph.validate import with_fallback
|
||
|
||
_CACHE_DIR = runtime_root() / ".cache" / "seeds"
|
||
|
||
|
||
def _cache_key(country: str, provider: str, cfg: Dict[str, Any]) -> str:
|
||
"""缓存键含「配置指纹」：改了种子相关参数（数量/上下文上限）即换新键重新生成，
|
||
避免命中旧参数生成的种子；旧文件保留（不删缓存，取最新）。"""
|
||
fp = hashlib.md5(
|
||
json.dumps(
|
||
{k: cfg.get(k) for k in ("max_seeds", "max_style_seeds", "max_related_seeds",
|
||
"trending_context_limit", "history_limit")},
|
||
sort_keys=True, ensure_ascii=False,
|
||
).encode("utf-8")
|
||
).hexdigest()[:8]
|
||
return f"{country}-{provider}-{fp}-{datetime.date.today().isoformat()}"
|
||
|
||
|
||
def _cache_get(key: str):
|
||
try:
|
||
p = _CACHE_DIR / f"{key}.json"
|
||
if p.exists():
|
||
return json.loads(p.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _cache_set(key: str, val: Dict[str, Any]):
|
||
try:
|
||
_CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||
(_CACHE_DIR / f"{key}.json").write_text(
|
||
json.dumps(val, ensure_ascii=False), encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
@with_fallback("seed")
|
||
def seed_node(state: Dict[str, Any]) -> Dict[str, Any]:
|
||
country = state["country"]
|
||
config = state["config"]
|
||
cc = dict(state.get("country_config") or {})
|
||
errors = list(state.get("errors") or [])
|
||
|
||
provider = (config.get("seed_provider") or "mock").strip().lower()
|
||
cfg = config.get("seed_provider_cfg") or {}
|
||
trending_limit = int(cfg.get("trending_context_limit", 15))
|
||
history_limit = int(cfg.get("history_limit", 20))
|
||
max_seeds = int(cfg.get("max_seeds", 0))
|
||
max_style = int(cfg.get("max_style_seeds", 12))
|
||
max_related = int(cfg.get("max_related_seeds", 12))
|
||
guard = COMMON_RISK_WORDS + [b.lower() for b in (config.get("blacklist") or [])]
|
||
|
||
ckey = _cache_key(country, provider, cfg)
|
||
cached = _cache_get(ckey) if provider != "static" else None
|
||
from_cache = cached is not None
|
||
|
||
if cached is not None:
|
||
res = cached
|
||
context: Dict[str, Any] = {
|
||
"country": country,
|
||
"max_seeds": max_seeds,
|
||
"max_style_seeds": max_style,
|
||
"max_related_seeds": max_related,
|
||
}
|
||
else:
|
||
# 1) 收集上下文
|
||
context: Dict[str, Any] = {
|
||
"country": country,
|
||
"max_seeds": max_seeds,
|
||
"max_style_seeds": max_style,
|
||
"max_related_seeds": max_related,
|
||
}
|
||
try:
|
||
tl = int((cc.get("trending") or {}).get("limit", 40))
|
||
rows = fetch_trending(geo=country, limit=min(tl, 40))
|
||
# 保留 rows 自带的 source=gt_trending，filter_person_names 的人名模式仅对该源生效
|
||
kept, _ = filter_query_noise(rows, enabled=True)
|
||
kept, _ = filter_person_names(kept)
|
||
kept = [r for r in kept if not any(w and w in r["topic"].lower() for w in guard)]
|
||
context["trending_seeds"] = [r["topic"] for r in kept][:trending_limit]
|
||
except Exception as e: # noqa: BLE001
|
||
print(f"[seed] trending 上下文收集失败(跳过): {e}")
|
||
context["trending_seeds"] = []
|
||
|
||
try:
|
||
p = os.path.join(state.get("output_dir", ""), "design_briefs.json")
|
||
if os.path.exists(p):
|
||
data = json.load(open(p, encoding="utf-8")).get("design_briefs", [])
|
||
safe = [d for d in data if d.get("risk_level") == "safe"]
|
||
safe.sort(key=lambda d: -(d.get("score") or 0))
|
||
hrows = [{"topic": d["topic"], "source": "history"} for d in safe]
|
||
hrows, _ = filter_person_names(hrows, pattern_sources={"history"})
|
||
hrows = [r for r in hrows if not any(w and w in r["topic"].lower() for w in guard)]
|
||
context["history_hotspots"] = [r["topic"] for r in hrows][:history_limit]
|
||
else:
|
||
context["history_hotspots"] = []
|
||
except Exception as e: # noqa: BLE001
|
||
print(f"[seed] 历史热点读取失败(跳过): {e}")
|
||
context["history_hotspots"] = []
|
||
|
||
# 月份/节日（按国家：各国节日表不同）
|
||
hol = build_holiday_context(country)
|
||
context["season"] = hol["season"]
|
||
context["year"] = hol["year"]
|
||
context["month"] = hol["month"]
|
||
context["date"] = hol["date"]
|
||
context["month_themes"] = hol["month_themes"]
|
||
context["upcoming_holidays"] = hol["upcoming_holidays"]
|
||
|
||
# 2) 选策略 + LLM 后端
|
||
strategy = get_seed_strategy(provider)
|
||
llm_backend = None
|
||
if provider != "static":
|
||
llm_backend = get_backend(provider)
|
||
# 注入 llm_screen 配置（api_key/base_url/model），否则 has_key 永远 False 降级 mock
|
||
try:
|
||
llm_backend.bind_config(config.get("llm_screen") or {})
|
||
except Exception as e: # noqa: BLE001
|
||
print(f"[seed] LLM 配置绑定失败: {e}")
|
||
if provider not in ("mock",) and not getattr(llm_backend, "has_key", False):
|
||
print(f"[seed] {provider} 未配置 API key，降级 mock 规则生成种子词")
|
||
llm_backend = get_backend("mock")
|
||
|
||
# 3) 生成/合并种子词
|
||
try:
|
||
res = strategy.resolve(country, cc, context, llm_backend)
|
||
except Exception as e: # noqa: BLE001
|
||
print(f"[seed] 策略解析失败，回退静态种子: {e}")
|
||
res = {
|
||
"style_seeds": list((cc.get("style", {}) or {}).get("seeds", []) or []),
|
||
"related_seeds": list((cc.get("related", {}) or {}).get("seed_keywords", []) or []),
|
||
"dynamic": False,
|
||
}
|
||
if provider != "static":
|
||
_cache_set(ckey, res)
|
||
|
||
# 4) 注入 cc
|
||
style_block = dict(cc.get("style") or {})
|
||
related_block = dict(cc.get("related") or {})
|
||
style_block["seeds"] = res["style_seeds"]
|
||
related_block["seed_keywords"] = res["related_seeds"]
|
||
cc["style"] = style_block
|
||
cc["related"] = related_block
|
||
|
||
stats = dict(state.get("stats") or {})
|
||
stats["seed"] = {
|
||
"provider": provider,
|
||
"dynamic": res.get("dynamic", False),
|
||
"from_cache": from_cache,
|
||
"style_count": len(res["style_seeds"]),
|
||
"related_count": len(res["related_seeds"]),
|
||
"trending_ctx": len(context.get("trending_seeds", [])),
|
||
"history_ctx": len(context.get("history_hotspots", [])),
|
||
"holidays": context.get("upcoming_holidays", []),
|
||
}
|
||
print(
|
||
f"[seed] provider={provider}{'（当日缓存命中）' if from_cache else ''} "
|
||
f"种子词 style={len(res['style_seeds'])} related={len(res['related_seeds'])}"
|
||
)
|
||
|
||
return {
|
||
"country_config": cc,
|
||
"seed_words": res,
|
||
"stats": stats,
|
||
"errors": errors,
|
||
}
|