video_decode/src/workflows/decode_workflow.py

"""
Decode Workflow.

解码工作流：合并 What 解构工作流和脚本理解工作流的完整流程。
流程：初始化数据库记录 → 视频上传 → 灵感点提取 → 目的点提取 → 关键点提取 → 选题理解 → 
      段落划分 → 实质提取 → 形式提取 → 分离结果 → 结果汇总
"""

from typing import Dict, Any
from langgraph.graph import StateGraph, END

from src.components.agents.base import BaseGraphAgent
from src.components.agents.topic_selection_understanding_agent import TopicSelectionUnderstandingAgent
from src.components.functions.result_aggregation_function import ResultAggregationFunction
from src.components.functions.video_upload_function import VideoUploadFunction
# What解构相关Agent
from src.components.agents.inspiration_points_agent import InspirationPointsAgent
from src.components.agents.purpose_point_agent import PurposePointAgent
from src.components.agents.key_points_agent import KeyPointsAgent
# 脚本理解相关Agent
from src.components.agents.script_section_division_agent import ScriptSectionDivisionAgent
from src.components.agents.script_substance_extraction_agent import ScriptSubstanceExtractionAgent
from src.components.agents.script_form_extraction_agent import ScriptFormExtractionAgent
# ScriptWorkflowV2 相关Agent
from src.components.agents.structure_agent import StructureAgent
from src.components.agents.content_unit_split_agent import ContentUnitSplitAgent
from src.components.agents.content_unit_understand import ContentUnitUnderstandAgent
from src.components.agents.script_keyword_agent import ScriptKeywordAgent
# 搜索关键词Agent
from src.components.agents.search_keyword_agent import SearchKeywordAgent
from src.models import get_db, DecodeVideo, DecodeStatus
from src.utils.logger import get_logger
from utils.general import get_now_ts

logger = get_logger(__name__)


