planning_agent.py 15 KB

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