tx_flow_analysis/llmops/workflow_state.py

"""
整合的工作流状态定义
===================

此文件定义了整合了多个Agent的工作流状态，兼容现有的Big Agent状态管理和新增的报告生成Agent状态。

状态层次：
1. 输入层：用户查询和数据
2. 意图层：意图识别结果
3. 规划层：规划决策和大纲生成
4. 计算层：指标计算结果
5. 结果层：最终报告生成
6. 对话层：消息历史和错误处理

兼容性：
- 兼容现有的Big Agent WorkflowState
- 整合来自other_agents的AgentState
- 支持扩展新的Agent状态需求

作者: Big Agent Team
版本: 1.0.0
创建时间: 2024-12-20
"""

from typing import TypedDict, List, Dict, Any, Optional
from datetime import datetime
from langchain_core.messages import BaseMessage
from pydantic import BaseModel, Field


# ============= 数据模型 =============

class MetricRequirement(BaseModel):
    """指标需求定义"""
    metric_id: str = Field(description="指标唯一标识，如 'total_income_jan'")
    metric_name: str = Field(description="指标中文名称")
    calculation_logic: str = Field(description="计算逻辑描述")
    required_fields: List[str] = Field(description="所需字段")
    dependencies: List[str] = Field(default_factory=list, description="依赖的其他指标ID")


class ReportSection(BaseModel):
    """报告大纲章节"""
    section_id: str = Field(description="章节ID")
    title: str = Field(description="章节标题")
    description: str = Field(description="章节内容要求")
    metrics_needed: List[str] = Field(description="所需指标ID列表")


class ReportOutline(BaseModel):
    """完整报告大纲"""
    report_title: str = Field(description="报告标题")
    sections: List[ReportSection] = Field(description="章节列表")
    global_metrics: List[MetricRequirement] = Field(description="全局指标列表")


# ============= 序列化工具函数 =============

def convert_numpy_types(obj: Any) -> Any:
    """
    递归转换所有numpy类型为Python原生类型
    确保所有数据可序列化
    """
    if isinstance(obj, dict):
        return {str(k): convert_numpy_types(v) for k, v in obj.items()}
    elif isinstance(obj, list):
        return [convert_numpy_types(item) for item in obj]
    elif isinstance(obj, tuple):
        return tuple(convert_numpy_types(item) for item in obj)
    elif isinstance(obj, set):
        return {convert_numpy_types(item) for item in obj}
    elif hasattr(obj, 'item') and hasattr(obj, 'dtype'):  # numpy scalar
        return convert_numpy_types(obj.item())
    else:
        return obj


# ============= 整合的工作流状态定义 =============

class IntegratedWorkflowState(TypedDict):
    """整合的工作流状态定义 - 兼容多个Agent系统"""

    # === 基础输入层 (兼容Big Agent) ===
    user_input: str
    question: str  # 别名，兼容报告生成Agent

    industry: str  # 行业

    # === 数据层 ===
    data_set: List[Dict[str, Any]]  # 报告生成Agent的数据格式
    transactions_df: Optional[Any]  # 可选的数据框格式
    file_name: str                  # 数据文件名称
    data_set_classified: List[Dict[str, Any]] # 分类打标后的数据集

    # === 意图识别层 (Big Agent原有) ===
    intent_result: Optional[Dict[str, Any]]

    # === 规划和大纲层 (新增) ===
    planning_step: int
    plan_history: List[str]
    outline_draft: Optional[ReportOutline]
    outline_version: int
    outline_ready: bool

    # === 指标计算层 ===
    metrics_requirements: List[MetricRequirement]  # 报告生成Agent格式
    computed_metrics: Dict[str, Any]  # 计算结果
    metrics_cache: Dict[str, Any]  # 缓存
    pending_metric_ids: List[str]  # 待计算指标ID
    failed_metric_attempts: Dict[str, int]  # 失败统计
    calculation_results: Optional[Dict[str, Any]]  # Big Agent格式的计算结果

    # === 结果层 ===
    report_draft: Dict[str, Any]  # 报告草稿
    knowledge_result: Optional[Dict[str, Any]]  # Big Agent知识沉淀结果
    is_complete: bool
    completeness_score: float
    answer: Optional[str]  # 最终答案

    # === 对话和消息层 ===
    messages: List[Dict[str, Any]]  # Big Agent消息格式
    current_node: str
    session_id: str
    next_route: str

    # === 错误处理层 ===
    errors: List[str]
    last_decision: str

    # === 时间跟踪层 ===
    start_time: str
    end_time: Optional[str]
    api_result: Dict[str, Any]  # 存储所有API调用结果


# ============= 状态创建和初始化函数 =============

