kb-deep-analyzer/SKILL.md

@@ -1,102 +0,0 @@
-# KB Deep Analyzer Skill
-
-**kb-deep-analyzer** - AI 深度内容分析工具。
-
-## Features
-
-- 多类型分析：视频（video）、文章（article）
-- 模板化输出
-- 质量评分
-- 自动标签
-- **多格式元数据支持**: 支持扁平结构 `{"title":"xxx"}` 和嵌套结构 `{"data":{"metadata":{...}}}`
-
-## 内容类型
-
-| 类型 | 参数 | 输出格式 |
-|------|------|----------|
-| 视频 | `--type video` | 摘要 + 主题 + 见解 + 时间轴 |
-| 文章 | `--type article` | 300字摘要 + 主题 + 原文 |
-
-## Commands
-
-```bash
-# 分析视频（默认）
-AI_API_KEY="your-api-key" AI_PROVIDER="zhipu" AI_MODEL="glm-5" \
-  python3 -u deep_analyzer.py --metadata meta.json --content content.json --output analysis.md --type video
-
-# 分析文章（300字摘要 + 原文）
-AI_API_KEY="your-api-key" AI_PROVIDER="zhipu" AI_MODEL="glm-5" \
-  python3 -u deep_analyzer.py --metadata meta.json --content content.json --output analysis.md --type article
-
-# 指定 AI 提供商
-python3 -u deep_analyzer.py -m meta.json -c content.json -o out.md --type video --provider zhipu --model glm-5
-```
-
-## 环境变量配置
-
-**必须设置以下环境变量**：
-
-```bash
-export AI_API_KEY="your-api-key"
-export AI_PROVIDER="zhipu"           # 或 openai, minimax
-export AI_MODEL="glm-5"           # 模型名称
-export AI_BASE_URL="https://..."    # 可选，自定义 API 地址
-```
-
-### 重要提示
-
-**必须使用 `-u` 参数运行 Python**，确保输出实时显示在终端：
-
-```bash
-python3 -u deep_analyzer.py ...
-```
-
-如果不使用 `-u` 参数，Python 的 stdout 缓冲会导致输出延迟或看不到进度。
-
-## 执行方式选择
-
-### 方式1: 并发执行（推荐，默认）
-- 使用 `batch_size=3`（默认）
-- 6个分片约 100 秒完成
-- 适合快速分析
-
-```bash
-python3 -u deep_analyzer.py --metadata meta.json --content content.json -o out.md
-```
-
-### 方式2: 顺序执行
-- 使用 `batch_size=1`
-- 6个分片约 735 秒完成
-- 适合调试和问题排查
-
-```bash
-# 在代码中设置 batch_size=1
-# analyzer.analyze(metadata, content, "video", batch_size=1)
-```
-
-## Metadata Format
-
-支持两种元数据格式：
-
-```json
-// 格式1: 扁平结构
-{
-  "title": "视频标题",
-  "author": "作者名",
-  "duration": 619,
-  "bvid": "BVxxx"
-}
-
-// 格式2: 嵌套结构（kb-bilibili-metadata 输出）
-{
-  "data": {
-    "metadata": {
-      "title": "视频标题",
-      "author": "作者名",
-      "duration": "10:19"
-    }
-  }
-}
-```
-
-会自动转换为扁平格式进行分析。

kb-deep-analyzer/ai_client.py

