video_decode/src/components/agents/search_keyword_agent.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
搜索关键词Agent

功能: 基于视频分析结果（灵感点、目的点、关键点、选题理解）生成抖音/TikTok组合搜索词列表
核心任务: 分析视频核心构成，输出用于检索竞品内容的搜索词组合
特征: 使用"元素+语境"组合法，生成精准搜索词
"""

from typing import Any, Dict, List
import json
import re

from src.components.agents.base import BaseLLMAgent
from src.utils.logger import get_logger

logger = get_logger(__name__)


class SearchKeywordAgent(BaseLLMAgent):
    """搜索关键词Agent - 生成抖音/TikTok组合搜索词列表"""

    def __init__(
        self,
        name: str = "search_keyword_agent",
        description: str = "搜索关键词Agent - 生成组合搜索词列表",
        model_provider: str = "google_genai",
        temperature: float = 0.7,
        max_tokens: int = 8192
    ):
        """
        初始化搜索关键词Agent

        Args:
            name: Agent名称
            description: Agent描述
            model_provider: 模型提供商 ("openai" 或 "google_genai")
            temperature: 生成温度，控制创造性
            max_tokens: 最大token数
        """
        system_prompt = self._build_system_prompt()
        super().__init__(
            name=name,
            description=description,
            model_provider=model_provider,
            system_prompt=system_prompt,
            temperature=temperature,
            max_tokens=max_tokens
        )

    def _build_system_prompt(self) -> str:
        """构建系统提示词 - 定义角色、能力、输出规范"""
        return """# Role
你是一位精通全网内容分发逻辑与搜索引擎优化（SEO）的数据分析师。

# Task
我将提供一段关于视频内容的详细结构化数据（JSON格式），包含【灵感点】、【目的点】和【关键点】。
你的任务是分析该视频的核心构成，并输出一份**抖音/TikTok组合搜索词列表**，用于检索与该视频**题材相同、切入点相似、受众重合**的竞品内容。

# Core Strategy: "元素+语境" 组合法
为了保证搜索结果的精准度，请严格遵循 **`[核心实体/具象元素]` + `[主题语境/内容属性]`** 的组合公式进行输出。

### 1. 定义"核心实体" (Left Side)
请从 JSON 的 `关键点 (Key Points)` 和 `视频信息` 中提取最具体的名词。
* **包括**：具体的人名/地名/产品名、特定的时间节点、核心物体、显性的视觉符号、特定的动作名称。
* **排除**：泛泛的大类词（如"生活"、"快乐"、"健康"）。

### 2. 定义"主题语境" (Right Side)
请从 JSON 的 `灵感点 (Inspiration)` 和 `目的点 (Purpose)` 中提取修饰性的词汇。
* **包括**：内容的体裁（教程/测评/Vlog）、情感色彩（治愈/吐槽/焦虑）、功能属性（避坑/省钱/变美）、特定的叙事逻辑（反转/揭秘/老话）。

# Execution Rules (执行规则)
请扫描 JSON 数据，针对不同维度生成搜索词：

1.  **提取"强钩子"组合**：
    * 找到视频开头最吸引人的视觉或概念元素（灵感点）。
    * *通用公式*：`[核心视觉/事件]` + `[钩子属性(如:真相/后续/反常)]`
2.  **提取"核心内容"组合**：
    * 找到视频主要传达的知识、剧情或展示物（关键点）。
    * *通用公式*：`[具体事物/动作]` + `[垂直领域词(如:做法/技巧/评价)]`
3.  **提取"受众痛点"组合**：
    * 找到视频想要解决的问题或满足的需求（目的点）。
    * *通用公式*：`[受众标签/场景]` + `[痛点/解决方案]`

# Output Format
请输出 10-12 个组合搜索词，按以下格式排列，并简要解释组合逻辑：
* **`关键词` + `语境词`** (例如：*空气炸锅 翻车* / *重阳节 禁忌* / *面试 自我介绍*)
    * *逻辑：提取了[X]元素，结合[Y]主题。*

