2441c9119c
包含七步编排入口、结构化/向量化/聚类/词频/报告模块与 prompts 配置；忽略原始 CSV 与本地密钥。 Co-authored-by: Cursor <cursoragent@cursor.com>
320 lines
10 KiB
Python
320 lines
10 KiB
Python
"""
|
||
外置 Prompt 加载器：每次调用均从磁盘重读 .md / .yaml（无缓存）。
|
||
|
||
编辑入口：
|
||
prompts/config.yaml — 数值、中文词频分类（产品/运营可改）
|
||
prompts/schema.yaml — 锁定 JSON 字段与标记（改前需开发确认）
|
||
prompts/extraction/ — 结构化提取规则与示例
|
||
prompts/report/ — 报告静态规则
|
||
prompts/word_freq/ — 词频分类/解读
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Tuple
|
||
|
||
import yaml
|
||
|
||
PROMPTS_ROOT = Path(__file__).resolve().parent
|
||
|
||
|
||
def get_config() -> Dict[str, Any]:
|
||
with (PROMPTS_ROOT / "config.yaml").open(encoding="utf-8") as f:
|
||
return yaml.safe_load(f) or {}
|
||
|
||
|
||
def get_schema() -> Dict[str, Any]:
|
||
with (PROMPTS_ROOT / "schema.yaml").open(encoding="utf-8") as f:
|
||
return yaml.safe_load(f) or {}
|
||
|
||
|
||
def read_text(relative_path: str) -> str:
|
||
return (PROMPTS_ROOT / relative_path).read_text(encoding="utf-8")
|
||
|
||
|
||
def render(relative_path: str, **kwargs: Any) -> str:
|
||
return read_text(relative_path).format(**kwargs).strip()
|
||
|
||
|
||
def word_categories(config: Dict[str, Any] | None = None) -> Tuple[str, ...]:
|
||
cfg = config or get_config()
|
||
return tuple(cfg.get("word_categories") or ())
|
||
|
||
|
||
def report_markers(schema: Dict[str, Any] | None = None) -> Tuple[str, ...]:
|
||
sch = schema or get_schema()
|
||
return tuple(sch.get("report", {}).get("markers") or ())
|
||
|
||
|
||
def product_feedback_categories(schema: Dict[str, Any] | None = None) -> frozenset[str]:
|
||
sch = schema or get_schema()
|
||
cats = sch.get("extraction", {}).get("product_feedback_categories") or []
|
||
return frozenset(str(c) for c in cats)
|
||
|
||
|
||
def _categories_literal(schema: Dict[str, Any]) -> str:
|
||
cats = schema.get("extraction", {}).get("product_feedback_categories") or []
|
||
return ", ".join(f"'{c}'" for c in cats)
|
||
|
||
|
||
def _sentiments_literal(schema: Dict[str, Any]) -> str:
|
||
sent = schema.get("extraction", {}).get("sentiments") or []
|
||
return ", ".join(f"'{s}'" for s in sent)
|
||
|
||
|
||
def load_extraction_examples() -> List[Dict[str, Any]]:
|
||
with (PROMPTS_ROOT / "extraction/examples.yaml").open(encoding="utf-8") as f:
|
||
return yaml.safe_load(f) or []
|
||
|
||
|
||
def format_examples(examples: List[Dict[str, Any]] | None = None) -> str:
|
||
items = examples if examples is not None else load_extraction_examples()
|
||
lines: List[str] = []
|
||
for i, ex in enumerate(items, 1):
|
||
output_json = json.dumps(ex["output"], ensure_ascii=False, separators=(",", ":"))
|
||
lines.append(f"=== 示例 {i} ===")
|
||
lines.append(f"[教学说明]: {ex['instruction']}")
|
||
lines.append(f"[输入评论]: \"{ex['review']}\"")
|
||
lines.append(f"[输出 JSON]:\n{output_json}\n")
|
||
return "\n".join(lines).strip()
|
||
|
||
|
||
def _extraction_vars(product_name: str, schema: Dict[str, Any]) -> Dict[str, Any]:
|
||
return {
|
||
"product_name": product_name,
|
||
"categories_literal": _categories_literal(schema),
|
||
"sentiments_literal": _sentiments_literal(schema),
|
||
"examples_str": format_examples(),
|
||
}
|
||
|
||
|
||
def build_batch_extraction_system(
|
||
industry: str,
|
||
product_name: str,
|
||
*,
|
||
n_keys: int,
|
||
keys_literal: str,
|
||
) -> str:
|
||
schema = get_schema()
|
||
common = _extraction_vars(product_name, schema)
|
||
parts = [
|
||
render("extraction/system_intro.md", industry=industry, product_name=product_name),
|
||
render(
|
||
"extraction/batch_output_format.md",
|
||
n_keys=n_keys,
|
||
keys_literal=keys_literal,
|
||
),
|
||
render("extraction/field_rules.md", **common),
|
||
render("extraction/format_constraints_batch.md"),
|
||
render("extraction/filter_irrelevant.md", product_name=product_name),
|
||
render(
|
||
"extraction/examples_section.md",
|
||
product_name=product_name,
|
||
examples_str=common["examples_str"],
|
||
),
|
||
render("extraction/examples_footer_batch.md"),
|
||
]
|
||
return "\n\n".join(p for p in parts if p.strip())
|
||
|
||
|
||
def build_batch_extraction_user(
|
||
*,
|
||
n_keys: int,
|
||
keys_literal: str,
|
||
tagged_input: str,
|
||
) -> str:
|
||
return render(
|
||
"extraction/user_batch.md",
|
||
n_keys=n_keys,
|
||
keys_literal=keys_literal,
|
||
tagged_input=tagged_input,
|
||
)
|
||
|
||
|
||
def build_single_extraction_template(industry: str, product_name: str) -> str:
|
||
schema = get_schema()
|
||
common = _extraction_vars(product_name, schema)
|
||
parts = [
|
||
render("extraction/system_intro.md", industry=industry, product_name=product_name),
|
||
render("extraction/field_rules.md", **common),
|
||
render("extraction/format_constraints_single.md"),
|
||
render("extraction/filter_irrelevant.md", product_name=product_name),
|
||
render(
|
||
"extraction/examples_section.md",
|
||
product_name=product_name,
|
||
examples_str=common["examples_str"],
|
||
),
|
||
render("extraction/user_single_tail.md"),
|
||
]
|
||
return "\n\n".join(p for p in parts if p.strip())
|
||
|
||
|
||
def _report_marker_kwargs(
|
||
config: Dict[str, Any],
|
||
schema: Dict[str, Any],
|
||
) -> Dict[str, Any]:
|
||
markers = list(report_markers(schema))
|
||
wf = config.get("word_freq", {})
|
||
rp = config.get("report", {})
|
||
return {
|
||
"marker_word_zh": markers[0],
|
||
"marker_word_category": markers[1],
|
||
"marker_cluster_names": markers[2],
|
||
"marker_report_html": markers[3],
|
||
"wordcloud_top_n": wf.get("wordcloud_top_n", 180),
|
||
"word_category_classify_n": wf.get("word_category_classify_n", 180),
|
||
"outlier_label_zh": rp.get("outlier_label_zh", "未归类"),
|
||
}
|
||
|
||
|
||
def build_report_system() -> str:
|
||
return render("report/system.md")
|
||
|
||
|
||
def build_report_analysis_requirements(
|
||
*,
|
||
product_name: str,
|
||
stage_1_audience: str,
|
||
) -> str:
|
||
return render(
|
||
"report/analysis_requirements.md",
|
||
product_name=product_name,
|
||
stage_1_audience=stage_1_audience,
|
||
)
|
||
|
||
|
||
def build_report_output_format() -> str:
|
||
return render("report/output_format.md", **_report_marker_kwargs(get_config(), get_schema()))
|
||
|
||
|
||
def _word_category_definitions() -> str:
|
||
return read_text("word_freq/category_definitions.md").strip()
|
||
|
||
|
||
def build_report_json_markers(*, cats_literal: str) -> str:
|
||
return render(
|
||
"report/json_markers.md",
|
||
cats_literal=cats_literal,
|
||
category_definitions=_word_category_definitions(),
|
||
**_report_marker_kwargs(get_config(), get_schema()),
|
||
)
|
||
|
||
|
||
def build_report_correction_message(*, err_block: str, raw_body: str, cats_literal: str) -> str:
|
||
return render(
|
||
"report/correction.md",
|
||
err_block=err_block,
|
||
raw_body=raw_body,
|
||
cats_literal=cats_literal,
|
||
**_report_marker_kwargs(get_config(), get_schema()),
|
||
)
|
||
|
||
|
||
def build_word_assign_prompts(
|
||
*,
|
||
industry: str,
|
||
product_name: str,
|
||
word_lines: str,
|
||
cats_literal: str,
|
||
) -> Tuple[str, str]:
|
||
system = render(
|
||
"word_freq/assign_system.md",
|
||
cats_literal=cats_literal,
|
||
category_definitions=_word_category_definitions(),
|
||
)
|
||
user = render(
|
||
"word_freq/assign_user.md",
|
||
industry=industry,
|
||
product_name=product_name,
|
||
word_lines=word_lines,
|
||
cats_literal=cats_literal,
|
||
)
|
||
return system, user
|
||
|
||
|
||
def build_category_analysis_prompts(
|
||
*,
|
||
industry: str,
|
||
product_name: str,
|
||
category_lines: str,
|
||
cats_literal: str,
|
||
) -> Tuple[str, str]:
|
||
system = render("word_freq/analysis_system.md")
|
||
user = render(
|
||
"word_freq/analysis_user.md",
|
||
industry=industry,
|
||
product_name=product_name,
|
||
category_lines=category_lines,
|
||
cats_literal=cats_literal,
|
||
)
|
||
return system, user
|
||
|
||
|
||
def sync_voc_report_constants(voc_report: Any) -> None:
|
||
"""从 config.yaml / schema.yaml 同步 voc_report 模块级常量（每次 generate_report 前调用）。"""
|
||
cfg = get_config()
|
||
sch = get_schema()
|
||
wf = cfg.get("word_freq", {})
|
||
rp = cfg.get("report", {})
|
||
|
||
voc_report.WORDCLOUD_TOP_N = int(wf.get("wordcloud_top_n", 180))
|
||
voc_report.WORD_FREQ_TABLE_N = int(wf.get("word_freq_table_n", 180))
|
||
voc_report.WORD_CATEGORY_CLASSIFY_N = int(wf.get("word_category_classify_n", 180))
|
||
voc_report.WORD_FREQ_PAGE_SIZE = int(wf.get("word_freq_page_size", 30))
|
||
voc_report.WORD_FREQ_PAGES = int(wf.get("word_freq_pages", 6))
|
||
voc_report.WORD_CATEGORY_ASSIGN_BATCH_SIZE = int(
|
||
wf.get("word_category_assign_batch_size", 40)
|
||
)
|
||
voc_report.WORD_CATEGORY_ANALYSIS_MAX_WORDS = int(
|
||
wf.get("word_category_analysis_max_words", 25)
|
||
)
|
||
voc_report.OUTLIER_LABEL_ZH = str(rp.get("outlier_label_zh", "未归类"))
|
||
voc_report.MAX_PHRASES_PER_CLUSTER = int(rp.get("max_phrases_per_cluster", 15))
|
||
voc_report.WORD_CATEGORIES = word_categories(cfg)
|
||
voc_report.REPORT_MARKERS = report_markers(sch)
|
||
|
||
|
||
def validate_prompt_files() -> List[str]:
|
||
"""检查外置文件能否正常加载与渲染；返回错误列表（空=通过）。"""
|
||
errors: List[str] = []
|
||
try:
|
||
cfg = get_config()
|
||
sch = get_schema()
|
||
if not word_categories(cfg):
|
||
errors.append("config.yaml: word_categories 为空")
|
||
if len(report_markers(sch)) != 4:
|
||
errors.append("schema.yaml: report.markers 须为 4 项")
|
||
if not product_feedback_categories(sch):
|
||
errors.append("schema.yaml: extraction.product_feedback_categories 为空")
|
||
format_examples()
|
||
build_batch_extraction_system(
|
||
"Test Industry",
|
||
"Test Product",
|
||
n_keys=2,
|
||
keys_literal='"C1", "C2"',
|
||
)
|
||
build_batch_extraction_user(n_keys=2, keys_literal='"C1", "C2"', tagged_input="[C1] hi")
|
||
build_single_extraction_template("Test", "Product")
|
||
build_report_system()
|
||
build_report_analysis_requirements(
|
||
product_name="P",
|
||
stage_1_audience="1_audience",
|
||
)
|
||
build_report_output_format()
|
||
build_report_json_markers(cats_literal="成分/原料")
|
||
build_word_assign_prompts(
|
||
industry="I",
|
||
product_name="P",
|
||
word_lines="dog\t10",
|
||
cats_literal="成分/原料",
|
||
)
|
||
build_category_analysis_prompts(
|
||
industry="I",
|
||
product_name="P",
|
||
category_lines="- 成分/原料：a(1)",
|
||
cats_literal="成分/原料",
|
||
)
|
||
except Exception as e:
|
||
errors.append(f"渲染失败: {e}")
|
||
return errors
|