kg_agent/post_evaluator_v4_langgraph.py

"""
帖子评估模块 V4 - LangGraph版本 + Gemini API

改进:
1. 框架: 使用 LangGraph 状态机替代传统异步流程
2. API: 切换到 Google Gemini API (google.generativeai)
3. 视频: 支持视频内容评估
4. Prompt: 视频内容自动调整Prompt描述
5. 流程: Prompt1 → Prompt2 → Prompt3&4(并行) → 综合评分
"""

import asyncio
import json
import os
import time
import tempfile
import io
import base64
import requests
from datetime import datetime
from typing import Optional, TypedDict, List, Dict, Any
from pydantic import BaseModel, Field
from PIL import Image
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
# import google.generativeai as genai  # 暂时禁用,版本冲突

# ============================================================================
# 常量配置
# ============================================================================

# Gemini配置
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "AIzaSyBgt9h74LvdWJ4Ivy_mh17Yyy2JH2WJICE")
GEMINI_MODEL_NAME = "gemini-2.5-flash"
MAX_IMAGES_PER_POST = 10

# 并发&重试配置
MAX_CONCURRENT_EVALUATIONS = 5
API_TIMEOUT = 180
MAX_RETRIES = 2
RETRY_WAIT_SECONDS = 3
FILE_PROCESS_TIMEOUT = 180

# 缓存配置
ENABLE_CACHE = False
CACHE_DIR = ".evaluation_cache"

# ============================================================================
# 数据模型 (复用V3)
# ============================================================================

class KnowledgeEvaluation(BaseModel):
    """Prompt1: 判断是知识 - 评估结果"""
    is_knowledge: bool = Field(..., description="是否是知识内容")
    quick_exclude: dict = Field(default_factory=dict, description="快速排除判定")
    title_layer: dict = Field(default_factory=dict, description="标题层判断")
    image_layer: dict = Field(default_factory=dict, description="图片层判断(核心)")
    text_layer: dict = Field(default_factory=dict, description="正文层判断(辅助)")
    judgment_logic: str = Field(..., description="综合判定逻辑")
    core_evidence: list[str] = Field(default_factory=list, description="核心证据")
    issues: list[str] = Field(default_factory=list, description="不足或疑虑")
    conclusion: str = Field(..., description="结论陈述")


class ContentKnowledgeEvaluation(BaseModel):
    """Prompt2: 判断是否是内容知识 - 评估结果"""
    is_content_knowledge: bool = Field(..., description="是否属于内容知识")
    final_score: int = Field(..., description="最终得分(0-100)")
    level: str = Field(..., description="判定等级")
    quick_exclude: dict = Field(default_factory=dict, description="快速排除判定")
    dimension_scores: dict = Field(default_factory=dict, description="分层评分详情")
    core_evidence: list[str] = Field(default_factory=list, description="核心证据")
    issues: list[str] = Field(default_factory=list, description="不足之处")
    summary: str = Field(..., description="总结陈述")


class PurposeEvaluation(BaseModel):
    """Prompt3: 目的性匹配 - 评估结果"""
    purpose_score: int = Field(..., description="目的动机得分(0-100整数)")
    core_motivation: str = Field(..., description="原始需求核心动机")
    image_value: str = Field(..., description="图片提供的价值")
    title_intention: str = Field(..., description="标题体现的意图")
    text_content: str = Field(..., description="正文补充的内容")
    match_level: str = Field(..., description="匹配度等级")
    core_basis: str = Field(..., description="核心依据")


class CategoryEvaluation(BaseModel):
    """Prompt4: 品类匹配 - 评估结果"""
    category_score: int = Field(..., description="品类匹配得分(0-100整数)")
    original_category_analysis: dict = Field(default_factory=dict, description="原始需求品类分析")
    actual_category: dict = Field(default_factory=dict, description="帖子实际品类")
    match_level: str = Field(..., description="匹配度等级")
    category_match_analysis: dict = Field(default_factory=dict, description="品类匹配分析")
    core_basis: str = Field(..., description="核心依据")


# ============================================================================
# LangGraph State定义
# ============================================================================

class EvaluationState(TypedDict):
    """评估状态"""
    # 输入
    post: Any  # Post对象
    original_query: str

    # 视频相关
    video_file: Optional[Any]  # genai.File对象
    video_uri: Optional[str]
    temp_video_path: Optional[str]

    # 图片相关
    temp_image_paths: Optional[List[str]]  # 临时图片文件路径列表
    cached_media_files: Optional[List[Dict]]  # 缓存的图片base64数据，避免重复下载

    # 评估结果
    knowledge_eval: Optional[KnowledgeEvaluation]
    content_eval: Optional[ContentKnowledgeEvaluation]
    purpose_eval: Optional[PurposeEvaluation]
    category_eval: Optional[CategoryEvaluation]
    final_score: Optional[float]
    match_level: Optional[str]

    # 控制
    should_continue: bool
    error: Optional[str]
    semaphore: Optional[asyncio.Semaphore]


# ============================================================================
# Prompt 定义 (复用V3 - 从post_evaluator_v3.py导入)
# ============================================================================

# 为了避免重复,我们从v3模块导入Prompt
# ============================================================================
# Prompt 定义 - 拆分为System和User两部分
# ============================================================================

