feat(kb-deep-analyzer): 改为批处理模式

- 10段 = 3+3+3+1 批次处理
- 每批最多3个并发
- 添加 on_batch_complete 回调支持
- timeout 保持 600s

skills/kb-deep-analyzer/deep_analyzer.py

@@ -225,8 +225,8 @@ class DeepAnalyzer:
         metadata: Dict[str, Any],
         content: Dict[str, Any],
         content_type: str = "video",
-        on_segment_complete: callable = None,
+        on_batch_complete: callable = None,
-        max_concurrent: int = 3
+        batch_size: int = 3
     ) -> AnalysisResult:
         """
         分析内容
@@ -235,8 +235,8 @@ class DeepAnalyzer:
             metadata: 元数据字典
             content: 内容数据字典（包含 subtitle, description 等）
             content_type: 内容类型
-            on_segment_complete: 每段完成时的回调函数 (idx, summary) => None
+            on_batch_complete: 每批完成时的回调函数 (batch_results) => None
-            max_concurrent: 最大并发数（默认3）
+            batch_size: 每批处理数（默认3）
 
         Returns:
             AnalysisResult 分析结果
@@ -264,11 +264,11 @@ class DeepAnalyzer:
 
         if has_content or has_segments:
             if use_chunking:
-                print(f"[DeepAnalyzer] 文本超过 {CHUNK_THRESHOLD} 字，使用分块分析（并发模式）")
+                print(f"[DeepAnalyzer] 文本超过 {CHUNK_THRESHOLD} 字，使用批处理分块分析")
                 result = self._analyze_chunks_concurrent(
                     title, text, duration, segments, metadata,
-                    max_concurrent=max_concurrent,
+                    batch_size=batch_size,
-                    on_segment_complete=on_segment_complete
+                    on_batch_complete=on_batch_complete
                 )
             elif duration < self.SHORT_VIDEO_THRESHOLD:
                 print(f"[DeepAnalyzer] 使用轻量分析策略 (时长 < 10分钟)")
@@ -498,7 +498,7 @@ class DeepAnalyzer:
         on_segment_complete: callable = None
     ) -> AnalysisResult:
         """
-        并发分块分析（支持流式输出）
+        批处理分块分析（流式输出，避免爆上下文）
 
         Args:
             title: 视频标题
@@ -506,24 +506,24 @@ class DeepAnalyzer:
             duration: 时长（秒）
             segments: SRT segments
             metadata: 元数据
