tech_design_document_update.md 63 KB
What解构业务技术设计文档 v2.0
文档目标: 为开发者提供清晰、可执行的技术实现指导
设计原则: 每个决策都有明确的PRD依据和技术理由
目录
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节):
- 递归深度: 最多10层
- 停止条件:
- 已是不可再分的最小单元
- 达到10层深度
- 解构原则:
- 由宏观到微观、由整体到局部
- 子节点不重不漏（充分必要）
- 知识驱动: 描述维度、分割判断都由知识库动态提供
- 客观性: 严格基于帖子内容，禁止臆测
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 的核心职责：
- 编排执行顺序: 按PRD 3.1节流程图编排各组件
- 管理状态传递: 维护 WhatDeconstructionState，在各节点间传递数据
- 控制递归深度: 确保不超过10层
- 并行执行控制: 图片和文本解构并行执行
- 知识query生成: 根据中间结果动态生成知识库query
4.2 工作流流程图
PRD依据: 3.1节"总体解构流程"
graph TD
START([开始]) --> N1[节点1: 帖子初理解<br/>PostUnderstandingAgent]
N1 --> N2[节点2: 评论理解<br/>CommentAnalysisAgent]
N2 --> PARALLEL{并行分支}
PARALLEL --> N3[节点3: 图片递归解构<br/>RecursiveImageDeconstructionAgent]
PARALLEL --> N4[节点4: 文本递归解构<br/>RecursiveTextDeconstructionAgent]
N3 --> N5[节点5: 帖子整体解构<br/>PostSummaryDeconstructionAgent]
N4 --> N5
N5 --> N6[节点6: 结果汇总<br/>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 工作流核心代码框架
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"]
关键设计点:
- 并行执行:
image_deconstruction和text_deconstruction从comment_analysis同时出发 - 状态传递: 通过
WhatDeconstructionState在节点间传递数据 - 递归封装: 图片/文本的递归逻辑封装在各自的Agent内部
- 知识库调用: 在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依据: 需要多模态理解（图+文），属于不确定性任务
核心职责:
- 综合分析图片和文本内容
- 推理提取品类（如"时尚美妆"、"美食"等）
- 总结主题（如"夏日穿搭"）
- 提取关键词（如"OOTD"、"小白裙"）
关键点:
| 关键点 | 实现方式 | 依据 |
|---|---|---|
| 多模态输入 | 将图片和文本一起输入给LLM | LLM具备多模态理解能力 |
| 结构化输出 | 使用JSON格式约束LLM输出 | 需要提取固定字段 |
| 品类识别准确性 | 在prompt中提供常见品类列表 | 提升品类识别准确性 |
代码框架:
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依据: 需要情感分析和语义理解，属于不确定性任务
核心职责:
- 分析评论区文字内容
- 识别消费者关注的亮点（对应帖子的哪些元素）
- 提取高频关注点和情绪共鸣点
关键点:
| 关键点 | 实现方式 | 依据 |
|---|---|---|
| 亮点映射 | 将评论中的关注点映射到帖子具体元素 | PRD 2.1.4节要求 |
| 情感分析 | 识别评论的情感倾向（共鸣、兴趣、疑问等） | 理解消费者反应 |
| 高频提取 | 聚合多条评论的共同关注点 | 识别关键吸引点 |
代码框架:
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模式）
核心职责:
- 粗理解图片，生成what字段
- 调用知识库获取描述维度
- 细致理解图片，填充描述值
- 判断是否需要分割（调用知识库）
- 如需分割，调用图片分割工具
- 递归处理子元素（最多10层）
关键点:
| 关键点 | 实现方式 | PRD依据 |
|---|---|---|
| ReAct模式 | 思考→行动→观察循环 | 需要多次调用工具 |
| 递归控制 | 深度参数 + 停止条件判断 | 2.2.3节（最多10层） |
| 知识驱动 | 描述维度和分割判断都由知识库提供 | 3.3.2节 |
| 分割优先判断 | 优先判断是否为多子图拼接 | 3.3.1节 |
| what字段生成 | 对图片的文本描述 | 2.2.2节 |
| 判断不依赖分割 | 判断是否拆解基于完整原图+文字描述，不依赖分割结果 | PRD 3.3.1节原则 |
递归流程图:
graph TD
START([接收图片节点]) --> DEPTH{检查深度<br/>是否≥10层?}
DEPTH -->|是| STOP([返回结果<br/>不再递归])
DEPTH -->|否| ROUGH[粗理解图片<br/>基于完整原图+节点描述<br/>生成what字段]
ROUGH --> QUERY1[生成query:<br/>获取描述维度]
QUERY1 --> KB1[调用知识库]
KB1 --> DETAIL[细致理解图片<br/>基于完整原图+节点描述+维度<br/>填充描述维度值]
DETAIL --> QUERY2[生成query:<br/>是否需要分割?]
QUERY2 --> KB2[调用知识库<br/>基于完整原图+what+描述]
KB2 --> DECISION{知识库判断:<br/>是否分割?}
DECISION -->|否| STOP
DECISION -->|是| IDENTIFY[识别子元素<br/>基于完整原图<br/>生成子元素描述]
IDENTIFY --> SEGMENT[调用图片分割工具<br/>传入完整原图<br/>获得子图片列表]
SEGMENT --> RECURSE[遍历子图片<br/>递归调用自身<br/>传递完整原图]
RECURSE --> STOP
style ROUGH fill:#e1f5ff
style DETAIL fill:#e1f5ff
style KB1 fill:#fff3cd
style KB2 fill:#fff3cd
style SEGMENT fill:#d4edda
代码框架:
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语义理解 + 动态调用工具
核心职责:
- 提取文本原文作为what字段（不可改写）
- 调用知识库获取描述维度
- 分析文本，填充描述值
- 判断是否需要切分
- 如需切分，调用文本切分函数
- 递归处理子文本
关键点:
| 关键点 | 实现方式 | PRD依据 |
|---|---|---|
| what字段是原文 | 直接使用文本原文，不可总结改写 | 2.2.2节 |
| 充分必要原则 | 子文本合并后必须等于父文本 | 3.4节 |
| 递归控制 | 深度限制 + 最小语义单元判断 | 2.2.3节 |
代码框架:
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依据: 需要综合分析和抽象提炼能力
核心职责:
- 调用知识库获取总结维度
- 综合所有子元素解构结果
- 从点线面体维度总结帖子
关键点:
| 关键点 | 实现方式 | PRD依据 |
|---|---|---|
| 动态总结维度 | 由知识库根据品类/主题/关键词提供 | 2.2.2节、3.2节 |
| 综合分析 | 整合图片、文本、评论的解构结果 | 3.1节步骤13 |
| 维度数量 | 不少于3个维度 | 2.2.2节 |
代码框架:
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调用
核心职责:
- 接收query和query类型
- 调用外部知识库API
- 根据query类型解析返回结果
关键点:
| 关键点 | 实现方式 | PRD依据 |
|---|---|---|
| 支持多种query类型 | description/tool/split_decision | 3.3.2节 |
| 降级策略 | 知识库不可用时使用LLM默认知识 | 容错处理 |
| 结果格式化 | 转换为Agent可用的结构化数据 | 提升可用性 |
代码框架:
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调用
核心职责:
- 接收图片路径和子元素描述
- 调用分割模型（SAM/GroundingDINO）
- 保存分割后的子图片
- 返回子图片路径列表
关键点:
| 关键点 | 实现方式 | PRD依据 |
|---|---|---|
| 支持多种分割模式 | grid（子图拼接）/ object（视觉元素） | 3.3.1节 |
| 分割质量验证 | 验证mask是否覆盖元素 | 保证准确性 |
| 文件管理 | 统一命名规范和存储路径 | 2.1.1节 |
代码框架:
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依据: 确定性字符串处理 + 工作流直接调用
核心职责:
- 接收文本和切分规则
- 按规则切分文本
- 保证不重不漏（子文本合并=父文本）
关键点:
| 关键点 | 实现方式 | PRD依据 |
|---|---|---|
| 充分必要原则 | 验证子文本合并后等于父文本 | 3.4节 |
| 多种切分策略 | sentence/paragraph/semantic | 3.5节 |
| 保留原文 | 不修改文本内容 | 2.2.2节 |
代码框架:
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依据: 确定性数据结构转换
核心职责:
- 接收所有解构结果
- 组装为树状JSON结构
- 生成节点ID
- 验证JSON格式
关键点:
| 关键点 | 实现方式 | PRD依据 |
|---|---|---|
| 递归构建树 | 递归函数处理children | 2.2.2节 |
| ID生成规则 | "1", "1_1", "1_1_2"格式 | 2.2.2节 |
| 格式验证 | 验证所有必需字段存在 | 2.2.2节 |
代码框架:
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节（输出）
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节 "输出格式定义"
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节
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