#!/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) }