From c91e7f8c2061a992a54410ef51d993a9f7748dde Mon Sep 17 00:00:00 2001 From: OnesvmWhoops Date: Wed, 17 Jun 2026 09:11:04 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BE=89=E5=93=A5=E7=89=88=E6=9C=AC=EF=BC=9A?= =?UTF-8?q?=E7=BB=93=E6=9E=84=E5=8C=96=E8=81=9A=E7=B1=BB=E6=BA=AF=E6=BA=90?= =?UTF-8?q?=E5=BD=92=E5=9B=A0=E4=B8=8E=E4=B8=9A=E5=8A=A1=E6=8A=A5=E5=91=8A?= =?UTF-8?q?=E5=A2=9E=E5=BC=BA=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 移除结构化 audience 字段，强化 voc_业务_2 源评论归因匹配与 Persona 引用展示，更新 README 与流水线默认清理 SQLite。 Co-authored-by: Cursor --- README.md | 433 +++++---- main_voc分析.py | 2 + prompts/README.md | 2 +- prompts/extraction/batch_output_format.md | 2 +- prompts/extraction/examples.yaml | 18 +- prompts/extraction/field_rules.md | 13 +- prompts/extraction/filter_irrelevant.md | 4 +- .../extraction/format_constraints_batch.md | 2 +- .../extraction/format_constraints_single.md | 2 +- prompts/schema.yaml | 2 +- voc_report.py | 2 +- voc_业务_2/build_report.py | 110 ++- voc_业务_2/config.yaml | 9 + voc_业务_2/data_loader.py | 187 +++- voc_业务_2/echarts_builder.py | 35 +- voc_业务_2/llm_analyzer.py | 65 +- voc_业务_2/prompts.yaml | 67 +- voc_业务_2/report_utils.py | 871 +++++++++++++++--- voc_业务_2/run_pipeline.py | 5 +- voc_业务_2/提示词编辑稿.md | 5 + 向量化.py | 77 +- 结构化_server.py | 57 +- 聚类.py | 125 +-- 23 files changed, 1493 insertions(+), 602 deletions(-) diff --git a/README.md b/README.md index 3e5ff93..a52b80b 100644 --- a/README.md +++ b/README.md @@ -1,257 +1,240 @@ -# VOC LLM 结构化分析 (VOC_LLM结构化) +# 亚马逊评论 VOC 结构化分析 -> 基于大语言模型（[DeepSeek](https://api.deepseek.com) OpenAI 兼容 API）与本地 MLX 向量的亚马逊 VOC（Voice of Customer）评论分析流水线：合并 CSV → 清洗 → LLM 结构化 → 向量化 → 聚类与词频 → 生成 HTML 分析报告。 +> **分支「辉哥版本」**：可结构化、可聚类、可溯源归因的亚马逊站内评论分析流水线。 +> 基于 DeepSeek Chat API + 本地 MLX 向量，完成：合并 CSV → 清洗 → LLM 结构化 → 向量化 → 聚类 → 词频 → **业务 HTML 报告**（Persona / 差评主题 / 根因 / KANO，每条洞察可回溯到真实评论）。 -## 📖 目录 +**仓库**：https://git.onesvm.com/whoops/amz_review_analyse -- [核心特性](#-核心特性) -- [环境要求](#-环境要求) -- [安装指南](#-安装指南) -- [使用说明](#-使用说明) -- [示例与输出](#-示例与输出) -- [项目结构](#-项目结构) -- [常见问题](#-常见问题) -- [参与贡献](#-参与贡献) -- [开源协议](#-开源协议) -- [联系方式与鸣谢](#-联系方式与鸣谢) +--- -## ✨ 核心特性 +## 目录 -- **七步全流程编排** — `main_voc分析.py` 一键串联：合并、清洗、结构化、向量化、聚类、词频、HTML 报告 -- **断点续跑** — 支持 `--from-step` / `--only-step`，从任意步骤恢复，调试时节省 API 成本 -- **LLM 结构化提取** — 从评论中抽取受众、痛点、方面、观点、情感等字段（`prompts/schema.yaml` 可配置） -- **本地向量化** — Apple Silicon 上运行 `Qwen3-Embedding-4B-mxfp8`（MLX），无需云端 Embedding API -- **语义聚类** — UMAP + HDBSCAN 多阶段聚类，辅以 LLM 评估簇质量自动调参 -- **词频分析** — LLM 归纳专有名词 + spaCy 全量词频统计，报告内嵌词云与六类归类 -- **可编辑 Prompt** — `prompts/` 目录下 Markdown / YAML 热加载，产品运营可直接改话术（见 `prompts/README.md`） -- **并行加速** — 聚类与词频在步骤 5–6 由线程池并行执行；结构化批间并行（默认 8 路） +- [核心能力](#核心能力) +- [快速开始（推荐）](#快速开始推荐) +- [主流程说明](#主流程说明) +- [结构化字段](#结构化字段) +- [溯源与归因](#溯源与归因) +- [环境要求](#环境要求) +- [安装](#安装) +- [项目结构](#项目结构) +- [常见问题](#常见问题) -## 🛠 环境要求 +--- + +## 核心能力 + +| 能力 | 说明 | +|------|------| +| **LLM 结构化提取** | 从评论抽取 `persona_signals`（画像信号）、`pain_points`（需求痛点）、`product_feedback`（方面/观点/情感/类别） | +| **本地向量化** | Apple Silicon 上运行 `Qwen3-Embedding-4B-mxfp8`（MLX），无需云端 Embedding | +| **语义聚类** | UMAP + HDBSCAN：`3a` 全量痛点、`3b` 按情感分桶的 aspect-opinion 聚类 | +| **Persona 发现** | 绑定聚类簇 + keywords 二次过滤，统计命中数与占比 | +| **差评主题 & 根因** | LLM 归纳主题；根因引用由系统从真实评论回填（结构化字段优先匹配） | +| **可溯源 HTML 报告** | Persona 卡片、根因区块展示源评论原文 + ASIN 链接；附录展示结构化/聚类抽样 | +| **断点续跑** | `main_voc分析.py --from-step` / `run_pipeline.py --from-step` | + +--- + +## 快速开始（推荐） + +业务报告入口在 `voc_业务_2/`，一键跑数据管道并生成 HTML： + +```bash +cd voc_业务_2 + +# 1. 编辑 config.yaml：input_dir 指向原始评论 CSV 目录 +# 2. 全流程（step 1–6 + 自动 build_report） +../310py/bin/python run_pipeline.py --input-dir "../你的评论CSV目录" +``` + +产物示例： + +- 根目录 SQLite：`voc_structured.sqlite`、`voc_embeddings.sqlite`、`voc_clustering.sqlite` +- HTML 报告：`voc_业务_2/output/{产品slug}-voc-report.html` + +仅重跑报告（数据库已就绪）： + +```bash +../310py/bin/python build_report.py --product "产品名" --industry "行业" +``` + +--- + +## 主流程说明 + +### 方式 A：`voc_业务_2/run_pipeline.py`（业务报告） + +``` +原始 CSV → main_voc分析 step 1–6 → build_report.py → HTML +``` + +- 每步默认**清理旧 SQLite**（不加 `--keep-db`），避免与历史 job 混用 +- 产品名/行业可在 `config.yaml` 留空，由 LLM 从评论样本自动识别 + +### 方式 B：`main_voc分析.py`（含经典 voc_report） + +```bash +./310py/bin/python main_voc分析.py \ + --input-dir "reviews_export" \ + --product "Bikini Trimmer" \ + --industry "个人护理" +``` + +七步：合并 → 清洗 → 结构化 → 向量化 → 聚类 → 词频 → HTML（`voc_report.py`）。 + +断点示例： + +```bash +./310py/bin/python main_voc分析.py --from-step 4 # 从向量化续跑 +./310py/bin/python main_voc分析.py --only-step 7 # 仅重生成 voc_report +``` + +更细步骤见 **[main_voc分析.md](main_voc分析.md)**、**[VOC分析方法论与报告生成逻辑.md](VOC分析方法论与报告生成逻辑.md)**。 + +--- + +## 结构化字段 + +当前 schema（`prompts/schema.yaml`）根字段为 **3 项**（已移除 `audience`）： + +```json +{ + "persona_signals": ["sensitive skin", "travel grooming"], + "pain_points": ["ingrown hair"], + "product_feedback": [ + { + "aspect": "battery life", + "opinion": "dies after one use", + "sentiment": "Negative", + "category": "Function" + } + ] +} +``` + +- 所有字段值须为**自然英文**（多语言评论先理解再英文输出） +- `product_feedback.category` 优先 8 类标准类别（Trust / Ingredient / Quality / Function / Appearance / Logistics / Customer Service / Price） + +Prompt 编辑入口：`prompts/extraction/`、`voc_业务_2/prompts.yaml`。 + +--- + +## 溯源与归因 + +整条链路通过 **`source_row`**（与 `merged_reviews_cleaned.csv` 行号一致）关联： + +``` +评论原文 (CSV) + ↓ source_row +comment_extractions (voc_structured.sqlite) + ↓ extraction_id / source_row +embedding_items (voc_embeddings.sqlite) + ↓ +cluster_assignments (voc_clustering.sqlite) → Persona 绑定簇 → source_rows 命中池 + ↓ +build_report.py HTML + ├── Persona 卡片：命中池内结构化/原文匹配，展示 1–3 条源评论 + ├── 根因分析：Negative product_feedback 优先匹配，展示 1–4 条 + └── 附录（voc_report）：结构化 JSON + 聚类短语抽样 +``` + +配置项（`voc_业务_2/config.yaml`）： + +```yaml +persona_quote_max: 3 # Persona 卡片最多展示条数 +rootcause_quote_max: 4 # 每条根因最多展示条数 +``` + +--- + +## 环境要求 | 依赖 | 说明 | |------|------| -| Python | >= 3.10（推荐 3.12，项目内 `310py`） | -| pip / uv | 安装 `requirements.txt` 中的包 | -| spaCy 英文模型 | 经 `uv pip` 安装 `en-core-web-sm`（见安装指南，词频步骤必需） | -| DeepSeek API Key | 结构化、聚类评估、词频、报告等 Chat 步骤 | -| 本地 Embedding 模型 | 目录 `Qwen3-Embedding-4B-mxfp8/`（约 4GB，已 gitignore，需自行下载） | -| Apple Silicon | 本地向量化依赖 MLX（M 系列芯片） | +| Python | ≥ 3.10（推荐 3.12，项目内 `310py`） | +| DeepSeek API Key | 结构化、Persona/主题/根因/KANO 等 Chat 步骤 | +| 本地 Embedding 模型 | `Qwen3-Embedding-4B-mxfp8/`（约 4GB，gitignore，需自行下载） | +| Apple Silicon | 本地 MLX 向量化 | +| spaCy `en_core_web_sm` | 词频步骤 | -**Chat API Key**（任选其一，勿提交到 Git）： +**API Key**（任选其一，勿提交 Git）： -1. 环境变量 `DEEPSEEK_API_KEY` -2. 环境变量 `DEEPSEEK_API_KEY_FILE` 指向单行密钥文件 -3. 项目根目录 `.deepseek_key`（单行，无引号） +```bash +export DEEPSEEK_API_KEY="sk-xxx" +# 或项目根 .deepseek_key（已被 .gitignore） +``` -可选：`DEEPSEEK_MODEL`（默认 `deepseek-v4-pro`）、`DEEPSEEK_BASE_URL`（默认 `https://api.deepseek.com`）。 +--- -## 📦 安装指南 - -1. 克隆项目到本地： +## 安装 ```bash git clone https://git.onesvm.com/whoops/amz_review_analyse.git -cd amz_review_analyse # 或你的本地目录名 -``` +cd amz_review_analyse -2. 创建虚拟环境并安装依赖（推荐）： - -```bash uv venv 310py --python 3.12 uv pip install --python 310py/bin/python -r requirements.txt uv pip install --python 310py/bin/python \ "en-core-web-sm @ https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.8.0/en_core_web_sm-3.8.0-py3-none-any.whl" ``` -3. 配置 DeepSeek API Key： - -```bash -export DEEPSEEK_API_KEY="sk-xxx" -# 或在项目根创建 .deepseek_key（已被 .gitignore 忽略） -``` - -4. 准备本地 Embedding 模型（首次向量化前）： - -将 `Qwen3-Embedding-4B-mxfp8` 放到项目根，或设置 `VOC_EMBED_MODEL_PATH` 指向模型目录。可从 [Hugging Face](https://huggingface.co/mlx-community/Qwen3-Embedding-4B-mxfp8) 下载。 - -5. 准备原始评论 CSV 目录（目录内所有 `*.csv` 表头须一致），例如亚马逊导出的 `*_realtime.csv`。 - -## 🚀 使用说明 - -### 全流程分析 - -```bash -./310py/bin/python main_voc分析.py \ - --input-dir "reviews_export" \ - --product "cat deterrent indoor" \ - --industry "Pet Supplies" -``` - -- `--input-dir`：原始 CSV 目录 -- `--product`：产品名（写入结构化任务与报告路径） -- `--industry`：行业名，默认 `-`（可在步骤 3 写入库） -- `--keep-db`：保留已有结构化 `voc_*.sqlite`，不覆盖删除 - -### 加速（批间并行，默认已开启） - -步骤 3 结构化默认多批并行 Chat 请求；步骤 4 向量为本地 MLX 串行批处理（勿对同一模型多线程）： - -```bash -# 全流程 -./310py/bin/python main_voc分析.py --input-dir "reviews_export" --product "产品名" - -# 调低结构化并发（遇 429 时） -./310py/bin/python main_voc分析.py --input-dir "reviews_export" --product "产品名" \ - --struct-workers 4 - -# 环境变量：export VOC_STRUCT_WORKERS=8 VOC_EMBED_BATCH_SIZE=16 -``` - -### 断点续跑 - -```bash -# 从向量化起续跑（步骤 4 起可省略 --product，自动读结构化库） -./310py/bin/python main_voc分析.py --from-step 4 --keep-db - -# 仅重跑词频（复用已有 voc_terms.json） -./310py/bin/python main_voc分析.py --from-step 6 --skip-wordfreq-llm - -# 仅重新生成 HTML 报告 -./310py/bin/python main_voc分析.py --only-step 7 -``` - -### 其他常用参数 - -| 参数 | 说明 | -|------|------| -| `--clean-intermediates` | 报告成功后删除中间 csv/sqlite，减少磁盘占用 | -| `--filter-small-clusters` | 报告仅保留簇内评论占比 ≥ 10% 的簇 | -| `--save-llm-raw` | 将报告 LLM 原文保存为 `report_llm_raw.txt`，调试时使用 | - -### 程序式调用 - -```python -from pathlib import Path -from main_voc分析 import run_voc_analysis - -result = run_voc_analysis( - input_dir=Path("reviews_export"), - industry="Pet Supplies", - product_name="cat deterrent indoor", - from_step=1, - clean_databases=True, -) -print(result["report_html"]) -``` - -### Prompt 验收（无需 API Key） - -```bash -./310py/bin/python prompts/smoke.py # 检查 prompt 能否加载 -./310py/bin/python prompts/smoke.py --live # 联调模型（需 DEEPSEEK_API_KEY） -``` - -### 可选变体：jieba 词频 - -中文或需 jieba 分词时，可使用 `main_voc分析_jieba.py`（词频走 `词频_jieba.py`，其余步骤与主流程一致）。 +Embedding 模型：从 [Hugging Face mlx-community/Qwen3-Embedding-4B-mxfp8](https://huggingface.co/mlx-community/Qwen3-Embedding-4B-mxfp8) 下载到项目根，或设置 `VOC_EMBED_MODEL_PATH`。 --- -更详细的步骤说明、算法与 SQLite 约定见 **[main_voc分析.md](main_voc分析.md)**。 - -## 📸 示例与输出 - -流程结束后，主要产物如下： - -| 路径 | 说明 | -|------|------| -| `merged_reviews.csv` | 多文件合并结果 | -| `merged_reviews_cleaned.csv` | 清洗、去重后的评论 | -| `voc_structured.sqlite` | LLM 结构化结果 | -| `voc_embeddings.sqlite` | 本地 Qwen3 向量（维度见库内 `dimensions` 字段） | -| `voc_clustering.sqlite` | 多阶段聚类标签 | -| `output/voc_terms.json` | 专有名词 / 停用词 | -| `output/word_freq.csv` | 全量词频表 | -| `output/{product}/{product}_voc_report.html` | **最终 VOC 分析报告**（词云、词频、分簇、AI 正文） | - -stdout 会打印 JSON 摘要（含 `report_html` 等键）。 - -## 📂 项目结构 +## 项目结构 ```text -VOC_LLM结构化/ -├── main_voc分析.py # 主流程编排入口（七步） -├── main_voc分析_jieba.py # 词频使用 jieba 的变体入口 -├── main_voc分析.md # 流程与算法详细说明 -├── voc_llm.py # DeepSeek Chat 密钥与客户端 -├── local_embedding.py # 本地 MLX Qwen3 向量化 -├── 合并评论数据.py # 步骤 1：多 CSV 合并 -├── content清洗.py # 步骤 2：评论清洗与去重 -├── 结构化_server.py # 步骤 3：LLM 结构化入库 -├── 结构化_Prompt.py # 结构化 prompt 组装 -├── 向量化.py # 步骤 4：本地 Embedding 入库 -├── 聚类.py # 步骤 5：UMAP + HDBSCAN -├── 词频.py / 词频_jieba.py # 步骤 6：术语提取 + 词频 -├── voc_report.py # 步骤 7：HTML 报告生成 -├── prompts/ # 可编辑 prompt、schema、配置 -├── Qwen3-Embedding-4B-mxfp8/ # 本地模型（gitignore，需自行放置） -├── requirements.txt -├── output/ # 报告与词频输出（gitignore） -└── README.md # 本文件 +├── main_voc分析.py # 七步主流程编排 +├── 结构化_server.py # LLM 结构化 → voc_structured.sqlite +├── 向量化.py # persona_signal / pain / aspect_opinion 向量 +├── 聚类.py # 3a 痛点 + 3b 情感分桶聚类 +├── voc_report.py # 经典 HTML 报告（Dashboard + 附录验证） +├── prompts/ # 结构化 & 报告 Prompt（可热加载） +├── voc_业务_2/ +│ ├── run_pipeline.py # ★ 业务一键流水线 +│ ├── build_report.py # ★ 业务 HTML 报告生成 +│ ├── llm_analyzer.py # Persona / 主题 / KANO / 根因 LLM +│ ├── report_utils.py # 统计、引用匹配、HTML 拼装 +│ ├── data_loader.py # SQLite / CSV 加载 +│ ├── config.yaml # 产品路径、并发、引用条数等 +│ └── template.html # 报告模板 +└── output/ # 报告与词频（gitignore） ``` -## ❓ 常见问题 - -**Q：提示缺少 `DEEPSEEK_API_KEY`？** -A：按上文配置环境变量或 `.deepseek_key`，并确认密钥未提交到仓库。 - -**Q：只有 `.dashscope_key` 报错？** -A：Chat 已切换为 DeepSeek，DashScope 密钥不能用于 `api.deepseek.com`，请改用 `.deepseek_key`。 - -**Q：步骤 4 向量化失败 / 找不到模型？** -A：确认 `Qwen3-Embedding-4B-mxfp8/` 在项目根，或设置 `VOC_EMBED_MODEL_PATH`；需在 Apple Silicon + Python 3.10+ 环境。 - -**Q：步骤 4 报错找不到 `product`？** -A：从步骤 1–3 完整跑过，或确保 `voc_structured.sqlite` 中已有最新 job；步骤 4 起可省略 `--product`。 - -**Q：词频步骤报 spaCy 模型缺失？** -A：执行 `uv pip install --python 310py/bin/python "en-core-web-sm @ https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.8.0/en_core_web_sm-3.8.0-py3-none-any.whl"`。 - -**Q：合并 CSV 失败？** -A：确保 `--input-dir` 下所有 CSV 表头完全一致。 - -**Q：修改 LLM 话术后报告解析失败？** -A：勿修改 `prompts/schema.yaml` 中 `report.markers` 四段标记名；改完运行 `./310py/bin/python prompts/smoke.py` 验收。 - -## 🤝 参与贡献 - -欢迎提交 Issue 与 Pull Request。建议流程： - -1. Fork 本仓库 -2. 创建特性分支：`git checkout -b feature/your-feature` -3. 提交更改：`git commit -m '简要说明变更'` -4. 推送并发起 Pull Request - -修改主流程或 CLI 时，请同步更新 `main_voc分析.md`；修改 `prompts/` 时请遵循 `prompts/README.md` 中的占位符与 schema 约定。 - -**安全提醒**：勿提交 `.deepseek_key`、`.dashscope_key`、`.env`、真实评论 CSV、`*.sqlite`、`output/` 及本地模型目录（见 `.gitignore`）。 - -## 📄 开源协议 - -本项目尚未在仓库中附带 `LICENSE` 文件。若为内部项目，请按组织规范使用；若计划开源，请补充协议文件（如 MIT）并更新本节链接。 - -## ✉️ 联系方式与鸣谢 - -- **详细技术文档**：[main_voc分析.md](main_voc分析.md)、[prompts/README.md](prompts/README.md) -- **项目仓库**：https://git.onesvm.com/whoops/amz_review_analyse - -### 鸣谢 - -- [DeepSeek API](https://api.deepseek.com) — Chat 结构化、聚类评估、词频与报告 -- [mlx-community/Qwen3-Embedding-4B-mxfp8](https://huggingface.co/mlx-community/Qwen3-Embedding-4B-mxfp8) — 本地向量化 -- [UMAP](https://umap-learn.readthedocs.io/)、[HDBSCAN](https://hdbscan.readthedocs.io/) — 聚类管线 -- [spaCy](https://spacy.io/) — 英文词频与 NLP - --- -*README 与 `main_voc分析.py` 七步流程保持一致；深度说明请参阅 `main_voc分析.md`。* +## 常见问题 + +**Q：Persona 显示「命中 N 条」但没有源评论？** +A：确保已用最新 `report_utils.pick_persona_quotes` 重跑 `build_report.py`；命中池有数据时会多层回退展示，优先结构化 `persona_signals` / `pain_points` 匹配。 + +**Q：结构化校验报 persona_signals 错误？** +A：「辉哥版本」已移除 `audience` 字段；请重跑 step 3 生成新格式 JSON，勿混用旧库。 + +**Q：提示缺少 API Key？** +A：配置 `DEEPSEEK_API_KEY` 或 `.deepseek_key`（Chat 已用 DeepSeek，DashScope 密钥不可用）。 + +**Q：向量化找不到模型？** +A：确认 `Qwen3-Embedding-4B-mxfp8/` 在项目根或设置 `VOC_EMBED_MODEL_PATH`。 + +--- + +## 分支说明 + +| 分支 | 说明 | +|------|------| +| `main` | 基础流水线 + DeepSeek + 本地 Embedding | +| **`辉哥版本`** | 业务报告（voc_业务_2）、结构化三字段、聚类溯源、Persona/根因源评论归因 | + +--- + +## 鸣谢 + +- [DeepSeek API](https://api.deepseek.com) — Chat 结构化与分析 +- [mlx-community/Qwen3-Embedding-4B-mxfp8](https://huggingface.co/mlx-community/Qwen3-Embedding-4B-mxfp8) — 本地向量化 +- [UMAP](https://umap-learn.readthedocs.io/)、[HDBSCAN](https://hdbscan.readthedocs.io/) — 聚类 + +--- + +*详细算法与 SQLite 表结构见 [main_voc分析.md](main_voc分析.md)。* diff --git a/main_voc分析.py b/main_voc分析.py index 34f9918..6229e8b 100644 --- a/main_voc分析.py +++ b/main_voc分析.py @@ -60,6 +60,8 @@ DEFAULT_INDUSTRY = "亚马逊电商" DEFAULT_PRODUCT = "亚马逊商品" + + def _safe_product_dir_name(product_name: str) -> str: """用于 output 子目录名；去除路径非法字符。""" name = (product_name or "product").strip() diff --git a/prompts/README.md b/prompts/README.md index b5676ea..81045b3 100644 --- a/prompts/README.md +++ b/prompts/README.md @@ -16,7 +16,7 @@ ## 占位符 `.md` 文件使用 Python `format` 语法，例如 `{product_name}`、`{industry}`。 -正文里需要字面量花括号时写双花括号：`{{"audience": "unknown"}}`。 +正文里需要字面量花括号时写双花括号：`{{"persona_signals": []}}`。 ## 改完怎么验收 diff --git a/prompts/extraction/batch_output_format.md b/prompts/extraction/batch_output_format.md index 98d5aab..fbecee2 100644 --- a/prompts/extraction/batch_output_format.md +++ b/prompts/extraction/batch_output_format.md @@ -2,6 +2,6 @@ 本批共 {n_keys} 条评论，用户消息中每条评论以 [C1]、[C2]… 前缀标识。 - 只输出一个 JSON 对象；顶层键必须且仅能是：{keys_literal} - 每个顶层键对应一条同前缀评论，不得遗漏、不得新增其他顶层键 -- 每个键的值是单条结构化对象，仅含 audience、pain_points、product_feedback 三个字段 +- 每个键的值是单条结构化对象，仅含 persona_signals、pain_points、product_feedback 三个字段 - 不要用 JSON 数组作为顶层；不要把多条评论合并进一个对象；不要用 results、data 等包裹层 - 不要输出 markdown 代码围栏或任何解释文字 diff --git a/prompts/extraction/examples.yaml b/prompts/extraction/examples.yaml index 09edce8..2797d0c 100644 --- a/prompts/extraction/examples.yaml +++ b/prompts/extraction/examples.yaml @@ -2,7 +2,9 @@ - instruction: "示例 1 教学：展示正常长评论如何标准提取，如何准确分类产品优点、缺陷以及物流问题。" review: "Bought this for my 12yo lab who struggles with stairs. It is very soft and helps her get onto the bed easily. But the zipper broke after a week and the shipping box was damaged." output: - audience: "12yo lab" + persona_signals: + - "large breed dog" + - "mobility limited pet" pain_points: - "struggles with stairs" - "difficulty getting onto bed" @@ -23,7 +25,8 @@ - instruction: "示例 2 教学：防混淆与隐性属性提炼。什么是「产品缺陷（如讲太快）」，它属于 product_feedback，绝不是 pain_points（用户需求）！并提炼出 teaching speed 这个隐性反馈对象。" review: "The course content is good, the teacher speaks too fast, can't keep up." output: - audience: "student" + persona_signals: + - "online course learner" pain_points: [] product_feedback: - aspect: "course content" @@ -38,7 +41,9 @@ - instruction: "示例 3 教学：深度上下文推理。如何通过动作（抓跳蚤）推理出使用者（pet），如何通过「全家能睡觉」推理出用户需求（失眠），以及如何将「Worth every penny」准确归类为对 Price（价格）的产品反馈。" review: "Finally found something that stops the midnight flea scratching! The chemical smell is a bit strong initially, but it fades. Worth every penny since our family can finally sleep." output: - audience: "pet" + persona_signals: + - "flea problem pet" + - "household sleep disruption" pain_points: - "midnight flea scratching" - "sleep deprivation" @@ -55,7 +60,8 @@ - instruction: "示例 4 教学：1) 非英语评论（例如西班牙语）输出必须全英文，禁止保留原文非英文片段，先理解语义再用英文短语表达；2) 「随附配件」不属于 8 个标准类别，演示如何补充一个简洁的英文新类别 'Accessories'。" review: "El producto en sí está muy bien, viene con 2 cristales templados y 2 grips para los joycon. Pero es lo que he leido en algunos comentarios, la funda huele como a tabaco jajaja es raro pero es así" output: - audience: "self" + persona_signals: + - "joycon grip user" pain_points: [] product_feedback: - aspect: "product" @@ -74,7 +80,9 @@ - instruction: "示例 5 教学：保留关键成分与症状等核心具体词汇。绝不能将带有具体成分/病症的词泛化提取。例如遇到「chicken flavor」（鸡肉风味）或「joint pain」（关节疼痛）时，必须保留核心修饰词，绝不能错误缩减提取为「flavor」或「pain」。" review: "My elderly cat suffers from severe joint pain. But he is very attracted by the chicken flavor of this supplement! It really helps him walk better." output: - audience: "elderly cat" + persona_signals: + - "senior cat" + - "joint pain cat" pain_points: - "severe joint pain" product_feedback: diff --git a/prompts/extraction/field_rules.md b/prompts/extraction/field_rules.md index f1471ae..5e29df8 100644 --- a/prompts/extraction/field_rules.md +++ b/prompts/extraction/field_rules.md @@ -1,12 +1,19 @@ ## 语言规则（红线）: - 评论输入可能是英语、西班牙语、法语、德语、日语等**任意语言**。 -- **所有输出字段值**（audience、pain_points 每一条、aspect、opinion）**必须是自然英文**。 +- **所有输出字段值**（persona_signals 每一条、pain_points 每一条、aspect、opinion）**必须是自然英文**。 - **禁止**保留原文非英文片段（如 está muy bien、huele como a tabaco、très bon）；先理解语义，再用英文短语表达。 ## 分析要求: -1. audience (为谁购买): - - 提取出实际的使用者，使用简短的英文名词。若无明确提及，请根据上下文推理；若完全无法推理则输出 'unknown'。 +1. persona_signals (用户画像信号): + - **字符串数组**；每条为 2–8 词的英文短语，描述身份、体质、场景或使用背景。 + - 从评论原文推断，典型信号： + - 体质/生理：`sensitive skin`、`coarse thick hair`、`pregnant`、`elderly` + - 场景/行为：`travel grooming`、`shower use`、`bikini area`、`first-time buyer` + - 自我标注句式：`I have …` / `As a …` / `My skin is …` 须提炼为短语写入此处 + - **禁止**仅写 `self` / `user` / `unknown` 等泛化词。 + - 不得把产品缺陷/差评（如 `pulls hair`、`battery dead`）写入 persona_signals；产品体验属于 product_feedback。 + - 若无任何可区分信号，可输出 `[]`。 2. pain_points (用户需求): - 仅限提取用户在购买前遇到的外部困扰、疾病、或具体场景（购买前尚未被本产品解决的需求）。 diff --git a/prompts/extraction/filter_irrelevant.md b/prompts/extraction/filter_irrelevant.md index b337ee7..313bc7d 100644 --- a/prompts/extraction/filter_irrelevant.md +++ b/prompts/extraction/filter_irrelevant.md @@ -1,3 +1,3 @@ 5. 无关评论过滤： - - 若某条评论明显与{product_name}无关（其他品类、其他 SKU 或完全跑题），该条输出：{{"audience": "unknown", "pain_points": [], "product_feedback": []}}。 - - 不得将无关内容填入 audience、pain_points 或 product_feedback。 + - 若某条评论明显与{product_name}无关（其他品类、其他 SKU 或完全跑题），该条输出：{{"persona_signals": [], "pain_points": [], "product_feedback": []}}。 + - 不得将无关内容填入 persona_signals、pain_points 或 product_feedback。 diff --git a/prompts/extraction/format_constraints_batch.md b/prompts/extraction/format_constraints_batch.md index 0f37d81..e645756 100644 --- a/prompts/extraction/format_constraints_batch.md +++ b/prompts/extraction/format_constraints_batch.md @@ -1,3 +1,3 @@ 4. 单条评论对象内的格式与字段约束： - - 每个评论对象只能包含 `audience`, `pain_points`, `product_feedback` 这 3 个字段。 + - 每个评论对象只能包含 `persona_signals`, `pain_points`, `product_feedback` 这 3 个字段。 - **绝对不要**在 JSON 中输出 `instruction`、`教学说明` 或其他任何多余字段。 diff --git a/prompts/extraction/format_constraints_single.md b/prompts/extraction/format_constraints_single.md index fa3119d..e807701 100644 --- a/prompts/extraction/format_constraints_single.md +++ b/prompts/extraction/format_constraints_single.md @@ -1,4 +1,4 @@ 4. 格式与字段约束： - - **你的 JSON 输出只能包含 `audience`, `pain_points`, `product_feedback` 这 3 个根字段。** + - **你的 JSON 输出只能包含 `persona_signals`, `pain_points`, `product_feedback` 这 3 个根字段。** - **绝对不要**在 JSON 中输出 `instruction`、`教学说明` 或其他任何多余字段。 - 必须以纯 JSON 格式输出结果，不要包含任何 markdown 标记（如 ```json ）或其他解释性文字。 diff --git a/prompts/schema.yaml b/prompts/schema.yaml index 0fa8bd6..6d26bb8 100644 --- a/prompts/schema.yaml +++ b/prompts/schema.yaml @@ -3,7 +3,7 @@ extraction: root_fields: - - audience + - persona_signals - pain_points - product_feedback product_feedback_fields: diff --git a/voc_report.py b/voc_report.py index 5b1c97a..ee73d2f 100644 --- a/voc_report.py +++ b/voc_report.py @@ -321,7 +321,7 @@ def _extraction_to_display_zh(ext: Dict[str, Any]) -> Dict[str, Any]: if not isinstance(pains, list): pains = [] return { - "受众": str(ext.get("audience", "")).strip(), + "画像信号": [str(s).strip() for s in ext.get("persona_signals") or [] if str(s).strip()], "需求/痛点": [str(p).strip() for p in pains if str(p).strip()], "产品反馈": feedback, } diff --git a/voc_业务_2/build_report.py b/voc_业务_2/build_report.py index 904c4bb..a3e3cb2 100644 --- a/voc_业务_2/build_report.py +++ b/voc_业务_2/build_report.py @@ -39,7 +39,7 @@ from llm_analyzer import ( from report_utils import ( enrich_theme_keywords, recalc_neg_priorities, compute_persona_pcts, assign_persona_clusters, validate_persona_physiological_labels, - enrich_persona_catalog_physio_counts, + prepare_persona_catalog_for_llm, pick_persona_quotes, fix_jtbd_fields, filter_matrix_rows, enrich_matrix_scene_evidence, build_matrix_table_rows_html, enrich_rootcause_quotes, build_executive_summary, enhanced_market_judgment, @@ -53,6 +53,7 @@ from report_utils import ( build_product_stopwords, build_kano_grid_html, build_neg_theme_summary_note, build_neg_theme_table_rows, build_pos_theme_table_rows, build_sentiment_keyword_groups, prepare_keyword_display, build_keyword_tables_html, + build_quote_blocks_html, ) from echarts_builder import ( get_echarts_script, build_all_charts, calc_theme_freq, calc_per_asin_theme_freq, @@ -134,13 +135,20 @@ def build_report_data(loader: DataLoader, cfg: dict) -> Dict[str, Any]: data["cluster_data"] = cluster_data all_clusters = loader.load_cluster_data() reviews = loader.load_reviews() + extractions = loader.load_comment_extractions() + data["extractions"] = extractions + persona_quote_max = int(cfg.get("persona_quote_max", 3)) + rootcause_quote_max = int(cfg.get("rootcause_quote_max", 4)) physio_min = int(cfg.get("persona_physio_min_reviews", 5)) - enrich_persona_catalog_physio_counts( + min_hit_count = int(cfg.get("persona_min_hit_count", 5)) + data["persona_min_hit_count"] = min_hit_count + prepare_persona_catalog_for_llm( cluster_data["persona_cluster_catalog"], all_clusters, reviews, + min_physio_reviews=physio_min, ) # ── Persona ── - personas = discover_personas(cluster_data) + personas = discover_personas(cluster_data, product_name=product, industry=industry) personas = assign_persona_clusters(personas, all_clusters) personas = validate_persona_physiological_labels( personas, cluster_data, reviews, all_clusters, min_review_count=physio_min, @@ -156,11 +164,15 @@ def build_report_data(loader: DataLoader, cfg: dict) -> Dict[str, Any]: data["neg_themes"] = neg_themes data["pos_themes"] = pos_themes - # ── 主题频次统计 ── - neg_freq = calc_theme_freq(neg_themes, reviews, is_neg=True) - pos_freq = calc_theme_freq(pos_themes, reviews, is_neg=False) - per_asin_neg = calc_per_asin_theme_freq(neg_themes, reviews, is_neg=True) - per_asin_pos = calc_per_asin_theme_freq(pos_themes, reviews, is_neg=False) + # ── 主题频次统计（结构化 category+aspect 优先） ── + neg_freq = calc_theme_freq(neg_themes, reviews, is_neg=True, extractions=extractions) + pos_freq = calc_theme_freq(pos_themes, reviews, is_neg=False, extractions=extractions) + per_asin_neg = calc_per_asin_theme_freq( + neg_themes, reviews, is_neg=True, extractions=extractions, + ) + per_asin_pos = calc_per_asin_theme_freq( + pos_themes, reviews, is_neg=False, extractions=extractions, + ) neg_themes = recalc_neg_priorities( neg_themes, neg_freq, stats.neg_review_count, len(stats.asins), per_asin_neg, ) @@ -174,7 +186,10 @@ def build_report_data(loader: DataLoader, cfg: dict) -> Dict[str, Any]: personas = compute_persona_pcts(personas, reviews, all_clusters) personas = sort_personas_by_evidence(personas) data["personas"] = personas - data["persona_quotes"] = pick_persona_quotes(personas, reviews) + data["persona_quotes"] = pick_persona_quotes( + personas, reviews, extractions, + max_quotes=persona_quote_max, + ) # ── KANO + JTBD + 情感关键词（三者并发） ── neg_kw_groups = build_sentiment_keyword_groups(neg_themes, reviews, limit=6, is_neg=True) @@ -212,21 +227,16 @@ def build_report_data(loader: DataLoader, cfg: dict) -> Dict[str, Any]: data["matrix"] = matrix # ── 根因分析（各 Persona 并发） ── - per_aud = loader.load_per_audience_clusters() - per_aud_prompt = {} - for al, d in per_aud.items(): - per_aud_prompt[al] = { - "pains": [{"label": c.cluster_label, "top_phrases": c.top_phrases} for c in d["pain"]], - "negative": [{"label": c.cluster_label, "top_phrases": c.top_phrases} for c in d["negative"]], - "positive": [{"label": c.cluster_label, "top_phrases": c.top_phrases} for c in d["positive"]], - } rootcauses = analyze_all_rootcauses( - personas, per_aud_prompt, neg_themes, max_workers=llm_workers, - product_name=product, industry=industry, min_hit_count=10, + personas, neg_themes, all_clusters=all_clusters, + max_workers=llm_workers, + product_name=product, industry=industry, min_hit_count=min_hit_count, ) rootcauses = enrich_rootcause_quotes( - rootcauses, personas, per_aud, loader, reviews, + rootcauses, personas, loader, reviews, persona_quotes=data["persona_quotes"], + extractions=extractions, + max_quotes=rootcause_quote_max, ) rootcauses = normalize_rootcauses(rootcauses, neg_themes) rootcauses = sort_rootcauses_by_evidence(rootcauses, personas) @@ -411,21 +421,20 @@ def render_html(template_path: Path, data: Dict[str, Any], cfg: dict) -> str: persona_cards = [] for i, p in enumerate(data["personas"]): pq = next((q for q in data["persona_quotes"] if q["persona"] == p["name"]), None) - quote_html = "" - if pq and pq.get("quote"): - q_cls = "neg" if pq.get("is_neg") else "pos" - star_hint = f" ({int(pq.get('rating', 0))}★)" if pq.get("rating") else "" - cn = pq.get("cn_summary") or quote_cn_summary(pq["quote"]) - quote_html = ( - f'
"{pq["quote"]}" ' - f'— ' - f'{asin_labels.get(pq["asin"], pq["asin"])}{star_hint}' - f'
摘要：{cn}
' - ) - elif pq: - quote_html = '
暂无与画像 keywords 匹配的代表性评论
' + quote_items = (pq.get("quotes") or []) if pq else [] + if not quote_items and pq and pq.get("quote"): + quote_items = [pq] + quote_html = build_quote_blocks_html( + quote_items, + asin_labels, + empty_msg="暂无命中池内的代表性评论", + ) normalize_persona_dimension(p) - meta = persona_display_meta(p, stats.total_reviews) + min_hit = int( + data.get("persona_min_hit_count") + or cfg.get("persona_min_hit_count", 5) + ) + meta = persona_display_meta(p, stats.total_reviews, min_hit_count=min_hit) persona_cards.append(f"""
{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, },