辉哥版本:结构化聚类溯源归因与业务报告增强。

移除结构化 audience 字段,强化 voc_业务_2 源评论归因匹配与 Persona 引用展示,更新 README 与流水线默认清理 SQLite。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
OnesvmWhoops 2026-06-17 09:11:04 +08:00
parent cb6692c0c0
commit c91e7f8c20
23 changed files with 1493 additions and 602 deletions

433
README.md
View file

@ -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)。*

View file

@ -60,6 +60,8 @@ DEFAULT_INDUSTRY = "亚马逊电商"
DEFAULT_PRODUCT = "亚马逊商品"
def _safe_product_dir_name(product_name: str) -> str:
"""用于 output 子目录名;去除路径非法字符。"""
name = (product_name or "product").strip()

View file

@ -16,7 +16,7 @@
## 占位符
`.md` 文件使用 Python `format` 语法,例如 `{product_name}`、`{industry}`。
正文里需要字面量花括号时写双花括号:`{{"audience": "unknown"}}`。
正文里需要字面量花括号时写双花括号:`{{"persona_signals": []}}`。
## 改完怎么验收

View file

@ -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 代码围栏或任何解释文字

View file

@ -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:

View file

@ -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 (用户需求):
- 仅限提取用户在购买前遇到的外部困扰、疾病、或具体场景(购买前尚未被本产品解决的需求)。

View file

@ -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。

View file

@ -1,3 +1,3 @@
4. 单条评论对象内的格式与字段约束:
- 每个评论对象只能包含 `audience`, `pain_points`, `product_feedback` 这 3 个字段。
- 每个评论对象只能包含 `persona_signals`, `pain_points`, `product_feedback` 这 3 个字段。
- **绝对不要**在 JSON 中输出 `instruction`、`教学说明` 或其他任何多余字段。

View file

@ -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 )或其他解释性文字。

View file

@ -3,7 +3,7 @@
extraction:
root_fields:
- audience
- persona_signals
- pain_points
- product_feedback
product_feedback_fields:

View file

@ -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,
}

View file

@ -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'<div class="quote {q_cls}">"{pq["quote"]}" '
f'— <a href="{pq["amazon_url"]}" target="_blank" rel="noopener">'
f'{asin_labels.get(pq["asin"], pq["asin"])}</a>{star_hint}'
f'<div class="quote-cn">摘要:{cn}</div></div>'
)
elif pq:
quote_html = '<div class="quote-empty">暂无与画像 keywords 匹配的代表性评论</div>'
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"""<div class="persona">
<div class="p-name">{p.get("name", "?")}</div>
<div class="p-meta">{meta}</div>
@ -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 调用,仅渲染模板")

View file

@ -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)

View file

@ -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,

View file

@ -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()}

View file

@ -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}")

View file

@ -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

File diff suppressed because it is too large Load diff

View file

@ -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"]])

View file

@ -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}}` 等 — 聚类摘要(自动填入)

View file

@ -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,

View file

@ -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"

125
聚类.py
View file

@ -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,
},