-            max_concurrent: 最大并发数（默认3）
+            batch_size: 每批处理数（默认3）
-            on_segment_complete: 每段完成时的回调函数 (idx, summary) => None
+            on_batch_complete: 每批完成时的回调 (batch_results) => None
 
         Returns:
             AnalysisResult 分析结果
         """
         import concurrent.futures
-        import threading
 
         result = AnalysisResult()
         CHUNK_MAX_CHARS = 3000
         total_chars = len(text)
+        num_chunks = 0
 
         # 生成分块
         has_valid_segments = segments and len(segments) > 0 and all('start' in seg and 'text' in seg for seg in segments)
 
         if has_valid_segments:
-            print(f"[DeepAnalyzer] 使用 SRT 时间戳进行分块（并发模式）")
+            print(f"[DeepAnalyzer] 使用 SRT 时间戳进行分块")
             chunks = []
             chunk_times = []
             current_chunk = ""
@@ -544,17 +544,19 @@ class DeepAnalyzer:
             if current_chunk:
                 chunks.append(current_chunk.strip())
                 chunk_times.append((current_start, current_end))
+
+            num_chunks = len(chunks)
         else:
-            print(f"[DeepAnalyzer] 使用字符位置分块（并发模式）")
+            print(f"[DeepAnalyzer] 使用字符位置分块")
             min_required = total_chars // CHUNK_MAX_CHARS + 1
-            num_segments = min(max(min_required, 3), 20)
+            num_chunks = min(max(min_required, 3), 20)
-            chunk_size = total_chars // num_segments
+            chunk_size = total_chars // num_chunks
             chunks = []
             chunk_times = []
 
-            for i in range(num_segments):
+            for i in range(num_chunks):
                 start = i * chunk_size
-                if i == num_segments - 1:
+                if i == num_chunks - 1:
                     end = total_chars
                 else:
                     end = self._find_breakpoint(text, start, start + chunk_size)
@@ -564,50 +566,62 @@ class DeepAnalyzer:
                 end_sec = int(end / char_per_second)
                 chunk_times.append((self._format_time(start_sec), self._format_time(end_sec)))
 
-        print(f"[DeepAnalyzer] 分块: {len(chunks)} 段，并发数: {max_concurrent}")
+        print(f"[DeepAnalyzer] 分块: {num_chunks} 段，批次大小: {batch_size}")
 
-        def analyze_chunk_wrapper(args):
+        # 批处理执行
-            """包装分析任务"""
+        batch_results = {}  # 存储各批结果
-            idx, chunk_text, start_time, end_time = args
+        all_summaries = []
-            summary = self._analyze_chunk(chunk_text, idx + 1, metadata)
-            return {
-                'index': idx,
-                'start_time': start_time,
-                'end_time': end_time,
-                'summary': summary
-            }
 
-        # 并发执行
+        for batch_idx in range(0, num_chunks, batch_size):
-        chunk_summaries = [None] * len(chunks)
+            batch_num = batch_idx // batch_size + 1
-        tasks = [(i, chunks[i], chunk_times[i][0], chunk_times[i][1]) for i in range(len(chunks))]
+            total_batches = (num_chunks + batch_size - 1) // batch_size
+            batch_end = min(batch_idx + batch_size, num_chunks)
+            batch_chunks = chunks[batch_idx:batch_end]
+            batch_times = chunk_times[batch_idx:batch_end]
 
-        with concurrent.futures.ThreadPoolExecutor(max_workers=max_concurrent) as executor:
+            print(f"[DeepAnalyzer] 批次 {batch_num}/{total_batches}: 段 {batch_idx+1}-{batch_end}")
-            futures = {executor.submit(analyze_chunk_wrapper, task): task[0] for task in tasks}
 
-            for future in concurrent.futures.as_completed(futures):
+            # 并发执行当前批次
-                idx = futures[future]
+            batch_tasks = [(i, batch_chunks[j], batch_times[j][0], batch_times[j][1]) 
-                try:
+                          for j, i in enumerate(range(batch_idx, batch_end))]
-                    result_item = future.result()
+
-                    chunk_summaries[idx] = result_item
+            with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size) as executor:
-                    print(f"[DeepAnalyzer] 段 {idx + 1}/{len(chunks)} 完成")
+                futures = {executor.submit(self._analyze_chunk, task[1], task[0] + 1, metadata): task[0] 
-                    # 回调通知
+                          for task in batch_tasks}
-                    if on_segment_complete:
+
-                        on_segment_complete(idx, result_item['summary'])
+                for future in concurrent.futures.as_completed(futures):
-                except Exception as e:
+                    idx = futures[future]
-                    print(f"[DeepAnalyzer] 段 {idx + 1} 分析失败: {e}")
+                    try:
-                    chunk_summaries[idx] = {
+                        summary = future.result()
-                        'index': idx,
+                        batch_results[idx] = {
-                        'start_time': chunk_times[idx][0],
+                            'index': idx,
-                        'end_time': chunk_times[idx][1],
+                            'start_time': chunk_times[idx][0],
-                        'summary': f"分析失败: {str(e)}"
+                            'end_time': chunk_times[idx][1],
-                    }
+                            'summary': summary
+                        }
+                        print(f"[DeepAnalyzer] 段 {idx + 1}/{num_chunks} 完成")
+                    except Exception as e:
+                        print(f"[DeepAnalyzer] 段 {idx + 1} 失败: {e}")
+                        batch_results[idx] = {
+                            'index': idx,
+                            'start_time': chunk_times[idx][0],
+                            'end_time': chunk_times[idx][1],
+                            'summary': f"分析失败: {str(e)}"
+                        }
+
+            # 回调通知当前批次完成
+            batch_summaries = [batch_results[i] for i in range(batch_idx, batch_end) if i in batch_results]
+            if on_batch_complete:
+                on_batch_complete(batch_summaries)
+
+            all_summaries.extend(batch_summaries)
 
         # 按顺序整理结果
-        chunk_summaries = [c for c in chunk_summaries if c is not None]
+        all_summaries = [all_summaries[i] for i in range(len(all_summaries))]
 
         # 合并结果
-        result = self._merge_summaries(chunk_summaries, title)
+        result = self._merge_summaries(all_summaries, title)
-        result.timeline = self._build_timeline_from_chunks(chunk_summaries)
+        result.timeline = self._build_timeline_from_chunks(all_summaries)
 
         return result