### Discord Bot簡章:Discord 機器人結合 OpenAI GPT
這份簡章旨在詳細介紹如何利用 Python、OpenAI GPT API 以及 Discord API 建立一個智能聊天機器人,此機器人能夠自動回應用戶消息並進行互動。
#### 功能概述
1. **消息處理與回應**:
- 機器人會監聽 Discord 服務器上的消息。
- 當收到用戶消息時,機器人將該消息保存到本地 JSON 文件以記錄聊天歷史。
- 利用 OpenAI GPT 產生回應並將回應發送給用戶。
2. **聊天歷史管理**:
- 機器人將聊天歷史保存在本地 JSON 文件中。
- 每當收到或發送消息時更新聊天歷史。
- 定期清理超過 20 分鐘的舊消息以節省空間並保持歷史的相關性。
3. **自動重啟功能**:
- 每 40 分鐘後自動重啟機器人,以確保系統穩定運行。
- 重啟後重新連接 Discord 並繼續處理消息。
#### 主要程式碼段落
1. **初始化與設定**:
- 設定 OpenAI API 鑰匙和 Discord 連接令牌。
- 設定機器人的意圖,允許讀取和處理消息。
```python
openai_api_key = "your_openai_api_key"
discord_token = "your_discord_token"
openai.api_key = openai_api_key
intents = discord.Intents.default()
intents.messages = True
intents.message_content = True
bot = discord.Client(intents=intents)
```
2. **消息處理**:
- 監聽來自 Discord 的消息。
- 使用 OpenAI 的 GPT 模型生成回應。
```python
@bot.event
async def on_message(message):
if message.author == bot.user:
return
save_message("user", str(message.author.id), message.content)
# 構建消息列表供 OpenAI API 使用
chat_history = load_chat_history()
messages = [{"role": msg["role"], "content": msg["content"]} for msg in chat_history]
try:
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages,
)
reply_text = response.choices[0].message['content'].strip()
await message.channel.send(reply_text)
save_message("assistant", "AI", reply_text)
except Exception as e:
print(f'發生錯誤:{e}')
await message.channel.send("抱歉發生錯誤,無法處理您的消息。")
```
#### 使用與部署指南
1. **設定 API 鑰匙**:確保您已有有效的 OpenAI API 鑰匙和 Discord 機器人令牌。
2. **部署機器人**:將此程式碼部署到您的服務器或任何支持 Python 的運行環境。
3. **邀請機器人加入 Discord 服務器**:使用
Discord 提供的機制將機器人加入到您的服務器中。
這個機器人展示了如何結合最新的 AI 技術與即時通訊工具來創建一個互動式聊天機器人,這種機器人可被應用於客服、娛樂或教育等多種場景。
```python
import discord
import openai
import json
import datetime
import asyncio
openai_api_key = "your_openai_api_key"
discord_token = "your_discord_token"
openai.api_key = openai_api_key
intents = discord.Intents.default()
intents.messages = True
intents.message_content = True
bot = discord.Client(intents=intents)
chat_history_file = "chat_history.json"
def load_chat_history():
"""從JSON文件中加載聊天歷史"""
try:
with open(chat_history_file, "r", encoding='utf-8') as file:
return json.load(file)
except FileNotFoundError:
return []
def save_message(role, author_id, message_content):
"""將消息及其角色保存到JSON文件中,帶有時間戳"""
chat_history = load_chat_history()
chat_history.append({"role": role, "author_id": author_id, "content": message_content, "timestamp": datetime.datetime.now().isoformat()})
with open(chat_history_file, "w", encoding='utf-8') as file:
json.dump(chat_history, file, indent=4)
async def restart_bot(bot):
"""定時重啟機器人"""
await bot.wait_until_ready()
while not bot.is_closed():
print("計劃在40分鐘後重啟...")
await asyncio.sleep(2400) # 等待40分鐘
print("正在重啟機器人...")
await bot.close() # 關閉機器人連接
await bot.start(discord_token) # 重新啟動機器人
print("機器人已重啟")
async def clean_old_messages():
"""定期清理超過20分鐘的消息,保留system消息"""
while True:
try:
with open(chat_history_file, "r+", encoding='utf-8') as file:
chat_history = json.load(file)
current_time = datetime.datetime.now()
# 過濾出含有 timestamp 字段且時間在20分鐘內的記錄,或者是系統消息
filtered_history = [
msg for msg in chat_history
if ('timestamp' not in msg and msg.get('role') == 'system') or
('timestamp' in msg and (current_time - datetime.datetime.fromisoformat(msg['timestamp'])) < datetime.timedelta(minutes=15))
]
file.seek(0)
file.truncate()
json.dump(filtered_history, file, indent=4, ensure_ascii=False)
except Exception as e:
print(f"更新聊天歷史時出錯: {e}")
await asyncio.sleep(60) # 每分鐘檢查一次
@bot.event
async def on_ready():
print(f'已登錄為 {bot.user.name}')
try:
with open(chat_history_file, "r", encoding='utf-8') as file:
print("聊天歷史文件已存在。")
except FileNotFoundError:
with open(chat_history_file, "w", encoding='utf-8') as file:
json.dump([], file)
print("已創建聊天歷史文件。")
bot.loop.create_task(restart_bot(bot))
bot.loop.create_task(clean_old_messages()) # 開始執行定期清理任務
@bot.event
async def on_message(message):
if message.author == bot.user:
return
# 保存用戶消息
save_message("user", str(message.author.id), message.content)
# 加載聊天歷史,為API調用準備消息列表
chat_history = load_chat_history()
messages = [{"role": msg["role"], "content": msg["content"]} for msg in chat_history if "role" in msg and "content" in msg]
messages.append({"role": "user", "content": message.content})
try:
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=messages,
)
reply_text = response.choices[0].message['content'].strip()
# 保存機器人的回復
save_message("assistant", "AI", reply_text)
except Exception as e:
print(f'發生錯誤:{e}')
reply_text = "抱歉發生錯誤,無法處理您的消息。"
await message.channel.send(reply_text)
bot.run(discord_token)
```