content-finder-agent/examples/content_finder/tools/gemini_api.py

"""
Gemini API 工具

提供内容分析和比较功能。
"""

import json
import os
from typing import List, Dict, Any, Optional
import httpx

from agent.tools import tool, ToolResult


# API 基础配置
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "")
GEMINI_API_URL = "https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent"
DEFAULT_TIMEOUT = 60.0


@tool(
    display={
        "zh": {
            "name": "Gemini内容分析",
            "params": {
                "content": "要分析的内容",
                "analysis_type": "分析类型",
                "context": "上下文信息"
            }
        }
    }
)
async def gemini_analyze_content(
    content: str,
    analysis_type: str = "general",
    context: Optional[str] = None,
) -> ToolResult:
    """
    使用Gemini分析内容

    Args:
        content: 要分析的内容（视频标题、描述等）
        analysis_type: 分析类型，可选：
            - general: 通用分析
            - emotion: 情感分析
            - audience: 受众分析
            - quality: 质量评估
        context: 额外的上下文信息

    Returns:
        ToolResult 包含分析结果：
        {
            "analysis": "分析结果文本",
            "structured_data": {
                "emotion": "感动",
                "target_audience": "50+老年人",
                "key_themes": ["抗日", "英雄", "历史"],
                "quality_score": 8.5,
                "recommendations": ["适合老年人观看", "具有教育意义"]
            }
        }
    """
    try:
        if not GEMINI_API_KEY:
            return ToolResult(
                title="配置错误",
                output="",
                error="未配置 GEMINI_API_KEY 环境变量"
            )

        # 构建提示词
        prompts = {
            "general": f"请分析以下内容，提供详细的分析报告：\n\n{content}",
            "emotion": f"请分析以下内容的情感倾向，识别主要情感和情感强度：\n\n{content}",
            "audience": f"请分析以下内容适合的目标受众，包括年龄、兴趣等特征：\n\n{content}",
            "quality": f"请评估以下内容的质量，包括吸引力、完整性、可信度等：\n\n{content}",
        }

        prompt = prompts.get(analysis_type, prompts["general"])

        if context:
            prompt += f"\n\n上下文信息：{context}"

        # 调用Gemini API
        payload = {
            "contents": [{
                "parts": [{
                    "text": prompt
                }]
            }]
        }

        headers = {
            "Content-Type": "application/json",
        }

        url = f"{GEMINI_API_URL}?key={GEMINI_API_KEY}"

        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
            response = await client.post(
                url,
                json=payload,
                headers=headers,
            )
            response.raise_for_status()
            data = response.json()

        # 提取分析结果
        candidates = data.get("candidates", [])
        if not candidates:
            return ToolResult(
                title="分析失败",
                output="",
                error="Gemini API 未返回结果"
            )

        analysis_text = candidates[0].get("content", {}).get("parts", [{}])[0].get("text", "")

        result = {
            "analysis": analysis_text,
            "analysis_type": analysis_type,
            "content_preview": content[:100] + "..." if len(content) > 100 else content
        }

        return ToolResult(
            title=f"Gemini分析: {analysis_type}",
            output=json.dumps(result, ensure_ascii=False, indent=2),
            long_term_memory=f"Analyzed content with Gemini ({analysis_type})"
        )
    except httpx.HTTPStatusError as e:
        return ToolResult(
            title="API调用失败",
            output="",
            error=f"HTTP error {e.response.status_code}: {e.response.text}"
        )
    except Exception as e:
        return ToolResult(
            title="分析失败",
            output="",
            error=str(e)
        )


@tool(
    display={
        "zh": {
            "name": "Gemini比较视频",
            "params": {
                "videos": "视频列表",
                "comparison_criteria": "比较标准"
            }
        }
    }
)
async def gemini_compare_videos(
    videos: List[Dict[str, Any]],
    comparison_criteria: Optional[List[str]] = None,
) -> ToolResult:
    """
    使用Gemini比较多个视频

    Args:
        videos: 视频列表，每个视频包含：
            - title: 标题
            - description: 描述
            - tags: 标签列表
            - stats: 数据统计
        comparison_criteria: 比较标准，可选：
            - similarity: 相似度
            - quality: 质量
            - audience_fit: 受众匹配度
            - engagement: 互动表现

    Returns:
        ToolResult 包含比较结果：
        {
            "comparison": "比较分析文本",
            "rankings": [
                {
                    "video_index": 0,
                    "title": "...",
                    "score": 8.5,
                    "strengths": ["..."],
                    "weaknesses": ["..."]
                }
            ],
            "recommendations": ["..."]
        }
    """
    try:
        if not GEMINI_API_KEY:
            return ToolResult(
                title="配置错误",
                output="",
                error="未配置 GEMINI_API_KEY 环境变量"
            )

        if len(videos) < 2:
            return ToolResult(
                title="参数错误",
                output="",
                error="至少需要2个视频进行比较"
            )

        # 构建比较提示词
        criteria_text = "、".join(comparison_criteria) if comparison_criteria else "相似度、质量、受众匹配度"

        prompt = f"请比较以下{len(videos)}个视频，从{criteria_text}等方面进行分析：\n\n"

        for i, video in enumerate(videos, 1):
            prompt += f"视频{i}：\n"
            prompt += f"标题：{video.get('title', 'N/A')}\n"
            prompt += f"描述：{video.get('description', 'N/A')}\n"
            prompt += f"标签：{', '.join(video.get('tags', []))}\n"
            if 'stats' in video:
                prompt += f"数据：{json.dumps(video['stats'], ensure_ascii=False)}\n"
            prompt += "\n"

        prompt += "请提供详细的比较分析，并给出推荐建议。"

        # 调用Gemini API
        payload = {
            "contents": [{
                "parts": [{
                    "text": prompt
                }]
            }]
        }

        headers = {
            "Content-Type": "application/json",
        }

        url = f"{GEMINI_API_URL}?key={GEMINI_API_KEY}"

        async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
            response = await client.post(
                url,
                json=payload,
                headers=headers,
            )
            response.raise_for_status()
            data = response.json()

        # 提取比较结果
        candidates = data.get("candidates", [])
        if not candidates:
            return ToolResult(
                title="比较失败",
                output="",
                error="Gemini API 未返回结果"
            )

        comparison_text = candidates[0].get("content", {}).get("parts", [{}])[0].get("text", "")

        result = {
            "comparison": comparison_text,
            "video_count": len(videos),
            "criteria": comparison_criteria or ["similarity", "quality", "audience_fit"]
        }

        return ToolResult(
            title=f"Gemini比较: {len(videos)}个视频",
            output=json.dumps(result, ensure_ascii=False, indent=2),
            long_term_memory=f"Compared {len(videos)} videos with Gemini"
        )
    except httpx.HTTPStatusError as e:
        return ToolResult(
            title="API调用失败",
            output="",
            error=f"HTTP error {e.response.status_code}: {e.response.text}"
        )
    except Exception as e:
        return ToolResult(
            title="比较失败",
            output="",
            error=str(e)
        )