基于 Python + DeepSeek 的 QQ 私聊智能机器人项目实战(附核心源代码)

发布时间:2026/5/24 11:30:06

基于 Python + DeepSeek 的 QQ 私聊智能机器人项目实战(附核心源代码) 基于 Python DeepSeek 的 QQ 私聊智能机器人项目实战附核心源代码最近自己做了一个 QQ 私聊 AI 机器人原本只是想简单接一个大模型接口做聊天后来越做越多就慢慢给它加上了 长期记忆、定时提醒、网页配置后台 这些功能。到现在这个项目已经不只是“发一句回一句”的聊天机器人了而更像是一个轻量级的个人智能助手系统。这篇文章我就把整个项目的思路、功能拆分和核心代码整理出来方便自己留档也给想做类似项目的朋友一个参考。---一、项目能做什么目前这个 QQ 机器人主要实现了下面几类功能1. QQ 私聊对话用户在 QQ 私聊窗口给机器人发消息机器人后端收到消息后会调用 DeepSeek API 生成回复再把结果发送回 QQ。整体流程是QQ用户发消息 - Python后端接收 - 调用DeepSeek - 回发到QQ---2. 长期记忆这个功能主要是为了让机器人不只是“看当前一句话”而是能逐渐记住用户。比如用户可以发记住我喜欢吃辣或者记住我平时用 Windows这些内容会被保存到本地 JSON 文件里。后续机器人在调用 DeepSeek 时会把这些记忆一起带上让回答更符合当前用户的习惯和信息。---3. 定时任务这个模块让机器人具备“主动发消息”的能力。比如每天晚上提醒学习固定时间发送通知后续扩展随机时间发消息任务信息会持久化保存因此程序重启后不会丢失。---4. 网页可视化配置为了避免每次改提示词或任务都去手动改代码我还做了一个简单的网页控制台。这个网页可以用来修改系统提示词配置机器人参数添加 / 删除定时任务导出配置---二、项目使用的技术栈这个项目目前的技术栈并不复杂主要包括PythonFlaskQQ 机器人接口DeepSeek API本地 JSON 存储HTML JavaScript整体目标不是堆很多框架而是尽量用简单、容易维护的方式把功能真正跑通。---三、项目目录结构我这里给一个比较简单的目录结构参考project/├── app.py├── config.json├── memories.json├── index.html说明app.py主后端程序config.json保存机器人配置和任务memories.json保存每个用户的长期记忆index.html网页配置后台---四、核心源代码下面给出一版核心代码示例。---1. 配置和记忆文件读写这一部分负责把配置和记忆持久化保存到本地。import osimport jsonfrom copy import deepcopyfrom threading import LockCONFIG_FILE config.jsonMEMORY_FILE memories.jsonLOCK Lock()DEFAULT_CONFIG {prompt: 你是一个有帮助的 QQ 私聊助手。回答简洁、自然、中文优先。,bot: {appId: ,secret: ,apiKey: ,baseUrl: https://api.deepseek.com,model: deepseek-chat,autoReply: True},tasks: []}def load_json_file(path, default_value):with LOCK:if not os.path.exists(path):with open(path, w, encodingutf-8) as f:json.dump(default_value, f, ensure_asciiFalse, indent2)return deepcopy(default_value)with open(path, r, encodingutf-8) as f:try:return json.load(f)except Exception:return deepcopy(default_value)def save_json_file(path, data):with LOCK:with open(path, w, encodingutf-8) as f:json.dump(data, f, ensure_asciiFalse, indent2)def load_config():data load_json_file(CONFIG_FILE, DEFAULT_CONFIG)merged deepcopy(DEFAULT_CONFIG)merged.update(data)if isinstance(data.get(bot), dict):merged[bot].update(data[bot])if not isinstance(merged.get(tasks), list):merged[tasks] []return mergeddef save_config(data):save_json_file(CONFIG_FILE, data)def load_memories():data load_json_file(MEMORY_FILE, {})return data if isinstance(data, dict) else {}def save_memories(data):save_json_file(MEMORY_FILE, data)---2. 长期记忆功能这一部分负责记忆的增删改查。def get_user_memories(user_openid):memories load_memories()items memories.get(user_openid, [])return items if isinstance(items, list) else []def add_user_memory(user_openid, content):memories load_memories()items memories.get(user_openid, [])if not isinstance(items, list):items []content content.strip()if not content:return Falseif content not in items:items.append(content)memories[user_openid] itemssave_memories(memories)return Truedef remove_user_memory(user_openid, content):memories load_memories()items memories.get(user_openid, [])if not isinstance(items, list):return Falseif content in items:items.remove(content)memories[user_openid] itemssave_memories(memories)return Truereturn Falsedef clear_user_memory(user_openid):memories load_memories()memories[user_openid] []save_memories(memories)return Truedef format_user_memories(user_openid):items get_user_memories(user_openid)if not items:return 暂无长期记忆。return 长期记忆\n \n.join(f- {x} for x in items)---3. DeepSeek 调用代码这里是调用大模型的核心逻辑。from openai import OpenAIdef build_deepseek_client():config load_config()bot_cfg config[bot]return OpenAI(api_keybot_cfg.get(apiKey, ),base_urlbot_cfg.get(baseUrl, https://api.deepseek.com))def build_messages(user_openid, user_text, session_histories):config load_config()prompt config.get(prompt, DEFAULT_CONFIG[prompt])memories get_user_memories(user_openid)messages [{role: system, content: prompt}]if memories:memory_text 以下是该用户的长期记忆请在回答时适当参考\n \n.join(f- {m} for m in memories)messages.append({role: system, content: memory_text})messages.extend(session_histories[user_openid])messages.append({role: user, content: user_text})return messagesdef ask_deepseek(user_openid, user_text, session_histories):config load_config()model config[bot].get(model, deepseek-chat)client build_deepseek_client()messages build_messages(user_openid, user_text, session_histories)response client.chat.completions.create(modelmodel,messagesmessages,streamFalse,timeout30,)reply response.choices[0].message.content or return reply.strip() or 我刚刚没有生成有效回复。---4. QQ 私聊消息处理这里是机器人最核心的交互部分。from collections import defaultdictsession_histories defaultdict(list)MAX_HISTORY 10def save_history(user_openid, user_text, assistant_text):history session_histories[user_openid]history.append({role: user, content: user_text})history.append({role: assistant, content: assistant_text})max_items MAX_HISTORY * 2if len(history) max_items:session_histories[user_openid] history[-max_items:]如果你已经接好了 QQ 机器人 SDK那么在收到私聊消息时可以这样处理命令def handle_user_message(user_openid, user_text):lower user_text.lower().strip()if lower in (我的记忆, 查看记忆, 记忆):return format_user_memories(user_openid)if lower in (清空记忆,):clear_user_memory(user_openid)return 长期记忆已清空。if user_text.startswith(记住) or user_text.startswith(记住:):content user_text.split(, 1)[1] if in user_text else user_text.split(:, 1)[1]content content.strip()if not content:return 要记住的内容不能为空。add_user_memory(user_openid, content)return f已记住{content}if user_text.startswith(忘记) or user_text.startswith(忘记:):content user_text.split(, 1)[1] if in user_text else user_text.split(:, 1)[1]content content.strip()ok remove_user_memory(user_openid, content)return 已删除该记忆。 if ok else 没有找到这条记忆。reply ask_deepseek(user_openid, user_text, session_histories)save_history(user_openid, user_text, reply)return reply---5. 定时任务系统这是一个最基础的每日提醒逻辑。import timefrom datetime import datetimedef normalize_task(task):task deepcopy(task)task.setdefault(id, str(int(time.time() * 1000)))task.setdefault(enabled, True)task.setdefault(type, daily)task.setdefault(time, 18:30)task.setdefault(content, )task.setdefault(targetOpenid, )task.setdefault(createdAt, datetime.now().isoformat())task.setdefault(lastRunAt, )return taskdef should_run_task(task, now_dt):if not task.get(enabled, True):return Falsetask_time task.get(time, 18:30)last_run_at task.get(lastRunAt, )try:hour, minute map(int, task_time.split(:))except Exception:return Falsetoday_run now_dt.replace(hourhour, minuteminute, second0, microsecond0)if now_dt today_run:return Falseif not last_run_at:return Truetry:last_run datetime.fromisoformat(last_run_at)except Exception:return Truereturn last_run.date() now_dt.date()发送消息的地方这里先用打印模拟def send_qq_message(task):target task.get(targetOpenid) or 未指定用户content task.get(content, ).strip()print( * 50)print(触发定时任务)print(目标:, target)print(内容:, content)print(时间:, datetime.now().strftime(%Y-%m-%d %H:%M:%S))print( * 50)return True调度循环def scheduler_loop():while True:try:config load_config()changed Falsenow_dt datetime.now()for i, raw_task in enumerate(config[tasks]):task normalize_task(raw_task)config[tasks][i] taskif should_run_task(task, now_dt):ok send_qq_message(task)if ok:task[lastRunAt] now_dt.isoformat()changed Trueif changed:save_config(config)except Exception as e:print(定时循环异常:, repr(e))time.sleep(5)---6. Flask 后台接口为了给网页配置页提供接口我这里加了 Flask。from flask import Flask, jsonify, request, send_from_directoryapp Flask(__name__, static_folder., static_url_path)app.route(/)def index():return send_from_directory(., index.html)app.route(/api/config, methods[GET])def get_config_api():return jsonify(load_config())app.route(/api/config, methods[POST])def update_config_api():data request.get_json(forceTrue)if not isinstance(data, dict):return jsonify({ok: False, error: 配置格式错误}), 400config deepcopy(DEFAULT_CONFIG)config.update(data)if isinstance(data.get(bot), dict):config[bot].update(data[bot])tasks data.get(tasks, [])if not isinstance(tasks, list):return jsonify({ok: False, error: tasks 必须是数组}), 400config[tasks] [normalize_task(t) for t in tasks]save_config(config)return jsonify({ok: True, config: config})启动import threadingdef start_scheduler():t threading.Thread(targetscheduler_loop, daemonTrue)t.start()if __name__ __main__:start_scheduler()app.run(host0.0.0.0, port5000, debugTrue)---五、前端网页配置页示例前端我做成了一个单文件 HTML双击也能看配合 Flask 也能跑。下面只放一个最精简的思路代码!DOCTYPE htmlhtml langzh-CNheadmeta charsetUTF-8titleQQ机器人控制台/title/headbodyh1QQ机器人配置/h1label系统提示词/labelbrtextarea idprompt rows6 cols60/textareabrbrbutton onclicksaveConfig()保存配置/buttonscriptasync function loadConfig() {const res await fetch(/api/config);const data await res.json();document.getElementById(prompt).value data.prompt || ;}async function saveConfig() {const prompt document.getElementById(prompt).value;const res await fetch(/api/config, {method: POST,headers: {Content-Type: application/json},body: JSON.stringify({prompt: prompt,bot: {},tasks: []})});const data await res.json();alert(data.ok ? 保存成功 : 保存失败);}loadConfig();/script/body/html---六、项目运行方式1. 安装依赖pip install flask openai如果你还要接 QQ 官方机器人 SDK再安装对应 SDK。2. 准备配置文件config.json{prompt: 你是一个有帮助的 QQ 私聊助手。回答简洁、自然、中文优先。,bot: {appId: ,secret: ,apiKey: 你的DeepSeekKey,baseUrl: https://api.deepseek.com,model: deepseek-chat,autoReply: true},tasks: []}memories.json{}3. 启动后端python app.py4. 浏览器打开后台http://127.0.0.1:5000---七、项目特点总结我觉得这个项目最有意思的地方在于它已经不是一个纯粹的“聊天 Demo”了而开始具备一些真正像助手的能力。它的特点可以概括成能聊天能记住用户能主动提醒能在网页里改配置结构清晰方便继续扩展后面如果继续往下做还可以继续加随机时间自动发消息工作流逻辑可视化任务管理更多记忆策略更完整的后台系统---八、总结这个 QQ 私聊机器人项目本质上是一个基于 Python 搭建的小型 AI 助手系统。它结合了 DeepSeek 的对话能力、长期记忆、定时任务以及网页配置后台不再只是一个“问一句答一句”的实验项目而是具备了进一步扩展成完整智能助手的基础。如果你也想自己练一个“从聊天机器人走向智能助手”的项目这个方向还是很适合拿来练手的。---如果你要我下一条可以继续给你两样东西里的一个1. 整合后的完整 app.py 成品版2. 适合 CSDN 直接发布的完整排版版文章

相关新闻