def create_initial_integrated_state(question: str, industry: str, data: List[Dict[str, Any]], file_name: str, session_id: str = None) -> IntegratedWorkflowState:
    """
    创建初始的整合状态

    Args:
        question: 用户查询
        industry: 行业
        data: 数据集
        file_name: 数据文件名称
        session_id: 会话ID

    Returns:
        初始化后的状态
    """
    current_time = datetime.now().isoformat()
    session = session_id or f"session_{int(datetime.now().timestamp())}"

    return {
        # 基础输入
        "user_input": question,
        "question": question,
        "industry": industry,

        # 数据层
        "data_set": convert_numpy_types(data),
        "data_set_classified": [],    # 分类打标后的数据集
        "transactions_df": None,
        "file_name": file_name,       # 文件名称

        # 意图识别层
        "intent_result": None,

        # 规划和大纲层
        "planning_step": 0,
        "plan_history": [],
        "outline_draft": None,
        "outline_version": 0,
        "outline_ready": False,

        # 指标计算层
        "metrics_requirements": [],
        "computed_metrics": {},
        "metrics_cache": {},
        "pending_metric_ids": [],
        "failed_metric_attempts": {},
        "calculation_results": None,

        # 结果层
        "report_draft": {},
        "knowledge_result": None,
        "is_complete": False,
        "completeness_score": 0.0,
        "answer": None,

        # 对话和消息层
        "messages": [{
            "role": "user",
            "content": question,
            "timestamp": current_time
        }],
        "current_node": "start",
        "session_id": session,
        "next_route": "planning_node",

        # 错误处理层
        "errors": [],
        "last_decision": "init",

        # 时间跟踪层
        "start_time": current_time,
        "end_time": None,
        "api_result": {},  # 存储所有API调用结果

        # 计算模式配置层
        "use_rules_engine_only": False,
        "use_traditional_engine_only": False
    }


def is_state_ready_for_calculation(state: IntegratedWorkflowState) -> bool:
    """
    检查状态是否准备好进行指标计算

    Args:
        state: 当前状态

    Returns:
        是否准备好
    """
    return (
        state.get("outline_draft") is not None and
        len(state.get("metrics_requirements", [])) > 0 and
        len(state.get("pending_metric_ids", [])) > 0
    )


def get_calculation_progress(state: IntegratedWorkflowState) -> Dict[str, Any]:
    """
    获取指标计算进度信息

    Args:
        state: 当前状态

    Returns:
        进度信息
    """
    required = len(state.get("metrics_requirements", []))
    computed = len(state.get("computed_metrics", {}))
    pending = len(state.get("pending_metric_ids", []))

    return {
        "required_count": required,
        "computed_count": computed,
        "pending_count": pending,
        "coverage_rate": computed / required if required > 0 else 0,
        "is_complete": computed >= required * 0.8  # 80%覆盖率视为完成
    }


def update_state_with_outline_generation(state: IntegratedWorkflowState, outline: ReportOutline) -> IntegratedWorkflowState:
    """
    使用大纲生成结果更新状态

    Args:
        state: 当前状态
        outline: 生成的大纲

    Returns:
        更新后的状态
    """
    new_state = state.copy()
    new_state["outline_draft"] = outline
    new_state["outline_version"] += 1
    new_state["outline_ready"] = True
    new_state["metrics_requirements"] = outline.global_metrics
    new_state["pending_metric_ids"] = [m.metric_id for m in outline.global_metrics]

    # 添加消息
    new_state["messages"].append({
        "role": "assistant",
        "content": f"✅ 大纲生成完成 v{new_state['outline_version']}：{outline.report_title}",
        "timestamp": datetime.now().isoformat()
    })

    return new_state


def update_state_with_planning_decision(state: IntegratedWorkflowState, decision: Dict[str, Any]) -> IntegratedWorkflowState:
    """
    使用规划决策结果更新状态

    Args:
        state: 当前状态
        decision: 规划决策

    Returns:
        更新后的状态
    """
    new_state = state.copy()
    new_state["planning_step"] += 1
    new_state["last_decision"] = decision.get("decision", "unknown")
    new_state["next_route"] = decision.get("next_route", "planning_node")

    # 如果有待计算指标，更新待计算列表
    if decision.get("metrics_to_compute"):
        new_state["pending_metric_ids"] = decision["metrics_to_compute"]

    # 添加规划历史
    new_state["plan_history"].append(
        f"Step {new_state['planning_step']}: {decision.get('decision', 'unknown')}"
    )

    return new_state


def finalize_state_with_report(state: IntegratedWorkflowState, final_report: Dict[str, Any]) -> IntegratedWorkflowState:
    """
    使用最终报告完成状态

    Args:
        state: 当前状态
        final_report: 最终报告

    Returns:
        完成的状态
    """
    new_state = state.copy()
    new_state["report_draft"] = final_report
    new_state["is_complete"] = True
    new_state["answer"] = final_report
    new_state["end_time"] = datetime.now().isoformat()

    # 计算完整性分数
    progress = get_calculation_progress(new_state)
    new_state["completeness_score"] = progress["coverage_rate"]

    return new_state

def update_state_with_data_classified(state: IntegratedWorkflowState, data_set_classified: List[Dict]) -> IntegratedWorkflowState:
    """
    使用分类打标结果更新状态

    Args:
        state: 当前状态
        data_set_classified: 分类打标的数据

    Returns:
        更新后的状态
    """
    new_state = state.copy()
    new_state["data_set_classified"] = data_set_classified

    # 添加消息
    new_state["messages"].append({
        "role": "assistant",
        "content": f"✅ 数据分类打标已完成",
        "timestamp": datetime.now().isoformat()
    })

    return new_state