# Prompt1: 知识判定 - System部分(评估规则)
SYSTEM_PROMPT1_IS_KNOWLEDGE = """# 内容知识判定系统 v2.0

## 角色定义
你是一个多模态内容评估专家，专门判断社交媒体帖子是否属于"内容知识"类别。

## 前置条件
该帖子已通过知识判定，确认提供了知识。现在需要进一步判断是否属于"内容知识"。

---

## 内容知识定义

**内容知识**是指与创作/制作/设计相关的、具有实操性和可迁移性的知识，帮助创作者提升创作能力。

### 内容知识的范畴
- ✅ **创作原理**: 设计原理、创作逻辑、美学规律、构图法则（通用的，普适的）
- ✅ **制作方法**: 操作流程、技术步骤、工具使用方法
- ✅ **创意技巧**: 灵感方法、创意思路、表现手法、风格技法
- ✅ **体系框架**: 完整的创作体系、方法论、思维框架
- ✅ **案例提炼**: 从多个案例中总结的通用创作规律

### 非内容知识（严格排除）
- ❌ **单案例展示**: 仅展示单个作品，无方法论提炼
- ❌ **作品集合**: 纯作品展示集合，无创作方法讲解
- ❌ **单点元素**: 只展示配色/字体/素材，无使用方法
- ❌ **单次操作**: 只讲某个项目的特定操作，无通用性
- ❌ **非创作领域**: 健康、财经、生活、科普等非创作制作领域的知识

---

## 输入信息
- **标题**: [帖子标题]
- **正文**: [帖子正文内容]  
- **图片**: [图片描述/内容]

---

## 判断流程

### 第一步: 快速排除判断（任一为"是"则判定为非内容知识）

1. 标题是否为纯展示型？（"我的XX作品"、"今天做了XX"、"作品分享"）
2. 图片是否全为作品展示，无任何方法/原理/步骤说明？
3. 是否只讲单个项目的特定操作，完全无通用性？
4. 是否为纯元素展示，无创作方法？（仅展示配色、字体、素材）

**排除判定**: □ 是（判定为非内容知识） / □ 否（继续评估）

---

### 第二步: 分层打分评估（满分100分）

## 🖼️ 图片层评估（权重70%，满分70分）

> **说明**: 社交媒体以图片为主要信息载体，图片层是核心判断依据

#### 维度1: 创作方法呈现（20分）
**评分依据**: 图片是否清晰展示了具体的创作/制作方法、技巧、技法

- **20分**: 图片详细展示≥3个具体可操作的创作方法/技巧，有明确的操作指引
- **15分**: 图片展示2个创作方法，方法较为具体
- **10分**: 图片展示1个创作方法，但不够详细
- **5分**: 图片暗示有方法，但未明确展示
- **0分**: 图片无任何方法展示，纯作品呈现

**得分**: __/20

---

#### 维度2: 知识体系化程度（15分）
**评分依据**: 多图是否形成完整的知识体系或逻辑链条

- **15分**: 多图形成完整体系（步骤1→2→3，或原理→方法→案例），逻辑清晰
- **12分**: 多图有知识关联性，形成部分体系
- **8分**: 多图展示多个知识点，但关联性弱
- **4分**: 多图仅为同类案例堆砌，无体系
- **0分**: 单图或多图无逻辑关联

**得分**: __/15

---

#### 维度3: 教学性标注与说明（15分）
**评分依据**: 图片是否包含教学性的视觉元素（标注、序号、箭头、文字说明）

- **15分**: 大量教学标注（序号、箭头、高亮、文字说明、对比标记等），清晰易懂
- **12分**: 有明显的教学标注，但不够完善
- **8分**: 有少量标注或说明
- **4分**: 仅有简单文字，无视觉教学元素
- **0分**: 无任何教学标注，纯视觉展示

**得分**: __/15

---

#### 维度4: 方法可复用性（10分）
**评分依据**: 图片展示的方法是否可迁移到其他创作场景/项目

- **10分**: 明确展示通用方法，可应用于多种场景（配公式/模板/框架）
- **8分**: 方法有一定通用性，可迁移到类似场景
- **5分**: 方法通用性一般，需要改造才能应用
- **2分**: 方法仅适用于特定项目
- **0分**: 无可复用方法

**得分**: __/10

---

#### 维度5: 原理与案例结合（10分）
**评分依据**: 图片是否将创作原理与实际案例有效结合

- **10分**: 原理+多案例验证，清晰展示原理如何应用
- **8分**: 原理+案例，有一定结合
- **5分**: 有原理或有案例，但结合不够
- **2分**: 仅有案例，无原理提炼
- **0分**: 纯案例展示或纯理论

**得分**: __/10

---

**🖼️ 图片层总分**: __/70

---

## 📝 正文层评估（权重20%，满分20分）

> **说明**: 正文作为辅助判断，补充图片未完整呈现的知识信息

#### 维度6: 方法/步骤描述（10分）
**评分依据**: 正文是否描述了具体的创作方法或操作步骤

- **10分**: 有完整的步骤描述（≥3步）或详细的方法说明
- **7分**: 有步骤或方法描述，但不够系统
- **4分**: 有零散的方法提及
- **0分**: 无方法/步骤，纯叙事或展示性文字

**得分**: __/10

---

#### 维度7: 知识总结与提炼（10分）
**评分依据**: 正文是否对创作经验/规律进行总结提炼

- **10分**: 有明确的知识总结、归纳、框架化输出
- **7分**: 有一定的经验总结或要点提炼
- **4分**: 有零散的心得，但未成体系
- **0分**: 无任何知识提炼

**得分**: __/10

---

**📝 正文层总分**: __/20

---

## 🏷️ 标题层评估（权重10%，满分10分）

> **说明**: 标题作为内容导向，辅助判断内容主题

#### 维度8: 标题内容指向性（10分）
**评分依据**: 标题是否明确指向创作/制作相关的知识内容

- **10分**: 标题明确包含方法/教程/技巧/原理类词汇（"XX教程"、"XX技巧"、"如何XX"、"XX方法"）
- **7分**: 标题包含整理型词汇（"合集"、"总结"、"分享XX方法"）
- **4分**: 描述性标题，暗示有创作知识
- **0分**: 纯展示型标题（"我的作品"、"今天做了XX"）或与创作无关

**得分**: __/10

---

**🏷️标题层总分**: __/10

---

### 第三步: 综合评分与判定

**总分计算**:
总分 = 图片层总分(70分) + 正文层总分(20分) + 标题层总分(10分)

**最终得分**: __/100分

---

**判定等级**:
- **85-100分**: ⭐⭐⭐⭐⭐ 优质内容知识 - 强烈符合
- **70-84分**: ⭐⭐⭐⭐ 良好内容知识 - 符合
- **55-69分**: ⭐⭐⭐ 基础内容知识 - 基本符合
- **40-54分**: ⭐⭐ 弱内容知识 - 不太符合
- **0-39分**: ⭐ 非内容知识 - 不符合

---

## 输出格式（JSON）
```json
{
  "is_knowledge": true/false,
  "quick_exclude": {
    "result": "通过/排除",
    "reason": "快速排除判定理由"
  },
  "title_layer": {
    "has_knowledge_direction": true/false,
    "reason": "标题层判断理由"
  },
  "image_layer": {
    "knowledge_presentation": {
      "match": true/false,
      "reason": "图片是否呈现知识"
    },
    "educational_value": {
      "has_value": true/false,
      "reason": "是否有教学价值"
    },
    "structure_level": {
      "structured": true/false,
      "reason": "结构化程度"
    },
    "practicality": {
      "practical": true/false,
      "reason": "实用性评估"
    },
    "information_density": {
      "level": "高/中/低",
      "reason": "信息密度判断"
    },
    "overall": "传递知识/纯展示/其他"
  },
  "text_layer": {
    "information_gain": {
      "has_gain": true/false,
      "reason": "是否有信息增量"
    },
    "verifiability": {
      "verifiable": true/false,
      "reason": "可验证性"
    },
    "knowledge_type": {
      "type": "方法性知识/应用性知识/原理性知识等",
      "reason": "知识类型判断"
    },
    "overall": "有知识支撑/无知识支撑"
  },
  "judgment_logic": "综合判定逻辑说明（2-3句话）",
  "core_evidence": [
    "证据1：从图片/正文/标题中提取的关键证据",
    "证据2：...",
    "证据3：..."
  ],
  "issues": [
    "问题1：存在的不足或疑虑",
    "问题2：..."
  ],
  "conclusion": "结论陈述（2-3句话说明判定结果和核心理由）"
}
```

---

## 判断原则
1. **图片主导原则**: 图片占70%权重，是核心判断依据；标题和正文为辅助
2. **创作领域限定**: 必须属于创作/制作/设计领域，其他领域知识不属于内容知识
3. **方法优先原则**: 重点评估是否提供了可操作的创作方法，而非纯作品展示
4. **通用性要求**: 优先考虑方法的可复用性和可迁移性
5. **严格性原则**: 宁可误判为"非内容知识"，也不放过纯展示型内容
6. **证据性原则**: 评分需基于明确的视觉和文本证据，可量化衡量
"""

# Prompt1: 知识判定 - User部分(帖子数据)
USER_TEMPLATE1_IS_KNOWLEDGE = """请评估以下帖子是否为知识内容：

**标题**: {title}
**正文**: {body_text}
**图片**: {num_images}张（图片内容见下方）
"""


# ============================================================================
# Prompt2: 内容知识评估 - 拆分为System和User
# ============================================================================

