starmen1265/openclaw-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(kb-deep-analyzer): 实现每段前加概述功能

Browse Source 
- 添加元数据（标题、描述、来源）到每段分析提示词
- 移除 mock 相关代码，使用真实 AI（zhipu/glm-4.7）
- 配置环境变量支持 AI_API_KEY、AI_PROVIDER、AI_MODEL、AI_BASE_URL
This commit is contained in:
starmen1265
2026-02-13 13:00:22 +08:00
parent b7b40b1ece
commit 3c422994b9
1 changed files with 39 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/kb-deep-analyzer/deep_analyzer.py
+39 -56
View File
@@ -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", "")
)