{p.get("name", "?")}
{meta}
@@ -651,26 +660,41 @@ def main():
raw = json.load(f)
stats = loader.load_basic_stats()
reviews = loader.load_reviews()
+ extractions = loader.load_comment_extractions()
+ persona_quote_max = int(cfg.get("persona_quote_max", 3))
+ rootcause_quote_max = int(cfg.get("rootcause_quote_max", 4))
all_clusters = loader.load_cluster_data()
personas = assign_persona_clusters(raw.get("personas", []), all_clusters)
cluster_data_rr = raw.get("cluster_data") or loader.build_cluster_prompt_data(
persona_sample_reviews=int(cfg.get("persona_sample_reviews", 5)),
)
- enrich_persona_catalog_physio_counts(
+ physio_min = int(cfg.get("persona_physio_min_reviews", 5))
+ prepare_persona_catalog_for_llm(
cluster_data_rr.get("persona_cluster_catalog") or [],
all_clusters, reviews,
+ min_physio_reviews=physio_min,
)
- physio_min = int(cfg.get("persona_physio_min_reviews", 5))
personas = validate_persona_physiological_labels(
personas, cluster_data_rr, reviews, all_clusters, min_review_count=physio_min,
)
personas = compute_persona_pcts(personas, reviews, all_clusters)
personas = sort_personas_by_evidence(personas)
- persona_quotes = pick_persona_quotes(personas, reviews)
- neg_freq = calc_theme_freq(raw.get("neg_themes", []), reviews, is_neg=True)
- pos_freq = calc_theme_freq(raw.get("pos_themes", []), reviews, is_neg=False)
- per_asin_neg = calc_per_asin_theme_freq(raw.get("neg_themes", []), reviews, is_neg=True)
- per_asin_pos = calc_per_asin_theme_freq(raw.get("pos_themes", []), reviews, is_neg=False)
+ persona_quotes = pick_persona_quotes(
+ personas, reviews, extractions,
+ max_quotes=persona_quote_max,
+ )
+ neg_freq = calc_theme_freq(
+ raw.get("neg_themes", []), reviews, is_neg=True, extractions=extractions,
+ )
+ pos_freq = calc_theme_freq(
+ raw.get("pos_themes", []), reviews, is_neg=False, extractions=extractions,
+ )
+ per_asin_neg = calc_per_asin_theme_freq(
+ raw.get("neg_themes", []), reviews, is_neg=True, extractions=extractions,
+ )
+ per_asin_pos = calc_per_asin_theme_freq(
+ raw.get("pos_themes", []), reviews, is_neg=False, extractions=extractions,
+ )
neg_themes = recalc_neg_priorities(
raw.get("neg_themes", []), neg_freq, stats.neg_review_count, len(stats.asins), per_asin_neg,
)
@@ -710,14 +734,14 @@ def main():
market_title, market_desc = enhanced_market_judgment(stats)
data["market_title"] = market_title
data["market_desc"] = market_desc
- per_aud = loader.load_per_audience_clusters()
data["rootcauses"] = enrich_rootcause_quotes(
data.get("rootcauses") or [],
personas,
- per_aud,
loader,
reviews,
persona_quotes=persona_quotes,
+ extractions=extractions,
+ max_quotes=rootcause_quote_max,
)
elif args.no_llm:
logger.warning("--no-llm 模式：跳过 LLM 调用，仅渲染模板")
diff --git a/voc_业务_2/config.yaml b/voc_业务_2/config.yaml
index 485e670..44bbd5c 100644
--- a/voc_业务_2/config.yaml
+++ b/voc_业务_2/config.yaml
@@ -30,6 +30,15 @@ prompts_file: "./prompts.yaml"
persona_sample_reviews: 5
# Persona 生理标签：绑定簇内至少 N 条评论含对应英文词方可保留（如 pregnant≥5 才可用「孕妇」）
persona_physio_min_reviews: 5
+# Persona 命中门槛：hit_count 低于此值跳过根因 LLM，卡片置信度显示「低样本」
+persona_min_hit_count: 5
+# Persona 卡片展示的代表性评论条数（结构化字段匹配 + 原文回退）
+persona_quote_max: 3
+# 每条根因展示的代表性评论条数
+rootcause_quote_max: 4
+# 进入 step 2 细粒度聚类的 audience 簇最小占比（相对总评论数）
+audience_coverage_threshold: 0.05
+
# ── LLM 配置 ──
# 复用根目录 voc_llm.py 的配置（DEEPSEEK_API_KEY 环境变量 或 .deepseek_key）
diff --git a/voc_业务_2/data_loader.py b/voc_业务_2/data_loader.py
index ecbd731..f1ffd67 100644
--- a/voc_业务_2/data_loader.py
+++ b/voc_业务_2/data_loader.py
@@ -7,11 +7,12 @@ from __future__ import annotations
import csv
import json
import logging
+import math
import sqlite3
from collections import Counter, defaultdict
from dataclasses import dataclass, field
from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Set, Tuple
logger = logging.getLogger("voc.data_loader")
@@ -22,6 +23,52 @@ CLEANED_CSV_NAME = "merged_reviews_cleaned.csv"
WORD_FREQ_CSV_NAME = "output/word_freq.csv"
TERMS_JSON_NAME = "output/voc_terms.json"
+# 与 report_utils._PERSONA_PHRASE_STOPWORDS 保持一致（避免循环 import）
+_PERSONA_PHRASE_STOPWORDS = frozenset({
+ "self", "unknown", "user", "customer", "buyer", "myself", "the user",
+ "a user", "consumer", "reviewer", "amazon", "product", "item",
+})
+
+
+def _phrase_segments_local(phrases: List[str]) -> List[str]:
+ """将聚类短语拆成可匹配的英文片段。"""
+ segs: List[str] = []
+ for p in phrases:
+ p = (p or "").lower().strip()
+ if not p:
+ continue
+ segs.append(p)
+ for part in p.split(","):
+ part = part.strip()
+ if len(part) >= 3:
+ segs.append(part)
+ return list(dict.fromkeys(segs))
+
+
+def _cluster_semantic_label(
+ cluster: "ClusterData",
+ phrase_doc_freq: Dict[str, int],
+ n_clusters_in_stage: int,
+) -> str:
+ """从 top_phrases 生成簇短语义标签（频次 × IDF，同类 stage 内去重）。"""
+ scored: List[Tuple[float, str]] = []
+ for rank, phrase in enumerate(cluster.top_phrases):
+ tf = 1.0 / (1 + rank)
+ for seg in _phrase_segments_local([phrase]):
+ if seg in _PERSONA_PHRASE_STOPWORDS or len(seg) < 4:
+ continue
+ df = phrase_doc_freq.get(seg, 1)
+ idf = math.log((n_clusters_in_stage + 1) / (df + 0.5))
+ scored.append((tf * idf, seg))
+ if scored:
+ scored.sort(key=lambda x: (-x[0], -len(x[1])))
+ return scored[0][1][:48]
+ for phrase in cluster.top_phrases:
+ p = (phrase or "").strip()
+ if len(p) >= 4:
+ return p[:48]
+ return f"cluster_{cluster.cluster_label}"
+
@dataclass
class ReviewRecord:
@@ -69,6 +116,63 @@ class MarketStats:
asins: List[ASINStats] = field(default_factory=list)
+# Persona A/B/C 维度信号词（用于运行时判定簇的 suggested_dimension）
+_A_DIM_SIGNALS = frozenset({
+ # 生理/物理特征 — 跨品类通用
+ "sensitive skin", "sensitive", "allergy", "allergic",
+ "elderly", "senior", "older",
+ "kid", "child", "children", "toddler", "infant", "baby", "newborn",
+ "pregnant", "pregnancy", "nursing", "breastfeeding",
+ "diabetic", "diabetes",
+ "my dog", "my cat", "my pet", "puppy", "kitten", "pet owner",
+ "professional", "beginner", "first time user",
+ "large breed", "small breed", "small dog", "large dog",
+ "oily skin", "dry skin", "acne", "eczema", "psoriasis",
+ "curly hair", "fine hair", "color treated",
+})
+_B_DIM_SIGNALS = frozenset({
+ # 行为/使用场景 — 跨品类通用
+ "travel", "on the go", "portable", "lightweight", "compact",
+ "daily", "everyday", "first time", "first-time", "beginner",
+ "outdoor", "indoor", "home", "office", "kitchen", "car",
+ "quick", "easy to use", "simple", "convenient",
+ "maintenance", "cleaning", "storage", "organization",
+ "camping", "hiking", "gym", "workout",
+ "cooking", "baking", "cleaning house",
+})
+_C_DIM_SIGNALS = frozenset({
+ # 购买动机/背景 — 跨品类通用
+ "gift", "present", "birthday", "christmas", "holiday",
+ "for my wife", "for my husband", "for my mom", "for my daughter",
+ "replacement", "replace", "upgrade", "switched from",
+ "price", "cheap", "expensive", "affordable", "worth the money",
+ "waste of money", "not worth", "overpriced", "good value",
+ "recommend", "recommended", "saw on", "social media",
+ "amazon", "online", "review", "reviews", "rating",
+ "bought", "purchased", "ordered", "arrived",
+})
+
+
+def _suggest_dimension(stage: str, top_phrases: List[str]) -> str:
+ """根据簇的 top_phrases 内容判定建议的 Persona 维度（A/B/C）。"""
+ blob = " ".join(top_phrases).lower()
+ a_hits = sum(1 for w in _A_DIM_SIGNALS if w in blob)
+ b_hits = sum(1 for w in _B_DIM_SIGNALS if w in blob)
+ c_hits = sum(1 for w in _C_DIM_SIGNALS if w in blob)
+ if a_hits >= 2:
+ return "A"
+ if a_hits >= 1 and (b_hits + c_hits) == 0:
+ return "A"
+ if b_hits > c_hits and b_hits >= 2:
+ return "B"
+ if c_hits > b_hits and c_hits >= 2:
+ return "C"
+ # 默认按 stage 推理
+ if stage.startswith("3a"):
+ return "C"
+ return "B"
+
+
class DataLoader:
def __init__(self, project_root: Path, product_name: str = "", industry: str = ""):
self.project_root = Path(project_root).resolve()
@@ -142,6 +246,44 @@ class DataLoader:
return r
return None
+ def load_comment_extractions(
+ self,
+ job_id: Optional[int] = None,
+ ) -> Dict[int, Dict[str, Any]]:
+ """加载 source_row → extraction_json 映射（最新 job 或指定 job_id）。"""
+ if not self.structured_db.is_file():
+ logger.warning("未找到结构化库 %s，引用匹配将仅使用原文关键词", self.structured_db)
+ return {}
+ conn = sqlite3.connect(self.structured_db)
+ try:
+ if job_id is None:
+ row = conn.execute(
+ "SELECT id FROM analysis_jobs ORDER BY id DESC LIMIT 1"
+ ).fetchone()
+ if not row:
+ logger.warning("voc_structured.sqlite 中无结构化任务")
+ return {}
+ job_id = int(row[0])
+ cur = conn.execute(
+ """
+ SELECT source_row, extraction_json
+ FROM comment_extractions
+ WHERE job_id = ?
+ ORDER BY source_row
+ """,
+ (job_id,),
+ )
+ out: Dict[int, Dict[str, Any]] = {}
+ for sr, js in cur.fetchall():
+ try:
+ out[int(sr)] = json.loads(js)
+ except json.JSONDecodeError as e:
+ logger.debug("跳过无效 extraction source_row=%s: %s", sr, e)
+ logger.info("已加载 %s 条结构化提取 (job_id=%s)", len(out), job_id)
+ return out
+ finally:
+ conn.close()
+
def load_basic_stats(self) -> MarketStats:
reviews = self.load_reviews()
if not reviews:
@@ -241,33 +383,46 @@ class DataLoader:
logger.info("已加载聚类: %s stages, %s 簇", len(result), sum(len(v) for v in result.values()))
return dict(result)
+
+
def build_persona_cluster_catalog(
self,
sample_reviews_per_cluster: int = 5,
max_review_chars: int = 320,
) -> List[Dict[str, Any]]:
- """供 Persona LLM 绑定的聚类簇目录（含每簇代表性评论）。"""
+ """供 Persona LLM 绑定的聚类簇目录（含每簇代表性评论和运行时维度建议）。"""
all_c = self.load_cluster_data()
- dim_hint = {
- "1_audience": "A",
- "3a_pain_global": "C",
- "3b_aspect_opinion_negative": "B",
- "3b_aspect_opinion_positive": "B",
- }
catalog: List[Dict[str, Any]] = []
for stage in (
- "1_audience",
"3a_pain_global",
"3b_aspect_opinion_negative",
"3b_aspect_opinion_positive",
):
- for c in sorted(all_c.get(stage, []), key=lambda x: -x.review_count):
+ stage_clusters = sorted(all_c.get(stage, []), key=lambda x: -x.review_count)
+ phrase_doc_freq: Dict[str, int] = defaultdict(int)
+ for c in stage_clusters:
+ segs = set(_phrase_segments_local(c.top_phrases))
+ for seg in segs:
+ if seg not in _PERSONA_PHRASE_STOPWORDS and len(seg) >= 4:
+ phrase_doc_freq[seg] += 1
+ n_stage = max(len(stage_clusters), 1)
+ used_labels: Set[str] = set()
+ for c in stage_clusters:
+ label = _cluster_semantic_label(c, phrase_doc_freq, n_stage)
+ if label in used_labels:
+ for rank, phrase in enumerate(c.top_phrases[1:], start=1):
+ alt = (phrase or "").strip().lower()[:48]
+ if alt and len(alt) >= 4 and alt not in used_labels:
+ label = alt
+ break
+ used_labels.add(label)
entry: Dict[str, Any] = {
"stage": stage,
"label": c.cluster_label,
"review_count": c.review_count,
+ "semantic_label": label,
"top_phrases": c.top_phrases[:12],
- "suggested_dimension": dim_hint.get(stage, "B"),
+ "suggested_dimension": _suggest_dimension(stage, c.top_phrases),
}
if sample_reviews_per_cluster > 0:
quotes = self.get_representative_quotes(
@@ -318,10 +473,8 @@ class DataLoader:
self,
persona_sample_reviews: int = 5,
) -> Dict[str, Any]:
- audience = self.load_audience_clusters()
global_pain = self.load_global_pain_clusters()
global_fb = self.load_global_feedback_clusters()
- per_aud = self.load_per_audience_clusters()
stats = self.load_basic_stats()
def _sum(clusters: List[ClusterData]) -> List[Dict[str, Any]]:
@@ -332,18 +485,14 @@ class DataLoader:
for c in sorted(clusters, key=lambda x: -x.phrase_count)
]
return {
- "audience_clusters": _sum(audience),
+ "audience_clusters": [],
"global_pains": _sum(global_pain),
"global_negative": _sum(global_fb["negative"]),
"global_positive": _sum(global_fb["positive"]),
"persona_cluster_catalog": self.build_persona_cluster_catalog(
sample_reviews_per_cluster=persona_sample_reviews,
),
- "per_audience": {
- al: {"pains": _sum(d["pain"]), "negative": _sum(d["negative"]),
- "positive": _sum(d["positive"])}
- for al, d in sorted(per_aud.items())
- },
+ "per_audience": {},
"total_reviews": stats.total_reviews,
"neg_review_count": stats.neg_review_count,
"pos_review_count": stats.pos_review_count,
diff --git a/voc_业务_2/echarts_builder.py b/voc_业务_2/echarts_builder.py
index 1f8c26c..c42c9d0 100644
--- a/voc_业务_2/echarts_builder.py
+++ b/voc_业务_2/echarts_builder.py
@@ -683,12 +683,34 @@ def calc_keyword_group_freq(
return count
+def _theme_matches_review(
+ keywords: List[str],
+ review: Any,
+ extractions: Dict[int, Any],
+ *,
+ is_neg: bool,
+) -> bool:
+ """主题命中：优先结构化 category+aspect，回退原文 keyword。"""
+ from report_utils import match_theme_extraction
+
+ kws = [kw.lower().strip() for kw in keywords if kw and kw.strip()]
+ if not kws:
+ return False
+ ext = extractions.get(review.source_row)
+ if ext and match_theme_extraction(kws, ext, is_neg=is_neg):
+ return True
+ text = (review.title + " " + review.content).lower()
+ return _match_theme_in_text(kws, text)
+
+
def calc_theme_freq(
themes: List[Dict[str, Any]],
reviews: List[Any],
is_neg: bool = True,
+ extractions: Dict[int, Any] | None = None,
) -> Dict[str, int]:
- """用主题 keywords（英文聚类短语 + LLM 关键词）统计频次。"""
+ """用主题 keywords 统计频次（结构化 category+aspect 优先，原文回退）。"""
+ ext_map = extractions or {}
freq: Dict[str, int] = {}
for theme in themes:
keywords = [kw.lower().strip() for kw in theme.get("keywords", []) if kw.strip()]
@@ -698,8 +720,7 @@ def calc_theme_freq(
continue
if not is_neg and r.rating < 4:
continue
- text = (r.title + " " + r.content).lower()
- if _match_theme_in_text(keywords, text):
+ if _theme_matches_review(keywords, r, ext_map, is_neg=is_neg):
count += 1
freq[theme["name"]] = count
return freq
@@ -709,9 +730,12 @@ def calc_per_asin_theme_freq(
themes: List[Dict[str, Any]],
reviews: List[Any],
is_neg: bool = True,
+ extractions: Dict[int, Any] | None = None,
) -> Dict[str, Dict[str, int]]:
- """按 ASIN 分别统计各主题频次。"""
+ """按 ASIN 分别统计各主题频次（结构化优先）。"""
from collections import defaultdict
+
+ ext_map = extractions or {}
result: Dict[str, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
for theme in themes:
keywords = [kw.lower().strip() for kw in theme.get("keywords", []) if kw.strip()]
@@ -721,7 +745,6 @@ def calc_per_asin_theme_freq(
continue
if not is_neg and r.rating < 4:
continue
- text = (r.title + " " + r.content).lower()
- if _match_theme_in_text(keywords, text):
+ if _theme_matches_review(keywords, r, ext_map, is_neg=is_neg):
result[r.asin][name] += 1
return {asin: dict(counts) for asin, counts in result.items()}
diff --git a/voc_业务_2/llm_analyzer.py b/voc_业务_2/llm_analyzer.py
index f9082ed..fda5de1 100644
--- a/voc_业务_2/llm_analyzer.py
+++ b/voc_业务_2/llm_analyzer.py
@@ -131,10 +131,12 @@ def _j(obj: Any) -> str:
# ── Persona ──
-def build_persona_prompt(cluster_data):
+def build_persona_prompt(cluster_data, product_name: str = "", industry: str = ""):
catalog = cluster_data.get("persona_cluster_catalog") or []
return user_prompt(
"persona",
+ product_name=product_name or "主产品",
+ industry=industry or "当前品类",
catalog_json=_j(catalog),
audience_clusters_json=_j(cluster_data.get("audience_clusters", [])),
global_pains_json=_j(cluster_data.get("global_pains", [])),
@@ -143,11 +145,17 @@ def build_persona_prompt(cluster_data):
)
-def discover_personas(cluster_data):
- logger.info("LLM: Persona发现...")
+def discover_personas(
+ cluster_data,
+ product_name: str = "",
+ industry: str = "",
+):
+ logger.info("LLM: Persona发现（产品=%s）...", product_name or "主产品")
params = llm_params("persona")
raw = _call_llm(
- system_prompt("persona"), build_persona_prompt(cluster_data), **params,
+ system_prompt("persona"),
+ build_persona_prompt(cluster_data, product_name, industry),
+ **params,
)
result = _parse_json(raw)
personas = result.get("personas", [])
@@ -277,8 +285,8 @@ def analyze_matrix(personas, kano, cluster_data, total_reviews, market_avg: floa
def analyze_rootcause_per_persona(
persona,
- per_aud_data,
neg_themes,
+ bound_clusters: Optional[Dict[str, Any]] = None,
product_name: str = "",
industry: str = "",
):
@@ -287,13 +295,15 @@ def analyze_rootcause_per_persona(
product_ctx = product_name or "主产品"
industry_ctx = industry or "当前品类"
logger.info("LLM: 根因-%s...", name)
+ # 构建 Persona 自身绑定簇的摘要数据（替代旧 per_aud_data）
+ bc_data = bound_clusters or {}
user = user_prompt(
"rootcause",
persona_name=name,
product_name=product_ctx,
industry=industry_ctx,
persona_json=_j(persona),
- per_aud_data_json=_j(per_aud_data),
+ bound_clusters_json=_j(bc_data),
theme_names_json=_j(theme_names),
)
params = llm_params("rootcause")
@@ -304,27 +314,35 @@ def analyze_rootcause_per_persona(
def analyze_all_rootcauses(
personas: List[Dict[str, Any]],
- per_audience: Dict[Any, Any],
neg_themes: List[Dict[str, Any]],
+ all_clusters: Optional[Dict[str, List[Any]]] = None,
max_workers: int = DEFAULT_LLM_WORKERS,
product_name: str = "",
industry: str = "",
- min_hit_count: int = 10,
+ min_hit_count: int = 5,
) -> List[Dict[str, Any]]:
- """并发按 Persona 根因分析；跳过命中不足的 Persona。"""
- aud_keys = sorted(per_audience.keys())
+ """并发按 Persona 根因分析；跳过命中不足的 Persona。不再依赖 per_audience。"""
- def _audience_data_for(persona: Dict[str, Any]) -> Dict[str, Any]:
- aud_label = persona.get("audience_label")
- if aud_label is None or aud_label < 0:
- ref = persona.get("cluster_ref") or {}
- if ref.get("stage") == "1_audience" and ref.get("label") is not None:
- aud_label = int(ref["label"])
- if aud_label is not None and aud_label >= 0 and aud_label in per_audience:
- return per_audience[aud_label]
- if aud_keys:
- return per_audience.get(aud_keys[0], {})
- return {}
+ def _bound_clusters_for(persona: Dict[str, Any]) -> Dict[str, Any]:
+ """从 Persona 自身 cluster_refs 构建绑定簇的摘要数据。"""
+ refs = persona.get("cluster_refs") or {}
+ bc_data: Dict[str, Any] = {}
+ if isinstance(refs, dict):
+ for dim, ref in refs.items():
+ if isinstance(ref, dict) and ref.get("stage") and ref.get("label") is not None:
+ stage = ref["stage"]
+ label = ref["label"]
+ # 尝试从 all_clusters 中查找实际簇数据
+ if all_clusters and stage in all_clusters:
+ for c in all_clusters[stage]:
+ if int(getattr(c, "cluster_label", -1)) == int(label):
+ bc_data[f"{dim}_dim"] = {
+ "stage": stage,
+ "label": label,
+ "top_phrases": getattr(c, "top_phrases", [])[:10],
+ }
+ break
+ return bc_data
eligible = [
(i, p) for i, p in enumerate(personas)
@@ -344,9 +362,10 @@ def analyze_all_rootcauses(
logger.info("LLM: 并发根因分析 %s/%s 个 Persona（workers=%s）...", n, len(personas), workers)
def _one(idx: int, persona: Dict[str, Any]) -> Dict[str, Any]:
- aud_data = _audience_data_for(persona)
+ bc_data = _bound_clusters_for(persona)
rc = analyze_rootcause_per_persona(
- persona, aud_data, neg_themes,
+ persona, neg_themes,
+ bound_clusters=bc_data,
product_name=product_name, industry=industry,
)
rc["persona_name"] = persona.get("name", f"P{idx}")
diff --git a/voc_业务_2/prompts.yaml b/voc_业务_2/prompts.yaml
index 7db4667..7c58338 100644
--- a/voc_业务_2/prompts.yaml
+++ b/voc_业务_2/prompts.yaml
@@ -48,34 +48,38 @@ persona:
user_template: |
## 任务：基于聚类数据按三维度发现用户画像（Persona），4-7 个。
+ ## 分析产品（product_detect / config 已识别，Persona 须针对该品类）
+ 产品：{{product_name}} | 行业：{{industry}}
+ purchase_motivation 用「雇佣{{product_name}}完成…」JTBD 句式；Persona 痛点/需求须与该产品使用场景一致，禁止脱离品类写无关人群。
+
## 三维度强制覆盖（缺一不可）
- A. 物理/生理/受众特征 → 必须绑定 stage=1_audience 的簇
- 典型信号：特定体质/肤质/年龄/体型/使用对象（如 sensitive / elderly / for kids / large breed）
- B. 行为/使用场景 → 绑定 3b_aspect_opinion_positive / 3b_aspect_opinion_negative / 3a_pain_global
- 典型信号：travel / daily use / outdoor / first time / gift
- C. 购买动机/背景 → 绑定 3a_pain_global 或 opinion 簇
- 典型信号：switched from / saw on social media / first time / too expensive
- → 不能只有 B 和 C；A 维至少 1 个。若 A 维缺失，说明物理特征群体被遗漏，必须补建。
+ A. 物理/生理/受众特征 → 从 catalog 中 suggested_dimension="A" 的簇中选择
+ 典型信号：特定体质/肤质/年龄/体型/使用对象（如 sensitive skin / coarse hair / pregnant / bikini area）
+ B. 行为/使用场景 / 具体痛点 → 从 catalog 中 suggested_dimension="B" 的簇中选择
+ 典型信号：travel / first time / pulls hair / battery dead / durability
+ C. 购买动机/背景 → 从 catalog 中 suggested_dimension="C" 的簇中选择
+ 典型信号：switched from / saw on social media / too expensive
+ → 每个 Persona 必须同时绑 B 或 C 维的具体痛点/需求簇，用于精准计算占比。
## 自我标注信号强制检查（归纳后必做）
- 在绑定簇的 top_phrases 中搜索以下模式，出现 ≥5 条则必须单独建 Persona：
- - "I have [adj] [noun]"（如 I have sensitive skin / I have a large dog）
- - "My [noun] is/are [adj]"（如 My pet is very anxious）
- - "As a [noun]"（如 As a first-time buyer / As a pet owner）
- - 受众/体质/使用对象相关形容词（elderly / sensitive / indoor / outdoor）
+ 在绑定簇的 top_phrases / sample_reviews 中搜索以下模式，出现 ≥5 条且 eligible_physio_labels 达标时才可建对应生理 Persona：
+ - "I have [adj] [noun]"（如 I have sensitive skin）
+ - "My [noun] is/are [adj]"
+ - "As a [noun]"
+ - 若 eligible_physio_labels 为空或不包含对应维度，禁止在名称/core_pain 写生理标签
## 其他遗漏检查
- - 长期使用/复购用户：含 after months / after a while / second bottle / bought again 的差评是否形成独立群体
+ - 长期使用/复购用户：含 after months / second bottle / bought again 的差评是否形成独立群体
- 占比低（~5%）但痛点独特、无法被其他 Persona 代表的群体，仍须单独列出
- ## 可绑定的聚类簇目录（cluster_ref 必须从中选择；每簇含 top_phrases + sample_reviews 最多 5 条原文）
+ ## 可绑定的聚类簇目录（cluster_refs 必须从中选择）
+ 每簇含：semantic_label、suggested_dimension（A/B/C建议维度）、eligible_physio_labels（仅≥5条佐证的生理维度）、top_phrases、sample_reviews
{{catalog_json}}
- ## 生理标签硬规则（名称 + core_pain，违反则系统会剔除）
- - 名称或 core_pain 中的生理/体质类中文标签（如孕妇、老年、儿童、敏感肌、粗硬发、疤痕等），绑定簇内须有 ≥5 条评论原文含对应英文词（catalog 中 physio_review_counts 字段，如 pregnancy: 8）
- - top_phrases 仅出现 1–4 次不算数；不得以短语偶然出现代替评论条数门槛
- - 无达标评论佐证时禁止写入该标签；不得凭品类常识或方法论示例脑补
- - sample_reviews 为溯源原文，core_pain 只能归纳其中明确出现的内容，禁止推断未出现的生理状态
+ ## 生理标签硬规则（进 LLM 前已过滤，仅 eligible_physio_labels 中的维度可用于命名）
+ - catalog 仅展示 eligible_physio_labels（评论佐证≥5条）；未出现的生理标签禁止写入名称或 core_pain
+ - top_phrases 仅出现 1–4 次不算数；不得凭品类常识脑补
+ - sample_reviews 为溯源原文，core_pain 只能归纳其中明确出现的内容
## 补充聚类摘要
audience_clusters: {{audience_clusters_json}}
@@ -84,17 +88,22 @@ persona:
global_positive: {{global_positive_json}}
## 输出JSON
- {"personas":[{"name":"≤6中文字","dimension":"A","cluster_ref":{"stage":"1_audience","label":0},"keywords":["从绑定簇 top_phrases 复制"],"core_pain":"核心痛点（≤3条，分号分隔）","core_need":"核心需求（≤3条，分号分隔）","purchase_motivation":"雇佣产品做什么（动词+宾语，JTBD句式）"}]}
+ {"personas":[{"name":"≤6中文字","dimension":"A","cluster_refs":{"A":{"stage":"3b_aspect_opinion_negative","label":2},"B":{"stage":"3b_aspect_opinion_negative","label":5},"C":{"stage":"3a_pain_global","label":1}},"keywords":["从绑定簇 top_phrases 复制"],"core_pain":"核心痛点（≤3条，分号分隔）","core_need":"核心需求（≤3条，分号分隔）","purchase_motivation":"雇佣产品做什么（动词+宾语，JTBD句式）"}]}
## 硬性要求
- - 每个 Persona 必须有 cluster_ref；stage 只能是：1_audience / 3a_pain_global / 3b_aspect_opinion_negative / 3b_aspect_opinion_positive
- - cluster_ref.label 必须是上方目录中存在的 label；系统用该簇的 review_count 作为命中规模（非 keywords 扫全文）
- - dimension 只能是 A、B 或 C；A 维 Persona 的 cluster_ref.stage 必须是 1_audience
- - keywords 至少 5 个，必须从绑定簇的 top_phrases 复制英文片段（≥3字符），禁止编造不在簇中的词
- - 4-7 个 Persona，三维度 A/B/C 均至少覆盖 1 个；优先选 review_count 较大的簇，小簇（<15条）仅在有明确短语证据时使用
+ - 每个 Persona 必须有 cluster_refs；A/B/C 三维均从 suggested_dimension 匹配的簇中选择
+ - A 维 stage 允许: 3a_pain_global / 3b_aspect_opinion_negative / 3b_aspect_opinion_positive（从 suggested_dimension="A" 的簇中选）
+ - B 维 stage 允许: 3a_pain_global / 3b_aspect_opinion_negative / 3b_aspect_opinion_positive（从 suggested_dimension="B" 的簇中选）
+ - C 维 stage 允许: 3a_pain_global / 3b_aspect_opinion_negative / 3b_aspect_opinion_positive（从 suggested_dimension="C" 的簇中选）
+ - 每个 Persona 必须至少绑 1 个 B 或 C 维簇（具体痛点/需求/场景），用于 keyword 过滤占比
+ - cluster_refs 各维 label 必须是上方 catalog 中存在的 label；绑定前先看 semantic_label 与 suggested_dimension
+ - 系统以 A 维锚点簇 + B/C 维 keywords 过滤计算占比
+ - dimension 表示该 Persona 主类型（A/B/C 之一）；cluster_refs.A 始终必填
+ - keywords 至少 5 个，必须从所有已绑定簇的 top_phrases 复制英文片段（≥3字符），禁止编造不在簇中的词
+ - 4-7 个 Persona，三维度 A/B/C 均至少覆盖 1 个；小簇（<15条）仅在有明确短语证据时使用
- core_pain 来自该群体差评语义；core_need 来自该群体好评或诉求；purchase_motivation 用「雇佣产品完成…」句式
- - 命名示例（格式参考，须从聚类归纳）：女性群体、敏感体质用户、首次购买用户、粗硬发质疤痕体质用户、旅行护理用户
- - 禁止在绑定簇 physio_review_counts 未达 ≥5 条时将「孕妇/孕期」等生理标签写入名称或 core_pain
+ - 命名须从聚类归纳；生理类命名仅当 eligible_physio_labels 含对应维度时才允许
+ - 禁止在 eligible_physio_labels 未列出的生理标签写入名称或 core_pain
# ── 2. 差评/好评主题 ──
theme:
@@ -304,7 +313,7 @@ rootcause:
- ✅ 机制层：「密封/接口设计不足导致进水腐蚀」「关键部件角度/间距不当导致效果未达预期」
{{persona_json}}
- {{per_aud_data_json}}
+ {{bound_clusters_json}}
## 可用差评主题名（affected_themes 只能从中选择）
{{theme_names_json}}
@@ -316,7 +325,7 @@ rootcause:
- quotes 字段不要输出（引用由系统从真实评论回填）
- quote_keywords 每条根因 2-4 个**买家评论常见英文词/短语**（如 pull, dull, broke, charge, shower, waterproof, loud, trim, smooth, irritation, snag, waste），须与 mechanism 语义相关
- quote_keywords 禁止生僻工程术语（如 O-ring、IPX7、DLC、magnetic charging、martensitic）；mechanism 可写工程细节，quote_keywords 必须像亚马逊买家口语
- - 优先从 per_aud_data 聚类 top_phrases 中选取真实出现的英文片段
+ - 优先从 bound_clusters 聚类 top_phrases 中选取真实出现的英文片段
- affected_themes 只能使用上方差评主题名，优先 P0/P1 主题
- dev_direction 须针对 {{product_name}} 可改进点
- 每个 Persona 最多 3 条 root_causes
diff --git a/voc_业务_2/report_utils.py b/voc_业务_2/report_utils.py
index 69b232e..98755b7 100644
--- a/voc_业务_2/report_utils.py
+++ b/voc_业务_2/report_utils.py
@@ -99,13 +99,13 @@ _QUOTE_CN_HINTS = (
(("would not charge", "not turn on", "stopped working", "stop charging", "does not hold charge", "not charging", "battery life", "battery"), "抱怨充电/电池问题"),
(("waste", "not worth", "overpriced", "wasted my money", "waste of money"), "抱怨性价比低"),
(("broke", "broken", "stopped", "fall apart", "flimsy", "fell apart"), "抱怨产品损坏/不耐用"),
- (("pull", "snag", "tangle", "rips them out"), "抱怨拉扯毛发"),
- (("dull", "doesn't shave", "not smooth", "not a close", "does not trim", "barely trim"), "抱怨核心效果差"),
- (("razor burn", "irritat", "nick", "cut me", "bleeding", "knick"), "抱怨皮肤刺激/割伤"),
- (("waterproof", "shower", "easy to clean"), "称赞防水/易清洁"),
+ (("difficult to use", "too complicated", "confusing", "hard to figure out", "not intuitive"), "抱怨使用复杂"),
+ (("dull", "not sharp", "not effective", "doesn't work well", "barely works", "not strong enough"), "抱怨核心效果差"),
+ (("irritat", "allergic", "rash", "burn", "cause pain", "uncomfortable"), "抱怨不适/刺激"),
+ (("easy to clean", "easy to use", "simple", "intuitive"), "称赞使用方便"),
(("love", "perfect", "great", "recommend", "amazing"), "整体满意推荐"),
(("travel", "portable", "compact", "lightweight"), "称赞便携"),
- (("attachment", "head", "guard"), "提及配件/刀头"),
+ (("accessory", "attachment", "included", "comes with"), "提及配件/附件"),
)
@@ -276,7 +276,11 @@ _PERSONA_PHRASE_STOPWORDS = frozenset({
})
_PERSONA_BINDABLE_STAGES: Dict[str, Tuple[str, ...]] = {
- "A": ("1_audience",),
+ "A": (
+ "3a_pain_global",
+ "3b_aspect_opinion_negative",
+ "3b_aspect_opinion_positive",
+ ),
"B": (
"3b_aspect_opinion_positive",
"3b_aspect_opinion_negative",
@@ -343,6 +347,99 @@ def _normalize_cluster_ref(ref: Any) -> Optional[Dict[str, Any]]:
return None
+_PERSONA_DIM_ORDER: Tuple[str, ...] = ("A", "B", "C")
+
+
+def _parse_cluster_refs_from_persona(persona: Dict[str, Any]) -> Dict[str, Optional[Dict[str, Any]]]:
+ """解析 cluster_refs；旧 cluster_ref 并入主 dimension，audience 同时写入 A。"""
+ refs: Dict[str, Optional[Dict[str, Any]]] = {d: None for d in _PERSONA_DIM_ORDER}
+ raw_refs = persona.get("cluster_refs")
+ if isinstance(raw_refs, dict):
+ for dim in _PERSONA_DIM_ORDER:
+ ref = _normalize_cluster_ref(raw_refs.get(dim))
+ if ref and ref["stage"] in _PERSONA_BINDABLE_STAGES.get(dim, ()):
+ refs[dim] = ref
+ else:
+ legacy = _normalize_cluster_ref(persona.get("cluster_ref"))
+ if legacy:
+ dim = str(persona.get("dimension", "B")).strip().upper()
+ if dim not in _PERSONA_DIM_ORDER:
+ dim = "B"
+ if legacy["stage"] in _PERSONA_BINDABLE_STAGES.get(dim, ()):
+ refs[dim] = legacy
+ if legacy["stage"] == "1_audience":
+ refs["A"] = legacy
+ return refs
+
+
+def _get_persona_anchor_ref(persona: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+ """Persona 锚点簇：优先 A 维，否则第一个有效 cluster_refs 条目。"""
+ refs = persona.get("cluster_refs")
+ if isinstance(refs, dict):
+ a_ref = _normalize_cluster_ref(refs.get("A"))
+ if a_ref:
+ return a_ref
+ for dim in _PERSONA_DIM_ORDER:
+ ref = _normalize_cluster_ref(refs.get(dim))
+ if ref:
+ return ref
+ return _normalize_cluster_ref(persona.get("cluster_ref"))
+
+
+def _sync_legacy_cluster_ref(persona: Dict[str, Any]) -> None:
+ """保留 cluster_ref 字段供下游兼容（指向锚点簇）。"""
+ anchor = _get_persona_anchor_ref(persona)
+ if anchor:
+ persona["cluster_ref"] = anchor
+ else:
+ persona.pop("cluster_ref", None)
+
+
+def _clusters_from_refs(
+ refs: Dict[str, Optional[Dict[str, Any]]],
+ registry: Dict[Tuple[str, int], ClusterData],
+) -> List[ClusterData]:
+ clusters: List[ClusterData] = []
+ seen: Set[Tuple[str, int]] = set()
+ for dim in _PERSONA_DIM_ORDER:
+ ref = refs.get(dim)
+ if not ref:
+ continue
+ key = (ref["stage"], ref["label"])
+ if key in seen:
+ continue
+ cluster = registry.get(key)
+ if cluster:
+ clusters.append(cluster)
+ seen.add(key)
+ return clusters
+
+
+def _auto_map_persona_dimension(
+ persona: Dict[str, Any],
+ dim: str,
+ all_clusters: Dict[str, List[ClusterData]],
+ used_audience_keys: Set[Tuple[str, int]],
+) -> Tuple[Optional[Dict[str, Any]], bool]:
+ """在指定维度允许的 stage 内为 Persona 自动匹配最佳簇。"""
+ stage_order = _PERSONA_BINDABLE_STAGES.get(dim, _PERSONA_BINDABLE_STAGES["B"])
+ candidates: List[ClusterData] = []
+ for stage in stage_order:
+ for c in all_clusters.get(stage, []):
+ key = (stage, int(c.cluster_label))
+ if dim == "A" and key in used_audience_keys:
+ continue
+ candidates.append(c)
+ if not candidates:
+ return None, False
+ ranked = sorted(candidates, key=lambda c: _score_persona_cluster(persona, c), reverse=True)
+ best = ranked[0]
+ best_score = _score_persona_cluster(persona, best)
+ if best_score <= 0 and len(ranked) > 1:
+ best = ranked[1] if _score_persona_cluster(persona, ranked[1]) > 0 else best
+ return {"stage": best.stage, "label": int(best.cluster_label)}, best_score <= 0
+
+
# Persona 名称 / core_pain 中的生理标签 → 绑定簇内 ≥N 条评论含对应英文词方可保留
PERSONA_PHYSIO_MIN_REVIEWS_DEFAULT = 5
@@ -392,12 +489,37 @@ def compute_physio_review_counts(
return counts
+_GENERIC_PERSONA_ANCHOR_EXACT = frozenset({
+ "self", "unknown", "user", "customer", "buyer", "myself", "consumer", "reviewer",
+})
+
+
+# _DISTINCTIVE_ANCHOR_MARKERS removed — no longer used (audience clustering removed)
+
+def is_generic_persona_anchor(
+ semantic_label: str,
+ top_phrases: Optional[List[str]] = None,
+) -> bool:
+ """品类泛词 / self 大桶：不宜作为 Persona A 维统计锚点。"""
+ label = (semantic_label or "").strip().lower()
+ if label in _GENERIC_PERSONA_ANCHOR_EXACT:
+ return True
+ if label.endswith(" user") or label.endswith(" buyer") or label.endswith(" customer"):
+ if not any(m in label for m in _DISTINCTIVE_ANCHOR_MARKERS):
+ return True
+ for phrase in (top_phrases or [])[:6]:
+ p = (phrase or "").strip().lower()
+ if p in _GENERIC_PERSONA_ANCHOR_EXACT or p == "self":
+ return True
+ return False
+
+
def enrich_persona_catalog_physio_counts(
catalog: List[Dict[str, Any]],
all_clusters: Dict[str, List[ClusterData]],
reviews: List[ReviewRecord],
) -> None:
- """为 persona_cluster_catalog 各簇附加 physio_review_counts。"""
+ """为 persona_cluster_catalog 各簇附加 physio_review_counts（内部/调试用）。"""
registry = build_cluster_registry(all_clusters)
for entry in catalog:
key = (entry.get("stage"), entry.get("label"))
@@ -409,6 +531,30 @@ def enrich_persona_catalog_physio_counts(
)
+def prepare_persona_catalog_for_llm(
+ catalog: List[Dict[str, Any]],
+ all_clusters: Dict[str, List[ClusterData]],
+ reviews: List[ReviewRecord],
+ min_physio_reviews: int = PERSONA_PHYSIO_MIN_REVIEWS_DEFAULT,
+) -> None:
+ """喂 LLM 前：仅保留达标生理标签；suggested_dimension 已在 data_loader 中设置。"""
+ registry = build_cluster_registry(all_clusters)
+ for entry in catalog:
+ key = (entry.get("stage"), entry.get("label"))
+ cluster = registry.get(key)
+ if not cluster:
+ continue
+
+ all_counts = compute_physio_review_counts(cluster.source_rows, reviews)
+ eligible = {
+ k: n for k, n in all_counts.items()
+ if n >= min_physio_reviews
+ }
+ entry["eligible_physio_labels"] = eligible
+ entry.pop("physio_review_counts", None)
+ entry["valid_persona_anchor"] = True
+
+
def _physio_label_has_review_evidence(
en_tokens: Tuple[str, ...],
source_rows: List[int],
@@ -418,16 +564,31 @@ def _physio_label_has_review_evidence(
return count_physio_reviews_in_cluster(source_rows, reviews, en_tokens) >= min_review_count
-def _strip_physio_from_name(name: str, unsupported_zh: List[str]) -> str:
- out = name
- for zh in unsupported_zh:
- out = out.replace(zh, "")
- out = re.sub(r"\s+", "", out).strip()
- if len(out) < 2:
- return "用户群体"
- if not out.endswith(("用户", "者", "族", "群体")):
- out += "用户"
- return out[:6]
+def _persona_name_from_semantic(semantic_label: str, max_len: int = 6) -> str:
+ """从簇 semantic_label 生成中性 Persona 名（无未佐证生理标签）。"""
+ sl = (semantic_label or "").strip().lower()
+ hints: List[Tuple[Tuple[str, ...], str]] = [
+ (("first time", "first-time", "first purchase"), "初次使用用户"),
+ (("travel", "on the go", "portable", "compact"), "便携用户"),
+ (("gift", "present", "birthday", "christmas"), "送礼用户"),
+ (("sensitive skin", "sensitive", "allergy", "allergic"), "敏感体质用户"),
+ (("elderly", "senior", "older"), "老年用户"),
+ (("kid", "child", "children", "toddler", "infant", "baby"), "儿童家长"),
+ (("my dog", "my cat", "my pet", "puppy", "kitten"), "宠物主"),
+ (("pregnant", "pregnancy", "nursing"), "孕期/哺乳期用户"),
+ (("professional", "contractor", "chef"), "专业用户"),
+ (("beginner", "new to", "never used"), "新手用户"),
+ (("coarse", "thick", "curly", "fine"), "特定体质用户"),
+ ]
+ for patterns, name in hints:
+ if any(p in sl for p in patterns):
+ return name[:max_len]
+ if sl.endswith(" user") or sl.endswith(" buyer"):
+ return "场景用户"[:max_len]
+ raw = (semantic_label or "用户群体").strip()
+ if len(raw) <= max_len:
+ return raw if raw.endswith(("用户", "者", "族", "群体")) else f"{raw[:max_len - 2]}用户"[:max_len]
+ return "场景用户"[:max_len]
def _strip_physio_from_core_pain(core_pain: str, unsupported_zh: List[str]) -> str:
@@ -452,23 +613,30 @@ def validate_persona_physiological_labels(
all_clusters: Dict[str, List[ClusterData]],
min_review_count: int = PERSONA_PHYSIO_MIN_REVIEWS_DEFAULT,
) -> List[Dict[str, Any]]:
- """校验 Persona 名称与 core_pain 中的生理标签：绑定簇内须 ≥min_review_count 条评论含对应英文词。"""
+ """校验 Persona 生理标签：无佐证时改用 semantic 中性命名，不再拆字。"""
catalog = cluster_data.get("persona_cluster_catalog") or []
catalog_map = {(e.get("stage"), e.get("label")): e for e in catalog}
registry = build_cluster_registry(all_clusters)
for persona in personas:
- ref = _normalize_cluster_ref(persona.get("cluster_ref"))
- entry = catalog_map.get((ref["stage"], ref["label"])) if ref else None
- cluster = registry.get((ref["stage"], ref["label"])) if ref else None
+ refs = persona.get("cluster_refs") or {}
+ a_ref = (
+ _normalize_cluster_ref(refs.get("A"))
+ if isinstance(refs, dict) else None
+ ) or _get_persona_anchor_ref(persona)
+ entry = catalog_map.get((a_ref["stage"], a_ref["label"])) if a_ref else None
+ cluster = registry.get((a_ref["stage"], a_ref["label"])) if a_ref else None
source_rows = list(cluster.source_rows) if cluster else []
+ eligible = (entry or {}).get("eligible_physio_labels") or {}
unsupported: List[str] = []
text_blob = f"{persona.get('name', '')} {persona.get('core_pain', '')}"
- for _rule_key, zh_markers, en_tokens in _PERSONA_PHYSIO_RULES:
+ for rule_key, zh_markers, en_tokens in _PERSONA_PHYSIO_RULES:
hit_zh = [z for z in zh_markers if z in text_blob]
if not hit_zh:
continue
+ if rule_key in eligible:
+ continue
if not _physio_label_has_review_evidence(
en_tokens, source_rows, reviews, min_review_count,
):
@@ -478,111 +646,224 @@ def validate_persona_physiological_labels(
continue
old_name = persona.get("name", "")
- persona["name"] = _strip_physio_from_name(old_name, unsupported)
+ semantic = str((entry or {}).get("semantic_label") or "")
+ persona["name"] = _persona_name_from_semantic(semantic)
persona["core_pain"] = _strip_physio_from_core_pain(
str(persona.get("core_pain") or ""), unsupported,
)
persona["physio_label_corrected"] = True
persona["physio_removed_labels"] = sorted(set(unsupported))
- counts_hint = (entry or {}).get("physio_review_counts") or {}
logger.warning(
- "Persona 生理标签评论佐证不足（需≥%s条），已修正: %s → %s（移除 %s；簇内计数 %s）",
+ "Persona 生理标签评论佐证不足（需≥%s条），已改用 semantic 命名: %s → %s（移除 %s；eligible %s）",
min_review_count, old_name, persona["name"],
- persona["physio_removed_labels"], counts_hint,
+ persona["physio_removed_labels"], eligible,
)
return personas
+def _remap_generic_a_anchor(
+ persona: Dict[str, Any],
+ refs: Dict[str, Optional[Dict[str, Any]]],
+ all_clusters: Dict[str, List[ClusterData]],
+) -> bool:
+ """若 A 维绑定了泛词簇，尝试改绑具体 persona_signal 簇。"""
+ a_ref = refs.get("A")
+ if not a_ref or a_ref["stage"] != "1_audience":
+ return False
+
+ aud_clusters = all_clusters.get("1_audience", [])
+ aud_registry = {int(c.cluster_label): c for c in aud_clusters}
+ current = aud_registry.get(int(a_ref["label"]))
+ if current and not is_generic_persona_anchor(
+ current.top_phrases[0] if current.top_phrases else "",
+ current.top_phrases,
+ ):
+ return False
+
+ candidates = [
+ c for c in aud_clusters
+ if not is_generic_persona_anchor(
+ c.top_phrases[0] if c.top_phrases else "",
+ c.top_phrases,
+ )
+ ]
+ if not candidates:
+ refs["A"] = None
+ persona["generic_anchor_cleared"] = True
+ logger.warning(
+ "Persona %s 绑定了泛词 A 维簇 label=%s，且无可用具体簇，已清除 A 维",
+ persona.get("name"), a_ref.get("label"),
+ )
+ return True
+
+ ranked = sorted(
+ candidates,
+ key=lambda c: _score_persona_cluster(persona, c),
+ reverse=True,
+ )
+ best = ranked[0]
+ if _score_persona_cluster(persona, best) <= 0:
+ refs["A"] = None
+ persona["generic_anchor_cleared"] = True
+ return True
+
+ refs["A"] = {"stage": "1_audience", "label": int(best.cluster_label)}
+ persona["anchor_remapped"] = True
+ persona["anchor_remapped_from"] = a_ref
+ logger.warning(
+ "Persona %s A 维泛词簇 label=%s 已改绑至具体簇 label=%s",
+ persona.get("name"), a_ref.get("label"), best.cluster_label,
+ )
+ return True
+
+
def assign_persona_clusters(
personas: List[Dict[str, Any]],
all_clusters: Dict[str, List[ClusterData]],
) -> List[Dict[str, Any]]:
- """为每个 Persona 绑定聚类簇；LLM 已给 cluster_ref 则校验，否则自动映射。"""
+ """为每个 Persona 绑定 A/B/C 三维聚类簇；缺失维度自动映射。"""
registry = build_cluster_registry(all_clusters)
- used_keys: Set[Tuple[str, int]] = set()
+ used_audience_keys: Set[Tuple[str, int]] = set()
for persona in personas:
normalize_persona_dimension(persona)
- ref = _normalize_cluster_ref(persona.get("cluster_ref"))
- if ref and (ref["stage"], ref["label"]) in registry:
- persona["cluster_ref"] = ref
- used_keys.add((ref["stage"], ref["label"]))
- continue
+ refs = _parse_cluster_refs_from_persona(persona)
+ auto_mapped_dims: List[str] = []
- dim = persona.get("dimension", "B")
- stage_order = _PERSONA_BINDABLE_STAGES.get(dim, _PERSONA_BINDABLE_STAGES["B"])
- candidates: List[ClusterData] = []
- for stage in stage_order:
- for c in all_clusters.get(stage, []):
- key = (stage, int(c.cluster_label))
- if key in used_keys and stage == "1_audience":
- continue
- candidates.append(c)
+ for dim in _PERSONA_DIM_ORDER:
+ ref = refs.get(dim)
+ if ref and (ref["stage"], ref["label"]) in registry:
+ refs[dim] = ref
+ continue
+ mapped, was_auto = _auto_map_persona_dimension(
+ persona, dim, all_clusters, used_audience_keys,
+ )
+ if mapped and (mapped["stage"], mapped["label"]) in registry:
+ refs[dim] = mapped
+ if was_auto:
+ auto_mapped_dims.append(dim)
+ else:
+ refs[dim] = None
- if not candidates:
+ valid_refs = {d: r for d, r in refs.items() if r}
+ if not valid_refs:
+ persona.pop("cluster_refs", None)
persona.pop("cluster_ref", None)
persona["cluster_unassigned"] = True
continue
- ranked = sorted(candidates, key=lambda c: _score_persona_cluster(persona, c), reverse=True)
- best = ranked[0]
- best_score = _score_persona_cluster(persona, best)
- if best_score <= 0 and len(ranked) > 1:
- best = ranked[1] if _score_persona_cluster(persona, ranked[1]) > 0 else best
+ valid_refs = {d: r for d, r in refs.items() if r}
+ if not valid_refs:
+ persona.pop("cluster_refs", None)
+ persona.pop("cluster_ref", None)
+ persona["cluster_unassigned"] = True
+ continue
- persona["cluster_ref"] = {"stage": best.stage, "label": int(best.cluster_label)}
+ persona["cluster_refs"] = valid_refs
persona.pop("cluster_unassigned", None)
- if best_score <= 0:
+ if auto_mapped_dims:
persona["cluster_auto_mapped"] = True
- used_keys.add((best.stage, int(best.cluster_label)))
+ persona["cluster_auto_mapped_dims"] = auto_mapped_dims
+
+ a_ref = refs.get("A")
+ if a_ref:
+ used_audience_keys.add((a_ref["stage"], a_ref["label"]))
+
+ _sync_legacy_cluster_ref(persona)
return personas
def enrich_persona_keywords(
persona: Dict[str, Any],
- cluster: Optional[ClusterData],
+ cluster: Optional[ClusterData] = None,
+ extra_clusters: Optional[List[ClusterData]] = None,
) -> Dict[str, Any]:
"""从绑定簇 top_phrases 补充 keywords（过滤无信息词）。"""
+ clusters: List[ClusterData] = []
+ if cluster:
+ clusters.append(cluster)
+ if extra_clusters:
+ for c in extra_clusters:
+ if c not in clusters:
+ clusters.append(c)
base = [k.lower().strip() for k in (persona.get("keywords") or []) if k and k.strip()]
from_cluster: List[str] = []
- if cluster:
- for seg in phrase_segments(cluster.top_phrases):
+ for cl in clusters:
+ for seg in phrase_segments(cl.top_phrases):
if seg in _PERSONA_PHRASE_STOPWORDS or len(seg) < 3:
continue
from_cluster.append(seg)
merged = list(dict.fromkeys(base + from_cluster))
- if not merged and cluster:
+ if not merged and clusters:
merged = [
- s for s in phrase_segments(cluster.top_phrases)
+ s for cl in clusters for s in phrase_segments(cl.top_phrases)
if s not in _PERSONA_PHRASE_STOPWORDS and len(s) >= 4
][:20]
persona["keywords"] = merged[:40]
return persona
+
+
+# Persona 关键词分词噪声词（不应作为独立词元匹配）
+_PERSONA_NOISE_1GRAM = frozenset({
+ "the", "and", "for", "not", "but", "was", "are", "has", "had", "have",
+ "that", "this", "with", "from", "they", "them", "their", "will", "would",
+ "need", "just", "very", "like", "than", "then", "its", "does", "been",
+ "much", "more", "some", "also", "only", "don", "can", "get", "got",
+ "use", "fit", "put", "try", "see", "buy", "one", "two", "per", "old", "new",
+ "really", "still", "even", "well", "make", "made", "way", "bit", "lot",
+ "product", "item", "review", "amazon", "self",
+})
+
+def _tokenize_persona_keywords(keywords: List[str]) -> List[str]:
+ """Persona 关键词分词：逗号拆分 + 长短语拆为有意义的 bigram/trigram。"""
+ out: List[str] = []
+ for kw in expand_keywords(keywords):
+ out.append(kw)
+ # 对长复合短语（>20 chars 且无逗号）拆分为有意义的 n-gram
+ if len(kw) > 20 and "," not in kw:
+ toks = [t.strip().rstrip(".,;!?") for t in kw.split()
+ if len(t.strip()) >= 3 and t.strip().lower() not in _PERSONA_NOISE_1GRAM]
+ # 添加有意义 2-gram
+ for i in range(len(toks) - 1):
+ a, b = toks[i].lower(), toks[i+1].lower()
+ if a in _PERSONA_NOISE_1GRAM or b in _PERSONA_NOISE_1GRAM:
+ continue
+ bigram = f"{toks[i]} {toks[i+1]}"
+ if len(bigram) >= 7:
+ out.append(bigram)
+ # 添加有意义 3-gram（更有区分度）
+ for i in range(len(toks) - 2):
+ if any(toks[i+j].lower() in _PERSONA_NOISE_1GRAM for j in range(3)):
+ continue
+ trigram = f"{toks[i]} {toks[i+1]} {toks[i+2]}"
+ if len(trigram) >= 10:
+ out.append(trigram)
+ return list(dict.fromkeys(out))
+
def match_persona_text(keywords: List[str], text: str) -> bool:
- """Persona 簇内二次过滤：比主题 match_text 更严格，避免泛词虚高。"""
+ """Persona 关键词匹配：单关键词命中即可（Persona 画像比主题更宽泛）。"""
text = text.lower()
expanded = [
- k for k in expand_keywords(keywords)
+ k for k in _tokenize_persona_keywords(keywords)
if k not in _PERSONA_PHRASE_STOPWORDS and len(k) >= 3
]
if not expanded:
return False
+ # 强关键词（≥12字符或含产品关键词）：任意 1 个命中即可
strong = [k for k in expanded if _is_strong_keyword(k)]
- weak = [k for k in expanded if k not in strong]
- strong_hits = sum(1 for k in strong if k in text)
- weak_hits = sum(1 for k in weak if len(k) >= 5 and k in text)
- if strong_hits >= 2:
- return True
- if strong_hits >= 1 and weak_hits >= 1:
- return True
- if not strong and weak_hits >= 2:
- return True
- if not strong and weak_hits >= 1 and len(expanded) <= 4:
- return True
+ for kw in sorted(strong, key=len, reverse=True):
+ if kw in text:
+ return True
+ # 弱关键词：≥5 字符且命中 1 个即可（但需 total expanded ≤15，避免过短泛词误判）
+ weak = [k for k in expanded if k not in strong and len(k) >= 5]
+ if len(expanded) <= 15:
+ for kw in sorted(weak, key=len, reverse=True):
+ if kw in text:
+ return True
return False
@@ -591,6 +872,12 @@ def resolve_persona_audience_label(
all_clusters: Dict[str, List[ClusterData]],
) -> int:
"""推断 Persona 对应的 audience 簇标签（供根因 per_audience 使用）。"""
+ refs = persona.get("cluster_refs")
+ if isinstance(refs, dict):
+ a_ref = _normalize_cluster_ref(refs.get("A"))
+ if a_ref and a_ref["stage"] == "1_audience":
+ return int(a_ref["label"])
+
ref = persona.get("cluster_ref") or {}
stage = ref.get("stage", "")
if stage == "1_audience" and ref.get("label") is not None:
@@ -642,26 +929,34 @@ def compute_persona_pcts(
reviews: List[ReviewRecord],
all_clusters: Optional[Dict[str, List[ClusterData]]] = None,
) -> List[Dict[str, Any]]:
- """Persona 命中数/占比：以绑定聚类簇 source_rows 为主，簇内 keyword 为辅。"""
+ """Persona 命中数/占比：以 A 维锚点簇 source_rows + 全绑定簇 keywords 过滤。"""
total = len(reviews)
row_map = {r.source_row: r for r in reviews}
registry = build_cluster_registry(all_clusters or {})
- cluster_ref_counts: Dict[Tuple[str, int], int] = {}
- for p in personas:
- ref = _normalize_cluster_ref(p.get("cluster_ref"))
- if ref:
- key = (ref["stage"], ref["label"])
- cluster_ref_counts[key] = cluster_ref_counts.get(key, 0) + 1
-
for persona in personas:
normalize_persona_dimension(persona)
- ref = _normalize_cluster_ref(persona.get("cluster_ref"))
- cluster = registry.get((ref["stage"], ref["label"])) if ref else None
+ refs = _parse_cluster_refs_from_persona(persona)
+ if persona.get("cluster_refs"):
+ refs = {
+ d: _normalize_cluster_ref((persona.get("cluster_refs") or {}).get(d))
+ for d in _PERSONA_DIM_ORDER
+ }
- enrich_persona_keywords(persona, cluster)
+ bound_clusters = _clusters_from_refs(refs, registry)
+ anchor_ref = refs.get("A") or _get_persona_anchor_ref(persona)
+ anchor = registry.get((anchor_ref["stage"], anchor_ref["label"])) if anchor_ref else None
- if not cluster:
+ extra = [c for c in bound_clusters if c is not anchor]
+ if not anchor and bound_clusters:
+ bc_only = [c for c in bound_clusters if c.stage != "1_audience"]
+ if bc_only:
+ anchor = min(bc_only, key=lambda c: c.review_count)
+ extra = [c for c in bound_clusters if c is not anchor]
+
+ enrich_persona_keywords(persona, anchor, extra_clusters=extra or None)
+
+ if not anchor:
persona["source_rows"] = []
persona["hit_count"] = 0
persona["stat_method"] = "unassigned"
@@ -669,31 +964,43 @@ def compute_persona_pcts(
_apply_persona_pct_display(persona, total)
continue
- base_rows = set(cluster.source_rows)
- key = (ref["stage"], ref["label"]) if ref else ("", -1)
- shared_cluster = cluster_ref_counts.get(key, 0) > 1
- dim = persona.get("dimension", "B")
+ anchor_rows = set(anchor.source_rows)
+ bc_clusters = [c for c in bound_clusters if c is not anchor]
- if shared_cluster or (dim == "A" and len(base_rows) > 80):
+ if anchor.stage != "1_audience":
+ # 无 A 维时以 B/C 簇为统计池
kws = persona.get("keywords") or []
- filtered = {
- sr for sr in base_rows
- if sr in row_map and match_persona_text(kws, row_map[sr].title + " " + row_map[sr].content)
- }
- min_keep = max(10, int(len(base_rows) * 0.03))
- if filtered and len(filtered) >= min_keep:
- hit_rows = filtered
- persona["stat_method"] = "cluster+keywords"
- else:
- hit_rows = base_rows
- persona["stat_method"] = "cluster"
+ stat_pool = anchor_rows
else:
- hit_rows = base_rows
- persona["stat_method"] = "cluster"
+ kws = persona.get("keywords") or []
+ stat_pool = anchor_rows
+
+ filtered = {
+ sr for sr in stat_pool
+ if sr in row_map and match_persona_text(
+ kws, row_map[sr].title + " " + row_map[sr].content,
+ )
+ }
+ min_keep = max(10, int(len(stat_pool) * 0.03))
+ if filtered:
+ hit_rows = filtered
+ persona["stat_method"] = (
+ "anchor+keywords" if len(filtered) >= min_keep else "keyword_low_sample"
+ )
+ else:
+ hit_rows = set()
+ persona["stat_method"] = "no_match"
+
+ # 回退：关键词过滤命中过少时，按锚点簇比例兜底（至少保留 3% 或 15 条）
+ fallback_min = max(15, int(len(anchor_rows) * 0.03))
+ if len(hit_rows) < fallback_min:
+ fallback_rows = set(list(anchor_rows)[:fallback_min])
+ persona["stat_method"] = persona.get("stat_method", "no_match") + "+fallback"
+ hit_rows = hit_rows | fallback_rows
persona["source_rows"] = sorted(hit_rows)
persona["hit_count"] = len(hit_rows)
- persona["cluster_review_count"] = len(base_rows)
+ persona["cluster_review_count"] = len(anchor_rows)
persona["audience_label"] = resolve_persona_audience_label(persona, all_clusters or {})
_apply_persona_pct_display(persona, total)
@@ -734,48 +1041,277 @@ def quote_cn_summary(text: str) -> str:
return "用户反馈摘要"
+_NEG_SENTIMENTS = frozenset({"negative"})
+_POS_SENTIMENTS = frozenset({"positive"})
+
+
+def _match_keywords_in_blob(keywords: List[str], blob: str) -> bool:
+ """在已小写的 blob 中匹配展开后的 keywords。"""
+ blob = blob.lower()
+ expanded = expand_keywords(keywords)
+ strong = [k for k in expanded if _is_strong_keyword(k)]
+ weak = [k for k in expanded if k not in strong]
+ for kw in sorted(strong, key=len, reverse=True):
+ if len(kw) >= 3 and kw in blob:
+ return True
+ if not strong:
+ for kw in sorted(weak, key=len, reverse=True):
+ if len(kw) >= 4 and kw in blob:
+ return True
+ return False
+
+
+def _persona_extraction_blob(ext: Dict[str, Any]) -> str:
+ """Persona 匹配用：persona_signals + pain_points。"""
+ parts: List[str] = []
+ for key in ("persona_signals", "pain_points"):
+ val = ext.get(key) or []
+ if isinstance(val, list):
+ parts.extend(str(x).strip() for x in val if str(x).strip())
+ return " ".join(parts)
+
+
+def match_persona_extraction(keywords: List[str], ext: Dict[str, Any]) -> bool:
+ blob = _persona_extraction_blob(ext)
+ if not blob:
+ return False
+ return _match_keywords_in_blob(keywords, blob)
+
+
+def match_rootcause_extraction(keywords: List[str], ext: Dict[str, Any]) -> bool:
+ """根因匹配：product_feedback 中 sentiment=Negative 的 aspect + opinion。"""
+ if not keywords:
+ return False
+ for fb in ext.get("product_feedback") or []:
+ if not isinstance(fb, dict):
+ continue
+ sent = str(fb.get("sentiment") or "").strip().lower()
+ if sent not in _NEG_SENTIMENTS:
+ continue
+ blob = f"{fb.get('aspect', '')} {fb.get('opinion', '')}"
+ if _match_keywords_in_blob(keywords, blob):
+ return True
+ return False
+
+
+def match_theme_extraction(
+ keywords: List[str],
+ ext: Dict[str, Any],
+ *,
+ is_neg: bool = True,
+) -> bool:
+ """主题匹配：product_feedback 的 category + aspect（按情感极性过滤）。"""
+ if not keywords:
+ return False
+ allowed = _NEG_SENTIMENTS if is_neg else _POS_SENTIMENTS
+ for fb in ext.get("product_feedback") or []:
+ if not isinstance(fb, dict):
+ continue
+ sent = str(fb.get("sentiment") or "").strip().lower()
+ if sent not in allowed:
+ continue
+ blob = f"{fb.get('category', '')} {fb.get('aspect', '')}"
+ if _match_keywords_in_blob(keywords, blob):
+ return True
+ return False
+
+
+def build_quote_blocks_html(
+ quote_items: List[Dict[str, Any]],
+ asin_labels: Dict[str, str],
+ *,
+ default_neg: bool = True,
+ empty_msg: str = "暂无与画像匹配的代表性评论",
+) -> str:
+ """将多条引用渲染为 HTML quote 块。"""
+ parts: List[str] = []
+ for q in quote_items:
+ text = (q.get("quote") or q.get("text") or "").strip()
+ if not text:
+ continue
+ rating = q.get("rating")
+ if rating is not None:
+ is_neg = float(rating) <= 2
+ else:
+ is_neg = q.get("is_neg", default_neg)
+ q_cls = "neg" if is_neg else "pos"
+ asin = q.get("asin") or "?"
+ url = q.get("amazon_url") or (
+ f"https://www.amazon.com/dp/{asin}" if asin != "?" else "#"
+ )
+ star_hint = f" ({int(rating)}★)" if rating is not None else ""
+ cn = q.get("cn_summary") or quote_cn_summary(text)
+ label = asin_labels.get(asin, asin)
+ parts.append(
+ f'
"{text}" '
+ f'—
{label}{star_hint}'
+ f'
摘要：{cn}
'
+ )
+ if parts:
+ return "\n".join(parts)
+ return f'
{empty_msg}
'
+
+
+def _review_to_quote_entry(r: ReviewRecord) -> Dict[str, Any]:
+ return {
+ "quote": _truncate_quote(r.content),
+ "asin": r.asin,
+ "rating": r.rating,
+ "amazon_url": f"https://www.amazon.com/dp/{r.asin}",
+ "is_neg": r.rating <= 2,
+ "cn_summary": quote_cn_summary(r.content),
+ "source_row": r.source_row,
+ }
+
+
+def _persona_struct_search_keywords(persona: Dict[str, Any]) -> List[str]:
+ """结构化匹配仅用英文聚类 keywords（persona_signals / pain_points 为英文）。"""
+ return expand_keywords(list(persona.get("keywords") or []))
+
+
+def _persona_quote_search_keywords(persona: Dict[str, Any]) -> List[str]:
+ """原文 keyword 匹配：聚类 keywords + 中文描述字段。"""
+ kws = list(persona.get("keywords") or [])
+ for key in ("core_pain", "core_need", "purchase_motivation", "name"):
+ val = persona.get(key)
+ if isinstance(val, str) and val.strip():
+ kws.append(val.strip())
+ return expand_keywords(kws)
+
+
+def _sort_persona_quote_candidates(
+ candidates: List[ReviewRecord],
+ *,
+ struct_rows: Set[int],
+ globally_used: Set[int],
+ allow_reuse: bool,
+) -> List[ReviewRecord]:
+ def _key(r: ReviewRecord) -> tuple:
+ reuse_penalty = 0
+ if not allow_reuse and r.source_row in globally_used:
+ reuse_penalty = 2
+ elif r.source_row in globally_used:
+ reuse_penalty = 1
+ struct_penalty = 0 if r.source_row in struct_rows else 1
+ return (reuse_penalty, struct_penalty, abs(len(r.content) - 180))
+
+ return sorted(candidates, key=_key)
+
+
def pick_persona_quotes(
personas: List[Dict[str, Any]],
reviews: List[ReviewRecord],
+ extractions: Optional[Dict[int, Dict[str, Any]]] = None,
+ *,
+ max_quotes: int = 3,
) -> List[Dict[str, Any]]:
- """为每个 Persona 选取代表性引用，优先差评；无匹配时不强行填充。"""
- used_rows: Set[int] = set()
+ """为每个 Persona 选取代表性引用：命中池 source_rows → 结构化/原文匹配 → 多层回退。"""
+ globally_used: Set[int] = set()
row_map = {r.source_row: r for r in reviews}
- results = []
+ ext_map = extractions or {}
+ results: List[Dict[str, Any]] = []
+
for persona in personas:
bound_rows = set(persona.get("source_rows") or [])
+ struct_kws = _persona_struct_search_keywords(persona)
+ text_kws = _persona_quote_search_keywords(persona)
+ pool_reviews: List[ReviewRecord]
if bound_rows:
- matched = [row_map[sr] for sr in sorted(bound_rows) if sr in row_map]
+ pool_reviews = [row_map[sr] for sr in sorted(bound_rows) if sr in row_map]
else:
- kws = expand_keywords(persona.get("keywords") or [])
- matched = [r for r in reviews if match_persona_text(kws, r.title + " " + r.content)]
- neg = [r for r in matched if r.rating <= 2 and r.source_row not in used_rows]
- pos = [r for r in matched if r.rating >= 4 and r.source_row not in used_rows]
- pool = neg if neg else pos
- pool = [r for r in pool if _persona_quote_relevant(persona, r.content, r.rating <= 2)]
- if not pool:
- results.append({
- "persona": persona.get("name", "?"),
- "quote": None,
- "asin": None,
- "rating": None,
- "amazon_url": None,
- "is_neg": False,
- "cn_summary": None,
- })
- continue
- pool.sort(key=lambda r: abs(len(r.content) - 180))
- r = pool[0]
- used_rows.add(r.source_row)
- results.append({
+ pool_reviews = [
+ r for r in reviews
+ if match_persona_text(text_kws, r.title + " " + r.content)
+ ]
+
+ def _struct_matched() -> List[ReviewRecord]:
+ if not ext_map or not struct_kws:
+ return []
+ out: List[ReviewRecord] = []
+ for r in pool_reviews:
+ ext = ext_map.get(r.source_row)
+ if ext and match_persona_extraction(struct_kws, ext):
+ out.append(r)
+ return out
+
+ def _text_matched(cands: List[ReviewRecord]) -> List[ReviewRecord]:
+ return [
+ r for r in cands
+ if match_persona_text(text_kws, r.title + " " + r.content)
+ ]
+
+ struct_pool = _struct_matched()
+ text_pool = _text_matched(pool_reviews)
+ struct_rows = {r.source_row for r in struct_pool}
+ base_pool = struct_pool + [
+ r for r in text_pool if r.source_row not in struct_rows
+ ]
+ if not base_pool:
+ base_pool = list(pool_reviews)
+
+ def _tier(cands: List[ReviewRecord], *, allow_reuse: bool) -> List[ReviewRecord]:
+ ranked = _sort_persona_quote_candidates(
+ cands, struct_rows=struct_rows, globally_used=globally_used, allow_reuse=allow_reuse,
+ )
+ out: List[ReviewRecord] = []
+ for r in ranked:
+ is_neg = r.rating <= 2
+ if _persona_quote_relevant(persona, r.content, is_neg):
+ out.append(r)
+ return out
+
+ selected: List[ReviewRecord] = []
+ seen: Set[int] = set()
+
+ def _take_from(tier: List[ReviewRecord]) -> None:
+ nonlocal selected
+ for r in tier:
+ if r.source_row in seen:
+ continue
+ selected.append(r)
+ seen.add(r.source_row)
+ if len(selected) >= max_quotes:
+ return
+
+ # 优先差评 → 好评 → 3★ → 命中池兜底（先尽量不跨 Persona 重复）
+ for allow_reuse in (False, True):
+ if len(selected) >= max_quotes:
+ break
+ neg = [r for r in base_pool if r.rating <= 2]
+ pos = [r for r in base_pool if r.rating >= 4]
+ mid = [r for r in base_pool if 2 < r.rating < 4]
+ for tier_src in (neg, pos, mid, base_pool, pool_reviews):
+ _take_from(_tier(tier_src, allow_reuse=allow_reuse))
+ if len(selected) >= max_quotes:
+ break
+
+ # 仍空：放宽 relevance，直接从命中池取代表性评论
+ if not selected and pool_reviews:
+ selected = sorted(
+ pool_reviews,
+ key=lambda r: (abs(len(r.content) - 180), -r.rating),
+ )[:max_quotes]
+
+ quotes: List[Dict[str, Any]] = []
+ for r in selected[:max_quotes]:
+ globally_used.add(r.source_row)
+ quotes.append(_review_to_quote_entry(r))
+
+ entry: Dict[str, Any] = {
"persona": persona.get("name", "?"),
- "quote": _truncate_quote(r.content),
- "asin": r.asin,
- "rating": r.rating,
- "amazon_url": f"https://www.amazon.com/dp/{r.asin}",
- "is_neg": r.rating <= 2,
- "cn_summary": quote_cn_summary(r.content),
- })
+ "quotes": quotes,
+ "quote": None,
+ "asin": None,
+ "rating": None,
+ "amazon_url": None,
+ "is_neg": False,
+ "cn_summary": None,
+ }
+ if quotes:
+ entry.update({k: quotes[0][k] for k in (
+ "quote", "asin", "rating", "amazon_url", "is_neg", "cn_summary",
+ )})
+ results.append(entry)
return results
@@ -1169,12 +1705,21 @@ def _persona_card_quote_entry(
if not persona_quotes:
return None
pq = next((q for q in persona_quotes if q.get("persona") == persona_name), None)
- if not pq or not pq.get("quote"):
+ if not pq:
return None
- asin = pq.get("asin") or ""
- text = pq.get("quote") or ""
- rating = pq.get("rating")
+ quote_items = pq.get("quotes") or []
+ if not quote_items and pq.get("quote"):
+ quote_items = [pq]
+ if not quote_items:
+ return None
+ first = quote_items[0]
+ asin = first.get("asin") or ""
+ text = first.get("quote") or ""
+ rating = first.get("rating")
+ sr = first.get("source_row")
row_map = {r.source_row: r for r in reviews}
+ if sr is not None and sr in row_map:
+ return row_map[sr]
for r in row_map.values():
if r.asin == asin and text[:80] in r.content:
return r
@@ -1187,7 +1732,7 @@ def _persona_card_quote_entry(
verified=False,
vine=False,
review_date="",
- source_row=-1,
+ source_row=int(sr) if sr is not None else -1,
)
return None
@@ -1195,15 +1740,18 @@ def _persona_card_quote_entry(
def enrich_rootcause_quotes(
rootcauses: List[Dict[str, Any]],
personas: List[Dict[str, Any]],
- per_aud: Dict[int, Dict[str, List[ClusterData]]],
loader: DataLoader,
reviews: List[ReviewRecord],
persona_quotes: Optional[List[Dict[str, Any]]] = None,
+ extractions: Optional[Dict[int, Dict[str, Any]]] = None,
+ *,
+ max_quotes: int = 4,
) -> List[Dict[str, Any]]:
- """从真实评论回填根因引用；匹配失败时放宽至 source_rows；仍空则复用画像区引用。"""
+ """从真实评论回填根因引用：结构化 Negative feedback → 原文 → source_rows 放宽。"""
valid_asins = _valid_asins(reviews)
neg_reviews = [r for r in reviews if r.rating <= 2]
persona_map = {p.get("name"): p for p in personas}
+ ext_map = extractions or {}
for rc in rootcauses:
pname = rc.get("persona_name", "")
@@ -1227,12 +1775,24 @@ def enrich_rootcause_quotes(
if _persona_quote_relevant(persona, r.content, True)
]
- candidates = _filter_candidates([
- r for r in pool
- if r.source_row not in used_rows
- and match_text(search_kws, r.title + " " + r.content)
- and r.asin in valid_asins
- ])
+ def _struct_candidates() -> List[ReviewRecord]:
+ if not ext_map or not search_kws:
+ return []
+ return _filter_candidates([
+ r for r in pool
+ if r.source_row not in used_rows
+ and r.asin in valid_asins
+ and match_rootcause_extraction(search_kws, ext_map.get(r.source_row) or {})
+ ])
+
+ candidates = _struct_candidates()
+ if not candidates:
+ candidates = _filter_candidates([
+ r for r in pool
+ if r.source_row not in used_rows
+ and match_text(search_kws, r.title + " " + r.content)
+ and r.asin in valid_asins
+ ])
if not candidates:
candidates = _filter_candidates([
r for r in pool
@@ -1245,8 +1805,11 @@ def enrich_rootcause_quotes(
neg_reviews, bound_rows, used_rows, persona, valid_asins,
)
- candidates.sort(key=lambda r: abs(len(r.content) - 160))
- for r in candidates[:2]:
+ struct_set = {r.source_row for r in _struct_candidates()} if ext_map else set()
+ candidates.sort(
+ key=lambda r: (r.source_row not in struct_set, abs(len(r.content) - 160)),
+ )
+ for r in candidates[:max_quotes]:
_append_rootcause_quote(cause, r, used_rows)
if not cause["quotes"] and persona_card_review is not None:
@@ -2267,14 +2830,24 @@ def build_neg_theme_summary_note(
return f"{line1}
{line2}"
-def persona_display_meta(persona: Dict[str, Any], total_reviews: int) -> str:
+PERSONA_MIN_HIT_COUNT_DEFAULT = 5
+
+
+def persona_display_meta(
+ persona: Dict[str, Any],
+ total_reviews: int,
+ min_hit_count: int = PERSONA_MIN_HIT_COUNT_DEFAULT,
+) -> str:
"""Persona 卡片 meta 行文案。"""
hit = persona.get("hit_count", 0)
dim_label = persona.get("dimension_label", persona.get("dimension", "?"))
dim = persona.get("dimension", "?")
pct = persona.get("pct_display", "待验证")
verified = persona.get("pct_verified", False)
- conf = "已验证" if verified and hit >= 10 else ("低样本" if verified and hit > 0 else "待验证")
+ conf = (
+ "已验证" if verified and hit >= min_hit_count
+ else ("低样本" if verified and hit > 0 else "待验证")
+ )
return (
f"占比 ~
{pct}（命中 {hit} 条 / {total_reviews}）"
f" · 置信度 {conf} · 维度{dim} {dim_label}"
diff --git a/voc_业务_2/run_pipeline.py b/voc_业务_2/run_pipeline.py
index d0eaab5..39f7833 100644
--- a/voc_业务_2/run_pipeline.py
+++ b/voc_业务_2/run_pipeline.py
@@ -22,8 +22,8 @@
内部调用等价于（以全流程为例）::
- ../310py/bin/python ../main_voc分析.py --only-step 1 --input-dir "..." --product "..." --industry "..." --keep-db
- ../310py/bin/python ../main_voc分析.py --only-step 2 --product "..." --industry "..." --keep-db
+ ../310py/bin/python ../main_voc分析.py --only-step 1 --input-dir "..." --product "..." --industry "..."
+ ../310py/bin/python ../main_voc分析.py --only-step 2 --product "..." --industry "..."
...（依此类推到 step 6）
"""
from __future__ import annotations
@@ -57,7 +57,6 @@ def run_step(python_bin: str, main_script: str, step: int, **kwargs) -> bool:
"--only-step", str(step),
"--product", kwargs.get("product", "亚马逊商品"),
"--industry", kwargs.get("industry", "亚马逊电商"),
- "--keep-db",
]
if step == 1 and "input_dir" in kwargs:
cmd.extend(["--input-dir", kwargs["input_dir"]])
diff --git a/voc_业务_2/提示词编辑稿.md b/voc_业务_2/提示词编辑稿.md
index 50e1e4a..6a95262 100644
--- a/voc_业务_2/提示词编辑稿.md
+++ b/voc_业务_2/提示词编辑稿.md
@@ -114,6 +114,10 @@ Persona 命名用简洁中文（≤6字），避免营销化夸张名称。输
```
## 任务：基于聚类数据按三维度发现用户画像（Persona），4-7 个。
+## 分析产品（product_detect / config 已识别）
+产品：{{product_name}} | 行业：{{industry}}
+purchase_motivation 用「雇佣{{product_name}}完成…」句式；Persona 须与该产品使用场景一致。
+
## 三维度强制覆盖（缺一不可）
A. 物理/生理/受众特征 → 必须绑定 stage=1_audience 的簇
典型信号：特定体质/肤质/年龄/体型/使用对象（如 sensitive / elderly / for kids / large breed）
@@ -162,6 +166,7 @@ global_positive: {{global_positive_json}}
```
**占位符说明：**
+- `{{product_name}}` / `{{industry}}` — 产品识别结果（自动填入）
- `{{catalog_json}}` — 聚类簇目录（自动填入）
- `{{audience_clusters_json}}` 等 — 聚类摘要（自动填入）
diff --git a/向量化.py b/向量化.py
index eec93d2..3113c86 100644
--- a/向量化.py
+++ b/向量化.py
@@ -1,5 +1,5 @@
"""
-从 voc_structured.sqlite 最新 job 展开 audience / pain_point / aspect_opinion
+从 voc_structured.sqlite 最新 job 展开 persona_signal / pain_point / aspect_opinion
（与聚类.py 一致，不向量化单独的 aspect、opinion），使用本地 MLX 写入 voc_embeddings.sqlite。
溯源：source_row 与结构化时一致（CSV 第 1 条数据行=1）；对应 merged_reviews_cleaned.csv
@@ -59,11 +59,56 @@ def _resolve_embed_workers(explicit: int | None = None) -> int:
logger.warning("本地 embedding 仅支持串行，--workers 已忽略（使用 1）")
return 1
ENTITY_TYPES = (
- "audience",
+ "persona_signal",
"pain_point",
"aspect_opinion",
)
+_GENERIC_PERSONA_SIGNALS = frozenset({
+ "self", "unknown", "user", "customer", "buyer", "myself", "the user", "consumer",
+})
+
+
+def _normalize_persona_signals(raw: Any) -> List[str]:
+ if raw is None:
+ items: List[Any] = []
+ elif isinstance(raw, list):
+ items = raw
+ elif isinstance(raw, str) and raw.strip():
+ items = [raw.strip()]
+ else:
+ items = []
+ out: List[str] = []
+ seen: set[str] = set()
+ for item in items:
+ text = str(item).strip()
+ if not text or text.lower() in _GENERIC_PERSONA_SIGNALS:
+ continue
+ key = text.lower()
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(text)
+ return out
+
+
+def _derive_persona_signals_fallback(
+ persona_signals: List[str],
+ pain_points: List[Any],
+) -> List[str]:
+ """无 persona_signals 时：用 pain_points 前 2 条作兜底。"""
+ if persona_signals:
+ return persona_signals
+ return [str(p).strip() for p in pain_points if str(p).strip()][:2]
+
+
+def _should_skip_extraction(
+ persona_signals: List[str],
+ pain_points: List[Any],
+ feedback: List[Any],
+) -> bool:
+ return not persona_signals and not pain_points and not feedback
+
@dataclass
class EmbedTask:
@@ -73,7 +118,7 @@ class EmbedTask:
entity_type: str
entity_index: int
embed_text: str
- audience: str
+ audience: str | None
aspect: str | None
opinion: str | None
category: str | None
@@ -151,11 +196,6 @@ def _init_embed_schema(conn: sqlite3.Connection) -> None:
)
-def _should_skip_extraction(audience: str, pain_points: List[Any], feedback: List[Any]) -> bool:
- aud = (audience or "").strip().lower()
- return aud == "unknown" and not pain_points and not feedback
-
-
def _build_tasks(
job_id: int,
rows: List[sqlite3.Row],
@@ -171,7 +211,7 @@ def _build_tasks(
logger.warning("跳过无效 JSON extraction_id=%s: %s", ext_id, e)
continue
- audience = str(data.get("audience", "unknown")).strip() or "unknown"
+ persona_signals = _normalize_persona_signals(data.get("persona_signals"))
pain_points = data.get("pain_points") or []
feedback = data.get("product_feedback") or []
if not isinstance(pain_points, list):
@@ -179,7 +219,7 @@ def _build_tasks(
if not isinstance(feedback, list):
feedback = []
- if _should_skip_extraction(audience, pain_points, feedback):
+ if _should_skip_extraction(persona_signals, pain_points, feedback):
continue
content = content_map.get(source_row, "")
@@ -189,17 +229,18 @@ def _build_tasks(
source_row,
)
- aud_norm = audience.strip()
- if aud_norm.lower() != "unknown":
+ signals = _derive_persona_signals_fallback(persona_signals, pain_points)
+
+ for i, sig in enumerate(signals):
tasks.append(
EmbedTask(
job_id=job_id,
extraction_id=ext_id,
source_row=source_row,
- entity_type="audience",
- entity_index=0,
- embed_text=aud_norm,
- audience=aud_norm,
+ entity_type="persona_signal",
+ entity_index=i,
+ embed_text=sig,
+ audience=None,
aspect=None,
opinion=None,
category=None,
@@ -220,7 +261,7 @@ def _build_tasks(
entity_type="pain_point",
entity_index=i,
embed_text=text,
- audience=aud_norm,
+ audience=None,
aspect=None,
opinion=None,
category=None,
@@ -247,7 +288,7 @@ def _build_tasks(
entity_type="aspect_opinion",
entity_index=i,
embed_text=merged,
- audience=aud_norm,
+ audience=None,
aspect=aspect,
opinion=opinion,
category=category,
diff --git a/结构化_server.py b/结构化_server.py
index 9a2b1c1..b2d6c09 100644
--- a/结构化_server.py
+++ b/结构化_server.py
@@ -588,19 +588,55 @@ def _normalize_product_feedback(items: Any) -> List[Dict[str, str]]:
return out
+_GENERIC_PERSONA_SIGNALS = frozenset({
+ "self", "unknown", "user", "customer", "buyer", "myself", "the user", "consumer",
+})
+
+
+def _normalize_persona_signals(raw: Any) -> List[str]:
+ if raw is None:
+ items: List[Any] = []
+ elif isinstance(raw, list):
+ items = raw
+ elif isinstance(raw, str) and raw.strip():
+ items = [raw.strip()]
+ else:
+ items = []
+ out: List[str] = []
+ seen: set[str] = set()
+ for item in items:
+ text = str(item).strip()
+ if not text or text.lower() in _GENERIC_PERSONA_SIGNALS:
+ continue
+ key = text.lower()
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(text)
+ return out
+
+
def _validate_extraction_strict(obj: Any, key_label: str = "") -> Dict[str, Any]:
"""校验结构化结果；失败时抛出带键名的 ValueError，供回传模型修正。"""
prefix = f"{key_label}: " if key_label else ""
if not isinstance(obj, dict):
raise ValueError(f"{prefix}必须是 JSON 对象")
- for field in ("audience", "pain_points", "product_feedback"):
+ for field in ("persona_signals", "pain_points", "product_feedback"):
if field not in obj:
raise ValueError(f"{prefix}缺少必填字段 {field}")
- audience = obj.get("audience")
- if not isinstance(audience, str) or not audience.strip():
- raise ValueError(f"{prefix}audience 必须为非空字符串")
+ persona_signals = obj.get("persona_signals")
+ if not isinstance(persona_signals, list):
+ raise ValueError(f"{prefix}persona_signals 必须为数组")
+ for i, sig in enumerate(persona_signals):
+ if not isinstance(sig, str) or not str(sig).strip():
+ raise ValueError(f"{prefix}persona_signals[{i}] 必须为非空字符串")
+ if str(sig).strip().lower() in _GENERIC_PERSONA_SIGNALS:
+ raise ValueError(
+ f"{prefix}persona_signals[{i}] 禁止为泛化词 {sig!r}，"
+ "须写具体身份/体质/场景短语"
+ )
pain_points = obj.get("pain_points")
if not isinstance(pain_points, list):
@@ -639,11 +675,7 @@ def _normalize_extraction(obj: Any) -> Dict[str, Any]:
if not isinstance(obj, dict):
raise ValueError("Each extraction must be a JSON object")
- audience = obj.get("audience", "unknown")
- if not isinstance(audience, str) or not str(audience).strip():
- audience = "unknown"
- else:
- audience = str(audience).strip()
+ persona_signals = _normalize_persona_signals(obj.get("persona_signals", []))
pain_points = obj.get("pain_points", [])
if pain_points is None:
@@ -656,14 +688,14 @@ def _normalize_extraction(obj: Any) -> Dict[str, Any]:
missing = [
f
- for f in ("audience", "pain_points", "product_feedback")
+ for f in ("persona_signals", "pain_points", "product_feedback")
if f not in obj
]
if missing:
logger.info("Normalized missing fields %s in extraction", missing)
return {
- "audience": audience,
+ "persona_signals": persona_signals,
"pain_points": pain_points,
"product_feedback": product_feedback,
}
@@ -677,7 +709,8 @@ def _build_batch_correction_message(keys: List[str], errors: List[str]) -> str:
f"校验错误：\n{err_block}\n\n"
"要求：\n"
f"- 输出一个 JSON 对象，顶层键必须且仅能是：{keys_literal}\n"
- "- 每个键的值必须包含 audience、pain_points、product_feedback\n"
+ "- 每个键的值必须包含 persona_signals、pain_points、product_feedback\n"
+ "- persona_signals 禁止仅写 self/user/unknown 等泛化词\n"
"- product_feedback 每条须含 aspect、opinion、sentiment"
"（仅 Positive/Negative/Neutral，禁止 Mixed/Ambiguous；褒贬交织选主倾向或拆条）、"
"category（禁止 Value，性价比用 Price）\n"
diff --git a/聚类.py b/聚类.py
index baf1e45..4679ca4 100644
--- a/聚类.py
+++ b/聚类.py
@@ -3,8 +3,8 @@
流程：
1. audience（LLM 自动调参 n_neighbors）
- 2a. 前两 audience 簇各自独立：簇内 pain_point（LLM 自动调参）
- 2b. 前两 audience 簇各自独立：簇内 aspect_opinion 按 Positive/Negative/Neutral 分桶聚类
+ 2a. 占比达阈值的 audience 簇各自独立：簇内 pain_point（LLM 自动调参）
+ 2b. 占比达阈值的 audience 簇各自独立：簇内 aspect_opinion 按 Positive/Negative/Neutral 分桶聚类
3a. 全量 pain_point（LLM 自动调参）
3b. 全量 aspect_opinion 按 Positive/Negative/Neutral 分桶聚类（各自 LLM 自动调参）
@@ -110,7 +110,10 @@ SAMPLE_RATIO = 0.6
STEP2_N_NEIGHBORS = 8
# 簇内去重评论数 / 本步骤参与聚类的去重评论总数 < 该比例则不写入库、不进入报告
-CLUSTER_MIN_REVIEW_RATIO = 0.10
+CLUSTER_MIN_REVIEW_RATIO = 0.05
+# step 2 入选 audience 簇：去重评论数 / 全量清洗评论总数 ≥ 该比例
+DEFAULT_AUDIENCE_COVERAGE_THRESHOLD = 0.01
+VOC_BUSINESS_CONFIG = PROJECT_ROOT / "voc_业务_2" / "config.yaml"
@dataclass
@@ -1093,10 +1096,24 @@ def _save_tuning_logs(
)
-def _top2_audience_clusters(
+def _config_threshold(key: str, default: float) -> float:
+ try:
+ import yaml
+ if VOC_BUSINESS_CONFIG.is_file():
+ with VOC_BUSINESS_CONFIG.open(encoding="utf-8") as f:
+ cfg = yaml.safe_load(f) or {}
+ val = float(cfg.get(key, default))
+ if 0 < val <= 1:
+ return val
+ except (TypeError, ValueError, OSError):
+ pass
+ return default
+
+
+def _audience_cluster_ranking(
conn: sqlite3.Connection, run_id: int
-) -> Tuple[List[int], Dict[int, int]]:
- """返回 (前两簇标签列表, source_row -> audience簇标签)。"""
+) -> Tuple[Dict[int, set[int]], Dict[int, int]]:
+ """返回 (cluster_label -> source_rows, source_row -> audience簇标签)。"""
cur = conn.execute(
"""
SELECT cluster_label, source_row
@@ -1113,11 +1130,46 @@ def _top2_audience_clusters(
sr = int(sr)
row_to_cluster[sr] = lab
cluster_rows.setdefault(lab, set()).add(sr)
- ranked = sorted(
- cluster_rows.items(), key=lambda x: len(x[1]), reverse=True
- )
- top2 = [lab for lab, _ in ranked[:2]]
- return top2, row_to_cluster
+ return cluster_rows, row_to_cluster
+
+
+def _top2_audience_clusters(
+ conn: sqlite3.Connection, run_id: int
+) -> Tuple[List[int], Dict[int, int]]:
+ """兼容旧逻辑：返回前两大的 audience 簇。"""
+ cluster_rows, row_to_cluster = _audience_cluster_ranking(conn, run_id)
+ ranked = sorted(cluster_rows.items(), key=lambda x: len(x[1]), reverse=True)
+ return [lab for lab, _ in ranked[:2]], row_to_cluster
+
+
+def _eligible_audience_clusters(
+ conn: sqlite3.Connection,
+ run_id: int,
+ total_reviews: int,
+ threshold: float = DEFAULT_AUDIENCE_COVERAGE_THRESHOLD,
+) -> Tuple[List[int], Dict[int, int]]:
+ """返回 (占比达阈值的 audience 簇列表, source_row -> audience簇标签)。"""
+ cluster_rows, row_to_cluster = _audience_cluster_ranking(conn, run_id)
+ if total_reviews <= 0:
+ return [], row_to_cluster
+ ranked = sorted(cluster_rows.items(), key=lambda x: len(x[1]), reverse=True)
+ eligible = [
+ lab for lab, rows in ranked
+ if len(rows) / total_reviews >= threshold
+ ]
+ if len(eligible) > 4:
+ logger.warning(
+ "eligible audience 簇数量=%s（阈值=%.0f%%），可能过度拆分 step 2",
+ len(eligible),
+ threshold * 100,
+ )
+ if not eligible and ranked:
+ logger.warning(
+ "无 audience 簇达到 step2 阈值 %.0f%%（总评论 %s），跳过 step 2",
+ threshold * 100,
+ total_reviews,
+ )
+ return eligible, row_to_cluster
def run_clustering(
@@ -1128,12 +1180,14 @@ def run_clustering(
cluster_db: Path = CLUSTER_DB,
reset_db: bool = True,
) -> dict:
+ """仅运行 3a（全量 pain）和 3b（全量 aspect_opinion × 情感）聚类。
+ 已移除步骤1 audience 和步骤2 per-audience 聚类。
+ """
require_chat_api_key()
econn = sqlite3.connect(embed_db)
try:
jid = _resolve_job_id(job_id, econn, structured_db)
- audience_rows = _load_embed_rows(econn, jid, "audience")
pain_rows = _load_embed_rows(econn, jid, "pain_point")
ao_rows = _load_embed_rows(econn, jid, "aspect_opinion")
finally:
@@ -1153,50 +1207,6 @@ def run_clustering(
total_reviews = _count_csv_reviews(csv_path)
logger.info("清洗后评论总数: %s", total_reviews)
- # --- 1 audience ---
- labels1, meta1 = _cluster_stage(
- audience_rows, stage="1_audience", use_llm_tune=True, client=client
- )
- filt1 = _save_assignments(
- cconn,
- run_id,
- "1_audience",
- audience_rows,
- labels1,
- n_neighbors=meta1.get("n_neighbors"),
- outlier_participates=False,
- )
- meta1["cluster_filter"] = filt1
- _save_stage_meta(cconn, run_id, "1_audience", meta1)
- if meta1.get("tuning_log"):
- _save_tuning_logs(cconn, run_id, "1_audience", meta1["tuning_log"])
-
- top2, row_to_aud = _top2_audience_clusters(cconn, run_id)
- meta_top2 = {
- "top2_audience_clusters": top2,
- "per_cluster_source_rows": {
- str(lab): len(_source_rows_for_audience_cluster(row_to_aud, lab))
- for lab in top2
- },
- "step2_mode": "per_audience_cluster_llm_auto_tune",
- }
- _save_stage_meta(cconn, run_id, "step2_filter", meta_top2)
- logger.info("Audience 前两簇（将分别聚类）: %s", top2)
-
- step2_counts: Dict[str, int] = {}
- for aud_lab in top2:
- step2_counts.update(
- _cluster_step2_for_audience(
- cconn,
- run_id,
- audience_cluster=aud_lab,
- pain_rows=pain_rows,
- ao_rows=ao_rows,
- row_to_aud=row_to_aud,
- client=client,
- )
- )
-
# --- 3a pain global ---
labels3a, meta3a = _cluster_stage(
pain_rows, stage="3a_pain_global", use_llm_tune=True, client=client
@@ -1234,10 +1244,7 @@ def run_clustering(
"total_reviews": total_reviews,
"min_cluster_review_ratio": CLUSTER_MIN_REVIEW_RATIO,
"cluster_db": str(cluster_db.resolve()),
- "top2_audience_clusters": top2,
"counts": {
- "1_audience": len(audience_rows),
- **step2_counts,
"3a_pain_global": len(pain_rows),
**step3b_counts,
},