diff --git a/skills/kb-deep-analyzer/deep_analyzer.py b/skills/kb-deep-analyzer/deep_analyzer.py index f25e02c..8bca524 100644 --- a/skills/kb-deep-analyzer/deep_analyzer.py +++ b/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, - max_concurrent: int = 3 + on_batch_complete: callable = None, + batch_size: int = 3 ) -> AnalysisResult: """ 分析内容 @@ -235,8 +235,8 @@ class DeepAnalyzer: metadata: 元数据字典 content: 内容数据字典（包含 subtitle, description 等） content_type: 内容类型 - on_segment_complete: 每段完成时的回调函数 (idx, summary) => None - max_concurrent: 最大并发数（默认3） + on_batch_complete: 每批完成时的回调函数 (batch_results) => None + 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, - on_segment_complete=on_segment_complete + batch_size=batch_size, + 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） - on_segment_complete: 每段完成时的回调函数 (idx, summary) => None + batch_size: 每批处理数（默认3） + 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) - chunk_size = total_chars // num_segments + num_chunks = min(max(min_required, 3), 20) + 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): - """包装分析任务""" - 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 - } + # 批处理执行 + batch_results = {} # 存储各批结果 + all_summaries = [] - # 并发执行 - chunk_summaries = [None] * len(chunks) - tasks = [(i, chunks[i], chunk_times[i][0], chunk_times[i][1]) for i in range(len(chunks))] + for batch_idx in range(0, num_chunks, batch_size): + batch_num = batch_idx // batch_size + 1 + 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: - futures = {executor.submit(analyze_chunk_wrapper, task): task[0] for task in tasks} + print(f"[DeepAnalyzer] 批次 {batch_num}/{total_batches}: 段 {batch_idx+1}-{batch_end}") - 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)}" - } + # 并发执行当前批次 + batch_tasks = [(i, batch_chunks[j], batch_times[j][0], batch_times[j][1]) + for j, i in enumerate(range(batch_idx, batch_end))] + + with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size) as executor: + futures = {executor.submit(self._analyze_chunk, task[1], task[0] + 1, metadata): task[0] + for task in batch_tasks} + + for future in concurrent.futures.as_completed(futures): + idx = futures[future] + try: + summary = future.result() + batch_results[idx] = { + 'index': idx, + 'start_time': chunk_times[idx][0], + '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.timeline = self._build_timeline_from_chunks(chunk_summaries) + result = self._merge_summaries(all_summaries, title) + result.timeline = self._build_timeline_from_chunks(all_summaries) return result