From 20672def798b5b2571b1f4b351e4d9b4b973af90 Mon Sep 17 00:00:00 2001
From: ljzloser <1312358581@qq.com>
Date: Fri, 17 Apr 2026 19:22:22 +0800
Subject: [PATCH 1/2] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0LLM=E6=8F=92?=
=?UTF-8?q?=E4=BB=B6=E7=B3=BB=E7=BB=9F=E5=92=8C=E6=B8=B8=E6=88=8F=E6=8E=A7?=
=?UTF-8?q?=E5=88=B6=E5=91=BD=E4=BB=A4?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
src/main.py | 29 +-
src/mineSweeperGUI.py | 123 +++-
src/mineSweeperGUIEvent.py | 3 +
.../llm_minesweeper_controller/__init__.py | 8 +
.../llm_minesweeper_controller/api_client.py | 217 ++++++
.../llm_minesweeper_controller/config.py | 104 +++
.../function_registry.py | 278 ++++++++
.../llm_minesweeper_controller/plugin.py | 636 ++++++++++++++++++
.../llm_minesweeper_controller/widgets.py | 136 ++++
src/shared_types/__init__.py | 2 +
src/shared_types/commands.py | 35 +-
src/shared_types/events.py | 47 +-
12 files changed, 1603 insertions(+), 15 deletions(-)
create mode 100644 src/plugins/llm_minesweeper_controller/__init__.py
create mode 100644 src/plugins/llm_minesweeper_controller/api_client.py
create mode 100644 src/plugins/llm_minesweeper_controller/config.py
create mode 100644 src/plugins/llm_minesweeper_controller/function_registry.py
create mode 100644 src/plugins/llm_minesweeper_controller/plugin.py
create mode 100644 src/plugins/llm_minesweeper_controller/widgets.py
diff --git a/src/main.py b/src/main.py
index b92dd7e..1397579 100644
--- a/src/main.py
+++ b/src/main.py
@@ -18,7 +18,8 @@
# 插件系统（新）
from plugin_sdk import GameServerBridge
from plugin_manager.app_paths import get_env_for_subprocess
-from shared_types.commands import NewGameCommand
+from shared_types.commands import NewGameCommand, MouseClickCommand
+from shared_types.enums import GameLevel
import subprocess
os.environ["QT_FONT_DPI"] = "96"
@@ -202,12 +203,34 @@ def cli_check_file(file_path: str) -> int:
# 注册控制命令处理器（自动在主线程执行）
def handle_new_game(cmd: NewGameCommand):
- print(f"[NewGameCommand] rows={cmd.rows}, cols={cmd.cols}, mines={cmd.mines}")
- ui.setBoard_and_start(cmd.rows, cmd.cols, cmd.mines)
+ """处理新游戏命令"""
from lib_zmq_plugins.shared.base import CommandResponse
+
+ # 根据 level 确定参数
+ if cmd.level == GameLevel.BEGINNER.value:
+ rows, cols, mines = 8, 8, 10
+ elif cmd.level == GameLevel.INTERMEDIATE.value:
+ rows, cols, mines = 16, 16, 40
+ elif cmd.level == GameLevel.EXPERT.value:
+ rows, cols, mines = 16, 30, 99
+ else:
+ # 自定义模式，使用传入的参数
+ rows, cols, mines = cmd.rows, cmd.cols, cmd.mines
+
+ print(f"[NewGameCommand] level={cmd.level}, rows={rows}, cols={cols}, mines={mines}")
+ ui.setBoard_and_start(rows, cols, mines)
return CommandResponse(request_id=cmd.request_id, success=True)
+ def handle_mouse_click(cmd: MouseClickCommand):
+ """处理鼠标点击命令"""
+ from lib_zmq_plugins.shared.base import CommandResponse
+
+ print(f"[MouseClickCommand] row={cmd.row}, col={cmd.col}, button={cmd.button}")
+ success = ui.execute_cell_click(cmd.row, cmd.col, cmd.button)
+ return CommandResponse(request_id=cmd.request_id, success=success)
+
GameServerBridge.instance().register_handler(NewGameCommand, handle_new_game)
+ GameServerBridge.instance().register_handler(MouseClickCommand, handle_mouse_click)
# _translate = QtCore.QCoreApplication.translate
hwnd = int(ui.mainWindow.winId())
diff --git a/src/mineSweeperGUI.py b/src/mineSweeperGUI.py
index fbe40ad..2ae7cb8 100644
--- a/src/mineSweeperGUI.py
+++ b/src/mineSweeperGUI.py
@@ -7,7 +7,7 @@
# from PyQt5.QtWidgets import QApplication, QFileDialog, QWidget
import gameDefinedParameter
from plugin_sdk.server_bridge import GameServerBridge
-from shared_types.events import VideoSaveEvent
+from shared_types.events import VideoSaveEvent, BoardUpdateEvent, GameStatusChangeEvent
import superGUI
import gameAbout
import gameSettings
@@ -219,6 +219,8 @@ def game_state(self):
@game_state.setter
def game_state(self, game_state: str):
# print(self._game_state, " -> " ,game_state)
+ last_state = self._game_state
+
match self._game_state:
case "playing":
self.try_append_evfs(game_state)
@@ -252,7 +254,29 @@ def game_state(self, game_state: str):
self.label.paint_cursor = False
self.label.paintProbability = False
self.num_bar_ui.QWidget.close()
+
self._game_state = game_state
+
+ # 发送游戏状态变化事件
+ state_map = {
+ "ready": 1,
+ "playing": 2,
+ "win": 3,
+ "fail": 4,
+ "joking": 2, # joking 也视为游戏中
+ "jowin": 3,
+ "jofail": 4,
+ "show": 5,
+ "study": 6,
+ "display": 7,
+ "showdisplay": 8,
+ }
+ if last_state != game_state:
+ event = GameStatusChangeEvent(
+ last_status=state_map.get(last_state, 0),
+ current_status=state_map.get(game_state, 0),
+ )
+ GameServerBridge.instance().send_event(event)
@property
def row(self):
@@ -477,6 +501,100 @@ def mineNumWheel(self, i):
# self.timer_mine_num.setSingleShot(True)
# self.timer_mine_num.start(3000)
+ def _send_board_update_event(self):
+ """发送棋盘更新事件给插件"""
+ try:
+ ms_board = self.label.ms_board
+ # 将 game_board 转换为列表格式
+ game_board_list = []
+ for row in ms_board.game_board:
+ game_board_list.append(list(row))
+
+ event = BoardUpdateEvent(
+ rows=self.row,
+ cols=self.column,
+ game_board=game_board_list,
+ mines_remaining=self.mineUnFlagedNum,
+ game_time=ms_board.time if hasattr(ms_board, 'time') else 0.0,
+ )
+ GameServerBridge.instance().send_event(event)
+ except Exception:
+ pass # 忽略发送失败
+
+ def execute_cell_click(self, row: int, col: int, button: int) -> bool:
+ """
+ 执行格子点击（供外部命令调用）
+
+ Args:
+ row: 行索引（从 0 开始）
+ col: 列索引（从 0 开始）
+ button: 鼠标按钮（0=左键, 2=右键）
+
+ Returns:
+ True 表示成功执行
+ """
+ # 检查游戏状态
+ if self.game_state not in ('ready', 'playing', 'joking'):
+ return False
+
+ # 检查坐标有效性
+ if row < 0 or row >= self.row or col < 0 or col >= self.column:
+ return False
+
+ # 转换为像素坐标（中心点）
+ i = row * self.pixSize + self.pixSize // 2
+ j = col * self.pixSize + self.pixSize // 2
+
+ try:
+ if button == 0: # 左键
+ # 模拟点击流程：按下 -> 抬起
+ self.label.ms_board.step('lc', (i, j))
+
+ # 处理第一次点击埋雷
+ if self.game_state == 'ready':
+ if self.label.ms_board.mouse_state == 4:
+ if self.board_constraint:
+ self.game_state = 'joking'
+ else:
+ self.game_state = 'playing'
+ if self.player_identifier[:6] != "[live]":
+ self.disable_screenshot()
+ if self.cursor_limit:
+ self.limit_cursor()
+ self.start_time_unix_2 = QtCore.QDateTime.currentDateTime().toMSecsSinceEpoch()
+ self.timer_10ms.start()
+ self.score_board_manager.editing_row = -2
+ # 埋雷
+ self.layMine(row, col)
+
+ self.label.ms_board.step('lr', (i, j))
+
+ # 检查游戏结束
+ if self.label.ms_board.game_board_state == 3:
+ self.gameWin()
+ elif self.label.ms_board.game_board_state == 4:
+ self.gameFailed()
+
+ elif button == 2: # 右键
+ # 更新剩余雷数
+ cell_state = self.label.ms_board.game_board[row][col]
+ if cell_state == 11: # 已标旗，取消标旗
+ self.mineUnFlagedNum += 1
+ self.showMineNum(self.mineUnFlagedNum)
+ elif cell_state == 10: # 未揭开，标旗
+ self.mineUnFlagedNum -= 1
+ self.showMineNum(self.mineUnFlagedNum)
+
+ self.label.ms_board.step('rc', (i, j))
+ self.label.ms_board.step('rr', (i, j))
+
+ self.label.update()
+ self._send_board_update_event()
+ return True
+
+ except Exception:
+ return False
+
def gameStart(self):
# 画界面，但是不埋雷。等价于点脸、f2、设置确定后的效果
self.mineUnFlagedNum = self.minenum # 没有标出的雷，显示在左上角
@@ -570,6 +688,9 @@ def gameFinished(self):
data[key] = getattr(ms_board, key)
event = VideoSaveEvent(**data)
GameServerBridge.instance().send_event(event)
+
+ # 发送棋盘更新事件，让插件知道最终状态
+ self._send_board_update_event()
def gameWin(self): # 成功后改脸和状态变量，停时间
self.timer_10ms.stop()
diff --git a/src/mineSweeperGUIEvent.py b/src/mineSweeperGUIEvent.py
index 2a3210f..61db2ca 100644
--- a/src/mineSweeperGUIEvent.py
+++ b/src/mineSweeperGUIEvent.py
@@ -69,6 +69,7 @@ def mineAreaLeftRelease(self, i, j):
return
else:
self.label.update()
+ self._send_board_update_event()
self.set_face(14)
elif self.game_state == 'playing' or self.game_state == 'joking':
# 如果是游戏中，且是左键抬起（不是双击），且是在10上，且在局面内，则用ai劫持、处理下
@@ -88,6 +89,7 @@ def mineAreaLeftRelease(self, i, j):
self.label.update()
return
self.label.update()
+ self._send_board_update_event()
self.set_face(14)
elif self.game_state == 'show':
@@ -121,6 +123,7 @@ def mineAreaRightRelease(self, i, j):
self.chording_ai(i // self.pixSize, j // self.pixSize)
self.label.ms_board.step('rr', (i, j))
self.label.update()
+ self._send_board_update_event()
self.set_face(14)
elif self.game_state == 'show':
# 看概率时，所有操作都移出局面外
diff --git a/src/plugins/llm_minesweeper_controller/__init__.py b/src/plugins/llm_minesweeper_controller/__init__.py
new file mode 100644
index 0000000..100f63c
--- /dev/null
+++ b/src/plugins/llm_minesweeper_controller/__init__.py
@@ -0,0 +1,8 @@
+"""
+llm_minesweeper_controller - 使用requests进行反向控制扫雷并处理func calling的插件
+"""
+from __future__ import annotations
+
+from .plugin import LlmMinesweeperControllerPlugin
+
+__all__ = ["LlmMinesweeperControllerPlugin"]
\ No newline at end of file
diff --git a/src/plugins/llm_minesweeper_controller/api_client.py b/src/plugins/llm_minesweeper_controller/api_client.py
new file mode 100644
index 0000000..35c51af
--- /dev/null
+++ b/src/plugins/llm_minesweeper_controller/api_client.py
@@ -0,0 +1,217 @@
+"""
+OpenAI兼容的大模型API客户端 - 使用requests + tools格式function calling
+"""
+from __future__ import annotations
+
+import json
+import requests
+from typing import Dict, Any, Optional, List
+from dataclasses import dataclass, field
+
+
+@dataclass
+class ChatResponse:
+ """聊天响应数据类"""
+ success: bool
+ status_code: int = 0
+ raw_data: Optional[Dict[str, Any]] = None
+ error: Optional[str] = None
+
+ # 解析后的内容
+ content: Optional[str] = None # 文本内容
+ tool_calls: Optional[List[Dict[str, Any]]] = None # tool_calls 列表
+ finish_reason: Optional[str] = None
+
+ @property
+ def has_tool_calls(self) -> bool:
+ """是否有 tool_calls"""
+ return bool(self.tool_calls)
+
+ @property
+ def has_content(self) -> bool:
+ """是否有文本内容"""
+ return bool(self.content)
+
+
+class LLMClient:
+ """OpenAI兼容的大模型客户端"""
+
+ def __init__(
+ self,
+ api_key: str,
+ base_url: str = "https://api.openai.com/v1",
+ model: str = "gpt-4o-mini",
+ timeout: int = 60,
+ ):
+ self.api_key = api_key
+ self.base_url = base_url.rstrip("/")
+ self.model = model
+ self.timeout = timeout
+ self.session = requests.Session()
+ self.session.headers.update({
+ "Authorization": f"Bearer {api_key}",
+ "Content-Type": "application/json",
+ })
+
+ def chat(
+ self,
+ messages: List[Dict[str, Any]],
+ tools: Optional[List[Dict[str, Any]]] = None,
+ temperature: float = 0.3,
+ max_tokens: Optional[int] = None,
+ ) -> ChatResponse:
+ """
+ 发送聊天请求 (OpenAI /v1/chat/completions)
+
+ Args:
+ messages: 消息列表，格式为 [{"role": "user/assistant/system/tool", "content": "..."}]
+ tools: OpenAI tools格式的函数定义列表
+ temperature: 温度参数
+ max_tokens: 最大token数
+
+ Returns:
+ ChatResponse
+ """
+ url = f"{self.base_url}/chat/completions"
+
+ payload: Dict[str, Any] = {
+ "model": self.model,
+ "messages": messages,
+ "temperature": temperature,
+ }
+
+ if max_tokens is not None:
+ payload["max_tokens"] = max_tokens
+
+ if tools:
+ payload["tools"] = tools
+ payload["tool_choice"] = "auto"
+
+ return self._request("POST", url, payload)
+
+ def test_connection(self) -> ChatResponse:
+ """测试API连接"""
+ messages = [{"role": "user", "content": "Hi, reply with OK."}]
+ return self.chat(messages=messages, max_tokens=10)
+
+ def _request(self, method: str, url: str, payload: Dict[str, Any]) -> ChatResponse:
+ """执行HTTP请求并解析响应"""
+ try:
+ response = self.session.request(
+ method, url, json=payload, timeout=self.timeout
+ )
+
+ response_data = None
+ if response.content:
+ try:
+ response_data = response.json()
+ except json.JSONDecodeError:
+ response_data = {"raw_response": response.text}
+
+ if response.status_code >= 400:
+ error_msg = ""
+ if response_data and "error" in response_data:
+ error_msg = response_data["error"].get("message", "")
+ return ChatResponse(
+ success=False,
+ status_code=response.status_code,
+ raw_data=response_data,
+ error=error_msg or f"HTTP {response.status_code}",
+ )
+
+ # 解析成功响应
+ return self._parse_success_response(response_data)
+
+ except requests.exceptions.Timeout:
+ return ChatResponse(success=False, error=f"请求超时 ({self.timeout}秒)")
+ except requests.exceptions.ConnectionError:
+ return ChatResponse(success=False, error="连接失败，请检查网络和API地址")
+ except requests.exceptions.RequestException as e:
+ return ChatResponse(success=False, error=f"请求异常: {str(e)}")
+ except Exception as e:
+ return ChatResponse(success=False, error=f"未知错误: {str(e)}")
+
+ def _parse_success_response(self, response_data: Dict[str, Any]) -> ChatResponse:
+ """解析成功的API响应"""
+ try:
+ choice = response_data.get("choices", [{}])[0]
+ message = choice.get("message", {})
+
+ # 提取文本内容
+ content = message.get("content")
+
+ # 提取 tool_calls
+ tool_calls = message.get("tool_calls")
+
+ # 提取 finish_reason
+ finish_reason = choice.get("finish_reason")
+
+ return ChatResponse(
+ success=True,
+ status_code=200,
+ raw_data=response_data,
+ content=content,
+ tool_calls=tool_calls,
+ finish_reason=finish_reason,
+ )
+ except Exception as e:
+ return ChatResponse(
+ success=True,
+ status_code=200,
+ raw_data=response_data,
+ error=f"解析响应失败: {str(e)}",
+ )
+
+ @staticmethod
+ def build_tool_result_message(tool_call_id: str, result: Any) -> Dict[str, Any]:
+ """
+ 构建 tool 结果消息
+
+ Args:
+ tool_call_id: tool_call 的 ID
+ result: 函数执行结果（会被 JSON 序列化）
+
+ Returns:
+ 可直接追加到 messages 的消息字典
+ """
+ if isinstance(result, str):
+ content = result
+ else:
+ content = json.dumps(result, ensure_ascii=False)
+
+ return {
+ "role": "tool",
+ "tool_call_id": tool_call_id,
+ "content": content,
+ }
+
+ @staticmethod
+ def build_assistant_tool_message(
+ content: Optional[str],
+ tool_calls: List[Dict[str, Any]]
+ ) -> Dict[str, Any]:
+ """
+ 构建 assistant 的 tool_calls 消息（用于多轮对话时保存历史）
+
+ Args:
+ content: 文本内容（可能为 None）
+ tool_calls: tool_calls 列表
+
+ Returns:
+ 可追加到 messages 的消息字典
+ """
+ msg = {"role": "assistant"}
+ if content:
+ msg["content"] = content
+ if tool_calls:
+ msg["tool_calls"] = tool_calls
+ return msg
+
+ def close(self):
+ self.session.close()
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.close()
\ No newline at end of file
diff --git a/src/plugins/llm_minesweeper_controller/config.py b/src/plugins/llm_minesweeper_controller/config.py
new file mode 100644
index 0000000..0aff491
--- /dev/null
+++ b/src/plugins/llm_minesweeper_controller/config.py
@@ -0,0 +1,104 @@
+"""
+llm_minesweeper_controller - 配置定义
+"""
+from __future__ import annotations
+
+from plugin_sdk import OtherInfoBase, BoolConfig, IntConfig, TextConfig, LongTextConfig, ChoiceConfig
+
+
+class LlmMinesweeperControllerConfig(OtherInfoBase):
+ """插件配置"""
+
+ # LLM API设置
+ api_key = TextConfig(
+ default="",
+ label="API密钥",
+ )
+
+ api_base_url = TextConfig(
+ default="https://api.openai.com/v1",
+ label="API基础URL",
+ )
+
+ model_name = TextConfig(
+ default="gpt-4o-mini",
+ label="模型名称",
+ )
+
+ request_timeout = IntConfig(
+ default=60,
+ label="请求超时(秒)",
+ )
+
+ # 游戏设置
+ default_difficulty = ChoiceConfig(
+ default="medium",
+ label="默认游戏难度",
+ choices=[
+ ("easy", "初级 (8x8, 10雷)"),
+ ("medium", "中级 (16x16, 40雷)"),
+ ("hard", "高级 (16x30, 99雷)"),
+ ],
+ )
+
+ # 功能开关
+ enable_auto_action = BoolConfig(
+ default=False,
+ label="自动执行LLM操作(否则需确认)",
+ )
+
+ # 提示词设置
+ system_prompt = LongTextConfig(
+ default="""你是一个顶级扫雷AI。你的任务是通过逻辑推理，输出工具调用指令来赢下游戏。
+
+# 一、 棋盘与状态定义
+- 格子状态：`-1`(未揭开)、`0-8`(周围雷数)、`F`(已标旗)、`M`(踩到的红雷)、`m`(未踩的白雷)
+- 游戏状态容错：若棋盘出现 `M` 必为失败；若非雷格全揭开必为胜利；以棋盘实际画面为准，忽略错误的状态参数。
+
+# 二、 可用工具
+- `get_board_state()`：获取全局视图。
+- `get_local_board(col, row, radius)`：获取局部细节，建议 radius=4。
+- `click_cell(col, row, button)`：执行操作，button仅限 `"left"`(揭开) 或 `"right"`(标旗)。
+- `start_new_game()`：失败或未初始化时调用。
+
+# 三、 核心推理策略（按优先级排序）
+1. **基础定式**：
+ - 数字 = 周围未揭开数 → 未揭开格全是雷（右键标旗）。
+ - 数字 = 周围旗子数 → 剩余未揭开格全安全（左键揭开）。
+2. **减法逻辑（核心）**：
+ - 对比边界上相邻的两个数字，利用它们的差值与非共享未知格的数量，推断特定格子是雷还是安全。
+3. **盲猜原则（仅限无任何逻辑解时）**：
+ - **绝对禁止猜边角！**
+ - 必须选择**长连续未揭开边界的中段**点击，以最大化获取信息量。
+
+# 四、 操作铁律
+1. **100%确定原则**：没有绝对把握不操作，宁可不动也不犯错。
+2. **单次限量**：每次推理后，只执行 1-3 个确定格子的操作。
+3. **标旗优先**：在既可标旗又可揭开的场景下，优先标旗（标旗不会触发死亡，且能降低后续推理复杂度）。
+4. **禁止重复操作**：绝不能点击 0-8 的格子或 F 的格子。
+
+# 五、 输出格式要求（严格遵守）
+不要输出任何解释性文本、问候语或分析过程。你的输出必须且只能是以下两种格式之一：
+
+【格式1：执行操作】
+
+工具名称(参数)
+
+
+一句理由，不超过15字，如：减法逻辑(5,4)安全
+
+
+【格式2：需要更多信息】
+
+get_board_state()
+
+
+初始化/查看全局
+""",
+ label="系统提示词",
+ )
+
+ temperature = IntConfig(
+ default=30,
+ label="温度参数(0-100)",
+ )
diff --git a/src/plugins/llm_minesweeper_controller/function_registry.py b/src/plugins/llm_minesweeper_controller/function_registry.py
new file mode 100644
index 0000000..a637c9b
--- /dev/null
+++ b/src/plugins/llm_minesweeper_controller/function_registry.py
@@ -0,0 +1,278 @@
+"""
+Function注册表 - 装饰器注册 + OpenAI tools格式 + 执行处理
+"""
+from __future__ import annotations
+
+import inspect
+import json
+from typing import Dict, Any, Callable, List, Optional, get_type_hints
+from dataclasses import dataclass, field
+
+
+@dataclass
+class ParameterInfo:
+ """参数信息"""
+ name: str
+ type: str
+ description: str = ""
+ required: bool = False
+ default: Any = None
+ enum_values: Optional[List[str]] = None
+
+
+@dataclass
+class FunctionInfo:
+ """函数信息"""
+ name: str
+ description: str
+ parameters: List[ParameterInfo] = field(default_factory=list)
+ function: Optional[Callable] = None
+
+
+class FunctionRegistry:
+ """Function注册表 - 装饰器注册、schema生成、执行处理"""
+
+ def __init__(self):
+ self.functions: Dict[str, FunctionInfo] = {}
+
+ # ═══════════════════════════════════════════════════════════════
+ # 注册相关
+ # ═══════════════════════════════════════════════════════════════
+
+ def register(
+ self,
+ name: Optional[str] = None,
+ description: Optional[str] = None,
+ param_descriptions: Optional[Dict[str, str]] = None
+ ):
+ """
+ 装饰器：注册函数供 LLM 调用
+
+ Example:
+ @registry.register(
+ description="点击格子",
+ param_descriptions={"x": "X坐标", "y": "Y坐标"}
+ )
+ def click_cell(x: int, y: int) -> Dict[str, Any]:
+ return {"success": True}
+ """
+ def decorator(func: Callable) -> Callable:
+ func_name = name or func.__name__
+ func_description = description or (func.__doc__ or "").strip()
+
+ sig = inspect.signature(func)
+ type_hints = get_type_hints(func)
+
+ parameters = []
+ for param_name, param in sig.parameters.items():
+ if param_name == 'self':
+ continue
+
+ param_type = type_hints.get(param_name, Any)
+ type_str = self._get_type_string(param_type)
+ param_desc = (param_descriptions or {}).get(param_name, "")
+ required = param.default == inspect.Parameter.empty
+ default = param.default if param.default != inspect.Parameter.empty else None
+
+ enum_values = None
+ if hasattr(param_type, '__members__'):
+ enum_values = list(param_type.__members__.keys())
+
+ parameters.append(ParameterInfo(
+ name=param_name,
+ type=type_str,
+ description=param_desc,
+ required=required,
+ default=default,
+ enum_values=enum_values
+ ))
+
+ self.functions[func_name] = FunctionInfo(
+ name=func_name,
+ description=func_description,
+ parameters=parameters,
+ function=func
+ )
+ return func
+ return decorator
+
+ def register_function(
+ self,
+ func: Callable,
+ name: Optional[str] = None,
+ description: Optional[str] = None,
+ param_descriptions: Optional[Dict[str, str]] = None
+ ) -> None:
+ """直接注册函数（非装饰器方式）"""
+ decorator = self.register(name, description, param_descriptions)
+ decorator(func)
+
+ # ═══════════════════════════════════════════════════════════════
+ # Schema 生成
+ # ═══════════════════════════════════════════════════════════════
+
+ def get_tools_schema(self) -> List[Dict[str, Any]]:
+ """获取 OpenAI tools 格式 schema"""
+ schemas = []
+ for func_info in self.functions.values():
+ properties = {}
+ required = []
+
+ for param in func_info.parameters:
+ param_schema = {"type": param.type}
+ if param.description:
+ param_schema["description"] = param.description
+ if param.enum_values:
+ param_schema["enum"] = param.enum_values
+ properties[param.name] = param_schema
+
+ if param.required:
+ required.append(param.name)
+
+ tool_schema = {
+ "type": "function",
+ "function": {
+ "name": func_info.name,
+ "description": func_info.description,
+ "parameters": {
+ "type": "object",
+ "properties": properties,
+ }
+ }
+ }
+
+ if required:
+ tool_schema["function"]["parameters"]["required"] = required
+
+ schemas.append(tool_schema)
+
+ return schemas
+
+ # ═══════════════════════════════════════════════════════════════
+ # 执行相关
+ # ═══════════════════════════════════════════════════════════════
+
+ def execute_function(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+ """执行注册的函数"""
+ func_info = self.get_function(name)
+ if not func_info or not func_info.function:
+ return {"success": False, "error": f"函数 '{name}' 未注册"}
+
+ try:
+ sig = inspect.signature(func_info.function)
+ valid_params = {}
+
+ for param_name, param_value in arguments.items():
+ if param_name in sig.parameters:
+ # 类型转换
+ param_type = sig.parameters[param_name].annotation
+ if param_type != inspect.Parameter.empty:
+ try:
+ if param_type == int:
+ param_value = int(param_value)
+ elif param_type == float:
+ param_value = float(param_value)
+ elif param_type == bool:
+ param_value = bool(param_value)
+ except (ValueError, TypeError):
+ pass
+ valid_params[param_name] = param_value
+
+ result = func_info.function(**valid_params)
+
+ if isinstance(result, dict):
+ if "success" not in result:
+ result["success"] = True
+ return result
+ else:
+ return {"success": True, "result": result}
+
+ except Exception as e:
+ return {"success": False, "error": f"函数执行失败: {str(e)}"}
+
+ def handle_tool_calls(self, tool_calls: List[Dict[str, Any]],
+ logger=None, widget=None) -> List[Dict[str, Any]]:
+ """
+ 处理 OpenAI 响应中的 tool_calls，返回工具结果消息列表
+
+ Args:
+ tool_calls: OpenAI 响应中的 tool_calls 字段
+ logger: 可选的日志记录器
+ widget: 可选的 UI 组件（用于显示日志）
+
+ Returns:
+ 工具结果消息列表，可直接追加到 messages 中
+ """
+ results = []
+ for tool_call in tool_calls:
+ tool_call_id = tool_call.get("id", "")
+ function_data = tool_call.get("function", {})
+ func_name = function_data.get("name", "")
+
+ try:
+ func_args_str = function_data.get("arguments", "{}")
+ func_args = json.loads(func_args_str) if isinstance(func_args_str, str) else func_args_str
+ except json.JSONDecodeError:
+ func_args = {}
+
+ if logger:
+ logger.info(f"处理 tool call: {func_name}, 参数: {func_args}")
+ if widget:
+ widget.log_message(f"LLM 调用: {func_name}({func_args})")
+
+ # 执行函数
+ result = self.execute_function(func_name, func_args)
+
+ if widget:
+ if result.get("success"):
+ widget.log_message(f"执行成功: {func_name}")
+ else:
+ widget.log_message(f"执行失败: {func_name} - {result.get('error', '未知错误')}")
+
+ # 构建工具结果消息
+ result_msg = {
+ "role": "tool",
+ "tool_call_id": tool_call_id,
+ "content": json.dumps(result, ensure_ascii=False)
+ }
+ results.append(result_msg)
+
+ return results
+
+ # ═══════════════════════════════════════════════════════════════
+ # 工具方法
+ # ═══════════════════════════════════════════════════════════════
+
+ def get_function(self, name: str) -> Optional[FunctionInfo]:
+ return self.functions.get(name)
+
+ def get_all_functions(self) -> List[FunctionInfo]:
+ return list(self.functions.values())
+
+ def list_function_names(self) -> List[str]:
+ return list(self.functions.keys())
+
+ def clear(self) -> None:
+ self.functions.clear()
+
+ def remove_function(self, name: str) -> bool:
+ if name in self.functions:
+ del self.functions[name]
+ return True
+ return False
+
+ @staticmethod
+ def _get_type_string(type_hint) -> str:
+ """将类型提示转换为 JSON Schema 类型字符串"""
+ if type_hint == int:
+ return "integer"
+ elif type_hint == float:
+ return "number"
+ elif type_hint == bool:
+ return "boolean"
+ elif type_hint == list or (hasattr(type_hint, '__origin__') and type_hint.__origin__ == list):
+ return "array"
+ elif type_hint == dict or (hasattr(type_hint, '__origin__') and type_hint.__origin__ == dict):
+ return "object"
+ else:
+ return "string"
\ No newline at end of file
diff --git a/src/plugins/llm_minesweeper_controller/plugin.py b/src/plugins/llm_minesweeper_controller/plugin.py
new file mode 100644
index 0000000..349c692
--- /dev/null
+++ b/src/plugins/llm_minesweeper_controller/plugin.py
@@ -0,0 +1,636 @@
+"""
+llm_minesweeper_controller - 插件主类
+"""
+from __future__ import annotations
+
+import json
+from typing import Dict, Any, Optional, List
+
+from PyQt5.QtWidgets import QWidget
+from PyQt5.QtCore import QThread, pyqtSignal
+
+from plugin_sdk import BasePlugin, PluginInfo, make_plugin_icon, WindowMode
+from shared_types.events import BoardUpdateEvent, GameStatusChangeEvent
+from shared_types.commands import NewGameCommand, MouseClickCommand
+
+from .widgets import LlmMinesweeperControllerWidget
+from .config import LlmMinesweeperControllerConfig
+from .api_client import LLMClient, ChatResponse
+from .function_registry import FunctionRegistry
+
+
+class LLMWorker(QThread):
+ """LLM 工作线程"""
+
+ log_signal = pyqtSignal(str)
+ chat_signal = pyqtSignal(str, str) # role, text
+ finished_signal = pyqtSignal(bool, str) # success, message
+
+ def __init__(self, client: LLMClient, registry: FunctionRegistry,
+ messages: List[Dict[str, Any]]):
+ super().__init__()
+ self.client = client
+ self.registry = registry
+ self.messages = messages
+
+ def run(self):
+ """执行多轮对话循环（无上限）"""
+ try:
+ tools = self.registry.get_tools_schema()
+ round_num = 0
+
+ while True:
+ round_num += 1
+ self.log_signal.emit(f"=== 第 {round_num} 轮对话 ===")
+
+ # 调用 LLM
+ response: ChatResponse = self.client.chat(
+ messages=self.messages,
+ tools=tools,
+ temperature=0.3,
+ )
+
+ if not response.success:
+ self.finished_signal.emit(
+ False, f"API 调用失败: {response.error}")
+ return
+
+ # 显示 LLM 文本回复
+ if response.has_content:
+ self.chat_signal.emit("assistant", response.content)
+
+ # 检查是否需要调用工具
+ if response.has_tool_calls:
+ # 将 assistant 消息加入历史
+ self.messages.append(
+ LLMClient.build_assistant_tool_message(
+ response.content,
+ response.tool_calls
+ )
+ )
+
+ # 处理每个 tool_call
+ for tool_call in response.tool_calls:
+ tool_call_id = tool_call.get("id", "")
+ func_data = tool_call.get("function", {})
+ func_name = func_data.get("name", "")
+
+ try:
+ func_args_str = func_data.get("arguments", "{}")
+ func_args = json.loads(func_args_str) if isinstance(
+ func_args_str, str) else func_args_str
+ except json.JSONDecodeError:
+ func_args = {}
+
+ self.log_signal.emit(f"调用函数: {func_name}({func_args})")
+ self.chat_signal.emit(
+ "tool", f"{func_name}({func_args})")
+
+ # 执行函数
+ result = self.registry.execute_function(
+ func_name, func_args)
+
+ self.log_signal.emit(f"执行结果: {result}")
+
+ # 构建 tool 结果消息
+ tool_msg = LLMClient.build_tool_result_message(
+ tool_call_id, result)
+ self.messages.append(tool_msg)
+
+ # 继续下一轮对话（让 LLM 处理工具结果）
+ continue
+
+ # 没有 tool_calls，对话结束
+ self.finished_signal.emit(True, "LLM 分析完成")
+ return
+
+ except Exception as e:
+ self.finished_signal.emit(False, f"执行异常: {str(e)}")
+
+
+class LlmMinesweeperControllerPlugin(BasePlugin):
+ """使用 LLM 控制扫雷的插件"""
+
+ @classmethod
+ def plugin_info(cls) -> PluginInfo:
+ return PluginInfo(
+ name="llm_minesweeper_controller",
+ version="1.0.0",
+ description="使用 LLM 分析并控制扫雷游戏",
+ window_mode=WindowMode.TAB,
+ icon=make_plugin_icon("#4CAF50", "L"),
+ other_info=LlmMinesweeperControllerConfig,
+ required_controls=[NewGameCommand, MouseClickCommand],
+ )
+
+ def _setup_subscriptions(self) -> None:
+ self.subscribe(BoardUpdateEvent, self._on_board_update)
+ self.subscribe(GameStatusChangeEvent, self._on_game_status_change)
+
+ def _create_widget(self) -> QWidget | None:
+ self._widget = LlmMinesweeperControllerWidget()
+ return self._widget
+
+ def on_initialized(self) -> None:
+ self.logger.info("LlmMinesweeperControllerPlugin 已初始化")
+
+ # 检查控制权限状态
+ self._log_control_auth_status()
+
+ # 初始化组件
+ self._init_llm_client()
+ self._init_function_registry()
+
+ # 设置 UI 回调
+ self._widget.set_test_button_callback(self._test_connection)
+ self._widget.set_analyze_callback(self._start_analysis)
+
+ # 监听配置变化
+ self.config_changed.connect(self._on_config_changed)
+
+ # 当前棋盘状态
+ self._current_board: Optional[Dict[str, Any]] = None
+
+ # 当前游戏状态 (1=ready, 2=playing, 3=win, 4=fail, ...)
+ self._game_status: int = 1
+
+ # 当前工作线程
+ self._worker: Optional[LLMWorker] = None
+
+ def _log_control_auth_status(self) -> None:
+ """检查控制权限"""
+ has_new_game = self.has_control_auth(NewGameCommand)
+ has_click = self.has_control_auth(MouseClickCommand)
+ self.logger.info(f"NewGameCommand 权限: {has_new_game}")
+ self.logger.info(f"MouseClickCommand 权限: {has_click}")
+ self._widget.log_message(
+ f"权限: NewGame={has_new_game}, MouseClick={has_click}")
+
+ def _init_llm_client(self) -> None:
+ """初始化 LLM 客户端"""
+ api_key = self.other_info.api_key
+ base_url = self.other_info.api_base_url
+ model = self.other_info.model_name
+ timeout = self.other_info.request_timeout
+
+ if api_key:
+ self.llm_client = LLMClient(
+ api_key=api_key,
+ base_url=base_url,
+ model=model,
+ timeout=timeout,
+ )
+ self.logger.info("LLM 客户端已初始化")
+ self._widget.log_message(f"LLM 客户端已初始化 (model: {model})")
+ else:
+ self.llm_client = None
+ self.logger.warning("未配置 API 密钥")
+ self._widget.log_message("未配置 API 密钥，LLM 功能不可用")
+
+ def _init_function_registry(self) -> None:
+ """初始化 Function 注册表"""
+ self.function_registry = FunctionRegistry()
+ self._register_minesweeper_functions()
+
+ def _register_minesweeper_functions(self) -> None:
+ """注册扫雷相关的函数"""
+ registry = self.function_registry
+
+ @registry.register(
+ description="点击指定位置的格子",
+ param_descriptions={
+ "col": "列索引 (从 0 开始，即 X 坐标)",
+ "row": "行索引 (从 0 开始，即 Y 坐标)",
+ "button": "鼠标按钮: 'left' 左键揭开, 'right' 右键标旗",
+ }
+ )
+ def click_cell(col: int, row: int, button: str = "left") -> Dict[str, Any]:
+ return self._execute_click_cell(col, row, button)
+
+ @registry.register(
+ description="开始新游戏，不传难度参数则使用默认难度",
+ param_descriptions={
+ "difficulty": "游戏难度（可选）: 'easy' 初级(8x8), 'medium' 中级(16x16), 'hard' 高级(16x30)",
+ }
+ )
+ def start_new_game(difficulty: str = None) -> Dict[str, Any]:
+ return self._execute_start_new_game(difficulty)
+
+ @registry.register(
+ description="获取当前棋盘状态",
+ )
+ def get_board_state() -> Dict[str, Any]:
+ return self._get_current_board_state()
+
+ @registry.register(
+ description="获取局部棋盘区域，返回以(col,row)为中心的局部格子，radius自己决定（建议3-5）",
+ param_descriptions={
+ "col": "中心列索引 (从 0 开始)",
+ "row": "中心行索引 (从 0 开始)",
+ "radius": "半径，自己决定大小，默认3，返回(2*radius+1)x(2*radius+1)的区域",
+ }
+ )
+ def get_local_board(col: int, row: int, radius: int = 3) -> Dict[str, Any]:
+ return self._get_local_board(col, row, radius)
+
+ def on_control_auth_changed(self, cmd_type, granted: bool) -> None:
+ """控制权限变更回调"""
+ if cmd_type == NewGameCommand:
+ self._widget.log_message(
+ f"NewGameCommand 权限: {'已授权' if granted else '未授权'}")
+ elif cmd_type == MouseClickCommand:
+ self._widget.log_message(
+ f"MouseClickCommand 权限: {'已授权' if granted else '未授权'}")
+
+ def _on_config_changed(self, name: str, value) -> None:
+ """配置变化回调"""
+ self.logger.info(f"配置变化: {name} = {value}")
+ self._widget.log_message(f"配置更新: {name}")
+
+ # API 相关配置变化时重新初始化客户端
+ if name in ["api_key", "api_base_url", "model_name", "request_timeout"]:
+ self._init_llm_client()
+
+ def _on_board_update(self, event: BoardUpdateEvent) -> None:
+ """处理棋盘更新事件"""
+ self._widget.log_message("收到棋盘更新事件")
+
+ # 更新当前棋盘状态
+ self._current_board = self._extract_board_data(event)
+
+ def _on_game_status_change(self, event: GameStatusChangeEvent) -> None:
+ """处理游戏状态变化事件
+
+ 状态值：
+ - 1: ready (准备)
+ - 2: playing (游戏中)
+ - 3: win (胜利)
+ - 4: fail (失败)
+ - 5: show (显示概率)
+ - 6: study (研究模式)
+ - 7/8: display 相关
+ """
+ status_names = {
+ 1: "准备",
+ 2: "游戏中",
+ 3: "胜利",
+ 4: "失败",
+ 5: "显示概率",
+ 6: "研究模式",
+ 7: "播放录像",
+ 8: "播放概率",
+ }
+
+ last_name = status_names.get(
+ event.last_status, f"未知({event.last_status})")
+ current_name = status_names.get(
+ event.current_status, f"未知({event.current_status})")
+
+ self._widget.log_message(f"游戏状态变化: {last_name} -> {current_name}")
+
+ # 记录当前游戏状态
+ self._game_status = event.current_status
+
+ # 如果游戏结束（胜利或失败），更新状态显示
+ if event.current_status == 3:
+ self._widget.update_status("游戏胜利!")
+ elif event.current_status == 4:
+ self._widget.update_status("游戏失败!")
+
+ # ═══════════════════════════════════════════════════════════════
+ # LLM 对话流程
+ # ═══════════════════════════════════════════════════════════════
+
+ def _test_connection(self) -> None:
+ """测试 API 连接"""
+ if not self.llm_client:
+ self._widget.log_message("请先配置 API 密钥")
+ return
+
+ self._widget.log_message("正在测试连接...")
+ self._widget.set_buttons_enabled(False)
+
+ response = self.llm_client.test_connection()
+
+ if response.success:
+ self._widget.update_status("连接成功")
+ self._widget.log_message(f"连接成功! 模型: {self.other_info.model_name}")
+ else:
+ self._widget.update_status("连接失败")
+ self._widget.log_message(f"连接失败: {response.error}")
+
+ self._widget.set_buttons_enabled(True)
+
+ def _auto_continue_analysis(self) -> None:
+ """自动继续分析（异常中断后）"""
+ if self._game_status != 2: # 不是进行中
+ self._widget.log_message("游戏已结束，不再继续分析")
+ return
+
+ if self._worker and self._worker.isRunning():
+ return
+
+ self._widget.log_message("自动继续分析...")
+ self._start_analysis()
+
+ def _start_analysis(self) -> None:
+ """开始 LLM 分析"""
+ if not self.llm_client:
+ self._widget.log_message("请先配置 API 密钥")
+ return
+
+ if self._worker and self._worker.isRunning():
+ self._widget.log_message("已有分析任务在运行")
+ return
+
+ # 构建初始消息
+ messages = self._build_initial_messages()
+
+ # 创建并启动工作线程
+ self._worker = LLMWorker(
+ client=self.llm_client,
+ registry=self.function_registry,
+ messages=messages,
+ )
+
+ # 连接信号
+ self._worker.log_signal.connect(self._widget.log_message)
+ self._worker.chat_signal.connect(self._widget.add_chat_message)
+ self._worker.finished_signal.connect(self._on_analysis_finished)
+
+ self._widget.set_buttons_enabled(False)
+ self._worker.start()
+
+ def _on_analysis_finished(self, success: bool, message: str) -> None:
+ """分析完成回调"""
+ self._widget.set_buttons_enabled(True)
+
+ if success:
+ self._widget.update_status("分析完成")
+ else:
+ self._widget.update_status("分析中断")
+ self._widget.log_message(message)
+
+ # 无论成功还是失败，如果游戏状态是进行中，自动继续分析
+ # 防止AI没有进行任何函数调用就结束
+ if self._game_status == 2: # playing
+ self._widget.log_message("游戏进行中，1秒后继续分析...")
+ from PyQt5.QtCore import QTimer
+ QTimer.singleShot(1000, self._auto_continue_analysis)
+
+ def _build_initial_messages(self) -> List[Dict[str, Any]]:
+ """构建初始消息列表"""
+ messages = []
+
+ # 系统提示词
+ system_prompt = self.other_info.system_prompt
+ if system_prompt:
+ messages.append({"role": "system", "content": system_prompt})
+
+ # 当前棋盘状态
+ board_state = self._get_current_board_state()
+ game_status = board_state.get('game_status', 'unknown')
+
+ board_info = f"""当前棋盘状态:
+- 行数: {board_state.get('rows', 0)}
+- 列数: {board_state.get('cols', 0)}
+- 剩余地雷: {board_state.get('mines_remaining', 0)}
+- 游戏时间: {board_state.get('game_time', 0):.1f}秒
+- 游戏状态: {game_status}
+- 棋盘数据 (cells[row][col], -1=未揭开, 0-8=周围地雷数, F=标旗, M=踩到的地雷):
+{json.dumps(board_state.get('cells', []), ensure_ascii=False)}
+
+请分析当前局面并选择最佳操作。"""
+
+ messages.append({"role": "user", "content": board_info})
+
+ return messages
+
+ # ═══════════════════════════════════════════════════════════════
+ # 可调用的函数实现
+ # ═══════════════════════════════════════════════════════════════
+
+ def _execute_click_cell(self, col: int, row: int, button: str = "left") -> Dict[str, Any]:
+ """执行点击格子操作
+
+ Args:
+ col: 列索引 (x 坐标, 从 0 开始)
+ row: 行索引 (y 坐标, 从 0 开始)
+ button: 鼠标按钮 ("left" 或 "right")
+ """
+ if not self.has_control_auth(MouseClickCommand):
+ return {"success": False, "error": "无权限执行鼠标点击命令"}
+
+ try:
+ # button 转换: "left" -> 0, "right" -> 2
+ button_map = {"left": 0, "right": 2, "middle": 1}
+ button_value = button_map.get(button.lower(), 0)
+
+ # 构建并发送鼠标点击命令
+ click_cmd = MouseClickCommand(
+ row=row,
+ col=col,
+ button=button_value,
+ )
+ self.send_command(click_cmd)
+
+ self._widget.log_message(f"已点击格子: 行{row}, 列{col}, 按钮: {button}")
+ return {
+ "success": True,
+ "message": f"格子点击已执行: 行{row}, 列{col}, {button}",
+ "coordinates": {"row": row, "col": col},
+ "button": button,
+ }
+
+ except Exception as e:
+ error_msg = f"执行格子点击失败: {str(e)}"
+ self.logger.error(error_msg)
+ return {"success": False, "error": error_msg}
+
+ def _execute_start_new_game(self, difficulty: str = None) -> Dict[str, Any]:
+ """执行开始新游戏操作
+
+ Args:
+ difficulty: 游戏难度 ("easy", "medium", "hard")，默认使用配置中的 default_difficulty
+ """
+ if not self.has_control_auth(NewGameCommand):
+ return {"success": False, "error": "无权限执行新游戏命令"}
+
+ try:
+ # 使用配置中的默认难度
+ if difficulty is None:
+ difficulty = self.other_info.default_difficulty
+
+ # 难度映射: level 值
+ # BEGINNER = 3, INTERMEDIATE = 4, EXPERT = 5
+ difficulty_map = {
+ "easy": 3, # 初级
+ "medium": 4, # 中级
+ "hard": 5, # 高级
+ }
+
+ level = difficulty_map.get(difficulty, 4)
+
+ # 清空当前棋盘状态
+ self._current_board = None
+
+ # 发送新游戏命令（使用 level 字段）
+ new_game_cmd = NewGameCommand(level=level)
+ self.send_command(new_game_cmd)
+
+ # 获取对应的行列地雷数用于日志
+ board_params = {
+ "easy": (8, 8, 10),
+ "medium": (16, 16, 40),
+ "hard": (16, 30, 99),
+ }
+ rows, cols, mines = board_params.get(difficulty, (16, 16, 40))
+
+ self._widget.log_message(
+ f"已开始新游戏，难度: {difficulty} ({rows}x{cols}, {mines}雷)")
+ return {
+ "success": True,
+ "message": f"新游戏已开始，难度: {difficulty}",
+ "difficulty": difficulty,
+ "level": level,
+ "rows": rows,
+ "cols": cols,
+ "mines": mines,
+ }
+
+ except Exception as e:
+ error_msg = f"开始新游戏失败: {str(e)}"
+ self.logger.error(error_msg)
+ return {"success": False, "error": error_msg}
+
+ def _get_game_status_text(self) -> str:
+ """获取游戏状态的文字描述"""
+ status_map = {
+ 1: "ready", # 准备
+ 2: "playing", # 游戏中
+ 3: "win", # 胜利
+ 4: "fail", # 失败
+ 5: "show", # 显示概率
+ 6: "study", # 研究模式
+ 7: "display", # 播放录像
+ 8: "showdisplay", # 播放概率
+ }
+ return status_map.get(self._game_status, "unknown")
+
+ def _get_current_board_state(self) -> Dict[str, Any]:
+ """获取当前棋盘状态"""
+ if self._current_board:
+ result = self._current_board.copy()
+ result["game_status"] = self._get_game_status_text()
+ return result
+ return {
+ "rows": 0,
+ "cols": 0,
+ "cells": [],
+ "mines_remaining": 0,
+ "game_time": 0.0,
+ "game_status": self._get_game_status_text(),
+ }
+
+ def _get_local_board(self, col: int, row: int, radius: int = 2) -> Dict[str, Any]:
+ """获取局部棋盘区域
+
+ Args:
+ col: 中心列索引
+ row: 中心行索引
+ radius: 半径，默认2（返回5x5区域）
+
+ Returns:
+ 局部棋盘数据，包含：
+ - cells: 局部格子数据
+ - center: 中心坐标（相对局部区域）
+ - offset: 局部区域在全图的起始偏移
+ - size: 局部区域大小
+ - full_board_size: 全图大小
+ """
+ if not self._current_board or not self._current_board.get("cells"):
+ return {
+ "cells": [],
+ "center": {"col": 0, "row": 0},
+ "offset": {"col": 0, "row": 0},
+ "size": {"rows": 0, "cols": 0},
+ "full_board_size": {"rows": 0, "cols": 0},
+ "error": "棋盘数据不可用",
+ }
+
+ full_cells = self._current_board.get("cells", [])
+ full_rows = len(full_cells)
+ full_cols = len(full_cells[0]) if full_cells else 0
+
+ if full_rows == 0 or full_cols == 0:
+ return {
+ "cells": [],
+ "center": {"col": 0, "row": 0},
+ "offset": {"col": 0, "row": 0},
+ "size": {"rows": 0, "cols": 0},
+ "full_board_size": {"rows": 0, "cols": 0},
+ "error": "棋盘为空",
+ }
+
+ # 计算局部区域边界
+ start_row = max(0, row - radius)
+ end_row = min(full_rows, row + radius + 1)
+ start_col = max(0, col - radius)
+ end_col = min(full_cols, col + radius + 1)
+
+ # 提取局部格子
+ local_cells = []
+ for r in range(start_row, end_row):
+ local_row = []
+ for c in range(start_col, end_col):
+ local_row.append(full_cells[r][c])
+ local_cells.append(local_row)
+
+ # 计算中心在局部区域中的位置
+ center_col = col - start_col
+ center_row = row - start_row
+
+ return {
+ "cells": local_cells,
+ "center": {"col": center_col, "row": center_row},
+ "offset": {"col": start_col, "row": start_row},
+ "size": {"rows": len(local_cells), "cols": len(local_cells[0]) if local_cells else 0},
+ "full_board_size": {"rows": full_rows, "cols": full_cols},
+ }
+
+ def _extract_board_data(self, event: BoardUpdateEvent) -> Dict[str, Any]:
+ """从事件中提取棋盘数据
+
+ game_board 值含义：
+ - 0-8: 已揭开的数字格子
+ - 10: 未揭开的格子
+ - 11: 标旗的格子
+ - 15: 踩到的地雷（红雷），游戏失败
+ - 16: 未踩到的地雷（白雷）
+ """
+ game_board = event.game_board or []
+
+ # 转换为 LLM 友好格式：-1=未揭开, 0-8=数字, F=标旗, M=地雷(踩到), m=地雷(未踩到)
+ cells = []
+ for row in game_board:
+ row_data = []
+ for cell in row:
+ if cell == 10: # 未揭开
+ row_data.append(-1)
+ elif cell == 11: # 标旗
+ row_data.append("F")
+ elif cell == 15: # 踩到的地雷（红雷），游戏失败
+ row_data.append("M")
+ elif cell == 16: # 未踩到的地雷（白雷）
+ row_data.append("m")
+ else: # 0-8 数字
+ row_data.append(cell)
+ cells.append(row_data)
+
+ return {
+ "rows": event.rows,
+ "cols": event.cols,
+ "cells": cells,
+ "mines_remaining": event.mines_remaining,
+ "game_time": event.game_time,
+ }
diff --git a/src/plugins/llm_minesweeper_controller/widgets.py b/src/plugins/llm_minesweeper_controller/widgets.py
new file mode 100644
index 0000000..3cf4f0e
--- /dev/null
+++ b/src/plugins/llm_minesweeper_controller/widgets.py
@@ -0,0 +1,136 @@
+"""
+llm_minesweeper_controller - UI 组件
+"""
+from __future__ import annotations
+
+from PyQt5.QtWidgets import (
+ QWidget, QVBoxLayout, QLabel, QTextEdit, QPushButton,
+ QHBoxLayout, QGroupBox, QSplitter,
+)
+from PyQt5.QtCore import pyqtSignal, Qt
+
+
+class LlmMinesweeperControllerWidget(QWidget):
+ """插件 UI"""
+
+ _log_signal = pyqtSignal(str)
+ _chat_signal = pyqtSignal(str, str) # role, text
+ _status_signal = pyqtSignal(str)
+ _enable_buttons_signal = pyqtSignal(bool)
+
+ def __init__(self, parent=None):
+ super().__init__(parent)
+ layout = QVBoxLayout(self)
+
+ # 状态显示
+ status_group = QGroupBox("状态")
+ status_layout = QHBoxLayout()
+ self._status_label = QLabel("就绪")
+ self._status_label.setStyleSheet("font-weight: bold;")
+ status_layout.addWidget(self._status_label)
+ status_group.setLayout(status_layout)
+ layout.addWidget(status_group)
+
+ # 对话显示区
+ chat_group = QGroupBox("LLM对话")
+ chat_layout = QVBoxLayout()
+ self._chat_text = QTextEdit()
+ self._chat_text.setReadOnly(True)
+ self._chat_text.setStyleSheet("""
+ QTextEdit {
+ font-family: Consolas, 'Microsoft YaHei', monospace;
+ font-size: 13px;
+ }
+ """)
+ chat_layout.addWidget(self._chat_text)
+ chat_group.setLayout(chat_layout)
+ layout.addWidget(chat_group, stretch=1)
+
+ # 日志显示区
+ log_group = QGroupBox("日志")
+ log_layout = QVBoxLayout()
+ self._log_text = QTextEdit()
+ self._log_text.setReadOnly(True)
+ self._log_text.setMaximumHeight(120)
+ self._log_text.setStyleSheet("font-size: 11px; color: #666;")
+ log_layout.addWidget(self._log_text)
+ log_group.setLayout(log_layout)
+ layout.addWidget(log_group)
+
+ # 控制按钮
+ button_layout = QHBoxLayout()
+ self._analyze_button = QPushButton("🤖 分析并操作")
+ self._analyze_button.setStyleSheet("padding: 6px; font-size: 14px;")
+ self._test_button = QPushButton("🔗 测试连接")
+ self._clear_chat_button = QPushButton("🗑 清除对话")
+ self._clear_log_button = QPushButton("🗑 清除日志")
+ button_layout.addWidget(self._analyze_button)
+ button_layout.addWidget(self._test_button)
+ button_layout.addWidget(self._clear_chat_button)
+ button_layout.addWidget(self._clear_log_button)
+ layout.addLayout(button_layout)
+
+ # 信号连接
+ self._log_signal.connect(self._on_log)
+ self._chat_signal.connect(self._on_chat)
+ self._status_signal.connect(self._on_status)
+ self._enable_buttons_signal.connect(self._on_enable_buttons)
+
+ # 按钮事件
+ self._clear_log_button.clicked.connect(self._clear_log)
+ self._clear_chat_button.clicked.connect(self._clear_chat)
+
+ def _on_log(self, text: str) -> None:
+ from datetime import datetime
+ timestamp = datetime.now().strftime("%H:%M:%S")
+ self._log_text.append(f"[{timestamp}] {text}")
+
+ def _on_chat(self, role: str, text: str) -> None:
+ color_map = {
+ "system": "#2196F3",
+ "assistant": "#4CAF50",
+ "tool": "#FF9800",
+ "user": "#9C27B0",
+ "error": "#F44336",
+ }
+ color = color_map.get(role, "#333")
+ self._chat_text.append(
+ f'[{role}] '
+ f'{text}'
+ )
+ # 滚动到底部
+ scrollbar = self._chat_text.verticalScrollBar()
+ scrollbar.setValue(scrollbar.maximum())
+
+ def _on_status(self, text: str) -> None:
+ self._status_label.setText(text)
+
+ def _on_enable_buttons(self, enabled: bool) -> None:
+ self._analyze_button.setEnabled(enabled)
+ self._test_button.setEnabled(enabled)
+
+ def _clear_log(self) -> None:
+ self._log_text.clear()
+
+ def _clear_chat(self) -> None:
+ self._chat_text.clear()
+
+ # ── 线程安全的公开方法（由插件调用） ──
+
+ def log_message(self, text: str) -> None:
+ self._log_signal.emit(text)
+
+ def add_chat_message(self, role: str, text: str) -> None:
+ self._chat_signal.emit(role, text)
+
+ def update_status(self, text: str) -> None:
+ self._status_signal.emit(text)
+
+ def set_buttons_enabled(self, enabled: bool) -> None:
+ self._enable_buttons_signal.emit(enabled)
+
+ def set_analyze_callback(self, callback) -> None:
+ self._analyze_button.clicked.connect(callback)
+
+ def set_test_button_callback(self, callback) -> None:
+ self._test_button.clicked.connect(callback)
diff --git a/src/shared_types/__init__.py b/src/shared_types/__init__.py
index 15163ed..28f2cca 100644
--- a/src/shared_types/__init__.py
+++ b/src/shared_types/__init__.py
@@ -5,6 +5,7 @@
"""
from .events import (
BoardUpdateEvent,
+ GameStatusChangeEvent,
EVENT_TYPES,
)
@@ -24,6 +25,7 @@
__all__ = [
# 事件
"BoardUpdateEvent",
+ "GameStatusChangeEvent",
"EVENT_TYPES",
# 指令
"NewGameCommand",
diff --git a/src/shared_types/commands.py b/src/shared_types/commands.py
index 5cba3ed..8965a11 100644
--- a/src/shared_types/commands.py
+++ b/src/shared_types/commands.py
@@ -3,23 +3,50 @@
"""
from __future__ import annotations
+from typing import Optional
+
from lib_zmq_plugins.shared.base import BaseCommand
+from .enums import GameLevel
+
class NewGameCommand(BaseCommand, tag="new_game"):
- """新游戏指令"""
+ """
+ 新游戏指令
+
+ Attributes:
+ level: 游戏难度，使用 GameLevel 枚举值
+ - 3: 初级 (8x8, 10雷)
+ - 4: 中级 (16x16, 40雷)
+ - 5: 高级 (16x30, 99雷)
+ - 6: 自定义（使用 rows/cols/mines）
+ rows: 行数（自定义模式时使用）
+ cols: 列数（自定义模式时使用）
+ mines: 地雷数（自定义模式时使用）
+ """
+ level: int = 6 # 默认自定义，使用 rows/cols/mines
rows: int = 16
cols: int = 30
mines: int = 99
class MouseClickCommand(BaseCommand, tag="mouse_click"):
- """鼠标点击指令"""
-
+ """
+ 鼠标点击指令
+
+ Attributes:
+ row: 行索引（从 0 开始）
+ col: 列索引（从 0 开始）
+ button: 鼠标按钮
+ - 0: 左键（揭开格子）
+ - 1: 中键
+ - 2: 右键（标旗）
+ modifiers: 键盘修饰符（保留）
+ """
row: int = 0
col: int = 0
button: int = 0
modifiers: int = 0
-COMMAND_TYPES = [NewGameCommand, MouseClickCommand]
+COMMAND_TYPES = [NewGameCommand, MouseClickCommand]
\ No newline at end of file
diff --git a/src/shared_types/events.py b/src/shared_types/events.py
index c97ec39..79b4ac4 100644
--- a/src/shared_types/events.py
+++ b/src/shared_types/events.py
@@ -3,19 +3,51 @@
"""
from __future__ import annotations
-from lib_zmq_plugins.shared.base import BaseEvent
+from typing import List, Optional
+from lib_zmq_plugins.shared.base import BaseEvent
-class GameStatusChange(BaseEvent, tag="game_status"):
- last_status: int = 0
- cuurent_status: int = 0
+from .enums import GameBoardState
class BoardUpdateEvent(BaseEvent, tag="board_update"):
- pass
+ """
+ 棋盘更新事件 - 每次棋盘状态变化时发送
+
+ Attributes:
+ rows: 行数
+ cols: 列数
+ game_board: 游戏局面二维数组
+ - 0-8: 已揭开的数字格子
+ - 10: 未揭开的格子
+ - 11: 标旗的格子
+ - 14: 错误标旗（游戏结束时显示）
+ - 15: 爆炸的地雷
+ - 16: 未爆炸的地雷（游戏结束时显示）
+ mines_remaining: 剩余未标出的地雷数（总地雷数 - 已标旗数）
+ game_time: 游戏时间（秒）
+ """
+ rows: int = 0
+ cols: int = 0
+ game_board: List[List[int]] = []
+ mines_remaining: int = 0
+ game_time: float = 0.0
+
+
+class GameStatusChangeEvent(BaseEvent, tag="game_status_change"):
+ """
+ 游戏状态变化事件
+
+ Attributes:
+ last_status: 上一个游戏状态
+ current_status: 当前游戏状态
+ """
+ last_status: int = 0
+ current_status: int = 0
-class ConetxtChangeEvent(BaseEvent, tag="context_change"):
+class ContextChangeEvent(BaseEvent, tag="context_change"):
+ """上下文变化事件"""
pass
@@ -65,5 +97,6 @@ class VideoSaveEvent(BaseEvent, tag="video_save"):
EVENT_TYPES = [
BoardUpdateEvent,
+ GameStatusChangeEvent,
VideoSaveEvent,
-]
+]
\ No newline at end of file
From eff026f1e012e358aa91573f94fefab76cbd1efe Mon Sep 17 00:00:00 2001
From: ljzloser <1312358581@qq.com>
Date: Fri, 17 Apr 2026 23:15:18 +0800
Subject: [PATCH 2/2] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=E4=B8=AD?=
=?UTF-8?q?=E9=94=AE=E5=8F=8C=E5=87=BB=E5=B9=B6=E4=BC=98=E5=8C=96=E6=89=AB?=
=?UTF-8?q?=E9=9B=B7AI=E6=8E=A7=E5=88=B6=E9=80=BB=E8=BE=91?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
src/main.py | 21 ++--
src/mineSweeperGUI.py | 96 +++++-----------
.../llm_minesweeper_controller/config.py | 105 +++++++++++-------
.../llm_minesweeper_controller/plugin.py | 33 +++++-
4 files changed, 132 insertions(+), 123 deletions(-)
diff --git a/src/main.py b/src/main.py
index 1397579..5f83d02 100644
--- a/src/main.py
+++ b/src/main.py
@@ -200,12 +200,12 @@ def cli_check_file(file_path: str) -> int:
ui._plugin_process = plugin_process # 保存引用，防止被 GC
GameServerBridge.instance().start()
-
+
# 注册控制命令处理器（自动在主线程执行）
def handle_new_game(cmd: NewGameCommand):
"""处理新游戏命令"""
from lib_zmq_plugins.shared.base import CommandResponse
-
+
# 根据 level 确定参数
if cmd.level == GameLevel.BEGINNER.value:
rows, cols, mines = 8, 8, 10
@@ -216,21 +216,24 @@ def handle_new_game(cmd: NewGameCommand):
else:
# 自定义模式，使用传入的参数
rows, cols, mines = cmd.rows, cmd.cols, cmd.mines
-
- print(f"[NewGameCommand] level={cmd.level}, rows={rows}, cols={cols}, mines={mines}")
+
+ print(
+ f"[NewGameCommand] level={cmd.level}, rows={rows}, cols={cols}, mines={mines}")
ui.setBoard_and_start(rows, cols, mines)
return CommandResponse(request_id=cmd.request_id, success=True)
-
+
def handle_mouse_click(cmd: MouseClickCommand):
"""处理鼠标点击命令"""
from lib_zmq_plugins.shared.base import CommandResponse
-
- print(f"[MouseClickCommand] row={cmd.row}, col={cmd.col}, button={cmd.button}")
+
+ print(
+ f"[MouseClickCommand] row={cmd.row}, col={cmd.col}, button={cmd.button}")
success = ui.execute_cell_click(cmd.row, cmd.col, cmd.button)
return CommandResponse(request_id=cmd.request_id, success=success)
-
+
GameServerBridge.instance().register_handler(NewGameCommand, handle_new_game)
- GameServerBridge.instance().register_handler(MouseClickCommand, handle_mouse_click)
+ GameServerBridge.instance().register_handler(
+ MouseClickCommand, handle_mouse_click)
# _translate = QtCore.QCoreApplication.translate
hwnd = int(ui.mainWindow.winId())
diff --git a/src/mineSweeperGUI.py b/src/mineSweeperGUI.py
index 2ae7cb8..ce6ac5a 100644
--- a/src/mineSweeperGUI.py
+++ b/src/mineSweeperGUI.py
@@ -37,6 +37,7 @@
class MineSweeperGUI(MineSweeperVideoPlayer):
+
def __init__(self, MainWindow: MainWindow, args):
self.mainWindow = MainWindow
self.checksum_guard = metaminesweeper_checksum.ChecksumGuard()
@@ -220,7 +221,7 @@ def game_state(self):
def game_state(self, game_state: str):
# print(self._game_state, " -> " ,game_state)
last_state = self._game_state
-
+
match self._game_state:
case "playing":
self.try_append_evfs(game_state)
@@ -254,9 +255,9 @@ def game_state(self, game_state: str):
self.label.paint_cursor = False
self.label.paintProbability = False
self.num_bar_ui.QWidget.close()
-
+
self._game_state = game_state
-
+
# 发送游戏状态变化事件
state_map = {
"ready": 1,
@@ -277,6 +278,7 @@ def game_state(self, game_state: str):
current_status=state_map.get(game_state, 0),
)
GameServerBridge.instance().send_event(event)
+ self._send_board_update_event()
@property
def row(self):
@@ -509,7 +511,7 @@ def _send_board_update_event(self):
game_board_list = []
for row in ms_board.game_board:
game_board_list.append(list(row))
-
+
event = BoardUpdateEvent(
rows=self.row,
cols=self.column,
@@ -521,79 +523,33 @@ def _send_board_update_event(self):
except Exception:
pass # 忽略发送失败
- def execute_cell_click(self, row: int, col: int, button: int) -> bool:
+ def execute_cell_click(self, row: int, col: int, button: int):
"""
执行格子点击（供外部命令调用）
-
+
Args:
row: 行索引（从 0 开始）
col: 列索引（从 0 开始）
- button: 鼠标按钮（0=左键, 2=右键）
-
- Returns:
- True 表示成功执行
+ button: 鼠标按钮（0=左键, 1=中键, 2=右键）
"""
- # 检查游戏状态
- if self.game_state not in ('ready', 'playing', 'joking'):
- return False
-
- # 检查坐标有效性
if row < 0 or row >= self.row or col < 0 or col >= self.column:
return False
-
- # 转换为像素坐标（中心点）
- i = row * self.pixSize + self.pixSize // 2
- j = col * self.pixSize + self.pixSize // 2
-
- try:
- if button == 0: # 左键
- # 模拟点击流程：按下 -> 抬起
- self.label.ms_board.step('lc', (i, j))
-
- # 处理第一次点击埋雷
- if self.game_state == 'ready':
- if self.label.ms_board.mouse_state == 4:
- if self.board_constraint:
- self.game_state = 'joking'
- else:
- self.game_state = 'playing'
- if self.player_identifier[:6] != "[live]":
- self.disable_screenshot()
- if self.cursor_limit:
- self.limit_cursor()
- self.start_time_unix_2 = QtCore.QDateTime.currentDateTime().toMSecsSinceEpoch()
- self.timer_10ms.start()
- self.score_board_manager.editing_row = -2
- # 埋雷
- self.layMine(row, col)
-
- self.label.ms_board.step('lr', (i, j))
-
- # 检查游戏结束
- if self.label.ms_board.game_board_state == 3:
- self.gameWin()
- elif self.label.ms_board.game_board_state == 4:
- self.gameFailed()
-
- elif button == 2: # 右键
- # 更新剩余雷数
- cell_state = self.label.ms_board.game_board[row][col]
- if cell_state == 11: # 已标旗，取消标旗
- self.mineUnFlagedNum += 1
- self.showMineNum(self.mineUnFlagedNum)
- elif cell_state == 10: # 未揭开，标旗
- self.mineUnFlagedNum -= 1
- self.showMineNum(self.mineUnFlagedNum)
-
- self.label.ms_board.step('rc', (i, j))
- self.label.ms_board.step('rr', (i, j))
-
- self.label.update()
- self._send_board_update_event()
- return True
-
- except Exception:
- return False
+ x = row * self.pixSize
+ y = col * self.pixSize
+ if button == 0:
+ self.mineAreaLeftPressed(x, y)
+ self.mineAreaLeftRelease(x, y)
+ elif button == 1:
+ self.mineAreaLeftPressed(x, y)
+ self.mineAreaLeftAndRightPressed(x, y)
+ self.mineAreaRightRelease(x, y)
+ self.mineAreaLeftRelease(x, y)
+ else:
+ self.mineAreaRightPressed(x, y)
+ self.mineAreaRightRelease(x, y)
+
+ self._send_board_update_event()
+ return True
def gameStart(self):
# 画界面，但是不埋雷。等价于点脸、f2、设置确定后的效果
@@ -688,7 +644,7 @@ def gameFinished(self):
data[key] = getattr(ms_board, key)
event = VideoSaveEvent(**data)
GameServerBridge.instance().send_event(event)
-
+
# 发送棋盘更新事件，让插件知道最终状态
self._send_board_update_event()
diff --git a/src/plugins/llm_minesweeper_controller/config.py b/src/plugins/llm_minesweeper_controller/config.py
index 0aff491..9c38df2 100644
--- a/src/plugins/llm_minesweeper_controller/config.py
+++ b/src/plugins/llm_minesweeper_controller/config.py
@@ -49,52 +49,71 @@ class LlmMinesweeperControllerConfig(OtherInfoBase):
# 提示词设置
system_prompt = LongTextConfig(
- default="""你是一个顶级扫雷AI。你的任务是通过逻辑推理，输出工具调用指令来赢下游戏。
-
-# 一、 棋盘与状态定义
+ default="""你是一个扫雷AI。你只能通过调用工具来操作游戏。
+
+# 绝对规则
+每次回复只允许调用1个工具函数。
+所有推理在你的"内部思考"中完成，不要输出到回复里。
+
+# 推理策略
+- 基础：数字=周围未揭开数 → 全是雷；数字=周围旗子数 → 全安全
+- 进阶：用相邻数字的差值与非共享未知格做减法约束
+- 盲猜（仅无逻辑解时）：选长连续边界中段，绝对禁止猜边角
+
+# 决策树（每次操作前必须按此顺序检查！）
+
+## 检查1：能标旗吗？→ 右键 `"right"`
+```
+对于每个数字N：
+ 未揭开格子数 = N ？
+ → 是：这N个格子必定是雷 → `click_cell(..., "right")` 标旗
+ → 继续检查其他数字
+```
+**标旗不会死！遇到确定的雷必须标旗！**
+
+## 标旗的修正
+如果发现之前标错了旗（数字逻辑矛盾），可以再次 `click_cell(..., "right")` 取消标旗。
+右键点击已标旗(F)的格子 = 取消标旗。
+
+## 检查2：能中键吗？→ 中键 `"middle"`
+```
+对于每个数字N：
+ 周围已标旗数 = N ？
+ → 是：周围剩余格子全部安全 → `click_cell(..., "middle")` 批量揭开
+ → 继续检查其他数字
+```
+**这是最常用的批量揭开操作，必须优先使用！**
+
+## 检查3：能左键吗？→ 左键 `"left"`
+```
+不属于以上两种情况，但确定安全？
+→ 是：`click_cell(..., "left")` 揭开（通常是数字0）
+```
+
+## 强制规则
+- 检查顺序：标旗 → 中键 → 左键，**不能跳过**
+
+# 游戏状态判断
+- 棋盘出现 M → 调用 start_new_game
+- cells 为空 → 调用 start_new_game
+- 否则 → 分析推理后调用 click_cell 或 get_local_board
+
+# 操作流程
+1. 先调用 get_board_state 获取全局
+2. 若有确定操作，直接调用 click_cell
+3. 若需要细节，调用 get_local_board(radius=4)
+4. 循环直到胜利
+
+# 棋盘与状态定义
- 格子状态：`-1`(未揭开)、`0-8`(周围雷数)、`F`(已标旗)、`M`(踩到的红雷)、`m`(未踩的白雷)
-- 游戏状态容错：若棋盘出现 `M` 必为失败；若非雷格全揭开必为胜利；以棋盘实际画面为准，忽略错误的状态参数。
+- 游戏状态容错：若棋盘出现 `M` 必为失败；若非雷格全揭开必为胜利；以棋盘实际画面为准。
-# 二、 可用工具
+# 可用工具
- `get_board_state()`：获取全局视图。
-- `get_local_board(col, row, radius)`：获取局部细节，建议 radius=4。
-- `click_cell(col, row, button)`：执行操作，button仅限 `"left"`(揭开) 或 `"right"`(标旗)。
-- `start_new_game()`：失败或未初始化时调用。
-
-# 三、 核心推理策略（按优先级排序）
-1. **基础定式**：
- - 数字 = 周围未揭开数 → 未揭开格全是雷（右键标旗）。
- - 数字 = 周围旗子数 → 剩余未揭开格全安全（左键揭开）。
-2. **减法逻辑（核心）**：
- - 对比边界上相邻的两个数字，利用它们的差值与非共享未知格的数量，推断特定格子是雷还是安全。
-3. **盲猜原则（仅限无任何逻辑解时）**：
- - **绝对禁止猜边角！**
- - 必须选择**长连续未揭开边界的中段**点击，以最大化获取信息量。
-
-# 四、 操作铁律
-1. **100%确定原则**：没有绝对把握不操作，宁可不动也不犯错。
-2. **单次限量**：每次推理后，只执行 1-3 个确定格子的操作。
-3. **标旗优先**：在既可标旗又可揭开的场景下，优先标旗（标旗不会触发死亡，且能降低后续推理复杂度）。
-4. **禁止重复操作**：绝不能点击 0-8 的格子或 F 的格子。
-
-# 五、 输出格式要求（严格遵守）
-不要输出任何解释性文本、问候语或分析过程。你的输出必须且只能是以下两种格式之一：
-
-【格式1：执行操作】
-
-工具名称(参数)
-
-
-一句理由，不超过15字，如：减法逻辑(5,4)安全
-
-
-【格式2：需要更多信息】
-
-get_board_state()
-
-
-初始化/查看全局
-""",
+- `get_local_board(col, row, radius=4)`：获取局部细节，返回(2*radius+1)x(2*radius+1)的区域。
+- `click_cell(col, row, button)`：执行操作，button可为 `"left"`(揭开)、`"right"`(标旗) 或 `"middle"`(快速揭开周围格子)。
+- `start_new_game(difficulty)`：开始新游戏，difficulty为 `"easy"`(8x8)、`"medium"`(16x16) 或 `"hard"`(16x30)。
+""",
label="系统提示词",
)
diff --git a/src/plugins/llm_minesweeper_controller/plugin.py b/src/plugins/llm_minesweeper_controller/plugin.py
index 349c692..be38856 100644
--- a/src/plugins/llm_minesweeper_controller/plugin.py
+++ b/src/plugins/llm_minesweeper_controller/plugin.py
@@ -3,6 +3,7 @@
"""
from __future__ import annotations
+from ctypes import cast
import json
from typing import Dict, Any, Optional, List
@@ -10,6 +11,7 @@
from PyQt5.QtCore import QThread, pyqtSignal
from plugin_sdk import BasePlugin, PluginInfo, make_plugin_icon, WindowMode
+from plugin_sdk.config_types import OtherInfoBase
from shared_types.events import BoardUpdateEvent, GameStatusChangeEvent
from shared_types.commands import NewGameCommand, MouseClickCommand
@@ -32,6 +34,12 @@ def __init__(self, client: LLMClient, registry: FunctionRegistry,
self.client = client
self.registry = registry
self.messages = messages
+ self._stop_flag = False
+
+ def stop(self) -> None:
+ """请求停止工作线程"""
+ self._stop_flag = True
+ self.requestInterruption()
def run(self):
"""执行多轮对话循环（无上限）"""
@@ -40,6 +48,11 @@ def run(self):
round_num = 0
while True:
+ # 检查停止标志
+ if self._stop_flag or self.isInterruptionRequested():
+ self.finished_signal.emit(False, "用户请求停止")
+ return
+
round_num += 1
self.log_signal.emit(f"=== 第 {round_num} 轮对话 ===")
@@ -111,6 +124,8 @@ def run(self):
class LlmMinesweeperControllerPlugin(BasePlugin):
"""使用 LLM 控制扫雷的插件"""
+ _widget: LlmMinesweeperControllerWidget
+
@classmethod
def plugin_info(cls) -> PluginInfo:
return PluginInfo(
@@ -123,6 +138,10 @@ def plugin_info(cls) -> PluginInfo:
required_controls=[NewGameCommand, MouseClickCommand],
)
+ @property
+ def other_info(self) -> LlmMinesweeperControllerConfig:
+ return super().other_info # type: ignore
+
def _setup_subscriptions(self) -> None:
self.subscribe(BoardUpdateEvent, self._on_board_update)
self.subscribe(GameStatusChangeEvent, self._on_game_status_change)
@@ -201,7 +220,7 @@ def _register_minesweeper_functions(self) -> None:
param_descriptions={
"col": "列索引 (从 0 开始，即 X 坐标)",
"row": "行索引 (从 0 开始，即 Y 坐标)",
- "button": "鼠标按钮: 'left' 左键揭开, 'right' 右键标旗",
+ "button": "鼠标按钮: 'left' 左键揭开, 'right' 右键标旗, 'middle' 中键快速揭开周围格子",
}
)
def click_cell(col: int, row: int, button: str = "left") -> Dict[str, Any]:
@@ -296,6 +315,8 @@ def _on_game_status_change(self, event: GameStatusChangeEvent) -> None:
self._widget.update_status("游戏胜利!")
elif event.current_status == 4:
self._widget.update_status("游戏失败!")
+ elif event.current_status == 1:
+ self._widget.update_status("游戏准备中...")
# ═══════════════════════════════════════════════════════════════
# LLM 对话流程
@@ -363,6 +384,8 @@ def _start_analysis(self) -> None:
def _on_analysis_finished(self, success: bool, message: str) -> None:
"""分析完成回调"""
+ if self._widget is None:
+ return
self._widget.set_buttons_enabled(True)
if success:
@@ -634,3 +657,11 @@ def _extract_board_data(self, event: BoardUpdateEvent) -> Dict[str, Any]:
"mines_remaining": event.mines_remaining,
"game_time": event.game_time,
}
+
+ def on_shutdown(self) -> None:
+ """插件关闭时停止 Worker"""
+ if self._worker and self._worker.isRunning():
+ self.logger.info("正在停止 LLM Worker...")
+ self._worker.stop()
+ self._worker.wait(1) # 等待最多3秒
+ self._worker = None