Ubuntu
303 lines
11 KiB
Python
Executable File
303 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
AI模型处理器 - 自动跟随nanobot配置的模型
|
|
用于文章翻译和总结
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
import requests
|
|
from typing import Dict, Optional, Tuple
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class AIProcessor:
|
|
"""AI模型处理器 - 自动使用nanobot配置的模型"""
|
|
|
|
def __init__(self, config_path=None):
|
|
"""初始化处理器"""
|
|
self.config_path = config_path or "/home/ubuntu/nanocode/it_automation_enhanced/config/config.json"
|
|
self.load_config()
|
|
|
|
# 导入配置加载器
|
|
try:
|
|
from config_loader import get_current_model_config
|
|
self.model_config = get_current_model_config()
|
|
self.provider = self.model_config['provider']
|
|
self.model = self.model_config['model']
|
|
logger.info(f"使用nanobot配置的模型: {self.provider}/{self.model}")
|
|
except Exception as e:
|
|
logger.warning(f"无法加载配置加载器: {e}，使用默认配置")
|
|
self.model_config = None
|
|
self.provider = 'moonshot'
|
|
self.model = 'kimi-k2.5'
|
|
|
|
def load_config(self):
|
|
"""加载项目配置"""
|
|
try:
|
|
with open(self.config_path, 'r', encoding='utf-8') as f:
|
|
self.config = json.load(f)
|
|
logger.info(f"项目配置加载成功")
|
|
except Exception as e:
|
|
logger.error(f"配置加载失败: {e}")
|
|
self.config = {}
|
|
|
|
def call_api(self, prompt: str, max_tokens: int = None) -> Optional[str]:
|
|
"""调用AI API - 自动使用nanobot配置的模型"""
|
|
|
|
# 使用配置加载器获取的模型配置
|
|
if self.model_config:
|
|
api_key = self.model_config['api_key']
|
|
api_base = self.model_config['api_base']
|
|
model = self.model_config['model']
|
|
default_max_tokens = self.model_config['max_tokens']
|
|
else:
|
|
# 回退到项目配置
|
|
provider_config = self.config.get(self.provider, {})
|
|
api_key = provider_config.get('api_key')
|
|
api_base = provider_config.get('api_base', 'https://api.moonshot.cn/v1')
|
|
model = provider_config.get('model', 'kimi-k2.5')
|
|
default_max_tokens = 4000
|
|
|
|
if not api_key or api_key.startswith('{{'):
|
|
logger.warning(f"{self.provider} API密钥未配置，跳过API调用")
|
|
return None
|
|
|
|
headers = {
|
|
'Authorization': f'Bearer {api_key}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
data = {
|
|
'model': model,
|
|
'messages': [
|
|
{'role': 'user', 'content': prompt}
|
|
],
|
|
'max_tokens': max_tokens or default_max_tokens
|
|
}
|
|
|
|
# 构建API端点URL
|
|
if api_base.endswith('/v1'):
|
|
endpoint = f'{api_base}/chat/completions'
|
|
else:
|
|
endpoint = f'{api_base}/chat/completions'
|
|
|
|
try:
|
|
logger.info(f"调用 {self.provider} API: {model}")
|
|
response = requests.post(
|
|
endpoint,
|
|
headers=headers,
|
|
json=data,
|
|
timeout=60
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
result = response.json()
|
|
content = result['choices'][0]['message']['content']
|
|
logger.info(f"{self.provider} API调用成功，返回 {len(content)} 字符")
|
|
return content
|
|
else:
|
|
logger.error(f"{self.provider} API调用失败: {response.status_code} - {response.text[:200]}")
|
|
return None
|
|
|
|
except Exception as e:
|
|
logger.error(f"{self.provider} API调用异常: {e}")
|
|
return None
|
|
|
|
def translate_article(self, content: str, source_lang: str = 'en', target_lang: str = 'zh') -> Optional[str]:
|
|
"""翻译文章"""
|
|
if not self.config.get('processing', {}).get('enable_translation', True):
|
|
logger.info("翻译功能已禁用，跳过")
|
|
return content
|
|
|
|
# 使用配置加载器的提示词或项目配置的提示词
|
|
if self.model_config:
|
|
prompt_template = self.model_config.get('translation_prompt',
|
|
f"请将以下{source_lang}文技术文章翻译成{target_lang}文，保持技术准确性，语言自然流畅：\n\n{{content}}")
|
|
else:
|
|
provider_config = self.config.get(self.provider, {})
|
|
prompt_template = provider_config.get('translation_prompt',
|
|
f"请将以下{source_lang}文技术文章翻译成{target_lang}文，保持技术准确性，语言自然流畅：\n\n{{content}}")
|
|
|
|
prompt = prompt_template.format(content=content)
|
|
|
|
# 如果内容太长，分段处理
|
|
max_length = self.config.get('processing', {}).get('max_content_length', 10000)
|
|
if len(content) > max_length:
|
|
logger.warning(f"文章过长 ({len(content)}字符)，将分段翻译")
|
|
return self._translate_long_content(content, prompt, max_length)
|
|
|
|
translated = self.call_api(prompt)
|
|
return translated
|
|
|
|
def _translate_long_content(self, content: str, prompt_template: str, chunk_size: int = 3000) -> Optional[str]:
|
|
"""翻译长内容（分段处理）"""
|
|
# 按段落分割
|
|
paragraphs = content.split('\n\n')
|
|
translated_paragraphs = []
|
|
|
|
current_chunk = ""
|
|
chunk_count = 0
|
|
|
|
for i, paragraph in enumerate(paragraphs):
|
|
if len(current_chunk) + len(paragraph) > chunk_size:
|
|
# 翻译当前块
|
|
chunk_count += 1
|
|
logger.info(f"翻译第 {chunk_count} 块 ({len(current_chunk)} 字符)")
|
|
|
|
prompt = prompt_template.format(content=current_chunk)
|
|
translated_chunk = self.call_api(prompt)
|
|
|
|
if translated_chunk:
|
|
translated_paragraphs.append(translated_chunk)
|
|
else:
|
|
logger.warning(f"第 {chunk_count} 块翻译失败")
|
|
translated_paragraphs.append(current_chunk) # 保留原文
|
|
|
|
current_chunk = paragraph
|
|
else:
|
|
if current_chunk:
|
|
current_chunk += "\n\n" + paragraph
|
|
else:
|
|
current_chunk = paragraph
|
|
|
|
# 翻译最后一块
|
|
if current_chunk:
|
|
chunk_count += 1
|
|
logger.info(f"翻译第 {chunk_count} 块 ({len(current_chunk)} 字符)")
|
|
|
|
prompt = prompt_template.format(content=current_chunk)
|
|
translated_chunk = self.call_api(prompt)
|
|
|
|
if translated_chunk:
|
|
translated_paragraphs.append(translated_chunk)
|
|
else:
|
|
logger.warning(f"第 {chunk_count} 块翻译失败")
|
|
translated_paragraphs.append(current_chunk)
|
|
|
|
return "\n\n".join(translated_paragraphs)
|
|
|
|
def summarize_article(self, content: str) -> Optional[str]:
|
|
"""总结文章"""
|
|
if not self.config.get('processing', {}).get('enable_summarization', True):
|
|
logger.info("总结功能已禁用，跳过")
|
|
return None
|
|
|
|
# 使用配置加载器的提示词或项目配置的提示词
|
|
if self.model_config:
|
|
prompt_template = self.model_config.get('summarization_prompt',
|
|
"请用中文总结以下技术文章的核心内容，突出技术要点和创新点：\n\n{content}")
|
|
else:
|
|
provider_config = self.config.get(self.provider, {})
|
|
prompt_template = provider_config.get('summarization_prompt',
|
|
"请用中文总结以下技术文章的核心内容，突出技术要点和创新点：\n\n{content}")
|
|
|
|
# 限制内容长度
|
|
max_length = self.config.get('processing', {}).get('max_content_length', 10000)
|
|
if len(content) > max_length:
|
|
content = content[:max_length] + "...[内容已截断]"
|
|
|
|
prompt = prompt_template.format(content=content)
|
|
summary = self.call_api(prompt, max_tokens=1000)
|
|
|
|
return summary
|
|
|
|
def rewrite_article(self, content: str) -> Optional[str]:
|
|
"""改写文章（中文润色）"""
|
|
if not self.config.get('processing', {}).get('enable_translation', True):
|
|
logger.info("改写功能已禁用，跳过")
|
|
return content
|
|
|
|
prompt = f"""请对以下技术文章进行改写和润色：
|
|
|
|
要求：
|
|
1. 保持原意和技术准确性
|
|
2. 调整句式结构，避免重复表达
|
|
3. 优化语言流畅度
|
|
4. 保留关键技术术语
|
|
5. 输出约原文80%长度的内容
|
|
|
|
原文：
|
|
{content[:8000]}
|
|
"""
|
|
|
|
rewritten = self.call_api(prompt)
|
|
return rewritten
|
|
|
|
def process_article(self, title: str, content: str, url: str = "") -> Dict:
|
|
"""完整处理文章"""
|
|
logger.info(f"开始处理文章: {title}")
|
|
|
|
result = {
|
|
'title': title,
|
|
'original_content': content,
|
|
'original_length': len(content),
|
|
'url': url,
|
|
'translated': None,
|
|
'summary': None,
|
|
'processed': False
|
|
}
|
|
|
|
# 1. 翻译
|
|
if self.config.get('processing', {}).get('enable_translation', True):
|
|
logger.info("翻译文章...")
|
|
translated = self.translate_article(content)
|
|
if translated:
|
|
result['translated'] = translated
|
|
result['translated_length'] = len(translated)
|
|
logger.info(f"翻译完成: {len(translated)} 字符")
|
|
else:
|
|
logger.warning("翻译失败")
|
|
|
|
# 2. 总结
|
|
if self.config.get('processing', {}).get('enable_summarization', True):
|
|
logger.info("总结文章...")
|
|
summary = self.summarize_article(content)
|
|
if summary:
|
|
result['summary'] = summary
|
|
logger.info(f"总结完成: {len(summary)} 字符")
|
|
else:
|
|
logger.warning("总结失败")
|
|
|
|
result['processed'] = True
|
|
logger.info(f"文章处理完成: {title}")
|
|
|
|
return result
|
|
|
|
def test_processor():
|
|
"""测试处理器"""
|
|
import sys
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
processor = AIProcessor()
|
|
|
|
# 测试内容
|
|
test_content = """Artificial intelligence is transforming the technology landscape.
|
|
Recent advances in machine learning have enabled breakthroughs in natural language processing,
|
|
computer vision, and autonomous systems. Companies are investing heavily in AI research
|
|
and development to gain competitive advantages."""
|
|
|
|
print(f"测试AI处理器 (使用 {processor.provider}/{processor.model})...")
|
|
|
|
# 测试翻译
|
|
print("\n1. 测试翻译...")
|
|
translated = processor.translate_article(test_content)
|
|
if translated:
|
|
print(f"翻译结果: {translated[:100]}...")
|
|
else:
|
|
print("翻译测试跳过（API未配置）")
|
|
|
|
# 测试总结
|
|
print("\n2. 测试总结...")
|
|
summary = processor.summarize_article(test_content)
|
|
if summary:
|
|
print(f"总结结果: {summary[:100]}...")
|
|
else:
|
|
print("总结测试跳过（API未配置）")
|
|
|
|
print("\n测试完成")
|
|
|
|
if __name__ == "__main__":
|
|
test_processor() |