@@ -1,514 +0,0 @@
-"""Obsidian KB - AI客户端
-支持多个 AI 提供商和 fallback 机制
-"""
-import os
-import json
-import requests
-from typing import Dict, Any, List, Optional, Tuple
-from src.models import ContentType
-from src.config import get_config
-from src.content_chunker import ContentChunker, get_chunking_strategy, chunk_content
-
-
-class AIClient:
-    """
-    AI客户端 - 支持分段分析长视频内容和 fallback 机制
-    """
-
-    def __init__(self):
-        self.config = get_config()
-        self.chunker = ContentChunker()
-        # 获取 AI 配置
-        self._load_ai_config()
-
-    def _load_ai_config(self) -> None:
-        """加载 AI 配置（主 provider + fallbacks）"""
-        self.ai_configs = []
-
-        # 加载主配置
-        main_provider = self.config.get('ai.provider', 'openai')
-        main_config = {
-            'provider': main_provider,
-            'api_key': self.config.get('ai.api_key'),
-            'model': self.config.get('ai.model', 'gpt-4'),
-            'base_url': self.config.get('ai.base_url'),
-            'is_primary': True
-        }
-        self.ai_configs.append(main_config)
-
-        # 加载 fallback 配置
-        fallbacks = self.config.get('ai.fallbacks', [])
-        for fb in fallbacks:
-            self.ai_configs.append({
-                'provider': fb.get('provider'),
-                'api_key': fb.get('api_key'),
-                'model': fb.get('model'),
-                'base_url': fb.get('base_url'),
-                'is_primary': False
-            })
-
-    def _call_api_with_fallback(self, prompt: str, max_tokens: int = 1000, response_format: str = 'text') -> Tuple[bool, str]:
-        """
-        使用 fallback 机制调用 API
-
-        Args:
-            prompt: 提示词
-            max_tokens: 最大 token 数
-            response_format: 响应格式 ('text' 或 'json')
-
-        Returns:
-            (success, response) 元组
-        """
-        last_error = None
-
-        for i, ai_config in enumerate(self.ai_configs):
-            try:
-                if not ai_config.get('api_key'):
-                    print(f"[AIClient] 跳过 {ai_config.get('provider')} - API Key 为空")
-                    continue
-
-                provider_type = ai_config.get('provider')
-                print(f"[AIClient] 尝试使用 {provider_type} (优先级 {i+1}/{len(self.ai_configs)})...")
-
-                response = self._call_single_provider(ai_config, prompt, max_tokens, response_format)
-                return True, response
-
-            except Exception as e:
-                last_error = e
-                print(f"[AIClient] {provider_type} 调用失败: {e}")
-                continue
-
-        # 所有 provider 都失败了
-        error_msg = f"所有 AI 提供商均失败，最后错误: {last_error}" if last_error else "没有可用的 AI 配置"
-        print(f"[AIClient] {error_msg}")
-        return False, self._mock_generate(prompt, max_tokens)  # 返回 mock 响应作为兜底
-
-    def _call_single_provider(self, ai_config: Dict, prompt: str, max_tokens: int, response_format: str) -> str:
-        """
-        调用单个 AI 提供商
-
-        Args:
-            ai_config: AI 配置
-            prompt: 提示词
-            max_tokens: 最大 token 数
-            response_format: 响应格式
-
-        Returns:
-            API 响应文本
-        """
-        provider = ai_config.get('provider')
-        api_key = ai_config.get('api_key')
-        model = ai_config.get('model')
-        base_url = ai_config.get('base_url')
-
-        if provider == 'minimax':
-            return self._call_minimax(api_key, model, base_url, prompt, max_tokens, response_format)
-        elif provider == 'zhipu':
-            return self._call_zhipu(api_key, model, base_url, prompt, max_tokens, response_format)
-        elif provider == 'alibaba':
-            return self._call_alibaba(api_key, model, base_url, prompt, max_tokens, response_format)
-        elif provider == 'openai':
-            return self._call_openai(api_key, model, base_url, prompt, max_tokens, response_format)
-        else:
-            raise ValueError(f"不支持的 provider: {provider}")
-
-    def _call_minimax(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
-        """调用 MiniMax API (Anthropic Messages 兼容接口)"""
-        url = base_url or "https://api.minimaxi.com/anthropic/v1/messages"
-
-        headers = {
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": "application/json"
-        }
-
-        payload = {
-            "model": model,
-            "max_tokens": max_tokens,
-            "messages": [
-                {"role": "user", "content": prompt}
-            ]
-        }
-
-        if response_format == 'json':
-            payload['messages'][0]['content'] = prompt + "\n\n请直接返回 JSON 格式，不要有其他文字。"
-
-        response = requests.post(url, headers=headers, json=payload, timeout=30)
-        response.raise_for_status()
-        data = response.json()
-
-        # MiniMax 返回的是 Anthropic 格式，content 是一个数组
-        # 需要找到 type='text' 的元素
-        for item in data.get('content', []):
-            if item.get('type') == 'text':
-                return item.get('text', '')
-
-        # 如果没有找到 text 类型，返回第一个元素的值（兼容旧格式）
-        return data.get('content', [{}])[0].get('text', '')
-
-    def _call_zhipu(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
-        """调用智谱 AI API (OpenAI 兼容接口)"""
-        # 确保 base_url 包含 /chat/completions
-        if base_url and not base_url.endswith('/chat/completions'):
-            url = f"{base_url}/chat/completions"
-        else:
-            url = base_url or "https://open.bigmodel.cn/api/paas/v4/chat/completions"
-
-        headers = {
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": "application/json"
-        }
-
-        payload = {
-            "model": model,
-            "max_tokens": max_tokens,
-            "messages": [
-                {"role": "user", "content": prompt}
-            ]
-        }
-
-        response = requests.post(url, headers=headers, json=payload, timeout=30)
-        response.raise_for_status()
-        data = response.json()
-        return data['choices'][0]['message']['content']
-
-    def _call_alibaba(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
-        """调用阿里云通义千问 API (OpenAI 兼容接口)"""
-        url = f"{base_url}/chat/completions"
-
-        headers = {
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": "application/json"
-        }
-
-        payload = {
-            "model": model,
-            "max_tokens": max_tokens,
-            "messages": [
-                {"role": "user", "content": prompt}
-            ]
-        }
-
-        response = requests.post(url, headers=headers, json=payload, timeout=30)
-        response.raise_for_status()
-        data = response.json()
-        return data['choices'][0]['message']['content']
-
-    def _call_openai(self, api_key: str, model: str, base_url: str, prompt: str, max_tokens: int, response_format: str) -> str:
-        """调用 OpenAI API"""
-        url = base_url or "https://api.openai.com/v1/chat/completions"
-
-        headers = {
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": "application/json"
-        }
-
-        payload = {
-            "model": model,
-            "max_tokens": max_tokens,
-            "messages": [
-                {"role": "user", "content": prompt}
-            ]
-        }
-
-        response = requests.post(url, headers=headers, json=payload, timeout=30)
-        response.raise_for_status()
-        data = response.json()
-        return data['choices'][0]['message']['content']
-
-    def quick_generate(self, prompt: str, max_tokens: int = 100) -> str:
-        """快速生成（用于轻量扫描）"""
-        success, response = self._call_api_with_fallback(prompt, max_tokens, 'text')
-        return response
-    
-    def generate_json(self, prompt: str) -> Dict[str, Any]:
-        """生成JSON格式响应"""
-        success, response = self._call_api_with_fallback(prompt, 1000, 'json')
-        try:
-            return json.loads(response)
-        except:
-            return self._parse_json_from_text(response)
-    
-    def _deep_analyze(self, content_text: str, content_type: str, duration: Optional[int] = None) -> Dict[str, Any]:
-        """
-        C5.3: 深度分析入口，根据内容时长决定是否分段分析
-        
-        Args:
-            content_text: 内容文本
-            content_type: 内容类型
-            duration: 视频时长（秒）
-            
-        Returns:
-            分析结果字典
-        """
-        duration = duration or 0
-        
-        print(f"[AIClient] 深度分析: type={content_type}, duration={duration}s")
-        
-        # 长视频（>=30分钟）且是视频类型，使用分段分析
-        if duration >= 1800 and content_type == "video":
-            print(f"[AIClient] 使用分段分析策略")
-            return self._analyze_in_chunks(content_text, duration)
-        
-        # 短视频直接分析
-        print(f"[AIClient] 使用直接分析策略")
-        return self._analyze_direct(content_text)
-    
-    def _analyze_in_chunks(self, content_text: str, duration: int) -> Dict[str, Any]:
-        """
-        C5.3: 分段分析长视频
-        
-        Args:
-            content_text: SRT字幕文本
-            duration: 视频时长（秒）
-            
-        Returns:
-            合并后的分析结果
-        """
-        # 1. 获取分段策略
-        strategy = get_chunking_strategy(duration)
-        print(f"[AIClient] 分段策略: {strategy['description']}")
-        
-        # 2. 执行分片
-        chunks = chunk_content(content_text, duration)
-        print(f"[AIClient] 内容已分为 {len(chunks)} 段")
-        
-        # 3. 逐段分析
-        chunk_summaries = []
-        for i, chunk in enumerate(chunks):
-            print(f"[AIClient] 分析第 {i+1}/{len(chunks)} 段...")
-            summary = self._analyze_chunk(chunk.text, chunk.index)
-            
-            # 处理时间格式（可能是float秒数或字符串）
-            start_time = chunk.start_time
-            end_time = chunk.end_time
-            if isinstance(start_time, (int, float)):
-                start_time = self._format_time(start_time)
-            if isinstance(end_time, (int, float)):
-                end_time = self._format_time(end_time)
-            
-            chunk_summaries.append({
-                'index': chunk.index,
-                'start_time': start_time,
-                'end_time': end_time,
-                'summary': summary
-            })
-        
-        # 4. 合并所有段摘要
-        print(f"[AIClient] 合并 {len(chunk_summaries)} 段分析结果")
-        final_result = self._merge_summaries(chunk_summaries, "video")
-        
-        # 5. 构建完整时间轴
-        final_result['timeline'] = self._build_timeline_from_chunks(chunk_summaries)
-        
-        return final_result
-    
-    def _analyze_chunk(self, chunk_text: str, chunk_index: int) -> str:
-        """
-        C5.3: 分析单个内容分片
-        
-        Args:
-            chunk_text: 分片文本
-            chunk_index: 分片索引
-            
-        Returns:
-            分片摘要（限制500字以内）
-        """
-        # 截断文本避免过长
-        text_sample = chunk_text[:3000] if len(chunk_text) > 3000 else chunk_text
-        
-        prompt = f"""请对以下视频片段进行简要分析（限制在300字以内）：
-
-片段内容：
-{text_sample}
-
-请提供：
-1. 这段的核心主题（50字以内）
-2. 关键要点（3-5条）
-3. 重要细节或引用
-
-请用中文回答，控制在300字以内。"""
-        
-        try:
-            summary = self.quick_generate(prompt, max_tokens=500)
-            # 确保不超过500字
-            if len(summary) > 500:
-                summary = summary[:497] + "..."
-            return summary
-        except Exception as e:
-            print(f"[AIClient] 分片 {chunk_index} 分析失败: {e}")
-            return f"第{chunk_index+1}段分析失败"
-    
-    def _merge_summaries(self, chunk_summaries: List[Dict], content_type: str) -> Dict[str, Any]:
-        """
-        C5.3: 合并所有段摘要，生成完整分析报告
-        
-        Args:
-            chunk_summaries: 各段摘要列表
-            content_type: 内容类型
-            
-        Returns:
-            完整分析结果
-        """
-        # 构建合并提示
-        summaries_text = "\n\n".join([
-            f"=== 第{s['index']+1}段 ({s['start_time']}-{s['end_time']}) ===\n{s['summary']}"
-            for s in chunk_summaries
-        ])
-        
-        prompt = f"""基于以下各段分析，生成完整的视频分析报告：
-
-{summaries_text}
-
-请提供以下格式的JSON：
-{{
-    "summary": "整体摘要（200字以内）",
-    "topics": ["主题1", "主题2", "主题3", "主题4"],
-    "insights": ["深入见解1", "深入见解2", "深入见解3"],
-    "actions": ["建议行动1", "建议行动2"],
-    "full_analysis": "详细分析（500字以内）"
-}}"""
-        
-        try:
-            result = self.generate_json(prompt)
-            
-            # 确保必要字段存在
-            if 'topics' not in result:
-                result['topics'] = ["主题A", "主题B", "主题C"]
-            if 'insights' not in result:
-                result['insights'] = ["重要见解1", "重要见解2"]
-            if 'actions' not in result:
-                result['actions'] = []
-            if 'summary' not in result:
-                result['summary'] = "长视频分析完成"
-            if 'full_analysis' not in result:
-                result['full_analysis'] = summaries_text[:500]
-                
-            return result
-            
-        except Exception as e:
-            print(f"[AIClient] 合并摘要失败: {e}")
-            # 返回基础合并结果
-            return {
-                'summary': f'视频共分为{len(chunk_summaries)}个部分进行分析',
-                'topics': [f"主题{i+1}" for i in range(min(4, len(chunk_summaries)))],
-                'insights': ['从分段分析中综合得出'],
-                'actions': [],
-                'full_analysis': summaries_text[:800]
-            }
-    
-    def _analyze_direct(self, content_text: str) -> Dict[str, Any]:
-        """直接分析（不分段）"""
-        # 截断文本
-        text_sample = content_text[:5000] if len(content_text) > 5000 else content_text
-        
-        prompt = f"""请分析以下内容，提供JSON格式结果：
-
-{text_sample}
-
-返回格式：
-{{
-    "summary": "整体摘要（150字以内）",
-    "topics": ["主题1", "主题2", "主题3"],
-    "insights": ["见解1", "见解2"],
-    "actions": ["行动建议1"],
-    "timeline": [
-        {{"start_time": "00:00", "end_time": "05:00", "title": "第一部分", "summary": "..."}}
-    ]
-}}"""
-        
-        try:
-            return self.generate_json(prompt)
-        except Exception as e:
-            print(f"[AIClient] 直接分析失败: {e}")
-            return {
-                'summary': '分析失败',
-                'topics': ['无法解析'],
-                'insights': ['请检查内容格式'],
-                'actions': [],
-                'timeline': []
-            }
-    
-    def _format_time(self, seconds: float) -> str:
-        """格式化秒数为时间字符串"""
-        hours = int(seconds // 3600)
-        minutes = int((seconds % 3600) // 60)
-        secs = int(seconds % 60)
-        if hours > 0:
-            return f"{hours:02d}:{minutes:02d}:{secs:02d}"
-        return f"{minutes:02d}:{secs:02d}"
-
-    def _build_timeline_from_chunks(self, chunk_summaries: List[Dict]) -> List[Dict[str, Any]]:
-        """从分片摘要构建时间轴"""
-        timeline = []
-        for summary in chunk_summaries:
-            timeline.append({
-                'index': summary['index'],
-                'start_time': summary['start_time'],
-                'end_time': summary['end_time'],
-                'title': f"第{summary['index']+1}部分",
-                'summary': summary['summary'][:200] + "..." if len(summary['summary']) > 200 else summary['summary']
-            })
-        return timeline
-    
-    def _mock_generate(self, prompt: str, max_tokens: int) -> str:
-        """模拟AI生成（实际实现应调用真实API）"""
-        # 根据prompt内容返回模拟响应
-        prompt_lower = prompt.lower()
-        
-        if 'json' in prompt_lower or 'json格式' in prompt:
-            if '主题' in prompt or 'topic' in prompt_lower:
-                return json.dumps({
-                    'summary': '这是一个关于人工智能和技术发展的内容，涵盖了多个重要主题。',
-                    'topics': ['人工智能', '机器学习', '未来发展', '技术趋势'],
-                    'insights': [
-                        'AI正在改变我们的工作方式',
-                        '机器学习应用越来越广泛',
-                        '未来技术发展趋势值得关注'
-                    ],
-                    'actions': ['深入研究相关技术', '关注行业动态'],
-                    'full_analysis': '详细分析内容...'
-                }, ensure_ascii=False)
-            return json.dumps({'result': '模拟JSON响应'})
-        
-        if '摘要' in prompt or 'summary' in prompt_lower:
-            return '这是一段关于人工智能发展的重要讨论，涵盖了技术趋势和未来应用。'
-        
-        if '关键词' in prompt or 'keyword' in prompt_lower:
-            return '人工智能, 机器学习, 深度学习, 神经网络, 自然语言处理'
-        
-        return '模拟AI响应内容'
-    
-    def _parse_json_from_text(self, text: str) -> Dict[str, Any]:
-        """从文本中解析JSON"""
-        # 尝试提取JSON块
-        import re
-        json_match = re.search(r'\{[\s\S]*\}', text)
-        if json_match:
-            try:
-                return json.loads(json_match.group())
-            except:
-                pass
-        return {'raw_response': text}
-
-
-class MultiAIClient:
-    """
-    多AI客户端 - 支持多种AI提供商的客户端
-    """
-    
-    def __init__(self):
-        self.config = get_config()
-        self.primary_client = AIClient()
-        self.clients = {
-            'openai': self.primary_client,
-            'local': AIClient(),  # 可以配置为本地模型
-        }
-    
-    def _deep_analyze(self, content_text: str, content_type: str, duration: Optional[int] = None) -> Dict[str, Any]:
-        """使用主客户端进行深度分析"""
-        return self.primary_client._deep_analyze(content_text, content_type, duration)
-    
-    def quick_generate(self, prompt: str, max_tokens: int = 100, provider: str = None) -> str:
-        """快速生成"""
-        client = self.clients.get(provider, self.primary_client)
-        return client.quick_generate(prompt, max_tokens)

kb_workflow/pipeline/processors/analyze.py

@@ -9,7 +9,6 @@ from __future__ import annotations
 import json
 import logging
 import os
-import sys
 from pathlib import Path
 from typing import Optional, Dict, Any
 
@@ -17,12 +16,10 @@ from kb_workflow.pipeline.config import get_settings
 from kb_workflow.pipeline.schemas import AnalyzeInput, AnalyzeOutput
 from kb_workflow.pipeline.state import WorkflowStateManager
 from kb_workflow.pipeline.processors.base import BaseProcessor
+from kb_workflow.deep_analyzer import DeepAnalyzer, SimpleAIClient, generate_frontmatter
 
 logger = logging.getLogger(__name__)
 
-# Flag to ensure sys.path is only modified once for kb-deep-analyzer
-_deep_analyzer_path_added = False
-
 
 class AnalyzeProcessor(BaseProcessor):
     """Analyze 处理器 - AI 深度内容分析
@@ -199,16 +196,6 @@ class AnalyzeProcessor(BaseProcessor):
         Returns:
             AnalysisResult 对象
         """
-        # Lazy import with delayed sys.path modification for external kb-deep-analyzer
-        global _deep_analyzer_path_added
-        if not _deep_analyzer_path_added:
-            _kb_deep_analyzer_path = Path(__file__).parent.parent.parent / "kb-deep-analyzer"
-            if str(_kb_deep_analyzer_path) not in sys.path:
-                sys.path.insert(0, str(_kb_deep_analyzer_path))
-            _deep_analyzer_path_added = True
-
-        from deep_analyzer import DeepAnalyzer, SimpleAIClient
-
         # 初始化 AI 客户端
         ai_client = self._create_ai_client()
 
@@ -224,8 +211,6 @@ class AnalyzeProcessor(BaseProcessor):
         Returns:
             SimpleAIClient 实例
         """
-        from deep_analyzer import SimpleAIClient
-
         # 从环境变量获取配置
         api_key = os.environ.get("AI_API_KEY", "")
         provider = os.environ.get("AI_PROVIDER", "")
@@ -307,8 +292,6 @@ class AnalyzeProcessor(BaseProcessor):
             result: AnalysisResult 对象
             content_type: 内容类型
         """
-        from deep_analyzer import generate_frontmatter
-
         # 标准化元数据（扁平化嵌套结构）
         normalized_metadata = self._normalize_metadata(metadata)