llm接口聊天+对话记录到本地

模型用的Qwen2.5-7B-Instruct，快，就是比较傻

LLM_OPENAI_API_URL=https://api.siliconflow.cn/v1

LLM_MODEL_NAME=Qwen/Qwen2.5-7B-Instruct

用户: 神乐大战不是格斗游戏是养成游戏 AI: 谢谢你的澄清。《神乐大战》实际上是一款养成游戏和模拟游戏的结合，玩家需要培养和管理自己的偶像组合。如果你喜欢这种类型的游戏，可以考虑以下几款类似游戏： 1. **《BanG Dream! 少女乐团派对！》**：这是一款模拟乐队养成游戏，玩家可以组建自己的乐队，训练成员，参与演唱会。 2. **《心跳回忆（Beatmania IIDJ）》**：虽然这款游戏主要是音乐节奏游戏，但它也包含了一定的养成元素，玩家需要培养角色，提升技能。 3. **《偶像大师（Negima! 偶像大师）》**：这款游戏也是聚焦于偶像的养成和管理，玩家可以收集和培养各种偶像，参加各种演出活动。 4. **《Stardew Valley》**：虽然不是直接的偶像养成游戏，但这款游戏的日常管理和角色互动元素与养成游戏有相似之处，可以作为一种不同类型的参考。 如果你还有其他具体的兴趣或偏好，欢迎继续分享，我可以为你提供更多针对性的建议。 用户: 神乐大战是什么类型的游戏 AI: 《神乐大战》（Seikyuu Taisen）实际上是一款音乐节奏类游戏，但它与传统的节奏游戏有所不同。在游戏中，玩家需要管理和培养一支由女孩组成的偶像团体，让她们在舞台上演出，同时还要处理她们之间的关系和日常生活。 《神乐大战》的游戏类型可以归类为： 1. **音乐节奏游戏**：玩家需要按照音乐节奏点击屏幕上的图标。 2. **养成游戏**：玩家需要管理和发展角色，提升她们的各项能力。 3. **模拟游戏**：玩家需要模拟和管理一个虚拟的偶像团体的生活和工作。 如果你喜欢这种类型的游戏，除了《神乐大战》，还有其他一些类似的游戏也值得尝试，比如： - **《LoveLive! 学园偶像计划》**：这也是一个音乐节奏游戏，以管理和培养偶像团体为主题。 - **《偶像大师 虚卡莉》**：同样是一款节奏游戏，玩家可以培养和管理偶像。 - **《LoveLive! Sunshine!! Rising Star♪Road!!》**：这也是一个以培养和管理偶像团体为主题的音乐节奏游戏。 这些游戏通常都包含了养成、表演、角色互动等多个元素，非常适合喜欢这种类型游戏的玩家。希望这些推荐能够帮助你找到更多喜欢的游戏！

