OpenClaw Agent
5507b1d986
- kb-link-parser: Multi-platform URL parser (Bilibili, YouTube, Zhihu) - kb-bilibili-metadata: Bilibili video/article/up主 metadata extractor - kb-zhihu-metadata: Zhihu question/answer/user metadata extractor - kb-bilibili-content: Bilibili video download and content extraction - kb-ocr-processor: OCR text extraction from video frames - kb-deep-analyzer: AI-powered deep content analysis - kb-workflow: Workflow orchestration (inbox, archive, subscriptions, report) Also add: - ARCHITECTURE_FLOWCHART.md: System architecture diagrams - DETAILED_DESIGN.md: Complete design documentation
515 lines
19 KiB
Python
515 lines
19 KiB
Python
"""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)
|