M7：面向 LLM 的 API 形状（23 个工具）+ 17 项判据
新增 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 项变异全被抓
This commit is contained in:
JianFeeeee
443
mcfold/api.py
Normal file
443
mcfold/api.py
Normal file
@ -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)
|
||||
333
tests/test_api.py
Normal file
333
tests/test_api.py
Normal file
@ -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：全部通过")
|
||||
Reference in New Issue
Block a user