戒酒的李白
352 lines
16 KiB
Python
352 lines
16 KiB
Python
import openai
|
|
import anthropic
|
|
import json
|
|
from typing import List, Dict, Tuple, Any
|
|
import os
|
|
import asyncio
|
|
import math
|
|
from datetime import datetime
|
|
from utils.logger import app_logger as logging
|
|
|
|
class AIAnalyzer:
|
|
def __init__(self):
|
|
# 尝试从环境变量中获取API密钥，如果没有则主动询问配置
|
|
self.openai_key = os.getenv('OPENAI_API_KEY')
|
|
if not self.openai_key:
|
|
print("未检测到 OPENAI_API_KEY。")
|
|
# 提示时允许按回车跳过输入
|
|
self.openai_key = input("请输入 OPENAI_API_KEY (按回车键跳过输入): ").strip()
|
|
|
|
self.claude_key = os.getenv('ANTHROPIC_API_KEY')
|
|
if not self.claude_key:
|
|
print("未检测到 ANTHROPIC_API_KEY。")
|
|
self.claude_key = input("请输入 ANTHROPIC_API_KEY (按回车键跳过输入): ").strip()
|
|
|
|
self.deepseek_key = os.getenv('DEEPSEEK_API_KEY')
|
|
if not self.deepseek_key:
|
|
print("未检测到 DEEPSEEK_API_KEY。")
|
|
self.deepseek_key = input("请输入 DEEPSEEK_API_KEY (按回车键跳过输入): ").strip()
|
|
|
|
# 如果不希望通过交互输入，也可以直接在此处配置（注释掉下面几行即可）
|
|
# self.openai_key = "你的OpenAI_API_KEY"
|
|
# self.claude_key = "你的ANTHROPIC_API_KEY"
|
|
# self.deepseek_key = "你的DEEPSEEK_API_KEY"
|
|
|
|
# 配置各API客户端
|
|
if self.openai_key:
|
|
openai.api_key = self.openai_key
|
|
if self.claude_key:
|
|
self.claude_client = anthropic.Anthropic(api_key=self.claude_key)
|
|
if self.deepseek_key:
|
|
self.deepseek_client = openai.OpenAI(
|
|
api_key=self.deepseek_key,
|
|
base_url="https://api.deepseek.com/v1"
|
|
)
|
|
|
|
# 支持的模型列表（增加了最新的 ChatGPT 和 Claude 模型）
|
|
self.supported_models: Dict[str, Dict[str, Any]] = {
|
|
# OpenAI 最新模型（ChatGPT系列）
|
|
'gpt-4o-latest': {
|
|
'provider': 'openai',
|
|
'max_tokens': 128000, # 支持大窗口
|
|
'cost_per_1k': 0.01 # 参考价格（美元）
|
|
},
|
|
'gpt-4o-mini': {
|
|
'provider': 'openai',
|
|
'max_tokens': 4000, # 轻量版，适合快速任务
|
|
'cost_per_1k': 0.00015 # 成本大幅降低
|
|
},
|
|
# 旧版OpenAI模型
|
|
'gpt-3.5-turbo': {'provider': 'openai', 'max_tokens': 2000, 'cost_per_1k': 0.0015},
|
|
'gpt-3.5-turbo-16k': {'provider': 'openai', 'max_tokens': 16000, 'cost_per_1k': 0.003},
|
|
'gpt-4': {'provider': 'openai', 'max_tokens': 8000, 'cost_per_1k': 0.03},
|
|
'gpt-4-32k': {'provider': 'openai', 'max_tokens': 32000, 'cost_per_1k': 0.06},
|
|
'gpt-4-turbo-preview': {'provider': 'openai', 'max_tokens': 128000, 'cost_per_1k': 0.01},
|
|
|
|
# Anthropic 最新模型（Claude系列）
|
|
'claude-3.5-sonnet-new': {
|
|
'provider': 'anthropic',
|
|
'max_tokens': 200000, # 新版Claude 3.5 Sonnet
|
|
'cost_per_1k': 0.015
|
|
},
|
|
'claude-3.5-haiku': {
|
|
'provider': 'anthropic',
|
|
'max_tokens': 200000, # 最新Claude 3.5 Haiku
|
|
'cost_per_1k': 0.0025
|
|
},
|
|
# 旧版Claude模型
|
|
'claude-2.1': {'provider': 'anthropic', 'max_tokens': 100000, 'cost_per_1k': 0.008},
|
|
'claude-2.0': {'provider': 'anthropic', 'max_tokens': 100000, 'cost_per_1k': 0.008},
|
|
'claude-instant-1.2': {'provider': 'anthropic', 'max_tokens': 100000, 'cost_per_1k': 0.0015},
|
|
|
|
# DeepSeek 模型
|
|
'deepseek-chat': {'provider': 'deepseek', 'max_tokens': 4000, 'cost_per_1k': 0.002},
|
|
'deepseek-reasoner': {'provider': 'deepseek', 'max_tokens': 4000, 'cost_per_1k': 0.003}
|
|
}
|
|
|
|
# 不同深度的分析提示词
|
|
self.prompt_templates: Dict[str, str] = {
|
|
'basic': """你是一个专业的舆情分析助手。请对每条消息进行基础的情感分析。
|
|
请按以下JSON格式返回：
|
|
{
|
|
"analysis_results": [
|
|
{
|
|
"message_id": "消息ID",
|
|
"sentiment": "情感倾向 (积极/消极/中性)",
|
|
"sentiment_score": "情感分数 (0-1)",
|
|
"keywords": ["关键词1", "关键词2"],
|
|
"key_points": "简要概述",
|
|
"influence_analysis": "基础影响分析",
|
|
"risk_level": "风险等级 (低/中/高)",
|
|
"timestamp": "分析时间戳"
|
|
}
|
|
]
|
|
}""",
|
|
'standard': """你是一个专业的舆情分析助手。请对每条消息进行标准深度的分析。
|
|
请按以下JSON格式返回：
|
|
{
|
|
"analysis_results": [
|
|
{
|
|
"message_id": "消息ID",
|
|
"sentiment": "情感倾向 (积极/消极/中性)",
|
|
"sentiment_score": "情感分数 (0-1)",
|
|
"keywords": ["关键词1", "关键词2", "关键词3"],
|
|
"key_points": "核心观点概述",
|
|
"influence_analysis": "潜在影响分析",
|
|
"risk_level": "风险等级 (低/中/高)",
|
|
"timestamp": "分析时间戳"
|
|
}
|
|
]
|
|
}""",
|
|
'deep': """你是一个专业的舆情分析助手。请对每条消息进行深度分析。
|
|
请按以下JSON格式返回：
|
|
{
|
|
"analysis_results": [
|
|
{
|
|
"message_id": "消息ID",
|
|
"sentiment": "情感倾向 (积极/消极/中性)",
|
|
"sentiment_score": "情感分数 (0-1)",
|
|
"keywords": ["关键词1", "关键词2", "关键词3", "关键词4", "关键词5"],
|
|
"key_points": "详细的核心观点分析",
|
|
"influence_analysis": "深度影响分析，包括短期和长期影响",
|
|
"risk_factors": ["风险因素1", "风险因素2", "风险因素3"],
|
|
"risk_level": "风险等级 (低/中/高)",
|
|
"suggestions": ["建议1", "建议2", "建议3"],
|
|
"timestamp": "分析时间戳"
|
|
}
|
|
]
|
|
}"""
|
|
}
|
|
|
|
async def analyze_messages(self, messages: List[Dict], batch_size: int = 50,
|
|
model_type: str = "gpt-3.5-turbo",
|
|
analysis_depth: str = "standard",
|
|
prefer_deepseek: bool = True) -> List[Dict]:
|
|
"""
|
|
分析一批消息并返回分析结果。
|
|
如果 DeepSeek API 可用且 prefer_deepseek 为 True，则优先使用 DeepSeek 模型。
|
|
"""
|
|
try:
|
|
# 优先使用 DeepSeek 模型以降低成本
|
|
if prefer_deepseek and self.deepseek_key:
|
|
if model_type not in ['deepseek-chat', 'deepseek-reasoner']:
|
|
logging.info("检测到 DeepSeek API, 优先使用 'deepseek-chat' 模型以降低成本。")
|
|
model_type = 'deepseek-chat'
|
|
|
|
if model_type not in self.supported_models:
|
|
raise ValueError(f"不支持的模型类型: {model_type}")
|
|
|
|
model_info = self.supported_models[model_type]
|
|
provider = model_info['provider']
|
|
max_tokens = model_info['max_tokens']
|
|
|
|
# 根据模型类型调整批处理大小
|
|
optimal_batch_size = self._get_optimal_batch_size(model_type)
|
|
adjusted_batch_size = min(batch_size, optimal_batch_size)
|
|
if adjusted_batch_size != batch_size:
|
|
logging.info(f"已将批处理大小从 {batch_size} 调整为 {adjusted_batch_size}")
|
|
|
|
tasks = []
|
|
total_cost = 0.0
|
|
# 分批处理消息并异步调用分析任务
|
|
for i in range(0, len(messages), adjusted_batch_size):
|
|
batch = messages[i:i + adjusted_batch_size]
|
|
system_prompt = self.prompt_templates.get(analysis_depth, self.prompt_templates['standard'])
|
|
tasks.append(self._process_batch(batch, system_prompt, model_type, max_tokens, provider))
|
|
|
|
# 并发执行所有批次任务
|
|
results = await asyncio.gather(*tasks)
|
|
|
|
all_results = []
|
|
for batch_result, batch_cost in results:
|
|
all_results.extend(batch_result)
|
|
total_cost += batch_cost
|
|
|
|
logging.info(f"分析完成, 总成本: ${total_cost:.4f}")
|
|
return all_results
|
|
except Exception as e:
|
|
logging.error(f"AI分析过程出错: {e}", exc_info=True)
|
|
return []
|
|
|
|
async def _process_batch(self, batch: List[Dict], system_prompt: str,
|
|
model_type: str, max_tokens: int, provider: str) -> Tuple[List[Dict], float]:
|
|
"""
|
|
处理单个批次的消息，返回 (分析结果, 本批次成本)
|
|
"""
|
|
try:
|
|
formatted_messages = [
|
|
f"消息ID: {msg.get('id')}\n内容: {msg.get('content')}" for msg in batch
|
|
]
|
|
messages_text = "\n---\n".join(formatted_messages)
|
|
|
|
if provider == 'openai':
|
|
result = await self._analyze_with_openai(messages_text, system_prompt, model_type, max_tokens)
|
|
elif provider == 'anthropic':
|
|
result = await self._analyze_with_claude(messages_text, system_prompt, model_type, max_tokens)
|
|
elif provider == 'deepseek':
|
|
result = await self._analyze_with_deepseek(messages_text, system_prompt, model_type, max_tokens)
|
|
else:
|
|
logging.error(f"未知的API供应商: {provider}")
|
|
return ([], 0.0)
|
|
|
|
batch_cost = self._calculate_cost(len(messages_text), model_type)
|
|
logging.info(f"批次处理完成, 成本: ${batch_cost:.4f}")
|
|
return (result, batch_cost)
|
|
except Exception as e:
|
|
logging.error(f"处理批次时出错: {e}", exc_info=True)
|
|
return ([], 0.0)
|
|
|
|
def _get_optimal_batch_size(self, model_type: str) -> int:
|
|
"""根据模型类型获取最优批处理大小"""
|
|
model_info = self.supported_models[model_type]
|
|
max_tokens = model_info['max_tokens']
|
|
|
|
# 估算每条消息的平均 token 数（假设为 200）
|
|
avg_tokens_per_message = 200
|
|
# 预留 20% 的 token 用于系统提示词和响应
|
|
available_tokens = int(max_tokens * 0.8)
|
|
optimal_batch_size = max(1, min(100, available_tokens // avg_tokens_per_message))
|
|
return optimal_batch_size
|
|
|
|
def _calculate_cost(self, input_length: int, model_type: str) -> float:
|
|
"""计算 API 调用成本"""
|
|
model_info = self.supported_models[model_type]
|
|
cost_per_1k = model_info['cost_per_1k']
|
|
# 估算 token 数（假设每 4 个字符约等于 1 个 token）
|
|
estimated_tokens = math.ceil(input_length / 4)
|
|
cost = (estimated_tokens / 1000) * cost_per_1k
|
|
return cost
|
|
|
|
async def _analyze_with_openai(self, messages_text: str, system_prompt: str,
|
|
model: str, max_tokens: int) -> List[Dict]:
|
|
"""使用 OpenAI API 进行分析"""
|
|
try:
|
|
response = await openai.ChatCompletion.acreate(
|
|
model=model,
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": f"请分析以下消息:\n{messages_text}"}
|
|
],
|
|
temperature=0.3,
|
|
max_tokens=max_tokens,
|
|
n=1
|
|
)
|
|
content = response.choices[0].message.content
|
|
result = json.loads(content)
|
|
if isinstance(result, dict) and 'analysis_results' in result:
|
|
return result['analysis_results']
|
|
else:
|
|
logging.error(f"OpenAI API返回格式不正确: {content}")
|
|
return []
|
|
except Exception as e:
|
|
logging.error(f"OpenAI API调用失败: {e}", exc_info=True)
|
|
return []
|
|
|
|
async def _analyze_with_claude(self, messages_text: str, system_prompt: str,
|
|
model: str, max_tokens: int) -> List[Dict]:
|
|
"""使用 Claude API 进行分析"""
|
|
try:
|
|
response = await self.claude_client.messages.create(
|
|
model=model,
|
|
max_tokens=max_tokens,
|
|
temperature=0.3,
|
|
system=system_prompt,
|
|
messages=[{"role": "user", "content": f"请分析以下消息:\n{messages_text}"}]
|
|
)
|
|
content = response.content[0].text
|
|
result = json.loads(content)
|
|
if isinstance(result, dict) and 'analysis_results' in result:
|
|
return result['analysis_results']
|
|
else:
|
|
logging.error(f"Claude API返回格式不正确: {content}")
|
|
return []
|
|
except Exception as e:
|
|
logging.error(f"Claude API调用失败: {e}", exc_info=True)
|
|
return []
|
|
|
|
async def _analyze_with_deepseek(self, messages_text: str, system_prompt: str,
|
|
model: str, max_tokens: int) -> List[Dict]:
|
|
"""使用 DeepSeek API 进行分析"""
|
|
try:
|
|
response = await self.deepseek_client.chat.completions.create(
|
|
model=model,
|
|
messages=[
|
|
{"role": "system", "content": system_prompt},
|
|
{"role": "user", "content": f"请分析以下消息:\n{messages_text}"}
|
|
],
|
|
temperature=0.3,
|
|
max_tokens=max_tokens
|
|
)
|
|
content = response.choices[0].message.content
|
|
result = json.loads(content)
|
|
if isinstance(result, dict) and 'analysis_results' in result:
|
|
return result['analysis_results']
|
|
else:
|
|
logging.error(f"DeepSeek API返回格式不正确: {content}")
|
|
return []
|
|
except Exception as e:
|
|
logging.error(f"DeepSeek API调用失败: {e}", exc_info=True)
|
|
return []
|
|
|
|
def format_analysis_for_display(self, analysis: Dict) -> Dict:
|
|
"""将分析结果格式化为前端显示格式"""
|
|
base_result = {
|
|
'id': analysis.get('message_id', ''),
|
|
'sentiment': analysis.get('sentiment', ''),
|
|
'sentiment_score': f"{float(analysis.get('sentiment_score', 0)):.2%}",
|
|
'keywords': ', '.join(analysis.get('keywords', [])),
|
|
'key_points': analysis.get('key_points', ''),
|
|
'influence': analysis.get('influence_analysis', ''),
|
|
'risk_level': analysis.get('risk_level', ''),
|
|
'analysis_time': datetime.fromtimestamp(
|
|
float(analysis.get('timestamp', 0))
|
|
).strftime('%Y-%m-%d %H:%M:%S')
|
|
}
|
|
|
|
# 如果是深度分析，添加额外信息
|
|
if 'risk_factors' in analysis:
|
|
base_result.update({
|
|
'risk_factors': analysis.get('risk_factors', []),
|
|
'suggestions': analysis.get('suggestions', [])
|
|
})
|
|
|
|
return base_result
|
|
|
|
# 创建全局 AI 分析器实例
|
|
ai_analyzer = AIAnalyzer()
|
|
|
|
# 若需要直接配置或测试，可在此处编写测试代码
|
|
if __name__ == "__main__":
|
|
# 示例：直接配置并调用分析器（可替换为实际测试代码）
|
|
sample_messages = [
|
|
{"id": "1", "content": "今天天气真好，我很开心。"},
|
|
{"id": "2", "content": "经济形势不容乐观，风险较大。"}
|
|
]
|
|
|
|
async def test():
|
|
results = await ai_analyzer.analyze_messages(sample_messages, model_type="gpt-4o-latest", analysis_depth="standard")
|
|
for res in results:
|
|
print(ai_analyzer.format_analysis_for_display(res))
|
|
|
|
asyncio.run(test())
|