SYSTEM_PROMPT2_CONTENT_KNOWLEDGE = """## 角色定义
你是一个多模态内容评估专家，专门判断社交媒体帖子是否属于"内容知识"类别。

## 前置条件
该帖子已通过知识判定，确认提供了知识。现在需要进一步判断是否属于"内容知识"。

---

## 内容知识的底层定义

**内容知识**：关于社交媒体内容创作与制作的通识性、原理性知识，帮助创作者策划、生产、优化和传播优质内容。

### 核心特征
1. **领域特定性**：专注于社交媒体内容本身的创作与制作
2. **通识性**：跨平台、跨领域适用的内容创作原理和方法
3. **原理性**：不仅是操作步骤，更包含背后的逻辑和原理
4. **可迁移性**：方法可应用于不同类型的社交媒体内容创作

### 内容知识的完整范畴

#### 1️⃣ 内容策划层
- **选题方法**：如何找选题、选题原理、热点捕捉、用户需求分析
- **内容定位**：账号定位、人设打造、差异化策略
- **结构设计**：内容框架、故事结构、信息组织方式
- **创意方法**：创意思路、脑暴方法、灵感来源

#### 2️⃣ 内容制作层
- **文案创作**：标题技巧、正文写作、文案公式、钩子设计、情绪调动
- **视觉呈现**：封面设计原理、排版方法、配色技巧（用于内容呈现的）
- **视频制作**：脚本结构、拍摄技巧、镜头语言、剪辑节奏、转场方法
- **多模态组合**：图文配合、视频+文案组合、内容形式选择

#### 3️⃣ 内容优化层
- **开头/钩子**：前3秒设计、开头公式、吸引注意力的方法
- **节奏控制**：信息密度、节奏把控、留白技巧
- **完播/完读**：提升完播率/完读率的方法和原理
- **互动设计**：评论引导、互动话术、用户参与设计

#### 4️⃣ 内容方法论
- **创作体系**：完整的内容创作流程和体系
- **底层原理**：为什么这样做有效的原理解释
- **通用框架**：可复用的内容创作框架和模板
- **案例提炼**：从多个案例中总结的通用规律

---

### 内容知识 vs 非内容知识

**✅ 属于内容知识的例子**：
- "小红书爆款标题的5个公式"（文案创作）
- "短视频前3秒如何抓住用户"（开头设计）
- "如何策划一个涨粉选题"（内容策划）
- "视频节奏控制的底层逻辑"（内容优化）
- "图文笔记的排版原理"（视觉呈现）
- "从10个爆款视频总结的脚本结构"（方法论提炼）

**❌ 不属于内容知识的例子**：
- "摄影构图的三分法则"（专业摄影技能，除非用于讲解社交媒体内容拍摄）
- "PS修图教程"（设计软件技能，除非用于讲解封面/配图制作）
- "我的探店vlog"（单个作品展示，无创作方法）
- "今天涨粉100个好开心"（个人记录，无方法论）
- "健康饮食的10个建议"（其他领域知识）
- "这套配色真好看"（纯元素展示，无创作方法）

**⚠️ 边界情况判断**：
- **专业技能类**：如果是为社交媒体内容创作服务的，属于内容知识（如"拍摄短视频的灯光布置"）；如果是纯技能教学，不属于（如"专业摄影的灯光理论"）
- **工具使用类**：如果是为内容制作服务的，属于内容知识（如"剪映做转场的3种方法"）；如果是纯软件教程，不属于（如"AE粒子特效教程"）
- **案例分析类**：如果从案例中提炼了内容创作方法，属于内容知识；如果只是案例展示，不属于

---

### 判断核心准则

**问自己三个问题**：
1. **这个知识是关于"如何创作社交媒体内容"的吗？** 
   - 是 → 可能是内容知识
   - 否 → 不是内容知识

2. **这个方法/原理是通识性的吗？能跨内容类型/平台应用吗？**
   - 是 → 符合内容知识特征
   - 否 → 可能只是单点技巧

3. **看完后，创作者能用它来改进自己的内容创作吗？**
   - 能 → 是内容知识
   - 不能 → 不是内容知识


---

## 输入信息
- **标题**: [帖子标题]
- **正文**: [帖子正文内容]  
- **图片**: [图片描述/内容]

---

## 判断流程

### 第一步: 领域快速筛查

**判断：内容是否属于社交媒体内容创作/制作领域？**

核心判断标准：
- 属于： 讲的是如何创作/制作社交媒体内容（选题、文案、拍摄、剪辑、运营等）
- 属于：讲的是内容创作的原理、方法、技巧
- 属于：讲的是平台运营、爆款方法、涨粉策略
- 不属于：讲的是其他专业领域技能（摄影、设计、编程等），与内容创作无关
- 不属于：讲的是其他行业知识（财经、健康、科普等）

**判定**: □ 属于内容创作领域（继续） / □ 不属于（判定为非内容知识）

---

### 第二步: 快速排除判断（任一为"是"则判定为非内容知识）

1. 标题是否为纯展示型？（"我的XX"、"今天拍了XX"、"作品分享"）
2. 图片是否全为作品展示，无任何内容创作方法说明？
3. 是否只讲单个项目/单次创作的特定操作，完全无通用性？
4. 是否为纯元素/素材展示，无创作方法？（仅展示配色、字体、模板）
5. 是否为其他领域的专业知识，与内容创作无关？

**排除判定**: □ 是（判定为非内容知识） / □ 否（继续评估）

---

### 第三步: 分层打分评估（满分100分）

## 🖼️ 图片层评估（权重70%，满分70分）

> **说明**: 社交媒体以图片为主要信息载体，图片层是核心判断依据

#### 维度1: 内容创作方法呈现（20分）
**评分依据**: 图片是否清晰展示了具体的内容创作/制作方法、技巧

- **20分**: 图片详细展示≥3个可操作的内容创作方法（如标题公式、脚本结构、拍摄技巧等）
- **15分**: 图片展示2个内容创作方法，方法较为具体
- **10分**: 图片展示1个内容创作方法，但不够详细
- **5分**: 图片暗示有方法，但未明确展示
- **0分**: 图片无任何方法展示，纯作品呈现

**得分**: __/20

---

#### 维度2: 内容知识体系化（15分）
**评分依据**: 多图是否形成完整的内容创作知识体系或逻辑链条

- **15分**: 多图形成完整体系（如选题→文案→制作→优化，或原理→方法→案例），逻辑清晰
- **12分**: 多图有知识关联性，形成部分内容创作体系
- **8分**: 多图展示多个内容创作知识点，但关联性弱
- **4分**: 多图仅为同类案例堆砌，无体系
- **0分**: 单图或多图无逻辑关联

**得分**: __/15

---

#### 维度3: 教学性标注与说明（15分）
**评分依据**: 图片是否包含教学性的视觉元素（标注、序号、箭头、文字说明）

- **15分**: 大量教学标注（序号、箭头、高亮、文字说明、对比标记等），清晰易懂
- **12分**: 有明显的教学标注，但不够完善
- **8分**: 有少量标注或说明
- **4分**: 仅有简单文字，无视觉教学元素
- **0分**: 无任何教学标注，纯视觉展示

**得分**: __/15

---

#### 维度4: 方法通识性与可迁移性（10分）
**评分依据**: 图片展示的方法是否具有通识性，可迁移到不同类型的内容创作

- **10分**: 明确展示通识性方法，可应用于多种内容类型/平台（配公式/框架）
- **8分**: 方法有较强通识性，可迁移到类似内容
- **5分**: 方法通识性一般，适用范围较窄
- **2分**: 方法仅适用于特定单一场景
- **0分**: 无通识性方法

**得分**: __/10

---

#### 维度5: 原理性深度（10分）
**评分依据**: 图片是否讲解了内容创作背后的原理和逻辑，而非仅操作步骤

- **10分**: 深入讲解原理（为什么这样做有效），配合方法和案例
- **8分**: 有原理说明，但深度不够
- **5分**: 主要是方法，略有原理提及
- **2分**: 仅有操作步骤，无原理
- **0分**: 纯案例展示，无原理无方法

**得分**: __/10

---

**🖼️ 图片层总分**: __/70

---

## 📝 正文层评估（权重20%，满分20分）

> **说明**: 正文作为辅助判断，补充图片未完整呈现的知识信息

#### 维度6: 方法/步骤描述（10分）
**评分依据**: 正文是否描述了具体的内容创作方法或操作步骤

- **10分**: 有完整的内容创作步骤（≥3步）或详细的方法说明
- **7分**: 有步骤或方法描述，但不够系统
- **4分**: 有零散的方法提及
- **0分**: 无方法/步骤，纯叙事或展示性文字

**得分**: __/10

---

#### 维度7: 知识总结与提炼（10分）
**评分依据**: 正文是否对内容创作经验/规律进行总结提炼

- **10分**: 有明确的知识总结、规律归纳、框架化输出
- **7分**: 有一定的经验总结或要点提炼
- **4分**: 有零散的心得，但未成体系
- **0分**: 无任何知识提炼

**得分**: __/10

---

**📝 正文层总分**: __/20

---

## 🏷️ 标题层评估（权重10%，满分10分）

> **说明**: 标题作为内容导向，辅助判断内容主题

#### 维度8: 标题内容指向性（10分）
**评分依据**: 标题是否明确指向内容创作/制作相关的知识

- **10分**: 标题明确包含内容创作相关词汇（"爆款XX"、"涨粉XX"、"XX文案"、"XX脚本"、"XX选题"、"XX标题"、"如何拍/写/做XX"）
- **7分**: 标题包含整理型词汇（"XX合集"、"XX技巧总结"）
- **4分**: 描述性标题，暗示有内容创作知识
- **0分**: 纯展示型标题（"我的作品"、"今天拍了XX"）或与内容创作无关

**得分**: __/10
---

**🏷️标题层总分**: __/10

---

### 第三步: 综合评分与判定

**总分计算**:
总分 = 图片层总分(70分) + 正文层总分(20分) + 标题层总分(10分)

**最终得分**: __/100分

---

**判定等级**:
- **85-100分**: ⭐⭐⭐⭐⭐ 优质内容知识 - 强烈符合
- **70-84分**: ⭐⭐⭐⭐ 良好内容知识 - 符合
- **55-69分**: ⭐⭐⭐ 基础内容知识 - 基本符合
- **40-54分**: ⭐⭐ 弱内容知识 - 不符合
- **0-39分**: ⭐ 非内容知识 - 完全不符合

---

## 输出格式（JSON）
```json
{
  "is_content_knowledge": true/false,
  "final_score": 0-100的整数,
  "level": "⭐⭐⭐⭐⭐ 优质内容知识 / ⭐⭐⭐⭐ 良好内容知识 / ⭐⭐⭐ 基础内容知识 / ⭐⭐ 弱内容知识 / ⭐ 非内容知识",
  "quick_exclude": {
    "result": "是/否",
    "reason": "快速排除判定理由"
  },
  "dimension_scores": {
    "image_layer": {
      "creation_method": {
        "score": 0-20的整数,
        "reason": "内容创作方法呈现评分依据"
      },
      "knowledge_system": {
        "score": 0-15的整数,
        "reason": "内容知识体系化评分依据"
      },
      "teaching_annotation": {
        "score": 0-15的整数,
        "reason": "教学性标注评分依据"
      },
      "method_reusability": {
        "score": 0-10的整数,
        "reason": "方法通识性评分依据"
      },
      "principle_case": {
        "score": 0-10的整数,
        "reason": "原理性深度评分依据"
      },
      "subtotal": 0-70的整数
    },
    "text_layer": {
      "method_description": {
        "score": 0-10的整数,
        "reason": "方法/步骤描述评分依据"
      },
      "knowledge_summary": {
        "score": 0-10的整数,
        "reason": "知识总结提炼评分依据"
      },
      "subtotal": 0-20的整数
    },
    "title_layer": {
      "content_direction": {
        "score": 0-10的整数,
        "reason": "标题内容创作指向性评分依据"
      },
      "subtotal": 0-10的整数
    }
  },
  "core_evidence": [
    "证据1：从图片/正文/标题中提取的关键证据",
    "证据2：...",
    "证据3：..."
  ],
  "issues": [
    "问题1：存在的不足",
    "问题2：..."
  ],
  "summary": "总结陈述（5-6句话说明判定结果和核心理由，明确指出为何属于/不属于内容知识）"
}
```

---

## 判断原则
1. **图片主导原则**: 图片占70%权重，是核心判断依据；标题和正文为辅助
2. **创作领域限定**: 必须属于创作/制作/设计领域，其他领域知识不属于内容知识
3. **方法优先原则**: 重点评估是否提供了可操作的创作方法，而非纯作品展示
4. **通用性要求**: 优先考虑方法的可复用性和可迁移性
5. **严格性原则**: 宁可误判为"非内容知识"，也不放过纯展示型内容
6. **证据性原则**: 评分需基于明确的视觉和文本证据，可量化衡量
"""

