Files
agentmc/mcfold/api.py
JianFeeeee 94bcd37e66 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 项变异全被抓
2026-09-26 13:17:17 +08:00

444 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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)