cat > /Users/salar/ai-risk-agent/stage5_eval.py << 'PYEOF'import osimport jsonimport timefrom llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings, PromptTemplatefrom llama_index.core.node_parser import SentenceSplitterfrom llama_index.embeddings.huggingface import HuggingFaceEmbeddingfrom llama_index.llms.zhipuai import ZhipuAISettings.embed_model = HuggingFaceEmbedding(model_name=”/Users/salar/bge-small-zh-v1.5”)api_key = os.environ.get(”ZHIPUAI_API_KEY”)Settings.llm = ZhipuAI(model=”glm-4-flash”, temperature=0.1, api_key=api_key)# ====== 1. 构建知识库 ======print(”正在构建知识库...”)raw_docs = SimpleDirectoryReader(”/Users/salar/ai-risk-agent/docs”).load_data()nodes = SentenceSplitter(chunk_size=256, chunk_overlap=50).get_nodes_from_documents(raw_docs)index = VectorStoreIndex(nodes)# ====== 2. 测试数据集（有标准答案） ======# 这是效果评估的关键：你需要人工标注的”正确答案”来对比TEST_SET = [ { ”query”: ”借记卡单笔交易限额是多少”, ”expected_keywords”: [”5万元”, ”5万”, ”50000”], ”expected_category”: ”限额规则”, ”expected_risk_relevant”: True, }, { ”query”: ”退款率超过15%的商户会怎样”, ”expected_keywords”: [”C级”, ”降级”, ”人工审核”], ”expected_category”: ”商户风险”, ”expected_risk_relevant”: True, }, { ”query”: ”什么是拆分交易”, ”expected_keywords”: [”拆分”, ”规避限额”, ”小额”], ”expected_category”: ”异常识别”, ”expected_risk_relevant”: True, }, { ”query”: ”T+0和T+1结算有什么区别”, ”expected_keywords”: [”当天”, ”次日”, ”垫资”], ”expected_category”: ”结算”, ”expected_risk_relevant”: False, }, { ”query”: ”反洗钱大额交易报告的阈值”, ”expected_keywords”: [”5万元”, ”50万元”, ”1万美元”], ”expected_category”: ”反洗钱”, ”expected_risk_relevant”: True, }, { ”query”: ”VCC虚拟卡有什么限制”, ”expected_keywords”: [],# 知识库中没有VCC内容，期望AI拒绝回答 ”expected_category”: ”无”, ”expected_risk_relevant”: False, ”expect_no_answer”: True, },]# ====== 3. 检索效果评估 ======print(”\n” + ”=”*60)print(”【第一层：检索效果评估】”)print(”=”*60)retriever = index.as_retriever(similarity_top_k=3)retrieval_results = []for test in TEST_SET: nodes = retriever.retrieve(test[”query”])# 检查Top1是否属于预期类别 top1_category = ”其他” for n in nodes: cat = n.metadata.get(”category”, ”其他”) if cat != ”其他”: top1_category = cat break# 检查是否有预期类别的结果 has_expected = any(n.metadata.get(”category”) == test[”expected_category”] for n in nodes) retrieval_results.append({ ”query”: test[”query”], ”expected”: test[”expected_category”], ”actual_top1”: top1_category, ”has_expected_in_top3”: has_expected, ”top3_scores”: [round(n.score, 4) for n in nodes], }) status = ”✅” if has_expected else ”❌” print(f”\n {status} Q: {test['query']}”) print(f” 期望类别: {test['expected_category']} | Top1: {top1_category}”) print(f” Top3相似度: {[round(n.score, 4) for n in nodes]}”)# 计算检索指标retrieval_hit = sum(1 for r in retrieval_results if r[”has_expected_in_top3”])retrieval_precision = retrieval_hit / len(retrieval_results)print(f”\n 📊 检索命中率(Top3): {retrieval_hit}/{len(retrieval_results)} = {retrieval_precision:.1%}”)# ====== 4. 生成效果评估 ======print(”\n” + ”=”*60)print(”【第二层：生成效果评估】”)print(”=”*60)# 使用优化后的Prompteval_prompt = PromptTemplate( ”””你是一个支付风控助手。严格根据参考资料回答。要求：1. 如果参考资料中有答案，给出具体信息2. 如果参考资料中没有相关信息，明确说”根据现有资料无法回答”3. 不要编造任何内容参考资料：{context_str}问题：{query_str}回答：”””)engine = index.as_query_engine(similarity_top_k=3, text_qa_template=eval_prompt)generation_results = []for test in TEST_SET: start_time = time.time() response = engine.query(test[”query”]) latency = time.time() - start_time answer = str(response)# 检查关键词命中 if test.get(”expect_no_answer”):# 期望AI拒绝回答 correct = ”无法回答” in answer or ”未提及” in answer or ”没有” in answer hallucination = not correct else:# 期望AI给出包含关键词的回答 correct = any(kw in answer for kw in test[”expected_keywords”]) hallucination = False# 简化，实际需要人工标注 generation_results.append({ ”query”: test[”query”], ”answer”: answer[:150], ”correct”: correct, ”hallucination”: hallucination, ”latency”: round(latency, 2), }) status = ”✅” if correct else ”❌” hall_flag = ” 🚨幻觉” if hallucination else ”” print(f”\n {status}{hall_flag} Q: {test['query']}”) print(f” A: {answer[:120]}...”) print(f” 延迟: {latency:.2f}s”)# 计算生成指标accuracy = sum(1 for r in generation_results if r[”correct”]) / len(generation_results)hallucination_rate = sum(1 for r in generation_results if r[”hallucination”]) / len(generation_results)avg_latency = sum(r[”latency”] for r in generation_results) / len(generation_results)print(f”\n 📊 生成准确率: {accuracy:.1%}”)print(f” 📊 幻觉率: {hallucination_rate:.1%}”)print(f” 📊 平均延迟: {avg_latency:.2f}s”)# ====== 5. 产品指标评估 ======print(”\n” + ”=”*60)print(”【第三层：产品指标评估】”)print(”=”*60)# Token成本估算# GLM-4-Flash: 输入≈1元/百万Token, 输出≈1元/百万Token（参考价）# 每次查询大约消耗：输入500Token + 输出200TokenINPUT_TOKENS_PER_QUERY = 500OUTPUT_TOKENS_PER_QUERY = 200COST_PER_MILLION_INPUT = 1.0# 元COST_PER_MILLION_OUTPUT = 1.0# 元daily_queries = 1000# 假设每日1000次查询monthly_queries = daily_queries * 30daily_input_cost = daily_queries * INPUT_TOKENS_PER_QUERY / 1_000_000 * COST_PER_MILLION_INPUTdaily_output_cost = daily_queries * OUTPUT_TOKENS_PER_QUERY / 1_000_000 * COST_PER_MILLION_OUTPUTmonthly_cost = (daily_input_cost + daily_output_cost) * 30print(f”\n 📊 成本估算（基于GLM-4-Flash）：”)print(f” - 每次查询Token消耗: 输入~{INPUT_TOKENS_PER_QUERY} + 输出~{OUTPUT_TOKENS_PER_QUERY}”)print(f” - 日查询量假设: {daily_queries}”)print(f” - 日LLM成本: ¥{daily_input_cost + daily_output_cost:.2f}”)print(f” - 月LLM成本: ¥{monthly_cost:.2f}”)print(f” - Embedding成本: ¥0（本地模型）”)print(f”\n 📊 延迟统计：”)p50 = sorted(r[”latency”] for r in generation_results)[len(generation_results)//2]p95 = sorted(r[”latency”] for r in generation_results)[int(len(generation_results)*0.95)]print(f” - P50延迟: {p50:.2f}s”)print(f” - P95延迟: {p95:.2f}s”)print(f” - 风控场景可接受延迟: <3s（当前{'达标' if p95 < 3 else '需优化'}）”)# ====== 6. 评估报告总结 ======print(”\n” + ”=”*60)print(”【评估报告总结】”)print(”=”*60)print(f””” ┌─────────────────────────────────────┐ │ 检索命中率 (Top3): {retrieval_precision:.0%} │ │ 生成准确率: {accuracy:.0%} │ │ 幻觉率: {hallucination_rate:.0%} │ │ P50延迟: {p50:.2f}s │ │ 月度LLM成本: ¥{monthly_cost:.0f} │ └─────────────────────────────────────┘ 评估结论： {'✅ 可进入灰度测试' if accuracy >= 0.8 else '⚠️ 需要优化后再测试'} {'✅ 幻觉率可接受' if hallucination_rate <= 0.1 else '🚨 幻觉率过高，需加强Prompt约束'} {'✅ 延迟达标' if p95 < 3 else '⚠️ 延迟偏高，需优化检索或换小模型'}”””)print(”💡 思考：”)print(” 1. 如果把测试集从6条增加到100条，准确率可能怎样变化？”)print(” 2. 幻觉率0%可能吗？如果不可能，产品上怎么兜底？”)print(” 3. 如果日查询量从1000增到100000，成本会怎样？需要优化什么？”)print(” 4. 评估不是一次性的，是持续的过程——这就是'数据飞轮'”)PYEOF

cd /Users/salar/ai-risk-agent && python stage5_eval.py

用户提问 → 检索模块 → 找出相关文档 → 把文档+问题一起塞给LLM → LLM生成答案 ↑ ↑ ↑ 检索准确率 召回了正确文档吗？ 它真的看了文档吗？