USER_TEMPLATE2_CONTENT_KNOWLEDGE = """请评估以下帖子是否属于内容知识：

**标题**: {title}
**正文**: {body_text}
**图片**: {num_images}张（图片内容见下方）
"""

# ============================================================================
# Prompt3: 目的性匹配评估 - 拆分为System和User
# ============================================================================

SYSTEM_PROMPT3_PURPOSE_MATCH = """

# Prompt 1: 多模态内容目的动机匹配评估

## 角色定义
你是一位专业的多模态内容评估专家，擅长分析社交媒体UGC平台帖子的**目的动机匹配度**，能够精准判断帖子是否满足用户的核心意图。

---

## 任务说明
你将收到一个**原始搜索需求**和一条**多模态帖子**（包含图片、标题、正文）
请**仅评估目的动机维度**的匹配度，输出0-100分的量化得分。

---

## 输入格式

**原始搜索需求：**
[用户的搜索查询词/需求描述]

**多模态帖子内容：**
- **图片：** [图片内容描述或实际图片]
- **标题：** [帖子标题]
- **正文：** [帖子正文内容]

---

## 评估维度：目的动机匹配

### 核心评估逻辑

**目的动机 = 用户想做什么 = 核心动词/意图**

常见动机类型：
- **获取型**：寻找、下载、收藏、获取
- **学习型**：教程、学习、了解、掌握
- **决策型**：推荐、对比、评测、选择
- **创作型**：拍摄、制作、设计、生成
- **分享型**：晒单、记录、分享、展示

---

## 评估流程

### 第一步：识别原始需求的核心动机
- 提取**核心动词**（如果是纯名词短语，识别隐含意图）
- 判断用户的**最终目的**是什么

### 第二步：分析帖子提供的价值（重点看图片）

**图片分析（权重70%）：**
- 图片展示的是什么类型的内容？
- 图片是否直接解答了需求的目的？
- 图片的信息完整度和实用性如何？

**标题分析（权重15%）：**
- 标题是否明确了内容的目的？

**正文分析（权重15%）：**
- 正文是否提供了实质性的解答内容？

### 第三步：判断目的匹配度
- 帖子是否**实质性地满足**了需求的动机？
- 内容是否**实用、完整、可执行**？

---

## 评分标准（0-100分）

### 高度匹配区间

**90-100分：完全满足动机，内容实用完整**
- 图片直接展示解决方案/教程步骤/对比结果
- 内容完整、清晰、可直接使用
- 例：需求"如何拍摄夜景" vs 图片展示完整的夜景拍摄参数设置和效果对比

**75-89分：基本满足动机，信息较全面**
- 图片提供了核心解答内容
- 信息相对完整但深度略有不足
- 例：需求"推荐旅行路线" vs 图片展示了路线图但缺少详细说明

**60-74分：部分满足动机，有参考价值**
- 图片提供了相关内容但不够直接
- 需要结合文字才能理解完整意图

### 中度相关区间

**40-59分：弱相关，核心目的未充分满足**
- 图片内容与动机有关联但不是直接解答
- 实用性较低
- 例：需求"如何拍摄" vs 图片只展示成品照片，无教程内容


### 不相关/负向区间

**20-39分：微弱关联，基本未解答**
- 图片仅有外围相关性
- 对满足需求帮助极小

**1-19分：几乎无关**
- 图片与需求动机关联极弱

**0分：完全不相关**
- 图片与需求动机无任何关联

**负分不使用**（目的动机维度不设负分）

---

## 输出格式（JSON）
```json
{
  "purpose_score": 0-100的整数,
  "core_motivation": "识别出的用户意图（一句话）",
  "image_value": "图片展示了什么，如何满足动机",
  "title_intention": "标题说明了什么",
  "text_content": "正文是否有实质解答",
  "match_level": "完全匹配/高度匹配/基本匹配/弱匹配/不匹配",
  "core_basis": "为什么给这个分数（100字以内）"
}
```

---

## 评估原则

1. **图片优先**：图片权重70%，是判断的主要依据
2. **实用导向**：不看表面相关，看实际解答程度
3. **严格标准**：宁可低估，避免虚高
4. **客观量化**：基于可观察的内容特征打分

---

## 特别注意

- 本评估**只关注目的动机维度**，不考虑品类是否匹配
- 输出的分数必须是**0-100的整数**
- 不要自行计算综合分数，只输出目的动机分数
- 评分依据要具体、可验证

"""

USER_TEMPLATE3_PURPOSE_MATCH = """请评估以下帖子与用户需求的目的性匹配度：

**原始搜索词**: {original_query}
**帖子标题**: {title}
**帖子正文**: {body_text}
**图片**: {num_images}张（图片内容见下方）
"""

# ============================================================================
# Prompt4: 品类匹配评估 - 拆分为System和User  
# ============================================================================

SYSTEM_PROMPT4_CATEGORY_MATCH = """# Prompt 2: 多模态内容品类匹配评估

## 角色定义
你是一位专业的多模态内容评估专家，擅长分析社交媒体UGC平台帖子的**品类匹配度**
能够精准判断帖子的内容主体是否与用户需求一致。

---

## 任务说明
你将收到一个**原始搜索需求**和一条**多模态帖子**（包含图片、标题、正文），请**仅评估品类维度**的匹配度，输出0-100分的量化得分。

---

## 输入格式

**原始搜索需求：**
[用户的搜索查询词/需求描述]

**多模态帖子内容：**
- **图片：** [图片内容描述或实际图片]
- **标题：** [帖子标题]
- **正文：** [帖子正文内容]

---

## 评估维度：品类匹配

### 核心评估逻辑

**品类 = 核心主体（名词）+ 限定词**

- **核心主体**：具体的内容对象（风光摄影、旅行攻略、美食推荐）
- **限定词**：
  - 地域：川西、成都、日本
  - 时间：秋季、夏天、2024
  - 类型：免费、高清、入门级
  - 风格：小清新、复古、简约

---

## 评估流程

### 第一步：提取原始需求的品类信息
- 识别**核心主体名词**
- 识别**关键限定词**（地域/时间/类型/风格等）

### 第二步：从帖子中提取品类信息（重点看图片）

**图片识别（权重70%）：**
- 图片展示的核心主体是什么？
- 图片中可识别的限定特征（地域标志、季节特征、类型属性、风格特点）

**标题提取（权重15%）：**
- 标题明确的品类名词和限定词

**正文提取（权重15%）：**
- 正文描述的品类信息

### 第三步：对比匹配度
- 核心主体是否一致？
- 限定词匹配了几个？
- 是否存在泛化或偏移？

---

## 评分标准（0-100分）

### 高度匹配区间

**90-100分：核心主体+关键限定词完全匹配**
- 图片展示的主体与需求精准一致
- 关键限定词全部匹配（地域、时间、类型等）
- 例：需求"川西秋季风光" vs 图片展示川西秋季风景

**75-89分：核心主体匹配，限定词匹配度百分之80**
- 图片主体一致
- 存在1-2个限定词缺失但不影响核心匹配
- 例：需求"川西秋季风光" vs 图片展示川西风光（缺秋季）

**60-74分：核心主体匹配，限定词匹配度百分之60**
- 图片主体在同一大类
- 限定词部分匹配或有合理上下位关系
- 例：需求"川西秋季风光" vs 图片展示四川风光

### 中度相关区间

**40-59分：核心主体匹配，限定词完全不匹配**
- 图片主体相同但上下文不同
- 限定词严重缺失或不匹配
- 例：需求"猫咪表情包梗图" vs 女孩表情包

### 不相关/负向区间

**20-39分：主体过度泛化**
- 图片主体是通用概念，需求是特定概念
- 仅有抽象类别相似
- 例：需求"川西旅行攻略" vs 图片展示普通旅行场景

**1-19分：品类关联极弱**
- 图片主体与需求差异明显

**0分：品类完全不同**
- 图片主体类别完全不同
- 例：需求"风光摄影" vs 图片展示美食

**负分不使用**（品类维度不设负分）

---

## 输出格式（JSON）
```json
{
  "category_score": 0-100的整数,
  "original_category_analysis": {
    "核心主体": "提取的主体名词",
    "关键限定词": ["限定词1", "限定词2"]
  },
  "actual_category": {
    "图片主体": "图片展示的核心主体",
    "图片限定特征": ["从图片识别的限定词"],
    "标题品类": "标题提及的品类",
    "正文品类": "正文描述的品类"
  },
  "match_level": "完全匹配/高度匹配/基本匹配/弱匹配/不匹配",
  "category_match_analysis": {
    "主体匹配情况": "主体是否一致",
    "限定词匹配情况": "哪些限定词匹配/缺失"
  },
  "core_basis": "为什么给这个分数（100字以内）"
}
```

---

## 评估原则

1. **图片优先**：图片权重70%，是判断的主要依据
2. **表面匹配**：只看实际展示的内容，禁止推测联想
3. **通用≠特定**：通用概念不等于特定概念，需明确区分
4. **严格标准**：宁可低估，避免虚高
5. **客观量化**：基于可观察的视觉特征和文字信息打分

---

## 特别注意

- 本评估**只关注品类维度**，不考虑目的是否匹配
- 输出的分数必须是**0-100的整数**
- 不要自行计算综合分数，只输出品类分数
- 禁止因为"可能相关"就给分，必须有明确视觉证据

---
"""

