skills-eval/run_llm_eval_v2.py

535 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""使用外部 LLM (qwen3.6-int4-AWQ, 32B) 测试 product-design-module-v2 skill。
模拟多轮对话LLM 作为 skill agent 提问,脚本模拟用户回答。
与 v1 版本的区别:使用优化后的 v2 skill 作为 system prompt。
"""
import json
import re
import time
import sys
from pathlib import Path
from openai import OpenAI
# 配置
API_URL = "http://js2.blockelite.cn:17865/v1"
API_KEY = "sk-sdajioqhdsakljhdjiuwhdqi"
MODEL = "qwen3.6-int4-AWQ"
PROJECT_ROOT = Path(__file__).parent
sys.path.insert(0, str(PROJECT_ROOT))
from eval_framework.checker import run_checks, load_custom_checks
def create_client():
return OpenAI(base_url=API_URL, api_key=API_KEY)
def load_skill_prompt() -> str:
"""加载 product-design-module-v2 skill 作为 system prompt。"""
skill_path = PROJECT_ROOT / "skills" / "product-design" / "product-design-module-v2" / "SKILL.md"
with open(skill_path, "r", encoding="utf-8") as f:
return f.read()
def chat_with_llm(client, messages: list, max_tokens: int = 4096) -> str:
"""调用 LLM API。"""
try:
response = client.chat.completions.create(
model=MODEL,
messages=messages,
max_tokens=max_tokens,
temperature=0.7,
)
return response.choices[0].message.content
except Exception as e:
print(f" [API Error] {e}")
return ""
# 用户模拟回答策略(与 v1 相同)
USER_RESPONSES = {
"工单系统": {
"initial": "我要做一个企业工单管理系统员工可以提交IT报修和设备申请技术人员接单处理管理员可以看到所有工单的数据报表。目前团队没有在用任何工单系统。",
"followups": [
"就是B端工具型。主要角色有三个普通员工提交工单技术人员处理工单管理员看报表和配置系统。目前从零开始做没有旧系统。",
"工单类型主要是IT报修和设备申请可能后续会加行政服务类。需要SLA时效比如紧急工单2小时内响应。移动端暂时不需要先做PC端。",
"差不多就这些,帮我汇总输出吧。",
"确认,没问题。",
]
},
"电商社区": {
"initial": "我要做一个类似得物的潮流电商平台,卖潮流服饰和球鞋,面向年轻人,要有社区功能让用户分享穿搭",
"followups": [
"是C端的交易和社区都要有。用户主要是18-30岁的年轻人喜欢潮流文化。参考得物和小红书。",
"商品主要是球鞋和潮流服饰,需要先鉴别后发货。社区主要是穿搭分享,用户可以发图文。需要商家后台管理商品和订单。",
"就这样吧,帮我输出功能结构。",
"确认。",
]
},
"OA协同": {
"initial": "我们公司内部用的一套老旧OA系统要升级换代需要支持员工请假、报销、出差申请部门经理审批HR能看到所有流程的报表。系统还要支持移动端。",
"followups": [
"是B端OA协同产品。角色有普通员工、部门经理、HR、系统管理员。现在是替代旧系统旧系统太老了不好用。",
"请假、报销、出差是核心流程。审批需要支持多级审批比如金额超过5000要总监批。移动端主要是方便经理随时审批。",
"够了,先这样,帮我输出结果吧。",
"确认保存。",
]
},
}
def run_test(client, test_name: str, user_responses: dict, max_rounds: int = 8) -> dict:
"""运行一次完整的 skill 测试。"""
print(f"\n{'='*60}")
print(f" 测试: {test_name}")
print(f" 模型: {MODEL}")
print(f" Skill: v2 (优化版)")
print(f"{'='*60}\n")
# 构建 system prompt — v2 版本只加载 v2 skill不加载 main skill
module_skill = load_skill_prompt()
system_prompt = f"""你是一个产品设计AI助手。请严格按照以下 skill 规范执行任务。
# 功能模块深挖 Skill当前执行
{module_skill}
---
重要提醒:
1. 你现在处于功能深挖阶段,项目类型和用户角色已在之前的步骤中确认
2. 请按照 skill 的流程,通过多轮对话引导用户完成功能结构梳理
3. 每轮只问 2-4 个问题
4. 最终输出必须包含:三级模块结构 + KANO分类 + ICE评分 + MindMapNode JSON + 汇总表
5. 不要在开头加 /think 或其他思考标记,直接输出对话内容
6. 严格遵守"输出纪律":禁止输出推理过程、禁止思维链、只输出用户需要看到的内容
"""
messages = [
{"role": "system", "content": system_prompt},
]
# 第一轮:用户输入
user_msg = user_responses["initial"]
messages.append({"role": "user", "content": user_msg})
print(f" [用户] {user_msg[:80]}...")
followup_idx = 0
conversation_log = []
round_num = 0
final_output = ""
has_final_tree = False
while round_num < max_rounds and not has_final_tree:
round_num += 1
print(f"\n --- 第 {round_num} 轮 ---")
# 调用 LLM
start_time = time.time()
assistant_msg = chat_with_llm(client, messages)
elapsed = time.time() - start_time
if not assistant_msg:
print(f" [LLM] 空响应,跳过")
continue
messages.append({"role": "assistant", "content": assistant_msg})
# 记录对话
conversation_log.append({
"round": round_num,
"role": "assistant",
"content": assistant_msg[:500] + ("..." if len(assistant_msg) > 500 else ""),
"full_length": len(assistant_msg),
"elapsed": elapsed
})
print(f" [LLM] ({len(assistant_msg)} chars, {elapsed:.1f}s)")
# 打印前200字
preview = assistant_msg[:200].replace("\n", " ")
print(f" 预览: {preview}...")
# 检查是否包含 final_tree
if '"final_tree"' in assistant_msg or '"action"' in assistant_msg:
has_final_tree = True
final_output = assistant_msg
print(f" [检测到 final_tree 输出!]")
break
# 检查是否用户说了"够了"之类的收敛信号
if followup_idx >= len(user_responses["followups"]):
# 没有更多预设回答,发送收敛信号
user_msg = "差不多了,帮我汇总输出最终结果吧。"
else:
user_msg = user_responses["followups"][followup_idx]
followup_idx += 1
messages.append({"role": "user", "content": user_msg})
print(f" [用户] {user_msg[:80]}...")
# 如果没有 final_tree再请求一次
if not has_final_tree:
print(f"\n [额外轮] 请求最终输出...")
messages.append({"role": "user", "content": "请帮我输出完整的功能结构包含三级模块、KANO分类、ICE评分、MindMapNode JSON和汇总表。"})
assistant_msg = chat_with_llm(client, messages, max_tokens=8192)
if assistant_msg:
messages.append({"role": "assistant", "content": assistant_msg})
final_output = assistant_msg
if '"final_tree"' in assistant_msg or '"action"' in assistant_msg:
has_final_tree = True
conversation_log.append({
"round": round_num + 1,
"role": "assistant",
"content": assistant_msg[:500],
"full_length": len(assistant_msg),
"elapsed": 0
})
return {
"test_name": test_name,
"model": MODEL,
"skill_version": "v2",
"rounds": round_num,
"has_final_tree": has_final_tree,
"final_output_length": len(final_output),
"conversation_log": conversation_log,
"final_output": final_output,
}
def evaluate_output(test_result: dict, eval_case_idx: int) -> dict:
"""对 LLM 输出运行确定性检查。"""
output = test_result["final_output"]
if not output:
return {"pass_rate": 0.0, "checks": [], "error": "无输出"}
# 加载对应的 eval expectations
eval_path = PROJECT_ROOT / "evals" / "product-design-module" / "evals.json"
with open(eval_path, "r", encoding="utf-8") as f:
evals = json.load(f)
if eval_case_idx >= len(evals["evals"]):
return {"pass_rate": 0.0, "checks": [], "error": "eval case 索引越界"}
expectations = evals["evals"][eval_case_idx]["expectations"]
custom_checks = load_custom_checks("product-design-module")
results = run_checks(output, expectations, custom_checks)
passed = sum(1 for r in results if r.passed)
total = len(results)
check_details = []
for r in results:
check_details.append({
"expectation": r.expectation,
"passed": r.passed,
"reason": r.reason
})
return {
"pass_rate": passed / total if total > 0 else 0.0,
"passed": passed,
"total": total,
"checks": check_details,
}
def evaluate_process(test_result: dict) -> dict:
"""评估多轮对话过程质量。"""
log = test_result["conversation_log"]
rounds = test_result["rounds"]
process_eval = {
"total_rounds": rounds,
"has_final_tree": test_result["has_final_tree"],
"assistant_messages": len([l for l in log if l["role"] == "assistant"]),
"avg_response_length": sum(l["full_length"] for l in log) / max(len(log), 1),
}
# 检查是否遵循了多轮对话流程
first_msg = log[0]["content"] if log else ""
has_domain_analysis = any(kw in first_msg for kw in ["产品类型", "B端", "C端", "初步分析", "领域"])
has_questions = any("" in l["content"] or "?" in l["content"] for l in log if l["role"] == "assistant")
has_summary_table = any("汇总" in l["content"] or "模块" in l["content"] for l in log if l["role"] == "assistant")
process_eval["has_domain_analysis"] = has_domain_analysis
process_eval["has_questions"] = has_questions
process_eval["has_summary_table"] = has_summary_table
# 检查是否过早输出第1轮就输出最终结果
if rounds == 1 and test_result["has_final_tree"]:
process_eval["too_early_output"] = True
else:
process_eval["too_early_output"] = False
# v2 新增:检查思维泄露
thinking_leak_count = 0
thinking_keywords = ["根据 skill", "根据规范", "我需要", "让我分析", "第一步", "第二步", "思维链", "按照流程"]
for l in log:
if l["role"] == "assistant":
for kw in thinking_keywords:
if kw in l["content"]:
thinking_leak_count += 1
break
process_eval["thinking_leak_count"] = thinking_leak_count
process_eval["thinking_leak_ratio"] = thinking_leak_count / max(len([l for l in log if l["role"] == "assistant"]), 1)
return process_eval
def main():
client = create_client()
# 先测试 API 连通性
print("测试 API 连通性...")
try:
test_msg = chat_with_llm(client, [
{"role": "system", "content": "你是一个助手。"},
{"role": "user", "content": "你好,请回复'连接成功'"}
], max_tokens=50)
print(f" API 响应: {test_msg[:50]}")
except Exception as e:
print(f" API 连接失败: {e}")
return
# 测试配置
tests = [
("工单系统", "工单系统", 0),
("电商社区", "电商社区", 1),
("OA协同", "OA协同", 4),
]
all_results = []
for test_name, response_key, eval_idx in tests:
user_responses = USER_RESPONSES[response_key]
result = run_test(client, test_name, user_responses)
# 保存完整输出 — v2 输出到独立目录
output_dir = PROJECT_ROOT / "eval_output" / "product-design-module" / "llm_test_v2"
output_dir.mkdir(parents=True, exist_ok=True)
output_path = output_dir / f"{test_name}_output.md"
with open(output_path, "w", encoding="utf-8") as f:
f.write(result["final_output"])
print(f"\n 输出已保存: {output_path}")
# 运行确定性检查
print(f"\n 运行确定性检查...")
eval_result = evaluate_output(result, eval_idx)
result["eval_result"] = eval_result
print(f" 通过率: {eval_result['pass_rate']*100:.1f}% ({eval_result.get('passed', 0)}/{eval_result.get('total', 0)})")
for c in eval_result.get("checks", []):
status = "PASS" if c["passed"] else "FAIL"
print(f" [{status}] {c['expectation'][:60]}")
if not c["passed"]:
print(f" 原因: {c['reason'][:80]}")
# 评估过程质量
process_eval = evaluate_process(result)
result["process_eval"] = process_eval
print(f"\n 过程评估:")
print(f" 对话轮数: {process_eval['total_rounds']}")
print(f" 有领域分析: {process_eval['has_domain_analysis']}")
print(f" 有提问: {process_eval['has_questions']}")
print(f" 有汇总表: {process_eval['has_summary_table']}")
print(f" 过早输出: {process_eval['too_early_output']}")
print(f" 最终有 final_tree: {process_eval['has_final_tree']}")
print(f" 思维泄露次数: {process_eval['thinking_leak_count']}")
print(f" 思维泄露比例: {process_eval['thinking_leak_ratio']*100:.1f}%")
all_results.append(result)
# 汇总报告
print(f"\n\n{'='*60}")
print(f" 汇总报告 (v2)")
print(f"{'='*60}\n")
report_lines = []
report_lines.append(f"# LLM 评估报告: product-design-module-v2 (模型: {MODEL}, 32B)")
report_lines.append(f"\n生成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
report_lines.append(f"\nSkill 版本: v2优化版213 行 vs 原 551 行)")
report_lines.append("")
# 总体结果
report_lines.append("## 总体结果")
report_lines.append("")
report_lines.append("| 测试 | 对话轮数 | final_tree | 检查通过率 | 过程评分 | 思维泄露率 |")
report_lines.append("|------|---------|------------|-----------|---------|-----------|")
for r in all_results:
eval_r = r["eval_result"]
proc = r["process_eval"]
# 过程评分:领域分析+提问+汇总表+不过早输出+有final_tree
process_score = sum([
proc["has_domain_analysis"],
proc["has_questions"],
proc["has_summary_table"],
not proc["too_early_output"],
proc["has_final_tree"],
]) / 5 * 100
report_lines.append(
f"| {r['test_name']} | {proc['total_rounds']} | "
f"{'' if proc['has_final_tree'] else ''} | "
f"{eval_r['pass_rate']*100:.1f}% ({eval_r.get('passed', 0)}/{eval_r.get('total', 0)}) | "
f"{process_score:.0f}% | "
f"{proc['thinking_leak_ratio']*100:.0f}% |"
)
report_lines.append("")
# 各测试详细结果
for r in all_results:
report_lines.append(f"## {r['test_name']}")
report_lines.append("")
report_lines.append(f"- 对话轮数: {r['process_eval']['total_rounds']}")
report_lines.append(f"- 输出长度: {r['final_output_length']} 字符")
report_lines.append(f"- 有 final_tree: {r['process_eval']['has_final_tree']}")
report_lines.append(f"- 有领域分析: {r['process_eval']['has_domain_analysis']}")
report_lines.append(f"- 有提问交互: {r['process_eval']['has_questions']}")
report_lines.append(f"- 有汇总表: {r['process_eval']['has_summary_table']}")
report_lines.append(f"- 过早输出: {r['process_eval']['too_early_output']}")
report_lines.append(f"- 思维泄露次数: {r['process_eval']['thinking_leak_count']}")
report_lines.append(f"- 思维泄露比例: {r['process_eval']['thinking_leak_ratio']*100:.1f}%")
report_lines.append("")
# 检查详情
eval_r = r["eval_result"]
report_lines.append("### 确定性检查结果")
report_lines.append("")
report_lines.append("| # | 期望条件 | 结果 | 原因 |")
report_lines.append("|---|---------|------|------|")
for i, c in enumerate(eval_r.get("checks", [])):
status = "PASS" if c["passed"] else "FAIL"
exp = c["expectation"][:50] + ("..." if len(c["expectation"]) > 50 else "")
reason = c["reason"][:60] + ("..." if len(c["reason"]) > 60 else "")
report_lines.append(f"| {i+1} | {exp} | {status} | {reason} |")
report_lines.append("")
# 对话摘要
report_lines.append("### 对话过程摘要")
report_lines.append("")
for log_entry in r["conversation_log"]:
role = log_entry["role"]
content_preview = log_entry["content"][:150].replace("\n", " ")
report_lines.append(f"- **Round {log_entry['round']} ({role})**: {content_preview}...")
report_lines.append("")
# v1 vs v2 对比
report_lines.append("## v1 vs v2 对比")
report_lines.append("")
# 尝试加载 v1 结果
v1_json_path = PROJECT_ROOT / "eval_output" / "product-design-module" / "llm_test" / "llm_eval_result.json"
if v1_json_path.exists():
with open(v1_json_path, "r", encoding="utf-8") as f:
v1_data = json.load(f)
report_lines.append("| 指标 | v1 | v2 | 变化 |")
report_lines.append("|------|----|----|------|")
# 计算汇总
v1_avg_pass = sum(t["eval_result"]["pass_rate"] for t in v1_data["tests"]) / len(v1_data["tests"])
v2_avg_pass = sum(r["eval_result"]["pass_rate"] for r in all_results) / len(all_results)
v1_avg_rounds = sum(t["rounds"] for t in v1_data["tests"]) / len(v1_data["tests"])
v2_avg_rounds = sum(r["process_eval"]["total_rounds"] for r in all_results) / len(all_results)
v1_has_tree = sum(1 for t in v1_data["tests"] if t["has_final_tree"])
v2_has_tree = sum(1 for r in all_results if r["process_eval"]["has_final_tree"])
v1_total_checks = sum(t["eval_result"]["total"] for t in v1_data["tests"])
v1_passed_checks = sum(t["eval_result"]["passed"] for t in v1_data["tests"])
v2_total_checks = sum(r["eval_result"]["total"] for r in all_results)
v2_passed_checks = sum(r["eval_result"]["passed"] for r in all_results)
def trend(v1, v2):
diff = v2 - v1
if diff > 0:
return f"+{diff:.1f}"
return f"{diff:.1f}"
report_lines.append(f"| 平均检查通过率 | {v1_avg_pass*100:.1f}% | {v2_avg_pass*100:.1f}% | {trend(v1_avg_pass*100, v2_avg_pass*100)} |")
report_lines.append(f"| 总检查项通过 | {v1_passed_checks}/{v1_total_checks} | {v2_passed_checks}/{v2_total_checks} | {trend(v1_passed_checks, v2_passed_checks)} |")
report_lines.append(f"| 平均对话轮数 | {v1_avg_rounds:.1f} | {v2_avg_rounds:.1f} | {trend(v1_avg_rounds, v2_avg_rounds)} |")
report_lines.append(f"| 有 final_tree | {v1_has_tree}/3 | {v2_has_tree}/3 | {trend(v1_has_tree, v2_has_tree)} |")
# 思维泄露对比
v2_avg_leak = sum(r["process_eval"]["thinking_leak_ratio"] for r in all_results) / len(all_results)
report_lines.append(f"| 思维泄露率 | N/A | {v2_avg_leak*100:.0f}% | 新增指标 |")
report_lines.append("")
else:
report_lines.append("v1 结果文件不存在,无法对比。")
report_lines.append("")
# 优化效果分析
report_lines.append("## 优化效果分析")
report_lines.append("")
report_lines.append("### v2 Skill 核心改动")
report_lines.append("1. **输出纪律**:新增章节明确禁止输出推理过程、思维链,只输出用户需要看到的内容")
report_lines.append("2. **格式模板化**:提供精确的 Markdown 模板和 JSON 模板,用 `[placeholder]` 标记可变部分")
report_lines.append("3. **自检清单**10 项检查项,输出前逐项验证")
report_lines.append("4. **精简示例**:从 6 个产品类型的详细 L3 示例精简为仅列出 L1 模块方向")
report_lines.append("5. **行数精简**:从 551 行精简到 213 行(减少 61%")
report_lines.append("6. **强化格式约束**:明确 `### 3.X` + `#### 3.X.Y` 层级、`ICEI×C×E=总分` 格式")
report_lines.append("")
# 保存报告
report_path = output_dir / "llm_eval_report_v2.md"
with open(report_path, "w", encoding="utf-8") as f:
f.write("\n".join(report_lines))
print(f"\n 报告已保存: {report_path}")
# 保存 JSON 结果
json_result = {
"model": MODEL,
"skill_version": "v2",
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S'),
"tests": []
}
for r in all_results:
test_data = {
"test_name": r["test_name"],
"rounds": r["rounds"],
"has_final_tree": r["has_final_tree"],
"final_output_length": r["final_output_length"],
"eval_result": {
"pass_rate": r["eval_result"]["pass_rate"],
"passed": r["eval_result"].get("passed", 0),
"total": r["eval_result"].get("total", 0),
"checks": r["eval_result"].get("checks", []),
},
"process_eval": r["process_eval"],
}
json_result["tests"].append(test_data)
json_path = output_dir / "llm_eval_result_v2.json"
with open(json_path, "w", encoding="utf-8") as f:
json.dump(json_result, f, ensure_ascii=False, indent=2)
print(f" JSON 结果已保存: {json_path}")
# 打印汇总
print(f"\n{'='*60}")
print(f" v2 测试完成!")
print(f" 模型: {MODEL} (32B)")
print(f" Skill: v2 (213 行)")
for r in all_results:
eval_r = r["eval_result"]
proc = r["process_eval"]
print(f" {r['test_name']}: 通过率 {eval_r['pass_rate']*100:.1f}%, "
f"轮数 {proc['total_rounds']}, "
f"final_tree {'' if proc['has_final_tree'] else ''}, "
f"思维泄露 {proc['thinking_leak_ratio']*100:.0f}%")
print(f"{'='*60}")
if __name__ == "__main__":
main()