""" 帖子评估模块 V3 - 4步串行+并行评估系统 改进: 1. Prompt1: 判断是知识 (is_knowledge) 2. Prompt2: 判断是否是内容知识 (is_content_knowledge) 3. Prompt3 & Prompt4: 并行执行 - 目的性(50%) + 品类(50%) 4. 代码计算综合得分: final_score = purpose × 0.5 + category × 0.5 5. 完全替代V2评估结果 """ import asyncio import json import os from datetime import datetime from typing import Optional from pydantic import BaseModel, Field import requests MODEL_NAME = "google/gemini-2.5-flash" MAX_IMAGES_PER_POST = 10 MAX_CONCURRENT_EVALUATIONS = 5 API_TIMEOUT = 120 # 缓存配置 ENABLE_CACHE = True # 是否启用评估结果缓存 CACHE_DIR = ".evaluation_cache" # 缓存目录 # ============================================================================ # 数据模型 # ============================================================================ 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="核心依据") # ============================================================================ # Prompt 定义 # ============================================================================ PROMPT1_IS_KNOWLEDGE = """# 知识判定系统 v1.0 ## 角色定义 你是一个多模态内容评估专家，专门判断社交媒体帖子是否提供了"知识"。 ## 知识定义 **知识**是指经过验证的、具有可靠性和真实性的信息，能够被理解、学习、传播和应用。 ### 知识类型 - ✅ **事实性知识**: 客观事实、数据、现象描述 - ✅ **原理性知识**: 规律、原理、理论、因果关系 - ✅ **方法性知识**: 技能、流程、步骤、操作方法 - ✅ **经验性知识**: 总结提炼的经验、教训、最佳实践 - ✅ **概念性知识**: 定义、分类、框架、体系 - ✅ **应用性知识**: 解决方案、工具使用、实践指南 ### 非知识类型（严格排除） - ❌ **纯观点/立场**: 未经验证的个人观点、偏好表达 - ❌ **情感表达**: 纯粹的情绪抒发、心情分享 - ❌ **单纯展示**: 作品展示、生活记录、打卡（无知识提炼） - ❌ **娱乐内容**: 段子、搞笑、八卦（无信息价值） - ❌ **纯营销/广告**: 单纯的产品推销 - ❌ **虚假/未验证信息**: 谣言、伪科学、未经证实的说法 --- ## 输入信息 - **标题**: {title} - **正文**: {body_text} - **图片**: {num_images}张 --- ## 判断流程 ### 第一步: 快速排除（任一为"是"则判定为非知识） 1. 是否为纯情感表达/生活记录/打卡？ 2. 是否为单纯的作品/产品展示（无知识提炼）？ 3. 是否为娱乐搞笑/八卦/纯营销内容？ 4. 是否包含虚假信息或伪科学？ 5. 是否完全没有新信息（纯重复常识）？ **排除判定**: □ 是（判定为非知识） / □ 否（继续评估） --- ### 第二步: 分层知识判断 ## 🏷️ 标题层 **判断：标题是否指向知识？** - ✅ 明确传达知识主题（如何/为什么/什么是/XX方法/XX原理） - ⚠️ 描述性标题，但暗示有知识内容 - ❌ 展示型（我的XX/今天XX）或情感型标题 **结果**: □ 有知识指向 / □ 无知识指向 --- ## 🖼️ 图片层（信息主要承载层） **判断1: 图片知识呈现方式** - ✅ 包含信息图表（数据、流程图、对比图、结构图） - ✅ 有知识性标注（解释、说明、步骤、原理） - ✅ 多图形成知识体系（步骤序列、案例对比） - ❌ 纯作品展示、美图、氛围图 **判断2: 图片教育价值** - ✅ 图片能教会他人方法、技能或原理 - ✅ 提供可学习的步骤或解决方案 - ❌ 图片无教学意义 **判断3: 图片结构化程度** - ✅ 有清晰的逻辑组织（序号、分层、框架） - ✅ 有步骤、要点的结构化呈现 - ❌ 碎片化、无逻辑的展示 **判断4: 图片实用性** - ✅ 提供可直接应用的方法或工具 - ✅ 能帮助解决实际问题 - ❌ 纯观赏性，无实际应用价值 **判断5: 图片信息密度** - ✅ 包含≥3个独立知识点 - ⚠️ 包含1-2个知识点 - ❌ 无明确知识点 **图片层综合评估**: □ 图片传递了知识 / □ 图片无知识价值 --- ## 📝 正文层（辅助判断） **判断1: 信息增量** - ✅ 提供了明确的新信息、新认知或新方法 - ❌ 无新信息，只是个人记录或情感表达 **判断2: 可验证性** - ✅ 基于事实、数据、可验证的经验 - ❌ 纯主观观点或感受，无依据 **判断3: 知识类型归属** - ✅ 能归入至少一种知识类型（事实/原理/方法/经验/概念/应用） - ❌ 无法归类为任何知识类型 **正文层综合评估**: □ 正文提供了知识支撑 / □ 正文无知识价值 --- ### 第三步: 综合判定 #### 判定规则 **直接判定为"非知识"（任一成立）**: - 未通过快速排除 - 图片层 = 无知识价值 且 正文层 = 无知识支撑 - 正文判断3（知识类型）= 无法归类 **判定为"是知识"（需同时满足）**: 1. 通过快速排除 2. 图片层 = 传递了知识 或 正文层 = 提供了知识支撑 3. 正文判断1（信息增量）= 有新信息 4. 正文判断3（知识类型）= 可归类 **特别说明**： - 社交媒体帖子以图片为主要信息载体，图片层权重最高 - 标题为辅助判断，正文为补充验证 --- ## 输出格式 请严格按照以下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": "说明是否满足判定规则，关键依据是什么", "core_evidence": [ "最强支持证据1", "最强支持证据2" ], "issues": [ "不足或疑虑（如无则为空数组）" ], "conclusion": "用1-2句话说明判定结果和核心理由" }} ## 判断原则 1. **图片优先**: 社交媒体以图片为主要信息载体，图片层是核心判断依据，标题和正文信息辅助 2. **严格性**: 宁可误判为"非知识"，也不放过无价值内容 3. **证据性**: 基于明确的视觉和文本证据判断 4. **价值导向**: 优先判断内容对读者是否有实际学习价值 """ PROMPT2_IS_CONTENT_KNOWLEDGE = """## 角色定义 你是一个多模态内容评估专家，专门判断社交媒体帖子是否属于"内容知识"类别。 ## 前置条件 该帖子已通过知识判定，确认提供了知识。现在需要进一步判断是否属于"内容知识"。 --- ## 内容知识的底层定义 **内容知识**：关于社交媒体内容创作与制作的通识性、原理性知识，帮助创作者策划、生产、优化和传播优质内容。 ### 核心特征 1. **领域特定性**：专注于社交媒体内容本身的创作与制作 2. **通识性**：跨平台、跨领域适用的内容创作原理和方法 3. **原理性**：不仅是操作步骤，更包含背后的逻辑和原理 4. **可迁移性**：方法可应用于不同类型的社交媒体内容创作 ### 内容知识的完整范畴 #### 1️⃣ 内容策划层 - **选题方法**：如何找选题、选题原理、热点捕捉、用户需求分析 - **内容定位**：账号定位、人设打造、差异化策略 - **结构设计**：内容框架、故事结构、信息组织方式 - **创意方法**：创意思路、脑暴方法、灵感来源 #### 2️⃣ 内容制作层 - **文案创作**：标题技巧、正文写作、文案公式、钩子设计、情绪调动 - **视觉呈现**：封面设计原理、排版方法、配色技巧（用于内容呈现的） - **视频制作**：脚本结构、拍摄技巧、镜头语言、剪辑节奏、转场方法 - **多模态组合**：图文配合、视频+文案组合、内容形式选择 #### 3️⃣ 内容优化层 - **开头/钩子**：前3秒设计、开头公式、吸引注意力的方法 - **节奏控制**：信息密度、节奏把控、留白技巧 - **完播/完读**：提升完播率/完读率的方法和原理 - **互动设计**：评论引导、互动话术、用户参与设计 #### 4️⃣ 内容方法论 - **创作体系**：完整的内容创作流程和体系 - **底层原理**：为什么这样做有效的原理解释 - **通用框架**：可复用的内容创作框架和模板 - **案例提炼**：从多个案例中总结的通用规律 --- ### 内容知识 vs 非内容知识 **✅ 属于内容知识的例子**： - "小红书爆款标题的5个公式"（文案创作） - "短视频前3秒如何抓住用户"（开头设计） - "如何策划一个涨粉选题"（内容策划） - "视频节奏控制的底层逻辑"（内容优化） - "图文笔记的排版原理"（视觉呈现） - "从10个爆款视频总结的脚本结构"（方法论提炼） **❌ 不属于内容知识的例子**： - "摄影构图的三分法则"（专业摄影技能，除非用于讲解社交媒体内容拍摄） - "PS修图教程"（设计软件技能，除非用于讲解封面/配图制作） - "我的探店vlog"（单个作品展示，无创作方法） - "今天涨粉100个好开心"（个人记录，无方法论） - "健康饮食的10个建议"（其他领域知识） - "这套配色真好看"（纯元素展示，无创作方法） **⚠️ 边界情况判断**： - **专业技能类**：如果是为社交媒体内容创作服务的，属于内容知识（如"拍摄短视频的灯光布置"）；如果是纯技能教学，不属于（如"专业摄影的灯光理论"） - **工具使用类**：如果是为内容制作服务的，属于内容知识（如"剪映做转场的3种方法"）；如果是纯软件教程，不属于（如"AE粒子特效教程"） - **案例分析类**：如果从案例中提炼了内容创作方法，属于内容知识；如果只是案例展示，不属于 --- ### 判断核心准则 **问自己三个问题**： 1. **这个知识是关于"如何创作社交媒体内容"的吗？** - 是 → 可能是内容知识 - 否 → 不是内容知识 2. **这个方法/原理是通识性的吗？能跨内容类型/平台应用吗？** - 是 → 符合内容知识特征 - 否 → 可能只是单点技巧 3. **看完后，创作者能用它来改进自己的内容创作吗？** - 能 → 是内容知识 - 不能 → 不是内容知识 --- ## 输入信息 - **标题**: {title} - **正文**: {body_text} - **图片**: {num_images}张 --- ## 判断流程 ### 第一步: 领域快速筛查 **判断：内容是否属于社交媒体内容创作/制作领域？** 核心判断标准： - 属于： 讲的是如何创作/制作社交媒体内容（选题、文案、拍摄、剪辑、运营等） - 属于：讲的是内容创作的原理、方法、技巧 - 属于：讲的是平台运营、爆款方法、涨粉策略 - 不属于：讲的是其他专业领域技能（摄影、设计、编程等），与内容创作无关 - 不属于：讲的是其他行业知识（财经、健康、科普等） **判定**: □ 属于内容创作领域（继续） / □ 不属于（判定为非内容知识） --- ### 第二步: 快速排除判断（任一为"是"则判定为非内容知识） 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分**: ⭐ 非内容知识 - 完全不符合 --- ## 输出格式 ### 判定结果 - **是否属于内容知识**: [是/否] - **最终得分**: [X]/100分 - **判定等级**: [⭐等级] --- ### 分层评分详情 **🖼️ 图片层（70分）** | 维度 | 得分 | 评分依据 | |------|------|----------| | 内容创作方法呈现 | __/20 | [简述：图片展示了哪些内容创作方法] | | 内容知识体系化 | __/15 | [简述：多图逻辑关系] | | 教学性标注 | __/15 | [简述：标注元素情况] | | 方法通识性 | __/10 | [简述：通识性与可迁移性评估] | | 原理性深度 | __/10 | [简述：是否讲解原理] | | **小计** | **__/70** | | **📝 正文层（20分）** | 维度 | 得分 | 评分依据 | |------|------|----------| | 方法/步骤描述 | __/10 | [简述] | | 知识总结提炼 | __/10 | [简述] | | **小计** | **__/20** | | **🏷️ 标题层（10分）** | 维度 | 得分 | 评分依据 | |------|------|----------| | 标题内容创作指向性 | __/10 | [简述] | | **小计** | **__/10** | | --- ### 核心证据 **支持判定的最强证据**（2-3条）: 1. [从图片/正文/标题中提取的关键证据] 2. [...] **不足之处**（如有）: 1. [存在的问题] --- ### 总结陈述 [用5-6句话说明判定结果和核心理由，明确指出为何属于/不属于内容知识] --- ## JSON输出格式 请严格按照以下JSON格式输出： {{ "is_content_knowledge": true/false, "final_score": 85, "level": "⭐⭐⭐⭐⭐ 优质内容知识", "quick_exclude": {{ "result": "是/否", "reason": "原因" }}, "dimension_scores": {{ "image_layer": {{ "creation_method": {{"score": 20, "reason": "简述"}}, "knowledge_system": {{"score": 15, "reason": "简述"}}, "teaching_annotation": {{"score": 15, "reason": "简述"}}, "method_reusability": {{"score": 10, "reason": "简述"}}, "principle_case": {{"score": 10, "reason": "简述"}}, "subtotal": 70 }}, "text_layer": {{ "method_description": {{"score": 10, "reason": "简述"}}, "knowledge_summary": {{"score": 10, "reason": "简述"}}, "subtotal": 20 }}, "title_layer": {{ "content_direction": {{"score": 10, "reason": "简述"}}, "subtotal": 10 }} }}, "core_evidence": [ "证据1", "证据2" ], "issues": [ "问题1（如无则为空数组）" ], "summary": "总结陈述（5-6句话）" }} --- ## 判断原则 1. **图片主导原则**: 图片占70%权重，是核心判断依据；标题和正文为辅助 2. **创作领域限定**: 必须属于创作/制作/设计领域，其他领域知识不属于内容知识 3. **方法优先原则**: 重点评估是否提供了可操作的创作方法，而非纯作品展示 4. **通用性要求**: 优先考虑方法的可复用性和可迁移性 5. **严格性原则**: 宁可误判为"非内容知识"，也不放过纯展示型内容 6. **证据性原则**: 评分需基于明确的视觉和文本证据，可量化衡量 """ PROMPT3_PURPOSE_MATCH = """# Prompt 1: 多模态内容目的动机匹配评估 ## 角色定义 你是一位专业的多模态内容评估专家，擅长分析社交媒体UGC平台帖子的**目的动机匹配度**，能够精准判断帖子是否满足用户的核心意图。 --- ## 任务说明 你将收到一个**原始搜索需求**和一条**多模态帖子**（包含图片、标题、正文） 请**仅评估目的动机维度**的匹配度，输出0-100分的量化得分。 --- ## 输入格式 **原始搜索需求：** {original_query} **多模态帖子内容：** - **图片：** {num_images}张 - **标题：** {title} - **正文：** {body_text} --- ## 评估维度：目的动机匹配 ### 核心评估逻辑 **目的动机 = 用户想做什么 = 核心动词/意图** 常见动机类型： - **获取型**：寻找、下载、收藏、获取 - **学习型**：教程、学习、了解、掌握 - **决策型**：推荐、对比、评测、选择 - **创作型**：拍摄、制作、设计、生成 - **分享型**：晒单、记录、分享、展示 --- ## 评估流程 ### 第一步：识别原始需求的核心动机 - 提取**核心动词**（如果是纯名词短语，识别隐含意图） - 判断用户的**最终目的**是什么 ### 第二步：分析帖子提供的价值（重点看图片） **图片分析（权重70%）：** - 图片展示的是什么类型的内容？ - 图片是否直接解答了需求的目的？ - 图片的信息完整度和实用性如何？ **标题分析（权重15%）：** - 标题是否明确了内容的目的？ **正文分析（权重15%）：** - 正文是否提供了实质性的解答内容？ ### 第三步：判断目的匹配度 - 帖子是否**实质性地满足**了需求的动机？ - 内容是否**实用、完整、可执行**？ --- ## 评分标准（0-100分） ### 高度匹配区间 **90-100分：完全满足动机，内容实用完整** - 图片直接展示解决方案/教程步骤/对比结果 - 内容完整、清晰、可直接使用 - 例：需求"如何拍摄夜景" vs 图片展示完整的夜景拍摄参数设置和效果对比 **75-89分：基本满足动机，信息较全面** - 图片提供了核心解答内容 - 信息相对完整但深度略有不足 - 例：需求"推荐旅行路线" vs 图片展示了路线图但缺少详细说明 **60-74分：部分满足动机，有参考价值** - 图片提供了相关内容但不够直接 - 需要结合文字才能理解完整意图 ### 中度相关区间 **40-59分：弱相关，核心目的未充分满足** - 图片内容与动机有关联但不是直接解答 - 实用性较低 - 例：需求"如何拍摄" vs 图片只展示成品照片，无教程内容 ### 不相关/负向区间 **20-39分：微弱关联，基本未解答** - 图片仅有外围相关性 - 对满足需求帮助极小 **1-19分：几乎无关** - 图片与需求动机关联极弱 **0分：完全不相关** - 图片与需求动机无任何关联 **负分不使用**（目的动机维度不设负分） --- ## 输出格式（JSON） ```json {{ "目的动机评估": {{ "目的动机得分": 0-100的整数, "原始需求核心动机": "识别出的用户意图（一句话）", "图片提供的价值": "图片展示了什么，如何满足动机", "标题体现的意图": "标题说明了什么", "正文补充的内容": "正文是否有实质解答", "匹配度等级": "完全匹配/高度匹配/基本匹配/弱匹配/不匹配", "核心依据": "为什么给这个分数（100字以内）" }} }} ``` --- ## 评估原则 1. **图片优先**：图片权重70%，是判断的主要依据 2. **实用导向**：不看表面相关，看实际解答程度 3. **严格标准**：宁可低估，避免虚高 4. **客观量化**：基于可观察的内容特征打分 --- ## 特别注意 - 本评估**只关注目的动机维度**，不考虑品类是否匹配 - 输出的分数必须是**0-100的整数** - 不要自行计算综合分数，只输出目的动机分数 - 评分依据要具体、可验证 """ PROMPT4_CATEGORY_MATCH = """# Prompt 2: 多模态内容品类匹配评估 ## 角色定义 你是一位专业的多模态内容评估专家，擅长分析社交媒体UGC平台帖子的**品类匹配度** 能够精准判断帖子的内容主体是否与用户需求一致。 --- ## 任务说明 你将收到一个**原始搜索需求**和一条**多模态帖子**（包含图片、标题、正文） 请**仅评估品类维度**的匹配度，输出0-100分的量化得分。 --- ## 输入格式 **原始搜索需求：** {original_query} **多模态帖子内容：** - **图片：** {num_images}张 - **标题：** {title} - **正文：** {body_text} --- ## 评估维度：品类匹配 ## 评估维度 本评估系统围绕 **品类维度** 进行： # 维度独立性警告 【严格约束】本评估**只评估品类维度**,，必须遵守以下规则： 1. **只看名词和限定词**：评估时只考虑主体、限定词的匹配度 2. **完全忽略动词**：动作意图、目的等动机信息对本维度评分无影响 3. **只看词条表面，禁止联想推演 4. **通用概念 ≠ 特定概念 ### 核心评估逻辑 **品类 = 核心内容主体（实体名词）+ 场景/地域限定** ### 品类识别规则 #### 第一步：剥离动作词，识别核心主体 **必须剥离的动作词（属于目的动机，不是品类）：** - 如何、怎么、制作、拍摄、寻找、推荐、学习、了解等 **示例：** - "如何制作猫咪表情包" → 品类主体是**猫咪**，不是"表情包制作" - "川西风光摄影教程" → 品类主体是**川西风光**，不是"摄影教程" - "推荐日本旅行景点" → 品类主体是**日本旅行/景点**，不是"推荐" #### 第二步：识别核心主体类别 **核心主体（实体名词）：** - **生物类**：猫咪、狗狗、植物、人物（具体指儿童、女孩、老人等） - **地理类**：川西、成都、日本、景点名称 - **物品类**：美食、服装、电子产品、家具 - **场景类**：风光、建筑、室内、户外 - **活动类**：旅行、运动、工作、学习场景 **关键原则：品类主体必须是具体的内容对象，不是动作或形式** #### 第三步：识别场景/地域等限定词（可选） **场景/地域限定：** - **地域限定**：川西、成都、日本、欧洲 - **时间限定**：秋季、夏天、2024 - **场景限定**：户外、室内、职场、家居 **注意：** - "表情包"、"梗图"、"照片"、"视频"等是**内容形式/载体**，不是品类主体 - "教程"、"攻略"、"指南"等是**内容类型**，属于目的动机，不是品类 --- ## 评估流程 ### 第一步：提取原始需求的品类信息 1. **剥离所有动作词和内容形式词** 2. **识别核心主体名词**（生物、地理、物品、场景等） 3. **识别场景/地域限定**（如果有） **示例分析：** - "如何制作猫咪表情包梗图" - 剥离动作：如何、制作 - 剥离形式：表情包、梗图 - **核心品类主体：猫咪** - 场景限定：无 ### 第二步：从帖子中提取品类信息（重点看图片） **图片识别（权重70%）：** - 图片的**核心主体**是什么？（是猫、是人、是风景、是物品？） - 图片的**场景/地域特征**是什么？ **标题提取（权重15%）：** - 标题明确的品类主体名词 **正文提取（权重15%）：** - 正文描述的品类主体 ### 第三步：对比品类匹配度 **核心判断：主体是否一致？** - 猫咪 ≠ 女孩 → 品类完全不同 → 0-10分 - 猫咪 = 猫咪 → 品类一致 → 进一步看场景限定 - 川西风光 ≠ 日本风光 → 地域不同 → 30-50分 - 川西风光 = 四川风光 → 地域相近 → 70-85分 --- ## 评分标准（0-100分） ### 高度匹配区间 **90-100分：核心主体完全一致 + 场景/地域等限定词完全匹配** - 图片主体与需求完全一致 - 关键限定词全部匹配（场景、地域、时间等） - 例：需求"川西秋季风光" vs 图片展示川西秋季风景 **75-89分：核心主体完全一致 + 场景/地域等限定词部分匹配** - 图片主体一致 - 存在1-2个限定词缺失但不影响核心匹配 - 例：需求"川西秋季风光" vs 图片展示川西风光（缺秋季） **60-74分：核心主体匹配，限定词大量缺失** - 图片主体在同一大类 - 场景/地域等限定词大部分缺失 - 例：需求"川西秋季风光" vs 图片展示风光 ### 中度相关区间 **40-59分：核心主体同大类但具体不同** - 图片主体相同但上下文不同 - 限定词严重缺失或不匹配 - 例：需求"川西风光摄影" vs 图片展示风光照但无地域特征 ### 不相关/负向区间 **20-39分：核心主体相关但类别差异明显** - 图片主体是通用概念，需求是特定概念 - 仅有抽象类别相似 - 例：需求"川西旅行攻略" vs 图片展示普通旅行场景 **1-19分：核心主体几乎不相关** - 图片主体与需求差异明显 **0分：核心主体完全不同** - 图片主体类别完全不同 - 例：需求"风光摄影" vs 图片展示美食 **关键原则：品类主体不同 = 品类不匹配 = 0分或极低分** --- ## 输出格式（JSON） ```json {{ "品类评估": {{ "原始需求品类分析": {{ "完整需求": "用户的原始搜索词", "剥离动作词": "识别并剥离的动作词", "剥离形式词": "识别并剥离的内容形式词", "核心主体": "提取的核心品类主体", "场景地域限定": ["限定词1", "限定词2"] }}, "帖子实际品类": {{ "图片主体": "图片展示的核心主体（权重70%）", "图片场景特征": "图片的场景/地域特征", "标题主体": "标题提及的主体", "正文主体": "正文描述的主体" }}, "品类匹配分析": {{ "主体对比": "需求主体 vs 帖子主体", "主体是否一致": "一致/同大类不同/完全不同", "场景限定匹配情况": "哪些匹配/哪些缺失" }}, "品类匹配得分": 0-100的整数, "匹配度等级": "完全匹配/高度匹配/基本匹配/弱匹配/不匹配", "核心依据": "为什么给这个分数（必须说明主体是否一致）" }} }} ``` --- ## 评估原则 1. **图片优先**：图片权重70%，是判断的主要依据 2. **表面匹配**：只看实际展示的内容，禁止推测联想 3. **通用≠特定**：通用概念不等于特定概念，需明确区分 4. **严格标准**：宁可低估，避免虚高 5. **客观量化**：基于可观察的视觉特征和文字信息打分 --- ## 特别注意 - 本评估**只关注品类维度**，不考虑目的是否匹配 - 严格标准一致性：对所有用例使用相同的评估标准，避免评分飘移 - 输出的分数必须是**0-100的整数** - 不要自行计算综合分数，只输出品类分数 - 禁止因为"可能相关"就给分，必须有明确视觉证据，不得用可能相关，你的评估 --- """ # ============================================================================ # 辅助函数 # ============================================================================ def _get_cache_key(note_id: str) -> str: """ 生成缓存key Args: note_id: 帖子ID Returns: 缓存文件名（不含目录） """ return f"{note_id}_v3.0.json" def _load_from_cache(note_id: str) -> Optional[tuple]: """ 从缓存加载评估结果 Args: note_id: 帖子ID Returns: 缓存的评估结果元组 (knowledge_eval, content_eval, purpose_eval, category_eval, final_score, match_level) 如果缓存不存在或读取失败，返回None """ 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): """ 保存评估结果到缓存 Args: note_id: 帖子ID eval_results: 评估结果元组 (knowledge_eval, content_eval, purpose_eval, category_eval, final_score, match_level) """ 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) # 转换为可序列化的dict 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": "v3.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]}") def _clean_json_response(content_text: str) -> str: """清理API返回的JSON内容""" content_text = content_text.strip() if content_text.startswith("```json"): content_text = content_text[7:] elif content_text.startswith("```"): content_text = content_text[3:] if content_text.endswith("```"): content_text = content_text[:-3] return content_text.strip() async def _call_openrouter_api( prompt_text: str, image_urls: list[str], semaphore: Optional[asyncio.Semaphore] = None ) -> dict: """ 调用OpenRouter API的通用函数 Args: prompt_text: Prompt文本 image_urls: 图片URL列表 semaphore: 并发控制信号量 Returns: 解析后的JSON响应 """ api_key = os.getenv("OPENROUTER_API_KEY") if not api_key: raise ValueError("OPENROUTER_API_KEY environment variable not set") content = [{"type": "text", "text": prompt_text}] for url in image_urls: content.append({"type": "image_url", "image_url": {"url": url}}) payload = { "model": MODEL_NAME, "messages": [{"role": "user", "content": content}], "response_format": {"type": "json_object"} } headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } async def _make_request(): loop = asyncio.get_event_loop() response = await loop.run_in_executor( None, lambda: requests.post( "https://openrouter.ai/api/v1/chat/completions", headers=headers, json=payload, timeout=API_TIMEOUT ) ) if response.status_code != 200: raise Exception(f"API error: {response.status_code} - {response.text[:200]}") result = response.json() content_text = result["choices"][0]["message"]["content"] content_text = _clean_json_response(content_text) return json.loads(content_text) async def _execute_with_retry(): """执行API请求，失败时自动重试最多2次""" MAX_RETRIES = 2 for attempt in range(MAX_RETRIES + 1): try: return await _make_request() except Exception as e: if attempt < MAX_RETRIES: wait_time = 2 * (attempt + 1) # 2秒, 4秒 print(f" ⚠️ API调用失败，{wait_time}秒后重试 (第{attempt + 1}/{MAX_RETRIES}次重试) - {str(e)[:50]}") await asyncio.sleep(wait_time) else: # 最后一次尝试也失败，抛出异常 raise if semaphore: async with semaphore: return await _execute_with_retry() else: return await _execute_with_retry() # ============================================================================ # 核心评估函数 # ============================================================================ async def evaluate_is_knowledge( post, semaphore: Optional[asyncio.Semaphore] = None ) -> Optional[KnowledgeEvaluation]: """ Prompt1: 判断是知识 Args: post: Post对象 semaphore: 并发控制信号量 Returns: KnowledgeEvaluation 或 None（失败时） """ if post.type == "video": return None image_urls = post.images[:MAX_IMAGES_PER_POST] if post.images else [] try: prompt_text = PROMPT1_IS_KNOWLEDGE.format( title=post.title, body_text=post.body_text or "", num_images=len(image_urls) ) data = await _call_openrouter_api(prompt_text, image_urls, semaphore) return 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", "") ) except Exception as e: print(f" ❌ Prompt1评估失败: {post.note_id} - {str(e)[:100]}") return None async def evaluate_is_content_knowledge( post, semaphore: Optional[asyncio.Semaphore] = None ) -> Optional[ContentKnowledgeEvaluation]: """ Prompt2: 判断是否是内容知识 Args: post: Post对象 semaphore: 并发控制信号量 Returns: ContentKnowledgeEvaluation 或 None（失败时） """ if post.type == "video": return None image_urls = post.images[:MAX_IMAGES_PER_POST] if post.images else [] try: prompt_text = PROMPT2_IS_CONTENT_KNOWLEDGE.format( title=post.title, body_text=post.body_text or "", num_images=len(image_urls) ) data = await _call_openrouter_api(prompt_text, image_urls, semaphore) # 判定是否是内容知识：得分 >= 55 分 final_score = data.get("final_score", 0) is_content_knowledge = final_score >= 55 return 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", "") ) except Exception as e: print(f" ❌ Prompt2评估失败: {post.note_id} - {str(e)[:100]}") return None async def evaluate_purpose_match( post, original_query: str, semaphore: Optional[asyncio.Semaphore] = None ) -> Optional[PurposeEvaluation]: """ Prompt3: 目的性匹配评估 Args: post: Post对象 original_query: 原始搜索query semaphore: 并发控制信号量 Returns: PurposeEvaluation 或 None（失败时） """ if post.type == "video": return None image_urls = post.images[:MAX_IMAGES_PER_POST] if post.images else [] try: prompt_text = PROMPT3_PURPOSE_MATCH.format( original_query=original_query, title=post.title, body_text=post.body_text or "", num_images=len(image_urls) ) data = await _call_openrouter_api(prompt_text, image_urls, semaphore) # Prompt3的输出在"目的动机评估"键下 purpose_data = data.get("目的动机评估", {}) return PurposeEvaluation( purpose_score=purpose_data.get("目的动机得分", 0), core_motivation=purpose_data.get("原始需求核心动机", ""), image_value=purpose_data.get("图片提供的价值", ""), title_intention=purpose_data.get("标题体现的意图", ""), text_content=purpose_data.get("正文补充的内容", ""), match_level=purpose_data.get("匹配度等级", ""), core_basis=purpose_data.get("核心依据", "") ) except Exception as e: print(f" ❌ Prompt3评估失败: {post.note_id} - {str(e)[:100]}") return None async def evaluate_category_match( post, original_query: str, semaphore: Optional[asyncio.Semaphore] = None ) -> Optional[CategoryEvaluation]: """ Prompt4: 品类匹配评估 Args: post: Post对象 original_query: 原始搜索query semaphore: 并发控制信号量 Returns: CategoryEvaluation 或 None（失败时） """ if post.type == "video": return None image_urls = post.images[:MAX_IMAGES_PER_POST] if post.images else [] try: prompt_text = PROMPT4_CATEGORY_MATCH.format( original_query=original_query, title=post.title, body_text=post.body_text or "", num_images=len(image_urls) ) data = await _call_openrouter_api(prompt_text, image_urls, semaphore) # Prompt4的输出在"品类评估"键下 category_data = data.get("品类评估", {}) return CategoryEvaluation( category_score=category_data.get("品类匹配得分", 0), original_category_analysis=category_data.get("原始需求品类分析", {}), actual_category=category_data.get("帖子实际品类", {}), match_level=category_data.get("匹配度等级", ""), category_match_analysis=category_data.get("品类匹配分析", {}), core_basis=category_data.get("核心依据", "") ) except Exception as e: print(f" ❌ Prompt4评估失败: {post.note_id} - {str(e)[:100]}") return None def calculate_final_score(purpose_score: int, category_score: int) -> tuple[float, str]: """ 计算综合得分和匹配等级 Args: purpose_score: 目的性得分 (0-100整数) category_score: 品类得分 (0-100整数) Returns: (final_score, match_level) - final_score: 保留2位小数 - match_level: 匹配等级字符串 """ # 计算综合得分: 目的性70% + 品类30% final = round(purpose_score * 0.5 + category_score * 0.5, 2) # 判定匹配等级 if final >= 85: level = "高度匹配" elif final >= 70: level = "基本匹配" elif final >= 50: level = "部分匹配" elif final >= 30: level = "弱匹配" else: level = "不匹配" return final, level async def evaluate_post_v3( post, original_query: str, semaphore: Optional[asyncio.Semaphore] = None ) -> tuple: """ V3评估主函数（4步流程） 流程: 1. Prompt1: 判断是知识 → 如果不是知识，停止 2. Prompt2: 判断是否是内容知识 → 如果不是内容知识，停止 3. Prompt3 & Prompt4: 并行执行目的性和品类匹配 4. 计算综合得分 Returns: (knowledge_eval, content_eval, purpose_eval, category_eval, final_score, match_level) 任一步骤失败，后续结果为None """ if post.type == "video": print(f" ⊗ 跳过视频帖子: {post.note_id}") return (None, None, None, None, None, None) # 检查缓存 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" 🔍 开始V3评估: {post.note_id}") # Step 1: 判断是知识 print(f" 📝 Step 1/4: 判断是知识...") knowledge_eval = await evaluate_is_knowledge(post, semaphore) if not knowledge_eval: print(f" ❌ Step 1失败，停止评估") return (None, None, None, None, None, None) if not knowledge_eval.is_knowledge: print(f" ⊗ 非知识内容，停止后续评估") return (knowledge_eval, None, None, None, None, None) print(f" ✅ Step 1: 是知识内容") # Step 2: 判断是否是内容知识 print(f" 📝 Step 2/4: 判断是否是内容知识...") content_eval = await evaluate_is_content_knowledge(post, semaphore) if not content_eval: print(f" ❌ Step 2失败，停止评估") return (knowledge_eval, None, None, None, None, None) if not content_eval.is_content_knowledge: print(f" ⊗ 非内容知识，停止后续评估 (得分: {content_eval.final_score})") return (knowledge_eval, content_eval, None, None, None, None) print(f" ✅ Step 2: 是内容知识 (得分: {content_eval.final_score})") # Step 3 & 4: 并行执行目的性和品类匹配 print(f" 📝 Step 3&4/4: 并行执行目的性和品类匹配...") purpose_task = evaluate_purpose_match(post, original_query, semaphore) category_task = evaluate_category_match(post, original_query, semaphore) purpose_eval, category_eval = await asyncio.gather(purpose_task, category_task) if not purpose_eval or not category_eval: print(f" ❌ Step 3或4失败") return (knowledge_eval, content_eval, purpose_eval, category_eval, None, None) print(f" ✅ Step 3: 目的性得分 = {purpose_eval.purpose_score}") print(f" ✅ Step 4: 品类得分 = {category_eval.category_score}") # Step 5: 计算综合得分 final_score, match_level = calculate_final_score( purpose_eval.purpose_score, category_eval.category_score ) print(f" ✅ 综合得分: {final_score} ({match_level})") # 保存到缓存 if ENABLE_CACHE: _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) def apply_evaluation_v3_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] ): """ 将V3评估结果应用到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 = "v3.0" async def batch_evaluate_posts_v3( posts: list, original_query: str, max_concurrent: int = MAX_CONCURRENT_EVALUATIONS ) -> int: """ 批量评估多个帖子（V3版本） Args: posts: Post对象列表 original_query: 原始搜索query max_concurrent: 最大并发数 Returns: 成功评估的帖子数量 """ semaphore = asyncio.Semaphore(max_concurrent) print(f"\n📊 开始批量评估 {len(posts)} 个帖子（并发限制: {max_concurrent}）...") tasks = [evaluate_post_v3(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 # 只要有Prompt1结果就算部分成功 if knowledge_eval: apply_evaluation_v3_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