506 lines
20 KiB
Python
506 lines
20 KiB
Python
"""使用外部 LLM (qwen3.6-int4-AWQ) 测试 product-design-module skill。
|
||
|
||
模拟多轮对话：LLM 作为 skill agent 提问，脚本模拟用户回答。
|
||
"""
|
||
|
||
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 skill 作为 system prompt。"""
|
||
skill_path = PROJECT_ROOT / "skills" / "product-design" / "product-design-module" / "SKILL.md"
|
||
with open(skill_path, "r", encoding="utf-8") as f:
|
||
return f.read()
|
||
|
||
|
||
def load_main_skill_prompt() -> str:
|
||
"""加载 product-design-main skill 作为补充 system prompt。"""
|
||
skill_path = PROJECT_ROOT / "skills" / "product-design" / "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 ""
|
||
|
||
|
||
# 用户模拟回答策略
|
||
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"{'='*60}\n")
|
||
|
||
# 构建 system prompt
|
||
module_skill = load_skill_prompt()
|
||
main_skill = load_main_skill_prompt()
|
||
|
||
system_prompt = f"""你是一个产品设计AI助手。请严格按照以下 skill 规范执行任务。
|
||
|
||
# 主流程 Skill（上下文参考）
|
||
{main_skill}
|
||
|
||
# 功能模块深挖 Skill（当前执行）
|
||
{module_skill}
|
||
|
||
---
|
||
重要提醒：
|
||
1. 你现在处于 Phase 1 步骤 2（功能深挖），项目类型和用户角色已在之前的步骤中确认
|
||
2. 请按照 product-design-module skill 的流程，通过多轮对话引导用户完成功能结构梳理
|
||
3. 每轮只问 2-4 个问题
|
||
4. 最终输出必须包含：三级模块结构 + KANO分类 + ICE评分 + MindMapNode JSON + 汇总表 + 完整树JSON
|
||
5. 不要在开头加 /think 或其他思考标记，直接输出对话内容
|
||
"""
|
||
|
||
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,
|
||
"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
|
||
|
||
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)
|
||
|
||
# 保存完整输出
|
||
output_dir = PROJECT_ROOT / "eval_output" / "product-design-module" / "llm_test"
|
||
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']}")
|
||
|
||
all_results.append(result)
|
||
|
||
# 汇总报告
|
||
print(f"\n\n{'='*60}")
|
||
print(f" 汇总报告")
|
||
print(f"{'='*60}\n")
|
||
|
||
report_lines = []
|
||
report_lines.append(f"# LLM 评估报告: product-design-module (模型: {MODEL})")
|
||
report_lines.append(f"\n生成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
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}% |"
|
||
)
|
||
|
||
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("")
|
||
|
||
# 检查详情
|
||
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("")
|
||
|
||
# 改进建议
|
||
report_lines.append("## Skill 优化建议")
|
||
report_lines.append("")
|
||
|
||
# 基于结果分析
|
||
avg_pass_rate = sum(r["eval_result"]["pass_rate"] for r in all_results) / len(all_results)
|
||
all_have_tree = all(r["process_eval"]["has_final_tree"] for r in all_results)
|
||
any_too_early = any(r["process_eval"]["too_early_output"] for r in all_results)
|
||
any_no_domain = any(not r["process_eval"]["has_domain_analysis"] for r in all_results)
|
||
|
||
if avg_pass_rate < 0.8:
|
||
report_lines.append("### 输出质量优化")
|
||
report_lines.append("- 检查通过率较低，LLM 可能未严格遵循 skill 的输出格式要求")
|
||
report_lines.append("- 建议在 skill 中增加更明确的输出格式约束和示例")
|
||
report_lines.append("- 建议在 system prompt 中强调关键格式要求（如 MindMapNode JSON 格式）")
|
||
report_lines.append("")
|
||
|
||
if any_too_early:
|
||
report_lines.append("### 对话流程优化")
|
||
report_lines.append("- LLM 可能在第一轮就输出了最终结果，跳过了多轮对话流程")
|
||
report_lines.append("- 建议在 skill 中更强调'必须通过多轮对话逐步引导'")
|
||
report_lines.append("- 建议添加明确的'不要在第一轮就输出最终结果'的约束")
|
||
report_lines.append("")
|
||
|
||
if any_no_domain:
|
||
report_lines.append("### 领域推理优化")
|
||
report_lines.append("- LLM 可能跳过了步骤 0 的领域推理")
|
||
report_lines.append("- 建议在 skill 开头更强调'必须先展示推理结果再提问'")
|
||
report_lines.append("")
|
||
|
||
if not all_have_tree:
|
||
report_lines.append("### 最终输出优化")
|
||
report_lines.append("- LLM 可能未输出完整的 final_tree JSON")
|
||
report_lines.append("- 建议在 skill 中增加 final_tree 输出的检查清单")
|
||
report_lines.append("- 建议添加'如果对话结束但未输出 final_tree，必须补充输出'的约束")
|
||
report_lines.append("")
|
||
|
||
# 通用建议
|
||
report_lines.append("### 通用建议")
|
||
report_lines.append("1. **Skill 长度优化**: 当前 skill 约 550 行，对较小模型可能上下文过长，建议提取核心约束为精简版")
|
||
report_lines.append("2. **格式强调**: 对 JSON 输出格式做更强的约束，避免 LLM 自由发挥导致格式错误")
|
||
report_lines.append("3. **流程检查点**: 在每个步骤末尾添加自检提示，让 LLM 确认是否完成了当前步骤")
|
||
report_lines.append("4. **示例精简**: 保留最关键的 1-2 个示例，减少 token 消耗")
|
||
|
||
# 保存报告
|
||
report_path = output_dir / "llm_eval_report.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,
|
||
"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.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" 测试完成!")
|
||
print(f" 模型: {MODEL}")
|
||
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 '无'}")
|
||
print(f"{'='*60}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|