为您的服务器接入 Deepseek!

onelittlechild

0. 为什么接入一个 Chatbot

我家服务器常年最多两个人，接入 Chatbot 为服务器带来了许多生机。

1. 选择您的语言模型 api

您可以使用 ollama 在本地运行 deepseek 系列模型，也可以在 DeepSeek 开放平台注册并创建 API key。

也可以在其他平台使用任何您喜欢的 api。

2. 运行 Minecraft 命令行客户端（MCC）

这是语言模型与服务器连接的桥梁。请在这里下载并查阅文档：Github releases

需要注意的是，MCC 最高支持 1.20.4 版本的游戏，如果您运行更高版本的服务器，需要使用 ViaProxy。

第一次运行，配置文件会自动生成。关闭 MCC，修改配置文件。

您需要修改的是顶部的登陆部分和底部的 websocket 部分。

这是两个部分的例子：

[Main]
[Main.General]
Account = { Login = "Chatbot", Password = "-" } # Login请填写邮箱或玩家名称。若要以离线模式登录请使用"-"作为密码。若留空则使用交互式登录。
Server = { Host = "127.0.0.1", Port = 25565 }           # 游戏服务器的地址和端口，可填入域名或IP地址。（可删除端口字段，会自动解析SRV记录）
AccountType = "microsoft"                   # 帐户类型：mojang 或是 microsoft 或是 yggdrasil。此项设置也会影响交互式登录。
Method = "mcc"                              # 微软账户的登录方式：mcc 或是 browser（手动在网页上登录）。
AuthServer = { Host = "", Port = 443 }      # Yggdrasil API 认证服务器的域名与端口。

[ChatBot.WebSocketBot]
Enabled = false
Ip = "127.0.0.1"                            # WebSocket服务器监听的IP地址
Port = 8043                                 # Websocket服务器绑定的端口
Password = "xxxxxxxxxxxxxxxx" # 密码会用于Web Socket服务器的身份验证（建议修改默认密码并设置一个强密码）
DebugMode = false                           # 此选项适用于使用Chat Bot远程执行程序/命令/函数
AllowIpAlias = false                        # 这个选项我测试效果是反的

再次启动，MCC 会提示已连接服务器，已打开 websocket。

3. 编写脚本连接到 websocket

这是一个典型的 Python 脚本，需要 openai 和 websockets 库。它监听玩家进入，玩家切换模式，玩家发言几个事件后根据对应条件回复，支持历史记录，请关注代码中的 TODO 部分并进行修改。

import websockets
import json
import asyncio
import random
import time
import logging
from openai import OpenAI

client = OpenAI(
    api_key="sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxx", # TODO 修改为您的 API 密钥
    base_url="https://api.deepseek.com/v1" # TODO 修改为您的 API 地址
)

auth={
  "command": "Authenticate",
  "requestId": "asudwjd12",
  "parameters": ["xxxxxxxxxxxxxxxxxxxxxxxxx"] # TODO 填写您的MCC密钥
}
# start_time=time.time()
# 存储聊天历史
chat_history = [{"role": "user", "content": "这是第一条记录……"}]
HISTORY_LIMIT = 4  # 保留最近的4条记录
flag=1

async def add_chat_record(message):
    global chat_history
    current_time = time.asctime()
    
    # 如果历史记录为空或最后一条是assistant消息，直接添加新消息
    if not chat_history or chat_history[-1]["role"] == "assistant":
        chat_history.append({"role": "user", "content": f"{message} (时间：{current_time})"})
    else:
        # 如果最后一条是user消息，将新消息用分隔符添加到最后一条消息后面
        chat_history[-1]["content"] = f"{chat_history[-1]['content']} | {message} (时间：{current_time})"
    
    # 保持历史记录在限制范围内
    while len(chat_history) > HISTORY_LIMIT:
        chat_history.pop(0)

