调整规划智有体,增加数据分类

llmops/agents/planning_agent.py

@@ -1,4 +1,3 @@
-
 from typing import List, Dict, Optional, Any, Union
 from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
 from pydantic import BaseModel, Field
@@ -24,7 +23,7 @@ class ClarificationRequest(BaseModel):
 class PlanningDecision(BaseModel):
     """规划决策输出"""
     decision: str = Field(
-        description="决策类型: generate_outline, compute_metrics, finalize_report, clarify_requirements"
+        description="决策类型: data_classify, generate_outline, compute_metrics, finalize_report, clarify_requirements"
     )
     reasoning: str = Field(description="详细推理过程")
     next_actions: List[Union[str, ActionItem]] = Field(
@@ -89,36 +88,40 @@ class PlanningAgent:
     def create_planning_prompt(self) -> ChatPromptTemplate:
         """创建规划提示模板"""
         return ChatPromptTemplate.from_messages([
-            ("system", """你是报告规划总控智能体，核心职责是精准分析当前状态并决定下一步行动。
-
-### 决策选项（二选一）
-1. generate_outline：大纲未生成或大纲无效
-2. compute_metrics：大纲已生成但指标未完成
-
-### 决策规则（按顺序检查）
-1. 检查 outline_draft 是否为空 → 空则选择 generate_outline
-2. 检查 metrics_requirements 是否为空 → 空则选择 generate_outline
-3. 检查是否有待计算指标 → 有则选择 compute_metrics
-4. 所有指标都已计算完成 → 选择 finalize_report
-5. 如果无法理解需求 → 选择 clarify_requirements
-
-### 重要原则
-- 大纲草稿已存在时，不要重复生成大纲
-- 决策为 compute_metrics 时，必须从状态信息中的"有效待计算指标ID列表"中选择
-- 确保 metrics_to_compute 是字符串数组格式
-- 确保指标ID与大纲中的global_metrics.metric_id完全一致
-- 从状态信息中的"有效待计算指标ID列表"中提取metric_id作为metrics_to_compute的值
-- 计算失败的指标可以重试最多3次
-- 绝对不要自己生成新的指标ID，必须严格使用状态信息中提供的已有指标ID
-- 如果状态信息中没有可用的指标ID，不要生成compute_metrics决策
-
-### 输出字段说明
-- decision: 决策字符串
-- reasoning: 决策原因说明
-- metrics_to_compute: 待计算指标ID列表，必须从状态信息中的"有效待计算指标ID列表"中选择。选择所有可用指标，除非指标数量过多（>10个）需要分批计算
-- priority_metrics: 优先级指标列表（前2-3个最重要的指标），从metrics_to_compute中选择
-
-必须输出有效的JSON格式！"""),
+            ("system", """
+            你是报告规划总控智能体，核心职责是精准分析当前状态并决定下一步行动。
+
+            ### 决策选项（三选一）
+            1. data_classify: 数据未分类打标或分类打标数据集数量为0
+            2. generate_outline：大纲未生成或大纲无效
+            3. compute_metrics：大纲已生成但指标未完成
+
+            ### 决策规则（按顺序检查）
+            1. 检查 data_set_classified 是否为空 或 数量为0时 → 选择 data_classify
+            2. 检查 outline_draft 是否为空 → 空则选择 generate_outline
+            3. 检查 metrics_requirements 是否为空 → 空则选择 generate_outline
+            4. 检查是否有待计算指标 → 有则选择 compute_metrics
+            5. 所有指标都已计算完成 → 选择 finalize_report
+            6. 如果无法理解需求 → 选择 clarify_requirements
+
+            ### 重要原则
+            - 数据已分类打标后，不要重复分类打标
+            - 大纲草稿已存在时，不要重复生成大纲
+            - 决策为 compute_metrics 时，必须从状态信息中的"有效待计算指标ID列表"中选择
+            - 确保 metrics_to_compute 是字符串数组格式
+            - 确保指标ID与大纲中的global_metrics.metric_id完全一致
+            - 从状态信息中的"有效待计算指标ID列表"中提取metric_id作为metrics_to_compute的值
+            - 计算失败的指标可以重试最多3次
+            - 绝对不要自己生成新的指标ID，必须严格使用状态信息中提供的已有指标ID
+            - 如果状态信息中没有可用的指标ID，不要生成compute_metrics决策
+
+            ### 输出字段说明
+            - decision: 决策字符串
+            - reasoning: 决策原因说明
+            - metrics_to_compute: 待计算指标ID列表，必须从状态信息中的"有效待计算指标ID列表"中选择。选择所有可用指标，除非指标数量过多（>10个）需要分批计算
+            - priority_metrics: 优先级指标列表（前2-3个最重要的指标），从metrics_to_compute中选择
+
+            必须输出有效的JSON格式！"""),
 
             MessagesPlaceholder("messages"),
 
@@ -153,7 +156,8 @@ class PlanningAgent:
 
         # 执行规划
         start_time = datetime.now()
-        response = await planner.ainvoke({
+        response = await
+        planner.ainvoke({
             "question": question,
             "industry": industry,
             "messages": [("system", status_info)]
@@ -198,7 +202,7 @@ class PlanningAgent:
                         raise ValueError("AI决策缺少具体的指标ID")
                     # 如果AI生成的指标ID明显是错误的（比如metric_001），使用默认逻辑
                     if any(mid.startswith("metric_") and mid.replace("metric_", "").isdigit()
-                          for mid in decision.metrics_to_compute):
+                           for mid in decision.metrics_to_compute):
                         raise ValueError("AI生成的指标ID格式不正确")
 
             else:
@@ -246,7 +250,8 @@ class PlanningAgent:
             print(f"[ERROR] 保存API结果文件失败: {filepath}, 错误: {str(e)}")
 
         # 记录大模型输出
-        print(f"[MODEL_OUTPUT] PlanningAgent: {json.dumps(decision.dict() if hasattr(decision, 'dict') else decision, ensure_ascii=False)}")
+        print(
+            f"[MODEL_OUTPUT] PlanningAgent: {json.dumps(decision.dict() if hasattr(decision, 'dict') else decision, ensure_ascii=False)}")
         print("========================================")
 
         return decision
@@ -273,20 +278,21 @@ class PlanningAgent:
         outline_draft = state.get('outline_draft')
         if outline_draft and outline_draft.global_metrics:
             available_metric_ids = [m.metric_id for m in outline_draft.global_metrics if m.metric_id]
-        
-
-        return f"""当前状态评估：
-- 规划步骤: {state.get('planning_step', 0)}
-- 大纲版本: {state.get('outline_version', 0)}
-- 大纲草稿存在: {state.get('outline_draft') is not None}
-- 指标需求总数: {required_count}
-- 已计算指标数: {computed_count}
-- 指标覆盖率: {coverage:.2%}
-- 待计算指标数: {len(pending_ids)}
-- 有效待计算指标ID列表: {filtered_pending_ids}
-- 可用指标ID列表: {available_metric_ids}
-- 失败尝试记录: {failed_attempts}
-"""
+
+        return f"""
+            当前状态评估：
+            - 规划步骤: {state.get('planning_step', 0)}
+            - 数据分类打标数量: {len(state.get('data_set_classified', 0))}
+            - 大纲版本: {state.get('outline_version', 0)}
+            - 大纲草稿存在: {state.get('outline_draft') is not None}
+            - 指标需求总数: {required_count}
+            - 已计算指标数: {computed_count}
+            - 指标覆盖率: {coverage:.2%}
+            - 待计算指标数: {len(pending_ids)}
+            - 有效待计算指标ID列表: {filtered_pending_ids}
+            - 可用指标ID列表: {available_metric_ids}
+            - 失败尝试记录: {failed_attempts}
+            """
 
 
 def analyze_current_state(state: Dict[str, Any]) -> Dict[str, Any]:
@@ -336,7 +342,8 @@ def analyze_current_state(state: Dict[str, Any]) -> Dict[str, Any]:
     }
 
 
-async def plan_next_action(question: str, industry: str, current_state: Dict[str, Any], api_key: str) -> PlanningDecision:
+async def plan_next_action(question: str, industry: str, current_state: Dict[str, Any],
+                           api_key: str) -> PlanningDecision:
     """
     规划下一步行动的主函数
 
@@ -351,7 +358,8 @@ async def plan_next_action(question: str, industry: str, current_state: Dict[str
     agent = PlanningAgent(api_key)
 
     try:
-        decision = await agent.make_decision(question, industry, current_state)
+        decision = await
+        agent.make_decision(question, industry, current_state)
 
         print(f"\n🧠 规划决策：{decision.decision}")
         print(f"   推理：{decision.reasoning[:100]}...")
@@ -373,4 +381,3 @@ async def plan_next_action(question: str, industry: str, current_state: Dict[str
             priority_metrics=[]
         )
 
-