91e6c47fc0
统一 voc_llm 密钥解析与默认模型；向量化改为本地 mlx 模型；更新 README、gitignore 与流水线文档。 Co-authored-by: Cursor <cursoragent@cursor.com>
139 lines
4 KiB
Python
139 lines
4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Prompt 验收脚本（修改 prompts/ 后运行）。
|
||
|
||
./310py/bin/python prompts/smoke.py # 仅校验文件加载与渲染（无需 API Key）
|
||
./310py/bin/python prompts/smoke.py --live # 调用模型跑 smoke/reviews.yaml（需 DEEPSEEK_API_KEY）
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
if str(PROJECT_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
import yaml
|
||
|
||
from prompts.loader import (
|
||
build_batch_extraction_system,
|
||
build_batch_extraction_user,
|
||
get_schema,
|
||
product_feedback_categories,
|
||
validate_prompt_files,
|
||
)
|
||
|
||
|
||
def _load_smoke_reviews() -> dict:
|
||
path = Path(__file__).resolve().parent / "smoke/reviews.yaml"
|
||
with path.open(encoding="utf-8") as f:
|
||
return yaml.safe_load(f) or {}
|
||
|
||
|
||
def run_dry() -> int:
|
||
errors = validate_prompt_files()
|
||
if errors:
|
||
print("❌ Prompt 文件校验失败：")
|
||
for e in errors:
|
||
print(f" - {e}")
|
||
return 1
|
||
smoke = _load_smoke_reviews()
|
||
industry = smoke.get("industry", "Test")
|
||
product = smoke.get("product_name", "Test")
|
||
reviews = smoke.get("reviews") or []
|
||
keys = [str(r["id"]) for r in reviews]
|
||
tagged = "\n".join(f"[{r['id']}] {r['text']}" for r in reviews)
|
||
keys_literal = ", ".join(json.dumps(k) for k in keys)
|
||
system = build_batch_extraction_system(
|
||
industry,
|
||
product,
|
||
n_keys=len(keys),
|
||
keys_literal=keys_literal,
|
||
)
|
||
user = build_batch_extraction_user(
|
||
n_keys=len(keys),
|
||
keys_literal=keys_literal,
|
||
tagged_input=tagged,
|
||
)
|
||
print("✅ 外置 Prompt 加载与渲染通过")
|
||
print(f" 示例评论数: {len(reviews)}")
|
||
print(f" system 长度: {len(system)} 字符")
|
||
print(f" user 长度: {len(user)} 字符")
|
||
return 0
|
||
|
||
|
||
def run_live() -> int:
|
||
rc = run_dry()
|
||
if rc != 0:
|
||
return rc
|
||
try:
|
||
from 结构化_server import _call_dashscope_chat, _validate_extraction_strict
|
||
except ImportError as e:
|
||
print(f"❌ 无法导入结构化_server: {e}")
|
||
return 1
|
||
|
||
smoke = _load_smoke_reviews()
|
||
industry = smoke["industry"]
|
||
product = smoke["product_name"]
|
||
reviews = smoke["reviews"]
|
||
keys = [str(r["id"]) for r in reviews]
|
||
tagged = "\n".join(f"[{r['id']}] {r['text']}" for r in reviews)
|
||
keys_literal = ", ".join(json.dumps(k) for k in keys)
|
||
system = build_batch_extraction_system(
|
||
industry,
|
||
product,
|
||
n_keys=len(keys),
|
||
keys_literal=keys_literal,
|
||
)
|
||
user = build_batch_extraction_user(
|
||
n_keys=len(keys),
|
||
keys_literal=keys_literal,
|
||
tagged_input=tagged,
|
||
)
|
||
print("⏳ 调用模型进行 smoke 结构化…")
|
||
try:
|
||
raw = _call_dashscope_chat(
|
||
[{"role": "system", "content": system}, {"role": "user", "content": user}],
|
||
max_tokens=8192,
|
||
)
|
||
except Exception as e:
|
||
print(f"❌ 模型调用失败: {e}")
|
||
return 1
|
||
|
||
if not isinstance(raw, dict):
|
||
print(f"❌ 期望 JSON 对象，得到: {type(raw).__name__}")
|
||
return 1
|
||
|
||
ok = 0
|
||
for k in keys:
|
||
if k not in raw:
|
||
print(f"❌ 缺少键 {k}")
|
||
continue
|
||
try:
|
||
_validate_extraction_strict(raw[k], k)
|
||
ok += 1
|
||
print(f"✅ {k} 校验通过")
|
||
except ValueError as e:
|
||
print(f"❌ {k} 校验失败: {e}")
|
||
|
||
schema_cats = product_feedback_categories(get_schema())
|
||
print(f" 锁定 category 枚举: {', '.join(sorted(schema_cats))}")
|
||
return 0 if ok == len(keys) else 1
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="Prompt smoke 验收")
|
||
parser.add_argument(
|
||
"--live",
|
||
action="store_true",
|
||
help="调用 DashScope 跑 smoke 评论（需 API Key）",
|
||
)
|
||
args = parser.parse_args()
|
||
return run_live() if args.live else run_dry()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|