diff --git a/kb_workflow/deep_analyzer/__init__.py b/kb_workflow/deep_analyzer/__init__.py new file mode 100644 index 0000000..2f947ce --- /dev/null +++ b/kb_workflow/deep_analyzer/__init__.py @@ -0,0 +1,650 @@ +#!/usr/bin/env python3 +"""Deep Analyzer - AI 深度内容分析工具 + +一次性调用 AI 生成完整深度分析报告。 + +Usage: + python deep_analyzer.py --metadata meta.json --content content.json --output analysis.md [--type video] + python deep_analyzer.py --help +""" + +import argparse +import json +import os +import sys +import re +from datetime import datetime +from pathlib import Path +from typing import Dict, Any, List +from dataclasses import dataclass, field + + +# ============== 数据模型 ============== + +@dataclass +class AnalysisResult: + """分析结果""" + summary: str = "" + main_topics: List[str] = field(default_factory=list) + key_insights: List[str] = field(default_factory=list) + recommended_tags: List[str] = field(default_factory=list) + full_analysis: str = "" + original_text: str = "" # 文章类型的原文内容 + + +# ============== AI 客户端 ============== + +class SimpleAIClient: + """AI 客户端""" + + def __init__(self, api_key: str = None, provider: str = None, model: str = None, base_url: str = None): + self.api_key = api_key or os.environ.get("AI_API_KEY", "") + self.provider = provider or os.environ.get("AI_PROVIDER", "") + self.model = model or os.environ.get("AI_MODEL", "glm-5") + self.base_url = base_url or os.environ.get("AI_BASE_URL", "") + self.max_retries = 3 + + def generate(self, prompt: str, max_tokens: int = 16000) -> str: + """生成文本""" + api_key = os.environ.get("AI_API_KEY", self.api_key) + provider = os.environ.get("AI_PROVIDER", self.provider) + model = os.environ.get("AI_MODEL", self.model) + base_url = os.environ.get("AI_BASE_URL", self.base_url) + + if not api_key: + raise ValueError("AI_API_KEY 未设置") + + for attempt in range(self.max_retries): + try: + import time + start_time = time.time() + print(f"[AI Client] 开始调用 AI...") + + if provider == "openai": + result = self._call_openai(prompt, max_tokens, api_key, model, base_url) + elif provider == "zhipu": + result = self._call_zhipu(prompt, max_tokens, api_key, model, base_url) + elif provider == "minimax": + 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] AI 调用完成，耗时: {elapsed:.2f}秒") + return result + except Exception as e: + if attempt < self.max_retries - 1: + print(f"[AI Client] 第 {attempt + 1} 次调用失败，{self.max_retries - attempt - 1} 次重试剩余: {e}") + continue + else: + raise + + def _call_openai(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str: + import requests + url = f"{base_url}/chat/completions" if base_url else "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=600) + response.raise_for_status() + return response.json()['choices'][0]['message']['content'] + + def _call_zhipu(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str: + import requests + base_url = base_url or "https://open.bigmodel.cn/api/coding/paas/v4" + url = f"{base_url}/chat/completions" + headers = {"Authorization": f"Bearer {api_key}"} + payload = {"model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens} + response = requests.post(url, headers=headers, json=payload, timeout=600) + response.raise_for_status() + data = response.json() + message = data.get('choices', [{}])[0].get('message', {}) + content = message.get('content', '') + reasoning = message.get('reasoning_content', '') + if not content and reasoning: + sentences = reasoning.strip().split('。') + useful = [s.strip() for s in sentences if s.strip() and len(s.strip()) > 10] + if useful: + return '。'.join(useful[-3:]) + '。' + return reasoning.strip()[:500] + return content + + def _call_minimax(self, prompt: str, max_tokens: int, api_key: str, model: str, base_url: str) -> str: + import requests + 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}]} + response = requests.post(url, headers=headers, json=payload, timeout=600) + response.raise_for_status() + data = response.json() + for item in data.get('content', []): + if item.get('type') == 'text': + return item.get('text', '') + return str(data) + + +# ============== 分析器 ============== + +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: + """一次性调用 AI 生成完整分析报告 + + Args: + metadata: 元数据字典 + content: 内容字典 + content_type: 内容类型，"video" 或 "article" + """ + + # 标准化元数据 + metadata = self._normalize_metadata(metadata) + + title = metadata.get('title', '未知') + duration = metadata.get('duration', 0) + author = metadata.get('author', '') + pubtime = metadata.get('pubtime', '') + url = metadata.get('url', '') + bvid = metadata.get('bvid', '') + + # 推断 platform：优先使用元数据中的值，否则根据 URL 或 content_type 推断 + platform = metadata.get('platform', '') + if not platform: + if 'zhihu.com' in url or content_type in ('answer', 'zhuanlan', 'question'): + platform = 'zhihu' + elif 'bilibili.com' in url or bvid: + platform = 'bilibili' + else: + platform = 'bilibili' # 默认值 + + # 清理 author 字段：移除零宽度空格和前后空白 + author = self._clean_author(author) + + # 获取内容 - 支持多种字段名 + text = content.get('text', '') or content.get('subtitle', '') or content.get('subtitles', '') or content.get('description', '') + + print(f"[DeepAnalyzer] 开始分析: {title}") + print(f"[DeepAnalyzer] 类型: {content_type}, 内容长度: {len(text)} 字符") + + # 构建 prompt（根据类型选择不同模板） + prompt = self._build_prompt( + title=title, + duration=duration, + duration_str=self._format_duration(duration), + author=author, + pubtime=pubtime, + url=url, + bvid=bvid, + platform=platform, + text=text, + content_type=content_type + ) + + # 调用 AI + try: + full_analysis = self.ai_client.generate(prompt, max_tokens=16000) + except Exception as e: + print(f"[DeepAnalyzer] AI 调用失败: {e}") + raise + + # 解析结果 + result = self._parse_analysis(full_analysis, text, duration, content_type) + + # 文章/回答类型：追加原文 + if content_type in ("article", "answer", "zhuanlan"): + result.original_text = text + + return result + + def _build_prompt(self, title: str, duration: int, duration_str: str, author: str, + pubtime: str, url: str, bvid: str, platform: str, text: str, + content_type: str = "video") -> str: + """构建分析 prompt，根据内容类型选择不同模板""" + if content_type in ("article", "answer", "zhuanlan"): + return self._build_article_prompt(title, author, pubtime, url, platform, text) + else: + return self._build_video_prompt(title, duration, duration_str, author, pubtime, url, bvid, platform, text) + + def _build_video_prompt(self, title: str, duration: int, duration_str: str, author: str, + pubtime: str, url: str, bvid: str, platform: str, text: str) -> str: + """构建视频分析 prompt""" + + # 计算分片数（按10分钟分片） + num_chunks = max(1, (duration + 599) // 600) + + # 截取适当长度的文本（20-30K 字符） + max_chars = 30000 + if len(text) > max_chars: + text_sample = text[:max_chars] + "\n\n[...]" + else: + text_sample = text + + prompt = f"""你是一个专业的视频内容分析师。请根据以下信息生成深度分析报告： + +【视频信息】 +标题：{title} +作者：{author} +时长：{duration_str}（{duration}秒） +发布日期：{pubtime} +平台：{platform} +URL：{url} + +【内容】（完整字幕） +{text_sample} + +【输出要求】必须使用以下格式： + +## 标题 +[自己总结的简洁标题，禁止使用原标题] + +## 摘要 +[200-400字] + +## 主要主题 +[从内容中提取不超过10个主题，每个不超过10字] +- 主题1 +- 主题2 +... + +## 关键见解 +[若干条深刻见解] +- 见解1 +- 见解2 +... + +## 标签 +[从内容中提取3-5个精炼标签，每个不超过10字] +- 标签1 +- 标签2 +... + +## 时间轴 +### 第 1 部分 (00:00 - 10:00) +[详细分析，至少500字] + +### 第 2 部分 (10:00 - 20:00) +[详细分析，至少500字] + +...（以此类推） + +【重要】分析应足够详尽，读者阅读后可了解视频核心内容，无需再看原视频。""" + + return prompt + + def _build_article_prompt(self, title: str, author: str, pubtime: str, + url: str, platform: str, text: str) -> str: + """构建文章分析 prompt""" + + # 文章保留完整内容（不截断，因为需要追加原文） + # 但 prompt 中只取前 20000 字符供 AI 分析 + max_chars = 20000 + if len(text) > max_chars: + text_sample = text[:max_chars] + "\n\n[...内容过长已截断，但原文会完整保留...]" + else: + text_sample = text + + prompt = f"""你是一个专业的文章内容分析师。请根据以下信息生成摘要： + +【文章信息】 +标题：{title} +作者：{author} +发布日期：{pubtime} +平台：{platform} +URL：{url} + +【内容】 +{text_sample} + +【输出要求】必须使用以下格式： + +## 摘要 +[300字以内的摘要，描述文章的核心观点和主要内容] + +## 主要主题 +[从内容中提取不超过5个主题，每个不超过10字] +- 主题1 +- 主题2 +... + +## 标签 +[从内容中提取3-5个精炼标签，每个不超过10字] +- 标签1 +- 标签2 +... + +【重要】 +1. 摘要应精炼概括文章核心观点，让读者快速了解文章主旨 +2. 不要输出时间轴 +3. 不要输出关键见解（除非是特别重要的观点）""" + + return prompt + + def _parse_analysis(self, analysis_text: str, original_text: str, duration: int, + content_type: str = "video") -> AnalysisResult: + """解析 AI 返回的分析结果""" + result = AnalysisResult() + result.full_analysis = analysis_text + + # 提取各部分（使用正则） + lines = analysis_text.split('\n') + + # 简单解析：按 ## 标题分割 + current_section = "" + in_section = None + + for line in lines: + if line.startswith('##'): + # 保存上一个 section + if in_section: + self._store_section(result, in_section, current_section) + # 开始新 section + section_name = line.lstrip('#').strip() + in_section = section_name + current_section = "" + elif in_section: + current_section += line + "\n" + + # 存储最后一个 section + if in_section: + self._store_section(result, in_section, current_section) + + # 如果解析失败，使用原始文本 + if not result.summary: + result.summary = analysis_text[:500] + + return result + + def _store_section(self, result: AnalysisResult, section_name: str, content: str): + """存储解析到的 section""" + content = content.strip() + if not content: + return + + section_lower = section_name.lower() + + if '摘要' in section_lower: + result.summary = content[:500] + elif '主题' in section_lower: + # 提取主题 + topics = [] + for line in content.split('\n'): + line = line.strip().lstrip('-').strip() + if line and len(line) <= 15: + topics.append(line) + result.main_topics = topics[:10] + elif '见解' in section_lower or '观点' in section_lower: + # 提取见解 + insights = [] + for line in content.split('\n'): + line = line.strip().lstrip('-').strip() + if line and len(line) > 5: + insights.append(line) + result.key_insights = insights + elif '标签' in section_lower: + # 提取标签 + tags = [] + for line in content.split('\n'): + line = line.strip().lstrip('-').strip() + if line and len(line) <= 10: + tags.append(line) + result.recommended_tags = tags[:5] + + def _normalize_metadata(self, metadata: Dict) -> Dict: + """标准化元数据""" + if not metadata: + return {} + + # 如果已经扁平 + if 'title' in metadata: + duration = metadata.get('duration', 0) + if isinstance(duration, str) and ':' in duration: + parts = duration.split(':') + try: + metadata['duration'] = int(parts[0]) * 60 + int(parts[1]) + except: + pass + return metadata + + # 嵌套结构 + if 'metadata' in metadata: + meta = metadata['metadata'] + normalized = {} + for k, v in meta.items(): + normalized[k] = v + # 处理 duration + duration = normalized.get('duration', 0) + if isinstance(duration, str) and ':' in duration: + parts = duration.split(':') + try: + normalized['duration'] = int(parts[0]) * 60 + int(parts[1]) + except: + pass + return normalized + + return metadata + + def _format_duration(self, seconds: int) -> str: + """格式化时长""" + h = seconds // 3600 + m = (seconds % 3600) // 60 + s = seconds % 60 + if h > 0: + return f"{h}:{m:02d}:{s:02d}" + return f"{m}:{s:02d}" + + def _clean_author(self, author: str) -> str: + """清理 author 字段：移除零宽度空格和前后空白""" + if not author: + return '' + # 移除零宽度空格 (U+200B) 和其他不可见字符 + invisible_chars = ['\u200b', '\u200c', '\u200d', '\ufeff', '\u00ad'] + cleaned = author + for char in invisible_chars: + cleaned = cleaned.replace(char, '') + # 移除前后空白和换行 + cleaned = cleaned.strip() + # 如果包含换行符，只取第一行 + if '\n' in cleaned: + cleaned = cleaned.split('\n')[0].strip() + return cleaned + + +# ============== 工具函数 ============== + +def load_json(file_path: str) -> Dict: + """加载 JSON 文件""" + if not file_path or not os.path.exists(file_path): + return {} + with open(file_path, 'r', encoding='utf-8') as f: + return json.load(f) + + +def generate_frontmatter(metadata: Dict, result: AnalysisResult, content_type: str = "video") -> str: + """生成 Markdown frontmatter""" + # 提取标签 + tags = result.recommended_tags or [] + + # 格式化日期 + pubtime = metadata.get('pubtime', '') + if pubtime: + date_str = str(pubtime)[:10] + else: + date_str = datetime.now().strftime('%Y-%m-%d') + + # 清理 author 字段 + author = metadata.get('author', '未知') + author = _clean_string(author) + + # 推断 platform + url = metadata.get('url', '') + platform = metadata.get('platform', '') + if not platform: + if 'zhihu.com' in url or content_type in ('answer', 'zhuanlan', 'question'): + platform = 'zhihu' + elif 'bilibili.com' in url or metadata.get('bvid'): + platform = 'bilibili' + else: + platform = 'bilibili' + + fm = ["---", + f"title: \"{metadata.get('title', '未命名')}\"", + f"date: {date_str}", + f"author: {author}", + f"platform: {platform}"] + + # 视频类型添加 duration 和 bvid + if content_type == "video": + fm.insert(3, f"duration: {metadata.get('duration_str', metadata.get('duration', 0))}") + bvid = metadata.get('bvid', '') + if bvid: + fm.insert(5, f"bvid: {bvid}") + + # 添加 URL + url = metadata.get('url', '') + if url: + fm.append(f"url: {url}") + + if tags: + fm.append("tags: [" + ", ".join(f'"{t}"' for t in tags) + "]") + + fm.append("---") + return '\n'.join(fm) + + +def _clean_string(text: str) -> str: + """清理字符串：移除零宽度空格和前后空白""" + if not text: + return '' + # 移除零宽度空格 (U+200B) 和其他不可见字符 + invisible_chars = ['\u200b', '\u200c', '\u200d', '\ufeff', '\u00ad'] + cleaned = text + for char in invisible_chars: + cleaned = cleaned.replace(char, '') + # 移除前后空白和换行 + cleaned = cleaned.strip() + # 如果包含换行符，只取第一行 + if '\n' in cleaned: + cleaned = cleaned.split('\n')[0].strip() + return cleaned + + +# ============== 主程序 ============== + +def main(): + parser = argparse.ArgumentParser(description="Deep Analyzer - AI 深度内容分析工具") + parser.add_argument("--metadata", "-m", required=True, help="元数据 JSON 文件路径") + parser.add_argument("--content", "-c", required=True, help="内容 JSON 文件路径") + parser.add_argument("--output", "-o", required=True, help="输出 Markdown 文件路径") + parser.add_argument("--type", "-t", default="video", choices=["video", "article"], help="内容类型") + parser.add_argument("--api-key", "-k", default=None, help="AI API Key") + parser.add_argument("--provider", "-p", default=None, choices=["openai", "zhipu", "minimax"], help="AI 提供商") + parser.add_argument("--model", default=None, help="AI 模型") + parser.add_argument("--base-url", "-u", default=None, help="API Base URL") + parser.add_argument("--no-frontmatter", action="store_true", help="不生成 frontmatter") + + args = parser.parse_args() + + # 加载数据 + print(f"[Main] 加载元数据: {args.metadata}") + raw_metadata = load_json(args.metadata) + + print(f"[Main] 加载内容: {args.content}") + content = load_json(args.content) + + # 标准化元数据 + metadata = {} + if 'title' in raw_metadata: + metadata = raw_metadata.copy() + # 处理 duration + duration = metadata.get('duration', 0) + if isinstance(duration, str) and ':' in duration: + parts = duration.split(':') + try: + metadata['duration'] = int(parts[0]) * 60 + int(parts[1]) + except: + pass + metadata['duration_str'] = format_duration(metadata.get('duration', 0)) + elif 'metadata' in raw_metadata: + meta = raw_metadata['metadata'] + metadata = meta.copy() + duration = metadata.get('duration', 0) + if isinstance(duration, str) and ':' in duration: + parts = duration.split(':') + try: + metadata['duration'] = int(parts[0]) * 60 + int(parts[1]) + except: + pass + metadata['duration_str'] = format_duration(metadata.get('duration', 0)) + + # 初始化 + ai_client = SimpleAIClient( + api_key=args.api_key or os.environ.get("AI_API_KEY", ""), + provider=args.provider or os.environ.get("AI_PROVIDER", ""), + model=args.model or os.environ.get("AI_MODEL", "glm-5"), + base_url=args.base_url or os.environ.get("AI_BASE_URL", "") + ) + analyzer = DeepAnalyzer(ai_client) + + # 执行分析 + print(f"[Main] 开始分析...") + result = analyzer.analyze(metadata, content, args.type) + + # 使用 AI 生成的原始内容（保留 Markdown 格式） + full_analysis = result.full_analysis + + # 删除内容中的"标签" section（只保留 frontmatter 中的 tags） + # 匹配 ## 标签 或 标签（无##）到下一个 ## 或文件结尾之间的内容 + full_analysis = re.sub(r'(## ?)?标签\n[\s\S]*?(?=\n## ?|\n时间轴|---|\Z)', '', full_analysis) + # 清理多余的空行 + full_analysis = re.sub(r'\n{3,}', '\n\n', full_analysis) + + # 文章类型：追加原文 + if args.type == "article" and result.original_text: + full_analysis = full_analysis.rstrip() + "\n\n## 原文\n\n" + result.original_text + + # 写入输出 + os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True) + + with open(args.output, 'w', encoding='utf-8') as f: + # 添加 frontmatter + if not args.no_frontmatter: + fm = generate_frontmatter(metadata, result, args.type) + f.write(fm + "\n\n") + + # 写入分析内容 + f.write(full_analysis) + + # 输出结果 + output = { + "success": True, + "filename": Path(args.output).name, + "filepath": str(Path(args.output).resolve()), + "message": "深度分析完成", + "stats": { + "summary_length": len(result.summary) if result.summary else 0, + "topics_count": len(result.main_topics), + "insights_count": len(result.key_insights), + "tags_count": len(result.recommended_tags) + } + } + print(json.dumps(output, ensure_ascii=False, indent=2)) + + +def format_duration(seconds: int) -> str: + """格式化时长""" + h = seconds // 3600 + m = (seconds % 3600) // 60 + s = seconds % 60 + if h > 0: + return f"{h}:{m:02d}:{s:02d}" + return f"{m}:{s:02d}" + + +if __name__ == "__main__": + try: + main() + except Exception as e: + print(json.dumps({"success": False, "error": str(e)}, ensure_ascii=False, indent=2)) + sys.exit(1) diff --git a/kb_workflow/pipeline/processors/analyze.py b/kb_workflow/pipeline/processors/analyze.py index 042d75c..36b1851 100644 --- a/kb_workflow/pipeline/processors/analyze.py +++ b/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.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)