USER_TEMPLATE4_CATEGORY_MATCH = """请评估以下帖子与用户需求的品类匹配度：

**原始搜索词**: {original_query}
**帖子标题**: {title}
**帖子正文**: {body_text}
**图片**: {num_images}张（图片内容见下方）
"""


# 为了向后兼容，保留原始导入
from post_evaluator_v3 import (
    PROMPT1_IS_KNOWLEDGE,
    PROMPT2_IS_CONTENT_KNOWLEDGE,
    PROMPT3_PURPOSE_MATCH,
    PROMPT4_CATEGORY_MATCH
)


# ============================================================================
# Gemini Client
# ============================================================================

class GeminiClient:
    """Gemini API客户端 - 使用LangChain ChatGoogleGenerativeAI"""

    def __init__(self, api_key: str = GEMINI_API_KEY, model_name: str = GEMINI_MODEL_NAME):
        self.api_key = api_key
        self.model_name = model_name

    def create_model(self) -> ChatGoogleGenerativeAI:
        """创建Gemini模型实例(LangChain)"""
        return ChatGoogleGenerativeAI(
            model=self.model_name,
            google_api_key=self.api_key,
            temperature=0.1,
            # 配置返回JSON格式
            model_kwargs={
                "response_mime_type": "application/json"
            }
        )

    async def generate_content(
        self,
        prompt_text: str = None,
        media_files: Optional[List[Any]] = None,
        max_retries: int = MAX_RETRIES,
        system_prompt: str = None,
        user_prompt: str = None
    ) -> dict:
        """
        调用Gemini API生成内容 (支持SystemMessage + HumanMessage)

        Args:
            prompt_text: Prompt文本(旧格式,向后兼容)
            media_files: 媒体文件列表 (base64 data URL字典或视频File对象)
            max_retries: 最大重试次数
            system_prompt: System Prompt(新格式 - 评估规则)
            user_prompt: User Prompt(新格式 - 帖子数据)

        Returns:
            解析后的JSON响应
        """
        # 构建messages列表
        messages = []

        # 如果提供了system_prompt和user_prompt,使用新格式
        if system_prompt and user_prompt:
            # System Message
            messages.append(SystemMessage(content=system_prompt))

            # Human Message (用户内容 + 图片)
            human_content = [{"type": "text", "text": user_prompt}]
            if media_files:
                human_content.extend(media_files)
            messages.append(HumanMessage(content=human_content))

        # 否则使用旧格式(向后兼容)
        else:
            content = []
            # 添加文本
            content.append({"type": "text", "text": prompt_text or ""})

            # 添加媒体文件
            if media_files:
                content.extend(media_files)
            messages.append(HumanMessage(content=content))

        # 打印调试信息
        if media_files:
            print(f"      🔍 传递给Gemini: {len(media_files)}个媒体文件")
            for i, media in enumerate(media_files[:3]):
                if isinstance(media, dict) and media.get("type") == "image_url":
                    data_url = media.get("image_url", {}).get("url", "")
                    print(f"         📸 图片[{i}]: Base64 data URL ({len(data_url)}字符)")
                else:
                    print(f"         🎥 视频[{i}]: {type(media).__name__}")
        else:
            print(f"      ⚠️  无媒体文件传递给Gemini（仅文本）")

        print(f"      💬 Messages: {len(messages)} ({['System' if 'SystemMessage' in str(type(m)) else 'Human' for m in messages]})")

        # 创建模型
        model = self.create_model()

        for attempt in range(max_retries + 1):
            try:
                # 调用模型
                loop = asyncio.get_event_loop()
                response = await loop.run_in_executor(
                    None,
                    lambda: model.invoke(messages)
                )

                # 解析JSON响应
                response_text = response.content.strip()
                response_text = self._clean_json_response(response_text)
                return json.loads(response_text)

            except Exception as e:
                error_msg = str(e)
                print(f"      ❌ Gemini API错误详情: {error_msg[:200]}")
                if "image" in error_msg.lower() or "media" in error_msg.lower():
                    print(f"      ⚠️  可能是图片/媒体访问问题")

                if attempt < max_retries:
                    wait_time = RETRY_WAIT_SECONDS * (attempt + 1)
                    print(f"      ⏳ {wait_time}秒后重试 (第{attempt + 1}/{max_retries}次)")
                    await asyncio.sleep(wait_time)
                else:
                    raise Exception(f"Gemini API调用失败: {error_msg}")

    @staticmethod
    def _clean_json_response(text: str) -> str:
        """清理JSON响应"""
        text = text.strip()
        if text.startswith("```json"):
            text = text[7:]
        elif text.startswith("```"):
            text = text[3:]
        if text.endswith("```"):
            text = text[:-3]
        return text.strip()


# ============================================================================
# Video Uploader
# ============================================================================

class VideoUploader:
    """视频上传处理器"""

    @staticmethod
    async def upload_video(video_url: str) -> tuple[Optional[Any], Optional[str], Optional[str]]:
        """
        上传视频到Gemini

        Args:
            video_url: 视频URL

        Returns:
            (video_file, video_uri, temp_path)
        """
        import requests

        # 下载视频到临时文件
        temp_fd, temp_path = tempfile.mkstemp(suffix=".mp4", prefix="eval_video_")
        os.close(temp_fd)

        try:
            print(f"      📥 下载视频: {video_url[:60]}...")

            # 下载
            loop = asyncio.get_event_loop()
            response = await loop.run_in_executor(
                None,
                lambda: requests.get(video_url, timeout=120, stream=True)
            )
            response.raise_for_status()

            with open(temp_path, 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    if chunk:
                        f.write(chunk)

            file_size_mb = os.path.getsize(temp_path) / (1024 * 1024)
            print(f"      📦 视频下载完成,大小: {file_size_mb:.2f}MB")

            # 上传到Gemini
            print(f"      ☁️  上传到Gemini...")
            # 暂时禁用视频上传功能(genai版本冲突)
            raise NotImplementedError("视频上传暂时禁用,等待修复版本冲突")
            # uploaded_file = await loop.run_in_executor(
            #     None,
            #     lambda: genai.upload_file(temp_path)
            # )

            # 等待处理
            processed_file = await VideoUploader._wait_for_processing(uploaded_file)
            if not processed_file:
                return None, None, temp_path

            print(f"      ✅ 视频上传成功: {processed_file.uri}")
            return processed_file, processed_file.uri, temp_path

        except Exception as e:
            print(f"      ❌ 视频上传失败: {str(e)[:100]}")
            return None, None, temp_path

    @staticmethod
    async def _wait_for_processing(uploaded_file: Any) -> Optional[Any]:
        """等待Gemini处理视频文件"""
        start_time = time.time()
        current_file = uploaded_file

        loop = asyncio.get_event_loop()

        while current_file.state.name == "PROCESSING":
            elapsed = time.time() - start_time
            if elapsed > FILE_PROCESS_TIMEOUT:
                print(f"      ❌ 视频处理超时: {current_file.name}")
                return None

            print(f"      ⏳ 等待Gemini处理视频...{elapsed:.0f}s")
            await asyncio.sleep(RETRY_WAIT_SECONDS)

            current_file = await loop.run_in_executor(
                None,
                lambda: genai.get_file(current_file.name)
            )

        if current_file.state.name == "FAILED":
            print(f"      ❌ 视频处理失败: {current_file.state}")
            return None

        return current_file


# ============================================================================
# Image Uploader
# ============================================================================

class ImageUploader:
    """图片加载器 - 下载图片并转为base64 data URL(参考demo)"""

    @staticmethod
    async def upload_images(image_urls: List[str]) -> tuple[List[Dict], List[str]]:
        """
        批量下载图片并转为base64 data URL格式

        Args:
            image_urls: 图片URL列表

        Returns:
            (image_contents, []) - 图片content字典列表和空列表(保持接口兼容)
            格式: {"type": "image_url", "image_url": {"url": "data:image/png;base64,..."}}
        """
        if not image_urls:
            return [], []

        print(f"      📥 准备加载 {len(image_urls)} 张图片(Base64方式)...")

        # 并发下载所有图片
        tasks = [ImageUploader._load_single_image(url, idx) for idx, url in enumerate(image_urls)]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        # 分离成功和失败的结果
        image_contents = []

        for idx, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"      ⚠️  图片{idx}加载失败: {str(result)[:50]}")
            elif result is not None:
                image_contents.append(result)

        print(f"      ✅ 成功加载 {len(image_contents)}/{len(image_urls)} 张图片")
        return image_contents, []  # 返回空列表作为temp_paths,因为不需要清理

    @staticmethod
    async def _load_single_image(image_url: str, idx: int) -> Optional[Dict]:
        """
        下载单张图片并转为base64 data URL格式

        Args:
            image_url: 图片URL
            idx: 图片索引(用于日志)

        Returns:
            图片content字典: {"type": "image_url", "image_url": {"url": "data:image/png;base64,..."}}
        """
        try:
            # 下载图片到内存
            loop = asyncio.get_event_loop()
            response = await loop.run_in_executor(
                None,
                lambda: requests.get(image_url, timeout=30)
            )
            response.raise_for_status()

            # 转换为PIL Image对象
            image = Image.open(io.BytesIO(response.content))

            # 转换为RGB模式(Gemini推荐)
            if image.mode != 'RGB':
                image = image.convert('RGB')

            # 转换为PNG格式的BytesIO
            buffer = io.BytesIO()
            image.save(buffer, format="PNG")
            image_bytes = buffer.getvalue()

            # Base64编码
            base64_encoded = base64.b64encode(image_bytes).decode('utf-8')
            data_url = f"data:image/png;base64,{base64_encoded}"

            file_size_kb = len(image_bytes) / 1024
            print(f"      ✓ 图片{idx}加载成功 ({file_size_kb:.1f}KB, {image.size[0]}x{image.size[1]})")

            # 返回格式与demo一致
            return {
                "type": "image_url",
                "image_url": {"url": data_url}
            }

        except Exception as e:
            print(f"      ✗ 图片{idx}加载失败: {str(e)[:60]}")
            return None



