From 99164f7ce6d367b2495dbf117d52e8dbbd5a1cc3 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Feb 2026 15:49:17 +0800 Subject: [PATCH] =?UTF-8?q?fix(kb-deep-analyzer):=20=E6=B7=BB=E5=8A=A0?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6=E6=97=A5=E5=BF=97=E5=92=8C=E8=B6=85=E6=97=B6?= =?UTF-8?q?=E8=BF=BD=E8=B8=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SimpleAIClient.generate(): 添加 AI 调用开始/结束日志 - _analyze_chunk(): 添加分片分析耗时日志 - _analyze_chunks_concurrent(): 添加批次进度日志和总耗时统计 - 便于排查卡住问题和监控进度 --- memory/2026-02-13.md | 52 +++++++++++++++++------ skills/kb-deep-analyzer/deep_analyzer.py | 54 ++++++++++++++++++++---- 2 files changed, 85 insertions(+), 21 deletions(-) diff --git a/memory/2026-02-13.md b/memory/2026-02-13.md index a1efd2e..b76c7e6 100644 --- a/memory/2026-02-13.md +++ b/memory/2026-02-13.md @@ -23,38 +23,66 @@ - `94370f2` - feat: 改为批处理模式 - `c8ddc1e` - feat: 添加流式写入器 - `518f8e4` - fix: 优化文档格式 +- `c276df2` - feat(kb-workflow): 添加严格失败判断逻辑 -## 知识库添加：BV1W4f9B6EdC +## 严格失败判断规则（15:19 添加） + +### kb-deep-analyzer 更新 +- 新增判断逻辑：只有 AI 真正完成分析才算成功 +- 判断条件： + - `summary` 不能以"共"开头（降级处理的默认内容） + - `main_topics` 不能为空 + - `key_insights` 不能为空 + - `full_analysis` 必须有值 +- 失败时返回 `success: false` 并退出码 1 + +### kb-workflow 更新 +- 新增"严格失败处理原则"章节 +- 明确：深度分析失败 = 整个流程失败 +- 禁止降级处理或使用模板"假装成功" + +## 知识库添加：BV1W4f9B6EdC（第二次处理） ### 处理结果 - **标题**: 爱泼斯坦案惊天动地？美式司法："没事，已经结束了"【逸语道破】 - **作者**: 沈逸老师 +- **时长**: 32:38 - **Whisper**: ✅ 完成（8043字符，971段） -- **深度分析**: ⚠️ 代码错误，使用基础模板 -- **归档**: ✅ 完成 +- **深度分析**: ❌ 卡住超时 +- **归档**: ❌ 未执行 -### 发现并修复的 Bug +### 执行记录 +1. 环境变量检查 ✅ +2. 链接解析 ✅ (BV1W4f9B6EdC) +3. 元数据提取 ✅ +4. 视频下载 ✅ +5. Whisper 识别 ✅ (15:29 完成) +6. 深度分析 ❌ (15:29 开始，20分钟后仍卡住，终止) +### 发现的问题 1. **参数名不匹配错误** - 错误: `name 'on_batch_complete' is not defined` - - 原因: `analyze()` 用 `on_batch_complete`，内部 `_analyze_chunks_concurrent()` 用 `on_segment_complete` - 修复: 统一使用 `on_segment_complete` 作为内部参数名 2. **batch_size 未定义** - 错误: `name 'batch_size' is not defined` - - 原因: 方法内部多次使用但未作为参数传入 - 修复: 在 `_analyze_chunks_concurrent()` 签名中添加 `batch_size: int = 3` -### 当前状态 -- glm-4.7 深度分析暂时不可用（代码修复后需重新测试） -- 知识库工作流 Whisper 环节正常 -- 需要重新测试深度分析功能 +3. **深度分析卡住** + - 症状: 进程运行但无输出，20分钟后仍卡住 + - 原因: 可能是批处理模式逻辑问题或 API 调用卡住 + - 状态: 需要排查 -## 技能清单 +## 技能状态 | 步骤 | 技能 | 状态 | |------|------|------| | 链接解析 | kb-link-parser | ✅ | | 元数据提取 | kb-bilibili-metadata | ✅ | | 字幕下载 | kb-bilibili-content | ✅ | | 语音识别 | kb-whisper | ✅ | -| 深度分析 | kb-deep-analyzer | ⚠️ 需重测 | +| 深度分析 | kb-deep-analyzer | ❌ 卡住 | + +## 待解决问题 +- [ ] 排查深度分析卡住原因 +- [ ] 验证严格失败判断逻辑是否正确工作 +- [ ] 重新测试完整知识库流程 diff --git a/skills/kb-deep-analyzer/deep_analyzer.py b/skills/kb-deep-analyzer/deep_analyzer.py index 6c0067c..9243007 100644 --- a/skills/kb-deep-analyzer/deep_analyzer.py +++ b/skills/kb-deep-analyzer/deep_analyzer.py @@ -238,7 +238,7 @@ class SimpleAIClient: self.max_retries = 3 # 最大重试次数 self.max_concurrent = 3 # 最大并发数 - def generate(self, prompt: str, max_tokens: int = 1000) -> str: + def generate(self, prompt: str, max_tokens: int = 1000, task_info: str = None) -> str: """生成文本，支持重试""" # 每次调用时重新读取环境变量，解决 exec 传递问题 api_key = os.environ.get("AI_API_KEY", self.api_key) @@ -249,24 +249,35 @@ class SimpleAIClient: if not api_key: raise ValueError("AI_API_KEY 未设置，请配置环境变量") + task_desc = f"[{task_info}]" if task_info else "" + last_error = None for attempt in range(self.max_retries): try: + import time + start_time = time.time() + print(f"[AI Client] {task_desc} 开始调用 AI...") + if provider == "openai": - return self._call_openai(prompt, max_tokens, api_key, model, base_url) + result = self._call_openai(prompt, max_tokens, api_key, model, base_url) elif provider == "zhipu": - return self._call_zhipu(prompt, max_tokens, api_key, model, base_url) + result = self._call_zhipu(prompt, max_tokens, api_key, model, base_url) elif provider == "minimax": - return self._call_minimax(prompt, max_tokens, api_key, model, base_url) + result = self._call_minimax(prompt, max_tokens, api_key, model, base_url) else: raise ValueError(f"不支持的 AI 提供商: {provider}") + + elapsed = time.time() - start_time + print(f"[AI Client] {task_desc} AI 调用完成，耗时: {elapsed:.2f}秒") + return result except Exception as e: + elapsed = time.time() - start_time if 'start_time' in dir() else 0 last_error = e if attempt < self.max_retries - 1: - print(f"[AI Client] 第 {attempt + 1} 次调用失败，{self.max_retries - attempt - 1} 次重试剩余: {e}") + print(f"[AI Client] {task_desc} 第 {attempt + 1} 次调用失败 ({elapsed:.2f}秒)，{self.max_retries - attempt - 1} 次重试剩余: {e}") continue else: - print(f"[AI Client] 调用失败，已重试 {self.max_retries} 次: {e}") + print(f"[AI Client] {task_desc} 调用失败，已重试 {self.max_retries} 次: {e}") raise def generate_json(self, prompt: str) -> Dict[str, Any]: @@ -745,6 +756,9 @@ class DeepAnalyzer: print(f"[DeepAnalyzer] 分块: {num_chunks} 段，批次大小: {batch_size}") + import time + total_start_time = time.time() + # 批处理执行 batch_results = {} # 存储各批结果 all_summaries = [] @@ -756,7 +770,9 @@ class DeepAnalyzer: batch_chunks = chunks[batch_idx:batch_end] batch_times = chunk_times[batch_idx:batch_end] - print(f"[DeepAnalyzer] 批次 {batch_num}/{total_batches}: 段 {batch_idx+1}-{batch_end}") + batch_start_time = time.time() + print(f"[DeepAnalyzer] === 批次 {batch_num}/{total_batches} 开始 ===") + print(f"[DeepAnalyzer] 段 {batch_idx+1}-{batch_end}，预计 {len(batch_chunks)} 个并发调用") # 并发执行当前批次 batch_tasks = [(i, batch_chunks[j], batch_times[j][0], batch_times[j][1]) @@ -766,8 +782,10 @@ class DeepAnalyzer: futures = {executor.submit(self._analyze_chunk, task[1], task[0] + 1, metadata): task[0] for task in batch_tasks} + completed_in_batch = 0 for future in concurrent.futures.as_completed(futures): idx = futures[future] + completed_in_batch += 1 try: summary = future.result() batch_results[idx] = { @@ -776,7 +794,8 @@ class DeepAnalyzer: 'end_time': chunk_times[idx][1], 'summary': summary } - print(f"[DeepAnalyzer] 段 {idx + 1}/{num_chunks} 完成") + elapsed = time.time() - batch_start_time + print(f"[DeepAnalyzer] 段 {idx + 1}/{num_chunks} 完成 ({completed_in_batch}/{len(batch_chunks)}), 批次耗时: {elapsed:.2f}秒") except Exception as e: print(f"[DeepAnalyzer] 段 {idx + 1} 失败: {e}") batch_results[idx] = { @@ -792,12 +811,23 @@ class DeepAnalyzer: on_segment_complete(batch_summaries) all_summaries.extend(batch_summaries) + + batch_elapsed = time.time() - batch_start_time + print(f"[DeepAnalyzer] === 批次 {batch_num}/{total_batches} 完成，耗时: {batch_elapsed:.2f}秒 ===") + + total_elapsed = time.time() - total_start_time + print(f"[DeepAnalyzer] 所有分块分析完成，总耗时: {total_elapsed:.2f}秒") # 按顺序整理结果 all_summaries = [all_summaries[i] for i in range(len(all_summaries))] # 合并结果 + print(f"[DeepAnalyzer] 开始合并各段摘要...") + merge_start = time.time() result = self._merge_summaries(all_summaries, title) + merge_elapsed = time.time() - merge_start + print(f"[DeepAnalyzer] 摘要合并完成，耗时: {merge_elapsed:.2f}秒") + result.timeline = self._build_timeline_from_chunks(all_summaries) return result @@ -884,7 +914,13 @@ class DeepAnalyzer: 请用中文详细回答，包括核心主题、关键要点和重要细节。""" try: - return self.ai_client.generate(prompt, max_tokens=2000) + import time + start_time = time.time() + print(f"[DeepAnalyzer] 开始分析分片 {chunk_index}...") + result = self.ai_client.generate(prompt, max_tokens=2000, task_info=f"分片 {chunk_index}") + elapsed = time.time() - start_time + print(f"[DeepAnalyzer] 分片 {chunk_index} 分析完成，耗时: {elapsed:.2f}秒") + return result except Exception as e: print(f"[DeepAnalyzer] 分片 {chunk_index} 分析失败: {e}") return f"第{chunk_index}段内容分析"