complete_agent_flow_rule.py 38 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846
  1. """
  2. 完整的智能体工作流 (Complete Agent Flow)
  3. =====================================
  4. 此工作流整合了规划、大纲生成和指标计算四个核心智能体,实现完整的报告生成流程。
  5. 包含的智能体:
  6. 1. PlanningAgent (规划智能体) - 分析状态并做出决策
  7. 2. OutlineAgent (大纲生成智能体) - 生成报告结构和指标需求
  8. 3. MetricCalculationAgent (指标计算智能体) - 执行标准指标计算
  9. 4. RulesEngineMetricCalculationAgent (规则引擎指标计算智能体) - 执行规则引擎指标计算
  10. 工作流程:
  11. 1. 规划节点 → 分析当前状态,决定下一步行动
  12. 2. 大纲生成节点 → 生成报告大纲和指标需求
  13. 3. 指标判断节点 → 根据大纲确定需要计算的指标
  14. 4. 指标计算节点 → 执行具体的指标计算任务
  15. 技术特点:
  16. - 基于LangGraph的状态机工作流
  17. - 支持条件路由和状态管理
  18. - 完善的错误处理机制
  19. - 详细的执行日志记录
  20. 作者: Big Agent Team
  21. 版本: 1.0.0
  22. 创建时间: 2024-12-20
  23. """
  24. import asyncio
  25. from typing import Dict, Any, List
  26. from datetime import datetime
  27. from langgraph.graph import StateGraph, END
  28. from llmops.workflow_state import (
  29. IntegratedWorkflowState,
  30. create_initial_integrated_state,
  31. get_calculation_progress,
  32. update_state_with_outline_generation,
  33. update_state_with_planning_decision,
  34. update_state_with_data_classified,
  35. convert_numpy_types,
  36. update_state_with_data_standardize,
  37. update_state_with_report
  38. )
  39. from llmops.agents.outline_agent import generate_report_outline
  40. from llmops.agents.planning_agent import plan_next_action
  41. from llmops.agents.rules_engine_metric_calculation_agent import RulesEngineMetricCalculationAgent
  42. from llmops.agents.data_manager import DataManager
  43. import os
  44. from llmops.agents.data_classify_agent import data_classify
  45. from llmops.config import multimodal_api_url, LLM_API_KEY, LLM_BASE_URL, LLM_MODEL_NAME
  46. from llmops.agents.data_stardard import data_standardize
  47. from llmops.agents.report_agent import generate_report_section_content
  48. class CompleteAgentFlow:
  49. """完整的智能体工作流"""
  50. def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com", model_name: str = "deepseek-chat"):
  51. """
  52. 初始化完整的工作流
  53. Args:
  54. api_key: DeepSeek API密钥
  55. base_url: DeepSeek API基础URL
  56. model_name: 模型名称
  57. """
  58. self.api_key = api_key
  59. self.base_url = base_url
  60. self.model_name = model_name
  61. # 初始规则引擎智能体
  62. self.rules_engine_agent = RulesEngineMetricCalculationAgent(api_key, base_url)
  63. # 创建工作流图
  64. self.workflow = self._create_workflow()
  65. def _create_workflow(self) -> StateGraph:
  66. """创建LangGraph工作流"""
  67. workflow = StateGraph(IntegratedWorkflowState)
  68. # 添加节点
  69. workflow.add_node("planning_node", self._planning_node)
  70. workflow.add_node("outline_generator", self._outline_generator_node)
  71. workflow.add_node("metric_calculator", self._metric_calculator_node)
  72. workflow.add_node("data_classify", self._data_classify_node)
  73. workflow.add_node("data_standardize", self._data_standardize_node)
  74. workflow.add_node("report_generator", self._report_generator_node)
  75. # 设置入口点
  76. workflow.set_entry_point("planning_node")
  77. # 添加条件边 - 基于规划决策路由
  78. workflow.add_conditional_edges(
  79. "planning_node",
  80. self._route_from_planning,
  81. {
  82. "outline_generator": "outline_generator",
  83. "metric_calculator": "metric_calculator",
  84. "data_classify": "data_classify",
  85. "data_standardize": "data_standardize",
  86. "report_generator": "report_generator",
  87. END: END
  88. }
  89. )
  90. # 从各个节点返回规划节点重新决策
  91. workflow.add_edge("data_standardize", "planning_node")
  92. workflow.add_edge("data_classify", "planning_node")
  93. workflow.add_edge("outline_generator", "planning_node")
  94. workflow.add_edge("metric_calculator", "planning_node")
  95. workflow.add_edge("report_generator", END)
  96. return workflow
  97. def _route_from_planning(self, state: IntegratedWorkflowState) -> str:
  98. """
  99. 从规划节点路由到下一个节点
  100. Args:
  101. state: 当前状态
  102. Returns:
  103. 目标节点名称
  104. """
  105. print(f"\n🔍 [路由决策] 步骤={state['planning_step']}, "
  106. f"数据集分类打标数量={len(state.get('data_set_classified', []))}",
  107. f"大纲={state.get('outline_draft') is not None}, "
  108. f"指标需求={len(state.get('metrics_requirements', []))}")
  109. # 防止无限循环
  110. if state['planning_step'] > 30:
  111. print("⚠️ 规划步骤超过30次,强制结束流程")
  112. return END
  113. # 数据标准化状态 0 → 数据标准化
  114. if state.get("is_standardized", 0) == 0:
  115. print("→ 路由到 data_standardize(数据标准化)")
  116. return "data_standardize"
  117. # 数据分类打标数量为0 → 分类打标
  118. if len(state.get("data_set_classified", [])) == 0:
  119. print("→ 路由到 data_classify(分类打标)")
  120. return "data_classify"
  121. # 如果大纲为空 → 生成大纲
  122. if not state.get("outline_draft"):
  123. print("→ 路由到 outline_generator(生成大纲)")
  124. return "outline_generator"
  125. # 如果指标需求为空但大纲已生成 → 评估指标需求
  126. if not state.get("metrics_requirements") and state.get("outline_draft"):
  127. print("→ 路由到 metric_evaluator(评估指标需求)")
  128. return "metric_evaluator"
  129. # 计算覆盖率
  130. progress = get_calculation_progress(state)
  131. coverage = progress["coverage_rate"]
  132. print(f" 指标覆盖率 = {coverage:.2%}")
  133. # 如果有待计算指标且覆盖率 < 100% → 计算指标
  134. if state.get("pending_metric_ids") and coverage < 1.0:
  135. print(f"→ 路由到 metric_calculator(计算指标,覆盖率={coverage:.2%})")
  136. return "metric_calculator"
  137. # 检查是否应该结束流程
  138. pending_ids = state.get("pending_metric_ids", [])
  139. failed_attempts = state.get("failed_metric_attempts", {})
  140. max_retries = 3
  141. # 计算还有哪些指标可以重试(未达到最大重试次数)
  142. retryable_metrics = [
  143. mid for mid in pending_ids
  144. if failed_attempts.get(mid, 0) < max_retries
  145. ]
  146. # 如果覆盖率 >= 80%,或者没有可重试的指标 → 生成报告
  147. if coverage >= 0.8 or not retryable_metrics:
  148. reason = "覆盖率达到80%" if coverage >= 0.8 else "没有可重试指标"
  149. print(f"→ 指标计算完成,进入生成报告(覆盖率={coverage:.2%},原因:{reason})")
  150. return "report_generator"
  151. # 默认返回规划节点
  152. return "planning_node"
  153. async def _planning_node(self, state: IntegratedWorkflowState) -> IntegratedWorkflowState:
  154. """规划节点:分析状态并做出决策"""
  155. try:
  156. print("🧠 正在执行规划分析...")
  157. # 使用规划智能体做出决策
  158. decision = await plan_next_action(
  159. question=state["question"],
  160. industry=state["industry"],
  161. current_state=state,
  162. api_key=self.api_key,
  163. base_url=self.base_url,
  164. model_name=self.model_name
  165. )
  166. # 更新状态
  167. new_state = update_state_with_planning_decision(state, {
  168. "decision": decision.decision,
  169. "next_route": self._decision_to_route(decision.decision),
  170. "metrics_to_compute": decision.metrics_to_compute
  171. })
  172. # 添加决策消息
  173. decision_msg = self._format_decision_message(decision)
  174. new_state["messages"].append({
  175. "role": "assistant",
  176. "content": decision_msg,
  177. "timestamp": datetime.now().isoformat()
  178. })
  179. print(f"✅ 规划决策完成:{decision.decision}")
  180. return convert_numpy_types(new_state)
  181. except Exception as e:
  182. print(f"❌ 规划节点执行失败: {e}")
  183. new_state = state.copy()
  184. new_state["errors"].append(f"规划节点错误: {str(e)}")
  185. return convert_numpy_types(new_state)
  186. async def _outline_generator_node(self, state: IntegratedWorkflowState) -> IntegratedWorkflowState:
  187. """大纲生成节点"""
  188. try:
  189. print("📝 正在生成报告大纲...")
  190. # 生成大纲(支持重试机制)
  191. outline = await generate_report_outline(
  192. question=state["question"],
  193. industry=state["industry"],
  194. sample_data=state["data_set"][:3], # 使用前3个样本
  195. api_key=self.api_key,
  196. base_url=self.base_url,
  197. model_name=self.model_name,
  198. max_retries=3, # 最多重试5次
  199. retry_delay=3.0 # 每次重试间隔3秒
  200. )
  201. # 更新状态
  202. new_state = update_state_with_outline_generation(state, outline)
  203. print(f"✅ 大纲生成完成:{outline.report_title}")
  204. print(f" 包含 {len(outline.sections)} 个章节,{len(outline.global_metrics)} 个指标需求")
  205. # 分析并打印AI的指标选择推理过程
  206. self._print_ai_selection_analysis(outline)
  207. return convert_numpy_types(new_state)
  208. except Exception as e:
  209. print(f"❌ 大纲生成失败: {e}")
  210. new_state = state.copy()
  211. new_state["errors"].append(f"大纲生成错误: {str(e)}")
  212. return convert_numpy_types(new_state)
  213. async def _data_classify_node(self, state: IntegratedWorkflowState) -> IntegratedWorkflowState:
  214. """数据分类打标节点"""
  215. try:
  216. standardized_file_path = state["standardized_file_path"]
  217. file_name = os.path.basename(standardized_file_path)
  218. # 读取标准化后的数据文件
  219. data_set = DataManager.load_data_from_csv_file(standardized_file_path)
  220. # 加载测试数据集并展示两条样例
  221. print(f"📊 读取标准化数据文件: {file_name}, 加载 {len(data_set)} 条记录")
  222. print(f"测试数据样例: {data_set[0:1]}")
  223. print("📝 正在对数据进行分类打标...")
  224. # 对数据进行分类打标
  225. data_set_classified = await data_classify(
  226. industry=state["industry"],
  227. data_set=data_set,
  228. file_name=state["file_name"]
  229. )
  230. # 更新状态
  231. new_state = update_state_with_data_classified(state, data_set_classified)
  232. print(f"✅ 数据分类打标完成,打标记录数: {len(data_set_classified)}")
  233. return convert_numpy_types(new_state)
  234. except Exception as e:
  235. print(f"❌ 数据分类打标失败: {e}")
  236. new_state = state.copy()
  237. new_state["errors"].append(f"数据分类打标错误: {str(e)}")
  238. return convert_numpy_types(new_state)
  239. async def _data_standardize_node(self, state: IntegratedWorkflowState) -> IntegratedWorkflowState:
  240. """数据标准化节点"""
  241. try:
  242. print("📝 正在对数据进行标准化处理...")
  243. # 数据标准化处理
  244. result = await data_standardize(
  245. api_key=self.api_key,
  246. base_url=self.base_url,
  247. model_name=self.model_name,
  248. multimodal_api_url=multimodal_api_url,
  249. input_file_path=state["original_file_path"]
  250. )
  251. is_succ = 0
  252. standardized_file_path = None
  253. if result["status"] == "success": # 数据标准化成功
  254. is_succ = 1
  255. standardized_file_path = result["file_path"]
  256. # 更新状态
  257. new_state = update_state_with_data_standardize(state, is_succ, standardized_file_path)
  258. print(f"✅ 数据标准化完成,处理状态: {is_succ},标准化文件路径:{standardized_file_path}")
  259. return convert_numpy_types(new_state)
  260. except Exception as e:
  261. print(f"❌ 数据标准化失败: {e}")
  262. new_state = state.copy()
  263. new_state["errors"].append(f"数据标准化错误: {str(e)}")
  264. return convert_numpy_types(new_state)
  265. async def _report_generator_node(self, state: IntegratedWorkflowState) -> IntegratedWorkflowState:
  266. """报告完成节点:生成最终报告"""
  267. try:
  268. print("📋 正在生成最终报告...")
  269. # 获取大纲和计算结果
  270. outline = state.get("outline_draft")
  271. computed_metrics = state.get("computed_metrics", {})
  272. print(f"已经完成的计算指标:{computed_metrics}")
  273. if not outline:
  274. raise ValueError("没有可用的报告大纲")
  275. # 生成最终报告
  276. final_report = {
  277. "title": outline.report_title,
  278. "generated_at": datetime.now().isoformat(),
  279. "summary": {
  280. "total_sections": len(outline.sections),
  281. "total_metrics_required": len(outline.global_metrics),
  282. "total_metrics_computed": len(computed_metrics),
  283. "planning_steps": state.get("planning_step", 0),
  284. "completion_rate": len(computed_metrics) / len(
  285. outline.global_metrics) if outline.global_metrics else 0
  286. },
  287. "sections": [],
  288. "metrics_detail": {}
  289. }
  290. chapter_num = 0
  291. total_sections = len(outline.sections)
  292. # 构建章节内容
  293. for section in outline.sections:
  294. section_content = {
  295. "section_id": section.section_id,
  296. "title": section.title,
  297. "description": section.description,
  298. "metrics": {}
  299. }
  300. # 添加该章节的指标数据
  301. for metric_id in section.metrics_needed:
  302. if metric_id in computed_metrics:
  303. section_content["metrics"][metric_id] = computed_metrics[metric_id]
  304. else:
  305. if not metric_id.startswith("metric-"):
  306. # 指标缺少metric前缀,进行补充
  307. section_content["metrics"][metric_id] = computed_metrics["metric-"+metric_id]
  308. else:
  309. section_content["metrics"][metric_id] = "数据缺失"
  310. chapter_num += 1
  311. # 生成章节内容
  312. chapter_content = await generate_report_section_content(api_key=self.api_key, base_url=self.base_url, model_name=self.model_name, section=section_content, chapter_num=chapter_num, total_sections=total_sections)
  313. print(f"生成章节内容:{chapter_content}")
  314. section_content["content"] = chapter_content
  315. final_report["sections"].append(section_content)
  316. # 添加详细的指标信息
  317. for metric_req in outline.global_metrics:
  318. metric_id = metric_req.metric_id
  319. final_report["metrics_detail"][metric_id] = {
  320. "name": metric_req.metric_name,
  321. "logic": metric_req.calculation_logic,
  322. "required_fields": metric_req.required_fields,
  323. "computed": metric_id in computed_metrics,
  324. "value": computed_metrics.get(metric_id, {}).get("value", "N/A")
  325. }
  326. # 更新状态
  327. new_state = update_state_with_report(state, final_report)
  328. # 添加完成消息
  329. new_state["messages"].append({
  330. "role": "assistant",
  331. "content": f"🎉 完整报告生成流程完成:{outline.report_title}",
  332. "timestamp": datetime.now().isoformat()
  333. })
  334. print(f"✅ 最终报告生成完成:{outline.report_title}")
  335. print(f" 章节数:{len(final_report['sections'])}")
  336. print(f" 计算指标:{len(computed_metrics)}/{len(outline.global_metrics)}")
  337. print(".2%")
  338. return convert_numpy_types(new_state)
  339. except Exception as e:
  340. print(f"❌ 报告完成失败: {e}")
  341. new_state = state.copy()
  342. new_state["errors"].append(f"报告完成错误: {str(e)}")
  343. return convert_numpy_types(new_state)
  344. def _print_ai_selection_analysis(self, outline):
  345. """打印AI指标选择的推理过程分析 - 完全通用版本"""
  346. print()
  347. print('╔══════════════════════════════════════════════════════════════════════════════╗')
  348. print('║ 🤖 AI指标选择分析 ║')
  349. print('╚══════════════════════════════════════════════════════════════════════════════╝')
  350. print()
  351. # 计算总指标数 - outline可能是字典格式,需要适配
  352. if hasattr(outline, 'sections'):
  353. # Pydantic模型格式
  354. total_metrics = sum(len(section.metrics_needed) for section in outline.sections)
  355. sections = outline.sections
  356. else:
  357. # 字典格式
  358. total_metrics = sum(len(section.get('metrics_needed', [])) for section in outline.get('sections', []))
  359. sections = outline.get('sections', [])
  360. # 获取可用指标总数(这里可以从状态或其他地方动态获取)
  361. available_count = 26 # 这个可以从API调用中动态获取
  362. print('📊 选择统计:')
  363. print(' ┌─────────────────────────────────────────────────────────────────────┐')
  364. print(' │ 系统可用指标: {}个 │ AI本次选择: {}个 │ 选择率: {:.1f}% │'.format(
  365. available_count, total_metrics, total_metrics/available_count*100 if available_count > 0 else 0))
  366. print(' └─────────────────────────────────────────────────────────────────────┘')
  367. print()
  368. print('📋 AI决策过程:')
  369. print(' 大模型已根据用户需求从{}个可用指标中选择了{}个最相关的指标。'.format(available_count, total_metrics))
  370. print(' 选择过程完全由大模型基于语义理解和业务逻辑进行,不涉及任何硬编码规则。')
  371. print()
  372. print('🔍 选择结果:')
  373. print(' • 总章节数: {}个'.format(len(sections)))
  374. print(' • 平均每章节指标数: {:.1f}个'.format(total_metrics/len(sections) if sections else 0))
  375. print(' • 选择策略: 基于用户需求的相关性分析')
  376. print()
  377. print('🎯 AI Agent核心能力:')
  378. print(' • 语义理解: 理解用户查询的业务意图和分析需求')
  379. print(' • 智能筛选: 从海量指标中挑选最相关的组合')
  380. print(' • 逻辑推理: 为每个分析维度提供充分的选择依据')
  381. print(' • 动态适配: 根据不同场景自动调整选择策略')
  382. print()
  383. print('💡 关键洞察:')
  384. print(' AI Agent通过大模型的推理能力,实现了超越传统规则引擎的智能化指标选择,')
  385. print(' 能够根据具体业务场景动态调整分析框架,确保分析的针对性和有效性。')
  386. print()
  387. async def _metric_calculator_node(self, state: IntegratedWorkflowState) -> IntegratedWorkflowState:
  388. """指标计算节点"""
  389. try:
  390. # 检查计算模式
  391. use_rules_engine_only = state.get("use_rules_engine_only", False)
  392. use_traditional_engine_only = state.get("use_traditional_engine_only", False)
  393. if use_rules_engine_only:
  394. print("🧮 正在执行规则引擎指标计算(专用模式)...")
  395. elif use_traditional_engine_only:
  396. print("🧮 正在执行传统引擎指标计算(专用模式)...")
  397. else:
  398. print("🧮 正在执行指标计算...")
  399. new_state = state.copy()
  400. # 使用规划决策指定的指标批次,如果没有指定则使用所有待计算指标
  401. current_batch = state.get("current_batch_metrics", [])
  402. if current_batch:
  403. pending_ids = current_batch
  404. print(f"🧮 本次计算批次包含 {len(pending_ids)} 个指标")
  405. else:
  406. pending_ids = state.get("pending_metric_ids", [])
  407. print(f"🧮 计算所有待计算指标,共 {len(pending_ids)} 个")
  408. if not pending_ids:
  409. print("⚠️ 没有待计算的指标")
  410. return convert_numpy_types(new_state)
  411. # 获取指标需求信息
  412. metrics_requirements = state.get("metrics_requirements", [])
  413. if not metrics_requirements:
  414. print("⚠️ 没有指标需求信息")
  415. return convert_numpy_types(new_state)
  416. # 计算成功和失败的指标
  417. successful_calculations = 0
  418. failed_calculations = 0
  419. # 遍历待计算的指标(创建副本避免修改时遍历的问题)
  420. for metric_id in pending_ids.copy():
  421. try:
  422. # 找到对应的指标需求
  423. metric_req = next((m for m in metrics_requirements if m.metric_id == metric_id), None)
  424. if not metric_req:
  425. # 修复:找不到指标需求时,创建临时的指标需求结构,避免跳过指标
  426. print(f"⚠️ 指标 {metric_id} 找不到需求信息,创建临时配置继续计算")
  427. metric_req = type('MetricRequirement', (), {
  428. 'metric_id': metric_id,
  429. 'metric_name': metric_id.replace('metric-', '') if metric_id.startswith('metric-') else metric_id,
  430. 'calculation_logic': f'计算 {metric_id}',
  431. 'required_fields': ['transactions'],
  432. 'dependencies': []
  433. })()
  434. print(f"🧮 计算指标: {metric_id} - {metric_req.metric_name}")
  435. # 根据模式决定使用哪种计算方式
  436. if use_rules_engine_only:
  437. # 只使用规则引擎计算
  438. use_rules_engine = True
  439. print(f" 使用规则引擎模式")
  440. elif use_traditional_engine_only:
  441. # 只使用传统引擎计算
  442. use_rules_engine = False
  443. print(f" 使用传统引擎模式")
  444. else:
  445. # 自动选择计算方式:优先使用规则引擎,只在规则引擎不可用时使用传统计算
  446. use_rules_engine = True # 默认使用规则引擎计算所有指标
  447. if use_rules_engine:
  448. # 使用规则引擎计算
  449. # 现在metric_id已经是知识ID,直接使用它作为配置名
  450. config_name = metric_id # metric_id 已经是知识ID,如 "metric-分析账户数量"
  451. intent_result = {
  452. "target_configs": [config_name],
  453. "intent_category": "指标计算"
  454. }
  455. print(f" 使用知识ID: {config_name}")
  456. # 将打好标的数据集传入指标计算函数中
  457. data_set_classified = state.get("data_set_classified", [])
  458. results = await self.rules_engine_agent.calculate_metrics(intent_result, data_set_classified)
  459. else:
  460. # 使用传统指标计算(模拟)
  461. # 这里简化处理,实际应该根据配置文件调用相应的API
  462. results = {
  463. "success": True,
  464. "results": [{
  465. "config_name": metric_req.metric_id,
  466. "result": {
  467. "success": True,
  468. "data": f"传统引擎计算结果:{metric_req.metric_name}",
  469. "value": 100.0 # 模拟数值
  470. }
  471. }]
  472. }
  473. # 处理计算结果
  474. calculation_success = False
  475. for result in results.get("results", []):
  476. if result.get("result", {}).get("success"):
  477. # 计算成功
  478. new_state["computed_metrics"][metric_id] = result["result"]
  479. successful_calculations += 1
  480. calculation_success = True
  481. print(f"✅ 指标 {metric_id} 计算成功")
  482. break # 找到一个成功的就算成功
  483. else:
  484. # 计算失败
  485. failed_calculations += 1
  486. print(f"❌ 指标 {metric_id} 计算失败")
  487. # 初始化失败尝试记录
  488. if "failed_metric_attempts" not in new_state:
  489. new_state["failed_metric_attempts"] = {}
  490. # 根据计算结果处理指标
  491. if calculation_success:
  492. # 计算成功:从待计算列表中移除
  493. if metric_id in new_state["pending_metric_ids"]:
  494. new_state["pending_metric_ids"].remove(metric_id)
  495. # 重置失败计数
  496. new_state["failed_metric_attempts"].pop(metric_id, None)
  497. else:
  498. # 计算失败:记录失败次数,不从待计算列表移除
  499. new_state["failed_metric_attempts"][metric_id] = new_state["failed_metric_attempts"].get(metric_id, 0) + 1
  500. max_retries = 3
  501. if new_state["failed_metric_attempts"][metric_id] >= max_retries:
  502. print(f"⚠️ 指标 {metric_id} 已达到最大重试次数 ({max_retries}),从待计算列表中移除")
  503. if metric_id in new_state["pending_metric_ids"]:
  504. new_state["pending_metric_ids"].remove(metric_id)
  505. except Exception as e:
  506. print(f"❌ 计算指标 {metric_id} 时发生异常: {e}")
  507. failed_calculations += 1
  508. # 初始化失败尝试记录
  509. if "failed_metric_attempts" not in new_state:
  510. new_state["failed_metric_attempts"] = {}
  511. # 记录失败次数
  512. new_state["failed_metric_attempts"][metric_id] = new_state["failed_metric_attempts"].get(metric_id, 0) + 1
  513. max_retries = 3
  514. if new_state["failed_metric_attempts"][metric_id] >= max_retries:
  515. print(f"⚠️ 指标 {metric_id} 异常已达到最大重试次数 ({max_retries}),从待计算列表中移除")
  516. if metric_id in new_state["pending_metric_ids"]:
  517. new_state["pending_metric_ids"].remove(metric_id)
  518. # 更新计算结果统计
  519. new_state["calculation_results"] = {
  520. "total_configs": len(pending_ids),
  521. "successful_calculations": successful_calculations,
  522. "failed_calculations": failed_calculations
  523. }
  524. # 添加消息
  525. if use_rules_engine_only:
  526. message_content = f"🧮 规则引擎指标计算完成:{successful_calculations} 成功,{failed_calculations} 失败"
  527. elif use_traditional_engine_only:
  528. message_content = f"🧮 传统引擎指标计算完成:{successful_calculations} 成功,{failed_calculations} 失败"
  529. else:
  530. message_content = f"🧮 指标计算完成:{successful_calculations} 成功,{failed_calculations} 失败"
  531. new_state["messages"].append({
  532. "role": "assistant",
  533. "content": message_content,
  534. "timestamp": datetime.now().isoformat()
  535. })
  536. if use_rules_engine_only:
  537. print(f"✅ 规则引擎指标计算完成:{successful_calculations} 成功,{failed_calculations} 失败")
  538. elif use_traditional_engine_only:
  539. print(f"✅ 传统引擎指标计算完成:{successful_calculations} 成功,{failed_calculations} 失败")
  540. else:
  541. print(f"✅ 指标计算完成:{successful_calculations} 成功,{failed_calculations} 失败")
  542. return convert_numpy_types(new_state)
  543. except Exception as e:
  544. print(f"❌ 指标计算节点失败: {e}")
  545. new_state = state.copy()
  546. new_state["errors"].append(f"指标计算错误: {str(e)}")
  547. return convert_numpy_types(new_state)
  548. def _decision_to_route(self, decision: str) -> str:
  549. """将规划决策转换为路由"""
  550. decision_routes = {
  551. "data_classify": "data_classify",
  552. "generate_outline": "outline_generator",
  553. "compute_metrics": "metric_calculator",
  554. "finalize_report": END # 直接结束流程
  555. }
  556. return decision_routes.get(decision, "planning_node")
  557. def _format_decision_message(self, decision: Any) -> str:
  558. """格式化决策消息"""
  559. try:
  560. decision_type = getattr(decision, 'decision', 'unknown')
  561. reasoning = getattr(decision, 'reasoning', '')
  562. if decision_type == "compute_metrics" and hasattr(decision, 'metrics_to_compute'):
  563. metrics = decision.metrics_to_compute
  564. return f"🧮 规划决策:计算 {len(metrics)} 个指标"
  565. elif decision_type == "finalize_report":
  566. return f"✅ 规划决策:生成最终报告"
  567. elif decision_type == "generate_outline":
  568. return f"📋 规划决策:生成大纲"
  569. else:
  570. return f"🤔 规划决策:{decision_type}"
  571. except:
  572. return "🤔 规划决策已完成"
  573. async def run_workflow(self, question: str, industry: str, original_file_path: str, session_id: str = None, use_rules_engine_only: bool = False, use_traditional_engine_only: bool = False) -> Dict[str, Any]:
  574. """
  575. 运行完整的工作流
  576. Args:
  577. question: 用户查询
  578. industry: 行业
  579. original_file_path: 原始文件路径
  580. session_id: 会话ID
  581. use_rules_engine_only: 是否只使用规则引擎指标计算
  582. use_traditional_engine_only: 是否只使用传统引擎指标计算
  583. Returns:
  584. 工作流结果
  585. """
  586. try:
  587. print("🚀 启动完整智能体工作流...")
  588. print(f"问题:{question}")
  589. print(f"行业:{industry}")
  590. print(f"数据文件:{original_file_path}")
  591. if use_rules_engine_only:
  592. print("计算模式:只使用规则引擎")
  593. elif use_traditional_engine_only:
  594. print("计算模式:只使用传统引擎")
  595. else:
  596. print("计算模式:标准模式")
  597. # 创建初始状态
  598. initial_state = create_initial_integrated_state(question, industry, original_file_path, session_id)
  599. # 设置计算模式标记
  600. if use_rules_engine_only:
  601. initial_state["use_rules_engine_only"] = True
  602. initial_state["use_traditional_engine_only"] = False
  603. elif use_traditional_engine_only:
  604. initial_state["use_rules_engine_only"] = False
  605. initial_state["use_traditional_engine_only"] = True
  606. else:
  607. initial_state["use_rules_engine_only"] = False
  608. initial_state["use_traditional_engine_only"] = False
  609. # 编译工作流
  610. app = self.workflow.compile()
  611. # 执行工作流
  612. result = await app.ainvoke(initial_state)
  613. print("✅ 工作流执行完成")
  614. return {
  615. "success": True,
  616. "result": result,
  617. "answer": result.get("answer"),
  618. "report": result.get("report_draft"),
  619. "session_id": result.get("session_id"),
  620. "execution_summary": {
  621. "planning_steps": result.get("planning_step", 0),
  622. "outline_generated": result.get("outline_draft") is not None,
  623. "metrics_computed": len(result.get("computed_metrics", {})),
  624. "completion_rate": result.get("completeness_score", 0)
  625. }
  626. }
  627. except Exception as e:
  628. print(f"❌ 工作流执行失败: {e}")
  629. return {
  630. "success": False,
  631. "error": str(e),
  632. "result": None
  633. }
  634. # 便捷函数
  635. async def run_complete_agent_flow(question: str, industry: str, data: List[Dict[str, Any]], file_name: str, api_key: str, session_id: str = None, use_rules_engine_only: bool = False, use_traditional_engine_only: bool = False) -> Dict[str, Any]:
  636. """
  637. 运行完整智能体工作流的便捷函数
  638. Args:
  639. question: 用户查询
  640. data: 数据集
  641. file_name: 数据文件名称
  642. api_key: API密钥
  643. session_id: 会话ID
  644. use_rules_engine_only: 是否只使用规则引擎指标计算
  645. use_traditional_engine_only: 是否只使用传统引擎指标计算
  646. Returns:
  647. 工作流结果
  648. """
  649. workflow = CompleteAgentFlow(api_key)
  650. return await workflow.run_workflow(question, industry, data, file_name, session_id, use_rules_engine_only, use_traditional_engine_only)
  651. # 便捷函数
  652. async def run_flow(question: str, industry: str, original_file_path: str, api_key: str, base_url: str, model_name: str, session_id: str = None, use_rules_engine_only: bool = False, use_traditional_engine_only: bool = False) -> Dict[str, Any]:
  653. """
  654. 运行完整智能体工作流的便捷函数
  655. Args:
  656. question: 用户查询
  657. data: 数据集
  658. original_file_path: 原始文件路径(pdf/img/csv)
  659. api_key: API密钥
  660. base_url: LLM base url
  661. model_name: LLM model name
  662. session_id: 会话ID
  663. use_rules_engine_only: 是否只使用规则引擎指标计算
  664. use_traditional_engine_only: 是否只使用传统引擎指标计算
  665. Returns:
  666. 工作流结果
  667. """
  668. workflow = CompleteAgentFlow(api_key, base_url, model_name)
  669. return await workflow.run_workflow(question, industry, original_file_path, session_id, use_rules_engine_only, use_traditional_engine_only)
  670. # 主函数用于测试
  671. async def main():
  672. """主函数:执行系统测试"""
  673. import os
  674. os.environ["LANGCHAIN_TRACING_V2"] = "false"
  675. os.environ["LANGCHAIN_API_KEY"] = ""
  676. # 禁用 LangGraph 的追踪
  677. os.environ["LANGSMITH_TRACING"] = "false"
  678. print("🚀 执行CompleteAgentFlow系统测试")
  679. print("=" * 50)
  680. # 行业
  681. industry = "农业"
  682. # 测试文件(pdf/img/csv)
  683. file_name = "11111.png"
  684. curr_dir = os.path.dirname(os.path.abspath(__file__))
  685. file_path = os.path.join(curr_dir, "..", "data_files", file_name)
  686. print(f"使用LLM:{LLM_MODEL_NAME}")
  687. # 执行测试
  688. result = await run_flow(
  689. question="请生成一份详细的农业经营贷流水分析报告,需要包含:1.总收入和总支出统计 2.收入笔数和支出笔数 3.各类型收入支出占比分析 4.交易对手收入支出TOP3排名 5.按月份的收入支出趋势分析 6.账户数量和交易时间范围统计 7.资金流入流出月度统计等全面指标",
  690. industry = industry,
  691. original_file_path=file_path,
  692. api_key=LLM_API_KEY,
  693. base_url=LLM_BASE_URL,
  694. model_name=LLM_MODEL_NAME,
  695. session_id="direct-test"
  696. )
  697. print(f"📋 结果: {'✅ 成功' if result.get('success') else '❌ 失败'}")
  698. print(f"{result}")
  699. if result.get('success'):
  700. summary = result.get('execution_summary', {})
  701. print(f" 规划步骤: {summary.get('planning_steps', 0)}")
  702. print(f" 指标计算: {summary.get('metrics_computed', 0)}")
  703. print("🎉 测试成功!")
  704. return result
  705. if __name__ == "__main__":
  706. import asyncio
  707. asyncio.run(main())