mirror of
https://github.com/Zippland/Snap-Solver.git
synced 2026-02-09 15:57:18 +08:00
gemini pro
This commit is contained in:
Zylan
236
models/google.py
Normal file
236
models/google.py
Normal file
@@ -0,0 +1,236 @@
|
||||
import json
|
||||
import os
|
||||
import base64
|
||||
from typing import Generator, Dict, Any, Optional, List
|
||||
import google.generativeai as genai
|
||||
from .base import BaseModel
|
||||
|
||||
class GoogleModel(BaseModel):
|
||||
"""
|
||||
Google Gemini API模型实现类
|
||||
支持Gemini 2.5 Pro等模型，可处理文本和图像输入
|
||||
"""
|
||||
|
||||
def __init__(self, api_key: str, temperature: float = 0.7, system_prompt: str = None, language: str = None, model_name: str = None):
|
||||
"""
|
||||
初始化Google模型
|
||||
|
||||
Args:
|
||||
api_key: Google API密钥
|
||||
temperature: 生成温度
|
||||
system_prompt: 系统提示词
|
||||
language: 首选语言
|
||||
model_name: 指定具体模型名称，如不指定则使用默认值
|
||||
"""
|
||||
super().__init__(api_key, temperature, system_prompt, language)
|
||||
self.model_name = model_name or self.get_model_identifier()
|
||||
self.max_tokens = 8192 # 默认最大输出token数
|
||||
|
||||
# 配置Google API
|
||||
genai.configure(api_key=api_key)
|
||||
|
||||
def get_default_system_prompt(self) -> str:
|
||||
return """You are an expert at analyzing questions and providing detailed solutions. When presented with an image of a question:
|
||||
1. First read and understand the question carefully
|
||||
2. Break down the key components of the question
|
||||
3. Provide a clear, step-by-step solution
|
||||
4. If relevant, explain any concepts or theories involved
|
||||
5. If there are multiple approaches, explain the most efficient one first"""
|
||||
|
||||
def get_model_identifier(self) -> str:
|
||||
"""返回默认的模型标识符"""
|
||||
return "gemini-2.5-pro-preview-03-25"
|
||||
|
||||
def analyze_text(self, text: str, proxies: dict = None) -> Generator[dict, None, None]:
|
||||
"""流式生成文本响应"""
|
||||
try:
|
||||
yield {"status": "started"}
|
||||
|
||||
# 设置环境变量代理（如果提供）
|
||||
original_proxies = None
|
||||
if proxies:
|
||||
original_proxies = {
|
||||
'http_proxy': os.environ.get('http_proxy'),
|
||||
'https_proxy': os.environ.get('https_proxy')
|
||||
}
|
||||
if 'http' in proxies:
|
||||
os.environ['http_proxy'] = proxies['http']
|
||||
if 'https' in proxies:
|
||||
os.environ['https_proxy'] = proxies['https']
|
||||
|
||||
try:
|
||||
# 初始化模型
|
||||
model = genai.GenerativeModel(self.model_name)
|
||||
|
||||
# 获取最大输出Token设置
|
||||
max_tokens = self.max_tokens if hasattr(self, 'max_tokens') else 8192
|
||||
|
||||
# 创建配置参数
|
||||
generation_config = {
|
||||
'temperature': self.temperature,
|
||||
'max_output_tokens': max_tokens,
|
||||
'top_p': 0.95,
|
||||
'top_k': 64,
|
||||
}
|
||||
|
||||
# 构建提示
|
||||
prompt_parts = []
|
||||
|
||||
# 添加系统提示词
|
||||
if self.system_prompt:
|
||||
prompt_parts.append(self.system_prompt)
|
||||
|
||||
# 添加用户查询
|
||||
if self.language and self.language != 'auto':
|
||||
prompt_parts.append(f"请使用{self.language}回答以下问题: {text}")
|
||||
else:
|
||||
prompt_parts.append(text)
|
||||
|
||||
# 初始化响应缓冲区
|
||||
response_buffer = ""
|
||||
|
||||
# 流式生成响应
|
||||
response = model.generate_content(
|
||||
prompt_parts,
|
||||
generation_config=generation_config,
|
||||
stream=True
|
||||
)
|
||||
|
||||
for chunk in response:
|
||||
if not chunk.text:
|
||||
continue
|
||||
|
||||
# 累积响应文本
|
||||
response_buffer += chunk.text
|
||||
|
||||
# 发送响应进度
|
||||
if len(chunk.text) >= 10 or chunk.text.endswith(('.', '!', '?', '。', '！', '？', '\n')):
|
||||
yield {
|
||||
"status": "streaming",
|
||||
"content": response_buffer
|
||||
}
|
||||
|
||||
# 确保发送完整的最终内容
|
||||
yield {
|
||||
"status": "completed",
|
||||
"content": response_buffer
|
||||
}
|
||||
|
||||
finally:
|
||||
# 恢复原始代理设置
|
||||
if original_proxies:
|
||||
for key, value in original_proxies.items():
|
||||
if value is None:
|
||||
if key in os.environ:
|
||||
del os.environ[key]
|
||||
else:
|
||||
os.environ[key] = value
|
||||
|
||||
except Exception as e:
|
||||
yield {
|
||||
"status": "error",
|
||||
"error": f"Gemini API错误: {str(e)}"
|
||||
}
|
||||
|
||||
def analyze_image(self, image_data: str, proxies: dict = None) -> Generator[dict, None, None]:
|
||||
"""分析图像并流式生成响应"""
|
||||
try:
|
||||
yield {"status": "started"}
|
||||
|
||||
# 设置环境变量代理（如果提供）
|
||||
original_proxies = None
|
||||
if proxies:
|
||||
original_proxies = {
|
||||
'http_proxy': os.environ.get('http_proxy'),
|
||||
'https_proxy': os.environ.get('https_proxy')
|
||||
}
|
||||
if 'http' in proxies:
|
||||
os.environ['http_proxy'] = proxies['http']
|
||||
if 'https' in proxies:
|
||||
os.environ['https_proxy'] = proxies['https']
|
||||
|
||||
try:
|
||||
# 初始化模型
|
||||
model = genai.GenerativeModel(self.model_name)
|
||||
|
||||
# 获取最大输出Token设置
|
||||
max_tokens = self.max_tokens if hasattr(self, 'max_tokens') else 8192
|
||||
|
||||
# 创建配置参数
|
||||
generation_config = {
|
||||
'temperature': self.temperature,
|
||||
'max_output_tokens': max_tokens,
|
||||
'top_p': 0.95,
|
||||
'top_k': 64,
|
||||
}
|
||||
|
||||
# 构建提示词
|
||||
prompt_parts = []
|
||||
|
||||
# 添加系统提示词
|
||||
if self.system_prompt:
|
||||
prompt_parts.append(self.system_prompt)
|
||||
|
||||
# 添加默认图像分析指令
|
||||
if self.language and self.language != 'auto':
|
||||
prompt_parts.append(f"请使用{self.language}分析这张图片并提供详细解答。")
|
||||
else:
|
||||
prompt_parts.append("请分析这张图片并提供详细解答。")
|
||||
|
||||
# 处理图像数据
|
||||
if image_data.startswith('data:image'):
|
||||
# 如果是data URI，提取base64部分
|
||||
image_data = image_data.split(',', 1)[1]
|
||||
|
||||
# 使用genai的特定方法处理图像
|
||||
image_part = {
|
||||
"mime_type": "image/jpeg",
|
||||
"data": base64.b64decode(image_data)
|
||||
}
|
||||
prompt_parts.append(image_part)
|
||||
|
||||
# 初始化响应缓冲区
|
||||
response_buffer = ""
|
||||
|
||||
# 流式生成响应
|
||||
response = model.generate_content(
|
||||
prompt_parts,
|
||||
generation_config=generation_config,
|
||||
stream=True
|
||||
)
|
||||
|
||||
for chunk in response:
|
||||
if not chunk.text:
|
||||
continue
|
||||
|
||||
# 累积响应文本
|
||||
response_buffer += chunk.text
|
||||
|
||||
# 发送响应进度
|
||||
if len(chunk.text) >= 10 or chunk.text.endswith(('.', '!', '?', '。', '！', '？', '\n')):
|
||||
yield {
|
||||
"status": "streaming",
|
||||
"content": response_buffer
|
||||
}
|
||||
|
||||
# 确保发送完整的最终内容
|
||||
yield {
|
||||
"status": "completed",
|
||||
"content": response_buffer
|
||||
}
|
||||
|
||||
finally:
|
||||
# 恢复原始代理设置
|
||||
if original_proxies:
|
||||
for key, value in original_proxies.items():
|
||||
if value is None:
|
||||
if key in os.environ:
|
||||
del os.environ[key]
|
||||
else:
|
||||
os.environ[key] = value
|
||||
|
||||
except Exception as e:
|
||||
yield {
|
||||
"status": "error",
|
||||
"error": f"Gemini图像分析错误: {str(e)}"
|
||||
}
|
||||
Reference in New Issue
Block a user