# What解构业务技术设计文档 v2.0 > **文档目标**: 为开发者提供清晰、可执行的技术实现指导 > > **设计原则**: 每个决策都有明确的PRD依据和技术理由 --- ## 目录 1. [PRD目标解析](#1-prd目标解析) 2. [功能需求拆解](#2-功能需求拆解) 3. [工作流 vs 组件职责划分](#3-工作流-vs-组件职责划分) 4. [工作流设计](#4-工作流设计) 5. [组件设计](#5-组件设计) 6. [数据结构设计](#6-数据结构设计) 7. [实现路线图](#7-实现路线图) --- ## 1. PRD目标解析 ### 1.1 核心目标是什么？ **PRD原文** (1. 需求目标): > 针对给定的小红书多模态内容（图文、视频、音频），从消费者视角进行充分必要的What要素逆向解构，通过层级化递归深入分析，识别和提取内容中所有构成成分"what"。 **核心目标拆解**: | 维度 | 具体含义 | PRD依据 | |------|---------|---------| | **输入对象** | 小红书多模态内容（图+文+视频+音频） | 2.1.1节 | | **视角** | 从消费者视角分析 | 1. 需求目标 | | **方法** | 层级化递归深入分析 | 1. 需求目标、2.2.3节 | | **产出** | What要素树（JSON结构） | 2.2.1节、2.2.2节 | | **价值** | 通过聚合分析提取爆款特征和趋势 | 1. 需求目标（核心价值） | **目标提炼**: > **构建一个递归解构系统，将多模态帖子内容自上而下、从整体到局部地拆解为树状的"What元素"结构，每个元素包含动态描述维度（由知识库提供），并关联原始素材。** ### 1.2 关键约束条件 **PRD依据** (2.2.3节、4节): 1. **递归深度**: 最多10层 2. **停止条件**: - 已是不可再分的最小单元 - 达到10层深度 3. **解构原则**: - 由宏观到微观、由整体到局部 - 子节点不重不漏（充分必要） 4. **知识驱动**: 描述维度、分割判断都由知识库动态提供 5. **客观性**: 严格基于帖子内容，禁止臆测 --- ## 2. 功能需求拆解 ### 2.1 核心功能清单 基于PRD 3.1节"总体解构流程"和3.3节"对单一节点的递归解构"，提炼出以下核心功能： | 功能ID | 功能名称 | 功能描述 | PRD依据 | 实现方式 | |--------|---------|---------|---------|---------| | **F1** | 帖子初理解 | 分析帖子整体，提取品类、主题、关键词 | 3.1节步骤1 | Agent组件 | | **F2** | 知识需求生成 | 根据品类/主题/关键词生成解构所需的query | 3.2节 | 工作流逻辑 | | **F3** | 知识检索 | 调用知识库获取描述维度、工具推荐、分割判断 | 3.2节、3.3.2节 | Tool组件 | | **F4** | 评论理解 | 分析评论，提取消费者关注的亮点 | 3.1节步骤4 | Agent组件 | | **F5** | 图片递归解构 | 递归解构图片元素，直到不可再分 | 3.1节步骤6-7、3.3节 | Agent组件 | | **F6** | 图片分割 | 将图片分割为多个独立的视觉元素 | 3.3.1节、3.5节 | Tool组件 | | **F7** | 文本递归解构 | 递归解构文本元素（标题/正文/标签） | 3.1节步骤8-11、3.3节 | Agent组件 | | **F8** | 文本切分 | 根据知识库指导切分文本 | 3.5节 | Function组件 | | **F9** | 帖子整体解构 | 从点线面体维度总结帖子 | 3.1节步骤13、3.2节 | Agent组件 | | **F10** | 结果汇总 | 将所有解构结果组装为树状JSON | 2.2.2节 | Function组件 | ### 2.2 功能依赖关系 ``` F1(帖子初理解) ↓ F2(知识需求生成) → F3(知识检索) ↓ F4(评论理解) ↓ [F5(图片递归解构) + F6(图片分割)] 并行 [F7(文本递归解构) + F8(文本切分)] ↓ F9(帖子整体解构) ↓ F10(结果汇总) ``` **关键观察**: - F5和F7内部都会多次调用F3(知识检索) - F6和F8是F5、F7的支撑工具 - F2是一个轻量级逻辑，不需要独立组件 --- ## 3. 工作流 vs 组件职责划分 ### 3.1 划分原则 **CLAUDE.md原则**: > - **工作流**：负责编排组件执行顺序和流程逻辑 > - **组件**：实现具体业务功能 ### 3.2 职责划分表 | 功能 | 实现方式 | 决策依据 | |------|---------|---------| | **F1 帖子初理解** | **Agent组件** | 需要多模态理解+语义推理（不确定性任务） | | **F2 知识需求生成** | **工作流逻辑** | 简单的字符串拼接，不需要独立组件 | | **F3 知识检索** | **Tool组件** | 需要被Agent调用 + 确定性API调用 | | **F4 评论理解** | **Agent组件** | 需要情感分析+语义理解（不确定性任务） | | **F5 图片递归解构** | **Agent组件** | 需要LLM视觉理解+ReAct模式调用工具 | | **F6 图片分割** | **Tool组件** | 需要被Agent调用 + 确定性模型调用 | | **F7 文本递归解构** | **Agent组件** | 需要LLM语义理解+ReAct模式调用工具 | | **F8 文本切分** | **Function组件** | 确定性字符串处理 + 工作流直接调用 | | **F9 帖子整体解构** | **Agent组件** | 需要综合分析+抽象提炼（不确定性任务） | | **F10 结果汇总** | **Function组件** | 确定性数据结构转换 + 工作流直接调用 | | **递归控制** | **工作流逻辑** | 编排递归流程，管理状态和深度 | | **并行执行** | **工作流逻辑** | 编排图片和文本并行解构 | **总结**: - **1个主工作流**: WhatDeconstructionWorkflow - **5个Agent组件**: F1, F4, F5, F7, F9 - **2个Tool组件**: F3, F6 - **2个Function组件**: F8, F10 --- ## 4. 工作流设计 ### 4.1 工作流核心职责 **WhatDeconstructionWorkflow** 的核心职责： 1. **编排执行顺序**: 按PRD 3.1节流程图编排各组件 2. **管理状态传递**: 维护 WhatDeconstructionState，在各节点间传递数据 3. **控制递归深度**: 确保不超过10层 4. **并行执行控制**: 图片和文本解构并行执行 5. **知识query生成**: 根据中间结果动态生成知识库query ### 4.2 工作流流程图 **PRD依据**: 3.1节"总体解构流程" ```mermaid graph TD START([开始]) --> N1[节点1: 帖子初理解
PostUnderstandingAgent] N1 --> N2[节点2: 评论理解
CommentAnalysisAgent] N2 --> PARALLEL{并行分支} PARALLEL --> N3[节点3: 图片递归解构
RecursiveImageDeconstructionAgent] PARALLEL --> N4[节点4: 文本递归解构
RecursiveTextDeconstructionAgent] N3 --> N5[节点5: 帖子整体解构
PostSummaryDeconstructionAgent] N4 --> N5 N5 --> N6[节点6: 结果汇总
ResultAggregationFunction] N6 --> END([结束]) style N3 fill:#e1f5ff style N4 fill:#e1f5ff style PARALLEL fill:#fff3cd ``` **流程说明**: | 节点 | 输入 | 输出 | PRD依据 | |------|------|------|---------| | N1 | 帖子多模态内容 | category, theme, keywords | 3.1节步骤1 | | N2 | 评论列表 | consumer_highlights | 3.1节步骤4 | | N3 | images列表 | image_deconstruction_results | 3.1节步骤6-7 | | N4 | text对象 | text_deconstruction_results | 3.1节步骤8-11 | | N5 | 所有子元素解构结果 | post_summary | 3.1节步骤13 | | N6 | 所有解构结果 | final_result (JSON树) | 2.2.2节 | ### 4.3 工作流核心代码框架 ```python from langgraph.graph import StateGraph, END from typing import TypedDict, List, Dict, Any # 1. 定义状态 class WhatDeconstructionState(TypedDict): """工作流状态定义""" # 输入 images: List[str] text: Dict[str, Any] # {title, body, hashtags} comments: List[Dict[str, str]] # 中间状态 category: str theme: str keywords: List[str] consumer_highlights: List[Dict] # 解构结果 image_deconstruction_results: List[Dict] text_deconstruction_results: List[Dict] post_summary: Dict[str, Any] # 最终输出 final_result: Dict[str, Any] # 2. 工作流类 class WhatDeconstructionWorkflow: """What解构主工作流""" def __init__(self, llm, knowledge_retrieval_tool, image_segment_tool): # 初始化所有组件 self.post_understanding_agent = PostUnderstandingAgent(llm) self.comment_analysis_agent = CommentAnalysisAgent(llm) self.image_agent = RecursiveImageDeconstructionAgent( llm, knowledge_retrieval_tool, image_segment_tool, max_depth=10 ) self.text_agent = RecursiveTextDeconstructionAgent( llm, knowledge_retrieval_tool, max_depth=10 ) self.summary_agent = PostSummaryDeconstructionAgent(llm, knowledge_retrieval_tool) self.aggregation_func = ResultAggregationFunction() # 构建图 self.graph = self._build_graph() def _build_graph(self) -> StateGraph: """构建LangGraph工作流""" workflow = StateGraph(WhatDeconstructionState) # 添加节点 workflow.add_node("post_understanding", self._node_post_understanding) workflow.add_node("comment_analysis", self._node_comment_analysis) workflow.add_node("image_deconstruction", self._node_image_deconstruction) workflow.add_node("text_deconstruction", self._node_text_deconstruction) workflow.add_node("post_summary", self._node_post_summary) workflow.add_node("result_aggregation", self._node_result_aggregation) # 定义边（流程） workflow.set_entry_point("post_understanding") workflow.add_edge("post_understanding", "comment_analysis") # 并行分支：评论理解后，图片和文本并行解构 workflow.add_edge("comment_analysis", "image_deconstruction") workflow.add_edge("comment_analysis", "text_deconstruction") # 汇聚：图片和文本都完成后，执行帖子整体解构 workflow.add_edge("image_deconstruction", "post_summary") workflow.add_edge("text_deconstruction", "post_summary") workflow.add_edge("post_summary", "result_aggregation") workflow.add_edge("result_aggregation", END) return workflow.compile() # 3. 节点实现 async def _node_post_understanding(self, state: WhatDeconstructionState): """节点1: 帖子初理解""" result = await self.post_understanding_agent.ainvoke({ "images": state["images"], "text": state["text"] }) return { "category": result["category"], "theme": result["theme"], "keywords": result["keywords"] } async def _node_comment_analysis(self, state: WhatDeconstructionState): """节点2: 评论理解""" result = await self.comment_analysis_agent.ainvoke({ "comments": state["comments"], "post_content": state["text"] }) return {"consumer_highlights": result["highlights"]} async def _node_image_deconstruction(self, state: WhatDeconstructionState): """节点3: 图片递归解构（处理所有图片）""" results = [] for idx, image_path in enumerate(state["images"]): result = await self.image_agent.ainvoke({ "image_path": image_path, "node_id": f"img_{idx+1}", "depth": 0, "category": state["category"], "theme": state["theme"] }) results.append(result) return {"image_deconstruction_results": results} async def _node_text_deconstruction(self, state: WhatDeconstructionState): """节点4: 文本递归解构（标题+正文+标签）""" results = [] text_data = state["text"] # 解构标题 if text_data.get("title"): title_result = await self.text_agent.ainvoke({ "text": text_data["title"], "text_type": "title", "node_id": "title", "depth": 0 }) results.append(title_result) # 解构正文 if text_data.get("body"): body_result = await self.text_agent.ainvoke({ "text": text_data["body"], "text_type": "body", "node_id": "body", "depth": 0 }) results.append(body_result) # 解构话题标签 if text_data.get("hashtags"): hashtags_result = await self.text_agent.ainvoke({ "text": " ".join(text_data["hashtags"]), "text_type": "hashtags", "node_id": "hashtags", "depth": 0 }) results.append(hashtags_result) return {"text_deconstruction_results": results} async def _node_post_summary(self, state: WhatDeconstructionState): """节点5: 帖子整体解构""" result = await self.summary_agent.ainvoke({ "category": state["category"], "theme": state["theme"], "keywords": state["keywords"], "image_results": state["image_deconstruction_results"], "text_results": state["text_deconstruction_results"], "consumer_highlights": state["consumer_highlights"] }) return {"post_summary": result["summary"]} async def _node_result_aggregation(self, state: WhatDeconstructionState): """节点6: 结果汇总""" final_result = self.aggregation_func.invoke({ "post_summary": state["post_summary"], "image_results": state["image_deconstruction_results"], "text_results": state["text_deconstruction_results"] }) return {"final_result": final_result} # 4. 入口方法 async def ainvoke(self, input_data: Dict) -> Dict: """执行工作流""" initial_state = WhatDeconstructionState( images=input_data["multimedia_content"]["images"], text=input_data["multimedia_content"]["text"], comments=input_data["comments"] ) result = await self.graph.ainvoke(initial_state) return result["final_result"] ``` **关键设计点**: 1. **并行执行**: `image_deconstruction` 和 `text_deconstruction` 从 `comment_analysis` 同时出发 2. **状态传递**: 通过 `WhatDeconstructionState` 在节点间传递数据 3. **递归封装**: 图片/文本的递归逻辑封装在各自的Agent内部 4. **知识库调用**: 在Agent内部调用 KnowledgeRetrievalTool --- ## 5. 组件设计 ### 5.1 Agent组件总览 **CLAUDE.md原则**: > **Agent组件特征**: 不确定性任务 + LLM智能能力 + 推理/多模态理解/语义分析 | Agent | 任务类型 | LLM能力需求 | PRD依据 | |-------|---------|------------|---------| | PostUnderstandingAgent | 多模态理解 | 图文融合理解 | 3.1节步骤1 | | CommentAnalysisAgent | 语义理解 | 情感分析、关键点提取 | 3.1节步骤4、2.1.4节 | | RecursiveImageDeconstructionAgent | 视觉理解+推理 | 图片理解、分割判断 | 3.3节、3.3.1节 | | RecursiveTextDeconstructionAgent | 语义理解+推理 | 文本分析、切分判断 | 3.3节 | | PostSummaryDeconstructionAgent | 综合分析 | 抽象提炼、多维度总结 | 3.2节 | --- ### 5.2 Agent组件详细设计 #### 5.2.1 PostUnderstandingAgent（帖子初理解） **组件类型**: Agent (继承 BaseLLMAgent) **选择依据**: - **PRD依据**: 3.1节步骤1 "帖子初理解，输出品类、主题、关键词" - **CLAUDE.md依据**: 需要多模态理解（图+文），属于不确定性任务 **核心职责**: 1. 综合分析图片和文本内容 2. 推理提取品类（如"时尚美妆"、"美食"等） 3. 总结主题（如"夏日穿搭"） 4. 提取关键词（如"OOTD"、"小白裙"） **关键点**: | 关键点 | 实现方式 | 依据 | |--------|---------|------| | **多模态输入** | 将图片和文本一起输入给LLM | LLM具备多模态理解能力 | | **结构化输出** | 使用JSON格式约束LLM输出 | 需要提取固定字段 | | **品类识别准确性** | 在prompt中提供常见品类列表 | 提升品类识别准确性 | **代码框架**: ```python from agents.base import BaseLLMAgent from typing import List, Dict, Any class PostUnderstandingAgent(BaseLLMAgent): """帖子初理解Agent PRD依据: 3.1节步骤1 功能: 分析帖子多模态内容，提取品类、主题、关键词 """ async def ainvoke(self, state: Dict[str, Any]) -> Dict[str, Any]: """ 执行帖子理解 Args: state: { "images": List[str], # 图片路径列表 "text": { # 文本内容 "title": str, "body": str, "hashtags": List[str] } } Returns: { "category": str, # 品类 "theme": str, # 主题 "keywords": List[str] # 关键词 } """ # 1. 构建多模态prompt prompt = self._build_prompt(state["images"], state["text"]) # 2. 调用LLM response = await self.llm.ainvoke(prompt) # 3. 解析结果 result = self._parse_response(response) return result def _build_prompt(self, images: List[str], text: Dict) -> str: """构建prompt 关键点: - 多模态输入（图片+文本） - 明确要求输出JSON格式 - 提供品类参考列表 """ return f""" 你是一个小红书内容分析专家。请分析以下帖子的多模态内容，提取品类、主题和关键词。 【文本内容】 标题: {text.get('title', '')} 正文: {text.get('body', '')} 话题标签: {', '.join(text.get('hashtags', []))} 【图片】 {self._format_images(images)} 【常见品类参考】 时尚美妆、美食、旅游、家居、母婴、运动健身、数码科技等 请以JSON格式输出: {{ "category": "品类", "theme": "主题", "keywords": ["关键词1", "关键词2", ...] }} """ def _parse_response(self, response: str) -> Dict: """解析LLM响应，提取JSON""" import json # 提取JSON部分 json_str = self._extract_json(response) return json.loads(json_str) ``` --- #### 5.2.2 CommentAnalysisAgent（评论理解） **组件类型**: Agent (继承 BaseLLMAgent) **选择依据**: - **PRD依据**: 2.1.4节 "评论信息用于提取内容消费者关注的亮点" - **CLAUDE.md依据**: 需要情感分析和语义理解，属于不确定性任务 **核心职责**: 1. 分析评论区文字内容 2. 识别消费者关注的亮点（对应帖子的哪些元素） 3. 提取高频关注点和情绪共鸣点 **关键点**: | 关键点 | 实现方式 | 依据 | |--------|---------|------| | **亮点映射** | 将评论中的关注点映射到帖子具体元素 | PRD 2.1.4节要求 | | **情感分析** | 识别评论的情感倾向（共鸣、兴趣、疑问等） | 理解消费者反应 | | **高频提取** | 聚合多条评论的共同关注点 | 识别关键吸引点 | **代码框架**: ```python class CommentAnalysisAgent(BaseLLMAgent): """评论理解Agent PRD依据: 2.1.4节 功能: 分析评论，提取消费者关注的亮点 """ async def ainvoke(self, state: Dict[str, Any]) -> Dict[str, Any]: """ 执行评论分析 Args: state: { "comments": List[Dict], # [{user, content}, ...] "post_content": Dict # 帖子内容（用于映射） } Returns: { "highlights": List[Dict] # 消费者关注的亮点列表 # [ # { # "element": "封面图中的小白裙", # "reason": "多条评论询问购买链接", # "emotion": "兴趣、购买意愿" # }, # ... # ] } """ prompt = self._build_prompt(state["comments"], state["post_content"]) response = await self.llm.ainvoke(prompt) return {"highlights": self._parse_highlights(response)} def _build_prompt(self, comments: List[Dict], post_content: Dict) -> str: """构建prompt 关键点: - 提供帖子内容作为参考（用于映射） - 要求识别具体元素 - 要求分析情感和关注原因 """ comments_text = "\n".join([ f"- {c['content']}" for c in comments ]) return f""" 你是一个内容分析专家。请分析评论区内容，提取消费者关注的亮点。 【帖子内容】 标题: {post_content.get('title', '')} 正文: {post_content.get('body', '')} 【评论区】 {comments_text} 请分析: 1. 评论中提到了帖子的哪些具体元素（封面、图片、标题、正文中的哪部分） 2. 消费者对这些元素的关注原因（为什么吸引他们） 3. 评论的情感倾向（共鸣、兴趣、疑问、赞美等） 以JSON格式输出亮点列表: {{ "highlights": [ {{ "element": "具体元素描述", "reason": "关注原因", "emotion": "情感倾向" }}, ... ] }} """ ``` --- #### 5.2.3 RecursiveImageDeconstructionAgent（图片递归解构） **组件类型**: Agent (继承 BaseReactAgent) **选择依据**: - **PRD依据**: 3.3节 "对单一节点的递归解构"、3.3.1节 "判断图片元素是否要继续分割拆解" - **CLAUDE.md依据**: 需要LLM视觉理解 + 动态调用工具（ReAct模式） **核心职责**: 1. 粗理解图片，生成what字段 2. 调用知识库获取描述维度 3. 细致理解图片，填充描述值 4. 判断是否需要分割（调用知识库） 5. 如需分割，调用图片分割工具 6. 递归处理子元素（最多10层） **关键点**: | 关键点 | 实现方式 | PRD依据 | |--------|---------|---------| | **ReAct模式** | 思考→行动→观察循环 | 需要多次调用工具 | | **递归控制** | 深度参数 + 停止条件判断 | 2.2.3节（最多10层） | | **知识驱动** | 描述维度和分割判断都由知识库提供 | 3.3.2节 | | **分割优先判断** | 优先判断是否为多子图拼接 | 3.3.1节 | | **what字段生成** | 对图片的文本描述 | 2.2.2节 | | **判断不依赖分割** | 判断是否拆解基于完整原图+文字描述，不依赖分割结果 | PRD 3.3.1节原则 | **递归流程图**: ```mermaid graph TD START([接收图片节点]) --> DEPTH{检查深度
是否≥10层?} DEPTH -->|是| STOP([返回结果
不再递归]) DEPTH -->|否| ROUGH[粗理解图片
基于完整原图+节点描述
生成what字段] ROUGH --> QUERY1[生成query:
获取描述维度] QUERY1 --> KB1[调用知识库] KB1 --> DETAIL[细致理解图片
基于完整原图+节点描述+维度
填充描述维度值] DETAIL --> QUERY2[生成query:
是否需要分割?] QUERY2 --> KB2[调用知识库
基于完整原图+what+描述] KB2 --> DECISION{知识库判断:
是否分割?} DECISION -->|否| STOP DECISION -->|是| IDENTIFY[识别子元素
基于完整原图
生成子元素描述] IDENTIFY --> SEGMENT[调用图片分割工具
传入完整原图
获得子图片列表] SEGMENT --> RECURSE[遍历子图片
递归调用自身
传递完整原图] RECURSE --> STOP style ROUGH fill:#e1f5ff style DETAIL fill:#e1f5ff style KB1 fill:#fff3cd style KB2 fill:#fff3cd style SEGMENT fill:#d4edda ``` **代码框架**: ```python from agents.base import BaseReactAgent from tools.base import BaseTool from typing import List, Dict, Any class RecursiveImageDeconstructionAgent(BaseReactAgent): """图片递归解构Agent PRD依据: 3.3节、3.3.1节 功能: 递归解构图片元素，最多10层 """ def __init__( self, llm, knowledge_retrieval_tool: BaseTool, image_segment_tool: BaseTool, max_depth: int = 10 ): super().__init__(llm, tools=[knowledge_retrieval_tool, image_segment_tool]) self.knowledge_tool = knowledge_retrieval_tool self.segment_tool = image_segment_tool self.max_depth = max_depth async def ainvoke(self, state: Dict[str, Any]) -> Dict[str, Any]: """ 递归解构图片 Args: state: { "original_image_path": str, # **完整原图路径**（PRD 3.3.1：始终传递未分割的完整图） "current_image_path": str, # 当前节点对应的图片路径（首次调用时=original_image_path） "node_description": str, # 当前节点的文字描述（首次为空，递归时为子元素描述） "node_id": str, # 节点ID（如"1", "1_1"） "depth": int, # 当前递归深度 "category": str, # 帖子品类（用于生成query） "theme": str # 帖子主题（用于生成query） } Returns: { "id": str, "what": str, "description": Dict, "children": List[Dict], # 子元素（递归结构） "importance_weight": float } """ # 步骤1: 检查递归深度（PRD 2.2.3节） if state["depth"] >= self.max_depth: return self._build_leaf_node(state) # 步骤2: 粗理解当前节点，生成what字段（PRD 3.3节流程步骤B0） # **关键**: 基于完整原图 + 节点描述进行理解（PRD 3.3.1原则） what_value = await self._rough_understanding( original_image=state["original_image_path"], current_image=state["current_image_path"], node_description=state.get("node_description", "") ) # 步骤3: 调用知识库获取描述维度（PRD 3.3.2节） description_query = self._generate_description_query(what_value) description_dimensions = await self.knowledge_tool.ainvoke({ "query": description_query, "query_type": "description" }) # 步骤4: 根据描述维度细致理解当前节点（PRD 3.3节流程步骤C） # **关键**: 基于完整原图 + 节点描述 + 描述维度进行理解（PRD 3.3.1原则） description_values = await self._detailed_understanding( original_image=state["original_image_path"], current_image=state["current_image_path"], node_description=state.get("node_description", ""), description_dimensions=description_dimensions ) # 步骤5: 判断是否需要分割（PRD 3.3节流程步骤D0） # **关键**: 判断分割的输入是完整原图 + what + 描述，不依赖分割结果（PRD 3.3.1原则） split_query = self._generate_split_query(what_value, description_values) split_decision = await self.knowledge_tool.ainvoke({ "query": split_query, "query_type": "split_decision" }) # 构建当前节点结果 result = { "id": state["node_id"], "what": what_value, "description": description_values, "children": [], "importance_weight": 0.5 # TODO: 计算权重逻辑 } # 步骤6: 如需分割，执行分割并递归（PRD 3.3节流程步骤E1-E2） if split_decision.get("should_split", False): # 识别子元素（基于完整原图 + 当前节点描述） sub_elements = await self._identify_sub_elements( original_image=state["original_image_path"], current_node_what=what_value, current_node_description=state.get("node_description", "") ) # 调用分割工具（输入：完整原图 + 子元素描述列表） sub_images = await self.segment_tool.ainvoke({ "image_path": state["original_image_path"], # **传入完整原图** "elements_description": sub_elements }) # 递归处理每个子元素 children = [] for idx, (sub_img_path, sub_element_desc) in enumerate(zip(sub_images, sub_elements)): child_state = { "original_image_path": state["original_image_path"], # **传递完整原图** "current_image_path": sub_img_path, # 分割后的子图片 "node_description": sub_element_desc, # 子元素的文字描述 "node_id": f"{state['node_id']}_{idx+1}", "depth": state["depth"] + 1, "category": state["category"], "theme": state["theme"] } child_result = await self.ainvoke(child_state) children.append(child_result) result["children"] = children return result async def _rough_understanding( self, original_image: str, current_image: str, node_description: str ) -> str: """粗理解当前节点，生成what字段 PRD依据: 3.3节步骤B0, 3.3.1节原则 关键点: 基于完整原图 + 节点描述进行理解 Args: original_image: 完整原图路径（提供上下文） current_image: 当前节点对应的图片路径 node_description: 当前节点的文字描述（首次为空，递归时为子元素描述） """ # 构建prompt：如果有节点描述，说明是递归调用，需要结合原图理解特定部分 if node_description: prompt = f""" 请基于完整原图的上下文，描述指定的视觉元素。 【完整原图】: {original_image} 【当前关注的元素描述】: {node_description} 【当前元素图片】: {current_image} 请用一句话描述这个视觉元素的核心内容。 要求: 简洁准确，聚焦于该元素本身。 """ else: # 首次调用，直接理解整图 prompt = f""" 请用一句话描述这张图片的核心内容和视觉元素。 要求: 简洁准确，涵盖主要视觉要素。 图片: {original_image} """ response = await self.llm.ainvoke(prompt) return response.strip() def _generate_description_query(self, what_value: str) -> str: """生成描述维度query PRD依据: 3.3.2节、2.2.2节 Query句式: PRD 3.3.2节定义 """ return f'刻画描述"{what_value}"核心特征的角度和维度有哪些？请尽可能不重不漏列举全。' def _generate_split_query(self, what_value: str, description: Dict) -> str: """生成分割判断query PRD依据: 3.3节步骤D0 """ return f'该节点"{what_value}"是否需要继续拆解分割？' async def _detailed_understanding( self, original_image: str, current_image: str, node_description: str, dimensions: Dict ) -> Dict: """根据描述维度细致理解当前节点 PRD依据: 3.3节步骤C, 3.3.1节原则 关键点: 基于完整原图 + 节点描述 + 描述维度进行理解 Args: original_image: 完整原图路径（提供上下文） current_image: 当前节点对应的图片路径 node_description: 当前节点的文字描述 dimensions: 知识库返回的描述维度 """ # 构建prompt：结合原图上下文和节点描述 if node_description: prompt = f""" 请基于完整原图的上下文，分析指定视觉元素的特征。 【完整原图】: {original_image} 【当前关注的元素描述】: {node_description} 【当前元素图片】: {current_image} 请从以下维度分析这个视觉元素: {self._format_dimensions(dimensions)} 以JSON格式输出各维度的值。 """ else: # 首次调用，分析整图 prompt = f""" 请从以下维度分析图片: {self._format_dimensions(dimensions)} 图片: {original_image} 以JSON格式输出各维度的值。 """ response = await self.llm.ainvoke(prompt) return self._parse_description(response) async def _identify_sub_elements( self, original_image: str, current_node_what: str, current_node_description: str ) -> List[str]: """识别当前节点中的子元素 PRD依据: 3.3.1节 关键点: 基于完整原图 + 当前节点描述识别子元素（不依赖分割结果） Args: original_image: 完整原图路径（提供上下文） current_node_what: 当前节点的what字段值 current_node_description: 当前节点的文字描述 """ # 构建prompt：基于原图和当前节点描述识别子元素 if current_node_description: prompt = f""" 请基于完整原图的上下文，识别指定视觉元素中包含的子元素。 【完整原图】: {original_image} 【当前关注的元素】: {current_node_what} 【当前元素的详细描述】: {current_node_description} 判断步骤: 1. 该元素是否为多张子图拼接？（如4宫格、9宫格） 2. 如果不是，该元素包含哪些独立的视觉对象？（如人物、商品、背景元素） 请列出所有子元素的文字描述，每个元素一行。 """ else: # 首次调用，分析整图 prompt = f""" 分析图片"{current_node_what}"，识别其中包含的子视觉元素。 【图片】: {original_image} 判断步骤: 1. 是否为多张子图拼接？（如4宫格、9宫格） 2. 如果不是，包含哪些独立的视觉对象？（如人物、商品、背景元素） 请列出所有子元素的文字描述，每个元素一行。 """ response = await self.llm.ainvoke(prompt) return [line.strip() for line in response.strip().split('\n') if line.strip()] ``` --- #### 5.2.4 RecursiveTextDeconstructionAgent（文本递归解构） **组件类型**: Agent (继承 BaseReactAgent) **选择依据**: - **PRD依据**: 3.3节 "对单一节点的递归解构" - **CLAUDE.md依据**: 需要LLM语义理解 + 动态调用工具 **核心职责**: 1. 提取文本原文作为what字段（不可改写） 2. 调用知识库获取描述维度 3. 分析文本，填充描述值 4. 判断是否需要切分 5. 如需切分，调用文本切分函数 6. 递归处理子文本 **关键点**: | 关键点 | 实现方式 | PRD依据 | |--------|---------|---------| | **what字段是原文** | 直接使用文本原文，不可总结改写 | 2.2.2节 | | **充分必要原则** | 子文本合并后必须等于父文本 | 3.4节 | | **递归控制** | 深度限制 + 最小语义单元判断 | 2.2.3节 | **代码框架**: ```python class RecursiveTextDeconstructionAgent(BaseReactAgent): """文本递归解构Agent PRD依据: 3.3节 功能: 递归解构文本元素（标题/正文/标签） """ def __init__( self, llm, knowledge_retrieval_tool: BaseTool, max_depth: int = 10 ): super().__init__(llm, tools=[knowledge_retrieval_tool]) self.knowledge_tool = knowledge_retrieval_tool self.max_depth = max_depth async def ainvoke(self, state: Dict[str, Any]) -> Dict[str, Any]: """ 递归解构文本 Args: state: { "text": str, # 当前文本内容 "text_type": str, # 文本类型（title/body/hashtags） "node_id": str, # 节点ID "depth": int # 当前递归深度 } Returns: { "id": str, "what": str, # 文本原文 "description": Dict, "children": List[Dict] } """ # 步骤1: 检查递归深度 if state["depth"] >= self.max_depth: return self._build_leaf_node(state) # 步骤2: what字段直接使用原文（PRD 2.2.2节） what_value = state["text"] # 步骤3: 调用知识库获取描述维度 description_query = self._generate_description_query( what_value, state["text_type"] ) description_dimensions = await self.knowledge_tool.ainvoke({ "query": description_query, "query_type": "description" }) # 步骤4: 根据描述维度分析文本 description_values = await self._analyze_text( what_value, description_dimensions, state["text_type"] ) # 步骤5: 判断是否需要切分 split_query = self._generate_split_query(what_value) split_decision = await self.knowledge_tool.ainvoke({ "query": split_query, "query_type": "split_decision" }) result = { "id": state["node_id"], "what": what_value, # 原文，不可改写 "description": description_values, "children": [] } # 步骤6: 如需切分，执行切分并递归 if split_decision.get("should_split", False): # 获取切分规则 split_rules = split_decision.get("split_rules", {}) # 执行文本切分（调用Function组件） sub_texts = self._split_text(what_value, split_rules) # 递归处理子文本 children = [] for idx, sub_text in enumerate(sub_texts): child_state = { "text": sub_text, "text_type": state["text_type"], "node_id": f"{state['node_id']}_{idx+1}", "depth": state["depth"] + 1 } child_result = await self.ainvoke(child_state) children.append(child_result) result["children"] = children return result def _split_text(self, text: str, split_rules: Dict) -> List[str]: """切分文本 PRD依据: 3.5节 关键点: 保证子文本合并后等于父文本（充分必要） """ strategy = split_rules.get("strategy", "sentence") if strategy == "sentence": # 按句子切分 import re sentences = re.split(r'[。！？\n]', text) return [s.strip() for s in sentences if s.strip()] elif strategy == "paragraph": # 按段落切分 paragraphs = text.split('\n\n') return [p.strip() for p in paragraphs if p.strip()] else: return [text] ``` --- #### 5.2.5 PostSummaryDeconstructionAgent（帖子整体解构） **组件类型**: Agent (继承 BaseLLMAgent) **选择依据**: - **PRD依据**: 3.2节 "帖子整体解构环节" - **CLAUDE.md依据**: 需要综合分析和抽象提炼能力 **核心职责**: 1. 调用知识库获取总结维度 2. 综合所有子元素解构结果 3. 从点线面体维度总结帖子 **关键点**: | 关键点 | 实现方式 | PRD依据 | |--------|---------|---------| | **动态总结维度** | 由知识库根据品类/主题/关键词提供 | 2.2.2节、3.2节 | | **综合分析** | 整合图片、文本、评论的解构结果 | 3.1节步骤13 | | **维度数量** | 不少于3个维度 | 2.2.2节 | **代码框架**: ```python class PostSummaryDeconstructionAgent(BaseLLMAgent): """帖子整体解构Agent PRD依据: 3.2节 功能: 从点线面体维度总结帖子 """ def __init__(self, llm, knowledge_retrieval_tool: BaseTool): super().__init__(llm) self.knowledge_tool = knowledge_retrieval_tool async def ainvoke(self, state: Dict[str, Any]) -> Dict[str, Any]: """ 执行帖子整体解构 Args: state: { "category": str, "theme": str, "keywords": List[str], "image_results": List[Dict], "text_results": List[Dict], "consumer_highlights": List[Dict] } Returns: { "summary": Dict # 动态维度的总结 } """ # 步骤1: 生成总结维度query（PRD 3.2节） summary_query = self._generate_summary_query( state["category"], state["theme"], state["keywords"] ) # 步骤2: 调用知识库获取总结维度 summary_dimensions = await self.knowledge_tool.ainvoke({ "query": summary_query, "query_type": "description" }) # 步骤3: 根据总结维度分析帖子 prompt = self._build_summary_prompt( summary_dimensions, state["image_results"], state["text_results"], state["consumer_highlights"] ) response = await self.llm.ainvoke(prompt) summary = self._parse_summary(response) return {"summary": summary} def _generate_summary_query( self, category: str, theme: str, keywords: List[str] ) -> str: """生成总结维度query PRD依据: 3.2节 Query句式: PRD 3.2节定义 """ keywords_str = ', '.join(keywords) return f'对于一篇主题为"{theme}"，品类为"{category}"，关键词包含"{keywords_str}"的多模态社交媒体帖子，从内容创作者视角进行What要素的初步识别和分类，需要使用哪些通用工具？' ``` --- ### 5.3 Tool组件详细设计 **CLAUDE.md原则**: > **Tool组件特征**: 确定性任务 + 可被Agent调用 #### 5.3.1 KnowledgeRetrievalTool（知识检索工具） **组件类型**: Tool (继承 BaseTool) **选择依据**: - **PRD依据**: 3.2节、3.3.2节 "对单一节点解构如何获取知识" - **CLAUDE.md依据**: 确定性API调用 + 需要被Agent调用 **核心职责**: 1. 接收query和query类型 2. 调用外部知识库API 3. 根据query类型解析返回结果 **关键点**: | 关键点 | 实现方式 | PRD依据 | |--------|---------|---------| | **支持多种query类型** | description/tool/split_decision | 3.3.2节 | | **降级策略** | 知识库不可用时使用LLM默认知识 | 容错处理 | | **结果格式化** | 转换为Agent可用的结构化数据 | 提升可用性 | **代码框架**: ```python from tools.base import BaseTool from typing import Dict, Any class KnowledgeRetrievalTool(BaseTool): """知识检索工具 PRD依据: 3.2节、3.3.2节 功能: 根据query检索解构所需的知识 """ name: str = "knowledge_retrieval" description: str = "根据query检索What解构所需的知识（描述维度、工具推荐、分割判断等）" def __init__(self, knowledge_base_url: str, fallback_llm=None): super().__init__() self.knowledge_base_url = knowledge_base_url self.fallback_llm = fallback_llm # 降级LLM async def _arun( self, query: str, query_type: str = "description" ) -> Dict[str, Any]: """ 调用知识库检索 Args: query: 查询语句（PRD 3.3.2节定义的句式） query_type: 查询类型 - description: 获取描述维度 - tool: 获取工具推荐 - split_decision: 获取分割判断 Returns: 根据query_type返回不同格式的结果 """ try: # 调用知识库API response = await self._call_knowledge_base(query) # 根据query类型解析结果 if query_type == "description": return self._parse_description_dimensions(response) elif query_type == "tool": return self._parse_tool_recommendations(response) elif query_type == "split_decision": return self._parse_split_decision(response) return response except Exception as e: # 降级策略: 使用LLM默认知识 if self.fallback_llm: return await self._fallback_with_llm(query, query_type) raise e def _parse_description_dimensions(self, response) -> Dict[str, Any]: """解析描述维度 返回格式: { "dimensions": ["维度1", "维度2", "维度3", ...], "count": 3 } """ dimensions = response.get("dimensions", []) return { "dimensions": dimensions, "count": len(dimensions) } def _parse_split_decision(self, response) -> Dict[str, Any]: """解析分割判断 返回格式: { "should_split": True/False, "split_rules": { # 如果should_split=True "strategy": "sentence/paragraph/grid", "params": {...} } } """ return { "should_split": response.get("should_split", False), "split_rules": response.get("split_rules", {}) } ``` **复用决策**: - **现有组件**: `tools/knowledge_retrieval_tools.py` - **决策**: **复用并修改** - **修改点**: 增强支持多种query_type，增加降级策略 --- #### 5.3.2 ImageSegmentTool（图片分割工具） **组件类型**: Tool (继承 BaseTool) **选择依据**: - **PRD依据**: 3.3.1节 "判断图片元素是否要继续分割拆解"、3.5节 - **CLAUDE.md依据**: 确定性模型调用 + 需要被Agent调用 **核心职责**: 1. 接收图片路径和子元素描述 2. 调用分割模型（SAM/GroundingDINO） 3. 保存分割后的子图片 4. 返回子图片路径列表 **关键点**: | 关键点 | 实现方式 | PRD依据 | |--------|---------|---------| | **支持多种分割模式** | grid（子图拼接）/ object（视觉元素） | 3.3.1节 | | **分割质量验证** | 验证mask是否覆盖元素 | 保证准确性 | | **文件管理** | 统一命名规范和存储路径 | 2.1.1节 | **代码框架**: ```python from tools.base import BaseTool from typing import List class ImageSegmentTool(BaseTool): """图片分割工具 PRD依据: 3.3.1节、3.5节 功能: 根据文字描述将图片分割为多个独立的视觉元素 """ name: str = "image_segment" description: str = "根据文字描述将图片分割为多个独立的视觉元素" def __init__(self, segment_model, output_dir: str): super().__init__() self.segment_model = segment_model self.output_dir = output_dir async def _arun( self, image_path: str, elements_description: List[str], segment_type: str = "object" ) -> List[str]: """ 分割图片 Args: image_path: 原始图片路径 elements_description: 子元素文字描述列表 segment_type: 分割类型 - object: 视觉元素分割（抠图） - grid: 子图拼接拆解（简单切割） Returns: 分割后的子图片路径列表 """ # 1. 加载图片 image = self._load_image(image_path) # 2. 根据分割类型选择策略 if segment_type == "grid": # 拼接图拆解（简单切割） sub_images = self._grid_split(image, len(elements_description)) else: # 视觉元素分割（调用分割模型） sub_images = await self._segment_objects(image, elements_description) # 3. 保存子图片 saved_paths = [] for idx, sub_img in enumerate(sub_images): save_path = f"{self.output_dir}/{self._generate_filename(image_path, idx)}" self._save_image(sub_img, save_path) saved_paths.append(save_path) return saved_paths def _grid_split(self, image, num_elements: int): """拼接图拆解 PRD依据: 3.3.1节（优先判断是否为多子图拼接） 支持: 2x2, 3x3, 1xN等常见拼接模式 """ # TODO: 根据num_elements推测拼接模式 pass async def _segment_objects(self, image, descriptions: List[str]): """视觉元素分割 调用SAM/GroundingDINO等模型 """ masks = await self.segment_model.segment(image, descriptions) return self._extract_objects(image, masks) ``` **复用决策**: - **现有组件**: `tools/segment_tools.py` - **决策**: **直接复用** - **理由**: 已实现图片分割功能，完全符合PRD需求 --- ### 5.4 Function组件详细设计 **CLAUDE.md原则**: > **Function组件特征**: 确定性任务 + 不需要被Agent调用 #### 5.4.1 TextSplitFunction（文本切分函数） **组件类型**: Function (继承 BaseFunction) **选择依据**: - **PRD依据**: 3.5节 "文本切分工具" - **CLAUDE.md依据**: 确定性字符串处理 + 工作流直接调用 **核心职责**: 1. 接收文本和切分规则 2. 按规则切分文本 3. 保证不重不漏（子文本合并=父文本） **关键点**: | 关键点 | 实现方式 | PRD依据 | |--------|---------|---------| | **充分必要原则** | 验证子文本合并后等于父文本 | 3.4节 | | **多种切分策略** | sentence/paragraph/semantic | 3.5节 | | **保留原文** | 不修改文本内容 | 2.2.2节 | **代码框架**: ```python from functions.base import BaseFunction from typing import List, Dict class TextSplitFunction(BaseFunction): """文本切分函数 PRD依据: 3.5节 功能: 根据切分规则切分文本 """ def invoke(self, text: str, split_rules: Dict) -> List[str]: """ 切分文本 Args: text: 待切分文本 split_rules: 切分规则 { "strategy": "sentence/paragraph/semantic", "params": {...} } Returns: 切分后的子文本列表 """ strategy = split_rules.get("strategy", "sentence") if strategy == "sentence": sub_texts = self._split_by_sentence(text) elif strategy == "paragraph": sub_texts = self._split_by_paragraph(text) elif strategy == "semantic": sub_texts = self._split_by_semantic(text, split_rules.get("params")) else: sub_texts = [text] # 验证充分必要原则（PRD 3.4节） self._validate_completeness(text, sub_texts) return sub_texts def _split_by_sentence(self, text: str) -> List[str]: """按句子切分""" import re sentences = re.split(r'[。！？\n]', text) return [s.strip() for s in sentences if s.strip()] def _validate_completeness(self, parent_text: str, child_texts: List[str]): """验证子文本合并后是否等于父文本 PRD依据: 3.4节 "充分必要原则" """ merged = ''.join(child_texts) if merged.replace(' ', '') != parent_text.replace(' ', ''): raise ValueError("子文本合并后不等于父文本，违反充分必要原则") ``` **复用决策**: - **现有组件**: 无 - **决策**: **完全新增** - **理由**: 项目中不存在文本切分功能 --- #### 5.4.2 ResultAggregationFunction（结果汇总函数） **组件类型**: Function (继承 BaseFunction) **选择依据**: - **PRD依据**: 2.2.2节 "输出格式定义" - **CLAUDE.md依据**: 确定性数据结构转换 **核心职责**: 1. 接收所有解构结果 2. 组装为树状JSON结构 3. 生成节点ID 4. 验证JSON格式 **关键点**: | 关键点 | 实现方式 | PRD依据 | |--------|---------|---------| | **递归构建树** | 递归函数处理children | 2.2.2节 | | **ID生成规则** | "1", "1_1", "1_1_2"格式 | 2.2.2节 | | **格式验证** | 验证所有必需字段存在 | 2.2.2节 | **代码框架**: ```python from functions.base import BaseFunction from typing import Dict, List class ResultAggregationFunction(BaseFunction): """结果汇总函数 PRD依据: 2.2.2节 功能: 将所有解构结果组装为树状JSON """ def invoke(self, state: Dict) -> Dict: """ 汇总所有解构结果 Args: state: { "post_summary": Dict, "image_results": List[Dict], "text_results": List[Dict] } Returns: 最终的树状JSON结构（PRD 2.2.2节格式） """ # 1. 提取各部分结果 post_summary = state["post_summary"] image_results = state["image_results"] text_results = state["text_results"] # 2. 构建元素列表 elements = [] # 添加图片元素 for idx, img_result in enumerate(image_results): element = self._build_element_tree( id=str(idx + 1), result=img_result ) elements.append(element) # 添加文本元素 text_id_start = len(image_results) + 1 for idx, text_result in enumerate(text_results): element = self._build_element_tree( id=str(text_id_start + idx), result=text_result ) elements.append(element) # 3. 组装最终JSON final_result = { "帖子总结": post_summary, "帖子包含元素": elements } # 4. 验证格式 self._validate_result(final_result) return final_result def _build_element_tree(self, id: str, result: Dict) -> Dict: """递归构建元素树 PRD依据: 2.2.2节 """ element = { "id": id, "what": result["what"], "描述": result["description"], "子节点元素关系": result.get("relationships", []), "元素重要性权重": result.get("importance_weight", 0.5), "子节点元素": [] } # 递归处理children if "children" in result and result["children"]: for idx, child in enumerate(result["children"]): child_element = self._build_element_tree( id=f"{id}_{idx+1}", result=child ) element["子节点元素"].append(child_element) return element def _validate_result(self, result: Dict): """验证JSON格式 PRD依据: 2.2.2节 检查: 必需字段、维度数量≥3等 """ # 验证帖子总结 assert "帖子总结" in result, "缺少'帖子总结'字段" assert len(result["帖子总结"]) >= 3, "帖子总结维度不足3个" # 验证帖子包含元素 assert "帖子包含元素" in result, "缺少'帖子包含元素'字段" # 递归验证每个元素节点 for element in result["帖子包含元素"]: self._validate_element(element) def _validate_element(self, element: Dict): """递归验证元素节点""" required_fields = ["id", "what", "描述", "子节点元素关系", "元素重要性权重", "子节点元素"] for field in required_fields: assert field in element, f"元素节点缺少'{field}'字段" # 验证描述维度数量 assert len(element["描述"]) >= 3, f"元素{element['id']}的描述维度不足3个" # 递归验证子节点 for child in element["子节点元素"]: self._validate_element(child) ``` **复用决策**: - **现有组件**: `functions/result_aggregation.py` - **决策**: **复用并修改** - **理由**: 可复用结果聚合逻辑，需新增树状结构构建 --- ## 6. 数据结构设计 ### 6.1 WhatDeconstructionState（工作流状态） **PRD依据**: 2.1节（输入）、2.2节（输出） ```python from typing import TypedDict, List, Dict, Any, Optional class WhatDeconstructionState(TypedDict): """What解构工作流状态 PRD依据: 2.1节、2.2节 """ # ========== 输入数据（PRD 2.1节） ========== images: List[str] # 图片路径列表（1-9张） text: Dict[str, Any] # {title, body, hashtags} comments: List[Dict[str, str]] # [{user, content}, ...] creator_info: Optional[Dict] # 创作者信息（可选） # ========== 中间状态 ========== category: str # 品类（F1输出） theme: str # 主题（F1输出） keywords: List[str] # 关键词（F1输出） consumer_highlights: List[Dict] # 消费者亮点（F4输出） # ========== 解构结果 ========== image_deconstruction_results: List[Dict] # 图片解构结果（F5输出） text_deconstruction_results: List[Dict] # 文本解构结果（F7输出） post_summary: Dict[str, Any] # 帖子总结（F9输出） # ========== 最终输出（PRD 2.2.2节） ========== final_result: Dict[str, Any] # 树状JSON结果 ``` ### 6.2 ElementNode（元素节点结构） **PRD依据**: 2.2.2节 "输出格式定义" ```python class ElementNode(TypedDict): """元素节点结构（递归树的节点） PRD依据: 2.2.2节 """ id: str # 节点ID（如"1", "1_1", "1_2_3"） what: str # 元素概述（文本原文 or 图片描述） 描述: Dict[str, Any] # 动态描述维度（≥3个） 子节点元素关系: List[str] # 子节点之间的关系描述 元素重要性权重: float # 0-1之间 子节点元素: List['ElementNode'] # 递归的子节点列表 ``` ### 6.3 FinalOutput（最终输出结构） **PRD依据**: 2.2.2节 ```python class FinalOutput(TypedDict): """最终输出的JSON结构 PRD依据: 2.2.2节 """ 帖子总结: Dict[str, Any] # 动态总结维度（≥3个） 帖子包含元素: List[ElementNode] # 元素树的根节点列表 ``` --- ## 7. 实现路线图 ### Phase 1: 基础框架搭建 **目标**: 建立工作流骨架和基础组件 | 任务 | 组件 | 优先级 | 依赖 | |------|------|--------|------| | 1.1 定义数据结构 | WhatDeconstructionState | P0 | 无 | | 1.2 实现PostUnderstandingAgent | Agent | P0 | 1.1 | | 1.3 修改KnowledgeRetrievalTool | Tool | P0 | 无 | | 1.4 搭建工作流骨架 | Workflow | P0 | 1.1, 1.2, 1.3 | | 1.5 端到端测试（简化版） | - | P0 | 1.4 | **验收标准**: - 工作流能够运行 - PostUnderstandingAgent能够输出品类/主题/关键词 - 知识库工具能够返回模拟数据 ### Phase 2: 核心解构能力 **目标**: 实现图片和文本的递归解构 | 任务 | 组件 | 优先级 | 依赖 | |------|------|--------|------| | 2.1 实现RecursiveImageDeconstructionAgent | Agent | P0 | Phase 1 | | 2.2 复用ImageSegmentTool | Tool | P0 | 2.1 | | 2.3 实现RecursiveTextDeconstructionAgent | Agent | P0 | Phase 1 | | 2.4 实现TextSplitFunction | Function | P0 | 2.3 | | 2.5 集成测试（单个图片+单个文本） | - | P0 | 2.1-2.4 | **验收标准**: - 能够递归解构单张图片（至少2层） - 能够递归解构单段文本（至少2层） - 递归深度控制生效（不超过10层） ### Phase 3: 增强功能 **目标**: 完成评论理解、帖子总结、结果汇总 | 任务 | 组件 | 优先级 | 依赖 | |------|------|--------|------| | 3.1 实现CommentAnalysisAgent | Agent | P1 | Phase 2 | | 3.2 实现PostSummaryDeconstructionAgent | Agent | P1 | Phase 2 | | 3.3 修改ResultAggregationFunction | Function | P0 | Phase 2 | | 3.4 完整端到端测试 | - | P0 | 3.1-3.3 | **验收标准**: - 能够分析评论并提取亮点 - 能够生成帖子整体总结 - 能够输出符合PRD 2.2.2节格式的完整JSON ### Phase 4: 完善优化 **目标**: 优化性能、容错、测试覆盖 | 任务 | 优先级 | |------|--------| | 4.1 递归深度优化（提前停止） | P2 | | 4.2 错误处理和降级策略 | P1 | | 4.3 并行执行优化 | P2 | | 4.4 单元测试覆盖 | P1 | | 4.5 性能基准测试 | P2 | **验收标准**: - 单元测试覆盖率 > 80% - 错误情况下有合理降级 - 性能满足预期（待定义） --- ## 8. 决策依据总结表 ### 8.1 工作流设计决策 | 决策点 | 选择 | PRD依据 | CLAUDE.md依据 | 技术理由 | |--------|------|---------|---------------|---------| | 工作流框架 | LangGraph | - | 要求使用LangGraph | 支持状态管理、节点编排 | | 并行执行 | 图片和文本并行 | 3.1节流程图 | 工作流负责编排 | 两者无依赖，可并行提升效率 | | 递归深度 | 最多10层 | 2.2.3节 | - | 防止无限递归，控制成本 | | 知识query生成 | 工作流逻辑 | 3.2节 | 轻量级逻辑不需组件 | 简单字符串拼接 | ### 8.2 组件类型选择决策 | 组件 | 类型 | PRD依据 | CLAUDE.md依据 | 选择理由 | |------|------|---------|---------------|---------| | PostUnderstandingAgent | Agent | 3.1步骤1 | 多模态理解（不确定性） | 需要LLM理解图文 | | CommentAnalysisAgent | Agent | 2.1.4节 | 语义理解（不确定性） | 需要LLM情感分析 | | RecursiveImageDeconstructionAgent | Agent (ReAct) | 3.3节 | 视觉理解+工具调用 | 需要多次调用知识库和分割工具 | | RecursiveTextDeconstructionAgent | Agent (ReAct) | 3.3节 | 语义理解+工具调用 | 需要多次调用知识库 | | PostSummaryDeconstructionAgent | Agent | 3.2节 | 综合分析（不确定性） | 需要LLM抽象提炼 | | KnowledgeRetrievalTool | Tool | 3.3.2节 | Agent需调用 | 确定性API调用 | | ImageSegmentTool | Tool | 3.3.1节 | Agent需调用 | 确定性模型调用 | | TextSplitFunction | Function | 3.5节 | 确定性+工作流调用 | 简单字符串处理 | | ResultAggregationFunction | Function | 2.2.2节 | 确定性+工作流调用 | 数据结构转换 | ### 8.3 复用决策 | 组件 | 决策 | 理由 | 修改点 | |------|------|------|--------| | knowledge_retrieval_tools.py | 复用并修改 | 已有基础查询功能 | 增强query类型支持、降级策略 | | segment_tools.py | 直接复用 | 功能完全匹配 | 无 | | result_aggregation.py | 复用并修改 | 可复用聚合逻辑 | 新增树状结构构建 | | TextSplitFunction | 完全新增 | 项目中无此功能 | - | | 所有Agent | 完全新增 | PRD业务逻辑全新 | - | --- ## 附录: Query模板速查表 **PRD依据**: 3.2节、3.3.2节 | Query类型 | 句式模板 | 使用场景 | |-----------|---------|---------| | 帖子整体总结维度 | 对于一篇主题为"{主题}"，品类为"{品类}"，关键词包含"{关键词}"的多模态社交媒体帖子，从内容创作者视角进行What要素的初步识别和分类，需要使用哪些通用工具？ | PostSummaryDeconstructionAgent | | 元素描述维度 | 刻画描述"{元素的what字段值}"核心特征的角度和维度有哪些？请尽可能不重不漏列举全。 | 所有递归解构Agent | | 分割判断 | 该节点"{元素的what字段值}"是否需要继续拆解分割？ | 所有递归解构Agent | | 图片整体分割工具 | 对一张描述为"{图片描述}"的图片，从内容创作者视角进行视觉元素分割，需要使用哪些图片分割/抠图工具？ | RecursiveImageDeconstructionAgent | | 文本段落分割工具 | 对一段描述为"{文本摘要}"的文本段落，从内容创作者视角进行结构化分割拆解，需要使用什么文本切分工具或文本结构化大模型？ | RecursiveTextDeconstructionAgent | --- **文档版本**: v2.0 **最后更新**: 2025-10-21 **适用PRD版本**: v1.3