93243165bf
用户裁定：完整版寻路 + 可达性判断，且这两个是**两个独立工具**； 未加载走到边界为止并如实报告；执行方式两者都给。 新增 mcfold/pathfind.py： - 移动词表对齐 mineflayer movements.js：forward/step_up/step_down/drop/jump/dig/place - A*（八向 + 高度），可按「节点数/半径」有界 - reachable：只答「能不能到 + 大概多远」，**不给路线**，默认不挖不放 - find_path：给具体路线 + world_changes（要挖/放哪些方块，逐格列出） 修的真 bug（6 个）： 1. is_ground（这格是实心底）与 is_standable（脚能停这格）语义混淆，差一格 → 混用导致平地找不到落脚点，只肯搭桥 2. can_already_pass 判据用错对象（拿「更上方 standable」当「已能通过」， 而那是挖完才成立）→ dig 永不生成，2 格墙走不通 3. dig 只支持挖 1 格 → 4 格墙永远不通。改「挖穿一列」，代价按格数线性 4. 代价权重失衡 → 1 格台阶被挖而不是 step_up（挖穿一排墙） 调高 dig/place 代价 + 加「背离目标的额外代价」后修正 5. ★ 未加载被报成 exhausted（听起来像「地形不通」）：根因是 neighbors() 在未加载格不生成移动 ⇒ A* 的未加载检测永远执行不到 → 改为主动探测边界，现在报 hit_unloaded + 停在边界 + 已走步数 6. Terrain.known 调用签名错；挖清单会含空气 诚实性（用户最在意的一条）： - 未加载格所有判定回 None，**不当成空旷** - 岩浆/水不当地面；solid=True 但致命的（仙人掌/岩浆块/篝火）也被排除 - 认不出的方块 → 不知道（None），不猜 - 碰撞粒度只有 block/empty 两档这个限制**写在返回值里** 判据 19 项；5 项变异全被抓。 ★ 「岩浆当地面」变异第一次**没被抓** —— 查下去发现我的判据测的是冗余代码 （lava/water 本来 solid=False，去掉 hazard 检查也不会红）； 真正承载语义的是 solid=True 但致命的那批方块 → 改用仙人掌才抓到。
514 lines
25 KiB
Python
514 lines
25 KiB
Python
"""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)
|