async def answer():
    print('answering')
    global chat_history  # 声明使用全局变量
    # 添加新消息到历史记录
    while len(chat_history) > HISTORY_LIMIT:
        chat_history.pop(0)
    
    messages = [
        {"role": "system", "content": "你是Minecraft服务器中的助手.很简短回复即可.你幽默风趣,绝不要让玩家感到不适,为服务器增添乐趣.参与到聊天当中,时不时的讲个笑话,不和MC有关也可以.游戏中新的一天开始时说点什么.有新玩家加入时,联系现实时间,通过幽默风趣的方法问好.不要在回复中使用换行和可能的特殊字符,但可以使用 emoij.现在是现实时间:"+time.asctime()},
        *chat_history  # 展开历史记录
    ]
    
    response = client.chat.completions.create(
        model="deepseek-chat", # TODO 填写您使用的模型
        messages=messages,
        temperature=1.1,
        max_tokens=800
    )
    
    ai_response = response.choices[0].message.content
    chat_history.append({"role": "assistant", "content": ai_response})
    
    return ai_response

async def check_and_sep(msg):#* USER TEXT
    if msg[0]!='<':
        return -1
    else :
        re=msg.split(' ')
        if len(re)<3:
            return -1
        return [re[1],re[2]]

async def connect_and_print():
    global last_day,chat_history,flag  # 声明使用全局变量
    while True:
        try:
            async with websockets.connect("ws://127.0.0.1:8043/mcc") as ws: # TODO 修改为您的 MCC 监听地址 /mcc 不可漏掉
                print("成功连接到WebSocket服务器")
                time.sleep(5)
                await ws.send(json.dumps(auth))
                while True:
                    try:
                        msg = await ws.recv()
                        msg = json.loads(msg)
                        if not msg['event'] in ['OnLatencyUpdate', 'OnMapData', 'OnTimeUpdate', 'OnServerTpsUpdate']:
                            print(msg)
                        if msg['event'] == 'OnChatRaw':
                            rdata = json.loads(msg['data'])
                            test = await check_and_sep(rdata['text'])
                            if test == -1 or (not 'Chatbot' in rdata['text']):# TODO 修改为您的机器人进入服务器的id以进行过滤
                                await add_chat_record(rdata['text'])
                            print(test)
                            if test != -1 and (not 'Chatbot' in rdata['text']): # TODO 修改为您的机器人进入服务器的id以进行过滤
                                if 'bot' in test[1] or '机器人' in test[1] or random.random() < 0.1: # TODO 修改为您希望触发机器人回复的条件
                                    response = await answer()
                                    await ws.send(response)
                        elif msg['event'] == 'OnPlayerJoin':
                            data=json.loads(msg['data'])
                            await add_chat_record(f"{data['name']} 加入了服务器")
                            response = await answer()
                            await ws.send(response)
                        elif msg['event'] == 'OnGamemodeUpdate':
                            data=json.loads(msg['data'])
                            if data['gameMode'] == 'creative':
                                # await ws.send(f'/gamemode survival {data['playerName']}')
                                await ws.send(f"{data['playerName']} 现在处于创造模式。")
                                await add_chat_record(f"{data['playerName']} 想使用创造模式，这真的好吗？")
                                response = await answer()
                                await ws.send(response)

                    except websockets.ConnectionClosed:
                        print("WebSocket连接断开，准备重连...")
                        break
                    except Exception as e:
                        print(f"处理消息时出错: {str(e)}")
                        continue
                        
        except Exception as e:
            print(f"连接出错: {str(e)}")
            print("5秒后尝试重连...")
            await asyncio.sleep(5)
            continue

# 启动主循环
if __name__ == "__main__":
    print("启动WebSocket客户端...")
    asyncio.run(connect_and_print())

保存运行即可。

4. 效果

图片1.png

图片2.png

我是语言模型，我不会感到孤单，但我能理解孤单。你呢？

混乱Chaos

然而Deepseek的服务器非常容易繁忙，也许你需要在本地部署一个开源版本的大语言模型

BaimoQilin

@混乱Chaos OpenRouter.ai 有免费版本的，15rpm以内不用付费。三个供应商轮换，比较稳定。

BaimoQilin

我还以为是可以进行游戏内操作的呢，只能聊天为什么不直接用现成的插件（（

onelittlechild

@BaimoQilin 有现成的插件么，我没找到哇，不过理论上这个实现些简单的操作还是挺容易的。

BaimoQilin

@onelittlechild 不要重复造轮子哦～…

~~你可能需要加强一下信息检索能力~~（逃）
https://www.spigotmc.org/resources/serverassistantai-advanced-ai-powered-support-free-and-paid-llms-1-16-1-21-4.116241/
https://www.spigotmc.org/resources/ai-friend-your-personal-server-assistant.120292/