From 520e30d02e95529f34a6fac0e577d9f73f6775a7 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Feb 2026 14:43:40 +0800 Subject: [PATCH] =?UTF-8?q?feat(kb-deep-analyzer):=20=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E8=B6=85=E6=97=B6=E5=92=8C=E5=B9=B6=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - timeout 从 120s 改为 600s - 添加 max_concurrent=3 并发支持 - 新增 _analyze_chunks_concurrent 方法支持流式输出 - analyze() 方法支持 on_segment_complete 回调 --- skills/kb-deep-analyzer/deep_analyzer.py | 155 +++++++++++++++++++++-- 1 file changed, 144 insertions(+), 11 deletions(-) diff --git a/skills/kb-deep-analyzer/deep_analyzer.py b/skills/kb-deep-analyzer/deep_analyzer.py index 995e285..f25e02c 100644 --- a/skills/kb-deep-analyzer/deep_analyzer.py +++ b/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: """ 在自然断点处分割文本