cat > /Users/salar/ai-risk-agent/stage6_prod.py << 'PYEOF'import osimport jsonimport timefrom datetime import datetimefrom 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. 生产化风控助手 ======# 日志系统class RiskLogger: def __init__(self): self.logs = [] def log(self, level, action, detail, merchant_id=None, txn_id=None): entry = { ”timestamp”: datetime.now().isoformat(), ”level”: level,# INFO / WARN / ERROR / CRITICAL ”action”: action, ”detail”: detail, ”merchant_id”: merchant_id, ”txn_id”: txn_id, } self.logs.append(entry) emoji = {”INFO”: ”ℹ️”, ”WARN”: ”⚠️”, ”ERROR”: ”❌”, ”CRITICAL”: ”🚨”}.get(level, ”📋”) print(f” {emoji} [{level}] {action}: {detail}”) return entry def summary(self): levels = {} for log in self.logs: levels[log[”level”]] = levels.get(log[”level”], 0) + 1 return levels# Token计数器class TokenCounter: def __init__(self): self.total_input = 0 self.total_output = 0 self.query_count = 0 def record(self, input_tokens, output_tokens): self.total_input += input_tokens self.total_output += output_tokens self.query_count += 1 def estimate_cost(self): ”””估算成本（GLM-4-Flash参考价）””” input_cost = self.total_input / 1_000_000 * 1.0 output_cost = self.total_output / 1_000_000 * 1.0 return {”input_cost”: input_cost, ”output_cost”: output_cost, ”total_cost”: input_cost + output_cost}logger = RiskLogger()token_counter = TokenCounter()# 安全策略DANGEROUS_ACTIONS = [”block_merchant”, ”freeze_account”, ”refund”]MAX_DAILY_QUERIES = 10000# 每日查询上限CONFIDENCE_THRESHOLD = 0.7# 置信度低于此值需人工复核HALLUCINATION_PATTERNS = [”我认为”, ”我猜测”, ”可能是”, ”大概”]# 疑似幻觉的模式词def production_risk_assessment(transaction: str) -> dict: ”””生产级风控评估：带完整异常处理和安全检查””” logger.log(”INFO”, ”开始评估”, f”交易: {transaction[:80]}”)# --- 安全检查1：速率限制 --- if token_counter.query_count >= MAX_DAILY_QUERIES: logger.log(”WARN”, ”速率限制”, f”已达日查询上限{MAX_DAILY_QUERIES}”) return {”risk_level”: ”无法判断”, ”reason”: ”已达日查询上限，请明日再试”, ”need_human”: True}# --- 安全检查2：输入验证 --- if not transaction or len(transaction) < 5: logger.log(”ERROR”, ”输入无效”, ”交易信息过短”) return {”risk_level”: ”无法判断”, ”reason”: ”交易信息不完整”, ”need_human”: True}# --- 执行RAG查询 --- try: start_time = time.time() engine = index.as_query_engine(similarity_top_k=3) response = engine.query( f”作为风控助手，请评估此交易的风险：{transaction}\n” f”请输出JSON：{{\”risk_level\”: \”低/中/高/极高\”, ” f”\”risk_factors\”: [...], \”action\”: \”...\”, \”confidence\”: 0.0-1.0}}” ) latency = time.time() - start_time answer = str(response).strip()# 估算Token消耗 est_input = len(transaction) * 2 + 500# 粗略估算 est_output = len(answer) * 2 token_counter.record(est_input, est_output) logger.log(”INFO”, ”LLM响应”, f”延迟{latency:.2f}s, 估算Token: 输入{est_input}/输出{est_output}”) except Exception as e: logger.log(”ERROR”, ”LLM调用失败”, str(e))# 兜底方案：返回保守的默认结果 return { ”risk_level”: ”中”, ”reason”: ”AI评估服务异常，默认中等风险”, ”action”: ”转人工审核”, ”need_human”: True, ”error”: str(e), }# --- 解析输出 ---# 去掉markdown标记 clean_answer = answer if clean_answer.startswith(”```”): clean_answer = clean_answer.split(”\n”, 1)[1] if ”\n” in clean_answer else clean_answer[3:] clean_answer = clean_answer.rsplit(”```”, 1)[0] try: result = json.loads(clean_answer) except json.JSONDecodeError: logger.log(”WARN”, ”输出解析失败”, ”AI未输出有效JSON，尝试正则提取”)# 兜底：尝试从自然语言中提取风险等级 result = {”risk_level”: ”中”, ”raw_output”: answer} if ”高” in answer or ”极高” in answer: result[”risk_level”] = ”高” elif ”低” in answer: result[”risk_level”] = ”低”# --- 安全检查3：幻觉检测 --- hallucination_detected = any(pattern in answer for pattern in HALLUCINATION_PATTERNS) if hallucination_detected: logger.log(”WARN”, ”疑似幻觉”, ”AI输出包含不确定表达”) result[”hallucination_flag”] = True# --- 安全检查4：置信度检查 --- confidence = result.get(”confidence”, 0.5) if isinstance(confidence, str): try: confidence = float(confidence) except ValueError: confidence = 0.5 if confidence < CONFIDENCE_THRESHOLD: logger.log(”WARN”, ”低置信度”, f”置信度{confidence:.2f}低于阈值{CONFIDENCE_THRESHOLD}”) result[”need_human”] = True result[”action”] = f”[需人工复核] {result.get('action', '')}”# --- 安全检查5：高风险操作拦截 --- action = result.get(”action”, ””) if any(da in action for da in DANGEROUS_ACTIONS): logger.log(”CRITICAL”, ”高风险操作”, f”AI建议: {action}”) result[”action”] = f”[需人工确认] {action}” result[”need_human”] = True logger.log(”CRITICAL”, ”安全拦截”, ”高风险操作已标记为需人工确认，AI不可自动执行”)# --- 延迟检查 --- if latency > 5.0: logger.log(”WARN”, ”延迟过高”, f”响应时间{latency:.2f}s超过5s阈值”) result[”latency_warning”] = True# 添加元信息 result[”latency”] = round(latency, 2) result[”token_estimate”] = {”input”: est_input, ”output”: est_output} result[”assessment_time”] = datetime.now().isoformat() return result# ====== 3. 测试生产化系统 ======print(”\n” + ”=”*60)print(”【生产化风控助手测试】”)print(”=”*60)test_cases = [ {”desc”: ”正常交易”, ”info”: ”商户M001，借记卡500元，正常营业时间”}, {”desc”: ”可疑交易”, ”info”: ”商户M002，贷记卡48000元，凌晨3点，退款率18%”}, {”desc”: ”高风险交易”, ”info”: ”商户M003，连续5笔4999元，疑似拆分交易”}, {”desc”: ”空输入”, ”info”: ””},# 测试异常处理 {”desc”: ”超长输入”, ”info”: ”商户” + ”A” * 5000},# 测试输入验证]for case in test_cases: print(f”\n{'─'*50}”) print(f”💳 场景: {case['desc']}”) result = production_risk_assessment(case[”info”]) print(f”\n 📋 评估结果：”) for k, v in result.items(): if k not in [”raw_output”]: print(f” - {k}: {v}”)# ====== 4. 生成运营报告 ======print(”\n\n” + ”=”*60)print(”【运营报告】”)print(”=”*60)cost = token_counter.estimate_cost()log_summary = logger.summary()print(f””” 📊 本次测试统计： - 总查询数: {token_counter.query_count} - 估算Token消耗: 输入{token_counter.total_input} + 输出{token_counter.total_output} - 估算成本: ¥{cost['total_cost']:.4f} 📋 日志级别分布: - INFO: {log_summary.get('INFO', 0)} - WARN: {log_summary.get('WARN', 0)} - ERROR: {log_summary.get('ERROR', 0)} - CRITICAL: {log_summary.get('CRITICAL', 0)} 🔒 安全事件: - 高风险操作拦截: {log_summary.get('CRITICAL', 0)}次 - 低置信度告警: {sum(1 for l in logger.logs if '低置信度' in l.get('detail', ''))}次 - 疑似幻觉标记: {sum(1 for l in logger.logs if '幻觉' in l.get('detail', ''))}次 💡 生产化关键决策： 1. 置信度<0.7 → 必须人工复核（不信任AI的模糊判断） 2. 冻结/退款操作 → 必须人工确认（AI只能建议，不能执行） 3. LLM调用失败 → 返回”中等风险”默认值（宁可误报不漏报） 4. 输出解析失败 → 正则提取兜底（不崩就行） 5. 延迟>5s → 标记告警（可能影响用户体验）”””)print(”💡 这就是AI PM的'最后一公里'：Demo能跑不等于能上线。”)print(”💡 生产化思维：错误兜底 > 功能完善，安全拦截 > 智能输出，可观测 > 黑盒运行”)PYEOF