tech_design_document.md 46 KB
What解构业务技术设计文档 v1.0
1. PRD目标解析
1.1 核心目标
针对小红书多模态内容（图文、视频、音频），从消费者视角进行充分必要的What要素逆向解构，通过层级化递归深入分析，识别和提取内容中所有构成成分"what"，并将这些要素结构化组织且直接关联原始多媒体素材。
1.2 核心价值
通过对大量内容的"What"要素进行聚合分析，提取爆款内容的内容特征与新兴趋势，为创作者提供可参考的、经过验证的成功要素组合，提升内容创作的确定性。
1.3 关键特征
- 递归深入：最多10层的递归解构
- 多模态：支持图片、视频、音频、文本
- 层级化：输出树状结构的元素关系
- 知识驱动：通过外部知识库动态获取解构维度
- 消费者视角：从内容消费者角度分析亮点和关键点
2. 功能需求拆解
2.1 核心功能清单
| 功能模块 | 功能描述 | 实现方式 |
|---|---|---|
| 帖子初理解 | 理解帖子整体，提取品类、主题、关键词 | Agent组件 |
| 知识检索 | 根据query动态获取解构所需的维度知识和工具知识 | Tool组件 |
| 评论理解 | 分析评论区内容，提取消费者关注的亮点 | Agent组件 |
| 图片元素解构 | 递归解构图片中的视觉元素 | Agent组件 |
| 图片分割 | 将图片分割为多个独立的视觉元素 | Tool组件 |
| 文本元素解构 | 递归解构标题、正文、话题标签 | Agent组件 |
| 文本切分 | 将文本按知识库指导切分为子元素 | Function组件 |
| 帖子整体解构 | 从点线面体维度解构帖子整体 | Agent组件 |
| 解构结果汇总 | 将所有解构结果组织为树状JSON结构 | Function组件 |
2.2 输入输出定义
2.2.1 系统输入
{
# 必选
"multimedia_content": {
"images": ["image_path1", "image_path2", ...], # 1-9张图片
"video": "video_path", # 可选
"audio": "audio_path", # 可选
"text": {
"title": "标题原文",
"body": "正文原文",
"hashtags": ["#标签1", "#标签2", ...]
},
"metadata": {
"resolution": "1920x1080",
"format": "jpg",
"duration": 120,
...
}
},
# 必选
"comments": [
{"user": "user1", "content": "评论内容1"},
{"user": "user2", "content": "评论内容2"},
...
],
# 可选
"creator_info": {
"nickname": "创作者昵称",
"avatar": "avatar_url",
"followers": 10000,
"category": "时尚美妆"
},
# 可选
"history_baseline": {
"recent_posts": [...], # 最近20-30个帖子
"style_features": {...}
}
}
2.2.2 系统输出
{
"帖子总结": {
"总结维度1": "值1", # 动态维度，由知识库决定
"总结维度2": "值2",
...
},
"帖子包含元素": [
{
"id": "1",
"what": "元素概述",
"描述": {
"描述维度1": "值1", # 动态维度，由知识库决定
"描述维度2": "值2",
...
},
"子节点元素关系": ["关系描述1", "关系描述2", ...],
"元素重要性权重": 0.7,
"子节点元素": [
{
"id": "1_1",
"what": "1_1元素概述",
"描述": {...},
"子节点元素关系": [...],
"元素重要性权重": 0.3,
"子节点元素": [...]
},
...
]
},
...
]
}
3. 架构设计
3.1 工作流 vs 组件划分原则
根据CLAUDE.md指导原则：
- 工作流：负责编排组件执行顺序和流程逻辑
- 组件：实现具体业务功能
3.1.1 工作流职责
- WhatDeconstructionWorkflow（主工作流）
- 编排整体解构流程
- 控制递归深度（最多10层）
- 管理状态传递
- 协调各Agent/Tool/Function的调用顺序
3.1.2 组件职责
根据任务特征选择组件类型：
- 不确定性任务 → Agent（需要LLM推理、多模态理解、语义分析）
- 确定性任务 + 需要Agent调用 → Tool
- 确定性任务 + 不需要Agent调用 → Function
4. 组件设计
4.1 Agent组件设计（核心业务逻辑）
4.1.1 PostUnderstandingAgent（帖子初理解Agent）
组件类型：Agent（BaseLLMAgent）
选择依据：
- PRD依据：3.1节 "帖子初理解，输出品类、主题、关键词"
- 需要LLM的语义理解能力
- 需要多模态理解能力（图文混合）
- 任务具有不确定性
功能定义：
- 输入：帖子的多模态内容（图片、文本）
- 输出：品类、主题、关键词列表
- 处理逻辑：综合分析图片视觉风格、文本语义、话题标签，推理帖子的核心主题
关键点：
- 多模态融合理解（图文结合）
- 提取结构化信息（品类、主题、关键词）
- 为后续知识检索提供query参数
代码框架：
class PostUnderstandingAgent(BaseLLMAgent):
"""帖子初理解Agent"""
async def ainvoke(self, state: WhatDeconstructionState) -> dict:
"""
分析帖子多模态内容，提取品类、主题、关键词
"""
# 1. 获取图片和文本内容
images = state.get("images")
text = state.get("text")
# 2. 构建多模态prompt
prompt = self._build_understanding_prompt(images, text)
# 3. 调用LLM进行理解
result = await self.llm.ainvoke(prompt)
# 4. 解析返回的品类、主题、关键词
parsed_result = self._parse_result(result)
return {
"category": parsed_result["category"],
"theme": parsed_result["theme"],
"keywords": parsed_result["keywords"]
}
4.1.2 CommentAnalysisAgent（评论理解Agent）
组件类型：Agent（BaseLLMAgent）
选择依据：
- PRD依据：2.1.4节 "评论信息用于提取内容消费者关注的亮点"
- 需要LLM的语义理解和情感分析能力
- 需要识别隐含的消费者关注点
- 任务具有不确定性
功能定义：
- 输入：评论区文字内容列表
- 输出：消费者关注的亮点列表（对应帖子的哪些元素）
- 处理逻辑：分析评论中的高频词、情感倾向、提及的具体元素
关键点：
- 情感分析（共鸣点、兴趣点）
- 关键点提取（评论中提到的具体内容元素）
- 聚合分析（多条评论的共同关注点）
- 与帖子内容的映射关系
代码框架：
class CommentAnalysisAgent(BaseLLMAgent):
"""评论理解Agent"""
async def ainvoke(self, state: WhatDeconstructionState) -> dict:
"""
分析评论区内容，提取消费者关注的亮点
"""
# 1. 获取评论数据
comments = state.get("comments")
post_content = state.get("text")
# 2. 构建分析prompt
prompt = self._build_comment_analysis_prompt(comments, post_content)
# 3. 调用LLM分析
result = await self.llm.ainvoke(prompt)
# 4. 提取亮点并与帖子内容映射
highlights = self._extract_highlights(result)
return {
"consumer_highlights": highlights
}
4.1.3 RecursiveImageDeconstructionAgent（图片递归解构Agent）
组件类型：Agent（BaseReactAgent）
选择依据：
- PRD依据：3.3节 "对单一节点的递归解构"
- 需要LLM的视觉理解能力
- 需要动态调用知识库和图片分割工具（ReAct模式）
- 需要递归判断是否继续分割
- 任务具有高度不确定性
功能定义：
- 输入：当前图片节点、父节点信息、递归深度
- 输出：图片的what描述、描述维度、子元素列表
- 处理逻辑：
- 粗理解图片，生成what字段值
- 根据what值生成query，调用知识库获取描述维度
- 根据描述维度细致理解图片，填充描述值
- 判断是否需要继续分割（调用知识库）
- 如需分割，调用图片分割工具，递归处理子元素
关键点：
- ReAct模式：思考→行动→观察的循环（调用知识库、调用分割工具）
- 递归控制：深度限制（10层）、停止条件判断
- 知识库交互：动态生成query获取解构维度
- 图片分割判断：优先判断是否为多子图拼接
- 多模态理解：将视觉信息转化为文字描述
代码框架：
class RecursiveImageDeconstructionAgent(BaseReactAgent):
"""图片递归解构Agent"""
def __init__(self, llm, tools: List[BaseTool], max_depth: int = 10):
super().__init__(llm, tools)
self.max_depth = max_depth
self.knowledge_retrieval_tool = None # 知识检索工具
self.image_segment_tool = None # 图片分割工具
async def ainvoke(self, state: WhatDeconstructionState) -> dict:
"""
递归解构图片元素
"""
# 1. 获取当前节点信息
current_image = state.get("current_image")
parent_id = state.get("parent_id", "")
depth = state.get("depth", 0)
# 检查递归深度
if depth >= self.max_depth:
return {"continue_decompose": False}
# 2. 粗理解图片（生成what字段）
what_value = await self._rough_understanding(current_image)
# 3. 生成query，获取描述维度
query = self._generate_description_query(what_value)
description_dimensions = await self.knowledge_retrieval_tool.ainvoke(query)
# 4. 根据描述维度细致理解图片
description_values = await self._detailed_understanding(
current_image, description_dimensions
)
# 5. 判断是否需要继续分割
split_query = self._generate_split_query(what_value, description_values)
should_split = await self.knowledge_retrieval_tool.ainvoke(split_query)
result = {
"what": what_value,
"description": description_values,
"continue_decompose": should_split
}
# 6. 如需分割，执行分割并递归
if should_split:
# 判断是否为多子图拼接
sub_elements = await self._identify_sub_elements(current_image)
if sub_elements:
# 调用图片分割工具
segmented_images = await self.image_segment_tool.ainvoke(
image=current_image,
elements_description=sub_elements
)
# 递归处理子元素
children = []
for idx, sub_img in enumerate(segmented_images):
child_state = {
"current_image": sub_img,
"parent_id": f"{parent_id}_{idx+1}",
"depth": depth + 1
}
child_result = await self.ainvoke(child_state)
children.append(child_result)
result["children"] = children
return result
4.1.4 RecursiveTextDeconstructionAgent（文本递归解构Agent）
组件类型：Agent（BaseReactAgent）
选择依据：
- PRD依据：3.3节 "对单一节点的递归解构"
- 需要LLM的文本语义理解能力
- 需要动态调用知识库和文本切分工具
- 需要递归判断是否继续切分
- 任务具有不确定性
功能定义：
- 输入：当前文本节点（标题/正文/标签）、父节点信息、递归深度
- 输出：文本的what描述、描述维度、子元素列表
- 处理逻辑：与图片解构类似，但针对文本元素
关键点：
- 文本原文保留：当节点是文本时，what字段必须是原文，不可改写
- 语义切分：根据知识库指导切分文本结构
- 递归控制：深度限制（10层）
- 充分必要原则：子节点文本必须完整覆盖父节点文本
代码框架：
class RecursiveTextDeconstructionAgent(BaseReactAgent):
"""文本递归解构Agent"""
def __init__(self, llm, tools: List[BaseTool], max_depth: int = 10):
super().__init__(llm, tools)
self.max_depth = max_depth
self.knowledge_retrieval_tool = None
self.text_split_tool = None
async def ainvoke(self, state: WhatDeconstructionState) -> dict:
"""
递归解构文本元素
"""
# 1. 获取当前节点信息
current_text = state.get("current_text")
text_type = state.get("text_type") # title/body/hashtags
parent_id = state.get("parent_id", "")
depth = state.get("depth", 0)
if depth >= self.max_depth:
return {"continue_decompose": False}
# 2. what字段直接使用原文
what_value = current_text
# 3. 生成query，获取描述维度
query = self._generate_description_query(what_value, text_type)
description_dimensions = await self.knowledge_retrieval_tool.ainvoke(query)
# 4. 根据描述维度分析文本
description_values = await self._analyze_text(
current_text, description_dimensions
)
# 5. 判断是否需要继续切分
split_query = self._generate_split_query(what_value, description_values)
should_split = await self.knowledge_retrieval_tool.ainvoke(split_query)
result = {
"what": what_value,
"description": description_values,
"continue_decompose": should_split
}
# 6. 如需切分，执行切分并递归
if should_split:
# 根据知识库指导切分文本
split_result = await self.text_split_tool.ainvoke(
text=current_text,
split_guidance=description_dimensions
)
# 递归处理子元素
children = []
for idx, sub_text in enumerate(split_result):
child_state = {
"current_text": sub_text,
"text_type": text_type,
"parent_id": f"{parent_id}_{idx+1}",
"depth": depth + 1
}
child_result = await self.ainvoke(child_state)
children.append(child_result)
result["children"] = children
return result
4.1.5 PostSummaryDeconstructionAgent（帖子整体解构Agent）
组件类型：Agent（BaseLLMAgent）
选择依据：
- PRD依据：3.2节 "帖子整体解构环节"
- 需要LLM的综合理解能力
- 需要从点线面体维度抽象总结
- 任务具有不确定性
功能定义：
- 输入：帖子的所有元素解构结果、知识库返回的总结维度
- 输出：帖子总结对象（动态维度）
- 处理逻辑：综合所有子元素的解构结果，从宏观维度总结帖子特征
关键点：
- 动态总结维度（由知识库决定）
- 综合多模态元素的解构结果
- 从整体视角抽象提炼
- 包含消费者亮点维度
代码框架：
class PostSummaryDeconstructionAgent(BaseLLMAgent):
"""帖子整体解构Agent"""
async def ainvoke(self, state: WhatDeconstructionState) -> dict:
"""
从点线面体维度解构帖子整体
"""
# 1. 获取所有子元素解构结果
image_results = state.get("image_deconstruction_results")
text_results = state.get("text_deconstruction_results")
consumer_highlights = state.get("consumer_highlights")
# 2. 获取总结维度（通过知识库）
category = state.get("category")
theme = state.get("theme")
keywords = state.get("keywords")
query = self._generate_summary_query(category, theme, keywords)
summary_dimensions = await self._call_knowledge_retrieval(query)
# 3. 根据总结维度综合分析
prompt = self._build_summary_prompt(
image_results, text_results, consumer_highlights, summary_dimensions
)
result = await self.llm.ainvoke(prompt)
# 4. 解析总结结果
summary = self._parse_summary(result, summary_dimensions)
return {
"post_summary": summary
}
def _generate_summary_query(self, category, theme, keywords):
"""生成总结维度query"""
return f"""对于一篇主题为"{theme}"，品类为"{category}"，关键词包含"{', '.join(keywords)}"的多模态社交媒体帖子，从内容创作者视角进行What要素的初步识别和分类，需要使用哪些通用工具？"""
4.2 Tool组件设计（可被Agent调用）
4.2.1 KnowledgeRetrievalTool（知识检索工具）
组件类型：Tool（BaseTool）
选择依据：
- PRD依据：3.2节、3.3.2节"对单一节点解构如何获取知识"
- 需要被Agent调用（各解构Agent都需要调用）
- 任务确定性（API调用）
功能定义：
- 输入：query字符串
- 输出：知识库返回的结果（描述维度、工具推荐、判断结果等）
- 处理逻辑：调用外部知识库API，解析返回结果
关键点：
- 支持多种query类型（描述维度、工具知识、分割判断）
- 结果格式化（转换为Agent可用的结构化数据）
- 错误处理和降级（知识库不可用时使用LLM默认知识）
代码框架：
from tools.base import BaseTool
from typing import Dict, Any
class KnowledgeRetrievalTool(BaseTool):
"""知识检索工具"""
name: str = "knowledge_retrieval"
description: str = "根据query检索What解构所需的知识（描述维度、工具推荐、分割判断等）"
def __init__(self, knowledge_base_url: str):
super().__init__()
self.knowledge_base_url = knowledge_base_url
async def _arun(self, query: str, query_type: str = "description") -> Dict[str, Any]:
"""
调用知识库检索
Args:
query: 查询语句
query_type: 查询类型（description/tool/split_decision）
"""
# 1. 调用知识库API
response = await self._call_knowledge_base(query)
# 2. 根据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
def _parse_description_dimensions(self, response) -> Dict[str, Any]:
"""解析描述维度"""
return {
"dimensions": response.get("dimensions", []),
"count": len(response.get("dimensions", []))
}
复用决策：
- 现有组件：
tools/knowledge_retrieval_tools.py已存在知识检索工具 - 决策：复用并修改
- 理由：现有工具实现了基本的知识库查询功能，但需要增强支持本PRD中的多种query类型
4.2.2 ImageSegmentTool（图片分割工具）
组件类型：Tool（BaseTool）
选择依据：
- PRD依据：3.3.1节 "判断图片元素是否要继续分割拆解"
- 需要被Agent调用（RecursiveImageDeconstructionAgent调用）
- 任务确定性（调用分割模型）
功能定义：
- 输入：图片路径、子元素文字描述列表
- 输出：分割后的子图片路径列表
- 处理逻辑：调用图片分割模型（如SAM、GroundingDINO等），根据文字描述分割图片
关键点：
- 支持多种分割模式（完整子图拼接拆解、视觉元素抠图）
- 保存分割结果并返回路径
- 分割质量验证
代码框架：
from tools.base import BaseTool
from typing import List
class ImageSegmentTool(BaseTool):
"""图片分割工具"""
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" # object/grid
) -> 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, 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(idx)}"
self._save_image(sub_img, save_path)
saved_paths.append(save_path)
return saved_paths
async def _segment_objects(self, image, descriptions):
"""调用分割模型分割视觉元素"""
# 调用SAM/GroundingDINO等模型
masks = await self.segment_model.segment(image, descriptions)
return self._extract_objects(image, masks)
复用决策：
- 现有组件：
tools/segment_tools.py已存在图片分割工具 - 决策：直接复用
- 理由：现有工具已实现"根据对象名称对图像进行分割，抽取指定对象的切片"功能，完全符合本PRD需求
4.3 Function组件设计（不可被Agent调用）
4.3.1 TextSplitFunction（文本切分函数）
组件类型：Function（BaseFunction）
选择依据：
- PRD依据：3.5节 "文本切分工具"
- 不需要被Agent直接调用（在工作流中调用）
- 任务确定性（字符串处理）
功能定义：
- 输入：文本字符串、切分规则（由知识库返回）
- 输出：切分后的子文本列表
- 处理逻辑：根据切分规则（句子、段落、语义块）切分文本
关键点：
- 多种切分策略（按句、按段、按语义）
- 保留原文完整性（子文本合并后等于父文本）
- 不重复不遗漏
代码框架：
from functions.base import BaseFunction
from typing import List, Dict
class TextSplitFunction(BaseFunction):
"""文本切分函数"""
def __init__(self):
super().__init__()
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":
return self._split_by_sentence(text)
elif strategy == "paragraph":
return self._split_by_paragraph(text)
elif strategy == "semantic":
return self._split_by_semantic(text, split_rules.get("params"))
return [text]
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 _split_by_paragraph(self, text: str) -> List[str]:
"""按段落切分"""
paragraphs = text.split('\n\n')
return [p.strip() for p in paragraphs if p.strip()]
def _split_by_semantic(self, text: str, params: Dict) -> List[str]:
"""按语义块切分（需要语义模型）"""
# TODO: 实现语义切分逻辑
pass
复用决策：
- 现有组件：无
- 决策：完全新增
- 理由：项目中不存在文本切分功能，需要从BaseFunction新建
4.3.2 ResultAggregationFunction（结果汇总函数）
组件类型：Function（BaseFunction）
选择依据：
- 任务确定性（数据结构转换）
- 不需要被Agent调用（工作流最后环节）
功能定义：
- 输入：各Agent的解构结果
- 输出：最终的树状JSON结构
- 处理逻辑：将分散的解构结果组装为符合PRD格式的JSON
关键点：
- 递归组装树状结构
- 字段格式验证
- ID生成规则
- 元素重要性权重计算
代码框架：
from functions.base import BaseFunction
from typing import Dict, List
class ResultAggregationFunction(BaseFunction):
"""结果汇总函数"""
def invoke(self, state: Dict) -> Dict:
"""
汇总所有解构结果为树状JSON
Args:
state: 工作流状态，包含所有解构结果
Returns:
最终的树状JSON结构
"""
# 1. 提取各部分结果
post_summary = state.get("post_summary")
image_results = state.get("image_deconstruction_results")
text_results = state.get("text_deconstruction_results")
# 2. 构建树状结构
elements = []
# 添加图片元素
for idx, img_result in enumerate(image_results):
element = self._build_element_tree(
id=f"{idx+1}",
result=img_result,
element_type="image"
)
elements.append(element)
# 添加文本元素
text_id_start = len(image_results) + 1
for idx, text_result in enumerate(text_results):
element = self._build_element_tree(
id=f"{text_id_start+idx}",
result=text_result,
element_type="text"
)
elements.append(element)
# 3. 组装最终JSON
final_result = {
"帖子总结": post_summary,
"帖子包含元素": elements
}
# 4. 验证JSON格式
self._validate_result(final_result)
return final_result
def _build_element_tree(self, id: str, result: Dict, element_type: str) -> Dict:
"""递归构建元素树"""
element = {
"id": id,
"what": result["what"],
"描述": result["description"],
"子节点元素关系": result.get("relationships", []),
"元素重要性权重": result.get("importance_weight", 0.5),
"子节点元素": []
}
# 递归处理子节点
if "children" in result:
for idx, child in enumerate(result["children"]):
child_element = self._build_element_tree(
id=f"{id}_{idx+1}",
result=child,
element_type=element_type
)
element["子节点元素"].append(child_element)
return element
复用决策：
- 现有组件：
functions/json_utils.py存在JSON处理函数 - 决策：复用并修改
- 理由：可复用JSON格式化和验证功能，但需新增树状结构构建逻辑
5. 工作流设计
5.1 WhatDeconstructionWorkflow（主工作流）
继承基类：BaseGraphAgent（LangGraph）
状态定义：
from typing import TypedDict, List, Dict, Any
class WhatDeconstructionState(TypedDict):
# 输入数据
images: List[str]
video: str
audio: str
text: Dict[str, Any] # {title, body, hashtags}
comments: List[Dict[str, str]]
creator_info: Dict[str, Any]
# 中间状态
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]
# 递归控制
current_depth: int
max_depth: int
节点定义：
nodes = {
"post_understanding": PostUnderstandingAgent,
"comment_analysis": CommentAnalysisAgent,
"image_deconstruction": RecursiveImageDeconstructionAgent,
"text_deconstruction": RecursiveTextDeconstructionAgent,
"post_summary": PostSummaryDeconstructionAgent,
"result_aggregation": ResultAggregationFunction
}
流程图：
graph TD
START([开始]) --> A[帖子初理解<br/>PostUnderstandingAgent]
A --> B[评论理解<br/>CommentAnalysisAgent]
B --> C{并行解构}
C --> D[图片递归解构<br/>RecursiveImageDeconstructionAgent]
C --> E[文本递归解构<br/>RecursiveTextDeconstructionAgent]
D --> F[帖子整体解构<br/>PostSummaryDeconstructionAgent]
E --> F
F --> G[结果汇总<br/>ResultAggregationFunction]
G --> END([结束])
核心代码：
from langgraph.graph import StateGraph
from agents.base import BaseGraphAgent
class WhatDeconstructionWorkflow(BaseGraphAgent):
"""What解构主工作流"""
def __init__(self, llm, tools: List[BaseTool]):
super().__init__()
self.llm = llm
self.tools = tools
# 初始化所有组件
self.post_understanding_agent = PostUnderstandingAgent(llm)
self.comment_analysis_agent = CommentAnalysisAgent(llm)
self.image_deconstruction_agent = RecursiveImageDeconstructionAgent(
llm, tools, max_depth=10
)
self.text_deconstruction_agent = RecursiveTextDeconstructionAgent(
llm, tools, max_depth=10
)
self.post_summary_agent = PostSummaryDeconstructionAgent(llm)
self.result_aggregation_func = ResultAggregationFunction()
# 构建工作流图
self.graph = self._build_graph()
def _build_graph(self) -> StateGraph:
"""构建LangGraph工作流"""
workflow = StateGraph(WhatDeconstructionState)
# 添加节点
workflow.add_node("post_understanding", self._post_understanding_node)
workflow.add_node("comment_analysis", self._comment_analysis_node)
workflow.add_node("image_deconstruction", self._image_deconstruction_node)
workflow.add_node("text_deconstruction", self._text_deconstruction_node)
workflow.add_node("post_summary", self._post_summary_node)
workflow.add_node("result_aggregation", self._result_aggregation_node)
# 定义边（流程顺序）
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.set_finish_point("result_aggregation")
return workflow.compile()
async def _post_understanding_node(self, state: WhatDeconstructionState):
"""节点：帖子初理解"""
result = await self.post_understanding_agent.ainvoke(state)
return {
"category": result["category"],
"theme": result["theme"],
"keywords": result["keywords"]
}
async def _comment_analysis_node(self, state: WhatDeconstructionState):
"""节点：评论理解"""
result = await self.comment_analysis_agent.ainvoke(state)
return {
"consumer_highlights": result["consumer_highlights"]
}
async def _image_deconstruction_node(self, state: WhatDeconstructionState):
"""节点：图片递归解构"""
images = state.get("images", [])
results = []
for idx, image_path in enumerate(images):
img_state = {
"current_image": image_path,
"parent_id": f"image_{idx+1}",
"depth": 0,
"max_depth": 10,
"category": state["category"],
"theme": state["theme"]
}
result = await self.image_deconstruction_agent.ainvoke(img_state)
results.append(result)
return {"image_deconstruction_results": results}
async def _text_deconstruction_node(self, state: WhatDeconstructionState):
"""节点：文本递归解构"""
text_data = state.get("text", {})
results = []
# 解构标题
if text_data.get("title"):
title_state = {
"current_text": text_data["title"],
"text_type": "title",
"parent_id": "title",
"depth": 0,
"max_depth": 10
}
title_result = await self.text_deconstruction_agent.ainvoke(title_state)
results.append(title_result)
# 解构正文
if text_data.get("body"):
body_state = {
"current_text": text_data["body"],
"text_type": "body",
"parent_id": "body",
"depth": 0,
"max_depth": 10
}
body_result = await self.text_deconstruction_agent.ainvoke(body_state)
results.append(body_result)
# 解构话题标签
if text_data.get("hashtags"):
hashtags_text = " ".join(text_data["hashtags"])
hashtags_state = {
"current_text": hashtags_text,
"text_type": "hashtags",
"parent_id": "hashtags",
"depth": 0,
"max_depth": 10
}
hashtags_result = await self.text_deconstruction_agent.ainvoke(hashtags_state)
results.append(hashtags_result)
return {"text_deconstruction_results": results}
async def _post_summary_node(self, state: WhatDeconstructionState):
"""节点：帖子整体解构"""
result = await self.post_summary_agent.ainvoke(state)
return {"post_summary": result["post_summary"]}
async def _result_aggregation_node(self, state: WhatDeconstructionState):
"""节点：结果汇总"""
final_result = self.result_aggregation_func.invoke(state)
return {"final_result": final_result}
async def ainvoke(self, input_data: Dict) -> Dict:
"""执行工作流"""
# 初始化状态
initial_state = WhatDeconstructionState(
images=input_data.get("multimedia_content", {}).get("images", []),
video=input_data.get("multimedia_content", {}).get("video", ""),
audio=input_data.get("multimedia_content", {}).get("audio", ""),
text=input_data.get("multimedia_content", {}).get("text", {}),
comments=input_data.get("comments", []),
creator_info=input_data.get("creator_info", {}),
current_depth=0,
max_depth=10
)
# 执行工作流
result = await self.graph.ainvoke(initial_state)
return result["final_result"]
6. 数据结构设计
6.1 WhatDeconstructionState（工作流状态）
from typing import TypedDict, List, Dict, Any, Optional
class WhatDeconstructionState(TypedDict):
"""What解构工作流状态"""
# ========== 输入数据 ==========
images: List[str] # 图片路径列表（1-9张）
video: Optional[str] # 视频路径
audio: Optional[str] # 音频路径
text: Dict[str, Any] # {title, body, hashtags}
comments: List[Dict[str, str]] # [{user, content}, ...]
creator_info: Optional[Dict[str, Any]] # 创作者信息（可选）
history_baseline: Optional[Dict[str, Any]] # 历史基准（可选）
# ========== 中间状态 ==========
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] # 最终的树状JSON结果
# ========== 递归控制 ==========
current_depth: int # 当前递归深度
max_depth: int # 最大递归深度（默认10）
6.2 ElementNode（元素节点结构）
class ElementNode(TypedDict):
"""元素节点结构（树状结构的节点）"""
id: str # 节点ID（如 "1", "1_1", "1_1_2"）
what: str # 元素概述（文本原文 or 图片描述）
描述: Dict[str, Any] # 动态描述维度（由知识库决定）
子节点元素关系: List[str] # 子节点之间的关系描述
元素重要性权重: float # 0-1之间的权重值
子节点元素: List['ElementNode'] # 递归的子节点列表
图片链接: str # 图片分解后的链接
6.3 FinalOutput（最终输出结构）
class FinalOutput(TypedDict):
"""最终输出的JSON结构"""
帖子总结: Dict[str, Any] # 动态总结维度（由知识库决定）
帖子包含元素: List[ElementNode] # 元素树的根节点列表
7. 技术决策依据总结
7.1 工作流设计决策
| 决策点 | 选择 | PRD依据 | 技术理由 |
|---|---|---|---|
| 工作流框架 | LangGraph | CLAUDE.md要求 | 支持状态管理、节点编排、递归控制 |
| 并行解构 | 图片和文本并行 | 3.1节流程图 | 提升效率，两者无依赖关系 |
| 递归深度 | 最多10层 | 2.2.3节 | 防止无限递归，控制计算成本 |
7.2 Agent组件决策
| Agent | 选择原因 | PRD依据 | 关键能力 |
|---|---|---|---|
| PostUnderstandingAgent | 多模态理解+语义推理 | 3.1节 | 提取品类/主题/关键词 |
| CommentAnalysisAgent | 情感分析+语义理解 | 2.1.4节 | 识别消费者关注点 |
| RecursiveImageDeconstructionAgent | ReAct模式+递归控制 | 3.3节+3.3.1节 | 动态调用工具+递归分割 |
| RecursiveTextDeconstructionAgent | ReAct模式+递归控制 | 3.3节 | 动态调用工具+递归切分 |
| PostSummaryDeconstructionAgent | 综合分析+抽象提炼 | 3.2节 | 点线面体维度总结 |
7.3 Tool组件决策
| Tool | 选择原因 | PRD依据 | 关键能力 |
|---|---|---|---|
| KnowledgeRetrievalTool | Agent需调用+确定性任务 | 3.2节+3.3.2节 | 知识库API调用 |
| ImageSegmentTool | Agent需调用+确定性任务 | 3.3.1节+3.5节 | 图片分割模型调用 |
7.4 Function组件决策
| Function | 选择原因 | PRD依据 | 关键能力 |
|---|---|---|---|
| TextSplitFunction | 不需Agent调用+确定性任务 | 3.5节 | 文本切分逻辑 |
| ResultAggregationFunction | 不需Agent调用+确定性任务 | 2.2.2节 | JSON结构组装 |
7.5 复用决策
| 组件 | 决策 | 理由 |
|---|---|---|
| knowledge_retrieval_tools.py | 复用并修改 | 已有基础查询功能，需增强query类型支持 |
| segment_tools.py | 直接复用 | 功能完全匹配PRD需求 |
| json_utils.py | 复用并修改 | 可复用JSON处理，需新增树状结构构建 |
| TextSplitFunction | 完全新增 | 项目中无此功能 |
| 所有Agent | 完全新增 | PRD业务逻辑全新 |
8. 实现优先级
Phase 1: 基础框架（核心路径）
- WhatDeconstructionState - 定义工作流状态
- PostUnderstandingAgent - 帖子初理解
- KnowledgeRetrievalTool（复用修改）- 知识检索
- WhatDeconstructionWorkflow - 主工作流骨架
Phase 2: 核心解构能力
- RecursiveImageDeconstructionAgent - 图片递归解构
- ImageSegmentTool（直接复用）- 图片分割
- RecursiveTextDeconstructionAgent - 文本递归解构
- TextSplitFunction（新增）- 文本切分
Phase 3: 增强功能
- CommentAnalysisAgent - 评论理解
- PostSummaryDeconstructionAgent - 帖子整体解构
- ResultAggregationFunction（复用修改）- 结果汇总
Phase 4: 完善优化
- 递归深度控制优化
- 错误处理和降级策略
- 性能优化和并行化
- 单元测试和集成测试
9. 关键技术挑战
9.1 递归控制
挑战：如何准确判断是否停止递归？ 方案：
- 依赖知识库返回的判断结果
- 设置硬性深度限制（10层）
- 最小元素判断（不可再分的最小语义/视觉单元）
9.2 知识库Query设计
挑战：如何动态生成有效的query？ 方案：
- 固定句式模板 + 动态内容填充
- 多种query类型（描述维度、工具推荐、分割判断）
- 降级策略（知识库不可用时使用LLM默认知识）
9.3 图片分割准确性
挑战：如何准确分割视觉元素？ 方案：
- 优先判断是否为多子图拼接（简单切割）
- 使用成熟的分割模型（SAM、GroundingDINO）
- LLM生成精准的元素描述文本
9.4 不重复不遗漏原则
挑战：如何保证子节点完整覆盖父节点？ 方案：
- 文本：子节点文本合并后必须等于父节点文本
- 图片：子元素的掩码合并后必须覆盖父图片区域
- LLM验证：最后由LLM验证是否符合"充分必要"原则
10. 附录：Query模板汇总
10.1 帖子整体解构Query
对于一篇主题为"{帖子主题}"，品类为"{帖子品类}"，关键词包含"{帖子关键词列表}"的多模态社交媒体帖子，从内容创作者视角进行What要素的初步识别和分类，需要使用哪些通用工具？
10.2 描述维度获取Query
刻画描述"{元素的what字段值}"核心特征的角度和维度有哪些？请尽可能不重不漏列举全。
10.3 分割判断Query
该节点"{元素的what字段值}"是否需要继续拆解分割？
10.4 图片整体分割工具Query
对一张描述为"{图片整体内容的简要描述}"的图片，从内容创作者视角进行视觉元素分割，需要使用哪些图片分割/抠图工具？
10.5 图片亮点提取工具Query
从内容创作者视角，对一张描述为"{图片整体内容的简要描述}"的图片，进行亮点或关键点的提取，需要使用什么图片分析工具或大模型视觉理解工具？
10.6 图片特定元素分割工具Query
对一张"{XXX图片整体描述}"的图片中的"{YYY具体视觉元素描述}"的视觉元素，从内容创作者视角进行细致分割拆解，需要使用哪些高精度图片分割/抠图工具或细粒度图像理解大模型？
10.7 图片特定元素亮点提取Query
从内容创作者视角，对一张"{XXX图片整体描述}"的图片中的"{YYY具体视觉元素描述}"的视觉元素，进行亮点或关键点的提取，需要使用什么图像特征提取工具或专业视觉大模型？
10.8 文本段落分割工具Query
对一段描述为"{文本段落内容摘要}"的文本段落，从内容创作者视角进行结构化分割拆解，需要使用什么文本切分工具或文本结构化大模型？
10.9 文本段落亮点提取Query
从内容创作者视角，对一段描述为"{文本段落内容摘要}"的文本段落（包含标题、正文、话题标签），进行亮点或关键点的提取，需要使用什么情感分析工具、语义理解工具或文本摘要大模型？
11. 文件结构规划
project/
├── workflows/
│ └── what_deconstruction_workflow.py # 主工作流（新增）
├── agents/
│ ├── post_understanding_agent.py # 帖子初理解Agent（新增）
│ ├── comment_analysis_agent.py # 评论理解Agent（新增）
│ ├── recursive_image_deconstruction_agent.py # 图片递归解构Agent（新增）
│ ├── recursive_text_deconstruction_agent.py # 文本递归解构Agent（新增）
│ └── post_summary_deconstruction_agent.py # 帖子整体解构Agent（新增）
├── tools/
│ ├── knowledge_retrieval_tools.py # 知识检索工具（复用修改）
│ └── segment_tools.py # 图片分割工具（直接复用）
├── functions/
│ ├── text_split_function.py # 文本切分函数（新增）
│ └── result_aggregation_function.py # 结果汇总函数（复用修改）
├── states/
│ └── what_deconstruction_state.py # 工作流状态定义（新增）
└── tests/
├── test_agents/
├── test_tools/
├── test_functions/
└── test_workflow/
12. 总结
本技术设计文档基于PRD v1.2，完整设计了What解构业务的技术实现方案：
12.1 核心设计原则
- 组件化：Agent/Tool/Function清晰分工
- 递归解构：层级化深入分析（最多10层）
- 知识驱动：动态从知识库获取解构维度
- 充分必要：子节点不重不漏覆盖父节点
12.2 技术架构
- 1个工作流：WhatDeconstructionWorkflow（LangGraph）
- 5个Agent：帖子初理解、评论理解、图片递归解构、文本递归解构、帖子整体解构
- 2个Tool：知识检索、图片分割
- 2个Function：文本切分、结果汇总
12.3 复用策略
- 直接复用：segment_tools.py
- 复用修改：knowledge_retrieval_tools.py、json_utils.py
- 完全新增：所有Agent、TextSplitFunction、主工作流
12.4 实现路径
按Phase 1-4完成全部路径