import tkinter as tk from tkinter import scrolledtext, messagebox, ttk import os from llm_server import LLMService def chat_with_memory(user_message, memory_file=None, use_memory=True): """ 简单的记忆聊天函数，将对话保存到文件 :param user_message: 用户消息 :param memory_file: 记忆文件路径 :param use_memory: 是否使用记忆功能 :return: AI回复内容 """ # 如果没有指定记忆文件路径，则使用当前脚本所在目录下的memory.txt if memory_file is None: current_dir = os.path.dirname(os.path.abspath(__file__)) memory_file = os.path.join(current_dir, "memory.txt") try: # 创建LLM服务实例 llm_service = LLMService() # 准备消息列表 messages = [] # 如果启用记忆功能且记忆文件存在，读取历史对话 if use_memory and os.path.exists(memory_file): with open(memory_file, 'r', encoding='utf-8') as f: content = f.read() # 这里需要解析历史对话，将它们转换为messages格式 # 简单示例：将历史对话按行分割并转换为消息列表 history_messages = parse_history_content(content) messages.extend(history_messages) # 添加当前用户消息 messages.append({"role": "user", "content": user_message}) # 调用LLM服务 result = llm_service.create(messages) # 获取AI回复 ai_response = result['choices'][0]['message']['content'] # 如果启用了记忆功能，则保存到记忆文件 if use_memory: try: # 保存到记忆文件（追加模式） with open(memory_file, 'a', encoding='utf-8') as f: f.write(f"用户: {user_message}\n") f.write(f"AI: {ai_response}\n\n") print(f"成功写入记忆文件: {memory_file}") except Exception as e: print(f"写入文件时出错: {e}") return ai_response except Exception as e: error_msg = f"连接LLM服务失败: {str(e)}" print(error_msg) return error_msg def parse_history_content(content): """ 解析历史对话内容，转换为messages格式 """ messages = [] lines = content.strip().split('\n') current_role = None current_content = [] for line in lines: line = line.strip() if line.startswith('用户:'): # 保存之前的角色内容 if current_role and current_content: messages.append({ "role": current_role, "content": '\n'.join(current_content).strip() }) # 开始新的用户消息 current_role = "user" current_content = [line[3:].strip()] # 去掉"用户:"前缀 elif line.startswith('AI:'): # 保存之前的角色内容 if current_role and current_content: messages.append({ "role": current_role, "content": '\n'.join(current_content).strip() }) # 开始新的AI消息 current_role = "assistant" current_content = [line[3:].strip()] # 去掉"AI:"前缀 elif line == "" and current_content: # 空行表示对话段结束 if current_role and current_content: messages.append({ "role": current_role, "content": '\n'.join(current_content).strip() }) current_role = None current_content = [] elif current_role: # 继续当前消息的内容 current_content.append(line) # 处理最后一条消息 if current_role and current_content: messages.append({ "role": current_role, "content": '\n'.join(current_content).strip() }) return messages class MemoryChatApp: def __init__(self, root): self.root = root self.root.title("AI记忆聊天") self.root.geometry("800x600") # 记忆开关变量 self.use_memory_var = tk.BooleanVar(value=True) # 创建界面 self.setup_ui() # 记忆文件路径 current_dir = os.path.dirname(os.path.abspath(__file__)) self.memory_file = os.path.join(current_dir, "memory.txt") # 加载历史对话 self.load_history() def setup_ui(self): # 顶部框架 - 记忆开关和清除按钮 top_frame = tk.Frame(self.root) top_frame.pack(fill=tk.X, padx=10, pady=5) # 记忆开关复选框 memory_check = tk.Checkbutton( top_frame, text="启用记忆功能", variable=self.use_memory_var ) memory_check.pack(side=tk.LEFT) # 清除对话按钮 clear_btn = tk.Button( top_frame, text="清除对话记录", command=self.clear_history ) clear_btn.pack(side=tk.RIGHT) # 聊天历史显示区域 self.chat_history = scrolledtext.ScrolledText( self.root, wrap=tk.WORD, state='disabled', height=25 ) self.chat_history.pack(padx=10, pady=5, fill=tk.BOTH, expand=True) # 输入区域框架 input_frame = tk.Frame(self.root) input_frame.pack(padx=10, pady=5, fill=tk.X) # 消息输入框 self.input_text = tk.Text(input_frame, height=5) self.input_text.pack(fill=tk.X, pady=5) # 底部按钮框架 button_frame = tk.Frame(input_frame) button_frame.pack(fill=tk.X) # 发送按钮 send_button = tk.Button( button_frame, text="发送 (Enter)", command=self.send_message ) send_button.pack(side=tk.RIGHT, padx=(5, 0)) # 绑定回车键发送消息（Shift+Enter 换行） self.input_text.bind('<Return>', self.on_enter_key) def on_enter_key(self, event): # 检测是否同时按下Shift键，如果是则换行，否则发送消息 if event.state & 0x1: # Shift键被按下 return # 允许正常的换行操作 else: self.send_message() return "break" # 阻止换行操作 def send_message(self): user_message = self.input_text.get("1.0", tk.END).strip() if not user_message: messagebox.showwarning("警告", "请输入消息内容") return # 清空输入框 self.input_text.delete("1.0", tk.END) # 显示用户消息 self.display_message("用户", user_message) # 获取AI回复 use_memory = self.use_memory_var.get() try: ai_response = chat_with_memory(user_message, self.memory_file, use_memory) # 显示AI回复 self.display_message("AI", ai_response) except Exception as e: error_msg = f"获取AI回复时出错: {str(e)}" self.display_message("系统", error_msg) def display_message(self, sender, message): self.chat_history.config(state='normal') self.chat_history.insert(tk.END, f"{sender}: {message}\n\n") self.chat_history.config(state='disabled') # 滚动到底部 self.chat_history.see(tk.END) def load_history(self): """加载历史对话到界面""" if os.path.exists(self.memory_file): with open(self.memory_file, 'r', encoding='utf-8') as f: content = f.read() # 解析历史内容并显示 history_messages = parse_history_content(content) for msg in history_messages: role = "用户" if msg["role"] == "user" else "AI" self.display_message(role, msg["content"]) def clear_history(self): """清除对话记录""" if messagebox.askyesno("确认", "确定要清除所有对话记录吗？"): try: # 清空记忆文件 with open(self.memory_file, 'w', encoding='utf-8') as f: f.write("") # 清空界面显示 self.chat_history.config(state='normal') self.chat_history.delete(1.0, tk.END) self.chat_history.config(state='disabled') messagebox.showinfo("提示", "对话记录已清除") except Exception as e: messagebox.showerror("错误", f"清除记录时出错: {str(e)}") def main(): root = tk.Tk() app = MemoryChatApp(root) root.mainloop() if __name__ == "__main__": main()

llm_server:

