feat(kb-deep-analyzer): 优化超时和并发

- timeout 从 120s 改为 600s
- 添加 max_concurrent=3 并发支持
- 新增 _analyze_chunks_concurrent 方法支持流式输出
- analyze() 方法支持 on_segment_complete 回调

skills/kb-deep-analyzer/deep_analyzer.py

@@ -60,6 +60,7 @@ class SimpleAIClient:
         self.model = model or os.environ.get("AI_MODEL", "gpt-4")
         self.base_url = base_url or os.environ.get("AI_BASE_URL", "")
         self.max_retries = 3  # 最大重试次数
+        self.max_concurrent = 3  # 最大并发数
 
     def generate(self, prompt: str, max_tokens: int = 1000) -> str:
         """生成文本，支持重试"""
@@ -117,7 +118,7 @@ class SimpleAIClient:
             "messages": [{"role": "user", "content": prompt}]
         }
 
-        response = requests.post(url, headers=headers, json=payload, timeout=120)
+        response = requests.post(url, headers=headers, json=payload, timeout=600)
         response.raise_for_status()
         response.raise_for_status()
         return response.json()['choices'][0]['message']['content']
@@ -135,7 +136,7 @@ class SimpleAIClient:
             "max_tokens": max_tokens
         }
 
-        response = requests.post(url, headers=headers, json=payload, timeout=120)
+        response = requests.post(url, headers=headers, json=payload, timeout=600)
         response.raise_for_status()
         response.raise_for_status()
         data = response.json()
@@ -197,7 +198,7 @@ class SimpleAIClient:
             "messages": [{"role": "user", "content": prompt}]
         }
 
-        response = requests.post(url, headers=headers, json=payload, timeout=120)
+        response = requests.post(url, headers=headers, json=payload, timeout=600)
         response.raise_for_status()
         response.raise_for_status()
         data = response.json()
@@ -219,8 +220,14 @@ class DeepAnalyzer:
     def __init__(self, ai_client: SimpleAIClient = None):
         self.ai_client = ai_client or SimpleAIClient()
 
