anomaly_recognizer_agent.py 47 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098
  1. from langgraph.prebuilt import create_react_agent
  2. from langchain_openai import ChatOpenAI
  3. from typing import Dict, List, Any, Optional
  4. import pandas as pd
  5. import json
  6. from datetime import datetime
  7. from pathlib import Path
  8. import numpy as np
  9. from llmops.agents.tools.balance_info_missing_recognizer import BalanceInfoMissingRecognizer
  10. from llmops.agents.tools.inactive_account_recognizer import InactiveAccountRecognizer
  11. from llmops.agents.tools.balance_recognizer import BalanceContinuityRecognizer
  12. from llmops.agents.tools.night_transaction_recognizer import NightTransactionRecognizer
  13. from llmops.agents.tools.high_frequency_transaction_recognizer import HighFrequencyTransactionRecognizer
  14. from llmops.agents.tools.large_amount_transaction_recognizer import LargeAmountTransactionRecognizer
  15. from llmops.agents.tools.occasional_high_integer_transaction_recognizer import OccasionalHighIntegerTransactionRecognizer
  16. from llmops.agents.tools.low_interest_rate_recognizer import LowInterestRateRecognizer
  17. from llmops.agents.tools.over_book_transaction_recognizer import OverBookTransactionRecognizer
  18. from llmops.agents.data_manager import DataManager
  19. from llmops.config import LLM_API_KEY, LLM_BASE_URL, LLM_MODEL_NAME, anomaly_recognizer_config
  20. class AnomalyRecognitionAgent:
  21. """异常识别智能体"""
  22. def __init__(self, csv_path: str, api_key: str, base_url: str = "https://api.deepseek.com",
  23. model_name: str = "deepseek-chat", config: Optional[Dict] = None):
  24. """
  25. 初始化异常识别智能体
  26. """
  27. self.csv_path = csv_path
  28. self.llm = ChatOpenAI(
  29. model=model_name,
  30. api_key=api_key,
  31. base_url=base_url,
  32. temperature=0.1
  33. )
  34. self.config = config or {}
  35. self.transaction_data = None
  36. self.data_summary = {}
  37. self.recognizer_tools = []
  38. self.agent = None
  39. self.recognition_results = {}
  40. # 初始化识别工具
  41. self._initialize_recognizers()
  42. # 如果提供了LLM,初始化Agent
  43. self._initialize_agent()
  44. def _initialize_recognizers(self):
  45. """初始化所有异常识别工具"""
  46. # 余额信息缺失检查
  47. if self.config.get('enable_balance_missing_check', True):
  48. balance_missing_config = self.config.get('balance_missing_check', {})
  49. self.recognizer_tools.append(BalanceInfoMissingRecognizer(
  50. csv_path=self.csv_path,
  51. config={'balance_missing_check': balance_missing_config}
  52. ))
  53. print(f"✅ 初始化余额信息缺失检查器(高优先级)")
  54. # 长期无交易账户识别器
  55. if self.config.get('enable_inactive_account_check', True):
  56. inactive_account_config = self.config.get('inactive_account_check', {})
  57. self.recognizer_tools.append(InactiveAccountRecognizer(
  58. csv_path=self.csv_path,
  59. config={'inactive_account_check': inactive_account_config}
  60. ))
  61. print(f"✅ 初始化长期无交易账户识别器(高优先级)")
  62. # 余额连续性识别
  63. if self.config.get('enable_balance_recognition', True):
  64. self.recognizer_tools.append(BalanceContinuityRecognizer(csv_path=self.csv_path))
  65. print(f"✅ 初始化余额连续性识别器")
  66. # 夜间交易识别
  67. if self.config.get('enable_night_recognition', True):
  68. night_config = self.config.get('night_recognition', {})
  69. self.recognizer_tools.append(NightTransactionRecognizer(
  70. csv_path=self.csv_path,
  71. config={'night_transaction': night_config}
  72. ))
  73. print(f"✅ 初始化夜间交易识别器")
  74. # 高频交易识别
  75. if self.config.get('enable_high_frequency_recognition', True):
  76. high_freq_config = self.config.get('high_frequency_recognition', {})
  77. self.recognizer_tools.append(HighFrequencyTransactionRecognizer(
  78. csv_path=self.csv_path,
  79. config={'high_frequency': high_freq_config}
  80. ))
  81. print(f"✅ 初始化高频交易识别器")
  82. # 大额交易识别
  83. if self.config.get('enable_large_amount_recognition', True):
  84. large_amount_recognition_config = self.config.get('large_amount_recognition', {})
  85. self.recognizer_tools.append(LargeAmountTransactionRecognizer(
  86. csv_path=self.csv_path,
  87. config={'large_amount_recognition': large_amount_recognition_config}
  88. ))
  89. print(f"✅ 初始化大额交易识别器")
  90. # 偶发大额整数交易识别
  91. if self.config.get('enable_occasional_high_integer_recognition', True):
  92. integer_config = self.config.get('occasional_high_integer_transaction', {})
  93. self.recognizer_tools.append(OccasionalHighIntegerTransactionRecognizer(
  94. csv_path=self.csv_path,
  95. config={'occasional_high_integer_transaction': integer_config}
  96. ))
  97. print(f"✅ 初始化偶发高额整数交易识别器")
  98. # 结算交易识别
  99. if self.config.get('enable_low_interest_rate_recognition', True):
  100. interest_config = self.config.get('low_interest_rate_recognition', {})
  101. self.recognizer_tools.append(LowInterestRateRecognizer(
  102. csv_path=self.csv_path,
  103. config={'interest_rate_check': interest_config}
  104. ))
  105. print(f"✅ 初始化低利率结息识别器(高优先级)")
  106. # 疑似过账交易识别
  107. if self.config.get('enable_over_book_transaction_recognition', True): # 使用 "over_book" 而不是 "overbook"
  108. overbook_config = self.config.get('over_book_transaction_recognition', {}) # 保持一致
  109. self.recognizer_tools.append(OverBookTransactionRecognizer(
  110. csv_path=self.csv_path,
  111. config={'over_book_transaction_recognition': overbook_config} # 保持一致
  112. ))
  113. print(f"✅ 初始化疑似过账流水交易识别器")
  114. print(f"📋 共初始化 {len(self.recognizer_tools)} 个识别器")
  115. def _initialize_agent(self):
  116. """初始化智能体 - 优化版本"""
  117. try:
  118. # 确保每个工具都有清晰的描述
  119. for tool in self.recognizer_tools:
  120. # 如果描述太短,添加说明
  121. if len(tool.description) < 30:
  122. tool.description = f"分析银行流水数据中的{tool.display_name}"
  123. # 创建Agent
  124. self.agent = create_react_agent(
  125. model=self.llm,
  126. tools=self.recognizer_tools
  127. )
  128. print("🤖 异常识别智能体初始化成功")
  129. print(f"🛠️ 加载了 {len(self.recognizer_tools)} 个工具:")
  130. for i, tool in enumerate(self.recognizer_tools, 1):
  131. print(f" {i}. {tool.display_name} ({tool.name})")
  132. print(f" 描述: {tool.description}")
  133. # 测试工具是否可用
  134. print("🧪 测试工具可用性...")
  135. for tool in self.recognizer_tools:
  136. try:
  137. # 测试工具的基本属性
  138. has_run = hasattr(tool, '_run')
  139. has_name = hasattr(tool, 'name')
  140. has_desc = hasattr(tool, 'description')
  141. print(
  142. f" ✓ {tool.name}: 接口完整" if all([has_run, has_name, has_desc]) else f" ⚠️ {tool.name}: 接口不完整")
  143. except:
  144. print(f" ❌ {tool.name}: 测试失败")
  145. except Exception as e:
  146. print(f"智能体初始化失败: {e}")
  147. import traceback
  148. traceback.print_exc()
  149. self.agent = None
  150. def load_transaction_data(self) -> pd.DataFrame:
  151. """加载交易数据"""
  152. try:
  153. print(f"📥 正在加载交易数据: {self.csv_path}")
  154. self.transaction_data = DataManager.load_from_standardized_csv(self.csv_path)
  155. self.data_summary = self._generate_data_summary()
  156. return self.transaction_data
  157. except Exception as e:
  158. print(f"数据加载失败: {e}")
  159. raise
  160. def _generate_data_summary(self) -> Dict[str, Any]:
  161. """生成数据摘要"""
  162. if self.transaction_data is None or len(self.transaction_data) == 0:
  163. return {}
  164. df = self.transaction_data
  165. summary = {
  166. 'transaction_count': len(df),
  167. 'date_range': {
  168. 'start': df['txDate'].min() if 'txDate' in df.columns else '未知',
  169. 'end': df['txDate'].max() if 'txDate' in df.columns else '未知'
  170. },
  171. 'total_amount': float(df['txAmount'].sum()) if 'txAmount' in df.columns else 0,
  172. 'income_amount': float(df[df['txDirection'] == '收入']['txAmount'].sum())
  173. if 'txAmount' in df.columns and 'txDirection' in df.columns else 0,
  174. 'expense_amount': float(df[df['txDirection'] == '支出']['txAmount'].sum())
  175. if 'txAmount' in df.columns and 'txDirection' in df.columns else 0,
  176. 'average_amount': float(df['txAmount'].mean()) if 'txAmount' in df.columns else 0,
  177. 'max_amount': float(df['txAmount'].max()) if 'txAmount' in df.columns else 0,
  178. 'min_amount': float(df['txAmount'].min()) if 'txAmount' in df.columns else 0,
  179. 'unique_days': df['datetime'].dt.date.nunique() if 'datetime' in df.columns else 0,
  180. 'direction_distribution': df['txDirection'].value_counts().to_dict()
  181. if 'txDirection' in df.columns else {}
  182. }
  183. return summary
  184. def execute_full_recognition(self) -> Dict[str, Any]:
  185. """执行完整异常识别"""
  186. if self.transaction_data is None:
  187. raise ValueError("请先加载交易数据")
  188. print("🔍 开始执行银行流水异常识别...")
  189. # 清空之前的结果
  190. self.recognition_results = {
  191. 'agent_results': None,
  192. 'direct_results': None,
  193. 'all_anomalies': [],
  194. 'summary': {}
  195. }
  196. # 先执行直接识别
  197. # try:
  198. # direct_results = self._execute_direct_recognition()
  199. # self.recognition_results['direct_results'] = direct_results
  200. # except Exception as e:
  201. # print(f"⚠️ 直接异常识别失败: {e}")
  202. # 执行Agent识别(如果可用)
  203. if self.agent:
  204. try:
  205. agent_results = self._execute_agent_recognition()
  206. self.recognition_results['agent_results'] = agent_results
  207. except Exception as e:
  208. print(f"⚠️ Agent异常识别失败: {e}")
  209. else:
  210. print("⚠️ Agent未初始化,跳过Agent识别")
  211. # 合并所有识别的异常
  212. self._consolidate_anomalies()
  213. # 生成识别摘要
  214. self._generate_recognition_summary()
  215. print("✅ 异常识别完成")
  216. return self.recognition_results
  217. def _execute_direct_recognition(self) -> Dict[str, Any]:
  218. """执行直接异常识别"""
  219. print("🚀 开始直接异常识别...")
  220. results = {}
  221. all_anomalies = []
  222. for recognizer in self.recognizer_tools:
  223. try:
  224. print(f" 🔍 执行 {recognizer.display_name}...")
  225. # 不传入任何参数,让识别器使用初始化时的csv_path
  226. result = recognizer._run()
  227. results[recognizer.display_name] = result
  228. # 处理结果
  229. if isinstance(result, str):
  230. # 字符串结果
  231. print(f" 📝 {recognizer.display_name}: {result[:100]}...")
  232. elif isinstance(result, dict):
  233. # 字典结果
  234. if 'identified_anomalies' in result:
  235. for anomaly in result['identified_anomalies']:
  236. anomaly['recognition_type'] = recognizer.display_name
  237. all_anomalies.append(anomaly)
  238. anomaly_count = result.get('identified_count', 0)
  239. status = result.get('recognition_status', '未知')
  240. print(f" ✅ {recognizer.display_name}: 识别完成,发现 {anomaly_count} 条异常 ({status})")
  241. except Exception as e:
  242. error_msg = f"{recognizer.display_name} 识别失败: {e}"
  243. print(f" ❌ {error_msg}")
  244. return {
  245. 'results': results,
  246. 'all_anomalies': all_anomalies,
  247. 'total_recognizers': len(self.recognizer_tools),
  248. 'completed_recognizers': len(results)
  249. }
  250. def _execute_agent_recognition(self) -> Dict[str, Any]:
  251. """执行Agent异常识别"""
  252. print("🤖 开始智能体异常识别...")
  253. try:
  254. agent_results = self.recognition_results.get('agent_results', {})
  255. if agent_results and 'all_anomalies' in agent_results:
  256. for anomaly in agent_results['all_anomalies']:
  257. if anomaly.get('check_type') == 'balance_info_missing':
  258. balance_missing_alert = f"""
  259. ⚠️ **重要提示**:
  260. 检测到数据完整性异常:银行流水缺少余额信息字段!
  261. 这会影响以下分析的准确性:
  262. 1. 余额连续性检查(可能无法执行)
  263. 2. 资金存量波动分析
  264. 3. 交易与余额的匹配验证
  265. 请在分析时考虑这一限制条件。
  266. """
  267. break
  268. # 准备工具信息
  269. tools_info = self._prepare_tools_info_for_prompt()
  270. # 生成通用提示词
  271. prompt = self._generate_universal_prompt(tools_info)
  272. # 创建初始状态
  273. initial_state = {
  274. "messages": [
  275. {
  276. "role": "system",
  277. "content": self._get_universal_system_prompt()
  278. },
  279. {
  280. "role": "user",
  281. "content": prompt
  282. }
  283. ]
  284. }
  285. print("🔄 正在执行Agent...")
  286. print("📋 提示词已发送:")
  287. print("-" * 50)
  288. print(prompt[:500] + "..." if len(prompt) > 500 else prompt)
  289. print("-" * 50)
  290. # 执行代理
  291. result = self.agent.invoke(initial_state)
  292. print(f"✅ Agent执行完成,共 {len(result['messages'])} 条消息")
  293. # 处理结果
  294. agent_output = self._process_agent_result(result)
  295. # 如果没有调用工具,尝试备用方案
  296. if len(agent_output['tool_calls']) == 0:
  297. print("⚠️ Agent没有调用工具,启动备用方案...")
  298. backup_result = self._execute_backup_recognition()
  299. agent_output['all_anomalies'].extend(backup_result['all_anomalies'])
  300. agent_output['backup_used'] = True
  301. print(f"🤖 最终统计: {len(agent_output['tool_calls'])} 次工具调用, {len(agent_output['all_anomalies'])} 条异常")
  302. return agent_output
  303. except Exception as e:
  304. error_msg = f"Agent识别执行失败: {str(e)}"
  305. print(f"❌ {error_msg}")
  306. import traceback
  307. traceback.print_exc()
  308. return {
  309. 'final_output': f"Agent识别失败: {error_msg}",
  310. 'tool_calls': [],
  311. 'tool_results': [],
  312. 'all_anomalies': [],
  313. 'error': str(e)
  314. }
  315. def _execute_backup_recognition(self) -> Dict[str, Any]:
  316. """备用识别方案:直接调用所有工具"""
  317. print("🔄 启动备用识别方案:直接调用所有工具...")
  318. backup_results = {
  319. 'all_anomalies': [],
  320. 'tool_results': [],
  321. 'tool_names': []
  322. }
  323. for recognizer in self.recognizer_tools:
  324. print(f" 🔧 调用 {recognizer.display_name}...")
  325. try:
  326. result = recognizer._run(csv_path=self.csv_path)
  327. backup_results['tool_results'].append(result)
  328. backup_results['tool_names'].append(recognizer.name)
  329. # 提取异常
  330. if isinstance(result, dict):
  331. if 'identified_anomalies' in result:
  332. anomalies = result['identified_anomalies']
  333. for anomaly in anomalies:
  334. standardized = self._standardize_anomaly_record(anomaly, result)
  335. backup_results['all_anomalies'].append(standardized)
  336. print(f" 发现 {len(anomalies)} 条异常")
  337. elif 'identified_count' in result:
  338. print(f" 工具返回 {result['identified_count']} 条异常(但未找到详细记录)")
  339. else:
  340. print(f" 工具返回非字典结果: {type(result)}")
  341. except Exception as e:
  342. print(f" ❌ 工具调用失败: {e}")
  343. print(f"✅ 备用方案完成: 调用了 {len(backup_results['tool_names'])} 个工具, 发现 {len(backup_results['all_anomalies'])} 条异常")
  344. return backup_results
  345. def _process_agent_result(self, result: Any) -> Dict[str, Any]:
  346. """处理Agent执行结果"""
  347. agent_output = {
  348. 'final_output': '',
  349. 'tool_calls': [],
  350. 'tool_results': [],
  351. 'all_anomalies': [],
  352. 'messages_analysis': []
  353. }
  354. # 分析消息流
  355. for i, message in enumerate(result["messages"]):
  356. msg_info = {
  357. 'index': i + 1,
  358. 'type': message.type,
  359. 'has_tool_calls': False,
  360. 'tool_call_count': 0
  361. }
  362. # 记录工具调用
  363. if hasattr(message, 'tool_calls') and message.tool_calls:
  364. tool_calls = message.tool_calls
  365. agent_output['tool_calls'].extend(tool_calls)
  366. msg_info['has_tool_calls'] = True
  367. msg_info['tool_call_count'] = len(tool_calls)
  368. print(f"🛠️ 消息{i + 1}: 发现 {len(tool_calls)} 个工具调用")
  369. for tc in tool_calls:
  370. print(f" 工具: {tc.get('name', '未知')}")
  371. print(f" 参数: {tc.get('args', {})}")
  372. # 处理工具返回结果
  373. if message.type == 'tool':
  374. content = message.content
  375. agent_output['tool_results'].append(content)
  376. # ============ 新增调试信息 ============
  377. print(f"\n🔍 工具返回内容类型: {type(content)}")
  378. if isinstance(content, dict):
  379. print(f"📋 工具返回字典键: {list(content.keys())}")
  380. if 'identified_count' in content:
  381. print(f"📊 工具报告的异常数量: {content['identified_count']}")
  382. if 'identified_anomalies' in content:
  383. print(f"📦 工具返回的异常列表长度: {len(content['identified_anomalies'])}")
  384. # 显示前几条异常详情
  385. for j, anomaly in enumerate(content['identified_anomalies'][:3], 1):
  386. print(
  387. f" 异常{j}: ID={anomaly.get('txId')}, 原因={anomaly.get('recognition_reason', '')[:50]}...")
  388. elif isinstance(content, str):
  389. print(f"📝 工具返回字符串长度: {len(content)}")
  390. print(f" 前200字符: {content[:200]}...")
  391. # ============ 调试信息结束 ============
  392. # 处理异常数据
  393. anomalies = self._extract_anomalies_from_content(content)
  394. if anomalies:
  395. print(f"✅ 从工具结果提取到 {len(anomalies)} 条异常")
  396. agent_output['all_anomalies'].extend(anomalies)
  397. else:
  398. print(f"⚠️ 从工具结果提取到 0 条异常")
  399. msg_info['content_type'] = type(content).__name__
  400. msg_info['content_length'] = len(str(content))
  401. # 记录最终AI输出
  402. if message.type == 'ai' and i == len(result["messages"]) - 1:
  403. agent_output['final_output'] = getattr(message, 'content', '')
  404. msg_info['is_final'] = True
  405. msg_info['output_length'] = len(agent_output['final_output'])
  406. print(f"🤖 最终AI输出 ({msg_info['output_length']} 字符):")
  407. print("-" * 40)
  408. print(agent_output['final_output'][:300] + "..." if len(agent_output['final_output']) > 300 else
  409. agent_output['final_output'])
  410. print("-" * 40)
  411. agent_output['messages_analysis'].append(msg_info)
  412. return agent_output
  413. def _extract_anomalies_from_content(self, content: Any) -> List[Dict[str, Any]]:
  414. """从工具结果中提取异常数据 - 修复版"""
  415. anomalies = []
  416. try:
  417. print(f"🔍 提取异常,输入类型: {type(content)}")
  418. # ============ 第一步:统一转换为字典 ============
  419. processed_content = None
  420. if isinstance(content, dict):
  421. print(f" ✅ 已经是字典,直接处理")
  422. processed_content = content
  423. elif isinstance(content, str):
  424. print(f" 📝 处理字符串内容,长度: {len(content)}")
  425. print(f" 预览: {content[:200]}...")
  426. # 尝试多种解析方式,传入初始深度0
  427. processed_content = self._parse_string_content(content, depth=0, max_depth=3)
  428. if processed_content is None:
  429. print(f" ❌ 无法解析字符串内容,返回空列表")
  430. return anomalies
  431. else:
  432. print(f" ⚠️ 未知内容类型: {type(content)},返回空列表")
  433. return anomalies
  434. # ============ 第二步:从字典中提取异常 ============
  435. if isinstance(processed_content, dict):
  436. print(f" 📋 处理字典,键: {list(processed_content.keys())}")
  437. # 可能包含异常的字段名列表(按优先级)
  438. anomaly_fields = [
  439. 'identified_anomalies',
  440. 'all_anomalies',
  441. 'anomalies',
  442. 'abnormal_records',
  443. 'identified_abnormalities'
  444. ]
  445. found_anomalies = False
  446. for field in anomaly_fields:
  447. if field in processed_content:
  448. anomaly_list = processed_content[field]
  449. print(f" ✅ 找到字段 '{field}',类型: {type(anomaly_list)}")
  450. if isinstance(anomaly_list, list):
  451. print(f" 列表长度: {len(anomaly_list)}")
  452. for i, anomaly in enumerate(anomaly_list):
  453. if isinstance(anomaly, dict):
  454. standardized = self._standardize_anomaly_record(anomaly, processed_content)
  455. anomalies.append(standardized)
  456. print(f" ✓ 标准化异常 {i + 1}: ID={anomaly.get('txId', '未知')}")
  457. else:
  458. print(f" ⚠️ 异常记录 {i + 1} 不是字典: {type(anomaly)}")
  459. # 尝试转换非字典异常
  460. if hasattr(anomaly, '__dict__'):
  461. anomaly_dict = anomaly.__dict__
  462. standardized = self._standardize_anomaly_record(anomaly_dict, processed_content)
  463. anomalies.append(standardized)
  464. found_anomalies = True
  465. print(f" 📊 从字段 '{field}' 提取到 {len(anomaly_list)} 条异常")
  466. break # 找到一个就停止
  467. else:
  468. print(f" ⚠️ 字段 '{field}' 不是列表类型: {type(anomaly_list)}")
  469. # 如果没有找到标准字段,搜索任何包含字典的列表字段
  470. if not found_anomalies:
  471. print(f" 🔎 没有找到标准异常字段,搜索其他列表字段...")
  472. for key, value in processed_content.items():
  473. if isinstance(value, list) and len(value) > 0:
  474. print(f" 发现列表字段 '{key}',长度: {len(value)},元素类型: {type(value[0])}")
  475. # 检查列表元素是否是字典(可能包含异常)
  476. if len(value) > 0 and isinstance(value[0], dict):
  477. print(f" ⚠️ 列表 '{key}' 包含字典,可能包含异常数据")
  478. # 可以选择是否提取这些数据
  479. # for item in value:
  480. # if isinstance(item, dict) and 'txId' in item:
  481. # standardized = self._standardize_anomaly_record(item, processed_content)
  482. # anomalies.append(standardized)
  483. print(f" 🎯 最终提取到 {len(anomalies)} 条异常")
  484. except Exception as e:
  485. print(f"❌ 提取异常数据时出错: {e}")
  486. import traceback
  487. traceback.print_exc()
  488. return anomalies
  489. def _standardize_anomaly_record(self, anomaly: Dict, source_content: Any) -> Dict[str, Any]:
  490. """标准化异常记录"""
  491. if not isinstance(anomaly, dict):
  492. anomaly = {'raw_data': str(anomaly)}
  493. # 提取识别器名称
  494. recognizer_name = ''
  495. if isinstance(source_content, dict):
  496. recognizer_name = source_content.get('recognition_type', '未知')
  497. # 从execution_info中提取更多信息
  498. if 'execution_info' in source_content:
  499. exec_info = source_content['execution_info']
  500. recognizer_name = exec_info.get('display_name', recognizer_name)
  501. # 确保有必要的字段
  502. standardized = {
  503. 'recognition_source': 'agent',
  504. 'recognition_type': recognizer_name,
  505. 'txId': str(anomaly.get('txId', anomaly.get('tx_id', ''))),
  506. 'txDate': str(anomaly.get('txDate', anomaly.get('tx_date', ''))),
  507. 'txTime': str(anomaly.get('txTime', anomaly.get('tx_time', ''))),
  508. 'txAmount': float(anomaly.get('txAmount', anomaly.get('tx_amount', 0))),
  509. 'txDirection': str(anomaly.get('txDirection', anomaly.get('tx_direction', ''))),
  510. 'recognition_reason': str(anomaly.get('recognition_reason', anomaly.get('reason', ''))),
  511. 'severity': str(anomaly.get('severity', 'medium')),
  512. 'status': str(anomaly.get('status', '待核查')),
  513. 'raw_anomaly': anomaly # 保留原始数据
  514. }
  515. # 添加datetime信息(如果存在)
  516. if 'datetime' in anomaly and pd.notna(anomaly['datetime']):
  517. standardized['datetime'] = str(anomaly['datetime'])
  518. return standardized
  519. def _generate_universal_prompt(self, tools_info: List[Dict[str, str]]) -> str:
  520. """生成通用提示词"""
  521. # 构建工具列表
  522. tools_list = "\n".join([
  523. f"{info['index']}. {info['display_name']} ({info['name']}): {info['short_desc']}"
  524. for info in tools_info
  525. ])
  526. # 构建建议顺序
  527. suggested_order = " → ".join([info['display_name'] for info in tools_info])
  528. # 构建工具调用示例
  529. first_tool = tools_info[0]
  530. example_call = f"""{first_tool['name']}(csv_path="{self.csv_path}")"""
  531. return f"""
  532. # 银行交易流水异常识别分析任务
  533. ## 数据文件:
  534. {self.csv_path}
  535. ## 可用分析工具(共{len(tools_info)}个):
  536. {tools_list}
  537. ## 执行要求:
  538. 1. **必须使用上述工具**进行分析,不能跳过工具调用
  539. 2. 建议按顺序执行:{suggested_order}
  540. 3. 每个工具都需要传入csv_path参数,值为:{self.csv_path}
  541. 4. 整合所有工具的结果生成综合报告
  542. ## 工具调用示例:
  543. 要调用第一个工具,使用:{example_call}
  544. ## 请开始执行:
  545. 请首先调用 {first_tool['display_name']} 工具开始分析。
  546. """
  547. def _prepare_tools_info_for_prompt(self) -> List[Dict[str, str]]:
  548. """为提示词准备工具信息"""
  549. tools_info = []
  550. for i, tool in enumerate(self.recognizer_tools, 1):
  551. tool_info = {
  552. 'index': i,
  553. 'name': tool.name,
  554. 'display_name': getattr(tool, 'display_name', tool.name),
  555. 'description': tool.description,
  556. 'short_desc': tool.description[:100] + "..." if len(tool.description) > 100 else tool.description
  557. }
  558. tools_info.append(tool_info)
  559. return tools_info
  560. def _get_universal_system_prompt(self) -> str:
  561. """获取通用系统提示词"""
  562. return """
  563. 你是一个银行流水异常识别专家AI助手。
  564. ## 核心规则:
  565. 1. 你必须使用提供的工具来分析数据
  566. 2. 不能跳过工具直接回答问题
  567. 3. 每次分析至少要调用一个工具
  568. 4. 等待工具返回结果后再继续分析
  569. 5. 基于工具结果生成报告
  570. ## 工具使用说明:
  571. - 每个工具都需要csv_path参数
  572. - 使用用户提供的文件路径
  573. - 可以按顺序调用多个工具
  574. - 记录每个工具的结果
  575. ## 输出要求:
  576. - 总结每个工具的分析结果
  577. - 列出所有发现的异常
  578. - 提供综合风险评估
  579. - 给出后续核查建议
  580. """
  581. def _debug_agent_execution(self, agent_output: Dict[str, Any]):
  582. """调试Agent执行过程"""
  583. print("\n🔍 Agent执行调试信息:")
  584. print(f" 工具调用次数: {len(agent_output['tool_calls'])}")
  585. for i, tool_call in enumerate(agent_output['tool_calls']):
  586. print(f" 工具调用 {i + 1}:")
  587. print(f" 名称: {tool_call.get('name', '未知')}")
  588. print(f" 参数: {tool_call.get('args', {})}")
  589. print(f" 工具结果数量: {len(agent_output['tool_results'])}")
  590. for i, result in enumerate(agent_output['tool_results']):
  591. print(f" 工具结果 {i + 1}: {str(result)[:150]}...")
  592. def _consolidate_anomalies(self):
  593. """合并所有识别的异常"""
  594. all_anomalies = []
  595. # # 从直接识别结果中收集异常
  596. # direct_results = self.recognition_results.get('direct_results', {})
  597. # if 'all_anomalies' in direct_results:
  598. # all_anomalies.extend(direct_results['all_anomalies'])
  599. # 从Agent结果中收集异常
  600. agent_results = self.recognition_results.get('agent_results')
  601. if agent_results and 'all_anomalies' in agent_results:
  602. all_anomalies.extend(agent_results['all_anomalies'])
  603. # 去重
  604. unique_anomalies = []
  605. seen = set()
  606. for anomaly in all_anomalies:
  607. key = f"{anomaly.get('txId', '')}_{anomaly.get('recognition_type', '')}"
  608. if key not in seen:
  609. seen.add(key)
  610. unique_anomalies.append(anomaly)
  611. self.recognition_results['all_anomalies'] = unique_anomalies
  612. print(f"📊 合并后共有 {len(unique_anomalies)} 条异常")
  613. def _generate_recognition_summary(self):
  614. """生成识别摘要"""
  615. all_anomalies = self.recognition_results.get('all_anomalies', [])
  616. summary = {
  617. 'total_transactions': self.data_summary.get('transaction_count', 0),
  618. 'total_identified_anomalies': len(all_anomalies),
  619. 'recognition_ratio': f"{(len(all_anomalies) / self.data_summary.get('transaction_count', 1) * 100):.2f}%"
  620. if self.data_summary.get('transaction_count', 0) > 0 else "0%",
  621. 'recognition_completion_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
  622. 'enabled_recognizers': len(self.recognizer_tools),
  623. 'anomaly_distribution': {}
  624. }
  625. # 按类型统计异常
  626. for anomaly in all_anomalies:
  627. anomaly_type = anomaly.get('recognition_type', '未知')
  628. summary['anomaly_distribution'][anomaly_type] = summary['anomaly_distribution'].get(anomaly_type, 0) + 1
  629. # 按严重程度统计
  630. severity_counts = {'high': 0, 'medium': 0, 'low': 0}
  631. for anomaly in all_anomalies:
  632. severity = anomaly.get('severity', 'medium')
  633. severity_counts[severity] = severity_counts.get(severity, 0) + 1
  634. summary['severity_distribution'] = severity_counts
  635. self.recognition_results['summary'] = summary
  636. # 以下方法保持不变...
  637. def generate_recognition_report(self, output_dir: str = "outputs/reports") -> str:
  638. """生成异常识别报告"""
  639. try:
  640. # 创建输出目录
  641. Path(output_dir).mkdir(parents=True, exist_ok=True)
  642. timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
  643. report_id = f"anomaly_report_{timestamp}"
  644. print(f"\n📊 生成异常识别报告...")
  645. # 1. 保存识别的异常记录(CSV格式)
  646. anomalies_path = self._save_anomalies_csv(output_dir, report_id)
  647. # 2. 生成详细识别报告(JSON格式)
  648. report_path = self._save_detailed_report(output_dir, report_id)
  649. # 3. 生成识别摘要(文本格式)
  650. summary_path = self._save_summary_txt(output_dir, report_id)
  651. print(f"✅ 报告生成完成")
  652. print(f" 异常记录: {anomalies_path}")
  653. print(f" 详细报告: {report_path}")
  654. print(f" 识别摘要: {summary_path}")
  655. return report_path
  656. except Exception as e:
  657. raise
  658. def _save_anomalies_csv(self, output_dir: str, report_id: str) -> str:
  659. """保存异常记录为CSV文件"""
  660. anomalies_df = pd.DataFrame(self.recognition_results['all_anomalies'])
  661. # 定义列顺序
  662. column_order = [
  663. 'recognition_type', 'txId', 'txDate', 'txTime', 'txAmount',
  664. 'txDirection', 'recognition_reason', 'severity', 'status'
  665. ]
  666. # 只保留存在的列
  667. existing_columns = [col for col in column_order if col in anomalies_df.columns]
  668. other_columns = [col for col in anomalies_df.columns if col not in column_order]
  669. # 重新排序列
  670. anomalies_df = anomalies_df[existing_columns + other_columns]
  671. # 保存CSV
  672. anomalies_path = Path(output_dir) / f"{report_id}_anomalies.csv"
  673. anomalies_df.to_csv(anomalies_path, index=False, encoding='utf-8-sig')
  674. return str(anomalies_path)
  675. # 其他方法保持不变...
  676. # anomaly_recognizer_agent.py
  677. # 在 AnomalyRecognitionAgent 类的末尾添加以下方法(在现有方法之后)
  678. def get_recognition_summary(self) -> Dict[str, Any]:
  679. """获取识别摘要"""
  680. return self.recognition_results.get('summary', {})
  681. def get_recognizer_stats(self) -> List[Dict[str, Any]]:
  682. """获取识别器统计信息"""
  683. stats = []
  684. for recognizer in self.recognizer_tools:
  685. stats.append(recognizer.get_summary())
  686. return stats
  687. def _save_detailed_report(self, output_dir: str, report_id: str) -> str:
  688. """保存详细识别报告(JSON格式)"""
  689. report_data = {
  690. 'report_metadata': {
  691. 'report_id': report_id,
  692. 'generation_time': datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
  693. 'data_source': self.csv_path,
  694. 'recognition_method': '混合模式' if self.recognition_results.get('agent_results') else '直接模式'
  695. },
  696. 'data_summary': self.data_summary,
  697. 'recognition_configuration': {
  698. 'enabled_recognizers': [
  699. {
  700. 'name': tool.name,
  701. 'display_name': tool.display_name,
  702. 'description': tool.description[:100] + '...' if len(
  703. tool.description) > 100 else tool.description
  704. }
  705. for tool in self.recognizer_tools
  706. ],
  707. 'total_recognizers': len(self.recognizer_tools),
  708. 'config': self.config
  709. },
  710. 'recognition_results': {
  711. 'summary': self.recognition_results.get('summary', {}),
  712. 'direct_results_summary': {},
  713. 'agent_results_summary': {}
  714. }
  715. }
  716. # 添加直接识别结果摘要
  717. # direct_results = self.recognition_results.get('direct_results', {})
  718. # if 'results' in direct_results:
  719. # for recognizer_name, result in direct_results['results'].items():
  720. # report_data['recognition_results']['direct_results_summary'][recognizer_name] = {
  721. # 'identified_count': result.get('identified_count', 0),
  722. # 'recognition_status': result.get('recognition_status', '未知'),
  723. # 'execution_time': result.get('execution_info', {}).get('execution_time', '')
  724. # }
  725. # 添加Agent识别结果摘要
  726. agent_results = self.recognition_results.get('agent_results')
  727. if agent_results:
  728. report_data['recognition_results']['agent_results_summary'] = {
  729. 'iterations': agent_results.get('iterations', 0),
  730. 'tool_calls_count': len(agent_results.get('tool_calls', [])),
  731. 'final_output_preview': agent_results.get('final_output', '')[:500] + '...'
  732. if agent_results.get('final_output') else '无'
  733. }
  734. # 保存JSON报告
  735. report_path = Path(output_dir) / f"{report_id}.json"
  736. def json_serializer(obj):
  737. if isinstance(obj, (pd.Timestamp, datetime)):
  738. return obj.strftime("%Y-%m-%d %H:%M:%S")
  739. elif isinstance(obj, np.integer):
  740. return int(obj)
  741. elif isinstance(obj, np.floating):
  742. return float(obj)
  743. elif isinstance(obj, np.ndarray):
  744. return obj.tolist()
  745. elif pd.isna(obj):
  746. return None
  747. elif hasattr(obj, '__dict__'):
  748. return str(obj)
  749. return str(obj)
  750. with open(report_path, 'w', encoding='utf-8') as f:
  751. json.dump(report_data, f, ensure_ascii=False, indent=2, default=json_serializer)
  752. return str(report_path)
  753. def _save_summary_txt(self, output_dir: str, report_id: str) -> str:
  754. """保存识别摘要(文本格式)"""
  755. summary = self.recognition_results.get('summary', {})
  756. anomaly_distribution = summary.get('anomaly_distribution', {})
  757. severity_distribution = summary.get('severity_distribution', {})
  758. summary_path = Path(output_dir) / f"{report_id}_summary.txt"
  759. with open(summary_path, 'w', encoding='utf-8') as f:
  760. f.write("=" * 70 + "\n")
  761. f.write("银行流水异常识别报告摘要\n")
  762. f.write("=" * 70 + "\n\n")
  763. # 报告信息
  764. f.write("📅 报告信息:\n")
  765. f.write(f" 报告ID: {report_id}\n")
  766. f.write(f" 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
  767. f.write(f" 数据源: {self.csv_path}\n\n")
  768. # 数据概览
  769. f.write("📈 数据概览:\n")
  770. f.write(f" 总交易笔数: {summary.get('total_transactions', 0):,}\n")
  771. f.write(
  772. f" 时间范围: {self.data_summary.get('date_range', {}).get('start', '未知')} 至 {self.data_summary.get('date_range', {}).get('end', '未知')}\n")
  773. f.write(f" 总交易金额: {self.data_summary.get('total_amount', 0):,.2f}元\n")
  774. f.write(f" 平均交易金额: {self.data_summary.get('average_amount', 0):,.2f}元\n\n")
  775. # 识别结果
  776. f.write("🔍 异常识别结果:\n")
  777. f.write(f" 启用的识别器: {summary.get('enabled_recognizers', 0)} 个\n")
  778. f.write(f" 识别出的异常: {summary.get('total_identified_anomalies', 0)} 条\n")
  779. f.write(f" 异常识别率: {summary.get('recognition_ratio', '0%')}\n\n")
  780. # 异常类型分布
  781. if anomaly_distribution:
  782. f.write("📊 异常类型分布:\n")
  783. for anomaly_type, count in anomaly_distribution.items():
  784. f.write(f" - {anomaly_type}: {count} 条\n")
  785. f.write("\n")
  786. # 严重程度分布
  787. if severity_distribution:
  788. f.write("⚠️ 严重程度分布:\n")
  789. for severity, count in severity_distribution.items():
  790. f.write(f" - {severity.upper()}: {count} 条\n")
  791. f.write("\n")
  792. f.write("\n" + "=" * 70 + "\n")
  793. f.write("报告生成完成\n")
  794. f.write("=" * 70 + "\n")
  795. return str(summary_path)
  796. def _parse_string_content(self, content: str, depth: int = 0, max_depth: int = 3) -> Optional[Dict]:
  797. """解析字符串内容为字典 - 支持多种格式,带递归深度控制"""
  798. # 递归深度保护
  799. if depth >= max_depth:
  800. print(f" ⚠️ 达到最大递归深度 {max_depth},停止解析")
  801. return None
  802. if not content or not isinstance(content, str):
  803. return None
  804. print(f" [{depth}] 解析字符串,长度: {len(content)}")
  805. # 尝试1: JSON解析(标准格式,双引号)
  806. try:
  807. import json
  808. parsed = json.loads(content)
  809. if isinstance(parsed, dict):
  810. print(f" [{depth}] ✅ JSON解析成功")
  811. return parsed
  812. else:
  813. print(f" [{depth}] ⚠️ JSON解析成功但不是字典: {type(parsed)}")
  814. # 如果是列表或其他类型,包装成字典
  815. return {
  816. 'parsed_content': parsed,
  817. 'original_type': type(parsed).__name__,
  818. 'parse_method': 'json'
  819. }
  820. except json.JSONDecodeError as e:
  821. print(f" [{depth}] ⚠️ JSON解析失败: {e}")
  822. # 尝试2: Python字典字符串表示(单引号)
  823. # 先清理字符串,移除可能的额外空白
  824. cleaned_content = content.strip()
  825. if cleaned_content.startswith('{') and cleaned_content.endswith('}'):
  826. try:
  827. import ast
  828. parsed = ast.literal_eval(cleaned_content) # 安全解析Python表达式
  829. if isinstance(parsed, dict):
  830. print(f" [{depth}] ✅ ast解析成功(Python字典字符串)")
  831. return parsed
  832. else:
  833. print(f" [{depth}] ⚠️ ast解析成功但不是字典: {type(parsed)}")
  834. return {
  835. 'parsed_content': parsed,
  836. 'original_type': type(parsed).__name__,
  837. 'parse_method': 'ast'
  838. }
  839. except (SyntaxError, ValueError, TypeError) as e:
  840. print(f" [{depth}] ⚠️ ast解析失败: {e}")
  841. # 尝试3: 包含字典的复杂字符串(如日志输出)
  842. # 查找第一个{和最后一个},尝试提取字典部分
  843. start_idx = content.find('{')
  844. end_idx = content.rfind('}')
  845. if start_idx >= 0 and end_idx > start_idx:
  846. dict_str = content[start_idx:end_idx + 1]
  847. # 避免提取的内容和原内容相同(会导致无限递归)
  848. if dict_str == content:
  849. print(f" [{depth}] ⚠️ 提取的子字符串与原字符串相同,跳过递归")
  850. return None
  851. print(f" [{depth}] 尝试提取子字符串,长度: {len(dict_str)}")
  852. print(f" [{depth}] 子字符串前100字符: {dict_str[:100]}...")
  853. # 递归尝试解析提取的部分,增加深度计数
  854. result = self._parse_string_content(dict_str, depth + 1, max_depth)
  855. if result:
  856. return result
  857. # 尝试4: 可能是eval安全的简单表示
  858. try:
  859. # 最后尝试:直接eval(仅用于调试,生产环境慎用)
  860. # 这里用更安全的方式
  861. import ast
  862. parsed = ast.literal_eval(content)
  863. print(f" [{depth}] ⚠️ 直接解析成功: {type(parsed)}")
  864. return {
  865. 'raw_content': content,
  866. 'parsed_content': parsed,
  867. 'original_type': type(parsed).__name__,
  868. 'parse_method': 'direct'
  869. }
  870. except Exception as e:
  871. print(f" [{depth}] ⚠️ 直接解析失败: {e}")
  872. print(f" [{depth}] ❌ 所有解析方式都失败")
  873. return None
  874. # 修改主程序的这一部分
  875. if __name__ == '__main__':
  876. import os
  877. os.environ["LANGCHAIN_TRACING_V2"] = "false"
  878. os.environ["LANGCHAIN_API_KEY"] = ""
  879. # 禁用 LangGraph 的追踪
  880. os.environ["LANGSMITH_TRACING"] = "false"
  881. file_name = "11111_data_standard_20260113_112906.csv"
  882. curr_dir = os.path.dirname(os.path.abspath(__file__))
  883. file_path = os.path.join(curr_dir, "..", "..", "data_files", file_name)
  884. print(f"csv文件:{file_path}, 是否存在: {os.path.exists(file_path)}")
  885. agent = AnomalyRecognitionAgent(csv_path=file_path, api_key=LLM_API_KEY, base_url=LLM_BASE_URL, model_name=LLM_MODEL_NAME, config=anomaly_recognizer_config)
  886. print("\n" + "=" * 60)
  887. print("开始运行异常识别流程")
  888. print("=" * 60)
  889. try:
  890. # 1. 加载数据
  891. print("\n📥 步骤1: 加载交易数据...")
  892. transaction_data = agent.load_transaction_data()
  893. print(f" 成功加载 {len(transaction_data)} 条交易记录")
  894. # 2. 执行异常识别
  895. print("\n🔍 步骤2: 执行异常识别...")
  896. results = agent.execute_full_recognition()
  897. # 3. 生成报告
  898. print("\n📊 步骤3: 生成识别报告...")
  899. report_path = agent.generate_recognition_report()
  900. except Exception as e:
  901. print(f"\n❌ 执行过程中发生错误: {e}")
  902. import traceback
  903. traceback.print_exc()