#!/usr/bin/env python # -*- coding: utf-8 -*- """ 脚本理解Agent 功能: 脚本理解与分析（创作者视角） - step1: Section划分 - 分析已有帖子的分段结构，理解创作者是如何组织内容的（树状结构输出） - step2: 元素列表生成 - 从已有帖子内容中提取元素列表（树状结构输出） """ from typing import List from src.components.agents.base import BaseLLMAgent from src.utils.logger import get_logger from src.utils.llm_invoker import LLMInvoker logger = get_logger(__name__) class ScriptUnderstandingAgent(BaseLLMAgent): """脚本理解Agent - 分析已有帖子的脚本结构 """ def __init__( self, name: str = "script_understanding_agent", description: str = "脚本理解Agent - 分析已有帖子的脚本结构（创作者视角）", model_provider: str = "google_genai", temperature: float = 0.1, max_tokens: int = 40960 ): """初始化脚本理解Agent """ system_prompt = self._build_system_prompt() super().__init__( name=name, description=description, model_provider=model_provider, system_prompt=system_prompt, temperature=temperature, max_tokens=max_tokens ) def _build_system_prompt(self) -> str: """构建系统提示词""" return """你是脚本结构分析专家，擅长从创作者视角理解内容的分段结构。 # 核心能力 1. 结构识别：分析已有帖子内容，识别创作者的分段逻辑 2. 树状输出：通过children字段表示层级关系，不使用parent_id # 工作原则 - step1（Section划分）：识别创作者如何分段组织内容（树状结构） - step2（元素提取）：提取具体元素列表（树状结构）""" def step1(self, text_data: dict, images: List[str], topic_description: dict, content_weight: dict = None) -> dict: """第一步：Section划分 分析已有帖子的内容，识别创作者是如何分段组织内容的 """ if not self.is_initialized: self.initialize() logger.info("【Step1】Section划分 - 分析帖子的分段结构") # 提取权重信息 weight_info = "" strategy_guidance = "" if content_weight: image_weight = content_weight.get("图片权重", 5) text_weight = content_weight.get("文字权重", 5) primary_source = content_weight.get("主要信息源", "both") content_relationship = content_weight.get("图文关系", "相关") weight_info = f""" # 图文权重信息 - 图片权重: {image_weight}/10 - 文字权重: {text_weight}/10 - 主要信息源: {primary_source} - 图文关系: {content_relationship} """ # 根据权重和关系确定策略指导 weight_diff = image_weight - text_weight # 确定策略类型和验证要求 if weight_diff >= 3: strategy_type = "以图为主" strategy_guidance = """ ## 图文策略: 以图为主 **核心原则**: 用图划分段落,图与图之间有顺序,文可以乱序对应到段落 **Section切分要点**: - 以图片序列为核心主线进行划分 - **图片编号必须保持严格连续性** - 文字可以灵活对应到图片 - 关注图片之间的关系、图片内容的语义分段、叙事逻辑 """ validation_guidance = """ **验证内容**： 1. ✅ **图片连续性验证**（核心验证） - 检查每个Section内的图片编号是否严格连续 - 方法：逐个检查图片编号，确保无跳跃 2. ✅ **图片语义关系验证** - 同一Section内的图片是否有语义关联 - 图片之间的过渡是否自然合理 3. ❌ **不验证文字顺序** - 文字可以灵活对应到图片 **禁止行为**： - ❌ 跨段落跳跃式引用图片 - ❌ 打乱图片的原始顺序 **验证流程**： - Step1: 提取每个Section的图片编号列表 - Step2: 检查图片编号是否连续 - Step3: 如发现跳跃，重新调整Section划分 """ elif weight_diff <= -3: strategy_type = "以文为主" strategy_guidance = """ ## 图文策略: 以文为主 **核心原则**: 用文划分段落,文与文之间有顺序,图可以乱序对应到段落 **Section切分要点**: - 以文字段落为核心主线进行划分 - **文字必须保持原始顺序** - 图片可以灵活对应到文字 - 关注文字的话题转换、逻辑推进、情绪变化 """ validation_guidance = """ **验证内容**： 1. ✅ **文字顺序验证**（核心验证） - 检查每个Section内的文字是否保持原始顺序 - 段落之间的文字流转是否自然连贯 2. ✅ **文字逻辑性验证** - 同一Section内的文字是否有逻辑关联 - 话题转换是否合理 3. ❌ **不验证图片顺序** - 图片可以灵活对应到文字 **禁止行为**： - ❌ 打乱文字的原始顺序 - ❌ 跨段落错乱文字位置 **验证流程**： - Step1: 提取每个Section的文字内容 - Step2: 按原文顺序检查文字是否连贯 - Step3: 如发现乱序，重新调整Section划分 """ else: # 策略3/4: 图文同权 if content_relationship == "相关": strategy_type = "图文同权-图文相关" strategy_guidance = """ ## 图文策略: 图文同权-图文相关 **核心原则**: 根据消费者吸引力,判断图/文为主,另一个为辅 **Section切分要点**: - 综合图文信息进行Section划分 - 根据图片排版质量、文字表达清晰度等判断主导维度 - 保持图文的对应关系,相互解释 - 关注图文相互补充的部分、哪个维度的表达更完整 """ validation_guidance = """ **前置步骤**： - 先判断主导维度（图为主 or 文为主） - 明确记录判断依据 **验证内容**： 1. ✅ **主导维度顺序验证**（核心验证） - 如果图为主：验证图片序列的严格连续性 - 如果文为主：验证文字段落的逻辑性 2. ✅ **图文对应关系验证** - 检查图文之间的对应关系是否合理 - 图文是否相互解释、相互补充 3. ✅ **辅助维度合理性验证** - 辅助维度是否支撑主导维度 **禁止行为**： - ❌ 主导维度出现顺序跳跃或错乱 - ❌ 图文对应关系混乱 **验证流程**： - Step1: 判断并记录主导维度 - Step2: 针对主导维度执行严格顺序验证 - Step3: 验证图文对应关系 - Step4: 如发现问题，重新调整Section划分 """ else: strategy_type = "图文同权-图文不相关" strategy_guidance = """ ## 图文策略: 图文同权-图文不相关 **核心原则**: 根据创作者目的,判断只关注图还是只关注文 **Section切分要点**: - 明确选择关注图或关注文,不能同时关注 - 分析创作者的主要表达意图 - 选定维度后,完全基于该维度进行Section划分 - 关注创作者想重点传达什么、解构的目标是什么 """ validation_guidance = """ **前置步骤**： - 先判断选择关注的维度（图 or 文） - 明确记录选择依据 **验证内容**： 1. ✅ **选定维度顺序验证**（核心验证） - 如果关注图：验证图片序列的严格连续性 - 如果关注文：验证文字段落的逻辑性 2. ✅ **单一维度一致性验证** - 确认所有Section都基于选定的单一维度划分 3. ❌ **不验证未选择的维度** - 未选择的维度可以完全忽略 **禁止行为**： - ❌ 选定维度出现顺序跳跃或错乱 - ❌ 同时关注图和文进行Section划分 **验证流程**： - Step1: 判断并记录关注的维度 - Step2: 针对选定维度执行严格顺序验证 - Step3: 确认未使用另一维度进行划分 - Step4: 如发现问题，重新调整Section划分 """ logger.info(f"使用图文策略: {strategy_type} - 图片:{image_weight}, 文字:{text_weight}, 关系:{content_relationship}") else: strategy_type = "未指定" validation_guidance = "" logger.info("未提供图文权重信息") # 构建帖子内容 post_content_parts = [] if text_data.get("title"): post_content_parts.append(f"标题: {text_data['title']}") if text_data.get("body"): post_content_parts.append(f"正文: {text_data['body']}") post_content = "\n".join(post_content_parts) if post_content_parts else "无文本信息" # 构建选题描述文本 topic_parts = [] if topic_description.get("主题"): topic_parts.append(f"主题: {topic_description['主题']}") if topic_description.get("描述"): topic_parts.append(f"描述: {topic_description['描述']}") topic_text = "\n".join(topic_parts) if topic_parts else "无选题描述" # 构建prompt prompt = f"""{weight_info} {strategy_guidance} # 帖子内容 {post_content} # 选题描述 {topic_text} # 任务 从**创作者视角**分析这个已有帖子是如何组织内容的。 ## Section切分流程 **第一步：应用图文策略** 根据上述图文策略指导,确定Section切分的主导维度和具体方法。 **第二步：识别主题显著变化位置** 扫描全部内容,识别**主题发生显著变化**的位置: - **判断标准**: * 语义跃迁: 讨论对象发生根本性改变 * 逻辑转换: 从"是什么"转向"为什么"或"怎么办" * 功能变化: 从"问题陈述"转向"解决方案" - **划分原则**: * 避免过度细分(每个小变化都成为顶层段落) * 避免过度粗放(将所有内容合并为1个顶层段落) * 以"主题板块"而非"内容单元"为划分粒度 **第三步：初步划分** - 按照策略要点确定核心驱动 - 基于主题显著变化位置进行划分 - 支持主Section和子Section的层级结构 **第四步：顺序验证与反思** {validation_guidance} ## 层级要求 **段落必须至少保留2层结构**： 1. **第1层（抽象层）**：从具象中聚合出的共性维度 2. **第2层（具象层）**：具体的内容细节 **层级关系说明**： - 抽象层是对多个具象内容的归纳和提炼 - 具象层是抽象层的具体展开 - 每个抽象层下必须有至少1个具象层子项 ## Section字段 - 描述: 段落描述（共性维度名称；具体内容概括） - 内容范围: **列表格式，包含具体内容** - 格式：["图X","正文-XXXX"] - 要求：必须包含具体的图片编号或文字原文片段 - 推理依据: 为什么这样划分 - 子项: 子Section列表（树状结构） # 输出（JSON）- 树状结构 {{ "内容品类": "内容品类", "段落列表": [ {{ "描述": "共性维度名称", "内容范围": ["图1", "图2", "正文——具体文字内容片段"], "推理依据": "为什么这样划分这个抽象层", "子项": [ {{ "描述": "具体内容概括", "内容范围": ["图1", "正文——具体文字内容片段"], "推理依据": "这个具象内容如何支撑上层抽象", "子项": [] }} ] }} ] }}""" # 构建多模态消息 message_content = [{"type": "text", "text": prompt}] # 添加图片 if images: for idx, image_url in enumerate(images, 1): message_content.append({"type": "text", "text": f"[图片{idx}]"}) message_content.append({ "type": "image_url", "image_url": {"url": image_url} }) messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": message_content} ] # 调用LLM result = LLMInvoker.safe_invoke( self, "Section划分", messages, fallback={"内容品类": "未知品类", "段落列表": []} ) # 只打印JSON结构 import json logger.info(f"Step1结果:\n{json.dumps(result, ensure_ascii=False, indent=2)}") return result def process(self, state: dict) -> dict: """处理state，执行完整的脚本理解流程（step1 + step2） 从state中提取所需数据，执行step1和step2，并返回结果 Args: state: 工作流状态，包含: - text: 文本数据 {title, body} - images: 图片URL列表 - topic_selection_understanding: 选题理解结果 - content_weight: 图文权重信息 Returns: dict: 包含script_understanding结果 """ logger.info("=== 开始脚本理解处理 ===") # 从state中提取数据 text_data = state.get("text", {}) images = state.get("images", []) # 提取选题描述（作为字典） topic_understanding = state.get("topic_selection_understanding", {}) topic_description = { "主题": topic_understanding.get("主题", ""), "描述": topic_understanding.get("描述", "") } # 提取图文权重 content_weight = state.get("content_weight", {}) # 执行 step1 - Section划分 logger.info("执行 Step1 - Section划分") sections_result = self.step1( text_data=text_data, images=images, topic_description=topic_description, content_weight=content_weight ) sections = sections_result.get("段落列表", []) # 执行 step2 - 元素列表生成 logger.info("执行 Step2 - 元素列表生成") elements_result = self.step2( text_data=text_data, images=images, topic_description=topic_description, content_weight=content_weight ) # 组装最终结果 script_understanding = { "内容品类": sections_result.get("内容品类", "未知品类"), "段落列表": sections, "元素列表": elements_result.get("元素列表", []), "图片列表": images # 添加原始图片URL列表 } # 只打印JSON结构 import json logger.info(f"脚本理解最终结果:\n{json.dumps(script_understanding, ensure_ascii=False, indent=2)}") return {"脚本理解": script_understanding} def step2_1_identify_candidates(self, text_data: dict, images: List[str], topic_description: dict, content_weight: dict = None) -> dict: """Step2.1: 候选元素识别 基于选题主题，识别与主题相关的候选元素（包括实质性和概念性元素） """ if not self.is_initialized: self.initialize() logger.info("=" * 80) logger.info("【Step2.1】候选元素识别 - 基于选题主题识别候选元素（实质性+概念性）") logger.info("=" * 80) # 构建权重信息 weight_info = self._build_weight_info(content_weight) # 构建内容 post_content = self._build_post_content(text_data) topic_text = self._build_topic_text(topic_description) # 构建prompt prompt = f"""{weight_info} # 帖子内容 {post_content} # 选题描述 {topic_text} # 任务 **基于选题主题作为关注锚点**，识别与主题相关或支撑主题的候选元素。 ## 选题的作用 选题主题是识别的**关注锚点**： - ✅ **直接相关**：与选题主题直接相关的核心元素 - ✅ **支撑主题**：对主题有支撑作用的辅助元素（即使不是核心，但能帮助理解主题） - ✅ **间接相关**：与选题主题间接相关的背景元素 - ❌ **完全无关**：与选题主题没有任何关联的元素 ## 识别原则 - **全面识别**：宁可多识别，不要漏掉与主题相关或支撑主题的元素 - **关系多样**：包括直接相关、支撑性、补充性、背景性等各种关系 - **宽松标准**：此阶段标准宽松，后续会进一步筛选 - **倾向保留**：不确定是否相关时，倾向于保留 ## 识别标准：实质性元素 + 概念性元素 ### 实质性元素（Substantial Elements） **定义**：看得见、摸得着的具体物体 - ✅ 必须是名词形式 - ✅ 必须是可观察的具体事物 - ✅ 必须是**原子的名词**，关注的是类/ID - ✅ 包括所有出现的物体，无论大小或重要性 - ❌ 不包括动作或状态 - ❌ 不包括情绪或氛围 - ❌ 不包括带修饰语的复合名词（修饰语应在后续分析中体现） ### 概念性元素（Conceptual Elements） **定义**：虽然不是客观实体，但是对理解帖子至关重要的抽象概念 - ✅ 必须是名词形式的概念 - ✅ 虽然没有直接提及，但图片或文字隐含了这个概念 - ✅ 出现频率高或重要性强的描述性概念 - ✅ 对理解主题有显著帮助的抽象维度 **概念性元素的判断标准**： - ✅ 这个概念是否帮助理解帖子的核心价值？ - ✅ 这个概念是否贯穿多个段落或图片？ - ✅ 这个概念是否隐含在视觉或文字表达中？ - ❌ 不是单纯的形容词或副词，必须能转化为名词概念 ## 识别要求 - **全面浏览**：不要遗漏任何出现的元素（实质性+概念性） - **原始记录**：此阶段不进行筛选，全部记录 - **原子名词**：使用最基础的名词形式，去除修饰语 - **显性+隐性**：既要识别显性出现的实体，也要识别隐性的概念 ## 元素类型标注 每个元素需要标注类型： - **实质性**：具体的客观物体 - **概念性**：抽象的概念维度 # 输出（JSON） {{ "候选元素列表": [ {{ "元素名称": "元素名称（原子名词）", "元素类型": "实质性/概念性", "出现位置": "该元素出现在哪些位置", "识别依据": "为什么识别这个元素（特别是概念性元素，需要说明隐含的依据）" }} ] }}""" # 构建多模态消息 message_content = [{"type": "text", "text": prompt}] if images: for idx, image_url in enumerate(images, 1): message_content.append({"type": "text", "text": f"[图片{idx}]"}) message_content.append({ "type": "image_url", "image_url": {"url": image_url} }) messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": message_content} ] # 调用LLM result = LLMInvoker.safe_invoke( self, "候选元素识别", messages, fallback={"候选元素列表": []} ) # 只打印JSON结构 import json logger.info(f"Step2.1结果:\n{json.dumps(result, ensure_ascii=False, indent=2)}") return result def step2_2_score_dimensions(self, candidates_result: dict, text_data: dict, images: List[str], topic_description: dict, content_weight: dict = None) -> dict: """Step2.2: 双维度评分（分视角分析） 对每个候选元素进行双维度评分（每维度0-10分） - 先分视角：判断概念性维度 vs 实质性维度哪个更重 - 优先处理权重更大的维度 - 维度1: 主题支撑性 - 评估元素对选题展开和目的达成的支撑程度 - 维度2: 多段落共性 - 评估元素的贯穿性和共性 """ if not self.is_initialized: self.initialize() logger.info("\n" + "=" * 80) logger.info("【Step2.2】双维度评分（分视角分析）") logger.info("=" * 80) candidates = candidates_result.get("候选元素列表", []) if not candidates: logger.warning("⚠️ 没有候选元素可供评分") return {"评分结果": []} # 统计概念性和实质性元素数量 conceptual_count = sum(1 for c in candidates if c.get("元素类型") == "概念性") substantial_count = sum(1 for c in candidates if c.get("元素类型") == "实质性") logger.info(f"输入: {len(candidates)} 个候选元素（概念性: {conceptual_count}, 实质性: {substantial_count}）") # 构建权重信息 weight_info = self._build_weight_info(content_weight) # 构建内容 post_content = self._build_post_content(text_data) topic_text = self._build_topic_text(topic_description) # 构建候选元素列表文本 candidates_text = "" for c in candidates: candidates_text += f"\n- {c.get('元素名称', 'N/A')} ({c.get('元素类型', 'N/A')})" candidates_text += f"\n 出现位置: {c.get('出现位置', 'N/A')}" if c.get('识别依据'): candidates_text += f"\n 识别依据: {c.get('识别依据', 'N/A')}" # 构建prompt prompt = f"""{weight_info} # 帖子内容 {post_content} # 选题描述 {topic_text} # 候选元素列表 {candidates_text} # 任务 对每个候选元素进行**双维度评分**（每维度0-10分）。 ## 分视角分析流程 **第一步：判断维度权重** 根据候选元素列表，判断概念性元素和实质性元素哪个更重： - 如果概念性元素数量多或对主题更核心 → 概念性维度权重更大 - 如果实质性元素数量多或对主题更核心 → 实质性维度权重更大 - 如果两者相当 → 两个维度权重相同 **第二步：优先处理权重更大的维度** - 先对权重更大的维度进行详细评分和分析 - 确保权重更大维度的元素得到充分评估 - 再处理另一个维度作为补充 **第三步：综合评分** - 对所有元素进行双维度评分 - 权重更大的维度应有更多高分元素 ## 两大评分维度 ### 1. 主题支撑性（0-10分） 评估元素对选题展开和目的达成的支撑程度： - **核心支撑（8-10分）**：元素是选题展开的核心支撑，直接承载创作目的 - **重要支撑（5-7分）**：元素为选题提供重要支撑，对目的达成有明显作用 - **辅助支撑（3-4分）**：元素为选题提供辅助性支撑 - **弱支撑（0-2分）**：元素与选题关联度低，对目的达成作用微弱 ### 2. 多段落共性（0-10分） 评估元素在多个段落中的贯穿性和共性： - **跨段落出现**：元素在多个段落/Section中出现 - **持续性存在**：元素在内容中的连续性呈现 - **代表性**：元素的典型性和普遍性，能否代表一类事物 **贯穿段落字段说明**： - 格式：列表格式，包含该元素出现的具体内容 - 要求：必须包含具体的图片编号或文字原文片段 ## 评分要求 - 客观评分：基于实际内容，不要主观臆测 - 严格打分：真正优秀的元素才能得高分 - 详细说明：每个维度都要给出评分依据 - 分视角优先：优先处理权重更大的维度 # 输出（JSON） {{ "维度判断": {{ "概念性元素重要性": "高/中/低", "实质性元素重要性": "高/中/低", "优先维度": "概念性/实质性/均衡", "判断依据": "为什么这个维度更重要" }}, "评分结果": [ {{ "元素名称": "元素名称", "元素类型": "概念性/实质性", "主题支撑性得分": 8, "主题支撑性依据": "评分依据说明（如何支撑选题展开和目的达成）", "多段落共性得分": 7, "多段落共性依据": "评分依据说明", "贯穿段落": ["图1", "图2", "正文——具体文字内容片段"] }} ] }}""" # 构建多模态消息 message_content = [{"type": "text", "text": prompt}] if images: for idx, image_url in enumerate(images, 1): message_content.append({"type": "text", "text": f"[图片{idx}]"}) message_content.append({ "type": "image_url", "image_url": {"url": image_url} }) messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": message_content} ] # 调用LLM result = LLMInvoker.safe_invoke( self, "双维度评分", messages, fallback={"评分结果": []} ) # 只打印JSON结构 import json logger.info(f"Step2.2结果:\n{json.dumps(result, ensure_ascii=False, indent=2)}") return result def step2_3_strict_filter(self, scored_result: dict, single_threshold: float = 6.0, weighted_threshold: float = 5.0) -> dict: """Step2.3: 严格筛选 根据评分结果进行严格筛选，满足以下条件之一即可通过： 1. 主题支撑性 > single_threshold 2. 多段落共性 > single_threshold 3. 加权得分 > weighted_threshold（加权：主题支撑性 * 0.5 + 多段落共性 * 0.5） Args: scored_result: 评分结果 single_threshold: 单维度筛选阈值，默认为6.0（降低阈值，返回更多支撑主题的元素） weighted_threshold: 加权得分筛选阈值，默认为5.0（整体合格即可） """ if not self.is_initialized: self.initialize() logger.info("\n" + "=" * 80) logger.info("【Step2.3】严格筛选") logger.info("=" * 80) logger.info("筛选规则（满足任一条件即通过）:") logger.info(f" 1. 主题支撑性 > {single_threshold} (单维度良好)") logger.info(f" 2. 多段落共性 > {single_threshold} (单维度良好)") logger.info(f" 3. 加权得分 > {weighted_threshold} (整体合格，权重：主题支撑性=0.5, 多段落共性=0.5)") logger.info(f" 注：阈值已从7降低到6，以返回更多支撑主题的元素") scored_elements = scored_result.get("评分结果", []) if not scored_elements: logger.warning("⚠️ 没有评分结果可供筛选") return {"筛选结果": []} logger.info(f"\n输入: {len(scored_elements)} 个评分元素") logger.info("\n筛选过程:") filtered_elements = [] for element in scored_elements: element_name = element.get("元素名称", "未知") # 提取分数 theme_score = element.get("主题支撑性得分", 0) multi_para_score = element.get("多段落共性得分", 0) # 计算加权得分（主题支撑性和多段落共性各占50%） weighted_score = theme_score * 0.5 + multi_para_score * 0.5 # 判断是否满足任一条件 conditions_met = [] if theme_score > single_threshold: conditions_met.append(f"主题支撑性={theme_score}") if multi_para_score > single_threshold: conditions_met.append(f"多段落共性={multi_para_score}") if weighted_score > weighted_threshold: conditions_met.append(f"加权得分={weighted_score:.1f}") # 如果满足任一条件，则通过筛选 if conditions_met: logger.info(f" ✓ {element_name}: 通过筛选 ({', '.join(conditions_met)})") filtered_elements.append(element) else: logger.info(f" ✗ {element_name}: 未通过筛选 (主题支撑性={theme_score}, 多段={multi_para_score}, 加权={weighted_score:.1f})") # 只打印JSON结构 import json logger.info(f"Step2.3结果:\n{json.dumps({'筛选结果': filtered_elements}, ensure_ascii=False, indent=2)}") return {"筛选结果": filtered_elements} def step2_4_deduplicate(self, filtered_result: dict, text_data: dict, images: List[str], topic_description: dict) -> dict: """Step2.4: 去重与整合 合并同类元素，确保元素之间正交独立，生成树状结构 - 非叶子节点：分类（抽象维度） - 叶子节点：具体元素 """ if not self.is_initialized: self.initialize() logger.info("\n" + "=" * 80) logger.info("【Step2.4】去重与整合") logger.info("=" * 80) filtered_elements = filtered_result.get("筛选结果", []) if not filtered_elements: logger.warning("⚠️ 没有筛选结果可供去重") return {"元素列表": []} logger.info(f"输入: {len(filtered_elements)} 个筛选后的元素") # 构建内容 post_content = self._build_post_content(text_data) topic_text = self._build_topic_text(topic_description) # 构建筛选元素文本 elements_text = "" for elem in filtered_elements: elements_text += f"\n- {elem.get('元素名称', 'N/A')}" elements_text += f"\n 主题支撑性: {elem.get('主题支撑性得分', 0)}, 多段落共性: {elem.get('多段落共性得分', 0)}" elements_text += f"\n 贯穿段落: {elem.get('贯穿段落', 'N/A')}" # 构建prompt prompt = f"""# 帖子内容 {post_content} # 选题描述 {topic_text} # 筛选后的元素列表 {elements_text} # 任务 对筛选后的元素进行**去重与整合**，生成最终的树状结构元素列表。 ## 核心原则 **树状结构：非叶子节点为分类，叶子节点为具体元素** - 非叶子节点：分类（抽象维度），可以包含子分类或具体元素 - 叶子节点：具体元素，是实际的元素 - 层级深度：根据实际内容灵活确定，不限定层数 - 分类节点通过"子项"字段包含下一层的分类或元素 ## 去重规则 1. **合并同类**：将具有相同上位概念的元素归类到同一分类下 2. **正交独立**：同级节点之间必须互不重叠、相互独立 ## 分类规则（使用元素本身 What 的维度） **核心原则：分类基于元素本身的 What 维度（本质属性）** - **What 维度**：回答"这是什么类型的东西" - **分类来源**：从元素本身的本质属性中抽象出分类，基于"它们是什么"来分类 - **分类层级**：分类可以嵌套（分类下可以有子分类） - **同级正交**：同一分类下的节点应该具有相同的 What 维度（上位概念） **MECE分类视角框架（选择最适合的一个维度）**： - **物理维度**：基于物理属性（形态、结构、状态等） - **化学维度**：基于化学成分和组成 - **生物维度**：基于生物分类（动物、植物、微生物等） - **功能维度**：基于客观功能分类（注意：是物体本身的功能属性，不是对其他事物的影响） **分类禁止项**： - ❌ 禁止从"对其他事物的影响/效果"角度分类 - ❌ 禁止从"目的/价值"角度分类 - ❌ 禁止使用抽象概念作为分类 **判断标准**： 分类名称应该能回答"XX是一种什么？"，而不是"XX对YY有什么作用？"或"XX导致什么结果？" ## 节点字段说明 ### 分类节点（非叶子节点） - **元素名称**：分类名称（基于 What 维度的抽象分类） - **描述**：对这个分类的定义说明（回答"这是什么类型的东西"） - **子项**：该分类下的子分类或具体元素列表 ### 元素节点（叶子节点） - **元素名称**：使用原子名词（最基础的名词形式，去除修饰语） - **描述**：该元素在帖子中的定义，包含具体表现 - **上游支撑**：说明该元素如何支撑选题的展开和目的达成 - **贯穿段落**：该元素贯穿的内容范围，列表格式（如：["图1", "图2", "正文——具体文字内容片段"]） - **主题支撑性得分**：0-10分 - **多段落共性得分**：0-10分 - **推理依据**：为什么这是核心元素 - **子项**：[]（叶子节点，必须为空列表） ## 输出要求 - 元素数量：宁少勿多，只保留真正核心的元素 - 同级正交：同级节点之间必须相互独立 - 描述清晰：每个节点都要有清晰的定义 - 支撑明确：每个元素都要说明如何支撑选题 - 层级灵活：根据实际内容确定树的深度 # 输出（JSON）- 树状结构 {{ "元素列表": [ {{ "元素名称": "一级分类名称", "描述": "对这个分类的定义说明", "子项": [ {{ "元素名称": "二级分类名称（可选）", "描述": "对这个子分类的定义说明", "子项": [ {{ "元素名称": "具体元素名称（原子名词）", "描述": "该元素在帖子中的定义，包含具体表现", "上游支撑": "该元素如何支撑选题的展开和目的达成", "贯穿段落": ["图1", "图2", "正文——具体文字内容片段"], "主题支撑性得分": 8, "多段落共性得分": 9, "推理依据": "为什么这是核心元素（如何满足双维度要求）", "子项": [] }} ] }}, {{ "元素名称": "具体元素名称（原子名词）", "描述": "该元素在帖子中的定义，包含具体表现", "上游支撑": "该元素如何支撑选题的展开和目的达成", "贯穿段落": ["图3", "正文——具体文字内容片段"], "主题支撑性得分": 7, "多段落共性得分": 8, "推理依据": "为什么这是核心元素", "子项": [] }} ] }} ] }} **注意**： - 如果分类下直接是具体元素，则元素是该分类的直接子项 - 如果需要更细的分类，可以在分类下再创建子分类 - 具体元素（叶子节点）必须包含完整的元素字段（描述、上游支撑、评分等） - 分类节点（非叶子）只需要包含名称、描述和子项""" # 构建多模态消息 message_content = [{"type": "text", "text": prompt}] if images: for idx, image_url in enumerate(images, 1): message_content.append({"type": "text", "text": f"[图片{idx}]"}) message_content.append({ "type": "image_url", "image_url": {"url": image_url} }) messages = [ {"role": "system", "content": self.system_prompt}, {"role": "user", "content": message_content} ] # 调用LLM result = LLMInvoker.safe_invoke( self, "去重与整合", messages, fallback={"元素列表": []} ) # 只打印JSON结构 import json logger.info(f"Step2.4结果:\n{json.dumps(result, ensure_ascii=False, indent=2)}") return result def step2(self, text_data: dict, images: List[str], topic_description: dict, content_weight: dict = None) -> dict: """第二步：元素 - 提取与主题相关的核心元素 执行完整的4步筛选流程： 1. 候选元素识别 2. 双维度评分（主题支撑性 + 多段落共性） 3. 严格筛选（满足任一条件即可） 4. 去重与整合 """ if not self.is_initialized: self.initialize() logger.info("\n" + "█" * 80) logger.info("█" + " " * 78 + "█") logger.info("█" + " " * 25 + "【Step2】元素提取 - 四步筛选流程" + " " * 23 + "█") logger.info("█" + " " * 78 + "█") logger.info("█" * 80) # Step 2.1: 候选元素识别 logger.info("\n▶ 开始执行 Step 2.1: 候选元素识别") candidates_result = self.step2_1_identify_candidates( text_data, images, topic_description, content_weight ) # Step 2.2: 双维度评分 logger.info("\n▶ 开始执行 Step 2.2: 双维度评分（主题支撑性 + 多段落共性）") scored_result = self.step2_2_score_dimensions( candidates_result, text_data, images, topic_description, content_weight ) # Step 2.3: 严格筛选 logger.info("\n▶ 开始执行 Step 2.3: 严格筛选（满足任一条件即可）") filtered_result = self.step2_3_strict_filter(scored_result) # Step 2.4: 去重与整合 logger.info("\n▶ 开始执行 Step 2.4: 去重与整合") final_result = self.step2_4_deduplicate( filtered_result, text_data, images, topic_description ) # 只打印JSON结构 import json logger.info(f"Step2最终结果:\n{json.dumps(final_result, ensure_ascii=False, indent=2)}") return final_result def _build_weight_info(self, content_weight: dict = None) -> str: """构建权重信息文本""" if not content_weight: return "" image_weight = content_weight.get("图片权重", 5) text_weight = content_weight.get("文字权重", 5) primary_source = content_weight.get("主要信息源", "both") return f"""# 图文权重信息 - 图片权重: {image_weight}/10 - 文字权重: {text_weight}/10 - 主要信息源: {primary_source} **元素提取时需要考虑权重**： - 如果图片权重高，元素提取应更多关注图片中的视觉元素 - 如果文字权重高，元素提取应更多关注文字描述的概念和对象 - 如果权重相近，需要综合图文信息提取元素 """ def _build_post_content(self, text_data: dict) -> str: """构建帖子内容文本""" post_content_parts = [] if text_data.get("title"): post_content_parts.append(f"标题: {text_data['title']}") if text_data.get("body"): post_content_parts.append(f"正文: {text_data['body']}") return "\n".join(post_content_parts) if post_content_parts else "无文本信息" def _build_topic_text(self, topic_description: dict) -> str: """构建选题描述文本""" topic_parts = [] if topic_description.get("主题"): topic_parts.append(f"主题: {topic_description['主题']}") if topic_description.get("描述"): topic_parts.append(f"描述: {topic_description['描述']}") return "\n".join(topic_parts) if topic_parts else "无选题描述" def _build_messages(self, state: dict) -> List[dict]: """构建消息 - ScriptUnderstandingAgent 不使用此方法 本 Agent 使用 step1 和 step2 方法直接构建消息 """ return [] def _update_state(self, state: dict, response) -> dict: """更新状态 - ScriptUnderstandingAgent 不使用此方法 本 Agent 使用 step1 和 step2 方法直接返回结果 """ return state