mirror of
https://github.com/Zippland/Bubbles.git
synced 2026-02-08 09:12:09 +08:00
优化 Gemini 类的初始化逻辑，新增默认模型、提示和安全设置，改进历史消息处理和错误处理机制，同时增强图片描述功能的 MIME 类型检测，提升代码的可读性和健壮性。
This commit is contained in:
Zylan
@@ -6,9 +6,11 @@ import logging
|
||||
import os
|
||||
import time
|
||||
import httpx
|
||||
import pathlib # 用于处理文件路径
|
||||
import mimetypes # 用于猜测图片类型
|
||||
import google.generativeai as genai
|
||||
from google.generativeai.types import generation_types, safety_types # 显式导入需要的类型
|
||||
from google.api_core.exceptions import GoogleAPICallError, ClientError
|
||||
from google.generativeai.types import BlockedPromptException, StopCandidateException
|
||||
|
||||
# 引入 MessageSummary 类型提示
|
||||
try:
|
||||
@@ -17,60 +19,87 @@ except ImportError:
|
||||
MessageSummary = object # Fallback
|
||||
|
||||
class Gemini:
|
||||
DEFAULT_MODEL = "gemini-1.5-pro-latest"
|
||||
DEFAULT_PROMPT = "You are a helpful assistant."
|
||||
DEFAULT_MAX_HISTORY = 15
|
||||
SAFETY_SETTINGS = { # 默认安全设置 - 可根据需要调整或从配置加载
|
||||
safety_types.HarmCategory.HARM_CATEGORY_HARASSMENT: safety_types.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
|
||||
safety_types.HarmCategory.HARM_CATEGORY_HATE_SPEECH: safety_types.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
|
||||
safety_types.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT: safety_types.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
|
||||
safety_types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT: safety_types.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
|
||||
}
|
||||
|
||||
def __init__(self, conf: dict, message_summary_instance: MessageSummary = None, bot_wxid: str = None) -> None:
|
||||
self._api_key = conf.get("api_key")
|
||||
# 优先使用用户配置的模型，否则默认为 gemini-1.5-pro-latest (一个常用的强大模型)
|
||||
self._model_name = conf.get("model_name", "gemini-1.5-pro-latest")
|
||||
self._prompt = conf.get("prompt")
|
||||
self._proxy = conf.get("proxy")
|
||||
self.max_history_messages = conf.get("max_history_messages", 15) # Gemini 通常上下文窗口较大，可适当增加默认值
|
||||
self.LOG = logging.getLogger("Gemini")
|
||||
self._api_key = conf.get("api_key")
|
||||
self._model_name = conf.get("model_name", self.DEFAULT_MODEL)
|
||||
# 存储原始配置的 prompt，用于初始化和可能的重载
|
||||
self._base_prompt = conf.get("prompt", self.DEFAULT_PROMPT)
|
||||
self._proxy = conf.get("proxy")
|
||||
self.max_history_messages = conf.get("max_history_messages", self.DEFAULT_MAX_HISTORY)
|
||||
|
||||
self.message_summary = message_summary_instance
|
||||
self.bot_wxid = bot_wxid
|
||||
self._model = None
|
||||
self.support_vision = False # 初始化时假设不支持，成功加载模型后再判断
|
||||
|
||||
if not self._api_key:
|
||||
self.LOG.error("Gemini API Key 未在配置中提供！")
|
||||
return # 没有 API Key 无法继续
|
||||
|
||||
if not self.message_summary:
|
||||
self.LOG.warning("MessageSummary 实例未提供给 Gemini，上下文功能将不可用！")
|
||||
if not self.bot_wxid:
|
||||
self.LOG.warning("bot_wxid 未提供给 Gemini，可能无法正确识别机器人自身消息！")
|
||||
|
||||
try:
|
||||
# 配置代理 (如果提供)
|
||||
# 1. 配置代理 (如果提供)
|
||||
transport = None
|
||||
if self._proxy:
|
||||
# 支持 http 和 socks5 代理
|
||||
proxies = {
|
||||
"http://": self._proxy,
|
||||
"https://": self._proxy,
|
||||
}
|
||||
transport = httpx.HTTPTransport(proxy=proxies)
|
||||
self.LOG.info(f"Gemini 使用代理: {self._proxy}")
|
||||
try:
|
||||
proxies = {"http://": self._proxy, "https://": self._proxy}
|
||||
transport = httpx.HTTPTransport(proxy=proxies)
|
||||
self.LOG.info(f"Gemini 使用代理: {self._proxy}")
|
||||
except Exception as proxy_err:
|
||||
self.LOG.error(f"配置 Gemini 代理失败: {proxy_err}", exc_info=True)
|
||||
# 代理配置失败，可以选择不使用代理继续或直接失败
|
||||
# 这里选择继续，不使用代理
|
||||
transport = None
|
||||
|
||||
# 2. 配置 Google AI Client
|
||||
genai.configure(api_key=self._api_key, transport=transport)
|
||||
# 初始化模型
|
||||
|
||||
# 3. 初始化模型
|
||||
# 将基础 prompt 作为 system_instruction 传递
|
||||
self._model = genai.GenerativeModel(
|
||||
self._model_name,
|
||||
# 默认系统提示，如果配置中未提供
|
||||
system_instruction=self._prompt if self._prompt else "You are a helpful assistant."
|
||||
system_instruction=self._base_prompt,
|
||||
safety_settings=self.SAFETY_SETTINGS # 应用安全设置
|
||||
)
|
||||
self.LOG.info(f"Gemini 模型 {self._model_name} 初始化成功")
|
||||
self.LOG.info(f"Gemini 模型 {self._model_name} 初始化成功，基础提示: '{self._base_prompt}'")
|
||||
|
||||
# 检查模型是否支持视觉 (简单检查，更可靠的方式是查询模型能力)
|
||||
# Gemini 1.5 Pro / Flash / 2.5 Pro 等较新模型通常都支持
|
||||
self.support_vision = "vision" in self._model_name or "pro" in self._model_name or "gemini-1.5" in self._model_name
|
||||
# 4. 检查视觉能力 (依赖模型名称的简单检查)
|
||||
# 注意：更可靠的方式是调用 list_models 并检查支持的方法
|
||||
if "vision" in self._model_name or "pro" in self._model_name or "gemini-1.5" in self._model_name or "flash" in self._model_name:
|
||||
self.support_vision = True
|
||||
self.LOG.info(f"模型 {self._model_name} 被认为支持视觉能力。")
|
||||
else:
|
||||
self.LOG.info(f"模型 {self._model_name} 根据名称判断可能不支持视觉能力。")
|
||||
|
||||
except (GoogleAPICallError, ClientError) as api_error:
|
||||
self.LOG.error(f"初始化 Gemini 时发生 API 错误: {api_error}", exc_info=True)
|
||||
self._model = None
|
||||
except Exception as e:
|
||||
self.LOG.error(f"初始化 Gemini 失败: {e}", exc_info=True)
|
||||
self._model = None # 标记初始化失败
|
||||
self.LOG.error(f"初始化 Gemini 时发生未知错误: {e}", exc_info=True)
|
||||
self._model = None
|
||||
|
||||
def __repr__(self):
|
||||
return 'Gemini'
|
||||
return f'Gemini(model={self._model_name}, initialized={self._model is not None})'
|
||||
|
||||
@staticmethod
|
||||
def value_check(conf: dict) -> bool:
|
||||
# 只需要 API Key 是必须的
|
||||
if conf and conf.get("api_key"):
|
||||
return True
|
||||
return False
|
||||
return bool(conf and conf.get("api_key"))
|
||||
|
||||
def _format_history(self, history: list) -> list:
|
||||
"""将数据库历史消息转换为 Gemini API 的 contents 格式"""
|
||||
@@ -78,32 +107,125 @@ class Gemini:
|
||||
for msg in history:
|
||||
role = "model" if msg.get("sender_wxid") == self.bot_wxid else "user"
|
||||
content = msg.get('content', '')
|
||||
sender_name = msg.get('sender', '未知用户') # 获取发送者名称
|
||||
|
||||
if content: # 避免添加空内容
|
||||
# Gemini 推荐 user role 包含发言者信息，model role 不需要
|
||||
if role == "user":
|
||||
sender_name = msg.get('sender', '未知用户')
|
||||
# Gemini API 不直接在消息体中标记发送者，但可以在内容中包含
|
||||
formatted_content = f"{sender_name}: {content}"
|
||||
contents.append({'role': role, 'parts': [formatted_content]})
|
||||
else: # 模型（机器人）的消息
|
||||
contents.append({'role': role, 'parts': [content]})
|
||||
formatted_content = f"[{sender_name}]: {content}" # 添加发送者标记
|
||||
contents.append({'role': role, 'parts': [{'text': formatted_content}]})
|
||||
else: # role == "model"
|
||||
contents.append({'role': role, 'parts': [{'text': content}]})
|
||||
return contents
|
||||
|
||||
def _generate_response(self, contents: list, generation_config_override: generation_types.GenerationConfig | None = None) -> str:
|
||||
"""内部方法，用于调用 Gemini API 并处理响应"""
|
||||
if not self._model:
|
||||
return "Gemini 模型未成功初始化，请检查配置和网络。"
|
||||
|
||||
# 配置生成参数 (可以从 config 中读取更多参数)
|
||||
# 默认使用适中的温度，可以根据需要调整
|
||||
default_config = generation_types.GenerationConfig(temperature=0.7)
|
||||
config_to_use = generation_config_override if generation_config_override else default_config
|
||||
|
||||
self.LOG.debug(f"发送给 Gemini API 的内容条数: {len(contents)}")
|
||||
# self.LOG.debug(f"使用的 GenerationConfig: {config_to_use}")
|
||||
# self.LOG.debug(f"发送内容详情: {contents}") # DEBUG: 打印发送内容
|
||||
|
||||
rsp_text = ""
|
||||
try:
|
||||
# 调用 API
|
||||
response = self._model.generate_content(
|
||||
contents=contents,
|
||||
generation_config=config_to_use,
|
||||
# safety_settings=... # 如果需要覆盖初始化时的安全设置
|
||||
stream=False # 非流式响应
|
||||
)
|
||||
|
||||
# 1. 检查 Prompt 是否被阻止 (在获取 candidates 之前)
|
||||
if response.prompt_feedback and response.prompt_feedback.block_reason:
|
||||
reason = response.prompt_feedback.block_reason.name
|
||||
self.LOG.warning(f"Gemini 提示被阻止，原因: {reason}")
|
||||
return f"抱歉，您的请求因包含不适内容而被阻止 (原因: {reason})。"
|
||||
|
||||
# 2. 检查 Candidates 和 Finish Reason
|
||||
# 尝试从 response 中安全地提取文本
|
||||
try:
|
||||
rsp_text = response.text # .text 是获取聚合文本的便捷方式
|
||||
except ValueError:
|
||||
# 如果 .text 不可用 (例如，因为 finish_reason 不是 STOP 或 MAX_TOKENS)
|
||||
# 检查具体的 finish_reason
|
||||
if response.candidates:
|
||||
candidate = response.candidates[0]
|
||||
finish_reason = candidate.finish_reason # 类型是 FinishReason 枚举
|
||||
finish_reason_name = finish_reason.name
|
||||
|
||||
if finish_reason == generation_types.FinishReason.SAFETY:
|
||||
self.LOG.warning(f"Gemini 响应被安全策略阻止。")
|
||||
# 可以尝试查看 safety_ratings 获取更详细信息
|
||||
ratings_info = getattr(candidate, 'safety_ratings', [])
|
||||
self.LOG.debug(f"Safety Ratings: {ratings_info}")
|
||||
return "抱歉，生成的响应可能包含不安全内容，已被阻止。"
|
||||
elif finish_reason == generation_types.FinishReason.RECITATION:
|
||||
self.LOG.warning(f"Gemini 响应因引用保护被阻止。")
|
||||
return "抱歉，回答可能包含受版权保护的内容，已被部分阻止。"
|
||||
elif finish_reason == generation_types.FinishReason.OTHER:
|
||||
self.LOG.warning(f"Gemini 响应因未知原因停止 (FinishReason: OTHER)")
|
||||
return "抱歉，生成响应时遇到未知问题。"
|
||||
else:
|
||||
# 包括 MAX_TOKENS, STOP 等预期情况，但没有文本
|
||||
self.LOG.warning(f"Gemini 未返回文本内容，但完成原因可接受: {finish_reason_name}")
|
||||
return f"生成内容时遇到问题 (完成原因: {finish_reason_name})"
|
||||
else:
|
||||
# 没有 candidates，也没有 prompt block，未知情况
|
||||
self.LOG.error("Gemini API 调用成功但未返回任何候选内容或提示反馈。")
|
||||
return "抱歉，Gemini未能生成响应。"
|
||||
|
||||
except (GoogleAPICallError, ClientError) as api_error:
|
||||
self.LOG.error(f"Gemini API 调用错误：{api_error}", exc_info=True)
|
||||
if "API key not valid" in str(api_error):
|
||||
return "Gemini API 密钥无效或已过期。"
|
||||
elif "quota" in str(api_error).lower():
|
||||
return "Gemini API 调用已达配额限制。"
|
||||
elif "Model not found" in str(api_error):
|
||||
return f"配置的 Gemini 模型 '{self._model_name}' 未找到或不可用。"
|
||||
elif "Resource has been exhausted" in str(api_error):
|
||||
return "Gemini API 资源耗尽，请稍后再试或检查配额。"
|
||||
else:
|
||||
return f"与 Gemini 通信时出错: {type(api_error).__name__}"
|
||||
except generation_types.StopCandidateException as sce: # 明确捕获这个
|
||||
self.LOG.error(f"Gemini API 响应被停止 (StopCandidateException): {sce}", exc_info=True)
|
||||
# 通常在流式处理中遇到，但也可能在非流式中因某些原因触发
|
||||
return "抱歉，Gemini 生成的响应被意外停止了。"
|
||||
# BlockedPromptException 似乎不直接抛出，而是通过 prompt_feedback 反馈
|
||||
# except generation_types.BlockedPromptException as bpe:
|
||||
# self.LOG.error(f"Gemini API 提示被阻止：{bpe}", exc_info=True)
|
||||
# return "抱歉，您的请求内容被 Gemini 阻止了。"
|
||||
except Exception as e:
|
||||
self.LOG.error(f"调用 Gemini 时发生未知错误: {e}", exc_info=True)
|
||||
return f"处理您的请求时发生未知错误: {type(e).__name__}"
|
||||
|
||||
return rsp_text.strip()
|
||||
|
||||
def get_answer(self, question: str, wxid: str, system_prompt_override=None, specific_max_history=None) -> str:
|
||||
if not self._model:
|
||||
return "Gemini 模型未成功初始化，请检查配置和网络。"
|
||||
|
||||
if not question:
|
||||
self.LOG.warning(f"尝试为 wxid={wxid} 获取答案，但问题为空。")
|
||||
return "您没有提问哦。"
|
||||
|
||||
# 1. 准备历史消息
|
||||
contents = []
|
||||
# 1. 处理历史消息
|
||||
if self.message_summary and self.bot_wxid:
|
||||
history = self.message_summary.get_messages(wxid)
|
||||
limit = specific_max_history if specific_max_history is not None else self.max_history_messages
|
||||
self.LOG.debug(f"获取 Gemini 历史 for {wxid}, 原始条数: {len(history)}, 使用限制: {limit}")
|
||||
|
||||
limit_to_use = specific_max_history if specific_max_history is not None else self.max_history_messages
|
||||
self.LOG.debug(f"获取 Gemini 历史记录 for {wxid}, 原始条数: {len(history)}, 使用限制: {limit_to_use}")
|
||||
|
||||
if limit_to_use is not None and limit_to_use > 0:
|
||||
history = history[-limit_to_use:]
|
||||
elif limit_to_use == 0:
|
||||
history = []
|
||||
if limit > 0:
|
||||
history = history[-limit:]
|
||||
elif limit == 0:
|
||||
history = [] # 明确清空历史
|
||||
|
||||
self.LOG.debug(f"应用限制后 Gemini 历史条数: {len(history)}")
|
||||
contents.extend(self._format_history(history))
|
||||
@@ -111,172 +233,167 @@ class Gemini:
|
||||
self.LOG.warning(f"无法为 wxid={wxid} 获取 Gemini 历史记录。")
|
||||
|
||||
# 2. 添加当前用户问题
|
||||
if question:
|
||||
contents.append({'role': 'user', 'parts': [question]})
|
||||
else:
|
||||
# 如果问题为空，可能不需要调用 API
|
||||
self.LOG.warning("尝试为 wxid={wxid} 获取答案，但问题为空。")
|
||||
# 可以返回空字符串或提示信息
|
||||
return "您没有提问哦。"
|
||||
# 注意：格式化时已包含发送者信息
|
||||
contents.append({'role': 'user', 'parts': [{'text': question}]})
|
||||
|
||||
# 3. 处理 System Prompt Override (如果提供)
|
||||
# 注意：Gemini API 目前不直接支持在 generate_content 中覆盖 system_instruction
|
||||
# 如果需要动态改变系统提示，通常需要重新初始化模型或在用户消息前插入一条 'user' role 的指令
|
||||
# 这里我们暂时忽略 system_prompt_override，因为标准 API 调用不支持
|
||||
if system_prompt_override:
|
||||
self.LOG.warning("Gemini API 当前不支持单次请求覆盖系统提示，将使用初始化时的提示。")
|
||||
# 可以考虑在这里将 override 的内容作为一条 user message 添加到 contents 开头
|
||||
# 例如: contents.insert(0, {'role': 'user', 'parts': [{'text': f"[System Instruction Override]: {system_prompt_override}"}]})
|
||||
# 但这会影响对话历史的结构，需要谨慎使用
|
||||
|
||||
# 3. 确定系统提示
|
||||
# 使用初始化时存储的 self._prompt 作为默认系统提示
|
||||
default_system_instruction = self._prompt if self._prompt else "You are a helpful assistant."
|
||||
effective_system_instruction = system_prompt_override if system_prompt_override else default_system_instruction
|
||||
# 添加当前时间到系统提示 (可选)
|
||||
# 4. 添加当前时间信息（可选，作为用户消息的一部分）
|
||||
now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
|
||||
time_mk = "\nCurrent time is: " # 加个换行
|
||||
final_system_instruction = f"{effective_system_instruction}{time_mk}{now_time}"
|
||||
# 可以将时间信息添加到最近的用户消息中，或作为一条新的 user message
|
||||
# 为了简单，暂不自动添加时间信息到内容中，如果需要，可以在 prompt 中说明
|
||||
|
||||
# 5. 调用内部生成方法
|
||||
return self._generate_response(contents)
|
||||
|
||||
rsp = ""
|
||||
try:
|
||||
# 配置生成参数 (可以从 config 中读取更多参数)
|
||||
generation_config = genai.types.GenerationConfig(
|
||||
# candidate_count=1, # 通常为 1
|
||||
# stop_sequences=["..."], # 可选的停止序列
|
||||
# max_output_tokens=2048, # 可选的最大输出 token
|
||||
temperature=0.7, # 控制随机性
|
||||
# top_p=1.0,
|
||||
# top_k=1,
|
||||
)
|
||||
|
||||
self.LOG.debug(f"发送给 Gemini API 的内容条数: {len(contents)}")
|
||||
# 注意: system_instruction 在 model 初始化时已设置，这里也可以覆盖
|
||||
# 但更推荐在初始化时设置，除非需要为单次请求定制
|
||||
response = self._model.generate_content(
|
||||
contents=contents,
|
||||
generation_config=generation_config,
|
||||
# 如果需要在每次调用时覆盖，则取消注释下一行
|
||||
# system_instruction=final_system_instruction
|
||||
)
|
||||
|
||||
# 处理可能的安全阻止等情况
|
||||
if response.candidates:
|
||||
# 检查完成原因
|
||||
finish_reason = response.candidates[0].finish_reason
|
||||
if finish_reason == genai.types.Candidate.FinishReason.SAFETY:
|
||||
rsp = "抱歉，您的请求可能包含不安全内容，已被阻止。"
|
||||
self.LOG.warning(f"Gemini 请求被安全策略阻止 (wxid: {wxid})")
|
||||
elif finish_reason == genai.types.Candidate.FinishReason.RECITATION:
|
||||
rsp = "抱歉，回答可能包含受版权保护的内容，已被部分阻止。"
|
||||
self.LOG.warning(f"Gemini 响应因引用保护被阻止 (wxid: {wxid})")
|
||||
elif response.text:
|
||||
rsp = response.text
|
||||
else: # 其他完成原因但没有文本
|
||||
rsp = f"生成内容时遇到问题 (完成原因: {finish_reason.name})"
|
||||
self.LOG.error(f"Gemini 未返回文本，完成原因: {finish_reason.name} (wxid: {wxid})")
|
||||
|
||||
elif response.prompt_feedback and response.prompt_feedback.block_reason:
|
||||
# 如果整个提示被阻止
|
||||
block_reason = response.prompt_feedback.block_reason.name
|
||||
rsp = f"抱歉，您的请求因包含不适内容而被阻止 (原因: {block_reason})。"
|
||||
self.LOG.warning(f"Gemini 提示被阻止，原因: {block_reason} (wxid: {wxid})")
|
||||
else:
|
||||
# 未知情况，没有候选也没有提示反馈
|
||||
rsp = "抱歉，Gemini未能生成响应。"
|
||||
self.LOG.error(f"Gemini 调用成功但未返回有效响应或错误信息 (wxid: {wxid})")
|
||||
|
||||
|
||||
except BlockedPromptException as bpe:
|
||||
self.LOG.error(f"Gemini API 提示被阻止：{bpe}", exc_info=True)
|
||||
rsp = "抱歉，您的请求内容被 Gemini 阻止了。"
|
||||
except StopCandidateException as sce:
|
||||
self.LOG.error(f"Gemini API 响应被停止：{sce}", exc_info=True)
|
||||
rsp = "抱歉，Gemini 生成的响应包含不适内容被停止了。"
|
||||
except (GoogleAPICallError, ClientError) as api_error:
|
||||
self.LOG.error(f"Gemini API 调用错误：{api_error}", exc_info=True)
|
||||
# 尝试提供更具体的错误信息
|
||||
if "API key not valid" in str(api_error):
|
||||
rsp = "Gemini API 密钥无效或已过期。"
|
||||
elif "quota" in str(api_error).lower():
|
||||
rsp = "Gemini API 调用已达配额限制。"
|
||||
else:
|
||||
rsp = f"与 Gemini 通信时出错: {type(api_error).__name__}"
|
||||
except Exception as e:
|
||||
self.LOG.error(f"调用 Gemini 时发生未知错误: {e}", exc_info=True)
|
||||
rsp = f"处理您的请求时发生未知错误: {type(e).__name__}"
|
||||
|
||||
return rsp.strip() # 移除可能的首尾空白
|
||||
|
||||
# 可以添加处理图片的方法
|
||||
def get_image_description(self, image_path: str, prompt: str = "请详细描述这张图片中的内容") -> str:
|
||||
if not self._model or not self.support_vision:
|
||||
return "当前 Gemini 模型未初始化或不支持图片理解。"
|
||||
if not self._model:
|
||||
return "Gemini 模型未初始化。"
|
||||
if not self.support_vision:
|
||||
return f"当前 Gemini 模型 '{self._model_name}' 不支持图片理解。"
|
||||
|
||||
if not os.path.exists(image_path):
|
||||
self.LOG.error(f"图片文件不存在: {image_path}")
|
||||
image_path_obj = pathlib.Path(image_path)
|
||||
if not image_path_obj.is_file():
|
||||
self.LOG.error(f"图片文件不存在或不是文件: {image_path}")
|
||||
return "无法读取图片文件"
|
||||
|
||||
try:
|
||||
self.LOG.info(f"使用 Gemini 分析图片: {image_path}")
|
||||
image_part = genai.types.Part.from_uri(
|
||||
mime_type="image/jpeg", # 假设是jpeg，可以根据文件扩展名判断
|
||||
uri=f"file://{os.path.abspath(image_path)}" # 使用文件URI
|
||||
)
|
||||
# 猜测 MIME 类型
|
||||
mime_type, _ = mimetypes.guess_type(image_path_obj)
|
||||
if not mime_type or not mime_type.startswith("image/"):
|
||||
self.LOG.warning(f"无法确定图片 MIME 类型或类型不是 image/*: {image_path}, 猜测为 jpeg")
|
||||
mime_type = "image/jpeg" # 使用默认值
|
||||
|
||||
# 构建包含文本和图片的消息
|
||||
self.LOG.info(f"使用 Gemini 分析图片: {image_path} (MIME: {mime_type})")
|
||||
|
||||
# 使用 pathlib 生成 file URI
|
||||
image_uri = image_path_obj.absolute().as_uri()
|
||||
|
||||
image_part = {'mime_type': mime_type, 'data': image_path_obj.read_bytes()}
|
||||
|
||||
# 构建包含文本提示和图片的消息
|
||||
contents = [
|
||||
{'role': 'user', 'parts': [prompt, image_part]}
|
||||
# Gemini 处理多模态输入时，推荐 prompt 和 image 都在 user role 的 parts 里
|
||||
{'role': 'user', 'parts': [
|
||||
{'text': prompt},
|
||||
image_part
|
||||
# 或者使用 from_uri (如果 API 支持且网络可访问该文件 URI):
|
||||
# genai.types.Part.from_uri(mime_type=mime_type, uri=image_uri)
|
||||
# 使用原始字节通常更可靠
|
||||
]}
|
||||
]
|
||||
|
||||
# 可以使用与 get_answer 类似的生成配置
|
||||
generation_config = genai.types.GenerationConfig(temperature=0.4)
|
||||
# 可以为图片分析设置不同的生成参数，例如更低的温度以获得更客观的描述
|
||||
image_gen_config = generation_types.GenerationConfig(temperature=0.4)
|
||||
|
||||
response = self._model.generate_content(
|
||||
contents=contents,
|
||||
generation_config=generation_config
|
||||
)
|
||||
|
||||
# 处理响应 (与 get_answer 类似)
|
||||
# ... [省略类似的响应处理逻辑] ...
|
||||
if response.text:
|
||||
return response.text.strip()
|
||||
else:
|
||||
# ... [省略类似的错误/阻止处理逻辑] ...
|
||||
return "Gemini 未能描述图片。"
|
||||
# 调用内部生成方法
|
||||
return self._generate_response(contents, generation_config_override=image_gen_config)
|
||||
|
||||
except FileNotFoundError:
|
||||
self.LOG.error(f"读取图片文件时发生 FileNotFoundError: {image_path}", exc_info=True)
|
||||
return "读取图片文件时出错。"
|
||||
except Exception as e:
|
||||
self.LOG.error(f"使用 Gemini 分析图片时出错: {e}", exc_info=True)
|
||||
return f"分析图片时出错: {type(e).__name__}"
|
||||
self.LOG.error(f"使用 Gemini 分析图片时发生未知错误: {e}", exc_info=True)
|
||||
return f"分析图片时发生未知错误: {type(e).__name__}"
|
||||
|
||||
# --- Main 测试部分 ---
|
||||
if __name__ == "__main__":
|
||||
# --- 简单的本地测试 ---
|
||||
print("运行 Gemini 本地测试...")
|
||||
logging.basicConfig(level=logging.DEBUG) # 设置日志级别
|
||||
print("--- 运行 Gemini 本地测试 ---")
|
||||
# 配置日志记录
|
||||
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
|
||||
# 模拟配置 (需要替换为你的真实 API Key)
|
||||
# 请确保从环境变量或安全途径获取 API Key，不要硬编码在代码中
|
||||
mock_config = {
|
||||
"api_key": os.environ.get("GEMINI_API_KEY", "YOUR_API_KEY"), # 从环境变量读取，或替换为你的key
|
||||
"model_name": "gemini-1.5-flash-latest", # 使用一个快速的模型测试
|
||||
"prompt": "你是一个乐于助人的AI助手。",
|
||||
"proxy": os.environ.get("HTTP_PROXY"), # 从环境变量读取代理
|
||||
"max_history_messages": 5
|
||||
}
|
||||
# --- 配置加载 ---
|
||||
# !!! 强烈建议从环境变量或安全的配置文件加载 API Key !!!
|
||||
# 例如: api_key = os.environ.get("GEMINI_API_KEY")
|
||||
# 不要将 API Key 硬编码在代码中提交
|
||||
api_key_from_env = os.environ.get("GEMINI_API_KEY")
|
||||
proxy_from_env = os.environ.get("HTTP_PROXY") # 支持 http/https 代理
|
||||
|
||||
if mock_config["api_key"] == "YOUR_API_KEY":
|
||||
print("警告：请设置 GEMINI_API_KEY 环境变量或替换 mock_config 中的 API Key！")
|
||||
exit()
|
||||
|
||||
# 不依赖 MessageSummary 和 bot_wxid 进行简单测试
|
||||
gemini_assistant = Gemini(mock_config)
|
||||
|
||||
if gemini_assistant._model: # 检查是否初始化成功
|
||||
test_question = "你好，Gemini！给我讲个关于太空旅行的笑话吧。"
|
||||
print(f"\n提问: {test_question}")
|
||||
answer = gemini_assistant.get_answer(test_question, "test_wxid")
|
||||
print(f"\nGemini 回答:\n{answer}")
|
||||
|
||||
# 测试图片描述 (可选, 需要有图片)
|
||||
# image_test_path = "path/to/your/test_image.jpg"
|
||||
# if os.path.exists(image_test_path):
|
||||
# print("\n测试图片描述...")
|
||||
# desc_prompt = "这张图片里有什么？"
|
||||
# description = gemini_assistant.get_image_description(image_test_path, desc_prompt)
|
||||
# print(f"\n图片描述:\n{description}")
|
||||
# else:
|
||||
# print(f"\n跳过图片测试，图片文件未找到: {image_test_path}")
|
||||
if not api_key_from_env:
|
||||
print("\n!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
|
||||
print("!!! 警告：环境变量 GEMINI_API_KEY 未设置。请设置该变量。 !!!")
|
||||
print("!!! 测试将无法连接到 Gemini API。 !!!")
|
||||
print("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!\n")
|
||||
# 可以选择退出或继续（如果只想测试初始化逻辑）
|
||||
# exit(1)
|
||||
api_key_to_use = "DUMMY_KEY_FOR_INIT_TEST" # 仅用于测试初始化日志，无法实际调用
|
||||
else:
|
||||
print("Gemini 初始化失败，无法进行测试。")
|
||||
api_key_to_use = api_key_from_env
|
||||
|
||||
mock_config = {
|
||||
"api_key": api_key_to_use,
|
||||
"model_name": "gemini-1.5-flash-latest", # 使用较快的模型测试
|
||||
"prompt": "你是一个风趣幽默的AI助手，擅长讲冷笑话。",
|
||||
"proxy": proxy_from_env,
|
||||
"max_history_messages": 3 # 测试时减少历史记录
|
||||
}
|
||||
print(f"测试配置: Model={mock_config['model_name']}, Proxy={'已设置' if mock_config['proxy'] else '未设置'}")
|
||||
|
||||
# --- 初始化 Gemini ---
|
||||
# 在测试中不依赖 MessageSummary
|
||||
print("\n--- 初始化 Gemini 实例 ---")
|
||||
gemini_assistant = Gemini(mock_config, bot_wxid="test_bot_wxid") # 提供一个测试 bot_wxid
|
||||
|
||||
# --- 测试文本生成 ---
|
||||
if gemini_assistant._model: # 检查模型是否成功初始化
|
||||
print("\n--- 测试文本生成 (get_answer) ---")
|
||||
test_question = "你好！今天天气怎么样？给我讲个关于程序员的冷笑话吧。"
|
||||
print(f"提问: {test_question}")
|
||||
start_time = time.time()
|
||||
answer = gemini_assistant.get_answer(test_question, "test_user_wxid") # 提供测试 wxid
|
||||
end_time = time.time()
|
||||
print(f"\nGemini 回答 (耗时: {end_time - start_time:.2f}s):\n{answer}")
|
||||
|
||||
# 测试空问题
|
||||
print("\n--- 测试空问题 ---")
|
||||
empty_answer = gemini_assistant.get_answer("", "test_user_wxid")
|
||||
print(f"空问题回答: {empty_answer}")
|
||||
|
||||
# 测试长对话历史（如果需要，可以手动构建一个 contents 列表来模拟）
|
||||
# print("\n--- 模拟长对话测试 ---")
|
||||
# mock_history = [
|
||||
# {'role': 'user', 'parts': [{'text': "[UserA]: 第一次提问"}]},
|
||||
# {'role': 'model', 'parts': [{'text': "第一次回答"}]},
|
||||
# {'role': 'user', 'parts': [{'text': "[UserB]: 第二次提问"}]},
|
||||
# {'role': 'model', 'parts': [{'text': "第二次回答"}]},
|
||||
# {'role': 'user', 'parts': [{'text': "[UserA]: 第三次提问，关于第一次提问的内容"}]},
|
||||
# ]
|
||||
# mock_history.append({'role': 'user', 'parts': [{'text': "当前的第四个问题"}]})
|
||||
# long_hist_answer = gemini_assistant._generate_response(mock_history)
|
||||
# print(f"长历史回答:\n{long_hist_answer}")
|
||||
|
||||
else:
|
||||
print("\n--- Gemini 初始化失败，跳过文本生成测试 ---")
|
||||
|
||||
# --- 测试图片描述 (可选) ---
|
||||
if gemini_assistant._model and gemini_assistant.support_vision:
|
||||
print("\n--- 测试图片描述 (get_image_description) ---")
|
||||
# 将 'path/to/your/test_image.jpg' 替换为实际的图片路径
|
||||
image_test_path_str = "test_image.jpg" # 假设图片在脚本同目录下
|
||||
image_test_path = pathlib.Path(image_test_path_str)
|
||||
|
||||
if image_test_path.exists():
|
||||
desc_prompt = "详细描述这张图片里的所有元素和场景氛围。"
|
||||
print(f"图片路径: {image_test_path.absolute()}")
|
||||
print(f"描述提示: {desc_prompt}")
|
||||
start_time = time.time()
|
||||
description = gemini_assistant.get_image_description(str(image_test_path), desc_prompt)
|
||||
end_time = time.time()
|
||||
print(f"\n图片描述 (耗时: {end_time - start_time:.2f}s):\n{description}")
|
||||
else:
|
||||
print(f"\n跳过图片测试，测试图片文件未找到: {image_test_path.absolute()}")
|
||||
print("请将一张名为 test_image.jpg 的图片放在脚本相同目录下进行测试。")
|
||||
elif gemini_assistant._model:
|
||||
print(f"\n--- 跳过图片测试，当前模型 {gemini_assistant._model_name} 不支持视觉 ---")
|
||||
else:
|
||||
print("\n--- Gemini 初始化失败，跳过图片描述测试 ---")
|
||||
|
||||
print("\n--- Gemini 本地测试结束 ---")
|
||||
Reference in New Issue
Block a user