video_decode/src/components/agents/script_keyword_agent.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
金句提取Agent（视频分析版本）

功能:
1. 从 state 中读取视频文件对象
2. 使用 Gemini 视频分析能力，直接基于视频内容提取"钩子"和"金句"
"""

from typing import Any, Dict

from src.components.agents.base import BaseLLMAgent
from src.utils.logger import get_logger
from src.utils.llm_invoker import LLMInvoker, get_video_file_from_state

logger = get_logger(__name__)


class ScriptKeywordAgent(BaseLLMAgent):
    """金句提取Agent - 基于视频分析提取钩子和金句"""

    def __init__(
        self,
        name: str = "script_keyword_agent",
        description: str = "金句提取Agent - 直接基于视频内容提取钩子和金句",
        model_provider: str = "google_genai",
        temperature: float = 0.4,
        max_tokens: int = 20480,
    ):
        # 这里的 system_prompt 仅用于初始化模型（提供 model_name），
        # 实际的视频分析逻辑全部通过 safe_invoke_video_analysis 的 prompt 控制
        system_prompt = "你是一名专注于中国60岁+老年群体短视频生态的内容分析专家。"
        super().__init__(
            name=name,
            description=description,
            model_provider=model_provider,
            system_prompt=system_prompt,
            temperature=temperature,
            max_tokens=max_tokens,
        )

    # ==================== 对外主入口 ====================

    def process(self, state: Dict[str, Any], config=None) -> Dict[str, Any]:
        """处理状态

        步骤：
        1. 从 state 中读取视频文件对象
        2. 直接调用 safe_invoke_video_analysis，基于视频内容提取钩子和金句
        3. 将金句提取结果写回 state
        """
        if not self.is_initialized:
            self.initialize()

        logger.info("金句提取Agent开始执行")

        # Step 0: 从 state 中获取视频文件对象
        video_file = get_video_file_from_state(state)
        if not video_file:
            logger.warning("无法获取视频文件对象，跳过金句提取")
            return {
                "script_keywords": {
                    "error": "无法获取视频文件对象",
                },
            }

        try:
            # 直接基于视频内容进行金句提取
            logger.info("基于视频内容进行金句提取...")
            keyword_prompt = self._build_keyword_prompt()

            keyword_result = LLMInvoker.safe_invoke_video_analysis(
                "金句提取(钩子和金句)",
                video_file,
                keyword_prompt,
                agent=self,
                fallback={
                    "script_type": "",
                    "hooks": [],
                    "golden_sentences": [],
                },
            )

            logger.info("金句提取Agent执行完成")

            return {
                "script_keywords": keyword_result,
            }

        except Exception as e:
            logger.error(f"金句提取Agent执行失败: {e}", exc_info=True)
            return {
                "script_keywords": {
                    "error": str(e),
                },
            }

    # ==================== Prompt 构建 ====================

    def _build_keyword_prompt(self) -> str:
        """构建基于视频内容的金句提取 Prompt"""
        return """
# Role
你是一名专注于中国 60岁+ 老年群体短视频生态的内容分析专家。你的核心能力是识别视频脚本中的"传播基因"和"心理抓手"。

# Task
请根据视频，利用【认知缺口】和【高能评判】两大底层逻辑，提取脚本中的"钩子（Hooks）"和"金句（Golden Sentences）"。

# Extraction Rules (必须严格遵守)
## 1. 场景预判 (Scenario Detection)
首先判断脚本属于以下哪种类型，并调整提取侧重：
- **类型 A：叙事/人物/科普类**（如名人故事、大国重器）
  -> 侧重提取：核心冲突、悲情叙事、民族情绪、正名/喊冤。
- **类型 B：盘点/猎奇/生活类**（如奇葩村庄、顺口溜、养生）
  -> 侧重提取：反常识特征、极低成本获益、顺口溜韵脚。

## 2. 定义与提取「钩子」(The Hooks)
*目标：提取制造悬念、打破常识或直击痛点的短句（用于留存）。*

请扫描脚本开头及段落转折处，提取符合以下逻辑的句子：
1.  **逻辑悬空 (The Open Loop):** 展示了惊人的结果或高唤醒度的状态，但隐去了原因或过程。（例如："看完才明白他的良苦用心"、"99%的人都不知道"）
2.  **反常识特征 (The Strange Feature):** 描述违背常理、伦理或物理规律的现象。**注意：将具体的生僻地名/人名抽象为特征描述。**（例如："男人一辈子只能生活在树上"、"娶媳妇只需要一头猪"）
3.  **痛点与利益 (Pain & Benefit):** 直接针对健康、养老金、子女关系的强引导。（例如："千万别再这样吃"、"过了60岁要注意"）

## 3. 定义与提取「金句」(The Golden Sentences)
*目标：提取用于社交货币、情绪宣泄或身份认同的短句（用于分享）。*
请扫描全文（特别是结尾和高潮处），提取符合以下逻辑的句子：
1.  **绝对化定性 (Absolute Judgment):** 脱离具体时空叙事，对人/事/物进行盖棺定论的总结。包含"一辈子"、"凡是"、"绝对"、"就是"等词。（例如："手里有钱才是硬道理"、"他一辈子就做了一件事"）
2.  **核心冲突与宣泄 (Conflict & Venting):** 描述 [付出 vs 误解]、[高尚 vs 庸俗]、[我们 vs 敌人] 的强烈对比。包含"嘲讽"、"心寒"、"咬牙切齿"、"不再看脸色"等词。（例如："把一生献给国家的科学家被人误解多年"）
3.  **乌托邦与治愈 (The Utopia):** 描绘老年人向往的理想状态，或否定现实压力。（例如："没有复杂的婆媳关系"、"把老年变玩年"）
4.  **韵律与警世 (Rhymes):** 押韵的顺口溜或具有视觉冲击力的警世格言。（例如："饭再好没有牙，钱再多床上趴"）

## 4. 负面清单 (Negative List - 不要提取)
- **纯事实叙事：** 具体的年份、复杂的数字、毫无特征的流水账（如"1996年他去了北京"）。
- **主播口水话：** "点好关注"、"屏幕前的朋友"、"您收到了吗"。
- **长难句：** 如果句子超过 30 字且无法拆解出独立观点，请忽略。
- **无意义连接词：** "话说"、"那么"、"之所以"。

# Output Format (JSON)
请直接输出 JSON 格式数据，不要包含任何解释性文字。
{
  "script_type": "判断是 类型A 或 类型B",
  "hooks": [
    "提取的钩子1",
    "提取的钩子2 (必须保留原文语气，如'竟然'、'千万')"
  ],
  "golden_sentences": [
    "提取的金句1",
    "提取的金句2",
    "提取的金句3 (必须是完整的观点或情绪表达)"
  ]
}
""".strip()

    # ==================== BaseLLMAgent 抽象方法占位实现 ====================

    def _build_messages(self, state: Dict[str, Any]) -> list:
        """满足 BaseLLMAgent 抽象要求（本 Agent 不通过该路径调用）"""
        return []

    def _update_state(self, state: Dict[str, Any], response: Any) -> Dict[str, Any]:
        """满足 BaseLLMAgent 抽象要求（本 Agent 不通过该路径调用）"""
        return state