"""M7 面向 LLM 的 API 形状：把 M3/M4/M5/M6 包成**agent 能自己摸清**的工具箱。 ## 为什么需要这一层 mineflayer 的 API 面向**程序员**（位置参数、抛异常、返回原始对象、读文档）。 LLM 的用法完全不同：它会**猜参数**、把失败当数据读、上下文有限。 plan.md §1d 列了五个维度的对策，本模块逐条落地： | 维度 | 这里怎么做 | |---|---| | 参数 | 全命名、少量必填；每个参数有默认值与合法范围 | | 返回 | 附 `describe` / `summary` 人类可读文本（省 token） | | 失败 | **不抛异常**，返回 `ok:false` + `why` + `how_to`（怎么办） | | 规模 | 默认限量 + 给 `total` 与 `truncated` | | 发现 | `describe()` 自省：列出工具、参数、当前会话状态 | ## 三条硬纪律 1. **不决策**：工具回答「世界是什么样 / 我能做什么」， 不回答「你应该去哪」。工具不替 agent 选路。 2. **失败要可读**：`{"ok": false, "why": ..., "how_to": ...}`。 抛异常等于让 LLM 面对 traceback —— 它会猜，而不是读说明。 3. **有界**：所有返回都有上限。超限时**明确告知被截断**， 并给出 `total`，让 agent 知道该缩小范围而不是以为看全了。 """ from __future__ import annotations import time from typing import Any, Callable from . import rough from .events import render_events from .session import StateSession from .state import WorldState # ---------- 工具自省（agent 靠这个摸索） ---------- #: 工具目录。这是**给 agent 看的契约**，字段名与语义都算公开接口， #: 改动会让依赖它的 agent prompt 失效 ⇒ 改动要当作破坏性变更对待。 TOOLS: dict[str, dict[str, Any]] = { # ---- 观察 ---- "where_am_i": { "summary": "我在哪、朝哪、踩着什么", "params": {}, "returns": "player/standing_on/feet/head/facing", }, "look_at": { "summary": "看向某个坐标（自动算朝向）", "params": {"x": "float，必填", "y": "float，必填", "z": "float，必填"}, "returns": "ok/yaw/pitch", }, "see": { "summary": "从当前位置能不能看见某点", "params": {"x": "float，必填", "y": "float，必填", "z": "float，必填"}, "returns": "visible/reason（clear/blocked/unknown_unloaded/target_unloaded/too_far）", }, "raycast": { "summary": "沿方向投一条线，看撞到什么", "params": {"direction": "'x'|'y'|'z'|'down'|'up'，默认取当前朝向", "max_distance": "float，默认 16"}, "returns": "hit/block/position/distance/reason", }, "block_at": { "summary": "查一个坐标的方块", "params": {"x": "int，必填", "y": "int，必填", "z": "int，必填"}, "returns": "name/state_id/source（registry/unknown-state/unloaded/out-of-world）", }, "column": { "summary": "看一整列的地貌剖面（地表在哪、有多厚）", "params": {"x": "int，缺省用玩家 x", "z": "int，缺省用玩家 z"}, "returns": "surface/non_air_count/profile", }, "surroundings": { "summary": "一次总览：在哪、踩什么、前面是什么、能否站立", "params": {"radius": "int，默认 4"}, "returns": "合并回答，省往返", }, "walkable": { "summary": "以我为中心的可站立分布（字符网格）", "params": {"radius": "int，默认 3"}, "returns": "grid（. 可站 / # 障碍 / ? 未知）+ stats", }, "slice": { "summary": "垂直/水平切面的 ASCII 图（看地形起伏）", "params": {"axis": "'x'|'z'|'y'，默认 'z'", "at": "int，缺省用玩家坐标", "width": "int，默认 16", "height": "int，默认 10"}, "returns": "rows_top_to_bottom/unknown_cells", }, "nearby_blocks": { "summary": "附近方块种类统计（有界）", "params": {"radius": "int，默认 4", "want_air": "bool，默认 False"}, "returns": "kinds/blocks/total/truncated", }, "nearby_entities": { "summary": "附近的实体（按距离排序）", "params": {"radius": "float，默认 16", "limit": "int，默认 20"}, "returns": "entities/count", }, "reachable": { "summary": "【工具1】可达性判断：到得了吗 / 大概多远（不给路线）", "params": {"x": "int，缺省=当前格", "y": "int，缺省=当前格", "z": "int，缺省=当前格", "max_radius": "int，默认 48", "allow_world_change": "bool，默认 False ⇒ 只答「现有地形」能不能过"}, "returns": "reachable/approximate_steps/cost/without_changing_world/stopped_because", }, "find_path": { "summary": "【工具2】完整版寻路：具体怎么走（含挖开/搭桥）", "params": {"x": "int，缺省=当前格", "y": "int，缺省=当前格", "z": "int，缺省=当前格", "max_radius": "int，默认 48", "allow_world_change": "bool，默认 True（允许挖/放）"}, "returns": "reached/moves/world_changes（要挖/放哪些方块）/collision_caveat", }, "inventory": { "summary": "我手上/背包里有什么", "params": {}, "returns": "held/hotbar（认不出的物品只给 id，不编名字）", }, # ---- 事件 ---- "events": { "summary": "取走自上次观察以来发生的变化", "params": {"limit": "int，默认 40", "fine_prefix": "可选，如 'block_changed:' 查细分层"}, "returns": "text（紧凑行）+ count/remaining", }, # ---- 操作 ---- "chat": {"summary": "发一句话", "params": {"text": "str，必填"}, "returns": "ok"}, "command": {"summary": "执行一条服务端命令", "params": {"cmd": "str，必填"}, "returns": "ok"}, "look": {"summary": "转视角（不移动）", "params": {"yaw": "float 0..360", "pitch": "float -90..90"}, "returns": "ok/clamped"}, "move": {"summary": "移动到指定坐标（服务端会做碰撞校验）", "params": {"x": "float", "y": "float", "z": "float"}, "returns": "ok/note"}, "dig": {"summary": "挖掉一个方块", "params": {"x": "int", "y": "int", "z": "int", "face": "str，默认 down"}, "returns": "ok/block（结果看后续 block_changed 事件）"}, "place": {"summary": "在指定面放置手上的方块", "params": {"x": "int", "y": "int", "z": "int", "face": "str，默认 top"}, "returns": "ok/held/warning"}, "use_item": {"summary": "使用手上的东西（吃、拉弓…）", "params": {}, "returns": "ok"}, "select_slot": {"summary": "切换热键栏（0..8）", "params": {"slot": "int 0..8"}, "returns": "ok"}, # ---- 会话 ---- "status": {"summary": "连接与世界的健康状况", "params": {}, "returns": "见返回"}, "describe": {"summary": "列出全部工具与当前会话状态", "params": {}, "returns": "tools/session"}, } class Tools: """给 agent 的工具箱。**不抛异常** —— 失败也返回可读结果。""" def __init__(self, session: StateSession): self.s = session # ---------- 自省 ---------- def describe(self) -> dict[str, Any]: """自省：有哪些工具、参数是什么、现在状态如何。 ★ 这是 LLM 摸索工具箱的主要入口，所以要**一次给全**， 让它不必逐个试错（试错既慢又烧 token）。 """ return { "tools": TOOLS, "tool_count": len(TOOLS), "session": self._session_health(), "conventions": { "坐标": "x/z 水平，y 垂直（可为负）", "方块 source": { "registry": "认得，且在本机注册表里", "unknown-state": "本机注册表没有这个 state id（**多半是 mod 方块**）—— 不会瞎猜", "unloaded": "该区块还没收到 ⇒ 不是空旷，是不知道", "out-of-world": "y 超出世界高度", }, "raycast reason": { "clear": "这段路是通的", "unloaded": "没加载 ⇒ 无法判断，不是通", "out-of-world": "射线出界", "solid": "撞到东西", }, "寻路两个工具的分工": { "reachable": "只答「能不能到 + 大概多远」，默认不挖不放（现有地形）", "find_path": "给具体路线，允许 dig/place；world_changes 列出要改的方块", "未加载": "走到边界为止并如实报告 —— 绝不把「没见过」当「空旷」", "碰撞局限": "数据只有 block/empty 两档，半砖/楼梯当整块 ⇒ 可能绕路但不撞墙", }, "can_see reason": { "clear": "看得见", "blocked": "被挡（by/at 说明被谁挡）", "unknown_unloaded": "这段没加载 ⇒ 无法断言", "target_unloaded": "目标格没加载", "too_far": "超过 max_distance", }, }, "tips": [ "不认识某方块时 source=unknown-state，说明它来自服务端但本机没有 —— " "此时不要假设它的样子或用途。", "unloaded 不等于空。要判断远处是否有东西，先让玩家走过去加载区块。", "操作结果**不乐观写进状态**：dig/place 之后看 events 里的 block_changed。", "本工具箱不做决策：它只回答「世界是什么样」和「我能做什么」。", ], } def _session_health(self) -> dict[str, Any]: c = self.s.client ws = self.s.state st = ws.summary() return { "connected_phase": getattr(c, "phase", "?"), "protocol": getattr(c, "proto_num", "?"), "position": {"x": ws.player.x, "y": ws.player.y, "z": ws.player.z}, "entity_id": ws.player.entity_id, "dimension": ws.player.dimension, "health": ws.player.health, "chunks_loaded": st["chunks_loaded"], "entities": st["entities"], "teleports_confirmed": getattr(self.s, "teleports_confirmed", 0), "decode_failures": len(getattr(c, "decode_failures", [])), "unknown_state_ids": st["unknown_state_ids"], "has_unresolved_unknown": st["has_unresolved_unknown"], "events": self.s.events.stats(), } def status(self) -> dict[str, Any]: """会话健康状况（也是一条诊断命令）。""" h = self._session_health() warns: list[str] = [] if h["decode_failures"]: warns.append(f"有 {h['decode_failures']} 个包解不开（已跳过，不断流）") if h["has_unresolved_unknown"]: warns.append(f"有 {len(h['unknown_state_ids'])} 个 state id 本机认不出" f"（多半是 mod 方块）") if h["chunks_loaded"] == 0: warns.append("一个区块都没加载 ⇒ 任何查询都会说 unloaded") return {"ok": True, **h, "warnings": warns, "describe": "用 describe() 看全部工具"} # ---------- 观察工具 ---------- def where_am_i(self) -> dict[str, Any]: su = rough.surroundings(self.s.state, radius=3) return { "ok": True, "position": su["player"], "standing_on": su["standing_on"], "feet": su["feet"], "head": su["head"], "can_stand_here": su["can_stand_here"], "facing": su["player"]["facing"], "describe": (f"在 ({su['player']['x']}, {su['player']['y']}, {su['player']['z']}) " f"朝向{su['player']['facing']}，" f"踩着 {su['standing_on']['name']}"), } def block_at(self, *, x: int, y: int, z: int) -> dict[str, Any]: r = self.s.state.block_at(int(x), int(y), int(z)) return {"ok": r["source"] == "registry", **r, "describe": f"({x},{y},{z}) = {r['describe']}"} def column(self, *, x: int | None = None, z: int | None = None) -> dict[str, Any]: p = self.s.state.player cx = int(p.x) if x is None else int(x) cz = int(p.z) if z is None else int(z) r = self.s.state.column_profile(cx, cz) return {"ok": True, **r, "describe": (f"({cx},{cz}) 地表 y={r['surface']['y'] if r['surface'] else '?'} " f"（{r['surface']['name'] if r['surface'] else '无'}），" f"非空气 {r['non_air_count']} 层")} def surroundings(self, *, radius: int = 4) -> dict[str, Any]: su = rough.surroundings(self.s.state, radius=radius) return {"ok": True, **su, "describe": (f"踩 {su['standing_on']['name']}，" f"前面 {su['ahead']['block'] or '通畅（' + su['ahead']['reason'] + '）'}")} def see(self, *, x: float, y: float, z: float, max_distance: float = 32.0) -> dict[str, Any]: r = rough.can_see(self.s.state, (float(x), float(y), float(z)), max_distance=max_distance) return {"ok": True, **r, "describe": { "clear": f"看得见 ({x},{y},{z})", "blocked": f"看不见，被 {r.get('by')} 挡住（位于 {r.get('at')}）", "unknown_unloaded": "无法判断：这段区块没加载", "target_unloaded": "无法判断：目标格没加载", "too_far": f"太远（{r.get('distance')} > {max_distance}）", "target_reached": f"目标本身是 {r.get('block')}", "same_point": "就是我自己", "out_of_world": "目标在世界外", }.get(r["reason"], r["reason"])} def raycast(self, *, direction: str | None = None, max_distance: float = 16.0) -> dict[str, Any]: ws = self.s.state d = _direction_vector(ws, direction) if d is None: return {"ok": False, "why": f"不认识的方向 {direction!r}", "how_to": "用 'x'/'-x'/'y'/'-y'/'z'/'-z'，或不传表示当前朝向"} r = rough.raycast(ws, ws.player.x, ws.player.y + 1.62, ws.player.z, *d, max_distance=max_distance) return {"ok": True, **r, "describe": (f"撞到 {r['block']} @ {r['position']}（距离 {r['distance']}）" if r["hit"] else f"未撞到东西：{r['reason']}")} def walkable(self, *, radius: int = 3) -> dict[str, Any]: r = rough.walkable(self.s.state, radius=radius) if not r.get("scanned"): return {"ok": False, "why": r["error"], "how_to": "减小 radius"} return {"ok": True, **r, "describe": (f"{r['stats']['open']} 可站 / {r['stats']['blocked']} 障碍 / " f"{r['stats']['unknown']} 未知（{r['row0_is']}）")} def slice(self, *, axis: str = "z", at: int | None = None, width: int = 16, height: int = 10) -> dict[str, Any]: r = rough.slice_view(self.s.state, axis=axis, at=at, width=width, height=height) if not r.get("scanned"): return {"ok": False, "why": r["error"], "how_to": "减小 width/height，或换一个 axis"} art = "\n".join(r["rows_top_to_bottom"]) return {"ok": True, **r, "ascii": art, "describe": f"{axis} 轴切面 {width}×{height}，{r['unknown_cells']} 格未知"} def nearby_blocks(self, *, radius: int = 4, want_air: bool = False) -> dict[str, Any]: p = self.s.state.player r = self.s.state.nearby_blocks(int(p.x), int(p.y), int(p.z), radius=radius, want_air=want_air) if not r.get("scanned"): return {"ok": False, "why": r["error"], "how_to": "减小 radius"} total_blocks = sum(b["count"] for b in r["blocks"]) return {"ok": True, **r, "total_blocks": total_blocks, "describe": f"半径 {radius} 内 {r['kinds']} 种方块，" f"未识别 {len(r['unknown_state_ids'])} 种 id"} def nearby_entities(self, *, radius: float = 16.0, limit: int = 20) -> dict[str, Any]: r = self.s.state.nearby_entities(radius=radius, limit=limit) return {"ok": True, **r, "describe": f"{r['count']} 个实体在 {radius} 格内"} def inventory(self) -> dict[str, Any]: ws = self.s.state hotbar = {} for i in range(9): n = ws.hotbar_item_name(i) if n: hotbar[i] = n held_idx = getattr(ws.player, "held_slot", None) return { "ok": True, "held_slot": held_idx, "held": ws.held_item_name(), "hotbar": hotbar, "known_slots": len(ws.slots), "note": ("物品只有数字 id（1.20.5+ 不传名称）⇒ 显示为 item#N；" "认不出就留空，不编名字" if hotbar else "还没收到 set_slot ⇒ 背包未知（可先读一次或用 command 领取）"), "describe": (f"手上（第 {held_idx} 格）: {ws.held_item_name() or '未知'}；" f"热键栏 {len(hotbar)} 格有东西"), } # ---------- 寻路（两个独立工具） ---------- def reachable(self, *, x: int | None = None, y: int | None = None, z: int | None = None, max_radius: int = 48, allow_world_change: bool = False) -> dict[str, Any]: """【工具1】**可达性判断**：到得了吗、大概多远。**不给路线**。 ★ 与 find_path 分开（用户明确要求）： 这个便宜、省 token，agent 常常只想知道「那边能过去吗」。 默认 `allow_world_change=False` ⇒ 答的是「**现有地形**能不能过」； 要「挖开/搭桥过去」请用 find_path。 """ from . import pathfind as _pf r = _pf.reachable(self.s.state, x, y, z, max_radius=max_radius, allow_world_change=allow_world_change) r["ok"] = True if r["reachable"]: r["describe"] = (f"能到 ({r['to']})，约 {r['approximate_steps']} 步" + ("（不需要挖/放）" if not allow_world_change else "")) else: r["describe"] = f"到不了 ({r['to']})：{r['note']}" return r def find_path(self, *, x: int | None = None, y: int | None = None, z: int | None = None, max_radius: int = 48, allow_world_change: bool = True) -> dict[str, Any]: """【工具2】**完整版寻路**：具体怎么走，含需要挖开/搭桥的步骤。 返回值里 `world_changes` 列出**要改哪些方块**， 配合 dig/place 工具执行；`needs_world_change=True` 的步骤可能失败 （权限/放不下），失败时要重新寻路。 """ from . import pathfind as _pf r = _pf.find_path(self.s.state, x, y, z, max_radius=max_radius, allow_world_change=allow_world_change) r["ok"] = True if r["reached"]: wc = r["world_changes"] r["describe"] = (f"{r['steps']} 步到 ({r['to']})，代价 {r['cost']}" + (f"；需改动 {wc['count']} 个方块" if wc["count"] else "；不需改世界")) else: r["describe"] = (f"未到达（{r.get('stopped_because')}）" f"，已走到 {r.get('stopped_at')}，共 {r['steps']} 步") if r["collision_caveat"]: r["caveat"] = r["collision_caveat"] return r # ---------- 事件 ---------- def events(self, *, limit: int = 40, fine_prefix: str | None = None) -> dict[str, Any]: if fine_prefix: got = self.s.events.drain_fine(fine_prefix, limit=limit) return {"ok": True, "kind": "fine", "prefix": fine_prefix, "count": len(got), "text": render_events(got)} got = self.s.events.drain(limit=limit) stats = self.s.events.stats() return { "ok": True, "kind": "coarse", "count": len(got), "remaining": stats["coarse_pending"], "dropped": stats["dropped"], "text": render_events(got), "describe": (f"{len(got)} 条新事件，剩余 {stats['coarse_pending']} 条未取" if stats["coarse_pending"] else f"{len(got)} 条新事件"), } # ---------- 操作（全部不抛异常） ---------- def chat(self, *, text: str) -> dict[str, Any]: self.s.chat(text) return {"ok": True, "sent": text, "describe": "已发出（回复走 events 里的 chat）"} def command(self, *, cmd: str) -> dict[str, Any]: self.s.command(cmd) return {"ok": True, "sent": cmd, "describe": "已发出（结果走 events：system_chat 或 block_changed）"} def look(self, *, yaw: float, pitch: float) -> dict[str, Any]: return {"ok": True, **self.s.actions.look(yaw, pitch)} def look_at(self, *, x: float, y: float, z: float) -> dict[str, Any]: return {"ok": True, **self.s.actions.look_at(x, y, z)} def move(self, *, x: float, y: float, z: float) -> dict[str, Any]: return {"ok": True, **self.s.actions.move(x, y, z)} def dig(self, *, x: int, y: int, z: int, face: str = "down") -> dict[str, Any]: r = self.s.actions.dig(int(x), int(y), int(z), face=face) if not r.get("ok"): return {"ok": False, "why": r.get("reason", "未知原因"), "how_to": "确认该区块已加载（先走过去），或换 face", "block": r.get("block"), "source": r.get("source")} return {"ok": True, **r, "describe": f"已发出挖掘意图（{r['block']}）；结果看 events 里的 block_changed"} def place(self, *, x: int, y: int, z: int, face: str = "top") -> dict[str, Any]: r = self.s.actions.place(int(x), int(y), int(z), face=face) if not r.get("ok"): return {"ok": False, "why": r.get("reason", "未知原因"), "how_to": "先看 inventory 确认手上有什么，必要时 command('give ...')"} out = {"ok": True, **r} if r.get("warning"): out["how_to"] = "手上物品未知，若失败先用 inventory 确认" out["describe"] = "已发出放置意图；结果看 events 里的 block_changed" return out def use_item(self) -> dict[str, Any]: return {"ok": True, **self.s.actions.use_item()} def select_slot(self, *, slot: int) -> dict[str, Any]: r = self.s.actions.select_slot(int(slot)) if not r.get("ok"): return {"ok": False, "why": r.get("reason"), "how_to": "槽位范围是 0..8"} return {"ok": True, **r} # ---------- 统一派发（便于函数调用式集成） ---------- def call(self, name: str, **kwargs: Any) -> dict[str, Any]: """按名字调一个工具。**未知工具名也给可读答复**（不抛 KeyError）。""" fn: Callable[..., dict[str, Any]] | None = getattr(self, name, None) if fn is None or name.startswith("_") or name not in TOOLS: return {"ok": False, "why": f"没有工具 {name!r}", "how_to": "用 describe() 列出全部工具", "available": sorted(TOOLS)} try: return fn(**kwargs) except TypeError as e: need = TOOLS.get(name, {}).get("params", {}) return {"ok": False, "why": f"参数不对：{e}", "how_to": f"这个工具的参数是：{need}"} except Exception as e: # 兜底：LLM 绝不该看到 traceback return {"ok": False, "why": f"{type(e).__name__}: {e}", "how_to": "用 status() 看会话是否还正常"} def _direction_vector(ws: WorldState, direction: str | None): if direction is None: import math p = ws.player y, pi = math.radians(p.yaw), math.radians(p.pitch) cp, sp = math.cos(pi), math.sin(pi) return (-math.sin(y) * cp, -sp, math.cos(y) * cp) table = { "x": (1, 0, 0), "-x": (-1, 0, 0), "y": (0, 1, 0), "-y": (0, -1, 0), "up": (0, 1, 0), "down": (0, -1, 0), "z": (0, 0, 1), "-z": (0, 0, -1), } return table.get(direction)