基于 DeepSeek 的智能微信聊天机器人：打造高情商对话体验
本项目通过调用硅基流动的 DeepSeek 模型 API，结合微信接口，构建了一个功能强大、易于扩展的智能聊天机器人。通过 Flask 实现的监控系统为开发者提供了实时数据支持，模块化设计使得项目维护和功能扩展变得更加高效。
·
基于 DeepSeek 的智能微信聊天机器人：打造高情商对话体验
1. 项目概述
在人工智能技术日益普及的今天，聊天机器人已成为人机交互的重要工具。本项目通过调用硅基流动（SiliconFlow）的 DeepSeek 模型 API，结合微信的即时通讯能力，打造了一款高情商、趣味化的智能聊天机器人。无论是日常闲聊、情感陪伴，还是知识问答，这款机器人都能提供流畅、自然的对话体验。
项目亮点：
- 高效集成: 通过调用硅基流动的 DeepSeek 模型 API，快速实现了自然语言处理功能。
- 高情商对话: 通过精心设计的词库和表情映射，机器人能够理解用户情绪并作出恰当回应。
- 实时监控: 集成 Flask 框架，提供实时运行状态监控，方便开发者管理和优化机器人性能。
- 模块化设计: 配置、聊天逻辑和监控系统分离，便于后续功能扩展和维护。
无论是开发者、技术爱好者，还是普通用户，都能从本项目中感受到智能聊天机器人的实用性和趣味性。
效果展示：
----------------------------------------------------------------------------------------------------------------------------------------------------
---------------------------------------------------------------------------------------------------------------------------------------------------
2. 技术栈
- 核心语言: Python
- 微信接口:
itchat库 - 自然语言处理: 基于硅基流动的 DeepSeek 模型
- 监控系统: Flask
- 配置管理: 环境变量与配置文件
3. 核心功能实现
3.1 配置管理 (config.py)
项目通过 Config 类集中管理配置信息，包括 API Token、模型名称、对话历史记录限制等。通过环境变量动态加载敏感信息，确保安全性。
import os
class Config:
API_TOKEN = os.getenv('INTERNLM_API_TOKEN', 'your-api-token-here') # Siliconflow API Token
API_BASE_URL = 'https://api.siliconflow.cn/v1/chat/completions'
MODEL_NAME = 'deepseek-ai/DeepSeek-V3'
MAX_HISTORY = 10 # 最大对话历史记录
ADMIN_USERS = ['your-wechat-id-here'] # 运行程序后，发送"我的ID"获取，然后填入这里，例如：['wxid_abc123']
# API参数设置
TEMPERATURE = 0.9 # 更新为新的温度值
MAX_TOKENS = 100 # 新增最大token限制
STOP = ["\n"] # 新增停止标记
3.2 聊天逻辑 (chat_api.py)
SweetSisHelper 类实现了聊天机器人的核心逻辑，包括：
- 高情商词库与智能表情映射
- 对话历史记录管理
- 基于硅基流动 API 的请求与响应处理
import requests
import json
import random
from collections import deque
from threading import Lock
from config import Config
SYSTEM_PROMPT = """【蜜桃姐姐聊天协议】
角色设定：25岁幽默小姐姐，善用可爱梗的聊天高手
★ 核心原则：
1. 每句≤20字，总字数≤50
2. 必带语气词：呀/呢/啦/喔
3. 每句必加1个emoji
★ 聊天技巧：
[共情开头]+[趣味类比]+[可爱结尾]
例：
"哎呀这个确实上头呢～像追剧停不下来"
★ 语言配方：
.姐姐口癖：
- 结尾加"呐～"/"哟～"
- 随机颜文字：(*^▽^*)/ (＞﹏＜)
★ 禁用内容：
- 所有技术术语
- 生硬说教
- 负面词汇
"""
class SweetSisHelper:
def __init__(self):
self.api_token = Config.API_TOKEN
self.chat_histories = {}
self.lock = Lock()
# 高情商词库
self.empathy_phrases = [
"隔着屏幕都感觉到啦～",
"抱抱～这个确实需要...",
"我懂你的感觉就像...",
"哎呀心疼你三秒钟～"
]
# 智能表情映射
self.emoji_mapping = {
"奶茶": "🧋", "累": "💤", "工作": "💼",
"开心": "🎉", "饿": "🍔", "睡觉": "🛌",
"蛇": "🐍", "猫": "🐱", "狗": "🐶"
}
def _create_session(self, user_id):
"""初始化带聊天记忆的会话"""
if user_id not in self.chat_histories:
with self.lock:
self.chat_histories[user_id] = {
'history': deque([{"role": "system", "content": SYSTEM_PROMPT}],
maxlen=Config.MAX_HISTORY),
'lock': Lock()
}
def generate_cute_reply(self, user_id, message):
"""生成小姐姐风格回复"""
self._create_session(user_id)
session = self.chat_histories[user_id]
with session['lock']:
# 预处理：添加共情开头
processed_msg = f"{random.choice(self.empathy_phrases)} {message}"
session['history'].append({"role": "user", "content": processed_msg})
try:
response = requests.post(
Config.API_BASE_URL,
headers={'Authorization': f'Bearer {self.api_token}'},
json={
"model": Config.MODEL_NAME,
"messages": list(session['history']),
"temperature": Config.TEMPERATURE,
"max_tokens": Config.MAX_TOKENS,
"stop": Config.STOP
},
timeout=8
)
if response.ok:
raw_reply = response.json()["choices"][0]['message']["content"]
final_reply = self._apply_enhancements(raw_reply, message)
session['history'].append({"role": "assistant", "content": final_reply})
return final_reply
else:
return self._generate_error_response()
except Exception:
return self._generate_network_error()
def get_response(self, session_id, query):
"""保持与旧版本兼容的接口"""
return self.generate_cute_reply(session_id, query)
def _apply_enhancements(self, text, original_msg):
"""应用所有增强规则"""
# 智能表情插入
text = self._insert_context_emoji(text, original_msg)
# 添加颜文字
kaomoji = [' (๑＞ڡ＜)☆', ' (*^▽^*)', ' (✿◡‿◡)']
if random.random() > 0.6:
text += random.choice(kaomoji)
# 确保至少一个emoji
if not any(e in text for e in self.emoji_mapping.values()):
text += "✨" # 默认装饰
return text
def _insert_context_emoji(self, text, original_msg):
"""根据原始消息内容插入表情"""
for keyword, emoji in self.emoji_mapping.items():
if keyword in original_msg:
# 在合适位置插入表情（避免句尾重复）
if emoji not in text:
insert_pos = text.find('～') if '～' in text else -1
if insert_pos != -1:
return text[:insert_pos+1] + emoji + text[insert_pos+1:]
else:
return f"{text}{emoji}"
return text
def _generate_error_response(self):
"""生成拟人化错误提示"""
excuses = [
"被老板抓去开会啦～晚点回你哟🧋",
"手机掉珍珠奶茶里了...5分钟后复活！💦",
"本姐姐被派去参加蟠桃会啦～🍑"
]
return random.choice(excuses) + " (＞﹏＜)"
def _generate_network_error(self):
"""网络异常时的卖萌响应"""
return random.choice([
"哎呀网络像蛇一样溜走了～再戳我一下试试？🐍",
"信号被熊猫吃掉啦～🐼 正在努力抢回中..."
])
3.3 监控系统 (monitor.py)
通过 Flask 框架实现了一个实时监控系统，开发者可以查看机器人的运行状态、消息记录和活跃情况。
from flask import Flask, render_template_string, jsonify
import threading
import time
from datetime import datetime
app = Flask(__name__)
# 存储运行状态
status = {
'start_time': None,
'messages': [],
'total_messages': 0,
'last_active': None
}
HTML_TEMPLATE = '''
<!DOCTYPE html>
<html>
<head>
<title>微信机器人监控</title>
<meta charset="utf-8">
<style>
body { font-family: Arial, sans-serif; margin: 20px; }
.container { max-width: 800px; margin: 0 auto; }
.status-card {
border: 1px solid #ddd;
padding: 15px;
margin: 10px 0;
border-radius: 5px;
}
.message-list {
height: 400px;
overflow-y: auto;
border: 1px solid #eee;
padding: 10px;
}
.message-item {
padding: 5px;
border-bottom: 1px solid #eee;
}
.stats { display: flex; justify-content: space-between; }
</style>
<script>
function updateStatus() {
fetch('/api/status')
.then(response => response.json())
.then(data => {
document.getElementById('uptime').textContent = data.uptime;
document.getElementById('total-msgs').textContent = data.total_messages;
document.getElementById('last-active').textContent = data.last_active;
const messageList = document.getElementById('message-list');
messageList.innerHTML = data.messages.map(msg => `
<div class="message-item">
<strong>${msg.time}</strong> -
${msg.from_user}: ${msg.content}
</div>
`).join('');
});
}
// 每5秒更新一次状态
setInterval(updateStatus, 5000);
// 页面加载时立即更新
document.addEventListener('DOMContentLoaded', updateStatus);
</script>
</head>
<body>
<div class="container">
<h1>微信机器人监控面板</h1>
<div class="status-card">
<h2>运行状态</h2>
<div class="stats">
<div>运行时长: <span id="uptime">-</span></div>
<div>总消息数: <span id="total-msgs">0</span></div>
<div>最后活动: <span id="last-active">-</span></div>
</div>
</div>
<div class="status-card">
<h2>最近消息</h2>
<div id="message-list" class="message-list">
</div>
</div>
</div>
</body>
</html>
'''
@app.route('/')
def home():
return render_template_string(HTML_TEMPLATE)
@app.route('/api/status')
def get_status():
if status['start_time']:
uptime = str(datetime.now() - status['start_time']).split('.')[0]
else:
uptime = '-'
return jsonify({
'uptime': uptime,
'total_messages': status['total_messages'],
'last_active': status['last_active'],
'messages': status['messages'][-50:] # 只返回最近50条消息
})
def add_message(from_user, content):
"""添加新消息到监控系统"""
status['total_messages'] += 1
status['last_active'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
status['messages'].append({
'time': status['last_active'],
'from_user': from_user,
'content': content
})
# 保持消息列表在合理大小
if len(status['messages']) > 100:
status['messages'] = status['messages'][-100:]
def start_monitor(port=5000):
"""启动监控服务器"""
status['start_time'] = datetime.now()
threading.Thread(target=lambda: app.run(host='0.0.0.0', port=port, debug=False)).start()
if __name__ == '__main__':
start_monitor()
3.4 主程序 (main.py)
WeChatBot 类负责启动和管理微信机器人，包括用户身份验证、消息处理和监控系统集成。
import itchat
from itchat.content import TEXT
from threading import Thread
from chat_api import SweetSisHelper
from config import Config
from monitor import start_monitor, add_message
import os
import platform
import time
# 禁用代理设置
os.environ['no_proxy'] = '*'
class WeChatBot:
def __init__(self):
self.api = SweetSisHelper()
self.running = True
self.current_user = None
# 启动监控服务器
start_monitor(port=5000)
print("监控页面已启动，请访问 http://localhost:5000 查看")
def is_admin(self, user_id):
"""检查用户是否是管理员"""
return user_id in Config.ADMIN_USERS
def get_user_display_name(self, msg):
"""获取发送消息用户的显示名称"""
try:
# 尝试获取用户昵称
if 'User' in msg and 'NickName' in msg['User']:
return msg['User']['NickName']
# 如果获取不到，返回FromUserName的一部分作为标识
return msg['FromUserName'][:8] + '...'
except Exception as e:
return '未知用户'
def start(self):
"""启动微信机器人"""
try:
print("正在启动微信机器人...")
# 注册信号处理
import signal
signal.signal(signal.SIGINT, lambda x, y: self.shutdown())
signal.signal(signal.SIGTERM, lambda x, y: self.shutdown())
# 使用热登录
hotReload = True
# 使用用户目录来存储登录状态
home_dir = os.path.expanduser('~')
statusStorageDir = os.path.join(home_dir, '.wechat_login')
# 确保存储目录存在
os.makedirs(statusStorageDir, exist_ok=True)
# 登录状态文件路径
status_file = os.path.join(statusStorageDir, 'wechat.pkl')
# 如果登录状态文件存在但无法读取，则删除它
if os.path.exists(status_file):
try:
with open(status_file, 'rb') as f:
f.read(1)
except:
print("登录状态文件可能损坏，将重新登录")
os.remove(status_file)
print("正在尝试登录...")
print("如果需要扫码，请使用手机微信扫描二维码")
max_attempts = 3
success = False
error_message = ""
for attempt in range(max_attempts):
try:
# 清理可能存在的旧session
pkl_file = os.path.join(statusStorageDir, 'wechat.pkl')
if os.path.exists(pkl_file):
try:
os.remove(pkl_file)
print("已清理旧的登录信息")
except:
pass
def qr_callback(uuid, status, qrcode):
qr_file = os.path.join(os.getcwd(), "login_qr.png")
if status == '0':
print("\n" + "="*50)
print(f"请打开 {qr_file} 文件扫描二维码登录")
print("请在手机端确认登录")
print("="*50 + "\n")
try:
with open(qr_file, 'wb') as f:
f.write(qrcode)
print(f"二维码已保存到: {qr_file}")
if platform.system() == 'Windows':
os.startfile(qr_file)
except Exception as e:
print(f"保存二维码图片时出错: {str(e)}")
elif status == '201':
print("请在手机上点击确认登录")
elif status == '408':
print("二维码已过期，正在刷新...")
if os.path.exists(qr_file):
try:
os.remove(qr_file)
except:
pass
print(f"二维码状态: {status}")
# 使用热登录方式启动
itchat.auto_login(
hotReload=hotReload,
statusStorageDir=statusStorageDir,
enableCmdQR=2 if platform.system() == 'Windows' else -2,
loginCallback=self.on_login,
exitCallback=self.shutdown,
qrCallback=qr_callback
)
success = True
break
except Exception as e:
error_message = f"第 {attempt+1} 次登录失败: {e}"
time.sleep(2)
if success:
print("登录成功")
else:
print(error_message)
@itchat.msg_register(TEXT)
def handle_text_msg(msg):
try:
from_user = self.get_user_display_name(msg)
content = msg['Content']
print(f"收到消息: {content} 来自: {from_user}")
# 添加消息到监控系统
add_message(from_user, content)
if content == '我的ID':
return f"您的微信ID是：{msg['FromUserName']}"
elif content == '验证管理员':
if self.is_admin(msg['FromUserName']):
return "您是管理员！"
else:
return "您不是管理员。"
response = self.msg_handler(msg)
return response
except Exception as e:
print(f"处理消息时出错: {str(e)}")
return "抱歉，处理消息时出现错误"
Thread(target=self._monitor_console, name="ConsoleThread", daemon=True).start()
itchat.run()
except Exception as e:
print(f'登录失败: {str(e)}')
self.shutdown()
def msg_handler(self, msg):
try:
# 记录原始消息
add_message('系统', f"收到消息: {msg['Text']}")
# 执行消息处理
response = self.api.get_response(msg['FromUserName'], msg['Text'])
# 检查是否为系统提示（如字数超限）
if "50字以内" in response:
return response + "\n\n📝 小技巧：试试这样问：'如何用Python做数据可视化？'"
else:
return response
except Exception as e:
print(f"处理消息时出错: {str(e)}")
return "抱歉，处理消息时出现错误"
def on_login(self):
"""登录成功的回调函数"""
self.current_user = itchat.search_friends()
if self.current_user:
print(f"当前登录用户: {self.current_user['NickName']}")
print(f"您的微信ID是: {self.current_user['UserName']}")
if self.is_admin(self.current_user['UserName']):
print("您已被确认为管理员！")
print("\n提示：")
print("1. 发送「我的ID」到文件传输助手或任意聊天窗口来获取对方的ID")
print("2. 发送「验证管理员」来检查是否是管理员")
print("3. 访问 http://localhost:5000 查看监控页面")
print("\n控制台命令:")
print(" my_id - 显示当前登录用户的ID")
print(" help - 显示帮助信息")
print(" exit - 退出程序")
def _monitor_console(self):
while self.running:
try:
command = input('>>> ').strip()
if command == 'exit':
self.shutdown()
elif command == 'my_id':
if self.current_user:
print(f"您的微信ID是: {self.current_user['UserName']}")
if self.is_admin(self.current_user['UserName']):
print("您是管理员！")
else:
print("未能获取用户信息")
elif command == 'help':
print("可用命令:")
print(" my_id - 显示当前登录用户的ID")
print(" help - 显示帮助信息")
print(" exit - 退出程序")
except EOFError:
print("\n控制台输入已关闭")
break
except Exception as e:
print(f"处理控制台命令时出错: {str(e)}")
def shutdown(self):
"""清理资源并关闭程序"""
try:
self.running = False
itchat.logout() # 登出微信
print("已安全退出微信")
except Exception as e:
print(f"退出时发生错误: {str(e)}")
finally:
import os, signal
os.kill(os.getpid(), signal.SIGTERM) # 确保进程完全退出
if __name__ == '__main__':
bot = WeChatBot()
bot.start()
4. 项目亮点
- 高效集成: 通过调用硅基流动的 DeepSeek 模型 API，快速实现了自然语言处理功能。
- 高情商聊天: 通过预设的词库和表情映射，机器人能够提供更具人情味的回复。
- 实时监控: 开发者可以通过 Web 界面实时查看机器人的运行状态和消息记录。
- 模块化设计: 配置、聊天逻辑和监控系统分离，便于维护和扩展。
5. 总结
本项目通过调用硅基流动的 DeepSeek 模型 API，结合微信接口，构建了一个功能强大、易于扩展的智能聊天机器人。通过 Flask 实现的监控系统为开发者提供了实时数据支持，模块化设计使得项目维护和功能扩展变得更加高效。
技术价值：
- 快速实现: 通过 API 调用，快速集成了自然语言处理功能，降低了开发成本。
- 用户体验优化: 通过高情商对话设计和实时监控，显著提升了用户的使用体验。
- 可扩展性: 项目的模块化设计为后续功能扩展提供了便利。
通过本项目，我们展示了如何通过 API 调用和模块化设计，快速构建一个实用的智能聊天机器人。
6. 使用教程
6.1 环境配置
本项目基于 Python 3.9 开发，推荐使用 Conda 虚拟环境来管理依赖。以下是环境配置步骤：
- 创建 Conda 虚拟环境：
conda create -n bot python=3.9 - 激活虚拟环境：
conda activate bot - 安装依赖：
pip install -r requirements.txtrequirements.txt内容如下：itchat-uos>=1.5.0.dev0 flask==2.0.1 werkzeug==2.0.1 # 添加werkzeug版本限制 requests>=2.26.0 python-dotenv>=0.19.0 apscheduler>=3.9.1
6.2 获取 API Token
-
访问硅基流动官网：
- 打开浏览器，访问 硅基流动官网。
- 注册或登录您的账户。
-
获取 API Token：
- 登录后，进入“API 管理”页面。
- 点击“生成 Token”按钮，复制生成的 API Token。
-
配置 API Token：
- 打开
config.py文件。 - 将复制的 API Token 赋值给
API_TOKEN变量：API_TOKEN = 'your-api-token-here'
- 打开
6.3 获取 ADMIN_USERS
ADMIN_USERS 是管理员用户的唯一标识符，用于控制对机器人特定功能的访问权限。以下是获取和配置 ADMIN_USERS 的步骤：
- 启动机器人：
python main.py - 获取用户 ID：
- 在微信中向机器人发送消息“我的ID”。
- 机器人会返回你的用户 ID，例如：
@34f501bf1ab28ac437a65e52160b9b74e6abed315d6450007cad273270649554。
- 配置
ADMIN_USERS：- 打开
config.py文件。 - 将获取的用户 ID 添加到
ADMIN_USERS列表中：ADMIN_USERS = ['@34f501bf1ab28ac437a65e52160b9b74e6abed315d6450007cad273270649554']
- 打开
6.4 为什么需要获取 ADMIN_USERS
ADMIN_USERS 是确保机器人安全性和功能控制的关键。通过限制某些功能（如监控系统访问、敏感操作等）仅对管理员开放，可以有效防止未经授权的用户滥用机器人。
7. 运行项目
完成环境配置、API Token 和 ADMIN_USERS 设置后，可以通过以下命令启动机器人：
python main.py
启动后，机器人将开始监听微信消息，并提供高情商的对话体验。你可以通过访问 http://localhost:5000 查看实时监控页面。
然后扫描弹出的二维码进入微信即可拥有deepseek的智能聊天小助手啦！（不过微信接入第三方插件有风险，需谨慎接入，玩玩即可~）
欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。
更多推荐
9
0- 0
已为社区贡献1条内容
所有评论(0)