planning_agent.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388
  1. from typing import List, Dict, Optional, Any, Union
  2. from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
  3. from pydantic import BaseModel, Field
  4. from langchain_openai import ChatOpenAI
  5. import json
  6. import os
  7. from datetime import datetime
  8. # 数据模型定义
  9. class ActionItem(BaseModel):
  10. """动作项定义"""
  11. action: str = Field(description="动作名称")
  12. parameters: Optional[Dict[str, Any]] = Field(default_factory=dict, description="动作参数")
  13. class ClarificationRequest(BaseModel):
  14. """澄清请求结构化格式"""
  15. questions: List[str] = Field(description="需要澄清的问题列表")
  16. missing_fields: List[str] = Field(default_factory=list, description="缺少的字段或信息")
  17. class PlanningDecision(BaseModel):
  18. """规划决策输出"""
  19. decision: str = Field(
  20. description="决策类型: data_classify, generate_outline, compute_metrics, finalize_report, clarify_requirements"
  21. )
  22. reasoning: str = Field(description="详细推理过程")
  23. next_actions: List[Union[str, ActionItem]] = Field(
  24. default_factory=list,
  25. description="下一步动作列表"
  26. )
  27. metrics_to_compute: List[str] = Field(
  28. default_factory=list,
  29. description="待计算指标ID列表(如 ['total_income', 'avg_balance'])"
  30. )
  31. priority_metrics: List[str] = Field(
  32. default_factory=list,
  33. description="优先级高的指标ID"
  34. )
  35. additional_requirements: Optional[
  36. Union[Dict[str, Any], List[Any], ClarificationRequest]
  37. ] = Field(default=None, description="额外需求或澄清信息")
  38. def normalize_requirements(req: Any) -> Optional[Dict[str, Any]]:
  39. """
  40. 规范化 additional_requirements
  41. 将列表转换为字典格式
  42. """
  43. if req is None:
  44. return None
  45. if isinstance(req, dict):
  46. return req
  47. if isinstance(req, list):
  48. # 如果LLM错误地返回了列表,转换为字典格式
  49. return {
  50. "questions": [str(item) for item in req],
  51. "missing_fields": []
  52. }
  53. return {"raw": str(req)}
  54. class PlanningAgent:
  55. """规划智能体:负责状态分析和决策制定"""
  56. def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com"):
  57. """
  58. 初始化规划Agent
  59. Args:
  60. api_key: DeepSeek API密钥
  61. base_url: DeepSeek API基础URL
  62. """
  63. self.llm = ChatOpenAI(
  64. model="deepseek-chat",
  65. api_key=api_key,
  66. base_url=base_url,
  67. temperature=0.1
  68. )
  69. # 初始化API调用跟踪
  70. self.api_calls = []
  71. def create_planning_prompt(self) -> ChatPromptTemplate:
  72. """创建规划提示模板"""
  73. return ChatPromptTemplate.from_messages([
  74. ("system", """
  75. 你是报告规划总控智能体,核心职责是精准分析当前状态并决定下一步行动。
  76. ### 决策选项(四选一)
  77. 1. data_standardize: 数据未做标准化或数据标准化失败
  78. 2. data_classify: 数据未分类打标或分类打标数据集数量为0
  79. 3. generate_outline:大纲未生成或大纲无效
  80. 4. compute_metrics:大纲已生成但指标未完成
  81. ### 决策规则(按顺序检查)
  82. 1. 检查 is_standardized 是否为0,值是0时 → 选择 data_standardize
  83. 2. 检查 data_set_classified 是否为空 或 数量为0时 → 选择 data_classify
  84. 3. 检查 outline_draft 是否为空 → 空则选择 generate_outline
  85. 4. 检查 metrics_requirements 是否为空 → 空则选择 generate_outline
  86. 5. 检查是否有待计算指标 → 有则选择 compute_metrics
  87. 6. 所有指标都已计算完成 → 选择 finalize_report
  88. 7. 如果无法理解需求 → 选择 clarify_requirements
  89. ### 重要原则
  90. - 数据标准化成功后,不要重复处理
  91. - 数据已分类打标后,不要重复分类打标
  92. - 大纲草稿已存在时,不要重复生成大纲
  93. - 决策为 compute_metrics 时,必须从状态信息中的"有效待计算指标ID列表"中选择
  94. - 确保 metrics_to_compute 是字符串数组格式
  95. - 确保指标ID与大纲中的global_metrics.metric_id完全一致
  96. - 从状态信息中的"有效待计算指标ID列表"中提取metric_id作为metrics_to_compute的值
  97. - 计算失败的指标可以重试最多3次
  98. - 绝对不要自己生成新的指标ID,必须严格使用状态信息中提供的已有指标ID
  99. - 如果状态信息中没有可用的指标ID,不要生成compute_metrics决策
  100. ### 输出字段说明
  101. - decision: 决策字符串
  102. - reasoning: 决策原因说明
  103. - metrics_to_compute: 待计算指标ID列表,必须从状态信息中的"有效待计算指标ID列表"中选择。选择所有可用指标,除非指标数量过多(>10个)需要分批计算
  104. - priority_metrics: 优先级指标列表(前2-3个最重要的指标),从metrics_to_compute中选择
  105. 必须输出有效的JSON格式!"""),
  106. MessagesPlaceholder("messages"),
  107. ("user", "报告需求:{question}\n\n请输出决策结果。")
  108. ])
  109. async def make_decision(self, question: str, industry: str, current_state: Dict[str, Any]) -> PlanningDecision:
  110. """
  111. 根据当前状态做出规划决策
  112. Args:
  113. question: 用户查询
  114. industry: 行业
  115. current_state: 当前状态信息
  116. Returns:
  117. 规划决策结果
  118. """
  119. planner = self.create_planning_prompt() | self.llm
  120. # 构建状态评估上下文
  121. status_info = self._build_status_context(current_state)
  122. # 记录大模型输入
  123. print("========================================")
  124. print("[AGENT] PlanningAgent (规划Agent)")
  125. print("[MODEL_INPUT] PlanningAgent:")
  126. print(f"[CONTEXT] 基于当前状态做出规划决策")
  127. print(f"Question: {question}")
  128. print(f"Status info: {status_info}")
  129. print("========================================")
  130. # 执行规划
  131. start_time = datetime.now()
  132. response = await planner.ainvoke({
  133. "question": question,
  134. "industry": industry,
  135. "messages": [("system", status_info)]
  136. })
  137. end_time = datetime.now()
  138. # 解析JSON响应
  139. try:
  140. # 从响应中提取JSON内容
  141. content = response.content if hasattr(response, 'content') else str(response)
  142. # 尝试找到JSON部分
  143. json_start = content.find('{')
  144. json_end = content.rfind('}') + 1
  145. if json_start >= 0 and json_end > json_start:
  146. json_str = content[json_start:json_end]
  147. decision_data = json.loads(json_str)
  148. # 预处理 additional_requirements 字段
  149. if "additional_requirements" in decision_data:
  150. req = decision_data["additional_requirements"]
  151. if isinstance(req, str):
  152. # 如果是字符串,尝试将其转换为合适的格式
  153. if req.strip():
  154. # 将字符串包装为字典格式
  155. decision_data["additional_requirements"] = {"raw_content": req}
  156. else:
  157. # 空字符串设为 None
  158. decision_data["additional_requirements"] = None
  159. elif isinstance(req, list):
  160. # 如果是列表,转换为字典格式
  161. decision_data["additional_requirements"] = {
  162. "questions": [str(item) for item in req],
  163. "missing_fields": []
  164. }
  165. # 如果已经是 dict 或其他允许的类型,保持不变
  166. decision = PlanningDecision(**decision_data)
  167. # 验证决策的合理性
  168. if decision.decision == "compute_metrics":
  169. if not decision.metrics_to_compute:
  170. raise ValueError("AI决策缺少具体的指标ID")
  171. # 如果AI生成的指标ID明显是错误的(比如metric_001),使用默认逻辑
  172. if any(mid.startswith("metric_") and mid.replace("metric_", "").isdigit()
  173. for mid in decision.metrics_to_compute):
  174. raise ValueError("AI生成的指标ID格式不正确")
  175. else:
  176. raise ValueError("No JSON found in response")
  177. except Exception as e:
  178. print(f"解析规划决策响应失败: {e},使用默认决策")
  179. # 返回默认决策
  180. decision = self._get_default_decision(current_state)
  181. # 记录API调用结果 - 简化版
  182. content = response.content if hasattr(response, 'content') else str(response)
  183. # 从content中提取JSON结果
  184. extracted_json = ""
  185. try:
  186. json_start = content.find('{')
  187. json_end = content.rfind('}') + 1
  188. if json_start >= 0 and json_end > json_start:
  189. extracted_json = content[json_start:json_end]
  190. except:
  191. extracted_json = content # 如果提取失败,使用原始content
  192. call_id = f"api_mll_规划决策_{'{:.2f}'.format((end_time - start_time).total_seconds())}"
  193. api_call_info = {
  194. "call_id": call_id,
  195. "response": extracted_json
  196. }
  197. self.api_calls.append(api_call_info)
  198. # 保存API结果到文件 - Markdown格式,更易阅读
  199. # 使用运行ID创建独立的文件夹
  200. run_id = os.environ.get('FLOW_RUN_ID', 'default')
  201. api_results_dir = f"api_results_{run_id}"
  202. os.makedirs(api_results_dir, exist_ok=True)
  203. timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
  204. filename = f"{timestamp}_{call_id}.md"
  205. filepath = os.path.join(api_results_dir, filename)
  206. try:
  207. with open(filepath, 'w', encoding='utf-8') as f:
  208. f.write("# 规划决策结果\n\n")
  209. f.write("## 决策结果\n\n")
  210. f.write("```json\n")
  211. f.write(api_call_info["response"])
  212. f.write("\n```\n")
  213. print(f"[API_RESULT] 保存API结果文件: {filepath}")
  214. except Exception as e:
  215. print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
  216. # 记录大模型输出
  217. print(
  218. f"[MODEL_OUTPUT] PlanningAgent: {json.dumps(decision.dict() if hasattr(decision, 'dict') else decision, ensure_ascii=False)}")
  219. print("========================================")
  220. return decision
  221. def _build_status_context(self, state: Dict[str, Any]) -> str:
  222. """构建状态评估上下文"""
  223. required_count = len(state.get("metrics_requirements", []))
  224. computed_count = len(state.get("computed_metrics", {}))
  225. coverage = computed_count / required_count if required_count > 0 else 0
  226. # 计算失败统计
  227. failed_attempts = state.get("failed_metric_attempts", {})
  228. pending_ids = state.get("pending_metric_ids", [])
  229. # 过滤掉失败次数过多的指标
  230. max_retry = 3
  231. filtered_pending_ids = [
  232. mid for mid in pending_ids
  233. if failed_attempts.get(mid, 0) < max_retry
  234. ]
  235. # 获取可用的指标ID
  236. available_metric_ids = []
  237. outline_draft = state.get('outline_draft')
  238. if outline_draft and outline_draft.global_metrics:
  239. available_metric_ids = [m.metric_id for m in outline_draft.global_metrics if m.metric_id]
  240. return f"""
  241. 当前状态评估:
  242. - 规划步骤: {state.get('planning_step', 0)}
  243. - 数据标准化: {state.get('is_standardized', 0)}
  244. - 数据分类打标数量: {len(state.get('data_set_classified', 0))}
  245. - 大纲版本: {state.get('outline_version', 0)}
  246. - 大纲草稿存在: {state.get('outline_draft') is not None}
  247. - 指标需求总数: {required_count}
  248. - 已计算指标数: {computed_count}
  249. - 指标覆盖率: {coverage:.2%}
  250. - 待计算指标数: {len(pending_ids)}
  251. - 有效待计算指标ID列表: {filtered_pending_ids}
  252. - 可用指标ID列表: {available_metric_ids}
  253. - 失败尝试记录: {failed_attempts}
  254. """
  255. def analyze_current_state(state: Dict[str, Any]) -> Dict[str, Any]:
  256. """
  257. 分析当前状态,返回关键信息
  258. Args:
  259. state: 当前状态
  260. Returns:
  261. 状态分析结果
  262. """
  263. required_metrics = state.get("metrics_requirements", [])
  264. computed_metrics = state.get("computed_metrics", {})
  265. # 计算覆盖率
  266. required_count = len(required_metrics)
  267. computed_count = len(computed_metrics)
  268. coverage = computed_count / required_count if required_count > 0 else 0
  269. # 找出未计算的指标
  270. computed_ids = set(computed_metrics.keys())
  271. pending_metrics = [
  272. m for m in required_metrics
  273. if m.metric_id not in computed_ids
  274. ]
  275. # 检查失败次数
  276. failed_attempts = state.get("failed_metric_attempts", {})
  277. max_retry = 3
  278. valid_pending_metrics = [
  279. m for m in pending_metrics
  280. if failed_attempts.get(m.metric_id, 0) < max_retry
  281. ]
  282. return {
  283. "has_outline": state.get("outline_draft") is not None,
  284. "required_count": required_count,
  285. "computed_count": computed_count,
  286. "coverage": coverage,
  287. "pending_metrics": pending_metrics,
  288. "valid_pending_metrics": valid_pending_metrics,
  289. "pending_ids": [m.metric_id for m in pending_metrics],
  290. "valid_pending_ids": [m.metric_id for m in valid_pending_metrics],
  291. "planning_step": state.get("planning_step", 0),
  292. "outline_version": state.get("outline_version", 0)
  293. }
  294. async def plan_next_action(question: str, industry: str, current_state: Dict[str, Any],
  295. api_key: str) -> PlanningDecision:
  296. """
  297. 规划下一步行动的主函数
  298. Args:
  299. question: 用户查询
  300. current_state: 当前状态
  301. api_key: API密钥
  302. Returns:
  303. 规划决策结果
  304. """
  305. agent = PlanningAgent(api_key)
  306. try:
  307. decision = await agent.make_decision(question, industry, current_state)
  308. print(f"\n🧠 规划决策:{decision.decision}")
  309. print(f" 推理:{decision.reasoning[:100]}...")
  310. if decision.metrics_to_compute:
  311. print(f" 待计算指标:{decision.metrics_to_compute}")
  312. return decision
  313. except Exception as e:
  314. print(f"⚠️ 规划决策出错: {e},使用默认决策")
  315. # 直接返回最基本的默认决策,避免复杂的默认决策逻辑
  316. return PlanningDecision(
  317. decision="finalize_report",
  318. reasoning="规划决策失败,使用默认的报告生成决策",
  319. next_actions=["生成最终报告"],
  320. metrics_to_compute=[],
  321. priority_metrics=[]
  322. )