import os from dotenv import load_dotenv import json import http.client from urllib.parse import urlparse import base64 import requests from io import BytesIO from PIL import Image class LLMService: def __init__(self): # 从环境变量中获取 DeepSeek 参数 # 加载 .env 文件中的环境变量 dotenv_path = r'E:\code\apikey\.env' load_dotenv(dotenv_path) self.api_url = os.getenv('LLM_OPENAI_API_URL') self.model_name = os.getenv('LLM_MODEL_NAME') self.api_key = os.getenv('LLM_OPENAI_API_KEY') # 检查必需的环境变量是否存在 if not self.api_url: raise ValueError("环境变量 'deepseek_OPENAI_API_URL' 未设置或为空") if not self.model_name: raise ValueError("环境变量 'deepseek_MODEL_NAME' 未设置或为空") if not self.api_key: raise ValueError("环境变量 'deepseek_OPENAI_API_KEY' 未设置或为空") print(f"LLM服务初始化完成，模型: {self.model_name}") def create(self, messages, tools=None): print("开始调用LLM服务") # 解析 URL（去掉协议部分） parsed = urlparse(f"{self.api_url}/chat/completions") host, path = parsed.hostname, parsed.path if not host: raise ValueError("API URL 无效，无法解析主机名") # 创建 HTTP 连接 conn = http.client.HTTPSConnection(host) # 构造请求体 request_body = { "model": self.model_name, "messages": messages, "tools": tools, "temperature": 0.9 # 添加温度参数 } # 发送 POST 请求 headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } conn.request( "POST", path, body=json.dumps(request_body), headers=headers ) # 获取响应 response = conn.getresponse() print(f"LLM服务响应状态码: {response.status}") if response.status != 200: error_msg = response.read().decode('utf-8') raise Exception(f"LLM服务器错误: {response.status} - {error_msg}") # 读取响应内容 response_data = response.read().decode('utf-8') data = json.loads(response_data) # 确保output目录存在 os.makedirs('output', exist_ok=True) # 将响应保存到文件 (修复路径分隔符问题) output_file_path = os.path.join('output', 'formatted_data.json') with open(output_file_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=4, ensure_ascii=False) # 关闭连接 conn.close() print("LLM服务调用完成") return data class VLMService: def __init__(self): # 从环境变量中获取 VLM 参数 dotenv_path = r'E:\code\apikey\.env' load_dotenv(dotenv_path) self.api_url = os.getenv('VLM_OPENAI_API_URL') self.model_name = os.getenv('VLM_MODEL_NAME') self.api_key = os.getenv('VLM_OPENAI_API_KEY') # 检查必需的环境变量是否存在 if not self.api_url: raise ValueError("环境变量 'VLM_OPENAI_API_URL' 未设置或为空") if not self.model_name: raise ValueError("环境变量 'VLM_MODEL_NAME' 未设置或为空") if not self.api_key: raise ValueError("环境变量 'VLM_OPENAI_API_KEY' 未设置或为空") print(f"VLM服务初始化完成，模型: {self.model_name}") def encode_image(self, image_source): """ 编码图像为base64字符串 支持URL和本地文件路径 """ try: if image_source.startswith(('http://', 'https://')): # 从URL获取图像 response = requests.get(image_source) image_data = response.content else: # 从本地文件路径获取图像 with open(image_source, "rb") as image_file: image_data = image_file.read() return base64.b64encode(image_data).decode('utf-8') except Exception as e: raise Exception(f"图像编码失败: {str(e)}") def create_with_image(self, messages, image_source=None, tools=None): print("开始调用VLM服务") # 如果提供了图像源，则处理图像 if image_source: # 编码图像 base64_image = self.encode_image(image_source) # 在第一条消息中添加图像 if messages and len(messages) > 0: # 构建包含图像的消息内容 image_content = { "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{base64_image}" } } # 如果第一条消息的内容是字符串，转换为列表格式 if isinstance(messages[0]["content"], str): messages[0]["content"] = [ { "type": "text", "text": messages[0]["content"] }, image_content ] # 如果已经是列表格式，追加图像内容 elif isinstance(messages[0]["content"], list): messages[0]["content"].append(image_content) # 解析 URL（去掉协议部分） parsed = urlparse(f"{self.api_url}/chat/completions") host, path = parsed.hostname, parsed.path if not host: raise ValueError("API URL 无效，无法解析主机名") # 创建 HTTP 连接 conn = http.client.HTTPSConnection(host) # 构造请求体 request_body = { "model": self.model_name, "messages": messages, "tools": tools, "temperature": 0.7 } # 发送 POST 请求 headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } conn.request( "POST", path, body=json.dumps(request_body), headers=headers ) # 获取响应 response = conn.getresponse() print(f"VLM服务响应状态码: {response.status}") if response.status != 200: error_msg = response.read().decode('utf-8') raise Exception(f"VLM服务器错误: {response.status} - {error_msg}") # 读取响应内容 response_data = response.read().decode('utf-8') data = json.loads(response_data) # 确保output目录存在 os.makedirs('output', exist_ok=True) # 将响应保存到文件 output_file_path = os.path.join('output', 'vlm_formatted_data.json') with open(output_file_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=4, ensure_ascii=False) # 关闭连接 conn.close() print("VLM服务调用完成") return data