-    def analyze(self, metadata: Dict[str, Any], content: Dict[str, Any],
-                content_type: str = "video") -> AnalysisResult:
+    def analyze(
+        self,
+        metadata: Dict[str, Any],
+        content: Dict[str, Any],
+        content_type: str = "video",
+        on_segment_complete: callable = None,
+        max_concurrent: int = 3
+    ) -> AnalysisResult:
         """
         分析内容
 
@@ -228,6 +235,8 @@ class DeepAnalyzer:
             metadata: 元数据字典
             content: 内容数据字典（包含 subtitle, description 等）
             content_type: 内容类型
+            on_segment_complete: 每段完成时的回调函数 (idx, summary) => None
+            max_concurrent: 最大并发数（默认3）
 
         Returns:
             AnalysisResult 分析结果
@@ -239,9 +248,6 @@ class DeepAnalyzer:
         duration = metadata.get('duration', 0)
         
         # 支持多种内容格式
-        # 1. Whisper 格式: {"text": "...", "segments": [...]}
-        # 2. 字幕格式: {"subtitle": "..."}
-        # 3. 描述格式: {"description": "..."}
         text = content.get('text', '') or content.get('subtitle', '') or content.get('description', '')
         segments = content.get('segments', [])
         
@@ -253,14 +259,17 @@ class DeepAnalyzer:
         print(f"[DeepAnalyzer] 文本存在: {has_content}, 分段数: {len(segments)}")
         print(f"[DeepAnalyzer] 文本长度: {len(text) if text else 0} 字符")
 
-        # 强制分块阈值：无论视频长短，超过3000字都用分块分析
         CHUNK_THRESHOLD = 3000
         use_chunking = has_content and len(text) > CHUNK_THRESHOLD
 
         if has_content or has_segments:
             if use_chunking:
-                print(f"[DeepAnalyzer] 文本超过 {CHUNK_THRESHOLD} 字，使用分块分析")
-                result = self._analyze_in_chunks_with_segments(title, text, duration, segments, metadata)
+                print(f"[DeepAnalyzer] 文本超过 {CHUNK_THRESHOLD} 字，使用分块分析（并发模式）")
+                result = self._analyze_chunks_concurrent(
+                    title, text, duration, segments, metadata,
+                    max_concurrent=max_concurrent,
+                    on_segment_complete=on_segment_complete
+                )
             elif duration < self.SHORT_VIDEO_THRESHOLD:
                 print(f"[DeepAnalyzer] 使用轻量分析策略 (时长 < 10分钟)")
                 result = self._analyze_short_with_subtitle(title, text, duration, segments)
@@ -478,6 +487,130 @@ class DeepAnalyzer:
 
         return result
 
+    def _analyze_chunks_concurrent(
+        self,
+        title: str,
+        text: str,
+        duration: int,
+        segments: list = None,
+        metadata: Dict = None,
+        max_concurrent: int = 3,
+        on_segment_complete: callable = None
+    ) -> AnalysisResult:
+        """
+        并发分块分析（支持流式输出）
+
+        Args:
+            title: 视频标题
+            text: 完整文本
+            duration: 时长（秒）
+            segments: SRT segments
+            metadata: 元数据
+            max_concurrent: 最大并发数（默认3）
+            on_segment_complete: 每段完成时的回调函数 (idx, summary) => None
+
+        Returns:
+            AnalysisResult 分析结果
+        """
+        import concurrent.futures
+        import threading
+
+        result = AnalysisResult()
+        CHUNK_MAX_CHARS = 3000
+        total_chars = len(text)
+
+        # 生成分块
+        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 时间戳进行分块（并发模式）")
+            chunks = []
+            chunk_times = []
+            current_chunk = ""
+            current_start = segments[0]['start']
+            current_end = ""
+
+            for seg in segments:
+                seg_text = f"[{seg['start']}] {seg['text']}\n"
+                if len(current_chunk) + len(seg_text) > CHUNK_MAX_CHARS and current_chunk:
+                    chunks.append(current_chunk.strip())
+                    chunk_times.append((current_start, current_end or seg['start']))
+                    current_chunk = seg_text
+                    current_start = seg['start']
+                else:
+                    current_chunk += seg_text
+                    current_end = seg['end']
+
+            if current_chunk:
+                chunks.append(current_chunk.strip())
+                chunk_times.append((current_start, current_end))
+        else:
+            print(f"[DeepAnalyzer] 使用字符位置分块（并发模式）")
+            min_required = total_chars // CHUNK_MAX_CHARS + 1
+            num_segments = min(max(min_required, 3), 20)
+            chunk_size = total_chars // num_segments
+            chunks = []
+            chunk_times = []
+
+            for i in range(num_segments):
+                start = i * chunk_size
+                if i == num_segments - 1:
+                    end = total_chars
+                else:
+                    end = self._find_breakpoint(text, start, start + chunk_size)
+                chunks.append(text[start:end])
+                char_per_second = total_chars / duration if duration > 0 else 1
+                start_sec = int(start / char_per_second)
+                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}")
+
+        def analyze_chunk_wrapper(args):
+            """包装分析任务"""
+            idx, chunk_text, start_time, end_time = args
+            summary = self._analyze_chunk(chunk_text, idx + 1, metadata)
+            return {
+                'index': idx,
+                'start_time': start_time,
+                'end_time': end_time,
+                'summary': summary
+            }
+
+        # 并发执行
+        chunk_summaries = [None] * len(chunks)
+        tasks = [(i, chunks[i], chunk_times[i][0], chunk_times[i][1]) for i in range(len(chunks))]
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=max_concurrent) as executor:
+            futures = {executor.submit(analyze_chunk_wrapper, task): task[0] for task in tasks}
+
+            for future in concurrent.futures.as_completed(futures):
+                idx = futures[future]
+                try:
+                    result_item = future.result()
+                    chunk_summaries[idx] = result_item
+                    print(f"[DeepAnalyzer] 段 {idx + 1}/{len(chunks)} 完成")
+                    # 回调通知
+                    if on_segment_complete:
+                        on_segment_complete(idx, result_item['summary'])
+                except Exception as e:
+                    print(f"[DeepAnalyzer] 段 {idx + 1} 分析失败: {e}")
+                    chunk_summaries[idx] = {
+                        'index': idx,
+                        'start_time': chunk_times[idx][0],
+                        'end_time': chunk_times[idx][1],
+                        'summary': f"分析失败: {str(e)}"
+                    }
+
+        # 按顺序整理结果
+        chunk_summaries = [c for c in chunk_summaries if c is not None]
+
+        # 合并结果
+        result = self._merge_summaries(chunk_summaries, title)
+        result.timeline = self._build_timeline_from_chunks(chunk_summaries)
+
+        return result
+
     def _find_breakpoint(self, text: str, start: int, preferred_end: int) -> int:
         """
         在自然断点处分割文本