2441c9119c
包含七步编排入口、结构化/向量化/聚类/词频/报告模块与 prompts 配置；忽略原始 CSV 与本地密钥。 Co-authored-by: Cursor <cursoragent@cursor.com>
1082 lines
36 KiB
Python
1082 lines
36 KiB
Python
"""
|
||
VOC 评论结构化服务（命令行 / 直接调用，无 MCP）。
|
||
|
||
----------------------------------------------------------------------
|
||
虚拟环境（推荐 Python 3.10+）
|
||
|
||
cd "/Users/onesvmwhoops/Cursor_Project/VOC_LLM结构化"
|
||
source 310py/bin/activate
|
||
pip install -r requirements.txt
|
||
|
||
----------------------------------------------------------------------
|
||
运行前提供密钥（任选其一；勿把密钥写进代码仓库）::
|
||
|
||
export DASHSCOPE_API_KEY="sk-xxx"
|
||
# 或项目根单行文件 .dashscope_key
|
||
# 或 export DASHSCOPE_API_KEY_FILE="/path/to/key.txt"
|
||
|
||
----------------------------------------------------------------------
|
||
用法::
|
||
|
||
python3 结构化_server.py --industry "Pet supplements" --product "Turkey tail mushroom for dogs" --file merged_reviews_cleaned.csv
|
||
|
||
python3 结构化_server.py --smoke
|
||
|
||
模型：默认 ``qwen3.6-flash``（OpenAI 兼容 Chat API；可通过环境变量 ``DASHSCOPE_MODEL`` 覆盖）。
|
||
|
||
地域：固定中国大陆华北2（北京）``https://dashscope.aliyuncs.com/compatible-mode/v1``，不使用新加坡/国际节点。
|
||
|
||
数据库：项目根目录 ``voc_structured.sqlite``；每次写入前会清理该库及下游 ``voc_embeddings.sqlite``、``voc_clustering.sqlite``（冒烟测试用临时库时不清理项目根文件）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import importlib.util
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import sys
|
||
from csv import DictReader
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Tuple
|
||
|
||
from prompts.loader import product_feedback_categories
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||
stream=sys.stderr,
|
||
)
|
||
logger = logging.getLogger("voc_structured")
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent
|
||
DB_PATH = PROJECT_ROOT / "voc_structured.sqlite"
|
||
EMBED_DB = PROJECT_ROOT / "voc_embeddings.sqlite"
|
||
CLUSTER_DB = PROJECT_ROOT / "voc_clustering.sqlite"
|
||
# 中国大陆华北2（北京）OpenAI 兼容模式
|
||
DASHSCOPE_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1"
|
||
MODEL_NAME = os.environ.get("DASHSCOPE_MODEL", "qwen3.6-flash").strip()
|
||
REVIEW_COLUMN_NAMES = ("content", "review", "评论")
|
||
BATCH_FETCH_MAX_ATTEMPTS = 3
|
||
BATCH_CORRECTION_MAX_ATTEMPTS = 3
|
||
|
||
_VALID_CATEGORIES = product_feedback_categories() # 模块加载时快照；校验时用 _get_valid_categories()
|
||
|
||
|
||
def _get_valid_categories() -> frozenset[str]:
|
||
return product_feedback_categories()
|
||
|
||
# 模型上下文上限（保守估）；单批仍用下方 DEFAULT_MAX_* 控制，保证 JSON 稳定
|
||
MODEL_MAX_INPUT_TOKENS = 991_800
|
||
MODEL_MAX_OUTPUT_TOKENS = 65_530
|
||
CHARS_PER_TOKEN_EST = 3.2
|
||
DEFAULT_MAX_BATCH_INPUT_TOKENS = 32_000
|
||
DEFAULT_MAX_BATCH_OUTPUT_TOKENS = 12_000
|
||
DEFAULT_OUTPUT_TOKENS_PER_REVIEW = 450
|
||
BATCH_COUNT_MIN = 1
|
||
BATCH_COUNT_MAX = 50
|
||
# 批量结构化单次请求输出 token（含 JSON 开销，上限不超过模型）
|
||
BATCH_OUTPUT_TOKEN_BUFFER = 1024
|
||
BATCH_OUTPUT_TOKEN_FLOOR = 4096
|
||
|
||
|
||
def _batch_max_output_tokens(review_count: int) -> int:
|
||
est = review_count * DEFAULT_OUTPUT_TOKENS_PER_REVIEW + BATCH_OUTPUT_TOKEN_BUFFER
|
||
return min(MODEL_MAX_OUTPUT_TOKENS, max(BATCH_OUTPUT_TOKEN_FLOOR, est))
|
||
|
||
|
||
def _resolve_dashscope_api_key() -> str:
|
||
"""环境变量 > DASHSCOPE_API_KEY_FILE > 项目根 .dashscope_key（均仅一行密钥，无引号）。"""
|
||
v = os.environ.get("DASHSCOPE_API_KEY", "").strip()
|
||
if v:
|
||
return v
|
||
fp = os.environ.get("DASHSCOPE_API_KEY_FILE", "").strip()
|
||
if fp:
|
||
p = Path(fp).expanduser()
|
||
if p.is_file():
|
||
return p.read_text(encoding="utf-8").strip().strip('"').strip("'")
|
||
local = PROJECT_ROOT / ".dashscope_key"
|
||
if local.is_file():
|
||
return local.read_text(encoding="utf-8").strip().strip('"').strip("'")
|
||
return ""
|
||
|
||
|
||
def _load_prompt_module():
|
||
path = PROJECT_ROOT / "结构化_Prompt.py"
|
||
spec = importlib.util.spec_from_file_location("voc_structured_prompts", path)
|
||
if spec is None or spec.loader is None:
|
||
raise RuntimeError(f"Cannot load prompt module from {path}")
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
return mod
|
||
|
||
|
||
_PROMPTS = _load_prompt_module()
|
||
|
||
|
||
def _clean_voc_sqlite_databases(structured_db: Path) -> None:
|
||
"""写入前删除 SQLite，避免与旧 job / 下游向量、聚类结果混用。"""
|
||
paths = [structured_db.resolve()]
|
||
if structured_db.resolve() == DB_PATH.resolve():
|
||
paths.extend([EMBED_DB.resolve(), CLUSTER_DB.resolve()])
|
||
for p in paths:
|
||
if p.is_file():
|
||
p.unlink()
|
||
logger.info("已清理 SQLite: %s", p.name)
|
||
|
||
|
||
def init_sqlite_schema(conn: sqlite3.Connection) -> None:
|
||
conn.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS analysis_jobs (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
created_at TEXT NOT NULL,
|
||
industry TEXT NOT NULL,
|
||
product_name TEXT NOT NULL,
|
||
source_file TEXT NOT NULL,
|
||
batch_size INTEGER NOT NULL,
|
||
model TEXT NOT NULL,
|
||
full_result_json TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS comment_extractions (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
job_id INTEGER NOT NULL,
|
||
source_row INTEGER NOT NULL,
|
||
batch_index INTEGER NOT NULL,
|
||
comment_key TEXT NOT NULL,
|
||
extraction_json TEXT NOT NULL,
|
||
UNIQUE(job_id, source_row),
|
||
FOREIGN KEY (job_id) REFERENCES analysis_jobs(id)
|
||
);
|
||
"""
|
||
)
|
||
|
||
|
||
def load_reviews_from_file(file_path: str) -> List[Tuple[int, str]]:
|
||
"""
|
||
返回 (source_row, text) 列表。
|
||
CSV：source_row 为表头下的数据行序号（第 1 行数据=1），空单元格跳过。
|
||
TXT：source_row 为文件物理行号（从 1 起），空行跳过。
|
||
"""
|
||
p = Path(file_path).expanduser().resolve()
|
||
if not p.is_file():
|
||
raise FileNotFoundError(f"File not found: {p}")
|
||
|
||
suffix = p.suffix.lower()
|
||
if suffix == ".csv":
|
||
return _load_csv_reviews(p)
|
||
if suffix == ".txt" or suffix == ".text":
|
||
return _load_txt_reviews(p)
|
||
raise ValueError(f"Unsupported file type {suffix!r}; use .csv or .txt")
|
||
|
||
|
||
def _load_txt_reviews(p: Path) -> List[Tuple[int, str]]:
|
||
out: List[Tuple[int, str]] = []
|
||
with p.open(encoding="utf-8-sig", newline="") as f:
|
||
for line_no, line in enumerate(f, start=1):
|
||
t = line.strip("\r\n").strip()
|
||
if not t:
|
||
continue
|
||
out.append((line_no, t))
|
||
return out
|
||
|
||
|
||
def _load_csv_reviews(p: Path) -> List[Tuple[int, str]]:
|
||
with p.open(encoding="utf-8-sig", newline="") as f:
|
||
reader = DictReader(f)
|
||
if not reader.fieldnames:
|
||
raise ValueError("CSV has no header row")
|
||
headers = [h.strip() for h in reader.fieldnames]
|
||
lower_map = {h.lower(): h for h in headers}
|
||
col: str | None = None
|
||
for name in REVIEW_COLUMN_NAMES:
|
||
if name.lower() in lower_map:
|
||
col = lower_map[name.lower()]
|
||
break
|
||
if col is None:
|
||
if len(headers) == 1:
|
||
col = headers[0]
|
||
else:
|
||
raise ValueError(
|
||
"CSV must contain column 'content', 'review', or '评论', "
|
||
f"or be single-column. Found columns: {headers}"
|
||
)
|
||
rows = list(reader)
|
||
out: List[Tuple[int, str]] = []
|
||
for idx, row in enumerate(rows, start=1):
|
||
cell = row.get(col)
|
||
if cell is None:
|
||
cell = ""
|
||
t = str(cell).strip()
|
||
if not t:
|
||
continue
|
||
out.append((idx, t))
|
||
return out
|
||
|
||
|
||
def _chunked(items: List[Tuple[int, str]], size: int) -> List[List[Tuple[int, str]]]:
|
||
return [items[i : i + size] for i in range(0, len(items), size)]
|
||
|
||
|
||
def _estimate_tokens(text: str) -> int:
|
||
"""英文评论粗略估 token（偏保守，避免顶满上下文）。"""
|
||
return max(1, int(len(text) / CHARS_PER_TOKEN_EST))
|
||
|
||
|
||
def _estimate_batch_input_tokens(
|
||
industry: str,
|
||
product_name: str,
|
||
batch: List[Tuple[int, str]],
|
||
) -> int:
|
||
tagged, keys, _ = _format_tagged_batch(batch)
|
||
system, user = _PROMPTS.build_batch_review_analysis_prompts(
|
||
industry=industry,
|
||
product_name=product_name,
|
||
keys=keys,
|
||
tagged_input=tagged,
|
||
)
|
||
return _estimate_tokens(system) + _estimate_tokens(user)
|
||
|
||
|
||
def _estimate_batch_output_tokens(review_count: int) -> int:
|
||
return review_count * DEFAULT_OUTPUT_TOKENS_PER_REVIEW
|
||
|
||
|
||
def chunk_reviews_by_token_budget(
|
||
reviews: List[Tuple[int, str]],
|
||
industry: str,
|
||
product_name: str,
|
||
*,
|
||
max_input_tokens: int = DEFAULT_MAX_BATCH_INPUT_TOKENS,
|
||
max_output_tokens: int = DEFAULT_MAX_BATCH_OUTPUT_TOKENS,
|
||
max_reviews_per_batch: int = BATCH_COUNT_MAX,
|
||
min_reviews_per_batch: int = BATCH_COUNT_MIN,
|
||
) -> List[List[Tuple[int, str]]]:
|
||
"""
|
||
按预估输入/输出 token 与单批条数上限，将评论动态打包。
|
||
短评可多条约一批，长评自动减少条数（单条超长则单独成批）。
|
||
"""
|
||
max_input_tokens = min(max_input_tokens, MODEL_MAX_INPUT_TOKENS)
|
||
max_output_tokens = min(max_output_tokens, MODEL_MAX_OUTPUT_TOKENS)
|
||
max_reviews_per_batch = max(min_reviews_per_batch, max_reviews_per_batch)
|
||
|
||
batches: List[List[Tuple[int, str]]] = []
|
||
i = 0
|
||
while i < len(reviews):
|
||
batch: List[Tuple[int, str]] = []
|
||
while i < len(reviews):
|
||
candidate = batch + [reviews[i]]
|
||
if len(candidate) > max_reviews_per_batch:
|
||
break
|
||
|
||
inp = _estimate_batch_input_tokens(industry, product_name, candidate)
|
||
out = _estimate_batch_output_tokens(len(candidate))
|
||
|
||
if batch and (inp > max_input_tokens or out > max_output_tokens):
|
||
break
|
||
|
||
batch = candidate
|
||
i += 1
|
||
|
||
if inp > max_input_tokens:
|
||
logger.warning(
|
||
"单条评论过长（约 %s input tokens），单独成批: source_row=%s",
|
||
inp,
|
||
batch[0][0],
|
||
)
|
||
break
|
||
|
||
if not batch:
|
||
break
|
||
batches.append(batch)
|
||
|
||
return batches
|
||
|
||
|
||
def _summarize_batch_plan(
|
||
batches: List[List[Tuple[int, str]]],
|
||
industry: str,
|
||
product_name: str,
|
||
) -> List[Dict[str, Any]]:
|
||
plan: List[Dict[str, Any]] = []
|
||
for idx, batch in enumerate(batches):
|
||
chars = sum(len(t) for _, t in batch)
|
||
plan.append(
|
||
{
|
||
"batch_index": idx,
|
||
"review_count": len(batch),
|
||
"total_chars": chars,
|
||
"est_input_tokens": _estimate_batch_input_tokens(
|
||
industry, product_name, batch
|
||
),
|
||
"est_output_tokens": _estimate_batch_output_tokens(len(batch)),
|
||
}
|
||
)
|
||
return plan
|
||
|
||
|
||
def _format_tagged_batch(batch: List[Tuple[int, str]]) -> Tuple[str, List[str], Dict[str, int]]:
|
||
"""返回 tagged 文本、键列表 C1..Cn、Ci -> source_row。"""
|
||
lines: List[str] = []
|
||
keys: List[str] = []
|
||
key_to_row: Dict[str, int] = {}
|
||
for i, (source_row, text) in enumerate(batch, start=1):
|
||
key = f"C{i}"
|
||
keys.append(key)
|
||
key_to_row[key] = source_row
|
||
safe = text.replace("\r\n", "\n").replace("\r", "\n")
|
||
lines.append(f"[{key}] {safe}")
|
||
return "\n".join(lines), keys, key_to_row
|
||
|
||
|
||
def _strip_code_fence(text: str) -> str:
|
||
t = text.strip()
|
||
if t.startswith("```"):
|
||
lines = t.split("\n")
|
||
if len(lines) >= 2 and lines[0].startswith("```"):
|
||
lines = lines[1:]
|
||
if lines and lines[-1].strip() == "```":
|
||
lines = lines[:-1]
|
||
t = "\n".join(lines).strip()
|
||
return t
|
||
|
||
|
||
def parse_model_json(raw: str) -> Any:
|
||
t = _strip_code_fence(raw)
|
||
try:
|
||
return json.loads(t)
|
||
except json.JSONDecodeError:
|
||
m = re.search(r"[\{\[][\s\S]*[\}\]]\s*$", t)
|
||
if not m:
|
||
raise
|
||
return json.loads(m.group(0))
|
||
|
||
|
||
def parse_model_json_object(raw: str) -> Dict[str, Any]:
|
||
parsed = parse_model_json(raw)
|
||
if isinstance(parsed, dict):
|
||
return parsed
|
||
raise ValueError(
|
||
f"模型 JSON 应为对象，实际为 {type(parsed).__name__}；请使用批量对象格式 {{\"C1\": ...}}"
|
||
)
|
||
|
||
|
||
def _call_dashscope_chat(
|
||
messages: List[Dict[str, str]],
|
||
*,
|
||
max_tokens: int | None = None,
|
||
) -> Any:
|
||
api_key = _resolve_dashscope_api_key()
|
||
if not api_key:
|
||
raise RuntimeError(
|
||
"Missing DashScope API key: set DASHSCOPE_API_KEY in the process environment, "
|
||
"or DASHSCOPE_API_KEY_FILE to a one-line key file, "
|
||
"or create a one-line file at project root: .dashscope_key"
|
||
)
|
||
|
||
from openai import OpenAI
|
||
|
||
client = OpenAI(api_key=api_key, base_url=DASHSCOPE_BASE_URL)
|
||
extra_body: Dict[str, Any] = {}
|
||
model_lower = MODEL_NAME.lower()
|
||
if model_lower.startswith(("qwen3.6", "qwen3.5", "qwen3")) or model_lower.startswith(
|
||
"deepseek-v4"
|
||
):
|
||
extra_body["enable_thinking"] = False
|
||
|
||
out_tokens = max_tokens if max_tokens is not None else DEFAULT_MAX_BATCH_OUTPUT_TOKENS
|
||
out_tokens = min(MODEL_MAX_OUTPUT_TOKENS, max(BATCH_OUTPUT_TOKEN_FLOOR, out_tokens))
|
||
|
||
resp = client.chat.completions.create(
|
||
model=MODEL_NAME,
|
||
messages=messages,
|
||
response_format={"type": "json_object"},
|
||
temperature=0.2,
|
||
max_tokens=out_tokens,
|
||
**({"extra_body": extra_body} if extra_body else {}),
|
||
)
|
||
msg = resp.choices[0].message
|
||
choice = msg.content
|
||
if not choice and getattr(msg, "reasoning_content", None):
|
||
choice = msg.reasoning_content
|
||
if not choice:
|
||
raise RuntimeError("Empty model response")
|
||
return parse_model_json(choice)
|
||
|
||
|
||
def _call_dashscope_model(system: str, user: str) -> Dict[str, Any]:
|
||
return _call_dashscope_chat(
|
||
[
|
||
{"role": "system", "content": system},
|
||
{"role": "user", "content": user},
|
||
]
|
||
)
|
||
|
||
|
||
def _coalesce_batch_response(raw: Dict[str, Any], keys: List[str]) -> Dict[str, Any]:
|
||
"""若模型把 C1..Cn 包在嵌套对象里，尝试展开到顶层。"""
|
||
if any(k in raw for k in keys):
|
||
return raw
|
||
for value in raw.values():
|
||
if isinstance(value, dict) and any(k in value for k in keys):
|
||
return value
|
||
return raw
|
||
|
||
|
||
def _normalize_batch_response(raw: Any, keys: List[str]) -> Dict[str, Any]:
|
||
"""
|
||
统一为 {C1: {...}, C2: {...}}。
|
||
兼容模型误返回 JSON 数组、嵌套对象或按序号的对象。
|
||
"""
|
||
if isinstance(raw, list):
|
||
out: Dict[str, Any] = {}
|
||
for i, key in enumerate(keys):
|
||
if i >= len(raw):
|
||
break
|
||
item = raw[i]
|
||
if not isinstance(item, dict):
|
||
continue
|
||
if len(item) == 1:
|
||
sole_k, sole_v = next(iter(item.items()))
|
||
if sole_k in keys and isinstance(sole_v, dict):
|
||
out[sole_k] = sole_v
|
||
continue
|
||
if any(k in item for k in keys):
|
||
for k in keys:
|
||
if k in item and isinstance(item[k], dict):
|
||
out[k] = item[k]
|
||
continue
|
||
out[key] = item
|
||
if out:
|
||
return out
|
||
raise ValueError(
|
||
"模型返回 JSON 数组，但无法按顺序映射为 "
|
||
+ ", ".join(keys[: min(len(raw), len(keys))])
|
||
)
|
||
|
||
if not isinstance(raw, dict):
|
||
raise TypeError(f"批量 JSON 应为 object 或 array，实际为 {type(raw).__name__}")
|
||
|
||
coalesced = _coalesce_batch_response(raw, keys)
|
||
if any(k in coalesced for k in keys):
|
||
return coalesced
|
||
|
||
# 如 {"1": {...}, "2": {...}} 映射到 C1, C2
|
||
remapped: Dict[str, Any] = {}
|
||
for i, key in enumerate(keys, start=1):
|
||
for alias in (str(i), f"C{i}"):
|
||
if alias in coalesced and isinstance(coalesced[alias], dict):
|
||
remapped[key] = coalesced[alias]
|
||
break
|
||
if remapped:
|
||
return remapped
|
||
|
||
return coalesced
|
||
|
||
|
||
_SENTIMENT_CANONICAL = {
|
||
"positive": "Positive",
|
||
"negative": "Negative",
|
||
"neutral": "Neutral",
|
||
}
|
||
|
||
# 模型常见误写 -> 合法 category（校验前自动映射，减少无效重试）
|
||
_CATEGORY_ALIASES: Dict[str, str] = {
|
||
"value": "Price",
|
||
"values": "Price",
|
||
"cost": "Price",
|
||
"pricing": "Price",
|
||
"value for money": "Price",
|
||
"worth": "Price",
|
||
"packaging": "Logistics",
|
||
"shipping": "Logistics",
|
||
"delivery": "Logistics",
|
||
"service": "Customer Service",
|
||
"customer support": "Customer Service",
|
||
"support": "Customer Service",
|
||
"design": "Appearance",
|
||
"look": "Appearance",
|
||
"performance": "Function",
|
||
"efficacy": "Function",
|
||
"effectiveness": "Function",
|
||
}
|
||
|
||
|
||
def _normalize_sentiment(raw: Any) -> str:
|
||
key = str(raw or "").strip().lower()
|
||
return _SENTIMENT_CANONICAL.get(key, "Neutral")
|
||
|
||
|
||
def _normalize_category(raw: Any) -> str | None:
|
||
valid = _get_valid_categories()
|
||
key = str(raw or "").strip()
|
||
if not key:
|
||
return None
|
||
if key in valid:
|
||
return key
|
||
mapped = _CATEGORY_ALIASES.get(key.lower())
|
||
if mapped:
|
||
return mapped
|
||
lower = key.lower()
|
||
for v in valid:
|
||
if v.lower() == lower:
|
||
return v
|
||
return None
|
||
|
||
|
||
def _normalize_product_feedback(items: Any) -> List[Dict[str, str]]:
|
||
if not isinstance(items, list):
|
||
return []
|
||
out: List[Dict[str, str]] = []
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
aspect = str(item.get("aspect", "")).strip()
|
||
opinion = str(item.get("opinion", "")).strip()
|
||
category = _normalize_category(item.get("category"))
|
||
if aspect and opinion and category:
|
||
out.append(
|
||
{
|
||
"aspect": aspect,
|
||
"opinion": opinion,
|
||
"sentiment": _normalize_sentiment(item.get("sentiment")),
|
||
"category": category,
|
||
}
|
||
)
|
||
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"):
|
||
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 必须为非空字符串")
|
||
|
||
pain_points = obj.get("pain_points")
|
||
if not isinstance(pain_points, list):
|
||
raise ValueError(f"{prefix}pain_points 必须为数组")
|
||
for i, p in enumerate(pain_points):
|
||
if not isinstance(p, str) or not str(p).strip():
|
||
raise ValueError(f"{prefix}pain_points[{i}] 必须为非空字符串")
|
||
|
||
pf = obj.get("product_feedback")
|
||
if not isinstance(pf, list):
|
||
raise ValueError(f"{prefix}product_feedback 必须为数组")
|
||
for i, item in enumerate(pf):
|
||
if not isinstance(item, dict):
|
||
raise ValueError(f"{prefix}product_feedback[{i}] 必须是对象")
|
||
for sub in ("aspect", "opinion", "sentiment", "category"):
|
||
if not str(item.get(sub, "")).strip():
|
||
raise ValueError(f"{prefix}product_feedback[{i}] 缺少或空的 {sub}")
|
||
sent_key = str(item.get("sentiment", "")).strip().lower()
|
||
if sent_key not in _SENTIMENT_CANONICAL:
|
||
raise ValueError(
|
||
f"{prefix}product_feedback[{i}].sentiment 必须为 "
|
||
"Positive、Negative 或 Neutral"
|
||
)
|
||
raw_cat = str(item.get("category", "")).strip()
|
||
cat = _normalize_category(raw_cat)
|
||
if cat is None:
|
||
raise ValueError(
|
||
f"{prefix}product_feedback[{i}].category 非法: {raw_cat!r}，"
|
||
f"允许: {', '.join(sorted(_get_valid_categories()))}"
|
||
)
|
||
|
||
return _normalize_extraction(obj)
|
||
|
||
|
||
def _normalize_extraction(obj: Any) -> Dict[str, Any]:
|
||
"""补齐缺省字段，避免模型漏写 pain_points / product_feedback 导致整条丢弃。"""
|
||
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()
|
||
|
||
pain_points = obj.get("pain_points", [])
|
||
if pain_points is None:
|
||
pain_points = []
|
||
if not isinstance(pain_points, list):
|
||
pain_points = [str(pain_points)] if str(pain_points).strip() else []
|
||
pain_points = [str(p).strip() for p in pain_points if str(p).strip()]
|
||
|
||
product_feedback = _normalize_product_feedback(obj.get("product_feedback", []))
|
||
|
||
missing = [
|
||
f
|
||
for f in ("audience", "pain_points", "product_feedback")
|
||
if f not in obj
|
||
]
|
||
if missing:
|
||
logger.info("Normalized missing fields %s in extraction", missing)
|
||
|
||
return {
|
||
"audience": audience,
|
||
"pain_points": pain_points,
|
||
"product_feedback": product_feedback,
|
||
}
|
||
|
||
|
||
def _build_batch_correction_message(keys: List[str], errors: List[str]) -> str:
|
||
keys_literal = ", ".join(json.dumps(k) for k in keys)
|
||
err_block = "\n".join(f"- {e}" for e in errors)
|
||
return (
|
||
"你上一次返回的 JSON 未通过服务端校验，请根据下列错误修正后重新输出。\n\n"
|
||
f"校验错误：\n{err_block}\n\n"
|
||
"要求：\n"
|
||
f"- 输出一个 JSON 对象，顶层键必须且仅能是：{keys_literal}\n"
|
||
"- 每个键的值必须包含 audience、pain_points、product_feedback\n"
|
||
"- product_feedback 每条须含 aspect、opinion、sentiment"
|
||
"（Positive/Negative/Neutral）、category（禁止 Value，性价比用 Price）\n"
|
||
"- 只输出一个 JSON 对象（不要用数组），不要 markdown 代码围栏或解释文字"
|
||
)
|
||
|
||
|
||
def _parse_batch_with_model_correction(
|
||
system: str,
|
||
user: str,
|
||
sub_keys: List[str],
|
||
*,
|
||
batch_index: int,
|
||
attempt_label: str,
|
||
) -> Tuple[Dict[str, Any], Dict[str, Dict[str, Any]], List[str]]:
|
||
"""
|
||
调用模型并在校验失败时将错误信息附在对话中重试。
|
||
返回 (原始 JSON, 已通过校验的条目, 仍失败的错误列表)。
|
||
"""
|
||
messages: List[Dict[str, str]] = [
|
||
{"role": "system", "content": system},
|
||
{"role": "user", "content": user},
|
||
]
|
||
raw_obj: Dict[str, Any] = {}
|
||
ok: Dict[str, Dict[str, Any]] = {}
|
||
errors: List[str] = []
|
||
|
||
for correction in range(BATCH_CORRECTION_MAX_ATTEMPTS):
|
||
logger.info(
|
||
"请求模型 batch %s（%s），校验轮次 %s/%s…",
|
||
batch_index,
|
||
attempt_label,
|
||
correction + 1,
|
||
BATCH_CORRECTION_MAX_ATTEMPTS,
|
||
)
|
||
try:
|
||
raw_parsed = _call_dashscope_chat(
|
||
messages, max_tokens=_batch_max_output_tokens(len(sub_keys))
|
||
)
|
||
raw_obj = _normalize_batch_response(raw_parsed, sub_keys)
|
||
except (RuntimeError, json.JSONDecodeError, ValueError, TypeError) as e:
|
||
errors = [f"模型响应无法解析为 JSON: {e}"]
|
||
ok = {}
|
||
if correction + 1 >= BATCH_CORRECTION_MAX_ATTEMPTS:
|
||
break
|
||
logger.warning(
|
||
"Batch %s parse error (correction %s): %s",
|
||
batch_index,
|
||
correction + 1,
|
||
e,
|
||
)
|
||
messages.append(
|
||
{
|
||
"role": "assistant",
|
||
"content": json.dumps(
|
||
{"error": "invalid_json", "detail": str(e)},
|
||
ensure_ascii=False,
|
||
),
|
||
}
|
||
)
|
||
messages.append(
|
||
{
|
||
"role": "user",
|
||
"content": _build_batch_correction_message(sub_keys, errors),
|
||
}
|
||
)
|
||
continue
|
||
|
||
ok = {}
|
||
errors = []
|
||
for sub_ck in sub_keys:
|
||
if sub_ck not in raw_obj:
|
||
errors.append(f"键 {sub_ck}：响应 JSON 缺少该顶层键")
|
||
continue
|
||
try:
|
||
ok[sub_ck] = _validate_extraction_strict(raw_obj[sub_ck], sub_ck)
|
||
except ValueError as e:
|
||
errors.append(str(e))
|
||
|
||
if not errors:
|
||
return raw_obj, ok, []
|
||
|
||
if correction + 1 >= BATCH_CORRECTION_MAX_ATTEMPTS:
|
||
break
|
||
|
||
logger.warning(
|
||
"Batch %s validation failed (correction %s/%s): %s",
|
||
batch_index,
|
||
correction + 1,
|
||
BATCH_CORRECTION_MAX_ATTEMPTS,
|
||
errors,
|
||
)
|
||
messages.append(
|
||
{"role": "assistant", "content": json.dumps(raw_obj, ensure_ascii=False)}
|
||
)
|
||
messages.append(
|
||
{
|
||
"role": "user",
|
||
"content": _build_batch_correction_message(sub_keys, errors),
|
||
}
|
||
)
|
||
|
||
return raw_obj, ok, errors
|
||
|
||
|
||
def _fetch_batch_extractions(
|
||
industry: str,
|
||
product_name: str,
|
||
batch: List[Tuple[int, str]],
|
||
b_idx: int,
|
||
) -> Dict[str, Any]:
|
||
"""请求一批评论的结构化结果；对缺失/无效键自动缩小批次重试。"""
|
||
_, all_keys, key_to_row = _format_tagged_batch(batch)
|
||
row_to_text = {row: text for row, text in batch}
|
||
pending_keys: List[str] = list(all_keys)
|
||
merged: Dict[str, Any] = {}
|
||
|
||
for attempt in range(BATCH_FETCH_MAX_ATTEMPTS):
|
||
if not pending_keys:
|
||
break
|
||
|
||
sub_batch = [
|
||
(key_to_row[ck], row_to_text[key_to_row[ck]]) for ck in pending_keys
|
||
]
|
||
tagged, sub_keys, sub_key_to_row = _format_tagged_batch(sub_batch)
|
||
# 重试时子批次会重新编号为 C1..Cn，需映射回原始键名
|
||
sub_to_orig = {sub_keys[i]: pending_keys[i] for i in range(len(sub_keys))}
|
||
|
||
system, user = _PROMPTS.build_batch_review_analysis_prompts(
|
||
industry=industry,
|
||
product_name=product_name,
|
||
keys=sub_keys,
|
||
tagged_input=tagged,
|
||
)
|
||
logger.info(
|
||
"请求模型 batch %s，第 %s 次尝试，本批 %s 条…",
|
||
b_idx,
|
||
attempt + 1,
|
||
len(sub_keys),
|
||
)
|
||
_, ok_map, val_errors = _parse_batch_with_model_correction(
|
||
system,
|
||
user,
|
||
sub_keys,
|
||
batch_index=b_idx,
|
||
attempt_label=f"attempt {attempt + 1}",
|
||
)
|
||
|
||
next_pending: List[str] = []
|
||
for sub_ck in sub_keys:
|
||
orig_ck = sub_to_orig[sub_ck]
|
||
if sub_ck in ok_map:
|
||
merged[orig_ck] = ok_map[sub_ck]
|
||
else:
|
||
next_pending.append(orig_ck)
|
||
|
||
if val_errors:
|
||
logger.warning(
|
||
"Batch %s still has validation issues after correction (attempt %s): %s",
|
||
b_idx,
|
||
attempt + 1,
|
||
val_errors,
|
||
)
|
||
|
||
pending_keys = next_pending
|
||
if pending_keys and attempt + 1 < BATCH_FETCH_MAX_ATTEMPTS:
|
||
logger.info(
|
||
"Retrying %s review(s) in batch %s (attempt %s)",
|
||
len(pending_keys),
|
||
b_idx,
|
||
attempt + 2,
|
||
)
|
||
|
||
if pending_keys:
|
||
logger.warning(
|
||
"Batch %s: gave up on %s review(s) after %s attempts: %s",
|
||
b_idx,
|
||
len(pending_keys),
|
||
BATCH_FETCH_MAX_ATTEMPTS,
|
||
pending_keys,
|
||
)
|
||
return merged
|
||
|
||
|
||
def run_analysis(
|
||
industry: str,
|
||
product_name: str,
|
||
file_path: str,
|
||
batch_size: int | None = None,
|
||
*,
|
||
clean_databases: bool = True,
|
||
max_batch_input_tokens: int = DEFAULT_MAX_BATCH_INPUT_TOKENS,
|
||
max_batch_output_tokens: int = DEFAULT_MAX_BATCH_OUTPUT_TOKENS,
|
||
max_reviews_per_batch: int = BATCH_COUNT_MAX,
|
||
) -> Dict[str, Any]:
|
||
src = str(Path(file_path).expanduser().resolve())
|
||
reviews = load_reviews_from_file(src)
|
||
if not reviews:
|
||
raise ValueError("No non-empty reviews found in file")
|
||
|
||
if batch_size is not None and batch_size > 0:
|
||
if batch_size < BATCH_COUNT_MIN:
|
||
raise ValueError(f"batch_size must be >= {BATCH_COUNT_MIN}, got {batch_size}")
|
||
batches = _chunked(reviews, batch_size)
|
||
batching_mode = "fixed"
|
||
batch_size_record = batch_size
|
||
else:
|
||
batches = chunk_reviews_by_token_budget(
|
||
reviews,
|
||
industry,
|
||
product_name,
|
||
max_input_tokens=max_batch_input_tokens,
|
||
max_output_tokens=max_batch_output_tokens,
|
||
max_reviews_per_batch=max_reviews_per_batch,
|
||
)
|
||
batching_mode = "dynamic"
|
||
batch_size_record = max_reviews_per_batch
|
||
|
||
batch_plan = _summarize_batch_plan(batches, industry, product_name)
|
||
total_batches = len(batches)
|
||
counts = [p["review_count"] for p in batch_plan]
|
||
logger.info(
|
||
"开始结构化：共 %s 条评论，模式=%s，共 %s 批，每批条数 min/med/max=%s/%s/%s",
|
||
len(reviews),
|
||
batching_mode,
|
||
total_batches,
|
||
min(counts) if counts else 0,
|
||
sorted(counts)[len(counts) // 2] if counts else 0,
|
||
max(counts) if counts else 0,
|
||
)
|
||
for p in batch_plan:
|
||
logger.info(
|
||
" 批次 %s: %s 条, %s 字, 约 %s in / %s out tokens",
|
||
p["batch_index"] + 1,
|
||
p["review_count"],
|
||
p["total_chars"],
|
||
p["est_input_tokens"],
|
||
p["est_output_tokens"],
|
||
)
|
||
extractions_by_row: Dict[str, Dict[str, Any]] = {}
|
||
batch_details: List[Dict[str, Any]] = []
|
||
sqlite_rows: List[Tuple[int, int, str, Dict[str, Any]]] = []
|
||
|
||
for b_idx, batch in enumerate(batches):
|
||
logger.info("批次 %s/%s：处理 %s 条评论…", b_idx + 1, total_batches, len(batch))
|
||
_, keys, key_to_row = _format_tagged_batch(batch)
|
||
raw_obj = _fetch_batch_extractions(industry, product_name, batch, b_idx)
|
||
logger.info(
|
||
"批次 %s/%s 完成：成功 %s/%s 条",
|
||
b_idx + 1,
|
||
total_batches,
|
||
len(raw_obj),
|
||
len(batch),
|
||
)
|
||
|
||
for ck in keys:
|
||
if ck not in raw_obj:
|
||
continue
|
||
validated = raw_obj[ck]
|
||
src_row = key_to_row[ck]
|
||
extractions_by_row[str(src_row)] = validated
|
||
sqlite_rows.append((src_row, b_idx, ck, validated))
|
||
|
||
batch_details.append(
|
||
{
|
||
"batch_index": b_idx,
|
||
"keys": keys,
|
||
"source_rows": [key_to_row[k] for k in keys],
|
||
"model_output": raw_obj,
|
||
}
|
||
)
|
||
|
||
created = datetime.now(timezone.utc).isoformat()
|
||
db_path = DB_PATH.resolve()
|
||
if clean_databases:
|
||
_clean_voc_sqlite_databases(db_path)
|
||
conn = sqlite3.connect(db_path)
|
||
try:
|
||
init_sqlite_schema(conn)
|
||
cur = conn.execute(
|
||
"""
|
||
INSERT INTO analysis_jobs
|
||
(created_at, industry, product_name, source_file, batch_size, model, full_result_json)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
created,
|
||
industry,
|
||
product_name,
|
||
src,
|
||
batch_size_record,
|
||
MODEL_NAME,
|
||
"{}",
|
||
),
|
||
)
|
||
job_id = int(cur.lastrowid)
|
||
|
||
for src_row, b_idx, ck, payload in sqlite_rows:
|
||
conn.execute(
|
||
"""
|
||
INSERT INTO comment_extractions
|
||
(job_id, source_row, batch_index, comment_key, extraction_json)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
job_id,
|
||
src_row,
|
||
b_idx,
|
||
ck,
|
||
json.dumps(payload, ensure_ascii=False),
|
||
),
|
||
)
|
||
|
||
result: Dict[str, Any] = {
|
||
"job_id": job_id,
|
||
"sqlite_path": str(db_path),
|
||
"industry": industry,
|
||
"product_name": product_name,
|
||
"source_file": src,
|
||
"batching_mode": batching_mode,
|
||
"batch_size": batch_size_record,
|
||
"batch_plan": batch_plan,
|
||
"max_batch_input_tokens": max_batch_input_tokens,
|
||
"max_batch_output_tokens": max_batch_output_tokens,
|
||
"model": MODEL_NAME,
|
||
"extractions": extractions_by_row,
|
||
"batches": batch_details,
|
||
}
|
||
full_json = json.dumps(result, ensure_ascii=False)
|
||
conn.execute(
|
||
"UPDATE analysis_jobs SET full_result_json = ? WHERE id = ?",
|
||
(full_json, job_id),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
return result
|
||
|
||
|
||
def _run_smoke_test() -> None:
|
||
"""
|
||
真实调用 DashScope 的冒烟：5 条英文评论、batch_size=5、一次模型请求。
|
||
使用临时目录下的 SQLite，不写项目根 voc_structured.sqlite。
|
||
"""
|
||
import tempfile
|
||
|
||
global DB_PATH
|
||
td = Path(tempfile.mkdtemp(prefix="voc_smoke_"))
|
||
csv_p = td / "reviews.csv"
|
||
csv_p.write_text(
|
||
"review\n"
|
||
"Soft ramp; zipper broke after one week.\n"
|
||
"Shipping box was crushed but product fine.\n"
|
||
"My senior dog uses it daily; worth the price.\n"
|
||
"Smells a bit chemical at first, smell fades.\n"
|
||
"Great for cats too, very stable.\n",
|
||
encoding="utf-8",
|
||
)
|
||
saved_db = DB_PATH
|
||
DB_PATH = td / "smoke.sqlite"
|
||
try:
|
||
out = run_analysis(
|
||
industry="Pet supplies",
|
||
product_name="Pet bed ramp",
|
||
file_path=str(csv_p),
|
||
)
|
||
slim = {
|
||
"ok": True,
|
||
"job_id": out.get("job_id"),
|
||
"model": out.get("model"),
|
||
"sqlite_path": out.get("sqlite_path"),
|
||
"extraction_row_keys": sorted(out.get("extractions", {}).keys()),
|
||
"extractions": out.get("extractions"),
|
||
}
|
||
print(json.dumps(slim, ensure_ascii=False, indent=2))
|
||
finally:
|
||
DB_PATH = saved_db
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(
|
||
description=f"VOC 评论结构化（{MODEL_NAME} + SQLite）"
|
||
)
|
||
parser.add_argument("--industry", required=True, help="行业，如 Pet supplements")
|
||
parser.add_argument("--product", required=True, help="产品名")
|
||
parser.add_argument("--file", required=True, type=Path, help="评论 CSV/TXT 路径")
|
||
parser.add_argument(
|
||
"--batch-size",
|
||
type=int,
|
||
default=None,
|
||
help="固定每批条数；不指定则按评论长度与 token 预算动态分批",
|
||
)
|
||
parser.add_argument(
|
||
"--max-batch-input-tokens",
|
||
type=int,
|
||
default=DEFAULT_MAX_BATCH_INPUT_TOKENS,
|
||
help=f"单批最大输入 token 估算上限（模型上限约 {MODEL_MAX_INPUT_TOKENS}）",
|
||
)
|
||
parser.add_argument(
|
||
"--max-batch-output-tokens",
|
||
type=int,
|
||
default=DEFAULT_MAX_BATCH_OUTPUT_TOKENS,
|
||
help=f"单批最大输出 token 估算上限（模型上限约 {MODEL_MAX_OUTPUT_TOKENS}）",
|
||
)
|
||
parser.add_argument(
|
||
"--max-reviews-per-batch",
|
||
type=int,
|
||
default=BATCH_COUNT_MAX,
|
||
help=f"动态模式下每批最多条数（短评可接近此值，默认 {BATCH_COUNT_MAX}）",
|
||
)
|
||
parser.add_argument(
|
||
"-o",
|
||
"--output-json",
|
||
type=Path,
|
||
default=None,
|
||
help="可选：将完整结果 JSON 写入该文件",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
result = run_analysis(
|
||
industry=args.industry,
|
||
product_name=args.product,
|
||
file_path=str(args.file),
|
||
batch_size=args.batch_size,
|
||
max_batch_input_tokens=args.max_batch_input_tokens,
|
||
max_batch_output_tokens=args.max_batch_output_tokens,
|
||
max_reviews_per_batch=args.max_reviews_per_batch,
|
||
)
|
||
text = json.dumps(result, ensure_ascii=False, indent=2)
|
||
if args.output_json:
|
||
args.output_json.write_text(text, encoding="utf-8")
|
||
logger.info("结果已写入 %s", args.output_json)
|
||
print(text)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) >= 2 and sys.argv[1] == "--smoke":
|
||
_run_smoke_test()
|
||
else:
|
||
main()
|