video_decode/src/components/agents/evaluate_agent.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
评估Agent

功能: 从待评估的视频列表中筛选出和原视频最匹配的内容
核心任务: 根据原视频的解构内容，对每个候选视频进行相关性评分和筛选
特征: 使用LLM进行相关性评估，输出评分和是否入选
"""

from typing import Any, Dict, List
import json

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

logger = get_logger(__name__)


class EvaluateAgent(BaseLLMAgent):
    """评估Agent - 筛选和原视频最匹配的视频"""

    def __init__(
        self,
        name: str = "evaluate_agent",
        description: str = "评估Agent - 筛选和原视频最匹配的视频",
        model_provider: str = "google_genai",
        temperature: float = 0.3,
        max_tokens: int = 20480
    ):
        """
        初始化评估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 """你是内容分析专家，擅长评估视频内容的相关性和匹配度。

# 任务
根据原视频的解构内容（包括标题、灵感点、目的点、关键点、选题理解等），对候选视频列表进行相关性评估和筛选。

# 评估维度
1. **内容主题匹配度**：候选视频的主题是否与原视频的主题一致或相关
2. **切入点相似度**：候选视频的切入角度是否与原视频相似
3. **受众重合度**：候选视频的目标受众是否与原视频重合
4. **表现形式相似度**：候选视频的表现形式（风格、结构等）是否与原视频相似

# 评分标准
- 相关性得分范围：0-100分
- 90-100分：高度相关，主题、切入点、受众、表现形式都高度匹配
- 70-89分：相关，在多个维度上匹配
- 50-69分：中等相关，在某些维度上匹配
- 30-49分：低相关，匹配度较低
- 0-29分：不相关，基本不匹配

# 输出要求
1. 对每个候选视频进行评分（0-100分）
2. 根据评分排序，默认选前50%进入候选
3. 输出结果保持与输入列表相同的顺序和字段，新增两个字段：
   - relevance_score: 相关性得分（0-100）
   - is_selected: 是否入选（true/false）
"""

    def process(self, state: Dict[str, Any], config=None) -> Dict[str, Any]:
        """处理状态 - 评估视频相关性并筛选
        
        Args:
            state: 状态字典，包含：
                - original_video_title: 原视频标题
                - original_video_content: 原视频解构内容（JSON格式）
                - search_result: 待评估的视频列表
                
        Returns:
            更新后的状态，包含：
                - evaluate_result: 评估结果列表（每个视频包含原始字段 + relevance_score + is_selected）
        """
        if not self.is_initialized:
            self.initialize()

        logger.info("开始评估视频相关性")

        try:
            # 从state获取数据
            original_video_title = state.get("original_video_title", "")
            original_video_content = state.get("original_video_content", {})
            search_result = state.get("search_result", [])

            if not search_result:
                logger.warning("待评估的视频列表为空")
                return {
                    "evaluate_result": []
                }

            if not original_video_title and not original_video_content:
                logger.warning("原视频信息为空，无法进行评估")
                return {
                    "evaluate_result": []
                }

            # 构建评估提示词
            prompt = self._build_evaluate_prompt(
                original_video_title,
                original_video_content,
                search_result
            )

            messages = [
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": prompt}
            ]

            # 调用LLM进行评估
            result = LLMInvoker.safe_invoke(
                self,
                "视频相关性评估",
                messages,
                fallback={"评估结果": []}
            )

            # 提取评估结果
            evaluate_result = result.get("评估结果", [])
            
            # 如果LLM返回的结果数量与输入不一致，进行修正
            if len(evaluate_result) != len(search_result):
                logger.warning(
                    f"LLM返回结果数量({len(evaluate_result)})与输入数量({len(search_result)})不一致，"
                    "将进行修正"
                )
                evaluate_result = self._fix_evaluate_result(search_result, evaluate_result)

            # 确保每个结果都有relevance_score和is_selected字段
            evaluate_result = self._ensure_evaluate_fields(search_result, evaluate_result)

            # 根据评分排序并标记前50%为入选
            evaluate_result = self._mark_selected_videos(evaluate_result)

            logger.info(f"评估完成，共评估{len(evaluate_result)}个视频")

            return {
                "evaluate_result": evaluate_result
            }

        except Exception as e:
            logger.error(f"视频评估失败: {e}", exc_info=True)
            # 返回原始列表，但添加默认的评分和选择状态
            search_result = state.get("search_result", [])
            return {
                "evaluate_result": [
                    {**video, "relevance_score": 0, "is_selected": False}
                    for video in search_result
                ]
            }

    def _build_evaluate_prompt(
        self,
        original_video_title: str,
        original_video_content: Dict[str, Any],
        search_result: List[Dict[str, Any]]
    ) -> str:
        """构建评估提示词"""
        
        # 格式化原视频内容
        content_str = json.dumps(original_video_content, ensure_ascii=False, indent=2)
        
        # 格式化候选视频列表
        candidates_text = ""
        for i, video in enumerate(search_result, 1):
            video_str = json.dumps(video, ensure_ascii=False, indent=2)
            candidates_text += f"\n## 候选视频 {i}\n{video_str}\n"

        prompt = f"""# 任务：评估视频相关性

## 原视频信息

### 标题
{original_video_title}

### 解构内容
{content_str}

## 候选视频列表
{candidates_text}

## 评估要求

1. **对每个候选视频进行相关性评分**（0-100分）
   - 考虑内容主题匹配度、切入点相似度、受众重合度、表现形式相似度
   - 评分要客观、准确

2. **输出格式要求**
   - 保持与输入列表相同的顺序
   - 保留原始字段不变
   - 新增两个字段：
     - `relevance_score`: 相关性得分（整数，0-100）
     - `is_selected`: 是否入选（布尔值，暂时设为false，后续会根据评分排序后标记前50%）

## 输出格式（JSON）

```json
{{
  "评估结果": [
    {{
      // 保留原始字段...
      "relevance_score": 85,
      "is_selected": false
    }}
  ]
}}
```

**重要**：
- 输出结果的数量必须与输入列表的数量完全一致
- 每个结果必须包含所有原始字段
- 每个结果必须包含relevance_score和is_selected字段
"""

        return prompt

    def _fix_evaluate_result(
        self,
        original_list: List[Dict[str, Any]],
        llm_result: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """修正评估结果，确保数量一致"""
        fixed_result = []
        
        # 创建LLM结果的索引（通过某些唯一字段匹配）
        llm_result_map = {}
        for item in llm_result:
            # 尝试通过video_id或其他唯一字段匹配
            video_id = item.get("video_id") or item.get("id") or item.get("videoId")
            if video_id:
                llm_result_map[str(video_id)] = item
        
        # 遍历原始列表，匹配LLM结果
        for i, original in enumerate(original_list):
            video_id = original.get("video_id") or original.get("id") or original.get("videoId")
            if video_id and str(video_id) in llm_result_map:
                # 找到匹配的结果，合并字段
                matched = llm_result_map[str(video_id)]
                fixed_item = {**original, **matched}
                fixed_result.append(fixed_item)
            elif i < len(llm_result):
                # 按索引匹配
                matched = llm_result[i]
                fixed_item = {**original, **matched}
                fixed_result.append(fixed_item)
            else:
                # 没有匹配的结果，使用原始数据并添加默认评分
                fixed_item = {**original, "relevance_score": 0, "is_selected": False}
                fixed_result.append(fixed_item)
        
        return fixed_result

    def _ensure_evaluate_fields(
        self,
        original_list: List[Dict[str, Any]],
        evaluate_result: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """确保每个评估结果都有必要的字段"""
        ensured_result = []
        
        for i, original in enumerate(original_list):
            if i < len(evaluate_result):
                item = evaluate_result[i]
                # 合并原始字段和评估字段
                merged_item = {**original}
                
                # 确保有relevance_score
                if "relevance_score" in item:
                    merged_item["relevance_score"] = item["relevance_score"]
                else:
                    merged_item["relevance_score"] = 0
                
                # 确保有is_selected（暂时设为false，后续会重新标记）
                merged_item["is_selected"] = False
                
                ensured_result.append(merged_item)
            else:
                # 如果LLM结果不足，使用原始数据并添加默认值
                ensured_result.append({
                    **original,
                    "relevance_score": 0,
                    "is_selected": False
                })
        
        return ensured_result

    def _mark_selected_videos(self, evaluate_result: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """根据评分排序并标记前50%为入选，然后恢复原始顺序"""
        if not evaluate_result:
            return evaluate_result
        
        # 为每个视频添加临时索引，以便后续恢复原始顺序
        indexed_result = [
            {**item, "_original_index": i}
            for i, item in enumerate(evaluate_result)
        ]
        
        # 按评分降序排序
        sorted_result = sorted(
            indexed_result,
            key=lambda x: x.get("relevance_score", 0),
            reverse=True
        )
        
        # 计算前50%的数量（向上取整）
        selected_count = max(1, (len(sorted_result) + 1) // 2)
        
        # 标记前50%为入选
        for i, item in enumerate(sorted_result):
            item["is_selected"] = (i < selected_count)
        
        # 恢复原始顺序
        sorted_result.sort(key=lambda x: x.get("_original_index", len(evaluate_result)))
        
        # 移除临时索引
        for item in sorted_result:
            item.pop("_original_index", None)
        
        return sorted_result

    def _build_messages(self, state: Dict[str, Any]) -> List[Dict[str, Any]]:
        """构建消息 - BaseLLMAgent要求实现（本Agent不使用此方法）"""
        return []

    def _update_state(self, state: Dict[str, Any], response: Any) -> Dict[str, Any]:
        """更新状态 - BaseLLMAgent要求实现（本Agent不使用此方法）"""
        return state