workflow_state.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440
  1. """
  2. 整合的工作流状态定义
  3. ===================
  4. 此文件定义了整合了多个Agent的工作流状态,兼容现有的Big Agent状态管理和新增的报告生成Agent状态。
  5. 状态层次:
  6. 1. 输入层:用户查询和数据
  7. 2. 意图层:意图识别结果
  8. 3. 规划层:规划决策和大纲生成
  9. 4. 计算层:指标计算结果
  10. 5. 结果层:最终报告生成
  11. 6. 对话层:消息历史和错误处理
  12. 兼容性:
  13. - 兼容现有的Big Agent WorkflowState
  14. - 整合来自other_agents的AgentState
  15. - 支持扩展新的Agent状态需求
  16. 作者: Big Agent Team
  17. 版本: 1.0.0
  18. 创建时间: 2024-12-20
  19. """
  20. from typing import TypedDict, List, Dict, Any, Optional
  21. from datetime import datetime
  22. from langchain_core.messages import BaseMessage
  23. from pydantic import BaseModel, Field
  24. # ============= 数据模型 =============
  25. class MetricRequirement(BaseModel):
  26. """指标需求定义"""
  27. metric_id: str = Field(description="指标唯一标识,如 'total_income_jan'")
  28. metric_name: str = Field(description="指标中文名称")
  29. calculation_logic: str = Field(description="计算逻辑描述")
  30. required_fields: List[str] = Field(description="所需字段")
  31. dependencies: List[str] = Field(default_factory=list, description="依赖的其他指标ID")
  32. class ReportSection(BaseModel):
  33. """报告大纲章节"""
  34. section_id: str = Field(description="章节ID")
  35. title: str = Field(description="章节标题")
  36. description: str = Field(description="章节内容要求")
  37. metrics_needed: List[str] = Field(description="所需指标ID列表")
  38. class ReportOutline(BaseModel):
  39. """完整报告大纲"""
  40. report_title: str = Field(description="报告标题")
  41. sections: List[ReportSection] = Field(description="章节列表")
  42. global_metrics: List[MetricRequirement] = Field(description="全局指标列表")
  43. # ============= 序列化工具函数 =============
  44. def convert_numpy_types(obj: Any) -> Any:
  45. """
  46. 递归转换所有numpy类型为Python原生类型
  47. 确保所有数据可序列化
  48. """
  49. if isinstance(obj, dict):
  50. return {str(k): convert_numpy_types(v) for k, v in obj.items()}
  51. elif isinstance(obj, list):
  52. return [convert_numpy_types(item) for item in obj]
  53. elif isinstance(obj, tuple):
  54. return tuple(convert_numpy_types(item) for item in obj)
  55. elif isinstance(obj, set):
  56. return {convert_numpy_types(item) for item in obj}
  57. elif hasattr(obj, 'item') and hasattr(obj, 'dtype'): # numpy scalar
  58. return convert_numpy_types(obj.item())
  59. else:
  60. return obj
  61. # ============= 整合的工作流状态定义 =============
  62. class IntegratedWorkflowState(TypedDict):
  63. """整合的工作流状态定义 - 兼容多个Agent系统"""
  64. # === 基础输入层 (兼容Big Agent) ===
  65. user_input: str
  66. question: str # 别名,兼容报告生成Agent
  67. industry: str # 行业
  68. # === 数据层 ===
  69. data_set: List[Dict[str, Any]] # 报告生成Agent的数据格式
  70. transactions_df: Optional[Any] # 可选的数据框格式
  71. file_name: str # 数据文件名称
  72. data_set_classified: List[Dict[str, Any]] # 分类打标后的数据集
  73. original_file_path: str # 上传文件绝对路径
  74. is_standardized: int # 数据是否已经标准化 0: 否 1: 是
  75. standardized_file_path: str # 数据标准化的文件路径
  76. # === 意图识别层 (Big Agent原有) ===
  77. intent_result: Optional[Dict[str, Any]]
  78. # === 规划和大纲层 (新增) ===
  79. planning_step: int
  80. plan_history: List[str]
  81. outline_draft: Optional[ReportOutline]
  82. outline_version: int
  83. outline_ready: bool
  84. # === 指标计算层 ===
  85. metrics_requirements: List[MetricRequirement] # 报告生成Agent格式
  86. computed_metrics: Dict[str, Any] # 计算结果
  87. metrics_cache: Dict[str, Any] # 缓存
  88. pending_metric_ids: List[str] # 待计算指标ID
  89. failed_metric_attempts: Dict[str, int] # 失败统计
  90. calculation_results: Optional[Dict[str, Any]] # Big Agent格式的计算结果
  91. # === 结果层 ===
  92. report_draft: Dict[str, Any] # 报告草稿
  93. knowledge_result: Optional[Dict[str, Any]] # Big Agent知识沉淀结果
  94. is_complete: bool
  95. completeness_score: float
  96. answer: Optional[str] # 最终答案
  97. # === 异常识别层 ===
  98. anomaly_recognition_completed: bool
  99. anomaly_recognition_results: Optional[Dict[str, Any]]
  100. anomaly_report_path: Optional[str]
  101. anomaly_summary: Dict[str, Any]
  102. # === 对话和消息层 ===
  103. messages: List[Dict[str, Any]] # Big Agent消息格式
  104. current_node: str
  105. session_id: str
  106. next_route: str
  107. # === 错误处理层 ===
  108. errors: List[str]
  109. last_decision: str
  110. # === 时间跟踪层 ===
  111. start_time: str
  112. end_time: Optional[str]
  113. api_result: Dict[str, Any] # 存储所有API调用结果
  114. # ============= 状态创建和初始化函数 =============
  115. def create_initial_integrated_state(question: str, industry: str, original_file_path: str, session_id: str = None) -> IntegratedWorkflowState:
  116. """
  117. 创建初始的整合状态
  118. Args:
  119. question: 用户查询
  120. industry: 行业
  121. original_file_path: 原始数据文件
  122. session_id: 会话ID
  123. Returns:
  124. 初始化后的状态
  125. """
  126. current_time = datetime.now().isoformat()
  127. session = session_id or f"session_{int(datetime.now().timestamp())}"
  128. return {
  129. # 基础输入
  130. "user_input": question,
  131. "question": question,
  132. "industry": industry,
  133. "original_file_path": original_file_path,
  134. # 数据层
  135. "is_standardized": 0, # 未标准化
  136. "data_set": [],
  137. "data_set_classified": [], # 分类打标后的数据集
  138. "transactions_df": None,
  139. "file_name": "", # 文件名称
  140. # 意图识别层
  141. "intent_result": None,
  142. # 规划和大纲层
  143. "planning_step": 0,
  144. "plan_history": [],
  145. "outline_draft": None,
  146. "outline_version": 0,
  147. "outline_ready": False,
  148. # 指标计算层
  149. "metrics_requirements": [],
  150. "computed_metrics": {},
  151. "metrics_cache": {},
  152. "pending_metric_ids": [],
  153. "failed_metric_attempts": {},
  154. "calculation_results": None,
  155. # 结果层
  156. "report_draft": {},
  157. "knowledge_result": None,
  158. "is_complete": False,
  159. "completeness_score": 0.0,
  160. "answer": None,
  161. # 对话和消息层
  162. "messages": [{
  163. "role": "user",
  164. "content": question,
  165. "timestamp": current_time
  166. }],
  167. "current_node": "start",
  168. "session_id": session,
  169. "next_route": "planning_node",
  170. # 错误处理层
  171. "errors": [],
  172. "last_decision": "init",
  173. # 时间跟踪层
  174. "start_time": current_time,
  175. "end_time": None,
  176. "api_result": {}, # 存储所有API调用结果
  177. # 计算模式配置层
  178. "use_rules_engine_only": False,
  179. "use_traditional_engine_only": False,
  180. # 异常识别层
  181. "anomaly_recognition_completed": False,
  182. "anomaly_recognition_results": None,
  183. "anomaly_report_path": None,
  184. "anomaly_summary": {},
  185. }
  186. def is_state_ready_for_calculation(state: IntegratedWorkflowState) -> bool:
  187. """
  188. 检查状态是否准备好进行指标计算
  189. Args:
  190. state: 当前状态
  191. Returns:
  192. 是否准备好
  193. """
  194. return (
  195. state.get("outline_draft") is not None and
  196. len(state.get("metrics_requirements", [])) > 0 and
  197. len(state.get("pending_metric_ids", [])) > 0
  198. )
  199. def get_calculation_progress(state: IntegratedWorkflowState) -> Dict[str, Any]:
  200. """
  201. 获取指标计算进度信息
  202. Args:
  203. state: 当前状态
  204. Returns:
  205. 进度信息
  206. """
  207. required = len(state.get("metrics_requirements", []))
  208. computed = len(state.get("computed_metrics", {}))
  209. pending = len(state.get("pending_metric_ids", []))
  210. return {
  211. "required_count": required,
  212. "computed_count": computed,
  213. "pending_count": pending,
  214. "coverage_rate": computed / required if required > 0 else 0,
  215. "is_complete": computed >= required * 0.8 # 80%覆盖率视为完成
  216. }
  217. def update_state_with_outline_generation(state: IntegratedWorkflowState, outline: ReportOutline) -> IntegratedWorkflowState:
  218. """
  219. 使用大纲生成结果更新状态
  220. Args:
  221. state: 当前状态
  222. outline: 生成的大纲
  223. Returns:
  224. 更新后的状态
  225. """
  226. new_state = state.copy()
  227. new_state["outline_draft"] = outline
  228. new_state["outline_version"] += 1
  229. new_state["outline_ready"] = True
  230. new_state["metrics_requirements"] = outline.global_metrics
  231. new_state["pending_metric_ids"] = [m.metric_id for m in outline.global_metrics]
  232. # 添加消息
  233. new_state["messages"].append({
  234. "role": "assistant",
  235. "content": f"✅ 大纲生成完成 v{new_state['outline_version']}:{outline.report_title}",
  236. "timestamp": datetime.now().isoformat()
  237. })
  238. return new_state
  239. def update_state_with_planning_decision(state: IntegratedWorkflowState, decision: Dict[str, Any]) -> IntegratedWorkflowState:
  240. """
  241. 使用规划决策结果更新状态
  242. Args:
  243. state: 当前状态
  244. decision: 规划决策
  245. Returns:
  246. 更新后的状态
  247. """
  248. new_state = state.copy()
  249. new_state["planning_step"] += 1
  250. new_state["last_decision"] = decision.get("decision", "unknown")
  251. new_state["next_route"] = decision.get("next_route", "planning_node")
  252. # 如果有待计算指标,更新待计算列表
  253. if decision.get("metrics_to_compute"):
  254. new_state["pending_metric_ids"] = decision["metrics_to_compute"]
  255. # 添加规划历史
  256. new_state["plan_history"].append(
  257. f"Step {new_state['planning_step']}: {decision.get('decision', 'unknown')}"
  258. )
  259. return new_state
  260. def update_state_with_report(state: IntegratedWorkflowState, final_report: Dict[str, Any]) -> IntegratedWorkflowState:
  261. """
  262. 使用最终报告完成状态
  263. Args:
  264. state: 当前状态
  265. final_report: 最终报告
  266. Returns:
  267. 完成的状态
  268. """
  269. new_state = state.copy()
  270. new_state["report_draft"] = final_report
  271. new_state["is_complete"] = True
  272. new_state["answer"] = final_report
  273. new_state["end_time"] = datetime.now().isoformat()
  274. # 计算完整性分数
  275. progress = get_calculation_progress(new_state)
  276. new_state["completeness_score"] = progress["coverage_rate"]
  277. return new_state
  278. def update_state_with_data_classified(state: IntegratedWorkflowState, data_set_classified: List[Dict]) -> IntegratedWorkflowState:
  279. """
  280. 使用分类打标结果更新状态
  281. Args:
  282. state: 当前状态
  283. data_set_classified: 分类打标的数据
  284. Returns:
  285. 更新后的状态
  286. """
  287. new_state = state.copy()
  288. new_state["data_set_classified"] = data_set_classified
  289. # 添加消息
  290. new_state["messages"].append({
  291. "role": "assistant",
  292. "content": f"✅ 数据分类打标已完成",
  293. "timestamp": datetime.now().isoformat()
  294. })
  295. return new_state
  296. def update_state_with_data_standardize(state: IntegratedWorkflowState, is_succ: int, standardized_file_path: str) -> IntegratedWorkflowState:
  297. """
  298. 根据数据标准化结果更新状态
  299. Args:
  300. state: 当前状态
  301. is_succ: 是否标准化数据成功 0:否 1:是
  302. standardized_file_path: 标准化后的文件路径
  303. Returns:
  304. 更新后的状态
  305. """
  306. import os
  307. new_state = state.copy()
  308. new_state["is_standardized"] = is_succ
  309. new_state["standardized_file_path"] = standardized_file_path
  310. new_state["file_name"] = os.path.basename(standardized_file_path)
  311. msg = "成功" if is_succ else "失败"
  312. # 添加消息
  313. new_state["messages"].append({
  314. "role": "assistant",
  315. "content": f"✅ 数据标准化完成,处理结果:{msg}",
  316. "timestamp": datetime.now().isoformat()
  317. })
  318. return new_state
  319. def update_state_with_anomaly_recognition(state: IntegratedWorkflowState, recognition_results: Dict[str, Any],
  320. report_path: str) -> IntegratedWorkflowState:
  321. """
  322. 使用异常识别结果更新状态
  323. Args:
  324. state: 当前状态
  325. recognition_results: 异常识别结果
  326. report_path: 异常报告路径
  327. Returns:
  328. 更新后的状态
  329. """
  330. new_state = state.copy()
  331. new_state["anomaly_recognition_completed"] = True
  332. new_state["anomaly_recognition_results"] = recognition_results
  333. new_state["anomaly_report_path"] = report_path
  334. # 提取摘要信息
  335. summary = recognition_results.get('summary', {})
  336. new_state["anomaly_summary"] = {
  337. "total_anomalies": summary.get('total_identified_anomalies', 0),
  338. "anomaly_ratio": summary.get('recognition_ratio', '0%'),
  339. "severity_distribution": summary.get('severity_distribution', {}),
  340. "anomaly_distribution": summary.get('anomaly_distribution', {})
  341. }
  342. # 添加消息
  343. new_state["messages"].append({
  344. "role": "assistant",
  345. "content": f"🔍 异常识别完成:发现 {summary.get('total_identified_anomalies', 0)} 条异常",
  346. "timestamp": datetime.now().isoformat()
  347. })
  348. return new_state