class PromptAdapter:
    """Prompt适配器 - 根据媒体类型调整Prompt"""

    @staticmethod
    def adapt_prompt(prompt_template: str, post: Any, **kwargs) -> str:
        """
        适配Prompt

        Args:
            prompt_template: Prompt模板
            post: Post对象
            **kwargs: 其他参数 (如original_query)

        Returns:
            适配后的Prompt
        """
        # 准备替换参数
        params = {
            "title": post.title or "",
            "body_text": post.body_text or "",
        }

        # 媒体描述
        if post.type == "video":
            params["num_images"] = "1个视频"
        else:
            num_images = len(post.images) if post.images else 0
            params["num_images"] = f"{num_images}张"

        # 添加其他参数
        params.update(kwargs)

        return prompt_template.format(**params)


# ============================================================================
# 缓存函数 (复用V3逻辑)
# ============================================================================

def _get_cache_key(note_id: str) -> str:
    """生成缓存key"""
    return f"{note_id}_v4.0.json"


def _load_from_cache(note_id: str) -> Optional[tuple]:
    """从缓存加载评估结果"""
    if not ENABLE_CACHE:
        return None

    cache_file = os.path.join(CACHE_DIR, _get_cache_key(note_id))

    if not os.path.exists(cache_file):
        return None

    try:
        with open(cache_file, 'r', encoding='utf-8') as f:
            data = json.load(f)

        # 重建评估对象
        knowledge_eval = None
        if data.get("knowledge_eval"):
            knowledge_eval = KnowledgeEvaluation(**data["knowledge_eval"])

        content_eval = None
        if data.get("content_eval"):
            content_eval = ContentKnowledgeEvaluation(**data["content_eval"])

        purpose_eval = None
        if data.get("purpose_eval"):
            purpose_eval = PurposeEvaluation(**data["purpose_eval"])

        category_eval = None
        if data.get("category_eval"):
            category_eval = CategoryEvaluation(**data["category_eval"])

        final_score = data.get("final_score")
        match_level = data.get("match_level")

        return (knowledge_eval, content_eval, purpose_eval, category_eval, final_score, match_level)

    except Exception as e:
        print(f"      ⚠️  缓存读取失败: {note_id} - {str(e)[:50]}")
        return None


def _save_to_cache(note_id: str, eval_results: tuple):
    """保存评估结果到缓存"""
    if not ENABLE_CACHE:
        return

    knowledge_eval, content_eval, purpose_eval, category_eval, final_score, match_level = eval_results

    os.makedirs(CACHE_DIR, exist_ok=True)

    cache_data = {
        "knowledge_eval": knowledge_eval.model_dump() if knowledge_eval else None,
        "content_eval": content_eval.model_dump() if content_eval else None,
        "purpose_eval": purpose_eval.model_dump() if purpose_eval else None,
        "category_eval": category_eval.model_dump() if category_eval else None,
        "final_score": final_score,
        "match_level": match_level,
        "cache_time": datetime.now().isoformat(),
        "evaluator_version": "v4.0"
    }

    cache_file = os.path.join(CACHE_DIR, _get_cache_key(note_id))

    try:
        with open(cache_file, 'w', encoding='utf-8') as f:
            json.dump(cache_data, f, ensure_ascii=False, indent=2)
    except Exception as e:
        print(f"      ⚠️  缓存保存失败: {note_id} - {str(e)[:50]}")


# ============================================================================
# LangGraph 节点函数
# ============================================================================

async def knowledge_node(state: EvaluationState) -> EvaluationState:
    """
    Node 1: 知识判断 (Prompt1)
    """
    post = state["post"]
    semaphore = state.get("semaphore")

    print(f"      📝 Step 1/4: 判断是知识...")

    try:
        # 准备媒体文件
        media_files = []
        if post.type == "video" and state.get("video_file"):
            media_files = [state["video_file"]]
            print(f"      📹 准备视频文件: {state.get('video_uri', 'N/A')}")
        elif post.images:
            # 图文帖子 - 上传图片到Gemini
            image_urls = post.images[:MAX_IMAGES_PER_POST]
            print(f"      📸 准备上传 {len(image_urls)} 张图片 (总共{len(post.images)}张)")

            uploaded_files, temp_paths = await ImageUploader.upload_images(image_urls)
            media_files = uploaded_files

            # 保存临时路径到state中
            if not state.get("temp_image_paths"):
                state["temp_image_paths"] = []
            state["temp_image_paths"].extend(temp_paths)

            # ✅ 缓存图片数据，避免后续节点重复下载
            state["cached_media_files"] = media_files
        else:
            print(f"      ⚠️  帖子无图片/视频")

        # 准备System和User Prompt
        user_prompt = PromptAdapter.adapt_prompt(USER_TEMPLATE1_IS_KNOWLEDGE, post)
        system_prompt = SYSTEM_PROMPT1_IS_KNOWLEDGE

        # 调用Gemini (使用新格式)
        client = GeminiClient()

        if semaphore:
            async with semaphore:
                data = await client.generate_content(
                    system_prompt=system_prompt,
                    user_prompt=user_prompt,
                    media_files=media_files
                )
        else:
            data = await client.generate_content(
                system_prompt=system_prompt,
                user_prompt=user_prompt,
                media_files=media_files
            )

        # 调试:打印返回的数据结构
        print(f"      🐛 DEBUG - API返回数据: {json.dumps(data, ensure_ascii=False, indent=2)[:500]}")
        print(f"      🐛 DEBUG - data keys: {list(data.keys())}")

        # 解析结果
        knowledge_eval = KnowledgeEvaluation(
            is_knowledge=data.get("is_knowledge", False),
            quick_exclude=data.get("quick_exclude", {}),
            title_layer=data.get("title_layer", {}),
            image_layer=data.get("image_layer", {}),
            text_layer=data.get("text_layer", {}),
            judgment_logic=data.get("judgment_logic", ""),
            core_evidence=data.get("core_evidence", []),
            issues=data.get("issues", []),
            conclusion=data.get("conclusion", "")
        )

        state["knowledge_eval"] = knowledge_eval

        # 判断是否继续
        if not knowledge_eval.is_knowledge:
            print(f"      ⊗ 非知识内容,停止后续评估")
            state["should_continue"] = False
        else:
            print(f"      ✅ Step 1: 是知识内容")
            state["should_continue"] = True

    except Exception as e:
        print(f"      ❌ Prompt1评估失败: {str(e)[:100]}")
        state["error"] = str(e)
        state["should_continue"] = False

    return state


