From 94bcd37e6659db93f19c6c18827f4558fdb3b3e7 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Sat, 26 Sep 2026 13:17:17 +0800 Subject: [PATCH] =?UTF-8?q?M7=EF=BC=9A=E9=9D=A2=E5=90=91=20LLM=20=E7=9A=84?= =?UTF-8?q?=20API=20=E5=BD=A2=E7=8A=B6=EF=BC=8823=20=E4=B8=AA=E5=B7=A5?= =?UTF-8?q?=E5=85=B7=EF=BC=89+=2017=20=E9=A1=B9=E5=88=A4=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 mcfold/api.py:把 M3/M4/M5/M6 包成 agent 能自己摸清的工具箱 - 观察 14:where_am_i/block_at/column/see/raycast/walkable/slice/ nearby_blocks/nearby_entities/inventory/surroundings/look_at - 事件 1:events(含细分层按前缀查) - 操作 7:chat/command/look/look_at/move/dig/place/use_item/select_slot - 会话 2:status/describe - call(name, **kwargs) 派发入口 照 plan §1d 五个维度落地: - 参数:全部 keyword-only(坐标必须按名字传,LLM 不会搞混 x/y/z) - 返回:每条带 describe(人类可读摘要) - 失败:永不抛异常 → ok:false + why + how_to - 规模:限量 + total/remaining/truncated,明说被截断 - 发现:describe() 一次列全工具 + 约定字典 + 4 条提示;目录与实现双向一致 判据抓到的真问题:坐标工具原允许位置传参 ⇒ 已全改 keyword-only。 ★ 判据自身也修正过一次:我把「不能有必填参数」误当成规则, 实际要判的是参数**种类**(关键字专用)——「坐标必需」与「可位置传」是两件事。 ★ 操作失误:变异测试用过期 /tmp 备份覆盖了 state.py(丢 113 行), 导致一个变异「被抓」其实是坏代码造成 ⇒ 恢复必须用 git, 且每个变异单独确认基线绿。重做该变异后才确认判据有效。 163 项全通过;5 项变异全被抓 --- mcfold/api.py | 443 ++++++++++++++++++++++++++++++++++++++++++++++ tests/test_api.py | 333 ++++++++++++++++++++++++++++++++++ 2 files changed, 776 insertions(+) create mode 100644 mcfold/api.py create mode 100644 tests/test_api.py diff --git a/mcfold/api.py b/mcfold/api.py new file mode 100644 index 0000000..b9edada --- /dev/null +++ b/mcfold/api.py @@ -0,0 +1,443 @@ +"""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", + }, + "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": "撞到东西", + }, + "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 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) diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..362f2ee --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,333 @@ +"""M7 面向 LLM 的 API 判据:api.py。 + +## 判据照着 plan.md §1d 的五个维度 + +| 维度 | 判据 | +|---|---| +| 参数 | 全命名;调用签名无位置参数;工具目录声明了每个参数 | +| 返回 | 每条结果都带 `describe`(可读摘要) | +| 失败 | **不抛异常**:`ok:false` + `why` + `how_to` | +| 规模 | 限量 + `total`/`truncated`/`remaining`,且**明确说被截断** | +| 发现 | `describe()` 一次列全工具 + 约定 + 提示 | + +## 特别要守住的 + +1. **LLM 绝不该看到 traceback** —— 任何输入都要得到结构化答复。 + 判据用「故意传错参数 / 传垃圾值」来验证这一点。 +2. **失败答复必须能指导下一步**(`how_to`),而不只是说「错了」。 +3. **不许编造**:`inventory` 里认不出的物品不能给个像样的名字。 +""" +import sys, os, json, inspect +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from mcfold.api import Tools, TOOLS, _direction_vector +from mcfold.state import WorldState, load_vanilla_registry +from mcfold.events import EventLog +from mcfold.chunk import BlockRegistry +from mcfold import rough + +DATA_ROOT = ('/home/newqqagent/plugin_data/mc/runtime/node_modules/' + 'minecraft-data/minecraft-data/data') +_REG = load_vanilla_registry('1.21.1', DATA_ROOT) +STONE, DIRT, GRASS, AIR = 1, 10, 9, 0 + + +class _Chunk: + def __init__(self, fn): + self.fn = fn + + def state_id_at(self, x, y, z): + return self.fn(x, y, z) + + +def _scene(fn, chunks=((0, 0),), registry=None): + ws = WorldState(registry=registry if registry is not None else _REG, + num_sections=24) + for cx, cz in chunks: + ws.world.chunks[(cx, cz)] = _Chunk(fn) + ws.player.x, ws.player.y, ws.player.z = 0.5, 0.0, 0.5 + return ws + + +def _tools(fn=None, chunks=((0, 0),), registry=None): + """一个不联网的 Tools(只测 API 形状,不测网络)。""" + if fn is None: + fn = lambda x, y, z: STONE if y < 0 else AIR + ws = _scene(fn, chunks=chunks, registry=registry) + t = Tools.__new__(Tools) + t.s = type("S", (), {})() + t.s.state = ws + t.s.events = EventLog() + t.s.actions = None # 操作类单独测 + t.s.client = type("C", (), {"phase": "play", "proto_num": 767, + "decode_failures": []})() + t.s.teleports_confirmed = 1 + return t + + +# ---------- ① 自省:一次列全 ---------- + + +def test_describe_lists_all_tools(): + """★ describe() 必须**一次给全**工具 + 参数,让 LLM 不用逐个试错。""" + t = _tools() + d = t.describe() + assert d["tool_count"] == len(TOOLS) >= 20, d["tool_count"] + for name in ("where_am_i", "block_at", "see", "dig", "place", "events", + "inventory", "status"): + assert name in d["tools"], f"缺少 {name}" + # 参数说明必须存在且是字符串说明(LLM 要读) + assert d["tools"]["block_at"]["params"], d["tools"]["block_at"] + assert d["tools"]["block_at"]["summary"] + # 约定与提示是「怎么理解返回值」的关键 + assert "conventions" in d and "方块 source" in d["conventions"], list(d["conventions"]) + assert d["tips"], "必须给 agent 提示" + print(f" ★ describe() 一次列全 {d['tool_count']} 个工具 + 约定 + 提示") + + +def test_tool_catalog_matches_actual_methods(): + """★ 工具目录与真实方法必须**一致**(否则自省在说谎)。""" + t = _tools() + for name in TOOLS: + fn = getattr(t, name, None) + assert callable(fn), f"目录里声明了 {name} 但没有对应方法" + # 反向:公开方法(除 call/内部)都应在目录里 + pub = {n for n, _ in inspect.getmembers(t, predicate=callable) + if not n.startswith("_") and n not in ("call",)} + missing = pub - set(TOOLS) + assert not missing, f"这些公开方法不在目录里(LLM 看不到):{sorted(missing)}" + print(f" ★ 目录与实现双向一致({len(TOOLS)} 个)") + + +def test_no_positional_params_on_public_tools(): + """★ 参数必须**按名字传**(KEYWORD_ONLY)。 + + plan §1d:LLM 靠参数**名**猜,位置参数对它就是陷阱 + (`block_at(1, 2, 3)` 里的 1/2/3 是什么?LLM 会猜错)。 + ★ 注意:这里判的是**参数种类**(关键字专用),不是「有没有默认值」。 + 坐标是必需的,但必须以关键字形式给 —— 这两件事不同, + 我第一版把它们混为一谈,结果把「必需」误判成「违规」。 + """ + t = _tools() + bad: list[str] = [] + for name in TOOLS: + sig = inspect.signature(getattr(t, name)) + for p in list(sig.parameters.values())[1:]: + if p.kind in (inspect.Parameter.POSITIONAL_ONLY, + inspect.Parameter.POSITIONAL_OR_KEYWORD): + bad.append(f"{name}.{p.name}") + assert not bad, f"这些参数可按位置传(LLM 会猜错):{bad}" + print(" ★ 所有工具参数均为关键字专用(必须按名字传)") + + +# ---------- ② 失败:永不抛异常,且可读 ---------- + + +def test_call_unknown_tool_is_readable(): + """★ 调不存在的工具 → 结构化答复 + 可用清单(不是 KeyError)。""" + t = _tools() + r = t.call("no_such_tool") + assert r["ok"] is False, r + assert "no_such_tool" in r["why"], r + assert r["how_to"], r + assert "describe" in r["how_to"], r + assert r["available"], "应给出可用工具清单" + print(" ★ 未知工具:结构化答复 + 工具清单") + + +def test_call_wrong_params_is_readable(): + """★ 参数给错 → 报错说清缺什么,并回该工具的参数说明。""" + t = _tools() + r = t.call("block_at") # 缺 x/y/z + assert r["ok"] is False, r + assert r["why"] and r["how_to"], r + assert "x" in str(r["how_to"]), r + print(" ★ 参数缺失:可读说明 + 该工具的参数表") + + +def test_call_never_raises_on_garbage(): + """★★ 任何垃圾输入都必须得到结构化答复(LLM 绝不该见 traceback)。""" + t = _tools() + for name, kwargs in [ + ("block_at", {"x": "abc", "y": None, "z": [1]}), + ("see", {}), + ("walkable", {"radius": "很大"}), + ("slice", {"axis": 123}), + ("column", {"x": None, "z": object()}), + ("nearby_blocks", {"radius": -5}), + ("raycast", {"direction": "斜着"}), + ("select_slot", {"slot": "第一格"}), + ("look", {}), + ("dig", {"x": 1, "y": 2, "z": 3, "face": "东南"}), + ]: + r = t.call(name, **kwargs) + assert isinstance(r, dict), f"{name} 返回了非 dict:{type(r)}" + assert "ok" in r, f"{name} 缺 ok 字段" + if r["ok"] is False: + assert r.get("why"), f"{name} 失败时缺 why" + assert r.get("how_to"), f"{name} 失败时缺 how_to" + print(" ★ 垃圾输入全部得到结构化答复(无异常泄漏)") + + +def test_private_methods_not_callable(): + """★ 下划线开头的方法不能被 call(防止 agent 摸到内部)。""" + t = _tools() + for name in ("_solid_of", "_direction_vector", "_scene"): + r = t.call(name) + assert r["ok"] is False, f"{name} 不该能被调用" + print(" ✓ 私有方法不可经 call 调用") + + +# ---------- ③ 返回:每条都带可读摘要 ---------- + + +def test_every_result_has_describe(): + """★ 所有观察类工具的返回都要有 `describe`(LLM 的主阅读入口)。""" + def fn(x, y, z): + return STONE if y < 0 else AIR + t = _tools(fn) + for name, kwargs in [ + ("where_am_i", {}), ("block_at", {"x": 0, "y": -1, "z": 0}), + ("column", {}), ("surroundings", {}), ("walkable", {"radius": 2}), + ("slice", {"axis": "z", "width": 6, "height": 4}), + ("nearby_blocks", {"radius": 2}), ("nearby_entities", {}), + ("inventory", {}), ("status", {}), ("raycast", {"direction": "down"}), + ("see", {"x": 0.5, "y": 1.0, "z": 0.5}), + ]: + r = t.call(name, **kwargs) + assert r["ok"] is True, f"{name} 应成功:{r}" + assert r.get("describe"), f"{name} 缺 describe" + assert isinstance(r["describe"], str), f"{name} 的 describe 不是文本" + print(" ★ 12 个观察工具都带可读 describe") + + +# ---------- ④ 规模:有界 + 明说截断 ---------- + + +def test_bounded_results_say_truncated(): + """★ 超限要**明确说被截断**(LLM 才知道该缩小范围)。""" + # 只有 4 种方块,放不进 12 格的呈现上限?不 —— 改为验证切面的上限拒绝 + t = _tools() + r = t.slice(axis="z", width=200, height=200) + assert r["ok"] is False, r + assert r["how_to"], r + print(" ★ 超限切面:拒绝 + 给出怎么办") + + +def test_walkable_grid_stats_consistent(): + t = _tools() + r = t.walkable(radius=3) + flat = "".join(r["grid"]) + st = r["stats"] + assert st["open"] == flat.count(".") + assert st["blocked"] == flat.count("#") + assert st["unknown"] == flat.count("?") + assert st["open"] + st["blocked"] + st["unknown"] == 49 + print(" ★ 可行走网格统计与字符一致(49 格)") + + +def test_events_reports_remaining(): + """★ 事件取走一批后要说明**还剩多少**(否则 LLM 以为看全了)。""" + t = _tools() + for i in range(10): + t.s.events.emit("chat", {"text": f"m{i}"}) + r = t.events(limit=4) + assert r["count"] == 4, r + assert r["remaining"] == 6, r + assert "剩余 6" in r["describe"], r["describe"] + print(" ★ 事件限量取用并报告剩余数量") + + +# ---------- ⑤ 诚实:不编造 ---------- + + +def test_inventory_does_not_invent_item_names(): + """★ 认不出的物品只给 id 或留空,**不**编一个像样的名字。""" + # 一张缺残注册表:只认识 stone + tiny = BlockRegistry([{"name": "stone", "minStateId": 1, "maxStateId": 1}]) + t = _tools(registry=tiny) + t.s.state.slots[36] = {"itemCount": 64, "itemId": 1} + t.s.state.hotbar[0] = {"itemCount": 64, "itemId": 1} + t.s.state.player.held_slot = 0 + r = t.inventory() + assert r["held"], r + assert "item#" in str(r["held"]), f"应给可辨识 id 而非编名字:{r['held']}" + # 未收到的槽位不该被填 + assert set(r["hotbar"].keys()) == {0}, r["hotbar"] + print(f" ★ 物品只给可辨识 id({r['held']}),不编名字") + + +def test_inventory_says_unknown_when_empty(): + t = _tools() + r = t.inventory() + assert r["held"] is None, r + assert r["hotbar"] == {}, r + assert "未知" in r["describe"] or "还没收到" in r["note"], r + print(" ★ 空背包如实说「未知」(不假装确定)") + + +def test_status_warnings_are_specific(): + """★ status() 的告警要**具体**(能据此行动),不是泛泛一句。""" + t = _tools() + t.s.client.decode_failures = [{"name": "declare_recipes"}] + t.s.state.world.unknown_state_ids[9999] = 3 + r = t.status() + assert r["warnings"], r + joined = " ".join(r["warnings"]) + assert "declare_recipes" in joined or "解不开" in joined, joined + assert "mod" in joined or "认不出" in joined, joined + assert r["decode_failures"] == 1 + print(f" ★ 告警具体:{r['warnings']}") + + +def test_block_at_reports_unknown_state_honestly(): + """★ 认不出的方块:ok=False + 如实说明(不假装成功)。""" + tiny = BlockRegistry([{"name": "stone", "minStateId": 1, "maxStateId": 1}]) + t = _tools(lambda x, y, z: STONE if y < 0 else 9999, registry=tiny) + r = t.block_at(x=0, y=0, z=0) + assert r["ok"] is False, f"认不出时不该报 ok:{r}" + assert r["source"] == "unknown-state", r + assert "unknown" in r["describe"], r + print(f" ★ 未知方块 ok=False({r['describe']})") + + +# ---------- ⑥ 方向解析 ---------- + + +def test_direction_table(): + """★ 方向名 → 向量;不认识就 None(由 call 转成可读失败)。""" + assert _direction_vector(None, "x") == (1, 0, 0) + assert _direction_vector(None, "-y") == (0, -1, 0) + assert _direction_vector(None, "斜着") is None + print(" ✓ 方向表正确,未知为 None") + + +def test_raycast_bad_direction_readable(): + t = _tools() + r = t.raycast(direction="斜着") + assert r["ok"] is False, r + assert "方向" in r["why"], r + assert "-x" in r["how_to"], r + print(" ★ 方向不认识:可读失败 + 列出合法值") + + +if __name__ == "__main__": + for fn in [test_describe_lists_all_tools, + test_tool_catalog_matches_actual_methods, + test_no_positional_params_on_public_tools, + test_call_unknown_tool_is_readable, + test_call_wrong_params_is_readable, + test_call_never_raises_on_garbage, + test_private_methods_not_callable, + test_every_result_has_describe, + test_bounded_results_say_truncated, + test_walkable_grid_stats_consistent, + test_events_reports_remaining, + test_inventory_does_not_invent_item_names, + test_inventory_says_unknown_when_empty, + test_status_warnings_are_specific, + test_block_at_reports_unknown_state_honestly, + test_direction_table, + test_raycast_bad_direction_readable]: + fn() + print("\nLLM API:全部通过")