From e3c6853eac1bdb41a3ccd2c643b5ea06c3196f2a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9E=97=E5=98=89=E7=83=A8?= Date: Fri, 16 Jan 2026 19:34:12 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=A0=E9=99=A4=20smart=5Fagent.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- smart_agent.py | 236 ------------------------------------------------- 1 file changed, 236 deletions(-) delete mode 100644 smart_agent.py diff --git a/smart_agent.py b/smart_agent.py deleted file mode 100644 index f1bfc37..0000000 --- a/smart_agent.py +++ /dev/null @@ -1,236 +0,0 @@ -import pandas as pd -import joblib -import json -import re - -# ========================================== -# 0. 基础工具类定义 -# ========================================== -class BaseTool: - def __init__(self, name, description): - self.name = name - self.description = description - - def run(self, *args, **kwargs): - raise NotImplementedError - -# ========================================== -# 1. 工具实现 -# ========================================== - -class MLPredictionTool(BaseTool): - """ - 工具 1: 机器学习预测工具 - 功能: 加载预训练模型，预测客户转化概率，并提供特征归因。 - """ - def __init__(self, artifact_path='model_artifacts.pkl'): - super().__init__("ML_Predictor", "输入客户特征，输出购买概率和关键影响因素") - print(f"[{self.name}] 正在加载模型资产...") - self.artifacts = joblib.load(artifact_path) - self.model = self.artifacts['model'] - self.encoders = self.artifacts['encoders'] - self.feature_meta = self.artifacts['feature_meta'] - # 获取特征重要性，用于简单的归因解释 - self.feature_importances = pd.Series( - self.model.feature_importances_, - index=self.feature_meta['all_cols'] - ).sort_values(ascending=False) - - def preprocess(self, customer_data): - df = pd.DataFrame([customer_data]) - if 'duration' in df.columns: df = df.drop('duration', axis=1) - - for col, le in self.encoders.items(): - if col in df.columns: - try: - df[col] = le.transform(df[col]) - except: - df[col] = 0 # 简单处理未知值 - - # 补齐可能缺失的列（全0填充）并保持顺序 - for col in self.feature_meta['all_cols']: - if col not in df.columns: - df[col] = 0 - - return df[self.feature_meta['all_cols']] - - def run(self, customer_data): - # 1. 预处理 - X = self.preprocess(customer_data) - - # 2. 预测 - prob = float(self.model.predict_proba(X)[0][1]) # 强制转换为 python float - - # 3. 归因 (Attribution) - # 简单逻辑：找出该客户数据中，属于 Top 5 重要特征的字段及其值 - top_features = self.feature_importances.head(5).index.tolist() - attribution = {feat: customer_data.get(feat, 'N/A') for feat in top_features} - - return { - "probability": round(prob, 4), - "risk_level": "High" if prob < 0.3 else ("Medium" if prob < 0.7 else "Low"), - "key_factors": attribution - } - -class StrategyRetrievalTool(BaseTool): - """ - 工具 2: 策略检索工具 - 功能: 根据客户分群或意向等级，检索对应的营销话术和产品包。 - """ - def __init__(self): - super().__init__("Strategy_Retriever", "根据意向分检索营销策略") - # 模拟向量数据库或规则库 - self.knowledge_base = { - "High_Intent": { - "segment": "VIP_Growth", - "channel": "Personal_Call", - "product": "大额存单/结构性存款", - "script_template": "尊贵的{name}，鉴于您良好的{key_factor}，我们要为您推荐专属..." - }, - "Medium_Intent": { - "segment": "Potential_Saver", - "channel": "SMS_Web", - "product": "灵活理财/定投", - "script_template": "你好，发现您对{key_factor}感兴趣，这里有一份理财攻略..." - }, - "Low_Intent": { - "segment": "General_Mass", - "channel": "Email", - "product": "货币基金/新人礼包", - "script_template": "本月财经快讯：如何打理您的零钱..." - } - } - - def run(self, probability): - if probability > 0.7: - key = "High_Intent" - elif probability > 0.4: - key = "Medium_Intent" - else: - key = "Low_Intent" - - return self.knowledge_base[key] - -# ========================================== -# 2. Agent 定义 (Orchestrator) -# ========================================== - -class SalesAgent: - def __init__(self): - self.tools = { - "predictor": MLPredictionTool(), - "retriever": StrategyRetrievalTool() - } - - def mock_llm_inference(self, prompt): - """ - 模拟 LLM 的生成能力。 - 在真实场景中，这里调用 openai.ChatCompletion.create(model="gpt-4", messages=...) - """ - # 从 Prompt 中解析 Context - # 这是一个 Mock，所以我们用正则或简单的逻辑把 Prompt 里的信息“反刍”出来 - # 实际上 LLM 会进行语义理解和润色 - - # 提取关键信息用于 Mock 输出 - try: - context_str = re.search(r"【Context】(.*?)【Instruction】", prompt, re.S).group(1) - context = json.loads(context_str) - - pred_result = context['prediction'] - strategy_result = context['strategy'] - customer_info = context['customer_raw'] - - # 模拟 LLM 生成话术 - script = strategy_result['script_template'].format( - name="客户", - key_factor=list(pred_result['key_factors'].keys())[0] - ) - - response = { - "thought_process": f"模型预测概率为 {pred_result['probability']}，属于 {strategy_result['segment']} 客群。已检索到对应策略，建议通过 {strategy_result['channel']} 触达。", - "final_decision": { - "action": strategy_result['channel'], - "product_recommendation": strategy_result['product'], - "personalized_script": script, - "attribution_explanation": f"预测模型显示该客户成交概率为 {pred_result['probability']*100}%，主要受 {json.dumps(pred_result['key_factors'], ensure_ascii=False)} 等因素影响。" - } - } - return json.dumps(response, ensure_ascii=False, indent=2) - except Exception as e: - return json.dumps({"error": f"LLM Mock Failed: {str(e)}"}) - - def process_request(self, customer_data): - print(f"\n[Agent] 收到新请求: {customer_data.get('job', 'Unknown')} | {customer_data.get('age')}岁") - - # --- Step 1: 调用 ML 工具进行预测 --- - print(f"[Agent] 调用工具: {self.tools['predictor'].name} ...") - pred_result = self.tools['predictor'].run(customer_data) - print(f" >>> 预测结果: 概率={pred_result['probability']}, 关键因素={list(pred_result['key_factors'].keys())}") - - # --- Step 2: 调用 检索工具获取策略 --- - print(f"[Agent] 调用工具: {self.tools['retriever'].name} ...") - strategy_result = self.tools['retriever'].run(pred_result['probability']) - print(f" >>> 检索结果: 渠道={strategy_result['channel']}, 产品={strategy_result['product']}") - - # --- Step 3: LLM 整合信息 --- - print(f"[Agent] 请求 LLM 进行最终决策与生成...") - - # 构建 Context - context = { - "customer_raw": customer_data, - "prediction": pred_result, - "strategy": strategy_result - } - - prompt = f""" - 你是一个智能营销助手。请根据以下上下文信息，生成结构化的营销建议。 - - 【Context】 - {json.dumps(context, ensure_ascii=False)} - - 【Instruction】 - 1. 解释模型预测结果。 - 2. 结合策略库，生成具体的话术。 - 3. 输出 JSON 格式。 - """ - - final_output = self.mock_llm_inference(prompt) - return final_output - -# ========================================== -# 3. 主程序入口 -# ========================================== -if __name__ == "__main__": - # 1. 准备数据 - df_raw = pd.read_csv('bank.csv') - - # 2. 初始化 Agent - agent = SalesAgent() - - # 3. 模拟场景 - print("\n" + "="*60) - print("场景演示: Agent 协调多个工具完成决策") - print("="*60) - - # 场景 A: 低意向客户 - customer_a = df_raw.iloc[1].to_dict() # 假设这是低概率 - if 'deposit' in customer_a: del customer_a['deposit'] - - result_a = agent.process_request(customer_a) - print("\n[Agent Final Output]") - print(result_a) - - print("-" * 60) - - # 场景 B: 高意向客户 (人工构造) - customer_b = customer_a.copy() - customer_b.update({ - 'poutcome': 'success', - 'duration': 1000, # 注意工具内部会移除 duration，这里只是模拟输入 - 'contact': 'cellular', - 'month': 'oct' - }) - - result_b = agent.process_request(customer_b) - print("\n[Agent Final Output]") - print(result_b)