"""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)