更新可视化

extract_topn_multimodal.py

@@ -529,8 +529,8 @@ if __name__ == "__main__":
     )
 
     # 默认路径配置
-    DEFAULT_CONTEXT_FILE = "input/test_case/output/knowledge_search_traverse/20251118/194351_e3/run_context_v3.json"
-    DEFAULT_OUTPUT_FILE = "input/test_case/output/knowledge_search_traverse/20251118/194351_e3/multimodal_extraction_topn_cleaned.json"
+    DEFAULT_CONTEXT_FILE = "input/test_case/output/knowledge_search_traverse/20251119/004308_d3/run_context_v3.json"
+    DEFAULT_OUTPUT_FILE = "input/test_case/output/knowledge_search_traverse/20251119/004308_d3/multimodal_extraction_topn_cleaned.json"
 
     # 添加参数
     parser.add_argument(
@@ -549,7 +549,7 @@ if __name__ == "__main__":
         '-n', '--top-n',
         dest='top_n',
         type=int,
-        default=10,
+        default=20,
         help='提取前N个帖子 (默认: 10)'
     )
     parser.add_argument(

knowledge_search_traverse.py

@@ -210,6 +210,12 @@ class RunContext(BaseModel):
     word_score_history: dict[str, float] = Field(default_factory=dict)
     # key: 词/组合文本, value: 最终得分
 
+    # 统计信息
+    stats_llm_calls: int = 0  # LLM评估调用次数
+    stats_sug_requests: int = 0  # 小红书SUG请求次数（包括缓存）
+    stats_sug_cache_hits: int = 0  # SUG缓存命中次数
+    stats_search_calls: int = 0  # 搜索调用次数
+
 
 # ============================================================================
 # Agent 定义
@@ -254,27 +260,24 @@ semantic_segmentation_instructions = """
 ---
 
 ### 2. 修饰词
-**定义**：对中心名词的限定和修饰的完整语义单元
-**包含**：形容词、时间词、地点词、程度词
-
-**注意**：多个连续的修饰词可以组合成一个片段
-
+**定义**：对中心名词的限定和修饰的完整语义单元，多个连续的修饰词可以组合成一个片段作为修饰词
+**包含**："X的Y"结构中"X的"是修饰词，"Y"是中心名词，X只能拆分为一个分段
 ---
 
 ### 3. 中心名词
-**定义**：动作和目标的核心对象
+**定义**：动作的核心对象，被修饰词修饰
 **包含**：
 - 核心名词：素材、梗图、表情包、教程
 - 复合名词：摄影素材、风光摄影素材、表情包梗图
 
 ---
 
-## 分段原则
+## 分段原则（务必遵守）
 
 1. **语义完整性**：每个片段应该是完整的语义单元
    - 动作目标：疑问词+动作词应该合并
    - 修饰词：连续的修饰成分可以合并
-   - 中心名词：复合名词保持完整
+   - 中心名词：复合名词保持完整，一个语句中务必只能分段出一个中心名词
 
 2. **维度互斥**：每个片段只能属于一种维度
 
@@ -308,32 +311,6 @@ semantic_segmentation_instructions = """
 }
 ```
 
-**Query**: "职场相关的网络热梗有哪些"
-
-**分段结果**:
-```json
-{
-  "segments": [
-    {
-      "segment_text": "职场相关的网络热",
-      "segment_type": "修饰词",
-      "reasoning": "多个修饰词组合，限定了梗的类型和范围"
-    },
-    {
-      "segment_text": "梗",
-      "segment_type": "中心名词",
-      "reasoning": "核心对象"
-    },
-    {
-      "segment_text": "有哪些",
-      "segment_type": "动作目标",
-      "reasoning": "完整的疑问表达，表达寻找/列举的动机"
-    }
-  ],
-  "overall_reasoning": "修饰词+名词+疑问表达的结构"
-}
-```
-
 ## 输出要求
 - segments: 片段列表
   - segment_text: 片段文本（必须来自原query）
@@ -1839,13 +1816,21 @@ def save_sug_cache(keyword: str, suggestions: list[str]):
         print(f"  ⚠️  写入SUG缓存失败（{keyword}）: {exc}")
 
 
-def get_suggestions_with_cache(keyword: str, api: XiaohongshuSearchRecommendations) -> list[str]:
+def get_suggestions_with_cache(keyword: str, api: XiaohongshuSearchRecommendations, context: RunContext | None = None) -> list[str]:
     """带持久化缓存的SUG获取"""
     cached = load_sug_cache(keyword)
     if cached is not None:
         print(f"    📦 SUG缓存命中: {keyword} ({len(cached)} 个)")
+        # 统计：SUG请求次数 + 缓存命中次数
+        if context is not None:
+            context.stats_sug_requests += 1
+            context.stats_sug_cache_hits += 1
         return cached
 
+    # 统计：SUG请求次数
+    if context is not None:
+        context.stats_sug_requests += 1
+
     suggestions = api.get_recommendations(keyword=keyword)
     if suggestions:
         save_sug_cache(keyword, suggestions)
@@ -2264,7 +2249,7 @@ def process_note_data(note: dict) -> Post:
     return post
 
 
-async def evaluate_with_o(text: str, o: str, cache: dict[str, tuple[float, str]] | None = None) -> tuple[float, str]:
+async def evaluate_with_o(text: str, o: str, cache: dict[str, tuple[float, str]] | None = None, context: RunContext | None = None, **kwargs) -> tuple[float, str]:
     """评估文本与原始问题o的相关度
 
     采用两阶段评估 + 代码计算规则：
@@ -2276,10 +2261,15 @@ async def evaluate_with_o(text: str, o: str, cache: dict[str, tuple[float, str]]
         text: 待评估的文本
         o: 原始问题
         cache: 评估缓存（可选），用于避免重复评估
+        context: 运行上下文（可选），用于统计
 
     Returns:
         tuple[float, str]: (最终相关度分数, 综合评估理由)
     """
+    # 统计LLM调用（无论是否缓存命中都计数，因为是"评估比对"次数）
+    if context is not None:
+        context.stats_llm_calls += 3  # 3个评估器
+
     # 检查缓存
     if cache is not None and text in cache:
         cached_score, cached_reason = cache[text]
@@ -2482,7 +2472,7 @@ async def evaluate_with_o_round0(text: str, o: str, cache: dict[str, tuple[float
     return 0.0, fallback_reason
 
 
-async def evaluate_within_scope(text: str, scope_text: str, cache: dict[str, tuple[float, str]] | None = None) -> tuple[float, str]:
+async def evaluate_within_scope(text: str, scope_text: str, cache: dict[str, tuple[float, str]] | None = None, context: RunContext | None = None) -> tuple[float, str]:
     """域内/域间专用评估函数（v124新增 - 需求2&3）
 
     用于评估词条与作用域词条（单域或域组合）的相关度
@@ -2497,10 +2487,15 @@ async def evaluate_within_scope(text: str, scope_text: str, cache: dict[str, tup
         text: 待评估的词条
         scope_text: 作用域词条（可以是单域词条或域组合词条）
         cache: 评估缓存（可选），用于避免重复评估
+        context: 运行上下文（可选），用于统计
 
     Returns:
         tuple[float, str]: (最终相关度分数, 综合评估理由)
     """
+    # 统计LLM调用（无论是否缓存命中都计数）
+    if context is not None:
+        context.stats_llm_calls += 2  # 2个评估器
+
     # 检查缓存
     cache_key = f"scope:{text}:{scope_text}"  # 添加前缀以区分不同评估类型
     if cache is not None and cache_key in cache:
@@ -2644,6 +2639,9 @@ async def evaluate_domain_combination_round1(
     Returns:
         (最终得分, 评估理由)
     """
+    # 统计LLM调用
+    context.stats_llm_calls += 1  # 1个评估器
+
     # 获取所属segment
     domain_idx = comb.domains[0] if comb.domains else 0
     segment = segments[domain_idx] if 0 <= domain_idx < len(segments) else None
@@ -2721,7 +2719,8 @@ async def evaluate_domain_combination_round2plus(
     base_score, base_reason = await evaluate_within_scope(
         comb.text,
         scope_text,
-        context.evaluation_cache
+        context.evaluation_cache,
+        context
     )
 
     # 步骤2: 计算加权系数
@@ -2817,7 +2816,7 @@ async def initialize(o: str, context: RunContext) -> tuple[list[Seg], list[Word]
     async def evaluate_seg(seg: Seg) -> Seg:
         async with seg_semaphore:
             # 初始化阶段的分词评估使用第一轮 prompt (round_num=1)
-            seg.score_with_o, seg.reason = await evaluate_with_o(seg.text, o, context.evaluation_cache, round_num=1)
+            seg.score_with_o, seg.reason = await evaluate_with_o(seg.text, o, context.evaluation_cache, context=context, round_num=1)
             return seg
 
     if seg_list:
@@ -2912,7 +2911,7 @@ async def run_round(
     sug_list_list = []  # list of list
     for q in q_list:
         print(f"\n  处理q: {q.text}")
-        suggestions = get_suggestions_with_cache(q.text, xiaohongshu_api)
+        suggestions = get_suggestions_with_cache(q.text, xiaohongshu_api, context)
 
         q_sug_list = []
         if suggestions:
@@ -2949,7 +2948,7 @@ async def run_round(
     async def evaluate_sug(sug: Sug) -> Sug:
         async with semaphore:  # 限制并发数
             # 根据轮次选择 prompt: 第一轮使用 round1 prompt，后续使用标准 prompt
-            sug.score_with_o, sug.reason = await evaluate_with_o(sug.text, o, context.evaluation_cache, round_num=round_num)
+            sug.score_with_o, sug.reason = await evaluate_with_o(sug.text, o, context.evaluation_cache, context=context, round_num=round_num)
             return sug
 
     if all_sugs:
@@ -3174,7 +3173,7 @@ async def run_round(
                 }
 
             # 正常评估，根据轮次选择 prompt
-            score, reason = await evaluate_with_o(combined, o, context.evaluation_cache, round_num=round_num)
+            score, reason = await evaluate_with_o(combined, o, context.evaluation_cache, context=context, round_num=round_num)
             return {
                 'word': comb.selected_word,
                 'query': combined,
@@ -3601,7 +3600,7 @@ async def run_round_v2(
     sug_details = {}
 
     for q in query_input:
-        suggestions = get_suggestions_with_cache(q.text, xiaohongshu_api)
+        suggestions = get_suggestions_with_cache(q.text, xiaohongshu_api, context)
         if suggestions:
             print(f"  {q.text}: 获取到 {len(suggestions)} 个SUG")
             for sug_text in suggestions:
@@ -3622,7 +3621,7 @@ async def run_round_v2(
         async def evaluate_sug(sug: Sug) -> Sug:
             async with semaphore:
                 sug.score_with_o, sug.reason = await evaluate_with_o(
-                    sug.text, o, context.evaluation_cache
+                    sug.text, o, context.evaluation_cache, context=context
                 )
                 return sug
 
@@ -3646,6 +3645,8 @@ async def run_round_v2(
     async def search_keyword(text: str, score: float, source_type: str) -> Search:
         """通用搜索函数"""
         print(f"    搜索: {text} (来源: {source_type})")
+        # 统计：搜索调用次数
+        context.stats_search_calls += 1
         try:
             search_result = xiaohongshu_search.search(keyword=text)
             notes = search_result.get("data", {}).get("data", [])
@@ -4029,6 +4030,10 @@ async def iterative_loop_v2(
     print(f"  总搜索次数: {len(all_search_list)}")
     print(f"  总帖子数: {sum(len(s.post_list) for s in all_search_list)}")
     # print(f"  提取帖子数: {len(all_extraction_results)}")  # 内容提取流程已断开
+    print(f"\n[统计信息]")
+    print(f"  LLM评估调用: {context.stats_llm_calls} 次")
+    print(f"  SUG请求: {context.stats_sug_requests} 次 (缓存命中: {context.stats_sug_cache_hits} 次)")
+    print(f"  搜索调用: {context.stats_search_calls} 次")
     print(f"{'='*60}")
 
     return all_search_list  # 不再返回提取结果
@@ -4099,6 +4104,10 @@ async def main(input_dir: str, max_rounds: int = 2, sug_threshold: float = 0.7,
         output += f"总搜索次数：{len(all_search_list)}\n"
         output += f"总帖子数：{sum(len(s.post_list) for s in all_search_list)}\n"
         # output += f"提取帖子数：{len(all_extraction_results)}\n"  # 内容提取流程已断开
+        output += f"\n统计信息：\n"
+        output += f"  LLM评估调用: {run_context.stats_llm_calls} 次\n"
+        output += f"  SUG请求: {run_context.stats_sug_requests} 次 (缓存命中: {run_context.stats_sug_cache_hits} 次)\n"
+        output += f"  搜索调用: {run_context.stats_search_calls} 次\n"
         output += "\n" + "="*60 + "\n"
 
         if all_search_list:

script/search/xiaohongshu_search.py

@@ -41,7 +41,7 @@ class XiaohongshuSearch:
     def search(
         self,
         keyword: str,
-        content_type: str = "不限",
+        content_type: str = "图文",
         sort_type: str = "综合",
         publish_time: str = "不限",
         cursor: str = "",

script/search_recommendations/xiaohongshu_search_recommendations.py

@@ -96,7 +96,7 @@ class XiaohongshuSearchRecommendations:
 
         return None
 
-    def get_recommendations(self, keyword: str, timeout: int = 300, max_retries: int = 4, retry_delay: int = 7, use_cache: bool = True) -> Dict[str, Any]:
+    def get_recommendations(self, keyword: str, timeout: int = 30, max_retries: int = 2, retry_delay: int = 6, use_cache: bool = True) -> Dict[str, Any]:
         """
         获取小红书搜索推荐词