输出格式为JSON：
```json
{
  "搜索词列表": [
    {
      "搜索词": "关键词 语境词",
      "组合逻辑": "提取了[X]元素，结合[Y]主题。"
    }
  ]
}
```"""

    def _build_messages(self, state: Dict[str, Any]) -> List[Dict[str, Any]]:
        """
        构建LLM消息

        Args:
            state: 包含视频和其他agent结果的状态字典,需要包含:
                - inspiration_points: 灵感点提取结果
                - purpose_point: 目的点提取结果
                - key_points: 关键点提取结果
                - topic_selection_understanding: 选题理解结果

        Returns:
            消息列表
        """
        # 提取输入数据
        inspiration_points = state.get("inspiration_points", [])
        purpose_point = state.get("purpose_point", {})
        key_points = state.get("key_points", {})
        topic_selection_understanding = state.get("topic_selection_understanding", {})

        # 构建提示词
        prompt = self._build_user_prompt(
            inspiration_points,
            purpose_point,
            key_points,
            topic_selection_understanding
        )

        # 构建消息
        return [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": prompt}
        ]

    def _update_state(self, state: Dict[str, Any], response: Any) -> Dict[str, Any]:
        """
        更新状态

        Args:
            state: 原始状态
            response: LLM响应

        Returns:
            更新后的状态
        """
        # 解析响应
        result = self._parse_response(response.content)

        return {
            "search_keywords": result
        }

    def _build_user_prompt(
        self,
        inspiration_points: Any,
        purpose_point: Dict[str, Any],
        key_points: Dict[str, Any],
        topic_selection_understanding: Dict[str, Any]
    ) -> str:
        """构建用户提示 - 提供数据和执行指令"""
        # 格式化各维度数据
        inspiration_section = self._format_inspiration_points(inspiration_points)
        purpose_section = self._format_purpose_point(purpose_point)
        key_points_section = self._format_key_points(key_points)
        topic_section = self._format_topic_selection_understanding(topic_selection_understanding)

        return f"""# Input Data

## 选题理解结果
{topic_section}

## 灵感点分析
{inspiration_section}

## 目的点分析
{purpose_section}

## 关键点分析
{key_points_section}

---

# 执行指令

基于以上数据，输出 10-12 个组合搜索词。

## 质量要求
- **搜索词数量**: 10-12 个
- **组合公式**: `[核心实体/具象元素]` + `[主题语境/内容属性]`
- **覆盖维度**: 需要覆盖"强钩子"、"核心内容"、"受众痛点"三个维度
- **精准度**: 搜索词应该能够检索到与该视频题材相同、切入点相似、受众重合的竞品内容
- **组合逻辑**: 每个搜索词需要简要说明提取了哪些元素，结合了哪些主题