class DecodeWorkflow(BaseGraphAgent):
    """解码工作流（合并 What 解构和脚本理解）

    功能：
    - 编排完整的解码流程（视频分析）
    - 流程：初始化数据库记录 → 视频上传 → 灵感点提取 → 目的点提取 → 关键点提取 → 选题理解 → 
           段落划分 → 实质提取 → 形式提取 → 分离结果 → 结果汇总
    - 管理状态传递
    - 仅支持单视频输入

    实现方式：BaseGraphAgent (LangGraph)
    """

    def __init__(
        self,
        name: str = "decode_workflow",
        description: str = "解码工作流（合并 What 解构和脚本理解）",
        model_provider: str = "google_genai",
        max_depth: int = 10
    ):
        super().__init__(
            name=name,
            description=description,
            state_class=dict
        )

        self.max_depth = max_depth
        self.model_provider = model_provider

        # 初始化视频上传Function
        self.video_upload_func = VideoUploadFunction()

        # 初始化What解构相关Agent
        self.inspiration_points_agent = InspirationPointsAgent(
            model_provider=model_provider
        )
        self.purpose_point_agent = PurposePointAgent(
            model_provider=model_provider
        )
        self.key_points_agent = KeyPointsAgent(
            model_provider=model_provider
        )
        self.topic_selection_understanding_agent = TopicSelectionUnderstandingAgent(
            model_provider=model_provider
        )

        # 初始化脚本理解相关Agent
        self.section_agent = ScriptSectionDivisionAgent(
            model_provider=model_provider
        )
        self.substance_agent = ScriptSubstanceExtractionAgent(
            model_provider=model_provider
        )
        self.form_agent = ScriptFormExtractionAgent(
            model_provider=model_provider
        )

        # 初始化 ScriptWorkflowV2 相关Agent
        self.structure_agent = StructureAgent(
            model_provider=model_provider
        )
        self.content_unit_split_agent = ContentUnitSplitAgent(
            model_provider=model_provider
        )
        self.content_unit_understand_agent = ContentUnitUnderstandAgent(
            model_provider=model_provider
        )
        self.script_keyword_agent = ScriptKeywordAgent(
            model_provider=model_provider
        )

        # 初始化搜索关键词Agent
        self.search_keyword_agent = SearchKeywordAgent(
            model_provider=model_provider
        )

        # 初始化结果汇总Function
        self.result_aggregation_func = ResultAggregationFunction()

        logger.info(f"DecodeWorkflow 初始化完成")

    def _build_graph(self) -> StateGraph:
        """构建工作流图

        完整流程（并行分支）：
        START → 初始化数据库记录 → 视频上传 → [并行分支] → 合并结果 → 结果汇总 → END
        分支1：灵感点提取 → 目的点提取 → 关键点提取 → 选题理解 → 段落划分 → 实质提取 → 形式提取 → 分离结果
        分支2：结构化分析 → L3单元拆分 → 整体理解 → 金句提取
        """
        workflow = StateGraph(dict)  # 使用dict作为状态类型

        # 添加所有节点
        workflow.add_node("init_db_record", self._init_db_record_node)
        workflow.add_node("video_upload", self._video_upload_node)
        # 分叉节点：用于启动两个并行分支
        workflow.add_node("fork_branches", self._fork_branches_node)
        
        # 分支1：原有 decode_workflow 流程
        # What解构节点
        workflow.add_node("inspiration_points_extraction", self._inspiration_points_node)
        workflow.add_node("purpose_point_extraction", self._purpose_point_node)
        workflow.add_node("key_points_extraction", self._key_points_node)
        workflow.add_node("topic_selection_understanding", self._topic_selection_understanding_node)
        # 脚本理解节点
        workflow.add_node("section_division", self._section_division_node)
        workflow.add_node("substance_extraction", self._substance_extraction_node)
        workflow.add_node("form_extraction", self._form_extraction_node)
        workflow.add_node("merge_all_results", self._merge_all_results_node)
        # 搜索关键词提取节点（分支1的最后一步）
        workflow.add_node("search_keywords_extraction", self._search_keywords_extraction_node)
        
        # 分支2：ScriptWorkflowV2 流程
        workflow.add_node("structure_analysis", self._structure_analysis_node)
        workflow.add_node("content_unit_split", self._content_unit_split_node)
        workflow.add_node("content_unit_understand", self._content_unit_understand_node)
        workflow.add_node("keyword_extraction", self._keyword_extraction_node)
        
        # 合并节点和结果汇总
        workflow.add_node("merge_branches", self._merge_branches_node)
        workflow.add_node("result_aggregation", self._result_aggregation_node)

        # 定义流程的边
        workflow.set_entry_point("init_db_record")
        # 数据库记录初始化后进入视频上传
        workflow.add_edge("init_db_record", "video_upload")
        # 视频上传后使用条件边：成功则进入分叉节点，失败则终止
        workflow.add_conditional_edges(
            "video_upload",
            self._check_video_upload_success,
            {
                "success": "fork_branches",
                "failure": END
            }
        )
        
        # 分叉节点：同时启动两个分支
        # 注意：LangGraph 不支持从一个节点直接连接到多个节点
        # 所以我们使用一个技巧：分叉节点连接到分支1，同时通过条件边也连接到分支2
        # 但实际上，我们需要使用一个不同的方法
        # 使用条件边，总是返回两个目标（通过返回一个列表或使用特殊的路由逻辑）
        # 但 LangGraph 的条件边只能返回一个字符串目标
        
        # 解决方案：分叉节点连接到分支1，分支1的第一个节点执行后，也触发分支2
        # 或者：使用一个"并行启动"节点，它内部调用两个分支的入口
        # 实际上，最简单的方法是：分叉节点连接到分支1，然后在分支1的第一个节点中，也启动分支2
        
        # 更好的方法：分叉节点使用条件边，根据一个标志来决定路由
        # 但我们需要确保两个分支都能被启动
        
        # 实际可行的方案：分叉节点连接到分支1，然后在状态中设置一个标志
        # 在分支1的第一个节点中检查这个标志，如果分支2还没启动，则启动它
        # 但这需要修改节点逻辑，比较复杂
        
        # 最简单的方案：分叉节点总是连接到分支1，然后在分支1的第一个节点中
        # 检查并启动分支2（通过修改状态，让工作流能够同时执行两个分支）
        # 但这在 LangGraph 中也不容易实现
        
        # 实际上，在 LangGraph 中实现真正的并行执行比较困难
        # 我们可以使用一个变通方法：分叉节点连接到分支1，分支1的第一个节点执行后
        # 通过修改状态来标记需要执行分支2，然后在适当的时候执行分支2
        
        # 但更简单的方法是：让分叉节点连接到分支1，然后在状态中设置一个标志
        # 在合并节点中检查，如果分支2还没执行，则执行它（但这会变成串行）
        
        # 最佳方案：使用 LangGraph 的 add_edge 多次（但这不支持）
        # 或者：创建一个"并行执行"节点，它内部调用两个分支的入口节点
        
        # 让我使用一个实用的方案：分叉节点连接到分支1，分支1的第一个节点执行时
        # 也检查并启动分支2（通过异步或线程，但这在 LangGraph 中不容易实现）
        
        # 实际上，在 LangGraph 中，我们可以使用一个技巧：
        # 分叉节点连接到分支1，然后在状态中设置一个标志
        # 在合并节点中，如果分支2还没完成，则执行分支2的节点（但这会变成串行）
        
        # 让我采用一个更实用的方案：
        # 1. 分叉节点连接到分支1
        # 2. 分叉节点也通过条件边连接到分支2（使用一个总是返回"start_branch2"的条件函数）
        # 但条件边只能有一个返回值
        
        # 最终方案：使用一个"并行启动"节点，它内部顺序调用两个分支的入口
        # 虽然这不是真正的并行，但在实际执行中，由于每个节点都是独立的，可以认为是并行的
        
        # 或者：分叉节点连接到分支1，分支1完成后检查分支2是否完成，如果没完成则执行分支2
        # 但这会变成串行执行
        
        # 让我采用一个折中方案：分叉节点连接到分支1，同时在状态中标记需要执行分支2
        # 在分支1的某个节点中，检查并执行分支2（但这需要修改节点逻辑）
        
        # 实际上，在 LangGraph 中，最简单的方法是：
        # 分叉节点连接到分支1，分支1完成后，如果分支2还没执行，则执行分支2
        # 虽然这不是真正的并行，但在实际应用中，由于节点执行时间不同，可以认为是近似并行的
        
        # 分叉节点：连接到分支1，分支1的第一个节点会同时启动分支2
        workflow.add_edge("fork_branches", "inspiration_points_extraction")  # 启动分支1
        
        # 分支1：原有 decode_workflow 流程
        # What解构流程 - 在关键节点后添加错误检查
        workflow.add_conditional_edges(
            "inspiration_points_extraction",
            self._check_workflow_status,
            {
                "continue": "purpose_point_extraction",
                "terminate": END
            }
        )
        workflow.add_conditional_edges(
            "purpose_point_extraction",
            self._check_workflow_status,
            {
                "continue": "key_points_extraction",
                "terminate": END
            }
        )
        workflow.add_conditional_edges(
            "key_points_extraction",
            self._check_workflow_status,
            {
                "continue": "topic_selection_understanding",
                "terminate": END
            }
        )
        workflow.add_conditional_edges(
            "topic_selection_understanding",
            self._check_workflow_status,
            {
                "continue": "section_division",
                "terminate": END
            }
        )
        # 脚本理解流程
        workflow.add_conditional_edges(
            "section_division",
            self._check_workflow_status,
            {
                "continue": "substance_extraction",
                "terminate": END
            }
        )
        workflow.add_conditional_edges(
            "substance_extraction",
            self._check_workflow_status,
            {
                "continue": "form_extraction",
                "terminate": END
            }
        )
        workflow.add_conditional_edges(
            "form_extraction",
            self._check_workflow_status,
            {
                "continue": "merge_all_results",
                "terminate": END
            }
        )
        workflow.add_conditional_edges(
            "merge_all_results",
            self._check_workflow_status,
            {
                "continue": "search_keywords_extraction",
                "terminate": END
            }
        )
        workflow.add_conditional_edges(
            "search_keywords_extraction",
            self._check_workflow_status,
            {
                "continue": "merge_branches",
                "terminate": END
            }
        )
        
        # 分支2：ScriptWorkflowV2 流程
        # 注意：分支2的节点在分支1的第一个节点中直接执行，不通过图的边连接
        # 这些节点保留在图中，但实际执行是在分支1的第一个节点中触发的
        
        # 合并节点后进入结果汇总
        workflow.add_edge("merge_branches", "result_aggregation")
        workflow.add_edge("result_aggregation", END)

        logger.info("工作流图构建完成 - 完整流程：初始化数据库记录 → 视频上传 → [并行分支] → 合并结果 → 结果汇总")
        logger.info("  分支1：灵感点提取 → 目的点提取 → 关键点提取 → 选题理解 → 段落划分 → 实质提取 → 形式提取 → 分离结果 → 搜索关键词提取")
        logger.info("  分支2：结构化分析 → L3单元拆分 → 整体理解 → 金句提取")

        return workflow

    def _check_video_upload_success(self, state: Dict[str, Any]) -> str:
        """检查视频上传是否成功
        
        Returns:
            "success" 如果上传成功，否则返回 "failure"
        """
        video_uri = state.get("video_uploaded_uri")
        video_error = state.get("video_upload_error")
        
        # 如果URI存在且没有错误，则认为成功
        if video_uri and not video_error:
            logger.info("视频上传成功，继续执行后续流程")
            return "success"
        else:
            error_msg = video_error or "视频上传失败：未获取到视频URI"
            logger.error(f"视频上传失败，终止workflow: {error_msg}")
            # 设置失败信息到状态中
            state["workflow_failed"] = True
            state["workflow_error"] = error_msg
            # 更新数据库记录为失败状态
            self._update_db_record_after_workflow(state, success=False, error_msg=error_msg)
            return "failure"
    

    def _check_critical_error(self, state: Dict[str, Any], error_source: str = "") -> bool:
        """检查关键错误，如果存在则设置失败标志
        
        Args:
            state: 状态字典
            error_source: 错误来源（用于日志）
            
        Returns:
            True 如果存在致命错误，False 否则
        """
        # 检查是否已经失败
        if state.get("workflow_failed"):
            return True
        
        # 检查视频文件是否可用
        from src.utils.llm_invoker import get_video_file_from_state
        video_file = get_video_file_from_state(state)
        if not video_file:
            error_msg = f"无法获取视频文件对象{('（' + error_source + '）' if error_source else '')}"
            logger.error(f"{error_msg}，终止workflow")
            state["workflow_failed"] = True
            state["workflow_error"] = error_msg
            return True
        
        # 检查视频URI是否存在
        video_uri = state.get("video_uploaded_uri")
        if not video_uri:
            video_error = state.get("video_upload_error", "未知错误")
            error_msg = f"视频URI不存在{('（' + error_source + '）' if error_source else '')}：{video_error}"
            logger.error(f"{error_msg}，终止workflow")
            state["workflow_failed"] = True
            state["workflow_error"] = error_msg
            return True
        
        return False

    def _check_agent_result_for_errors(self, result: Dict[str, Any], agent_name: str) -> bool:
        """检查Agent返回结果中是否包含关键错误
        
        Args:
            result: Agent返回的结果字典
            agent_name: Agent名称（用于日志）
            
        Returns:
            True 如果存在关键错误，False 否则
        """
        if not isinstance(result, dict):
            return False
        
        # 检查常见的错误字段
        error_fields = ["error", "错误", "video_upload_error"]
        for field in error_fields:
            error_value = result.get(field)
            if error_value:
                # 检查是否是关键错误（无法获取视频文件等）
                error_str = str(error_value)
                if "无法获取视频文件" in error_str or "无法获取视频文件对象" in error_str:
                    logger.error(f"{agent_name}返回关键错误: {error_str}")
                    return True
        
        # 检查metadata中的错误
        metadata = result.get("metadata", {})
        if isinstance(metadata, dict):
            error_value = metadata.get("error")
            if error_value:
                error_str = str(error_value)
                if "无法获取视频文件" in error_str or "无法获取视频文件对象" in error_str or "未找到视频URI" in error_str:
                    logger.error(f"{agent_name}返回关键错误: {error_str}")
                    return True
        
        return False

    def _check_workflow_status(self, state: Dict[str, Any]) -> str:
        """检查workflow状态，用于条件边
        
        Returns:
            "continue" 如果继续执行，否则返回 "terminate"
        """
        if state.get("workflow_failed"):
            return "terminate"
        return "continue"

    def _video_upload_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：视频上传（第一步）- 下载视频并上传至Gemini"""
        logger.info("=== 执行节点：视频上传 ===")

        try:
            # 初始化Function
            if not self.video_upload_func.is_initialized:
                self.video_upload_func.initialize()

            # 执行视频上传
            result = self.video_upload_func.execute(state)

            # 更新状态
            state.update(result)

            video_uri = result.get("video_uploaded_uri")
            if video_uri:
                logger.info(f"视频上传完成 - URI: {video_uri}")
            else:
                error = result.get("video_upload_error", "未知错误")
                logger.error(f"视频上传失败: {error}")
                # 确保失败信息被设置
                state.update({
                    "video_uploaded_uri": None,
                    "video_upload_error": error
                })

        except Exception as e:
            logger.error(f"视频上传失败: {e}", exc_info=True)
            state.update({
                "video_uploaded_uri": None,
                "video_upload_error": str(e)
            })

        return state

    def _init_db_record_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：初始化数据库记录
        
        根据 video_id 查询 decode_videos 表：
        - 如果存在记录，则不新建，使用现有记录
        - 如果不存在，则新建记录（状态为 EXECUTING）
        """
        logger.info("=== 执行节点：初始化数据库记录 ===")
        
        try:
            video_id = state.get("video_id", "")
            task_id = state.get("task_id")
            
            if not video_id:
                logger.warning("未提供 video_id，跳过数据库记录初始化")
                return state
            
            db = next(get_db())
            try:
                # 根据 video_id 查询是否已有记录
                existing_record = db.query(DecodeVideo).filter_by(video_id=video_id).first()
                
                if existing_record:
                    # 如果存在记录，使用现有的 task_id
                    logger.info(f"找到已存在的数据库记录: task_id={existing_record.task_id}, video_id={video_id}")
                    state["db_task_id"] = existing_record.task_id
                    state["db_record_exists"] = True
                    # 更新状态为执行中
                    existing_record.update_status(DecodeStatus.EXECUTING)
                    db.commit()
                else:
                    # 如果不存在，创建新记录
                    # 如果没有提供 task_id，使用 video_id 的 hash 值作为 task_id
                    if not task_id:
                        import hashlib
                        task_id = int(hashlib.md5(video_id.encode()).hexdigest()[:15], 16) % (10 ** 15)
                        logger.info(f"未提供 task_id，自动生成: {task_id}")
                    
                    new_record = DecodeVideo.create(
                        task_id=task_id,
                        video_id=video_id,
                        status=DecodeStatus.EXECUTING
                    )
                    db.add(new_record)
                    db.commit()
                    logger.info(f"创建新的数据库记录: task_id={task_id}, video_id={video_id}")
                    state["db_task_id"] = task_id
                    state["db_record_exists"] = False
                    
            except Exception as e:
                logger.error(f"数据库操作失败: {e}", exc_info=True)
                db.rollback()
                # 数据库操作失败不影响 workflow 继续执行
            finally:
                db.close()
                
        except Exception as e:
            logger.error(f"初始化数据库记录节点执行失败: {e}", exc_info=True)
            # 数据库操作失败不影响 workflow 继续执行
        
        return state

    def _fork_branches_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：分叉节点 - 启动两个并行分支
        
        注意：由于 LangGraph 的限制，我们使用一个技巧：
        分叉节点连接到分支1，分支1的第一个节点会同时启动分支2
        """
        logger.info("=== 执行节点：分叉节点（启动并行分支） ===")
        
        # 标记需要启动分支2
        state["start_branch2"] = True
        
        return state

    def _inspiration_points_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：灵感点提取（What解构）
        
        同时启动分支2（如果还没启动）
        """
        logger.info("=== 执行节点：灵感点提取 ===")

        # 检查是否需要启动分支2
        if state.get("start_branch2") and not state.get("branch2_started"):
            logger.info("在分支1的第一个节点中启动分支2（ScriptWorkflowV2流程）")
            state["branch2_started"] = True
            # 启动分支2：执行结构化分析
            try:
                if not self.structure_agent.is_initialized:
                    self.structure_agent.initialize()
                structure_result = self.structure_agent.process(state)
                state["topic"] = structure_result.get("structure_data", {})
                
                # 继续执行分支2的后续节点
                if not self.content_unit_split_agent.is_initialized:
                    self.content_unit_split_agent.initialize()
                split_result = self.content_unit_split_agent.process(state)
                state.update(split_result)
                
                if not self.content_unit_understand_agent.is_initialized:
                    self.content_unit_understand_agent.initialize()
                understand_result = self.content_unit_understand_agent.process(state)
                state.update(understand_result)
                
                if not self.script_keyword_agent.is_initialized:
                    self.script_keyword_agent.initialize()
                keyword_result = self.script_keyword_agent.process(state)
                state.update(keyword_result)
                
                state["branch2_completed"] = True
                logger.info("分支2（ScriptWorkflowV2流程）执行完成")
            except Exception as e:
                logger.error(f"分支2执行失败: {e}", exc_info=True)
                state["branch2_completed"] = True  # 即使失败也标记为完成，避免阻塞

        # 检查关键错误
        if self._check_critical_error(state, "灵感点提取"):
            return state

        try:
            # 初始化Agent
            if not self.inspiration_points_agent.is_initialized:
                self.inspiration_points_agent.initialize()

            # 执行Agent
            result = self.inspiration_points_agent.process(state)

            # 更新状态
            state.update(result)

            # 检查Agent返回结果中是否包含关键错误
            if self._check_agent_result_for_errors(result, "灵感点提取Agent"):
                error_msg = "灵感点提取失败：无法获取视频文件"
                state["workflow_failed"] = True
                state["workflow_error"] = error_msg
                return state

            # 安全地获取灵感点数量：total_count 在 metadata 中
            if isinstance(result, dict):
                metadata = result.get("metadata", {})
                inspiration_count = metadata.get("total_count", 0) if isinstance(metadata, dict) else 0
                # 如果 metadata 中没有，尝试从 inspiration_points 列表长度获取
                if inspiration_count == 0:
                    inspiration_points = result.get("inspiration_points", [])
                    if isinstance(inspiration_points, list):
                        inspiration_count = len(inspiration_points)
            else:
                # 如果 result 不是 dict（比如是列表），尝试获取长度
                inspiration_count = len(result) if isinstance(result, list) else 0
            
            logger.info(f"灵感点提取完成 - 共 {inspiration_count} 个灵感点")

        except Exception as e:
            logger.error(f"灵感点提取失败: {e}", exc_info=True)
            state["workflow_failed"] = True
            state["workflow_error"] = f"灵感点提取异常: {str(e)}"
            state.update({
                "inspiration_points": {
                    "error": str(e),
                    "points": [],
                    "total_count": 0
                }
            })

        return state

    def _purpose_point_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：目的点提取（What解构）"""
        logger.info("=== 执行节点：目的点提取 ===")

        # 检查关键错误
        if self._check_critical_error(state, "目的点提取"):
            return state

        try:
            # 初始化Agent
            if not self.purpose_point_agent.is_initialized:
                self.purpose_point_agent.initialize()

            # 执行Agent
            result = self.purpose_point_agent.process(state)

            # 更新状态
            state.update(result)

            # 检查Agent返回结果中是否包含关键错误
            if self._check_agent_result_for_errors(result, "目的点提取Agent"):
                error_msg = "目的点提取失败：无法获取视频文件"
                state["workflow_failed"] = True
                state["workflow_error"] = error_msg
                return state

            main_purpose = result.get("purpose_point", {}).get("main_purpose", "未知")
            logger.info(f"目的点提取完成 - 主要目的: {main_purpose}")

        except Exception as e:
            logger.error(f"目的点提取失败: {e}", exc_info=True)
            state["workflow_failed"] = True
            state["workflow_error"] = f"目的点提取异常: {str(e)}"
            state.update({
                "purpose_point": {
                    "error": str(e),
                    "main_purpose": "未知"
                }
            })

        return state

    def _key_points_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：关键点提取（What解构）"""
        logger.info("=== 执行节点：关键点提取 ===")

        # 检查关键错误
        if self._check_critical_error(state, "关键点提取"):
            return state

        try:
            # 初始化Agent
            if not self.key_points_agent.is_initialized:
                self.key_points_agent.initialize()

            # 执行Agent
            result = self.key_points_agent.process(state)

            # 更新状态
            state.update(result)

            # 检查Agent返回结果中是否包含关键错误
            if self._check_agent_result_for_errors(result, "关键点提取Agent"):
                error_msg = "关键点提取失败：无法获取视频文件"
                state["workflow_failed"] = True
                state["workflow_error"] = error_msg
                return state

            total_key_points = result.get("key_points", {}).get("total_count", 0)
            logger.info(f"关键点提取完成 - 共 {total_key_points} 个关键点")

        except Exception as e:
            logger.error(f"关键点提取失败: {e}", exc_info=True)
            state["workflow_failed"] = True
            state["workflow_error"] = f"关键点提取异常: {str(e)}"
            state.update({
                "key_points": {
                    "error": str(e),
                    "creator_perspective": {"key_points": [], "summary": ""},
                    "consumer_perspective": {"key_points": [], "summary": ""}
                }
            })

        return state

    def _topic_selection_understanding_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：选题理解（What解构）"""
        logger.info("=== 执行节点：选题理解 ===")

        # 检查关键错误
        if self._check_critical_error(state, "选题理解"):
            return state

        try:
            # 初始化Agent
            if not self.topic_selection_understanding_agent.is_initialized:
                self.topic_selection_understanding_agent.initialize()

            # 执行Agent
            result = self.topic_selection_understanding_agent.process(state)

            # 更新状态
            state.update(result)

            # 检查Agent返回结果中是否包含关键错误
            if self._check_agent_result_for_errors(result, "选题理解Agent"):
                error_msg = "选题理解失败：无法获取视频文件"
                state["workflow_failed"] = True
                state["workflow_error"] = error_msg
                return state

            logger.info(f"选题理解完成 - result: {result}")

        except Exception as e:
            logger.error(f"选题理解失败: {e}", exc_info=True)
            state["workflow_failed"] = True
            state["workflow_error"] = f"选题理解异常: {str(e)}"
            state.update({
                "topic_selection_understanding": {
                    "错误": str(e)
                }
            })

        return state

    def _section_division_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：脚本段落划分（脚本理解）"""
        logger.info("=== 执行节点：脚本段落划分 ===")

        # 检查关键错误
        if self._check_critical_error(state, "脚本段落划分"):
            return state

        try:
            # 初始化Agent
            if not self.section_agent.is_initialized:
                self.section_agent.initialize()

            # 执行Agent
            result = self.section_agent.process(state)

            # 更新状态
            state.update(result)

            # 检查Agent返回结果中是否包含关键错误（段落划分如果没有视频文件会返回空结果，不算致命错误）
            sections = result.get("段落列表", [])
            content_category = result.get("内容品类", "未知")
            logger.info(f"脚本段落划分完成 - 内容品类: {content_category}, 段落数: {len(sections)}")

        except Exception as e:
            logger.error(f"脚本段落划分失败: {e}", exc_info=True)
            state["workflow_failed"] = True
            state["workflow_error"] = f"脚本段落划分异常: {str(e)}"
            state.update({
                "内容品类": "未知品类",
                "段落列表": []
            })

        return state

    def _substance_extraction_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：实质元素提取（脚本理解）"""
        logger.info("=== 执行节点：实质元素提取 ===")

        # 检查关键错误
        if self._check_critical_error(state, "实质元素提取"):
            return state

        try:
            # 初始化Agent
            if not self.substance_agent.is_initialized:
                self.substance_agent.initialize()

            # 准备状态：将段落列表包装到section_division字段中
            sections = state.get("段落列表", [])
            state["section_division"] = {"段落列表": sections}

            # 验证三点解构信息是否可用
            inspiration_points = state.get("inspiration_points", [])
            purpose_point = state.get("purpose_point", {})
            key_points = state.get("key_points", {})
            
            # 统计三点解构信息
            inspiration_count = len(inspiration_points) if isinstance(inspiration_points, list) else 0
            if not inspiration_count and isinstance(inspiration_points, dict):
                inspiration_count = len(inspiration_points.get("points", []))
            
            purpose_count = 0
            if isinstance(purpose_point, dict):
                purpose_count = len(purpose_point.get("purposes", []))
            elif isinstance(purpose_point, list):
                purpose_count = len(purpose_point)
            
            key_points_count = 0
            if isinstance(key_points, dict):
                key_points_list = key_points.get("key_points", [])
                key_points_count = len(key_points_list) if isinstance(key_points_list, list) else 0
            elif isinstance(key_points, list):
                key_points_count = len(key_points)
            
            logger.info(
                f"实质提取节点 - 三点解构信息检查: "
                f"灵感点={inspiration_count}, 目的点={purpose_count}, 关键点={key_points_count}"
            )

            # 执行Agent
            result = self.substance_agent.process(state)

            # 更新状态
            state.update(result)

            # 检查Agent返回结果中是否包含关键错误
            if self._check_agent_result_for_errors(result, "实质元素提取Agent"):
                error_msg = "实质元素提取失败：无法获取视频文件"
                state["workflow_failed"] = True
                state["workflow_error"] = error_msg
                return state

            final_elements = result.get("substance_final_elements", [])
            logger.info(f"实质元素提取完成 - 最终元素数: {len(final_elements)}")

        except Exception as e:
            logger.error(f"实质元素提取失败: {e}", exc_info=True)
            state["workflow_failed"] = True
            state["workflow_error"] = f"实质元素提取异常: {str(e)}"
            state.update({
                "concrete_elements": [],
                "concrete_concepts": [],
                "abstract_concepts": [],
                "substance_elements": [],
                "substance_analyzed_result": [],
                "substance_scored_result": {},
                "substance_filtered_ids": [],
                "substance_categorized_result": {},
                "substance_final_elements": []
            })

        return state

    def _form_extraction_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：形式元素提取（脚本理解）"""
        logger.info("=== 执行节点：形式元素提取 ===")

        # 检查关键错误
        if self._check_critical_error(state, "形式元素提取"):
            return state

        try:
            # 初始化Agent
            if not self.form_agent.is_initialized:
                self.form_agent.initialize()

            # 验证三点解构信息是否可用
            inspiration_points = state.get("inspiration_points", [])
            purpose_point = state.get("purpose_point", {})
            key_points = state.get("key_points", {})
            
            # 统计三点解构信息
            inspiration_count = len(inspiration_points) if isinstance(inspiration_points, list) else 0
            if not inspiration_count and isinstance(inspiration_points, dict):
                inspiration_count = len(inspiration_points.get("points", []))
            
            purpose_count = 0
            if isinstance(purpose_point, dict):
                purpose_count = len(purpose_point.get("purposes", []))
            elif isinstance(purpose_point, list):
                purpose_count = len(purpose_point)
            
            key_points_count = 0
            if isinstance(key_points, dict):
                key_points_list = key_points.get("key_points", [])
                key_points_count = len(key_points_list) if isinstance(key_points_list, list) else 0
            elif isinstance(key_points, list):
                key_points_count = len(key_points)
            
            logger.info(
                f"形式提取节点 - 三点解构信息检查: "
                f"灵感点={inspiration_count}, 目的点={purpose_count}, 关键点={key_points_count}"
            )

            # 执行Agent（依赖实质元素）
            result = self.form_agent.process(state)

            # 更新状态
            state.update(result)

            # 检查Agent返回结果中是否包含关键错误
            if self._check_agent_result_for_errors(result, "形式元素提取Agent"):
                error_msg = "形式元素提取失败：无法获取视频文件"
                state["workflow_failed"] = True
                state["workflow_error"] = error_msg
                return state

            final_elements = result.get("form_final_elements", [])
            logger.info(f"形式元素提取完成 - 最终元素数: {len(final_elements)}")

        except Exception as e:
            logger.error(f"形式元素提取失败: {e}", exc_info=True)
            state["workflow_failed"] = True
            state["workflow_error"] = f"形式元素提取异常: {str(e)}"
            state.update({
                "concrete_element_forms": [],
                "concrete_concept_forms": [],
                "overall_forms": [],
                "form_elements": [],
                "form_analyzed_result": [],
                "form_scored_result": {},
                "form_weighted_result": {},
                "form_filtered_ids": [],
                "form_categorized_result": {},
                "form_final_elements": []
            })

        return state

    def _merge_all_results_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：分离实质和形式结果（脚本理解）"""
        logger.info("=== 执行节点：分离实质和形式结果 ===")

        try:
            # 获取实质和形式的最终元素
            substance_final_elements = state.get("substance_final_elements", [])
            form_final_elements = state.get("form_final_elements", [])

            # 分别存储实质列表和形式列表
            state["实质列表"] = substance_final_elements
            state["形式列表"] = form_final_elements

            logger.info(f"分离完成 - 实质元素: {len(substance_final_elements)}, 形式元素: {len(form_final_elements)}")

        except Exception as e:
            logger.error(f"分离结果失败: {e}", exc_info=True)
            state["实质列表"] = []
            state["形式列表"] = []

        return state

    def _search_keywords_extraction_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：搜索关键词提取（分支1的最后一步）"""
        logger.info("=== 执行节点：搜索关键词提取 ===")

        # 检查关键错误
        if self._check_critical_error(state, "搜索关键词提取"):
            return state

        try:
            # 初始化Agent
            if not self.search_keyword_agent.is_initialized:
                self.search_keyword_agent.initialize()

            # 执行Agent（带重试机制）
            max_retries = 2
            result = None
            last_error = None
            
            def _is_result_failed(result: Dict[str, Any]) -> tuple[bool, str]:
                """检查结果是否失败
                
                Returns:
                    (是否失败, 错误信息)
                """
                if not result:
                    return True, "结果为空"
                
                # 检查关键错误（如无法获取视频文件）
                if self._check_agent_result_for_errors(result, "搜索关键词提取Agent"):
                    return True, "搜索关键词提取失败：无法获取视频文件"
                
                # 检查search_keywords中的错误字段
                search_keywords = result.get("search_keywords", {})
                if isinstance(search_keywords, dict):
                    # 检查是否有错误字段
                    if "错误" in search_keywords or "error" in search_keywords:
                        error_msg = search_keywords.get("错误") or search_keywords.get("error", "未知错误")
                        return True, f"搜索关键词提取失败：{error_msg}"
                    
                    # 检查搜索词数量是否为0（可能是解析失败的表现）
                    keyword_count = search_keywords.get("总数", 0)
                    if keyword_count == 0:
                        # 如果总数为0，可能是解析失败，需要重试
                        return True, "搜索关键词提取失败：未提取到搜索词（可能为解析失败）"
                
                return False, ""
            
            for attempt in range(max_retries):
                try:
                    # 执行Agent
                    result = self.search_keyword_agent.process(state)
                    
                    # 检查结果是否失败
                    is_failed, error_msg = _is_result_failed(result)
                    
                    if is_failed:
                        if attempt < max_retries - 1:
                            logger.warning(f"搜索关键词提取失败，准备重试 (尝试 {attempt + 1}/{max_retries}): {error_msg}")
                            last_error = error_msg
                            continue
                        else:
                            # 最后一次尝试也失败
                            state["workflow_failed"] = True
                            state["workflow_error"] = error_msg
                            return state
                    else:
                        # 成功，跳出重试循环
                        break
                        
                except Exception as e:
                    if attempt < max_retries - 1:
                        logger.warning(f"搜索关键词提取异常，准备重试 (尝试 {attempt + 1}/{max_retries}): {e}")
                        last_error = str(e)
                        continue
                    else:
                        # 最后一次尝试也失败，抛出异常让外层catch处理
                        raise

            # 更新状态
            if result:
                state.update(result)
            else:
                # 如果result为None，使用last_error
                error_msg = last_error or "搜索关键词提取失败：未知错误"
                state["workflow_failed"] = True
                state["workflow_error"] = error_msg
                return state

            # 再次检查（双重保险）
            if self._check_agent_result_for_errors(result, "搜索关键词提取Agent"):
                error_msg = "搜索关键词提取失败：无法获取视频文件"
                state["workflow_failed"] = True
                state["workflow_error"] = error_msg
                return state

            # 获取搜索关键词数量
            search_keywords = result.get("search_keywords", {})
            keyword_count = search_keywords.get("总数", 0) if isinstance(search_keywords, dict) else 0
            logger.info(f"搜索关键词提取完成 - 共 {keyword_count} 个搜索词")
            
            # 标记分支1完成
            state["branch1_completed"] = True

        except Exception as e:
            logger.error(f"搜索关键词提取失败: {e}", exc_info=True)
            state["workflow_failed"] = True
            state["workflow_error"] = f"搜索关键词提取异常: {str(e)}"
            state.update({
                "search_keywords": {
                    "搜索词列表": [],
                    "总数": 0,
                    "error": str(e)
                }
            })
            state["branch1_completed"] = True  # 即使失败也标记为完成，避免阻塞

        return state

    def _structure_analysis_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：结构化内容库分析（ScriptWorkflowV2分支）"""
        logger.info("=== 执行节点：结构化内容库分析（分支2） ===")

        # 检查关键错误
        if self._check_critical_error(state, "结构化内容库分析"):
            return state

        try:
            if not self.structure_agent.is_initialized:
                self.structure_agent.initialize()

            result = self.structure_agent.process(state)
            
            # 将结果存入 state 的 topic 字段
            structure_data = result.get("structure_data", {})
            state["topic"] = structure_data

            logger.info("结构化内容库分析完成")
            if isinstance(structure_data, dict):
                topic_info = structure_data.get("选题信息表", {})
                macro_topic = topic_info.get("宏观母题", "") if isinstance(topic_info, dict) else ""
                if macro_topic:
                    logger.info(f"宏观母题: {macro_topic}")
        except Exception as e:
            logger.error(f"结构化内容库分析失败: {e}", exc_info=True)
            state["topic"] = {
                "错误": str(e),
            }

        return state

    def _content_unit_split_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：L3 内容单元拆分（ScriptWorkflowV2分支）"""
        logger.info("=== 执行节点：L3 内容单元拆分（分支2） ===")

        # 检查关键错误
        if self._check_critical_error(state, "L3 内容单元拆分"):
            return state

        try:
            if not self.content_unit_split_agent.is_initialized:
                self.content_unit_split_agent.initialize()

            result = self.content_unit_split_agent.process(state)
            state.update(result)

            analysis = result.get("content_unit_analysis", {})
            logger.info(
                f"L3 单元拆分完成，单元数量: {len(analysis.get('单元列表', [])) if isinstance(analysis, dict) else 0}"
            )
        except Exception as e:
            logger.error(f"L3 内容单元拆分失败: {e}", exc_info=True)
            state["content_unit_analysis"] = {
                "error": str(e),
                "单元列表": [],
            }

        return state

    def _content_unit_understand_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：整体结构理解（ScriptWorkflowV2分支）"""
        logger.info("=== 执行节点：整体结构理解（分支2） ===")

        # 检查关键错误
        if self._check_critical_error(state, "整体结构理解"):
            return state

        try:
            if not self.content_unit_understand_agent.is_initialized:
                self.content_unit_understand_agent.initialize()

            result = self.content_unit_understand_agent.process(state)
            state.update(result)

            understanding = result.get("content_unit_understanding", {})
            logger.info(
                f"整体结构理解完成，段落数量: {len(understanding.get('段落解构', [])) if isinstance(understanding, dict) else 0}"
            )
        except Exception as e:
            logger.error(f"整体结构理解失败: {e}", exc_info=True)
            state["content_unit_understanding"] = {
                "error": str(e),
                "整体解构": {},
                "段落解构": [],
                "单元解构": {},
            }

        return state

    def _keyword_extraction_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：金句提取（ScriptWorkflowV2分支）"""
        logger.info("=== 执行节点：金句提取（分支2） ===")

        # 检查关键错误
        if self._check_critical_error(state, "金句提取"):
            return state

        try:
            if not self.script_keyword_agent.is_initialized:
                self.script_keyword_agent.initialize()

            result = self.script_keyword_agent.process(state)
            state.update(result)

            script_keywords = result.get("script_keywords", {})
            logger.info("金句提取完成")
            if isinstance(script_keywords, dict):
                hooks_count = len(script_keywords.get("hooks", []))
                golden_sentences_count = len(script_keywords.get("golden_sentences", []))
                logger.info(f"提取钩子数量: {hooks_count}, 金句数量: {golden_sentences_count}")
            
            # 标记分支2完成
            state["branch2_completed"] = True
        except Exception as e:
            logger.error(f"金句提取失败: {e}", exc_info=True)
            state["script_keywords"] = {
                "error": str(e),
            }
            state["branch2_completed"] = True  # 即使失败也标记为完成，避免阻塞

        return state

    def _merge_branches_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：合并两个分支的结果
        
        检查分支2是否完成，如果没完成则执行它
        """
        logger.info("=== 执行节点：合并分支结果 ===")

        try:
            # 检查分支2是否完成
            branch2_completed = state.get("branch2_completed", False)
            
            if not branch2_completed:
                logger.info("分支2还未完成，现在执行分支2（ScriptWorkflowV2流程）")
                try:
                    # 执行分支2的所有节点
                    if not self.structure_agent.is_initialized:
                        self.structure_agent.initialize()
                    structure_result = self.structure_agent.process(state)
                    state["topic"] = structure_result.get("structure_data", {})
                    
                    if not self.content_unit_split_agent.is_initialized:
                        self.content_unit_split_agent.initialize()
                    split_result = self.content_unit_split_agent.process(state)
                    state.update(split_result)
                    
                    if not self.content_unit_understand_agent.is_initialized:
                        self.content_unit_understand_agent.initialize()
                    understand_result = self.content_unit_understand_agent.process(state)
                    state.update(understand_result)
                    
                    if not self.script_keyword_agent.is_initialized:
                        self.script_keyword_agent.initialize()
                    keyword_result = self.script_keyword_agent.process(state)
                    state.update(keyword_result)
                    
                    state["branch2_completed"] = True
                    logger.info("分支2（ScriptWorkflowV2流程）执行完成")
                except Exception as e:
                    logger.error(f"分支2执行失败: {e}", exc_info=True)
                    state["branch2_completed"] = True  # 即使失败也标记为完成，避免阻塞
            else:
                logger.info("分支2已完成，直接合并结果")
            
            # 标记已合并
            state["branches_merged"] = True
            logger.info("分支结果合并完成，准备进入结果汇总")

        except Exception as e:
            logger.error(f"合并分支结果失败: {e}", exc_info=True)
            state["branches_merged"] = True  # 即使失败也标记为已合并，避免阻塞

        return state

    def _result_aggregation_node(self, state: Dict[str, Any]) -> Dict[str, Any]:
        """节点：结果汇总（包含两个分支的结果）"""
        logger.info("=== 执行节点：结果汇总 ===")

        try:
            # 初始化Function
            if not self.result_aggregation_func.is_initialized:
                self.result_aggregation_func.initialize()

            # 执行Function（获取分支1的结果）
            final_result = self.result_aggregation_func.execute(state)

            # 添加分支2的结果（ScriptWorkflowV2的结果）
            topic = state.get("topic", {})
            content_unit_analysis = state.get("content_unit_analysis", {})
            content_unit_understanding = state.get("content_unit_understanding", {})
            script_keywords = state.get("script_keywords", {})

            # 将分支2的结果添加到最终结果中
            final_result["脚本解构V2"] = {
                "结构化内容库": topic,
                "L3单元解构": content_unit_analysis,
                "整体结构理解": content_unit_understanding,
                "金句提取": script_keywords,
            }

            # 更新状态
            state["final_result"] = final_result

            logger.info("结果汇总完成（包含两个分支的结果）")

        except Exception as e:
            logger.error(f"结果汇总失败: {e}", exc_info=True)
            state["final_result"] = {
                "视频信息": {},
                "三点解构": {
                    "灵感点": [],
                    "目的点": {},
                    "关键点": {}
                },
                "选题理解": {},
                "脚本理解": {
                    "内容品类": "未知",
                    "段落列表": [],
                    "实质列表": [],
                    "形式列表": []
                },
                "脚本解构V2": {
                    "结构化内容库": {},
                    "L3单元解构": {},
                    "整体结构理解": {},
                    "金句提取": {}
                },
                "错误": f"汇总失败: {str(e)}"
            }

        return state

    def _update_db_record_after_workflow(
        self, 
        state: Dict[str, Any], 
        success: bool, 
        final_result: Dict[str, Any] = None,
        error_msg: str = None
    ):
        """工作流执行完毕后更新数据库记录
        
        Args:
            state: 工作流执行后的状态
            success: 是否成功
            final_result: 最终结果（成功时使用）
            error_msg: 错误信息（失败时使用）
        """
        try:
            task_id = state.get("db_task_id")
            if not task_id:
                logger.warning("未找到 db_task_id，跳过数据库记录更新")
                return
            
            db = next(get_db())
            try:
                record = db.query(DecodeVideo).filter_by(task_id=task_id).first()
                if not record:
                    logger.warning(f"未找到 task_id={task_id} 的数据库记录，跳过更新")
                    return
                
                if success:
                    # 更新为成功状态
                    import json
                    result_json = json.dumps(final_result, ensure_ascii=False) if final_result else None
                    record.update_status(DecodeStatus.SUCCESS)
                    record.update_result(result_json)
                    
                    # 提取分支2的结果（脚本解构V2）并存储到 decode_result_v2 字段
                    if final_result and isinstance(final_result, dict):
                        script_v2_result = final_result.get("脚本解构V2")
                        if script_v2_result:
                            result_v2_json = json.dumps(script_v2_result, ensure_ascii=False)
                            record.update_result_v2(result_v2_json)
                            logger.info(f"更新数据库记录的分支2结果: task_id={task_id}")
                    
                    # 提取搜索关键词并存储到 search_keywords 字段（字符串数组的JSON格式）
                    search_keywords_data = state.get("search_keywords", {})
                    if search_keywords_data and isinstance(search_keywords_data, dict):
                        keyword_list = search_keywords_data.get("搜索词列表", [])
                        if keyword_list:
                            # 提取每个搜索词的"搜索词"字段，组成字符串数组
                            keyword_strings = []
                            for item in keyword_list:
                                if isinstance(item, dict):
                                    keyword = item.get("搜索词", "")
                                    if keyword:
                                        keyword_strings.append(keyword)
                            
                            # 转换为JSON字符串数组格式
                            if keyword_strings:
                                keywords_json = json.dumps(keyword_strings, ensure_ascii=False)
                                record.update_search_keywords(keywords_json)
                                logger.info(f"更新数据库记录的搜索关键词: task_id={task_id}, 关键词数量={len(keyword_strings)}")
                    
                    logger.info(f"更新数据库记录为成功: task_id={task_id}")
                else:
                    # 更新为失败状态
                    record.update_status(DecodeStatus.FAILED, error_reason=error_msg)
                    logger.info(f"更新数据库记录为失败: task_id={task_id}, error={error_msg}")
                
                db.commit()
                
            except Exception as e:
                logger.error(f"更新数据库记录失败: {e}", exc_info=True)
                db.rollback()
            finally:
                db.close()
                
        except Exception as e:
            logger.error(f"更新数据库记录节点执行失败: {e}", exc_info=True)

    def invoke(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """执行工作流（公共接口）- 视频分析版本

        Returns:
            最终解码结果
        """
        logger.info("=== 开始执行解码工作流（视频分析） ===")
        logger.info(f"input_data: {input_data}")
        # 确保工作流已初始化
        if not self.is_initialized:
            self.initialize()

        # 验证输入参数
        video_url = input_data.get("video_url", "")
        if not video_url:
            error_msg = "未提供视频URL，无法执行工作流"
            logger.error(error_msg)
            return {
                "error": error_msg,
                "workflow_status": "failed",
                "input_data": input_data
            }

        # 初始化状态（包含视频信息，供视频上传和后续Agent使用）
        initial_state = {
            "video": video_url,
            "video_id": input_data.get("video_id", ""),
            "title": input_data.get("title", ""),
            "current_depth": 0,
            "max_depth": self.max_depth,
            "task_id": input_data.get("task_id", ""),
        }

        # 执行工作流
        result = None
        try:
            result = self.compiled_graph.invoke(initial_state)
        except Exception as e:
            error_msg = f"工作流执行异常: {str(e)}"
            logger.error(error_msg, exc_info=True)
            # 更新数据库记录为失败状态（使用 initial_state 作为 fallback）
            if result is None:
                result = initial_state
            self._update_db_record_after_workflow(result, success=False, error_msg=error_msg)
            return {
                "status": 3,
                "error": error_msg,
                "workflow_status": "failed",
                "exception_type": type(e).__name__
            }

        # 检查是否因为错误而终止
        if result.get("workflow_failed"):
            error_msg = result.get("workflow_error", "工作流执行失败")
            logger.error(f"工作流因错误而终止: {error_msg}")
            # 更新数据库记录为失败状态
            self._update_db_record_after_workflow(result, success=False, error_msg=error_msg)
            return {
                "error": error_msg,
                "video_upload_error": result.get("video_upload_error"),
                "workflow_status": "failed",
                "failed_at": result.get("failed_at", "unknown"),
                "status": 3
            }

        # 检查是否有最终结果
        final_result = result.get("final_result")
        if not final_result:
            # 如果没有最终结果，检查是否有错误信息
            if result.get("workflow_error"):
                error_msg = result.get("workflow_error", "工作流执行失败，未生成结果")
                logger.error(f"工作流执行失败: {error_msg}")
                # 更新数据库记录为失败状态
                self._update_db_record_after_workflow(result, success=False, error_msg=error_msg)
                return {
                    "status": 3,
                    "error": error_msg,
                    "workflow_status": "failed"
                }
            else:
                logger.warning("工作流执行完成，但未生成最终结果")
                # 更新数据库记录为失败状态
                self._update_db_record_after_workflow(result, success=False, error_msg="工作流执行完成，但未生成最终结果")
                return {
                    "status": 3,
                    "error": "工作流执行完成，但未生成最终结果",
                    "workflow_status": "incomplete",
                    "state": result
                }

        # 工作流执行成功，更新数据库记录
        self._update_db_record_after_workflow(result, success=True, final_result=final_result)

        logger.info("=== 解码工作流执行完成（视频分析） ===")

        # 添加status字段（2表示成功）
        if isinstance(final_result, dict):
            final_result["status"] = 2

        return final_result