amz_review_analyse/voc_业务_2/prompt_loader.py
OnesvmWhoops cb6692c0c0 新增 voc_业务_2 报告流水线,并完善聚类与词频模块。
包含 Persona 锚定聚类、LLM 报告生成、方法论文档及 .gitignore 更新,便于在 Gitee 独立部署。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-06-15 17:12:39 +08:00

71 lines
2.1 KiB
Python

# -*- coding: utf-8 -*-
"""从 prompts.yaml 加载报告 LLM 提示词,供 llm_analyzer 使用。"""
from __future__ import annotations
import logging
from functools import lru_cache
from pathlib import Path
from typing import Any, Dict, Optional
import yaml
logger = logging.getLogger("voc.prompt_loader")
PROMPTS_FILE = Path(__file__).resolve().parent / "prompts.yaml"
def _render(template: str, **kwargs: Any) -> str:
"""将 {{key}} 替换为值;模板内 JSON 示例的花括号无需转义。"""
out = template
for key, val in kwargs.items():
out = out.replace("{{" + key + "}}", str(val))
return out
@lru_cache(maxsize=1)
def load_prompts(path: Optional[str] = None) -> Dict[str, Any]:
fp = Path(path) if path else PROMPTS_FILE
if not fp.is_file():
raise FileNotFoundError(f"提示词文件不存在: {fp}")
with fp.open(encoding="utf-8") as f:
data = yaml.safe_load(f) or {}
logger.debug("已加载提示词: %s", fp)
return data
def reload_prompts() -> None:
load_prompts.cache_clear()
def get_section(name: str, path: Optional[str] = None) -> Dict[str, Any]:
prompts = load_prompts(path)
sec = prompts.get(name)
if not sec:
raise KeyError(f"prompts.yaml 缺少段落: {name}")
return sec
def system_prompt(name: str, path: Optional[str] = None) -> str:
return (get_section(name, path).get("system") or "").strip()
def llm_params(name: str, path: Optional[str] = None) -> Dict[str, Any]:
sec = get_section(name, path)
return {
"temperature": float(sec.get("temperature", 0.3)),
"max_tokens": int(sec.get("max_tokens", 8000)),
}
def user_prompt(name: str, path: Optional[str] = None, **kwargs: Any) -> str:
tpl = (get_section(name, path).get("user_template") or "").strip()
return _render(tpl, **kwargs)
def optional_block(name: str, block_key: str, path: Optional[str] = None, **kwargs: Any) -> str:
"""加载可选子模板(如差评主题 extra 块)。"""
sec = get_section(name, path)
tpl = (sec.get(block_key) or "").strip()
if not tpl:
return ""
return _render(tpl, **kwargs)