**重要提醒**：
- 核心实体必须具体，避免泛泛的大类词
- 主题语境要能体现内容的体裁、情感、功能或叙事逻辑
- 搜索词不能出现雷同和近似的词语
- 搜索词应该符合抖音/TikTok用户的搜索习惯"""

    def _format_topic_selection_understanding(self, topic_selection_understanding: Dict[str, Any]) -> str:
        """格式化选题理解信息"""
        if not topic_selection_understanding:
            return "**未找到选题理解结果**"

        主题 = topic_selection_understanding.get("主题", "")
        描述 = topic_selection_understanding.get("描述", "")

        formatted = ""
        if 主题:
            formatted += f"**主题**: {主题}\n"
        if 描述:
            formatted += f"**描述**: {描述}\n"

        return formatted if formatted else "**未找到选题理解结果**"

    def _format_inspiration_points(self, inspiration_points: Any) -> str:
        """格式化灵感点信息"""
        # 新结构：inspiration_points 直接是列表
        if isinstance(inspiration_points, list):
            if not inspiration_points:
                return "**未提取到灵感点**"

            formatted = f"**灵感点总数**: {len(inspiration_points)}\n\n"

            for idx, point in enumerate(inspiration_points, 1):
                formatted += f"### 灵感点 {idx}: {point.get('灵感点', 'N/A')}\n"
                formatted += f"**分类**: {point.get('分类', '')}\n"
                formatted += f"**描述**: {point.get('描述', '')}\n"
                if point.get("推理"):
                    formatted += f"**推理**: {point.get('推理', '')}\n"
                formatted += "\n"

            return formatted

        # 旧结构：{"points": [...]} 或字典格式
        if isinstance(inspiration_points, dict):
            # 旧结构：points列表
            if "points" in inspiration_points:
                points = inspiration_points.get("points", [])
                if not points:
                    return "**未提取到灵感点**"

                formatted = f"**灵感点总数**: {len(points)}\n\n"

                for idx, point in enumerate(points, 1):
                    formatted += f"### 灵感点 {idx}: {point.get('灵感点表象', 'N/A')}\n"
                    formatted += f"**本质**: {point.get('灵感点本质', '')}\n"
                    formatted += f"{point.get('本质详细说明', '')}\n\n"

                return formatted

        return "**未提取到灵感点**"

    def _format_purpose_point(self, purpose_point: Dict[str, Any]) -> str:
        """格式化目的点信息"""
        if not purpose_point:
            return "**未提取到目的点**"

        # 新结构：{"perspective": "创作者视角", "purposes": [...], "total_count": 2}
        purposes = purpose_point.get("purposes", [])
        if not purposes:
            return "**未提取到目的点**"

        formatted = f"**目的点总数**: {len(purposes)}\n\n"

        for idx, purpose in enumerate(purposes, 1):
            维度 = purpose.get("维度", {})
            if isinstance(维度, dict):
                维度_str = f"{维度.get('一级分类', '')}/{维度.get('二级分类', '')}"
            else:
                维度_str = str(维度)

            目的点 = purpose.get("目的点", "N/A")
            描述 = purpose.get("描述", "")

            formatted += f"### 目的点 {idx}: {目的点}\n"
            formatted += f"**维度**: {维度_str}\n"
            if 描述:
                formatted += f"**描述**: {描述}\n"
            formatted += "\n"

        return formatted

    def _format_key_points(self, key_points: Dict[str, Any]) -> str:
        """格式化关键点信息 - 支持层级结构"""
        if not key_points or "key_points" not in key_points:
            return "**未提取到关键点**"

        points = key_points.get("key_points", [])
        if not points:
            return "**未提取到关键点**"

        total_count = key_points.get("total_count", len(points))
        root_count = key_points.get("root_count", len(points))
        hierarchy_enabled = key_points.get("hierarchy_enabled", False)

        formatted = f"**关键点总数**: {total_count}\n"
        if hierarchy_enabled:
            formatted += f"**一级关键点数**: {root_count}\n"
            formatted += f"**层级结构**: 已启用\n\n"
        else:
            formatted += "\n"

        # 递归格式化关键点树
        def format_point_tree(point: Dict[str, Any], level: int = 1, index: int = 1) -> str:
            indent = "  " * (level - 1)
            result = f"{indent}### 关键点 {index} (层级 {level}): {point.get('关键点', 'N/A')}\n"
            result += f"{indent}**维度**: [{point.get('维度大类', '')}] {point.get('维度细分', '')}\n"
            result += f"{indent}**描述**: {point.get('描述', '')}\n"

            # 递归处理子关键点
            children = point.get("children", [])
            if children:
                result += f"{indent}**子关键点**: {len(children)}个\n\n"
                for child_idx, child in enumerate(children, 1):
                    result += format_point_tree(child, level + 1, child_idx)
            else:
                result += "\n"

            return result

        # 格式化所有一级关键点
        for idx, point in enumerate(points, 1):
            formatted += format_point_tree(point, level=1, index=idx)

        return formatted

    def _parse_response(self, response: str) -> Dict[str, Any]:
        """解析LLM响应"""
        try:
            # 尝试提取JSON代码块
            json_match = re.search(r'```(?:json)?\s*(.*?)\s*```', response, re.DOTALL)
            if json_match:
                json_str = json_match.group(1)
            else:
                json_str = response

            # 解析JSON
            result = json.loads(json_str)

            # 确保包含必需字段
            if "搜索词列表" not in result:
                result["搜索词列表"] = []

            # 验证每个搜索词的结构
            search_keywords = result.get("搜索词列表", [])
            validated_keywords = []
            for keyword in search_keywords:
                if isinstance(keyword, dict):
                    # 确保包含必需字段
                    validated_keyword = {
                        "搜索词": keyword.get("搜索词", ""),
                        "组合逻辑": keyword.get("组合逻辑", "")
                    }
                    if validated_keyword["搜索词"]:  # 只保留有效的搜索词
                        validated_keywords.append(validated_keyword)

            result["搜索词列表"] = validated_keywords
            result["总数"] = len(validated_keywords)

            return result

        except json.JSONDecodeError as e:
            logger.error(f"JSON解析失败: {e}")
            logger.error(f"响应内容: {response[:500]}")
            # JSON解析失败，返回默认结构
            return {
                "搜索词列表": [],
                "总数": 0,
                "错误": "解析失败"
            }
        except Exception as e:
            logger.error(f"解析响应时发生错误: {e}", exc_info=True)
            return {
                "搜索词列表": [],
                "总数": 0,
                "错误": str(e)
            }