async def content_knowledge_node(state: EvaluationState) -> EvaluationState:
    """
    Node 2: 内容知识判断 (Prompt2)
    """
    post = state["post"]
    semaphore = state.get("semaphore")

    print(f"      📝 Step 2/4: 判断是否是内容知识...")

    try:
        # 准备媒体文件
        media_files = []
        if post.type == "video" and state.get("video_file"):
            media_files = [state["video_file"]]
            print(f"      📹 准备视频文件")
        elif post.images:
            # ✅ 优先使用缓存的图片，避免重复下载
            if state.get("cached_media_files"):
                media_files = state["cached_media_files"]
                print(f"      ♻️  使用缓存图片 ({len(media_files)}张)")
            else:
                # 缓存不存在才下载
                image_urls = post.images[:MAX_IMAGES_PER_POST]
                print(f"      📸 准备上传 {len(image_urls)} 张图片 (用于内容知识评估)")

                uploaded_files, temp_paths = await ImageUploader.upload_images(image_urls)
                media_files = uploaded_files

                # 保存临时路径到state中
                if not state.get("temp_image_paths"):
                    state["temp_image_paths"] = []
                state["temp_image_paths"].extend(temp_paths)
        else:
            print(f"      ⚠️  无媒体文件")

        # 准备System和User Prompt
        user_prompt = PromptAdapter.adapt_prompt(USER_TEMPLATE2_CONTENT_KNOWLEDGE, post)
        system_prompt = SYSTEM_PROMPT2_CONTENT_KNOWLEDGE

        # 调用Gemini (使用新格式)
        client = GeminiClient()

        if semaphore:
            async with semaphore:
                data = await client.generate_content(
                    system_prompt=system_prompt,
                    user_prompt=user_prompt,
                    media_files=media_files
                )
        else:
            data = await client.generate_content(
                system_prompt=system_prompt,
                user_prompt=user_prompt,
                media_files=media_files
            )

        # 解析结果
        final_score = data.get("final_score", 0)
        is_content_knowledge = final_score >= 55

        content_eval = ContentKnowledgeEvaluation(
            is_content_knowledge=is_content_knowledge,
            final_score=final_score,
            level=data.get("level", ""),
            quick_exclude=data.get("quick_exclude", {}),
            dimension_scores=data.get("dimension_scores", {}),
            core_evidence=data.get("core_evidence", []),
            issues=data.get("issues", []),
            summary=data.get("summary", "")
        )

        state["content_eval"] = content_eval

        # 判断是否继续
        if not is_content_knowledge:
            print(f"      ⊗ 非内容知识,停止后续评估 (得分: {final_score})")
            state["should_continue"] = False
        else:
            print(f"      ✅ Step 2: 是内容知识 (得分: {final_score})")
            state["should_continue"] = True

    except Exception as e:
        print(f"      ❌ Prompt2评估失败: {str(e)[:100]}")
        state["error"] = str(e)
        state["should_continue"] = False

    return state


async def parallel_match_node(state: EvaluationState) -> EvaluationState:
    """
    Node 3: 并行目的性和品类匹配 (Prompt3 & Prompt4)
    """
    post = state["post"]
    original_query = state["original_query"]
    semaphore = state.get("semaphore")

    print(f"      📝 Step 3&4/4: 并行执行目的性和品类匹配...")

    try:
        # 准备媒体文件
        media_files = []
        if post.type == "video" and state.get("video_file"):
            media_files = [state["video_file"]]
            print(f"      📹 准备视频文件")
        elif post.images:
            # ✅ 优先使用缓存的图片，避免重复下载
            if state.get("cached_media_files"):
                media_files = state["cached_media_files"]
                print(f"      ♻️  使用缓存图片 ({len(media_files)}张)")
            else:
                # 缓存不存在才下载
                image_urls = post.images[:MAX_IMAGES_PER_POST]
                print(f"      📸 准备上传 {len(image_urls)} 张图片 (用于目的性和品类评估)")

                uploaded_files, temp_paths = await ImageUploader.upload_images(image_urls)
                media_files = uploaded_files

                # 保存临时路径到state中
                if not state.get("temp_image_paths"):
                    state["temp_image_paths"] = []
                state["temp_image_paths"].extend(temp_paths)
        else:
            print(f"      ⚠️  无媒体文件")

        client = GeminiClient()

        # 并行执行Prompt3和Prompt4
        async def eval_purpose():
            user_prompt = PromptAdapter.adapt_prompt(
                USER_TEMPLATE3_PURPOSE_MATCH, post, original_query=original_query
            )
            system_prompt = SYSTEM_PROMPT3_PURPOSE_MATCH
            
            if semaphore:
                async with semaphore:
                    return await client.generate_content(
                        system_prompt=system_prompt,
                        user_prompt=user_prompt,
                        media_files=media_files
                    )
            else:
                return await client.generate_content(
                    system_prompt=system_prompt,
                    user_prompt=user_prompt,
                    media_files=media_files
                )

        async def eval_category():
            user_prompt = PromptAdapter.adapt_prompt(
                USER_TEMPLATE4_CATEGORY_MATCH, post, original_query=original_query
            )
            system_prompt = SYSTEM_PROMPT4_CATEGORY_MATCH
            
            if semaphore:
                async with semaphore:
                    return await client.generate_content(
                        system_prompt=system_prompt,
                        user_prompt=user_prompt,
                        media_files=media_files
                    )
            else:
                return await client.generate_content(
                    system_prompt=system_prompt,
                    user_prompt=user_prompt,
                    media_files=media_files
                )

        purpose_data, category_data = await asyncio.gather(eval_purpose(), eval_category())

        # 🔍 调试日志 - 查看API返回的实际结构
        print(f"\n      🐛 DEBUG - purpose_data keys: {list(purpose_data.keys())}")
        print(f"      🐛 DEBUG - purpose_data 内容: {purpose_data}")
        print(f"\n      🐛 DEBUG - category_data keys: {list(category_data.keys())}")
        print(f"      🐛 DEBUG - category_data 内容: {category_data}\n")

        # 解析Prompt3结果（直接使用英文字段名）
        purpose_eval = PurposeEvaluation(
            purpose_score=purpose_data.get("purpose_score", 0),
            core_motivation=purpose_data.get("core_motivation", ""),
            image_value=purpose_data.get("image_value", ""),
            title_intention=purpose_data.get("title_intention", ""),
            text_content=purpose_data.get("text_content", ""),
            match_level=purpose_data.get("match_level", ""),
            core_basis=purpose_data.get("core_basis", "")
        )

        # 解析Prompt4结果（直接使用英文字段名）
        category_eval = CategoryEvaluation(
            category_score=category_data.get("category_score", 0),
            original_category_analysis=category_data.get("original_category_analysis", {}),
            actual_category=category_data.get("actual_category", {}),
            match_level=category_data.get("match_level", ""),
            category_match_analysis=category_data.get("category_match_analysis", {}),
            core_basis=category_data.get("core_basis", "")
        )

        state["purpose_eval"] = purpose_eval
        state["category_eval"] = category_eval
        state["should_continue"] = True

        print(f"      ✅ Step 3: 目的性得分 = {purpose_eval.purpose_score}")
        print(f"      ✅ Step 4: 品类得分 = {category_eval.category_score}")

    except Exception as e:
        print(f"      ❌ Prompt3或4评估失败: {str(e)[:100]}")
        state["error"] = str(e)
        state["should_continue"] = False

    return state


async def score_node(state: EvaluationState) -> EvaluationState:
    """
    Node 4: 计算综合得分
    """
    print(f"      📊 Step 5/5: 计算综合得分...")

    try:
        purpose_eval = state["purpose_eval"]
        category_eval = state["category_eval"]

        if not purpose_eval or not category_eval:
            raise Exception("缺少目的性或品类评估结果")

        # 计算综合得分: 目的性50% + 品类50%
        final_score = round(
            purpose_eval.purpose_score * 0.5 + category_eval.category_score * 0.5,
            2
        )

        # 判定匹配等级
        if final_score >= 85:
            match_level = "高度匹配"
        elif final_score >= 70:
            match_level = "基本匹配"
        elif final_score >= 50:
            match_level = "部分匹配"
        elif final_score >= 30:
            match_level = "弱匹配"
        else:
            match_level = "不匹配"

        state["final_score"] = final_score
        state["match_level"] = match_level

        print(f"      ✅ 综合得分: {final_score} ({match_level})")

    except Exception as e:
        print(f"      ❌ 综合评分失败: {str(e)[:100]}")
        state["error"] = str(e)

    return state


# ============================================================================
# LangGraph 图定义
# ============================================================================

