From 3c422994b9114f5bdcdda3a2fc3984251a2948f4 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Feb 2026 13:00:22 +0800 Subject: [PATCH] =?UTF-8?q?feat(kb-deep-analyzer):=20=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E6=AF=8F=E6=AE=B5=E5=89=8D=E5=8A=A0=E6=A6=82=E8=BF=B0=E5=8A=9F?= =?UTF-8?q?=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 添加元数据（标题、描述、来源）到每段分析提示词 - 移除 mock 相关代码，使用真实 AI（zhipu/glm-4.7） - 配置环境变量支持 AI_API_KEY、AI_PROVIDER、AI_MODEL、AI_BASE_URL --- skills/kb-deep-analyzer/deep_analyzer.py | 95 ++++++++++-------------- 1 file changed, 39 insertions(+), 56 deletions(-) diff --git a/skills/kb-deep-analyzer/deep_analyzer.py b/skills/kb-deep-analyzer/deep_analyzer.py index c3dd188..a3b4339 100644 --- a/skills/kb-deep-analyzer/deep_analyzer.py +++ b/skills/kb-deep-analyzer/deep_analyzer.py @@ -56,14 +56,14 @@ class SimpleAIClient: 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", "mock") + self.provider = provider or os.environ.get("AI_PROVIDER", "") self.model = model or os.environ.get("AI_MODEL", "gpt-4") self.base_url = base_url or os.environ.get("AI_BASE_URL", "") def generate(self, prompt: str, max_tokens: int = 1000) -> str: """生成文本""" - if not self.api_key or self.provider == "mock": - return self._mock_generate(prompt, max_tokens) + if not self.api_key: + raise ValueError("AI_API_KEY 未设置，请配置环境变量") try: if self.provider == "openai": @@ -73,10 +73,10 @@ class SimpleAIClient: elif self.provider == "minimax": return self._call_minimax(prompt, max_tokens) else: - return self._mock_generate(prompt, max_tokens) + raise ValueError(f"不支持的 AI 提供商: {self.provider}") except Exception as e: print(f"[AI Client] 调用失败: {e}") - return self._mock_generate(prompt, max_tokens) + raise def generate_json(self, prompt: str) -> Dict[str, Any]: """生成 JSON 格式响应""" @@ -189,42 +189,6 @@ class SimpleAIClient: return item.get('text', '') return str(data) - def _mock_generate(self, prompt: str, max_tokens: int) -> str: - """模拟 AI 生成""" - prompt_lower = prompt.lower() - - if 'json' in prompt_lower or '请返回' in prompt: - # 返回模拟 JSON 响应 - if '主题' in prompt or 'topic' in prompt_lower: - return json.dumps({ - 'summary': '这是一个模拟的AI分析结果，包含内容的主要主题和关键见解。', - 'topics': ['主题一', '主题二', '主题三', '主题四'], - 'insights': [ - 'AI正在深刻改变内容消费方式', - '个性化推荐提升了用户体验', - '技术发展带来新的可能性' - ], - 'actions': ['深入研究相关技术', '关注行业动态'], - 'full_analysis': '详细分析内容基于模拟AI生成。' - }, ensure_ascii=False) - if '标签' in prompt or 'tag' in prompt_lower: - return json.dumps({'tags': ['技术', 'AI', '视频分析', '深度学习']}, ensure_ascii=False) - return json.dumps({'result': '模拟响应'}, ensure_ascii=False) - - if '摘要' in prompt or 'summary' in prompt_lower: - return '这是一段关于内容深度分析的重要讨论，涵盖了多个关键主题和见解。' - - return '模拟AI响应内容' - - def _parse_json_from_text(self, text: str) -> Dict[str, Any]: - """从文本中解析 JSON""" - json_match = re.search(r'\{[\s\S]*\}', text) - if json_match: - try: - return json.loads(json_match.group()) - except: - pass - return {'raw_response': text} # ============== 深度分析器 ============== @@ -279,13 +243,13 @@ class DeepAnalyzer: 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) + result = self._analyze_in_chunks_with_segments(title, text, duration, segments, metadata) elif duration < self.SHORT_VIDEO_THRESHOLD: print(f"[DeepAnalyzer] 使用轻量分析策略 (时长 < 10分钟)") result = self._analyze_short_with_subtitle(title, text, duration, segments) else: print(f"[DeepAnalyzer] 使用详细时间轴分析策略 (时长 >= 10分钟)") - result = self._analyze_long_with_subtitle(title, text, duration, segments, content_type) + result = self._analyze_long_with_subtitle(title, text, duration, segments, content_type, metadata) else: print(f"[DeepAnalyzer] 降级处理（无内容）") result = self._analyze_fallback(title, text, metadata) @@ -307,13 +271,14 @@ class DeepAnalyzer: return result def _analyze_long_with_subtitle(self, title: str, text: str, duration: int, - segments: list = None, content_type: str = None) -> AnalysisResult: + segments: list = None, content_type: str = None, + metadata: Dict = None) -> AnalysisResult: """有内容的长视频详细分析""" result = AnalysisResult() if duration >= self.SHORT_VIDEO_THRESHOLD and content_type == "video": # 10分钟以上使用分段分析 - result = self._analyze_in_chunks(title, text, duration) + result = self._analyze_in_chunks(title, text, duration, metadata) else: # 直接分析 text_sample = self._get_text_sample(text, max_chars=5000) @@ -327,7 +292,7 @@ class DeepAnalyzer: return result - def _analyze_in_chunks(self, title: str, subtitle: str, duration: int) -> AnalysisResult: + def _analyze_in_chunks(self, title: str, subtitle: str, duration: int, metadata: Dict = None) -> AnalysisResult: """分段分析长视频 - 动态分段策略""" result = AnalysisResult() @@ -387,8 +352,8 @@ class DeepAnalyzer: start_time = self._format_time(start_sec) end_time = self._format_time(end_sec) - # AI 分析每段 - chunk_summary = self._analyze_chunk(chunk_text, i + 1) + # AI 分析每段（传入元数据以添加概述） + chunk_summary = self._analyze_chunk(chunk_text, i + 1, metadata) chunk_summaries.append({ 'index': i, 'start_time': start_time, @@ -404,7 +369,7 @@ class DeepAnalyzer: return result - def _analyze_in_chunks_with_segments(self, title: str, text: str, duration: int, segments: list = None) -> AnalysisResult: + def _analyze_in_chunks_with_segments(self, title: str, text: str, duration: int, segments: list = None, metadata: Dict = None) -> AnalysisResult: """ 使用实际字幕时间戳进行分块分析 @@ -481,8 +446,8 @@ class DeepAnalyzer: for i, chunk_text in enumerate(chunks): start_time, end_time = chunk_times[i] - # AI 分析每段 - chunk_summary = self._analyze_chunk(chunk_text, i + 1) + # AI 分析每段（传入元数据以添加概述） + chunk_summary = self._analyze_chunk(chunk_text, i + 1, metadata) chunk_summaries.append({ 'index': i, 'start_time': start_time, @@ -548,13 +513,31 @@ class DeepAnalyzer: # 没找到自然断点，返回理想位置 return preferred_end - def _analyze_chunk(self, chunk_text: str, chunk_index: int) -> str: + def _analyze_chunk(self, chunk_text: str, chunk_index: int, metadata: Dict = None) -> str: """分析单个分片""" text_sample = self._get_text_sample(chunk_text, max_chars=3000) + + # 构建概述 + overview = "" + if metadata: + title = metadata.get('title', '') + description = metadata.get('description', '') + source = metadata.get('source', '') + + overview_parts = [] + if title: + overview_parts.append(f"视频标题：{title}") + if description: + overview_parts.append(f"视频描述：{description}") + if source: + overview_parts.append(f"来源：{source}") + + if overview_parts: + overview = "【概述】\n" + "\n".join(overview_parts) + "\n\n" prompt = f"""请对以下视频片段进行详细分析（限制在800字以内）： -片段内容： +{overview}--- 片段内容 --- {text_sample} 请用中文详细回答，包括核心主题、关键要点和重要细节。""" @@ -1225,8 +1208,8 @@ def main(): parser.add_argument("--api-key", "-k", default=None, help="AI API Key (或设置 AI_API_KEY 环境变量)") parser.add_argument("--provider", "-p", default=None, - choices=["openai", "zhipu", "minimax", "mock"], - help="AI 提供商 (默认: mock)") + choices=["openai", "zhipu", "minimax"], + help="AI 提供商") parser.add_argument("--model", default=None, help="AI 模型 (默认: gpt-4)") parser.add_argument("--base-url", "-u", default=None, @@ -1253,7 +1236,7 @@ def main(): # 初始化 ai_client = SimpleAIClient( api_key=args.api_key or os.environ.get("AI_API_KEY", ""), - provider=args.provider or os.environ.get("AI_PROVIDER", "mock"), + provider=args.provider or os.environ.get("AI_PROVIDER", ""), model=args.model or os.environ.get("AI_MODEL", "gpt-4"), base_url=args.base_url or os.environ.get("AI_BASE_URL", "") )