夜雨聆风Stage 6: 生产化 — 从Demo到能上线的系统
你要学什么
前5步是让AI"能跑"。这一步是让AI"跑得稳"。真实生产环境和Demo的区别：
运行代码
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 entrydef summary(self):levels = {}for log in self.logs:levels[log[”level”]] = levels.get(log[”level”], 0) + 1return levels# Token计数器class TokenCounter:def __init__(self):self.total_input = 0self.total_output = 0self.query_count = 0def record(self, input_tokens, output_tokens):self.total_input += input_tokensself.total_output += output_tokensself.query_count += 1def estimate_cost(self):”””估算成本（GLM-4-Flash参考价）”””input_cost = self.total_input / 1_000_000 * 1.0output_cost = self.total_output / 1_000_000 * 1.0return {”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_timeanswer = str(response).strip()# 估算Token消耗est_input = len(transaction) * 2 + 500# 粗略估算est_output = len(answer) * 2token_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 = answerif 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.5if confidence < CONFIDENCE_THRESHOLD:logger.log(”WARN”, ”低置信度”, f”置信度{confidence:.2f}低于阈值{CONFIDENCE_THRESHOLD}”)result[”need_human”] = Trueresult[”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”] = Truelogger.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
cd /Users/salar/ai-risk-agent && python stage6_prod.py跑完后想一想
空输入和超长输入，系统怎么处理的？如果没有这些检查会怎样？ AI建议"冻结商户"，但系统拦截了——为什么不让AI自动执行？ LLM调用失败时，返回"中等风险"默认值——这个决策对吗？如果返回"低风险"呢？ 你能从日志级别分布看出什么？WARN和CRITICAL分别意味着什么？
总结下项目
你得知道DEMO级和生产究竟差多少，PM应该对生产一直抱有敬畏。你是这个产品的“父母”
说实话，DEMO级和生产的，比天差地别还大。但不是技术上的差距大，是思维方式的差距大。我用我熟悉的支付领域打个比方：
Demo和生产的核心区别是：Demo是"正常情况下能用"，生产是"任何情况下都不能出大事"。
举两个你刚才跑出来的真实例子：
例子1：超长输入 → AI建议冻结账户 Demo思维："哇，AI好聪明，还能识别异常商户名！" 生产思维："等等，它凭什么说冻结就冻结？它有证据吗？证据链完整吗？误冻结了商户投诉谁负责？"
所以生产环境必须有安全拦截层——AI可以建议冻结，但必须人工确认才能执行。这跟支付里的"风控系统触发高风险规则，必须人工审核才能冻结商户"是一个逻辑。
例子2：LLM调用失败 Demo思维："报错了，重启一下" 生产思维："如果现在是双11高峰期，LLM挂了怎么办？交易全卡住吗？"
所以生产环境必须有降级策略——LLM挂了就返回默认值（中等风险+转人工），宁可误报不能漏报，系统不能崩。这跟支付系统的"通道降级"是一个道理：主通道挂了切备通道，交易不能断。总结一下你现在应该记住的3句话
"RAG效果评估要分层看，检索准确率和生成准确率是两回事——小数据集上的高准确率很可能是虚高。" "Agent最大的挑战不是能不能调用工具，而是'什么时候该调用、什么时候不该调用'——AI的工具调用意愿是不稳定的，生产环境必须有强制调用和兜底机制。" "Demo和生产的差距不是功能多寡，而是容错能力——正常情况下都能跑，异常情况下谁不崩谁就是生产级。"