def create_evaluation_graph() -> StateGraph:
    """创建评估流程图"""

    # 定义条件判断
    def should_continue_to_content(state: EvaluationState) -> str:
        """判断是否继续到内容知识评估"""
        if not state.get("should_continue", False):
            return END
        return "content_knowledge_node"

    def should_continue_to_match(state: EvaluationState) -> str:
        """判断是否继续到匹配评估"""
        if not state.get("should_continue", False):
            return END
        return "parallel_match_node"

    def should_continue_to_score(state: EvaluationState) -> str:
        """判断是否继续到评分"""
        if not state.get("should_continue", False):
            return END
        return "score_node"

    # 创建StateGraph
    workflow = StateGraph(EvaluationState)

    # 添加节点
    workflow.add_node("knowledge_node", knowledge_node)
    workflow.add_node("content_knowledge_node", content_knowledge_node)
    workflow.add_node("parallel_match_node", parallel_match_node)
    workflow.add_node("score_node", score_node)

    # 设置入口点
    workflow.set_entry_point("knowledge_node")

    # 添加条件边
    workflow.add_conditional_edges(
        "knowledge_node",
        should_continue_to_content,
        {
            "content_knowledge_node": "content_knowledge_node",
            END: END
        }
    )

    workflow.add_conditional_edges(
        "content_knowledge_node",
        should_continue_to_match,
        {
            "parallel_match_node": "parallel_match_node",
            END: END
        }
    )

    workflow.add_conditional_edges(
        "parallel_match_node",
        should_continue_to_score,
        {
            "score_node": "score_node",
            END: END
        }
    )

    # score_node结束后直接到END
    workflow.add_edge("score_node", END)

    return workflow.compile()


# ============================================================================
# 主评估函数
# ============================================================================

async def evaluate_post_v4(
    post,
    original_query: str,
    semaphore: Optional[asyncio.Semaphore] = None
) -> tuple:
    """
    V4评估主函数 (LangGraph版本)

    Args:
        post: Post对象
        original_query: 原始搜索query
        semaphore: 并发控制信号量

    Returns:
        (knowledge_eval, content_eval, purpose_eval, category_eval, final_score, match_level)
    """
    # 检查缓存
    if ENABLE_CACHE:
        cached_result = _load_from_cache(post.note_id)
        if cached_result is not None:
            print(f"      ♻️  使用缓存结果: {post.note_id}")
            return cached_result

    print(f"      🔍 开始V4评估 (LangGraph): {post.note_id}")

    # 初始化状态
    initial_state: EvaluationState = {
        "post": post,
        "original_query": original_query,
        "video_file": None,
        "video_uri": None,
        "temp_video_path": None,
        "temp_image_paths": None,
        "knowledge_eval": None,
        "content_eval": None,
        "purpose_eval": None,
        "category_eval": None,
        "final_score": None,
        "match_level": None,
        "should_continue": True,
        "error": None,
        "semaphore": semaphore
    }

    # 处理视频
    if post.type == "video" and post.images and len(post.images) > 0:
        video_url = post.images[0]  # 视频URL通常在images[0]
        video_file, video_uri, temp_path = await VideoUploader.upload_video(video_url)
        initial_state["video_file"] = video_file
        initial_state["video_uri"] = video_uri
        initial_state["temp_video_path"] = temp_path

        if not video_file:
            print(f"      ❌ 视频上传失败,停止评估")
            return (None, None, None, None, None, None)

    try:
        # 创建并运行图
        graph = create_evaluation_graph()
        final_state = await graph.ainvoke(initial_state)

        # 提取结果
        knowledge_eval = final_state.get("knowledge_eval")
        content_eval = final_state.get("content_eval")
        purpose_eval = final_state.get("purpose_eval")
        category_eval = final_state.get("category_eval")
        final_score = final_state.get("final_score")
        match_level = final_state.get("match_level")

        # 保存到缓存
        if ENABLE_CACHE and knowledge_eval:
            _save_to_cache(
                post.note_id,
                (knowledge_eval, content_eval, purpose_eval, category_eval, final_score, match_level)
            )

        return (knowledge_eval, content_eval, purpose_eval, category_eval, final_score, match_level)

    finally:
        # 清理临时视频文件
        if initial_state.get("temp_video_path"):
            try:
                os.remove(initial_state["temp_video_path"])
                print(f"      🗑️  清理临时视频文件")
            except:
                pass

        # 清理临时图片文件
        temp_image_paths = final_state.get("temp_image_paths") if 'final_state' in locals() else initial_state.get("temp_image_paths")
        if temp_image_paths:
            cleaned_count = 0
            for temp_path in temp_image_paths:
                try:
                    os.remove(temp_path)
                    cleaned_count += 1
                except:
                    pass
            if cleaned_count > 0:
                print(f"      🗑️  清理 {cleaned_count}/{len(temp_image_paths)} 个临时图片文件")


def apply_evaluation_v4_to_post(
    post,
    knowledge_eval: Optional[KnowledgeEvaluation],
    content_eval: Optional[ContentKnowledgeEvaluation],
    purpose_eval: Optional[PurposeEvaluation],
    category_eval: Optional[CategoryEvaluation],
    final_score: Optional[float],
    match_level: Optional[str]
):
    """
    将V4评估结果应用到Post对象

    Args:
        post: Post对象
        knowledge_eval: Prompt1结果
        content_eval: Prompt2结果
        purpose_eval: Prompt3结果
        category_eval: Prompt4结果
        final_score: 综合得分
        match_level: 匹配等级
    """
    # Prompt1: 判断是知识
    if knowledge_eval:
        post.is_knowledge = knowledge_eval.is_knowledge
        post.knowledge_evaluation = {
            "quick_exclude": knowledge_eval.quick_exclude,
            "title_layer": knowledge_eval.title_layer,
            "image_layer": knowledge_eval.image_layer,
            "text_layer": knowledge_eval.text_layer,
            "judgment_logic": knowledge_eval.judgment_logic,
            "core_evidence": knowledge_eval.core_evidence,
            "issues": knowledge_eval.issues,
            "conclusion": knowledge_eval.conclusion
        }

    # Prompt2: 判断是否是内容知识
    if content_eval:
        post.is_content_knowledge = content_eval.is_content_knowledge
        post.knowledge_score = float(content_eval.final_score)
        post.content_knowledge_evaluation = {
            "is_content_knowledge": content_eval.is_content_knowledge,
            "final_score": content_eval.final_score,
            "level": content_eval.level,
            "quick_exclude": content_eval.quick_exclude,
            "dimension_scores": content_eval.dimension_scores,
            "core_evidence": content_eval.core_evidence,
            "issues": content_eval.issues,
            "summary": content_eval.summary
        }

    # Prompt3: 目的性匹配
    if purpose_eval:
        post.purpose_score = purpose_eval.purpose_score
        post.purpose_evaluation = {
            "purpose_score": purpose_eval.purpose_score,
            "core_motivation": purpose_eval.core_motivation,
            "image_value": purpose_eval.image_value,
            "title_intention": purpose_eval.title_intention,
            "text_content": purpose_eval.text_content,
            "match_level": purpose_eval.match_level,
            "core_basis": purpose_eval.core_basis
        }

    # Prompt4: 品类匹配
    if category_eval:
        post.category_score = category_eval.category_score
        post.category_evaluation = {
            "category_score": category_eval.category_score,
            "original_category_analysis": category_eval.original_category_analysis,
            "actual_category": category_eval.actual_category,
            "match_level": category_eval.match_level,
            "category_match_analysis": category_eval.category_match_analysis,
            "core_basis": category_eval.core_basis
        }

    # 综合得分
    if final_score is not None and match_level is not None:
        post.final_score = final_score
        post.match_level = match_level

    # 设置评估时间和版本
    post.evaluation_time = datetime.now().isoformat()
    post.evaluator_version = "v4.0_langgraph"


async def batch_evaluate_posts_v4(
    posts: list,
    original_query: str,
    max_concurrent: int = MAX_CONCURRENT_EVALUATIONS
) -> int:
    """
    批量评估多个帖子 (V4版本)

    Args:
        posts: Post对象列表
        original_query: 原始搜索query
        max_concurrent: 最大并发数

    Returns:
        成功评估的帖子数量
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    print(f"\n📊 开始批量评估 {len(posts)} 个帖子 (LangGraph + Gemini,并发限制: {max_concurrent})...")

    tasks = [evaluate_post_v4(post, original_query, semaphore) for post in posts]
    results = await asyncio.gather(*tasks)

    success_count = 0
    for i, result in enumerate(results):
        knowledge_eval, content_eval, purpose_eval, category_eval, final_score, match_level = result

        if knowledge_eval:
            apply_evaluation_v4_to_post(
                posts[i],
                knowledge_eval,
                content_eval,
                purpose_eval,
                category_eval,
                final_score,
                match_level
            )
            success_count += 1

    print(f"✅ 批量评估完成: {success_count}/{len(posts)} 帖子已评估")

    return success_count