fix(kb-deep-analyzer): 添加进度日志和超时追踪
- SimpleAIClient.generate(): 添加 AI 调用开始/结束日志 - _analyze_chunk(): 添加分片分析耗时日志 - _analyze_chunks_concurrent(): 添加批次进度日志和总耗时统计 - 便于排查卡住问题和监控进度
This commit is contained in:
+40
-12
@@ -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 | ❌ 卡住 |
|
||||
|
||||
## 待解决问题
|
||||
- [ ] 排查深度分析卡住原因
|
||||
- [ ] 验证严格失败判断逻辑是否正确工作
|
||||
- [ ] 重新测试完整知识库流程
|
||||
|
||||
@@ -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}段内容分析"
|
||||
|
||||
Reference in New Issue
Block a user