diff --git a/README.MD b/README.MD index 60385fc8deab489f0d8339733aaaf450e7daedc4..3da98b651e78e57dabcec3b935bf030a6794af61 100644 --- a/README.MD +++ b/README.MD @@ -31,6 +31,8 @@ git clone https://gitee.com/lch0821/WeChatRobot.git python -m pip install -U pip # 安装必要依赖 pip install -r requirements.txt +# 国内用户可能会因为网络问题出现安装失败，届时可使用镜像源来下载 +pip install -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com -r requirements.txt # ChatGLM 还需要安装一个 kernel ipython kernel install --name chatglm3 --user ``` @@ -140,7 +142,25 @@ bard: # -----bard配置这行不填----- # 提示词尽可能用英文，bard对中文提示词的效果不是很理想，下方提示词为英语老师的示例，请按实际需要修改,默认设置的提示词为谷歌创造的AI大语言模型 # I want you to act as a spoken English teacher and improver. I will speak to you in English and you will reply to me in English to practice my spoken English. I want you to keep your reply neat, limiting the reply to 100 words. I want you to strictly correct my grammar mistakes, typos, and factual errors. I want you to ask me a question in your reply. Now let's start practicing, you could ask me a question first. Remember, I want you to strictly correct my grammar mistakes, typos, and factual errors. prompt: You am a large language model, trained by Google. + +deepseek: sk-xxxxxxxxxxxxxxx # -----deepseek配置这行不填----- + #思维链相关功能默认关闭，开启后会增加响应时间和消耗更多的token + key: # 填写你的 DeepSeek API Key + api: https://api.deepseek.com # DeepSeek API 地址 + model: deepseek-chat # 可选: deepseek-chat (DeepSeek-V3), deepseek-reasoner (DeepSeek-R1) + prompt: 你是智能聊天机器人，你叫 DeepSeek 助手 # 根据需要对角色进行设定 + enable_reasoning: false # 是否启用思维链功能，仅在使用 deepseek-reasoner 模型时有效 + show_reasoning: false # 是否在回复中显示思维过程，仅在启用思维链功能时有效 +``` + +## 至开发者 +``` +在接入图片生成的相关功能时，可将调用文件放入image文件夹内。 +在 image/__init__.py 文件内加入对应的模块儿，以便作为Python包来调用。 +在 configuration.py 内也要加入相关的代码，否则会初始化失败。 +文生图相关功能，全部默认开启，如果均未配置，全部转接至语言大模型。 ``` +[文生图功能的使用说明](./image/文生图功能的使用说明.MD) ## HTTP 如需要使用 HTTP 接口，请参考： diff --git a/base/func_deepseek.py b/base/func_deepseek.py new file mode 100644 index 0000000000000000000000000000000000000000..d0b1d174d9bcf2352302b64669383a866af80878 --- /dev/null +++ b/base/func_deepseek.py @@ -0,0 +1,150 @@ +#! /usr/bin/env python3 +# -*- coding: utf-8 -*- + +import logging +from datetime import datetime + +import httpx +from openai import APIConnectionError, APIError, AuthenticationError, OpenAI + + +class DeepSeek(): + def __init__(self, conf: dict) -> None: + key = conf.get("key") + api = conf.get("api", "https://api.deepseek.com") + proxy = conf.get("proxy") + prompt = conf.get("prompt") + self.model = conf.get("model", "deepseek-chat") + self.LOG = logging.getLogger("DeepSeek") + + self.reasoning_supported = (self.model == "deepseek-reasoner") + + if conf.get("enable_reasoning", False) and not self.reasoning_supported: + self.LOG.warning("思维链功能只在使用 deepseek-reasoner 模型时可用，当前模型不支持此功能") + + self.enable_reasoning = conf.get("enable_reasoning", False) and self.reasoning_supported + self.show_reasoning = conf.get("show_reasoning", False) and self.enable_reasoning + + if proxy: + self.client = OpenAI(api_key=key, base_url=api, http_client=httpx.Client(proxy=proxy)) + else: + self.client = OpenAI(api_key=key, base_url=api) + + self.conversation_list = {} + + self.system_content_msg = {"role": "system", "content": prompt} + + def __repr__(self): + return 'DeepSeek' + + @staticmethod + def value_check(conf: dict) -> bool: + if conf: + if conf.get("key") and conf.get("prompt"): + return True + return False + + def get_answer(self, question: str, wxid: str) -> str: + if question == "#清除对话": + if wxid in self.conversation_list.keys(): + del self.conversation_list[wxid] + return "已清除上下文" + + if question.lower() in ["#开启思维链", "#enable reasoning"]: + if not self.reasoning_supported: + return "当前模型不支持思维链功能，请使用 deepseek-reasoner 模型" + self.enable_reasoning = True + self.show_reasoning = True + return "已开启思维链模式，将显示完整的推理过程" + + if question.lower() in ["#关闭思维链", "#disable reasoning"]: + if not self.reasoning_supported: + return "当前模型不支持思维链功能，无需关闭" + self.enable_reasoning = False + self.show_reasoning = False + return "已关闭思维链模式" + + if question.lower() in ["#隐藏思维链", "#hide reasoning"]: + if not self.enable_reasoning: + return "思维链功能未开启，无法设置隐藏/显示" + self.show_reasoning = False + return "已设置隐藏思维链，但模型仍会进行深度思考" + + if question.lower() in ["#显示思维链", "#show reasoning"]: + if not self.enable_reasoning: + return "思维链功能未开启，无法设置隐藏/显示" + self.show_reasoning = True + return "已设置显示思维链" + + if wxid not in self.conversation_list: + self.conversation_list[wxid] = [] + if self.system_content_msg["content"]: + self.conversation_list[wxid].append(self.system_content_msg) + + self.conversation_list[wxid].append({"role": "user", "content": question}) + + try: + clean_messages = [] + for msg in self.conversation_list[wxid]: + clean_msg = {"role": msg["role"], "content": msg["content"]} + clean_messages.append(clean_msg) + + response = self.client.chat.completions.create( + model=self.model, + messages=clean_messages, + stream=False + ) + + if self.reasoning_supported and self.enable_reasoning: + # deepseek-reasoner模型返回的特殊字段: reasoning_content和content + # 单独处理思维链模式的响应 + reasoning_content = getattr(response.choices[0].message, "reasoning_content", None) + content = response.choices[0].message.content + + if self.show_reasoning and reasoning_content: + final_response = f"🤔思考过程：\n{reasoning_content}\n\n🎉最终答案：\n{content}" + #最好不要删除表情，因为微信内的信息没有办法做自定义显示，这里是为了做两个分隔，来区分思考过程和最终答案！💡 + else: + final_response = content + self.conversation_list[wxid].append({"role": "assistant", "content": content}) + else: + final_response = response.choices[0].message.content + self.conversation_list[wxid].append({"role": "assistant", "content": final_response}) + + # 控制对话长度，保留最近的历史记录 + # 系统消息(如果有) + 最近9轮对话(问答各算一轮) + max_history = 19 + if len(self.conversation_list[wxid]) > max_history: + has_system = self.conversation_list[wxid][0]["role"] == "system" + if has_system: + self.conversation_list[wxid] = [self.conversation_list[wxid][0]] + self.conversation_list[wxid][-(max_history-1):] + else: + self.conversation_list[wxid] = self.conversation_list[wxid][-max_history:] + + return final_response + + except (APIConnectionError, APIError, AuthenticationError) as e1: + self.LOG.error(f"DeepSeek API 返回了错误：{str(e1)}") + return f"DeepSeek API 返回了错误：{str(e1)}" + except Exception as e0: + self.LOG.error(f"发生未知错误：{str(e0)}") + return "抱歉，处理您的请求时出现了错误" + + +if __name__ == "__main__": + from configuration import Config + config = Config().DEEPSEEK + if not config: + exit(0) + + chat = DeepSeek(config) + + while True: + q = input(">>> ") + try: + time_start = datetime.now() + print(chat.get_answer(q, "wxid")) + time_end = datetime.now() + print(f"{round((time_end - time_start).total_seconds(), 2)}s") + except Exception as e: + print(e) diff --git a/config.yaml.template b/config.yaml.template index c26c7f7cbf6ca94bb2f8bf08fd27b6fb1b58c1c0..f2bb541d45c314b7b7cc7e196029dd9d887d8013 100644 --- a/config.yaml.template +++ b/config.yaml.template @@ -104,3 +104,42 @@ bard: # -----bard配置这行不填----- zhipu: # -----zhipu配置这行不填----- api_key: #api key model: # 模型类型 + +deepseek: # -----deepseek配置这行不填----- + #思维链相关功能默认关闭，开启后会增加响应时间和消耗更多的token + key: # 填写你的 DeepSeek API Key API Key的格式为sk-xxxxxxxxxxxxxxx + api: https://api.deepseek.com # DeepSeek API 地址 + model: deepseek-chat # 可选: deepseek-chat (DeepSeek-V3), deepseek-reasoner (DeepSeek-R1) + prompt: 你是智能聊天机器人，你叫 DeepSeek 助手 # 根据需要对角色进行设定 + enable_reasoning: false # 是否启用思维链功能，仅在使用 deepseek-reasoner 模型时有效 + show_reasoning: false # 是否在回复中显示思维过程，仅在启用思维链功能时有效 + +cogview: # -----智谱AI图像生成配置这行不填----- + # 此API请参考 https://www.bigmodel.cn/dev/api/image-model/cogview + enable: False # 是否启用图像生成功能，默认关闭，将False替换为true则开启，此模型可和其他模型同时运行。 + api_key: # 智谱API密钥，请填入您的API Key + model: cogview-4-250304 # 模型编码，可选：cogview-4-250304、cogview-4、cogview-3-flash + quality: standard # 生成质量，可选：standard（快速）、hd（高清） + size: 1024x1024 # 图片尺寸，可自定义，需符合条件 + trigger_keyword: 牛智谱 # 触发图像生成的关键词 + temp_dir: # 临时文件存储目录，留空则默认使用项目目录下的zhipuimg文件夹，如果要更改，例如 D:/Pictures/temp 或 /home/user/temp + fallback_to_chat: true # 当未启用绘画功能时：true=将请求发给聊天模型处理，false=回复固定的未启用提示信息 + +aliyun_image: # -----如果要使用阿里云文生图，取消下面的注释并填写相关内容，模型到阿里云百炼找通义万相-文生图2.1-Turbo----- + enable: true # 是否启用阿里文生图功能，false为关闭，默认开启，如果未配置，则会将消息发送给聊天大模型 + api_key: sk-xxxxxxxxxxxxxxxxxxxxxxxx # 替换为你的DashScope API密钥 + model: wanx2.1-t2i-turbo # 模型名称，默认使用wanx2.1-t2i-turbo(快),wanx2.1-t2i-plus（中）,wanx-v1（慢），会给用户不同的提示！ + size: 1024*1024 # 图像尺寸，格式为宽*高 + n: 1 # 生成图像的数量 + temp_dir: ./temp # 临时文件存储路径 + trigger_keyword: 牛阿里 # 触发词，默认为"牛阿里" + fallback_to_chat: true # 当服务不可用时是否转发给聊天模型处理 + +gemini_image: # -----谷歌AI画图配置这行不填----- + enable: true # 是否启用谷歌AI画图功能 + api_key: # 谷歌Gemini API密钥，必填 + model: gemini-2.0-flash-exp-image-generation # 模型名称，建议保持默认，只有这一个模型可以进行绘画 + temp_dir: ./geminiimg # 图片保存目录，可选 + trigger_keyword: 牛谷歌 # 触发词，默认为"牛谷歌" + fallback_to_chat: false # 未启用时是否回退到聊天模式 + proxy: http://127.0.0.1:7890 # 使用Clash代理，格式为：http://域名或者IP地址:端口号 diff --git a/configuration.py b/configuration.py index 0f8816da2bca6d8d4d9ff2bf06eadeb021f0ad69..10bf629800609ef26ac1996500c3658e790a48a8 100644 --- a/configuration.py +++ b/configuration.py @@ -4,6 +4,7 @@ import logging.config import os import shutil +from typing import Dict, List import yaml @@ -40,5 +41,8 @@ class Config(object): self.CHATGLM = yconfig.get("chatglm", {}) self.BardAssistant = yconfig.get("bard", {}) self.ZhiPu = yconfig.get("zhipu", {}) - + self.DEEPSEEK = yconfig.get("deepseek", {}) + self.COGVIEW = yconfig.get("cogview", {}) + self.ALIYUN_IMAGE = yconfig.get("aliyun_image", {}) + self.GEMINI_IMAGE = yconfig.get("gemini_image", {}) self.SEND_RATE_LIMIT = yconfig.get("send_rate_limit", 0) diff --git a/constants.py b/constants.py index 17e9590b0844d64200fabd9da0a3b529255260a5..59fc9d06989eee5157a29f6549fb5479171e6f8b 100644 --- a/constants.py +++ b/constants.py @@ -11,13 +11,14 @@ class ChatType(IntEnum): BardAssistant = 5 # Google Bard ZhiPu = 6 # ZhiPu OLLAMA = 7 # Ollama + DEEPSEEK = 8 # DeepSeek @staticmethod def is_in_chat_types(chat_type: int) -> bool: if chat_type in [ChatType.TIGER_BOT.value, ChatType.CHATGPT.value, ChatType.XINGHUO_WEB.value, ChatType.CHATGLM.value, ChatType.BardAssistant.value, ChatType.ZhiPu.value, - ChatType.OLLAMA]: + ChatType.OLLAMA.value, ChatType.DEEPSEEK.value]: return True return False diff --git a/image/__init__.py b/image/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..6849a9c33ebe520220ac8c140ab30a8b627cde07 --- /dev/null +++ b/image/__init__.py @@ -0,0 +1,13 @@ +"""图像生成功能模块 + +包含以下功能： +- CogView: 智谱AI文生图 +- AliyunImage: 阿里云文生图 +- GeminiImage: 谷歌Gemini文生图 +""" + +from .func_cogview import CogView +from .func_aliyun_image import AliyunImage +from .func_gemini_image import GeminiImage + +__all__ = ['CogView', 'AliyunImage', 'GeminiImage'] \ No newline at end of file diff --git a/image/func_aliyun_image.py b/image/func_aliyun_image.py new file mode 100644 index 0000000000000000000000000000000000000000..a93d8761bc8c3d451c82cea5de936bb5b1a8f631 --- /dev/null +++ b/image/func_aliyun_image.py @@ -0,0 +1,108 @@ +#! /usr/bin/env python3 +# -*- coding: utf-8 -*- + +import logging +import os +import time +from http import HTTPStatus +from urllib.parse import urlparse, unquote +from pathlib import PurePosixPath + +import requests +from dashscope import ImageSynthesis +import dashscope + +class AliyunImage(): + """阿里文生图API调用 + """ + + @staticmethod + def value_check(args: dict) -> bool: + try: + return bool(args and args.get("api_key", "") and args.get("model", "")) + except Exception: + return False + + def __init__(self, config={}) -> None: + self.LOG = logging.getLogger("AliyunImage") + if not config: + raise Exception("缺少配置信息") + + self.api_key = config.get("api_key", "") + self.model = config.get("model", "wanx2.1-t2i-turbo") + self.size = config.get("size", "1024*1024") + self.enable = config.get("enable", True) + self.n = config.get("n", 1) + self.temp_dir = config.get("temp_dir", "./temp") + + # 确保临时目录存在 + if not os.path.exists(self.temp_dir): + os.makedirs(self.temp_dir) + + # 设置API密钥 + dashscope.api_key = self.api_key + + # 不要记录初始化日志 + + def generate_image(self, prompt: str) -> str: + """生成图像并返回图像URL + + Args: + prompt (str): 图像描述 + + Returns: + str: 生成的图像URL或错误信息 + """ + if not self.enable or not self.api_key: + return "阿里文生图功能未启用或API密钥未配置" + + try: + rsp = ImageSynthesis.call( + api_key=self.api_key, + model=self.model, + prompt=prompt, + n=self.n, + size=self.size + ) + + if rsp.status_code == HTTPStatus.OK and rsp.output and rsp.output.results: + return rsp.output.results[0].url + else: + self.LOG.error(f"图像生成失败: {rsp.code}, {rsp.message}") + return f"图像生成失败: {rsp.message}" + except Exception as e: + error_str = str(e) + self.LOG.error(f"图像生成出错: {error_str}") + + if "Error code: 500" in error_str or "HTTP/1.1 500" in error_str: + self.LOG.warning(f"检测到违规内容请求: {prompt}") + return "很抱歉，您的请求可能包含违规内容，无法生成图像" + + return "图像生成失败，请调整您的描述后重试" + + def download_image(self, image_url: str) -> str: + """ + 下载图片并返回本地文件路径 + + Args: + image_url (str): 图片URL + + Returns: + str: 本地图片文件路径，下载失败则返回None + """ + try: + response = requests.get(image_url, stream=True, timeout=30) + if response.status_code == 200: + file_path = os.path.join(self.temp_dir, f"aliyun_image_{int(time.time())}.jpg") + with open(file_path, 'wb') as f: + for chunk in response.iter_content(chunk_size=1024): + if chunk: + f.write(chunk) + self.LOG.info(f"图片已下载到: {file_path}") + return file_path + else: + self.LOG.error(f"下载图片失败，状态码: {response.status_code}") + return None + except Exception as e: + self.LOG.error(f"下载图片过程出错: {str(e)}") + return None diff --git a/image/func_cogview.py b/image/func_cogview.py new file mode 100644 index 0000000000000000000000000000000000000000..a7a72b720a9e030879e61a73d0b7d54a04917ae2 --- /dev/null +++ b/image/func_cogview.py @@ -0,0 +1,99 @@ +import logging +import os +import requests +import tempfile +import time +from zhipuai import ZhipuAI + +class CogView(): + def __init__(self, conf: dict) -> None: + self.api_key = conf.get("api_key") + self.model = conf.get("model", "cogview-4-250304") # 默认使用最新模型 + self.quality = conf.get("quality", "standard") + self.size = conf.get("size", "1024x1024") + self.enable = conf.get("enable", True) + + project_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + default_img_dir = os.path.join(project_dir, "zhipuimg") + self.temp_dir = conf.get("temp_dir", default_img_dir) + + self.LOG = logging.getLogger("CogView") + + if self.api_key: + self.client = ZhipuAI(api_key=self.api_key) + else: + self.LOG.warning("未配置智谱API密钥，图像生成功能无法使用") + self.client = None + + os.makedirs(self.temp_dir, exist_ok=True) + + @staticmethod + def value_check(conf: dict) -> bool: + if conf and conf.get("api_key") and conf.get("enable", True): + return True + return False + + def __repr__(self): + return 'CogView' + + def generate_image(self, prompt: str) -> str: + """ + 生成图像并返回图像URL + + Args: + prompt (str): 图像描述 + + Returns: + str: 生成的图像URL或错误信息 + """ + if not self.client or not self.enable: + return "图像生成功能未启用或API密钥未配置" + + try: + response = self.client.images.generations( + model=self.model, + prompt=prompt, + quality=self.quality, + size=self.size, + ) + + if response and response.data and len(response.data) > 0: + return response.data[0].url + else: + return "图像生成失败，未收到有效响应" + except Exception as e: + error_str = str(e) + self.LOG.error(f"图像生成出错: {error_str}") + + if "Error code: 500" in error_str or "HTTP/1.1 500" in error_str or "code\":\"1234\"" in error_str: + self.LOG.warning(f"检测到违规内容请求: {prompt}") + return "很抱歉，您的请求可能包含违规内容，无法生成图像" + + return "图像生成失败，请调整您的描述后重试" + + def download_image(self, image_url: str) -> str: + """ + 下载图片并返回本地文件路径 + + Args: + image_url (str): 图片URL + + Returns: + str: 本地图片文件路径，下载失败则返回None + """ + try: + response = requests.get(image_url, stream=True, timeout=30) + if response.status_code == 200: + file_path = os.path.join(self.temp_dir, f"cogview_{int(time.time())}.jpg") + with open(file_path, 'wb') as f: + for chunk in response.iter_content(chunk_size=1024): + if chunk: + f.write(chunk) + self.LOG.info(f"图片已下载到: {file_path}") + return file_path + else: + self.LOG.error(f"下载图片失败，状态码: {response.status_code}") + return None + except Exception as e: + self.LOG.error(f"下载图片过程出错: {str(e)}") + return None diff --git a/image/func_gemini_image.py b/image/func_gemini_image.py new file mode 100644 index 0000000000000000000000000000000000000000..708a2bdc52573a02f4fe15e0912e3275952a68d7 --- /dev/null +++ b/image/func_gemini_image.py @@ -0,0 +1,113 @@ +#! /usr/bin/env python3 +# -*- coding: utf-8 -*- + +import logging +import os +import mimetypes +import time +import random +from google import genai +from google.genai import types + +class GeminiImage: + """谷歌AI画图API调用 + """ + + def __init__(self, config={}) -> None: + self.LOG = logging.getLogger("GeminiImage") + + self.enable = config.get("enable", True) + self.api_key = config.get("api_key", "") or os.environ.get("GEMINI_API_KEY", "") + self.model = config.get("model", "gemini-2.0-flash-exp-image-generation") + self.proxy = config.get("proxy", "") + + project_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + self.temp_dir = config.get("temp_dir", os.path.join(project_dir, "geminiimg")) + + if not os.path.exists(self.temp_dir): + os.makedirs(self.temp_dir) + + if not self.api_key: + self.enable = False + return + + try: + # 设置代理 + if self.proxy: + os.environ["HTTP_PROXY"] = self.proxy + os.environ["HTTPS_PROXY"] = self.proxy + + # 初始化客户端 + self.client = genai.Client(api_key=self.api_key) + except Exception: + self.enable = False + + def generate_image(self, prompt: str) -> str: + """生成图像并返回图像文件路径 + """ + try: + # 设置代理 + if self.proxy: + os.environ["HTTP_PROXY"] = self.proxy + os.environ["HTTPS_PROXY"] = self.proxy + + image_prompt = f"生成一张高质量的图片: {prompt}。请直接提供图像，不需要描述。" + + # 发送请求 + response = self.client.models.generate_content( + model=self.model, + contents=image_prompt, + config=types.GenerateContentConfig( + response_modalities=['Text', 'Image'] + ) + ) + + # 处理响应 + if hasattr(response, 'candidates') and response.candidates: + for candidate in response.candidates: + if hasattr(candidate, 'content') and candidate.content: + for part in candidate.content.parts: + if hasattr(part, 'inline_data') and part.inline_data: + # 保存图像 + file_name = f"gemini_image_{int(time.time())}_{random.randint(1000, 9999)}" + file_extension = mimetypes.guess_extension(part.inline_data.mime_type) or ".png" + file_path = os.path.join(self.temp_dir, f"{file_name}{file_extension}") + + with open(file_path, "wb") as f: + f.write(part.inline_data.data) + + return file_path + + # 如果没有找到图像，尝试获取文本响应 + try: + text_content = response.text + if text_content: + return f"模型未能生成图像: {text_content[:100]}..." + except (AttributeError, TypeError): + pass + + return "图像生成失败，可能需要更新模型或调整提示词" + + except Exception as e: + error_str = str(e) + self.LOG.error(f"图像生成出错: {error_str}") + + # 处理500错误 + if "500 INTERNAL" in error_str: + self.LOG.error("遇到谷歌服务器内部错误") + return "谷歌AI服务器临时故障，请稍后再试。这是谷歌服务器的问题，不是你的请求有误。" + + if "timeout" in error_str.lower(): + return "图像生成超时，请检查网络或代理设置" + + if "violated" in error_str.lower() or "policy" in error_str.lower(): + return "请求包含违规内容，无法生成图像" + + # 其他常见错误类型处理 + if "quota" in error_str.lower() or "rate" in error_str.lower(): + return "API使用配额已用尽或请求频率过高，请稍后再试" + + if "authentication" in error_str.lower() or "auth" in error_str.lower(): + return "API密钥验证失败，请联系管理员检查配置" + + return f"图像生成失败，错误原因: {error_str.split('.')[-1] if '.' in error_str else error_str}" \ No newline at end of file diff --git "a/image/\346\226\207\347\224\237\345\233\276\345\212\237\350\203\275\347\232\204\344\275\277\347\224\250\350\257\264\346\230\216.MD" "b/image/\346\226\207\347\224\237\345\233\276\345\212\237\350\203\275\347\232\204\344\275\277\347\224\250\350\257\264\346\230\216.MD" new file mode 100644 index 0000000000000000000000000000000000000000..dd563b2552d93cfe3a8db3f04db208e691d356eb --- /dev/null +++ "b/image/\346\226\207\347\224\237\345\233\276\345\212\237\350\203\275\347\232\204\344\275\277\347\224\250\350\257\264\346\230\216.MD" @@ -0,0 +1,72 @@ +# 图像生成配置说明 +#### 文生图相关功能的加入，可在此说明文件内加入贡献者的GitHub链接，方便以后的更新，以及BUG的修改！ + + + +智谱AI绘画：[JiQingzhe2004 (JiQingzhe)](https://github.com/JiQingzhe2004) + +阿里云AI绘画：[JiQingzhe2004 (JiQingzhe)](https://github.com/JiQingzhe2004) + +谷歌AI绘画：[JiQingzhe2004 (JiQingzhe)](https://github.com/JiQingzhe2004) + +------ + +在`config.yaml`中进行以下配置才可以调用： + +```yaml +cogview: # -----智谱AI图像生成配置这行不填----- + # 此API请参考 https://www.bigmodel.cn/dev/api/image-model/cogview + enable: False # 是否启用图像生成功能，默认关闭，将False替换为true则开启，此模型可和其他模型同时运行。 + api_key: # 智谱API密钥，请填入您的API Key + model: cogview-4-250304 # 模型编码，可选：cogview-4-250304、cogview-4、cogview-3-flash + quality: standard # 生成质量，可选：standard（快速）、hd（高清） + size: 1024x1024 # 图片尺寸，可自定义，需符合条件 + trigger_keyword: 牛智谱 # 触发图像生成的关键词 + temp_dir: # 临时文件存储目录，留空则默认使用项目目录下的zhipuimg文件夹，如果要更改，例如 D:/Pictures/temp 或 /home/user/temp + fallback_to_chat: true # 当未启用绘画功能时：true=将请求发给聊天模型处理，false=回复固定的未启用提示信息 + +aliyun_image: # -----如果要使用阿里云文生图，取消下面的注释并填写相关内容，模型到阿里云百炼找通义万相-文生图2.1-Turbo----- + enable: true # 是否启用阿里文生图功能，false为关闭，默认开启，如果未配置，则会将消息发送给聊天大模型 + api_key: sk-xxxxxxxxxxxxxxxxxxxxxxxx # 替换为你的DashScope API密钥 + model: wanx2.1-t2i-turbo # 模型名称，默认使用wanx2.1-t2i-turbo(快),wanx2.1-t2i-plus（中）,wanx-v1（慢），会给用户不同的提示！ + size: 1024*1024 # 图像尺寸，格式为宽*高 + n: 1 # 生成图像的数量 + temp_dir: ./temp # 临时文件存储路径 + trigger_keyword: 牛阿里 # 触发词，默认为"牛阿里" + fallback_to_chat: true # 当未启用绘画功能时：true=将请求发给聊天模型处理，false=回复固定的未启用提示信息 + +gemini_image: # -----谷歌AI画图配置这行不填----- + enable: true # 是否启用谷歌AI画图功能 + api_key: your-api-key-here # 谷歌Gemini API密钥，必填 + model: gemini-2.0-flash-exp-image-generation # 模型名称，建议保持默认，只有这一个模型可以进行绘画 + temp_dir: ./geminiimg # 图片保存目录，可选 + trigger_keyword: 牛谷歌 # 触发词，默认为"牛谷歌" + fallback_to_chat: false # 当未启用绘画功能时：true=将请求发给聊天模型处理，false=回复固定的未启用提示信息 +``` + +## 如何获取API密钥 + +1. 访问 [Google AI Studio](https://aistudio.google.com/) +2. 创建一个账号或登录 +3. 访问 [API Keys](https://aistudio.google.com/app/apikeys) 页面 +4. 创建一个新的API密钥 +5. 复制API密钥并填入配置文件 + +## 使用方法 + +直接发送消息或在群聊中@机器人，使用触发词加提示词，例如： + +# 单人聊天的使用 +``` +牛智谱 一只可爱的猫咪在阳光下玩耍 +牛阿里 一只可爱的猫咪在阳光下玩耍 +牛谷歌 一只可爱的猫咪在阳光下玩耍 +``` +## 群组的使用方法 +``` +@ 牛图图 一只可爱的猫咪在阳光下玩耍 + +需要接入机器人的微信名称叫做牛图图 +``` + +生成的图片会自动发送到聊天窗口。 diff --git a/main.py b/main.py index b24b38d3c99e26636d5a27bb8c126c8f795b2aa5..a294d64fe755d9c3bc8e790fb00e073771e42bc5 100644 --- a/main.py +++ b/main.py @@ -24,7 +24,15 @@ def main(chat_type: int): robot.LOG.info(f"WeChatRobot【{__version__}】成功启动···") # 机器人启动发送测试消息 - robot.sendTextMsg("机器人启动成功！", "filehelper") + robot.sendTextMsg("机器人启动成功！\n" + "绘画功能使用说明：\n" + "• 智谱绘画：牛智谱[描述]\n" + "• 阿里绘画：牛阿里[描述]\n" + "• 谷歌绘画：牛谷歌[描述]\n" + "实例：\n" + "牛阿里 画一张家乡\n" + "@XX 牛阿里 画一张家乡\n" + "聊天时直接发送消息即可", "filehelper") # 接收消息 # robot.enableRecvMsg() # 可能会丢消息？ diff --git a/requirements.txt b/requirements.txt index 589dd1c09d099cab644b264963d1423e24d77243..37a28fe22241f17ed2064a2ce7f8fd2752de5e32 100644 --- a/requirements.txt +++ b/requirements.txt @@ -13,6 +13,8 @@ pillow jupyter_client zhdate ipykernel -google-generativeai -zhipuai +google-generativeai>=0.3.0 +zhipuai>=1.0.0 ollama +dashscope +google-genai \ No newline at end of file diff --git a/robot.py b/robot.py index df1fe42ae727ba61a285ed55558886eb66b90fc9..ff4ddb86660c43678e19349201b1605e6aface39 100644 --- a/robot.py +++ b/robot.py @@ -6,7 +6,11 @@ import time import xml.etree.ElementTree as ET from queue import Empty from threading import Thread +import os +import random +import shutil from base.func_zhipu import ZhiPu +from image import CogView, AliyunImage, GeminiImage from wcferry import Wcf, WxMsg @@ -14,6 +18,7 @@ from base.func_bard import BardAssistant from base.func_chatglm import ChatGLM from base.func_ollama import Ollama from base.func_chatgpt import ChatGPT +from base.func_deepseek import DeepSeek from base.func_chengyu import cy from base.func_weather import Weather from base.func_news import News @@ -53,6 +58,8 @@ class Robot(Job): self.chat = ZhiPu(self.config.ZhiPu) elif chat_type == ChatType.OLLAMA.value and Ollama.value_check(self.config.OLLAMA): self.chat = Ollama(self.config.OLLAMA) + elif chat_type == ChatType.DEEPSEEK.value and DeepSeek.value_check(self.config.DEEPSEEK): + self.chat = DeepSeek(self.config.DEEPSEEK) else: self.LOG.warning("未配置模型") self.chat = None @@ -71,23 +78,215 @@ class Robot(Job): self.chat = BardAssistant(self.config.BardAssistant) elif ZhiPu.value_check(self.config.ZhiPu): self.chat = ZhiPu(self.config.ZhiPu) + elif DeepSeek.value_check(self.config.DEEPSEEK): + self.chat = DeepSeek(self.config.DEEPSEEK) else: self.LOG.warning("未配置模型") self.chat = None self.LOG.info(f"已选择: {self.chat}") + # 初始化图像生成服务 + self.cogview = None + self.aliyun_image = None + self.gemini_image = None + + # 初始化Gemini图像生成服务 + try: + if hasattr(self.config, 'GEMINI_IMAGE'): + self.gemini_image = GeminiImage(self.config.GEMINI_IMAGE) + else: + self.gemini_image = GeminiImage({}) + + if getattr(self.gemini_image, 'enable', False): + self.LOG.info("谷歌Gemini图像生成功能已启用") + except Exception as e: + self.LOG.error(f"初始化谷歌Gemini图像生成服务失败: {e}") + + # 初始化CogView和AliyunImage服务 + if hasattr(self.config, 'COGVIEW') and self.config.COGVIEW.get('enable', False): + try: + self.cogview = CogView(self.config.COGVIEW) + self.LOG.info("智谱CogView文生图功能已初始化") + except Exception as e: + self.LOG.error(f"初始化智谱CogView文生图服务失败: {str(e)}") + if hasattr(self.config, 'ALIYUN_IMAGE') and self.config.ALIYUN_IMAGE.get('enable', False): + try: + self.aliyun_image = AliyunImage(self.config.ALIYUN_IMAGE) + self.LOG.info("阿里Aliyun功能已初始化") + except Exception as e: + self.LOG.error(f"初始化阿里云文生图服务失败: {str(e)}") + @staticmethod def value_check(args: dict) -> bool: if args: return all(value is not None for key, value in args.items() if key != 'proxy') return False + def handle_image_generation(self, service_type, prompt, receiver, at_user=None): + """处理图像生成请求的通用函数 + :param service_type: 服务类型，'cogview'/'aliyun'/'gemini' + :param prompt: 图像生成提示词 + :param receiver: 接收者ID + :param at_user: 被@的用户ID，用于群聊 + :return: 处理状态，True成功，False失败 + """ + if service_type == 'cogview': + if not self.cogview or not hasattr(self.config, 'COGVIEW') or not self.config.COGVIEW.get('enable', False): + self.LOG.info(f"收到智谱文生图请求但功能未启用: {prompt}") + fallback_to_chat = self.config.COGVIEW.get('fallback_to_chat', False) if hasattr(self.config, 'COGVIEW') else False + if not fallback_to_chat: + self.sendTextMsg("报一丝，智谱文生图功能没有开启，请联系管理员开启此功能。（可以贿赂他开启）", receiver, at_user) + return True + return False + service = self.cogview + wait_message = "正在生成图像，请稍等..." + elif service_type == 'aliyun': + if not self.aliyun_image or not hasattr(self.config, 'ALIYUN_IMAGE') or not self.config.ALIYUN_IMAGE.get('enable', False): + self.LOG.info(f"收到阿里文生图请求但功能未启用: {prompt}") + fallback_to_chat = self.config.ALIYUN_IMAGE.get('fallback_to_chat', False) if hasattr(self.config, 'ALIYUN_IMAGE') else False + if not fallback_to_chat: + self.sendTextMsg("报一丝，阿里文生图功能没有开启，请联系管理员开启此功能。（可以贿赂他开启）", receiver, at_user) + return True + return False + service = self.aliyun_image + model_type = self.config.ALIYUN_IMAGE.get('model', '') + if model_type == 'wanx2.1-t2i-plus': + wait_message = "当前模型为阿里PLUS模型，生成速度较慢，请耐心等候..." + elif model_type == 'wanx-v1': + wait_message = "当前模型为阿里V1模型，生成速度非常慢，可能需要等待较长时间，请耐心等候..." + else: + wait_message = "正在生成图像，请稍等..." + elif service_type == 'gemini': + if not self.gemini_image or not getattr(self.gemini_image, 'enable', False): + self.sendTextMsg("谷歌文生图服务未启用", receiver, at_user) + return True + + service = self.gemini_image + wait_message = "正在通过谷歌AI生成图像，请稍等..." + else: + self.LOG.error(f"未知的图像生成服务类型: {service_type}") + return False + + self.LOG.info(f"收到图像生成请求 [{service_type}]: {prompt}") + self.sendTextMsg(wait_message, receiver, at_user) + + image_url = service.generate_image(prompt) + + if image_url and (image_url.startswith("http") or os.path.exists(image_url)): + try: + self.LOG.info(f"开始处理图片: {image_url}") + # 谷歌API直接返回本地文件路径，无需下载 + image_path = image_url if service_type == 'gemini' else service.download_image(image_url) + + if image_path: + # 创建一个临时副本，避免文件占用问题 + temp_dir = os.path.dirname(image_path) + file_ext = os.path.splitext(image_path)[1] + temp_copy = os.path.join( + temp_dir, + f"temp_{service_type}_{int(time.time())}_{random.randint(1000, 9999)}{file_ext}" + ) + + try: + # 创建文件副本 + shutil.copy2(image_path, temp_copy) + self.LOG.info(f"创建临时副本: {temp_copy}") + + # 发送临时副本 + self.LOG.info(f"发送图片到 {receiver}: {temp_copy}") + self.wcf.send_image(temp_copy, receiver) + + # 等待一小段时间确保微信API完成处理 + time.sleep(1.5) + + except Exception as e: + self.LOG.error(f"创建或发送临时副本失败: {str(e)}") + # 如果副本处理失败，尝试直接发送原图 + self.LOG.info(f"尝试直接发送原图: {image_path}") + self.wcf.send_image(image_path, receiver) + + # 安全删除文件 + self._safe_delete_file(image_path) + if os.path.exists(temp_copy): + self._safe_delete_file(temp_copy) + + else: + self.LOG.warning(f"图片下载失败，发送URL链接作为备用: {image_url}") + self.sendTextMsg(f"图像已生成，但无法自动显示，点链接也能查看:\n{image_url}", receiver, at_user) + except Exception as e: + self.LOG.error(f"发送图片过程出错: {str(e)}") + self.sendTextMsg(f"图像已生成，但发送过程出错，点链接也能查看:\n{image_url}", receiver, at_user) + else: + self.LOG.error(f"图像生成失败: {image_url}") + self.sendTextMsg(f"图像生成失败: {image_url}", receiver, at_user) + + return True + + def _safe_delete_file(self, file_path, max_retries=3, retry_delay=1.0): + """安全删除文件，带有重试机制 + + :param file_path: 要删除的文件路径 + :param max_retries: 最大重试次数 + :param retry_delay: 重试间隔(秒) + :return: 是否成功删除 + """ + if not os.path.exists(file_path): + return True + + for attempt in range(max_retries): + try: + os.remove(file_path) + self.LOG.info(f"成功删除文件: {file_path}") + return True + except Exception as e: + if attempt < max_retries - 1: + self.LOG.warning(f"删除文件 {file_path} 失败, 将在 {retry_delay} 秒后重试: {str(e)}") + time.sleep(retry_delay) + else: + self.LOG.error(f"无法删除文件 {file_path} 经过 {max_retries} 次尝试: {str(e)}") + + return False + def toAt(self, msg: WxMsg) -> bool: """处理被 @ 消息 :param msg: 微信消息结构 :return: 处理状态，`True` 成功，`False` 失败 """ + # CogView触发词 + cogview_trigger = self.config.COGVIEW.get('trigger_keyword', '牛智谱') if hasattr(self.config, 'COGVIEW') else '牛智谱' + # 阿里文生图触发词 + aliyun_trigger = self.config.ALIYUN_IMAGE.get('trigger_keyword', '牛阿里') if hasattr(self.config, 'ALIYUN_IMAGE') else '牛阿里' + # 谷歌AI画图触发词 + gemini_trigger = self.config.GEMINI_IMAGE.get('trigger_keyword', '牛谷歌') if hasattr(self.config, 'GEMINI_IMAGE') else '牛谷歌' + + content = re.sub(r"@.*?[\u2005|\s]", "", msg.content).replace(" ", "") + + # 阿里文生图处理 + if content.startswith(aliyun_trigger): + prompt = content[len(aliyun_trigger):].strip() + if prompt: + result = self.handle_image_generation('aliyun', prompt, msg.roomid, msg.sender) + if result: + return True + + # CogView处理 + elif content.startswith(cogview_trigger): + prompt = content[len(cogview_trigger):].strip() + if prompt: + result = self.handle_image_generation('cogview', prompt, msg.roomid, msg.sender) + if result: + return True + + # 谷歌AI画图处理 + elif content.startswith(gemini_trigger): + prompt = content[len(gemini_trigger):].strip() + if prompt: + return self.handle_image_generation('gemini', prompt, msg.roomid or msg.sender, msg.sender if msg.roomid else None) + else: + self.sendTextMsg(f"请在{gemini_trigger}后面添加您想要生成的图像描述", msg.roomid or msg.sender, msg.sender if msg.roomid else None) + return True + return self.toChitchat(msg) def toChengyu(self, msg: WxMsg) -> bool: @@ -167,18 +366,44 @@ class Robot(Job): elif msg.type == 10000: # 系统信息 self.sayHiToNewFriend(msg) - elif msg.type == 0x01: # 文本消息 - # 让配置加载更灵活，自己可以更新配置。也可以利用定时任务更新。 + elif msg.type == 0x01: if msg.from_self(): if msg.content == "^更新$": self.config.reload() self.LOG.info("已更新") else: + # 阿里文生图触发词处理 + aliyun_trigger = self.config.ALIYUN_IMAGE.get('trigger_keyword', '牛阿里') if hasattr(self.config, 'ALIYUN_IMAGE') else '牛阿里' + if msg.content.startswith(aliyun_trigger): + prompt = msg.content[len(aliyun_trigger):].strip() + if prompt: + result = self.handle_image_generation('aliyun', prompt, msg.sender) + if result: + return + + # CogView触发词处理 + cogview_trigger = self.config.COGVIEW.get('trigger_keyword', '牛智谱') if hasattr(self.config, 'COGVIEW') else '牛智谱' + if msg.content.startswith(cogview_trigger): + prompt = msg.content[len(cogview_trigger):].strip() + if prompt: + result = self.handle_image_generation('cogview', prompt, msg.sender) + if result: + return + + # 谷歌AI画图触发词处理 + gemini_trigger = self.config.GEMINI_IMAGE.get('trigger_keyword', '牛谷歌') if hasattr(self.config, 'GEMINI_IMAGE') else '牛谷歌' + if msg.content.startswith(gemini_trigger): + prompt = msg.content[len(gemini_trigger):].strip() + if prompt: + result = self.handle_image_generation('gemini', prompt, msg.sender) + if result: + return + self.toChitchat(msg) # 闲聊 def onMsg(self, msg: WxMsg) -> int: try: - self.LOG.info(msg) # 打印信息 + self.LOG.info(msg) self.processMsg(msg) except Exception as e: self.LOG.error(e) @@ -216,7 +441,7 @@ class Robot(Job): # 清除超过1分钟的记录 self._msg_timestamps = [t for t in self._msg_timestamps if now - t < 60] if len(self._msg_timestamps) >= self.config.SEND_RATE_LIMIT: - self.LOG.warning("发送消息过快，已达到每分钟"+self.config.SEND_RATE_LIMIT+"条上限。") + self.LOG.warning(f"发送消息过快，已达到每分钟{self.config.SEND_RATE_LIMIT}条上限。") return self._msg_timestamps.append(now)